import bisect
import csv
import io
from datetime import datetime
from pathlib import Path
from typing import Any

from sqlalchemy import func

from app.database import db_session
from app.models import Category, DataFile, DataPackage, DataPackageFile
from app.services.data_management_service import DataManagementService


class PackageManagementService:
    # CSV 存储目录与数据文件相同
    _PACKAGES_SUBDIR = 'packages'
    def __init__(self) -> None:
        self._dm = DataManagementService()

    # ── categories ──────────────────────────────────────────────────────────

    def get_category_tree(self) -> list[dict[str, Any]]:
        with db_session() as session:
            rows = (
                session.query(Category)
                .filter(Category.type == 'data_package')
                .filter(Category.parent_id.is_(None))
                .order_by(Category.sort_order.asc(), Category.id.asc())
                .all()
            )

        return [
            {
                'id': item.id,
                'name': item.name,
                'parent_id': None,
                'children': [],
            }
            for item in rows
        ]

    def create_category(self, name: str, parent_id: str = 'all') -> dict[str, Any]:
        if not name:
            raise ValueError('分类名称不能为空')
        if str(parent_id).strip().lower() not in {'', 'all', 'none', 'null'}:
            raise ValueError('当前仅支持一级分类')

        with db_session() as session:
            category = Category(
                name=name,
                type='data_package',
                parent_id=None,
                sort_order=0,
            )
            session.add(category)
            session.commit()
            session.refresh(category)
            return {
                'id': category.id,
                'name': category.name,
                'parent_id': None,
                'type': category.type,
                'sort_order': category.sort_order,
            }

    def update_category(self, category_id: str, name: str) -> dict[str, Any]:
        if category_id in {'', 'all'}:
            raise ValueError('分类ID不能为空')
        if not name:
            raise ValueError('分类名称不能为空')

        db_id = self._parse_int_id(category_id, '分类ID')
        with db_session() as session:
            category = (
                session.query(Category)
                .filter(Category.id == db_id, Category.type == 'data_package')
                .first()
            )
            if not category:
                raise ValueError('分类不存在')

            category.name = name
            session.commit()
            session.refresh(category)
            return {
                'id': category.id,
                'name': category.name,
                'parent_id': None,
                'type': category.type,
                'sort_order': category.sort_order,
            }

    def delete_category(self, category_id: str) -> None:
        if category_id in {'', 'all'}:
            raise ValueError('分类ID不能为空')

        db_id = self._parse_int_id(category_id, '分类ID')
        with db_session() as session:
            category = (
                session.query(Category)
                .filter(Category.id == db_id, Category.type == 'data_package')
                .first()
            )
            if not category:
                raise ValueError('分类不存在')

            pkg_count = (
                session.query(func.count(DataPackage.id))
                .filter(DataPackage.category_id == db_id)
                .scalar()
            )
            if pkg_count and pkg_count > 0:
                raise ValueError('该分类下仍有数据包，无法删除')

            session.delete(category)
            session.commit()

    # ── data files (for selection) ───────────────────────────────────────────

    def list_all_data_files(
        self,
        category_id: str = '',
        filename: str = '',
        remark: str = '',
    ) -> list[dict[str, Any]]:
        with db_session() as session:
            query = (
                session.query(DataFile, Category)
                .outerjoin(Category, DataFile.category_id == Category.id)
            )

            if category_id not in ('', 'all', None):
                db_id = self._parse_int_id(category_id, '分类ID')
                query = query.filter(DataFile.category_id == db_id)

            if filename:
                query = query.filter(DataFile.filename.like(f'%{filename}%'))

            if remark:
                query = query.filter(DataFile.remark.like(f'%{remark}%'))

            rows = (
                query
                .order_by(Category.name.asc(), DataFile.uploaded_at.desc(), DataFile.id.desc())
                .all()
            )
            return [
                {
                    'id': f.id,
                    'filename': f.filename,
                    'category_id': f.category_id,
                    'category_name': c.name if c else '',
                    'data_count': f.data_count,
                    'remark': f.remark or '',
                    'uploaded_at': f.uploaded_at.strftime('%Y-%m-%d %H:%M:%S') if f.uploaded_at else '',
                }
                for f, c in rows
            ]

    # ── packages ─────────────────────────────────────────────────────────────

    def list_packages(self, category_id: str = '', name: str = '') -> list[dict[str, Any]]:
        with db_session() as session:
            query = session.query(DataPackage)

            if category_id not in ('', 'all', None):
                db_id = self._parse_int_id(category_id, '分类ID')
                query = query.filter(DataPackage.category_id == db_id)

            if name:
                query = query.filter(DataPackage.name.like(f'%{name}%'))

            rows = query.order_by(DataPackage.created_at.desc(), DataPackage.id.desc()).all()
            return [self._pkg_to_dict(item) for item in rows]

    def create_package(
        self,
        name: str,
        category_id: str | None,
        remark: str | None,
        file_ids: list[int],
        clean_rules: dict | None = None,
        smooth: dict | None = None,
        row_start: int | None = None,
        row_end: int | None = None,
    ) -> dict[str, Any]:
        if not name:
            raise ValueError('数据包名称不能为空')
        if not file_ids:
            raise ValueError('请至少选择一个数据文件')
        if row_start is not None and row_end is not None and row_start > row_end:
            raise ValueError('起始行不能大于结束行')

        cat_db_id: int | None = None
        if category_id and str(category_id).strip().lower() not in {'', 'all', 'none', 'null'}:
            cat_db_id = self._parse_int_id(str(category_id), '分类ID')

        with db_session() as session:
            if cat_db_id is not None:
                cat = (
                    session.query(Category)
                    .filter(Category.id == cat_db_id, Category.type == 'data_package')
                    .first()
                )
                if not cat:
                    raise ValueError('数据包分类不存在')

            files = session.query(DataFile).filter(DataFile.id.in_(file_ids)).all()
            if len(files) != len(file_ids):
                raise ValueError('部分数据文件不存在')

            file_map = {f.id: f for f in files}

            # 始终执行完整合并，以便写入磁盘 CSV
            merged = self._merge_records(
                file_ids, file_map, limit=None,
                clean_rules=clean_rules,
                smooth=smooth,
                row_start=row_start, row_end=row_end,
            )
            total_count = merged['count']
            stored_rules: dict | None = clean_rules if (clean_rules and clean_rules.get('enabled')) else None
            if smooth and smooth.get('enabled'):
                if stored_rules is None:
                    stored_rules = {}
                stored_rules['smooth'] = smooth

            pkg = DataPackage(
                name=name,
                category_id=cat_db_id,
                remark=remark,
                data_count=total_count,
                clean_rules=stored_rules,
                row_start=row_start,
                row_end=row_end,
            )
            session.add(pkg)
            session.flush()  # 获取 pkg.id

            # 写入合并 CSV 文件
            pkg_dir = self._dm._upload_dir / 'packages'
            pkg_dir.mkdir(parents=True, exist_ok=True)
            ts = datetime.now().strftime('%Y%m%d%H%M%S%f')
            stored_name = f'pkg_{pkg.id}_{ts}.csv'
            self._write_records_csv(merged['records'], pkg_dir / stored_name)
            pkg.stored_name = stored_name
            pkg.file_path = f'/app/uploads/packages/{stored_name}'

            for idx, fid in enumerate(file_ids):
                pf = DataPackageFile(package_id=pkg.id, file_id=fid, sort_order=idx)
                session.add(pf)

            session.commit()
            session.refresh(pkg)
            return self._pkg_to_dict(pkg)

    def create_package_split(
        self,
        name: str,
        category_id: str | None,
        remark: str | None,
        file_ids: list[int],
        clean_rules: dict | None = None,
        smooth: dict | None = None,
        row_start: int | None = None,
        row_end: int | None = None,
    ) -> list[dict[str, Any]]:
        """按 70/15/15 比例将数据拆分为训练集/验证集/测试集三个数据包。"""
        if not name:
            raise ValueError('数据包名称不能为空')
        if not file_ids:
            raise ValueError('请至少选择一个数据文件')

        cat_db_id: int | None = None
        if category_id and str(category_id).strip().lower() not in {'', 'all', 'none', 'null'}:
            cat_db_id = self._parse_int_id(str(category_id), '分类ID')

        with db_session() as session:
            if cat_db_id is not None:
                cat = (
                    session.query(Category)
                    .filter(Category.id == cat_db_id, Category.type == 'data_package')
                    .first()
                )
                if not cat:
                    raise ValueError('数据包分类不存在')

            files = session.query(DataFile).filter(DataFile.id.in_(file_ids)).all()
            if len(files) != len(file_ids):
                raise ValueError('部分数据文件不存在')
            file_map = {f.id: f for f in files}

            # 先完整合并（含清洗/平滑/截取）
            merged = self._merge_records(
                file_ids, file_map, limit=None,
                clean_rules=clean_rules,
                smooth=smooth,
                row_start=row_start, row_end=row_end,
            )
            all_records = merged['records']
            total = len(all_records)

            # 按 70/15/15 计算切分点
            train_end = round(total * 0.70)
            val_end = train_end + round(total * 0.15)
            slices = [
                (f'{name}-训练集', all_records[:train_end]),
                (f'{name}-验证集', all_records[train_end:val_end]),
                (f'{name}-测试集', all_records[val_end:]),
            ]

            stored_rules: dict | None = clean_rules if (clean_rules and clean_rules.get('enabled')) else None
            if smooth and smooth.get('enabled'):
                if stored_rules is None:
                    stored_rules = {}
                stored_rules['smooth'] = smooth

            pkg_dir = self._dm._upload_dir / 'packages'
            pkg_dir.mkdir(parents=True, exist_ok=True)
            results: list[dict[str, Any]] = []

            for split_name, records in slices:
                pkg = DataPackage(
                    name=split_name,
                    category_id=cat_db_id,
                    remark=remark,
                    data_count=len(records),
                    clean_rules=stored_rules,
                    row_start=row_start,
                    row_end=row_end,
                )
                session.add(pkg)
                session.flush()

                ts = datetime.now().strftime('%Y%m%d%H%M%S%f')
                stored_name = f'pkg_{pkg.id}_{ts}.csv'
                self._write_records_csv(records, pkg_dir / stored_name)
                pkg.stored_name = stored_name
                pkg.file_path = f'/app/uploads/packages/{stored_name}'

                for idx, fid in enumerate(file_ids):
                    pf = DataPackageFile(package_id=pkg.id, file_id=fid, sort_order=idx)
                    session.add(pf)

                results.append(self._pkg_to_dict(pkg))

            session.commit()

        return results

    def update_package(
        self,
        package_id: str,
        name: str,
        category_id: str | None,
        remark: str | None,
    ) -> dict[str, Any]:
        db_id = self._parse_int_id(package_id, '数据包ID')

        cat_db_id: int | None = None
        if category_id and str(category_id).strip().lower() not in {'', 'all', 'none', 'null'}:
            cat_db_id = self._parse_int_id(str(category_id), '分类ID')

        with db_session() as session:
            pkg = session.query(DataPackage).filter(DataPackage.id == db_id).first()
            if not pkg:
                raise ValueError('数据包不存在')

            if cat_db_id is not None:
                cat = (
                    session.query(Category)
                    .filter(Category.id == cat_db_id, Category.type == 'data_package')
                    .first()
                )
                if not cat:
                    raise ValueError('数据包分类不存在')

            pkg.name = name
            pkg.category_id = cat_db_id
            pkg.remark = remark
            session.commit()
            session.refresh(pkg)
            return self._pkg_to_dict(pkg)

    def update_package(
        self,
        package_id: str,
        name: str,
        category_id: str | None,
        remark: str | None,
    ) -> dict[str, Any]:
        db_id = self._parse_int_id(package_id, '数据包ID')

        cat_db_id: int | None = None
        if category_id and str(category_id).strip().lower() not in {'', 'all', 'none', 'null'}:
            cat_db_id = self._parse_int_id(str(category_id), '分类ID')

        with db_session() as session:
            pkg = session.query(DataPackage).filter(DataPackage.id == db_id).first()
            if not pkg:
                raise ValueError('数据包不存在')

            if cat_db_id is not None:
                cat = (
                    session.query(Category)
                    .filter(Category.id == cat_db_id, Category.type == 'data_package')
                    .first()
                )
                if not cat:
                    raise ValueError('数据包分类不存在')

            pkg.name = name
            pkg.category_id = cat_db_id
            pkg.remark = remark
            session.commit()
            session.refresh(pkg)
            return self._pkg_to_dict(pkg)

    def delete_package(self, package_id: str) -> None:
        db_id = self._parse_int_id(package_id, '数据包ID')
        with db_session() as session:
            pkg = session.query(DataPackage).filter(DataPackage.id == db_id).first()
            if not pkg:
                raise ValueError('数据包不存在')
            stored_name = pkg.stored_name
            session.query(DataPackageFile).filter(DataPackageFile.package_id == db_id).delete()
            session.delete(pkg)
            session.commit()
        # 删除磁盘上的合并 CSV 文件
        if stored_name:
            pkg_file = self._dm._upload_dir / 'packages' / stored_name
            if pkg_file.exists():
                pkg_file.unlink()

    def get_package_records(self, package_id: str, limit: int | None = None) -> dict[str, Any]:
        db_id = self._parse_int_id(package_id, '数据包ID')

        with db_session() as session:
            pkg = session.query(DataPackage).filter(DataPackage.id == db_id).first()
            if not pkg:
                raise ValueError('数据包不存在')

            stored_clean_rules = pkg.clean_rules
            stored_row_start = pkg.row_start
            stored_row_end = pkg.row_end

            pf_rows = (
                session.query(DataPackageFile)
                .filter(DataPackageFile.package_id == db_id)
                .order_by(DataPackageFile.sort_order.asc())
                .all()
            )
            file_ids = [pf.file_id for pf in pf_rows]
            files = session.query(DataFile).filter(DataFile.id.in_(file_ids)).all()
            file_map = {f.id: f for f in files}

        return self._merge_records(
            file_ids, file_map, limit,
            clean_rules=stored_clean_rules,
            smooth=stored_clean_rules.get('smooth') if stored_clean_rules else None,
            row_start=stored_row_start, row_end=stored_row_end,
        )

    def preview_records(
        self,
        file_ids: list[int],
        limit: int | None = None,
        clean_rules: dict | None = None,
        smooth: dict | None = None,
        row_start: int | None = None,
        row_end: int | None = None,
    ) -> dict[str, Any]:
        if not file_ids:
            return {'records': [], 'count': 0}

        with db_session() as session:
            files = session.query(DataFile).filter(DataFile.id.in_(file_ids)).all()
            file_map = {f.id: f for f in files}

        return self._merge_records(
            file_ids, file_map, limit,
            clean_rules=clean_rules,
            smooth=smooth,
            row_start=row_start, row_end=row_end,
        )

    # ── helpers ──────────────────────────────────────────────────────────────

    def _merge_records(
        self,
        file_ids: list[int],
        file_map: dict[int, Any],
        limit: int | None,
        clean_rules: dict | None = None,
        smooth: dict | None = None,
        row_start: int | None = None,
        row_end: int | None = None,
    ) -> dict[str, Any]:
        use_filter = bool(clean_rules and clean_rules.get('enabled'))
        use_smooth = bool(smooth and smooth.get('enabled'))
        use_range = row_start is not None or row_end is not None
        need_full_read = use_filter or use_range or use_smooth
        all_records: list[dict[str, Any]] = []
        total_count = 0
        remaining = limit

        for fid in file_ids:
            if fid not in file_map:
                continue
            file_meta = file_map[fid]
            path = self._dm._resolve_local_file_path(file_meta.file_path, file_meta.stored_name)
            if need_full_read:
                records, count = self._dm._read_records(path, limit=None)
            else:
                read_limit = (remaining if remaining is not None and remaining > 0 else 0) if remaining is not None else None
                records, count = self._dm._read_records(path, limit=read_limit)
            total_count += count
            all_records.extend(records)
            if not need_full_read and remaining is not None:
                remaining -= len(records)
                if remaining <= 0:
                    break

        if use_filter:
            all_records = self._apply_clean_rules(all_records, clean_rules)

        if use_range:
            # row_start/row_end are 1-based and inclusive
            start_idx = (row_start - 1) if row_start and row_start >= 1 else 0
            end_idx = row_end if row_end else len(all_records)
            all_records = all_records[start_idx:end_idx]

        if use_smooth:
            all_records = self._apply_smooth(all_records, smooth.get('window', 5))

        total_count = len(all_records)

        if limit is not None and limit > 0:
            return {'records': all_records[:limit], 'count': total_count}
        return {'records': all_records, 'count': total_count}

    @staticmethod
    def _apply_clean_rules(records: list[dict[str, Any]], clean_rules: dict) -> list[dict[str, Any]]:
        c_min = clean_rules.get('current_min')
        c_max = clean_rules.get('current_max')
        v_min = clean_rules.get('voltage_min')
        v_max = clean_rules.get('voltage_max')
        t_min = clean_rules.get('temperature_min')
        t_max = clean_rules.get('temperature_max')
        newton_interp = bool(clean_rules.get('newton_interp', False))

        numeric_cols = ['current', 'voltage', 'set_temperature', 'actual_temperature']

        def _is_out_cur(v):
            if v is None:
                return False
            return (c_min is not None and v < c_min) or (c_max is not None and v > c_max)

        def _is_out_vol(v):
            if v is None:
                return False
            return (v_min is not None and v < v_min) or (v_max is not None and v > v_max)

        def _is_out_tmp(v):
            if v is None:
                return False
            return (t_min is not None and v < t_min) or (t_max is not None and v > t_max)

        result: list[dict[str, Any]] = []
        for r in records:
            cur_out = _is_out_cur(r.get('current'))
            vol_out = _is_out_vol(r.get('voltage'))
            tmp_out = _is_out_tmp(r.get('actual_temperature'))

            if not (cur_out or vol_out or tmp_out):
                result.append(r)
                continue

            if not newton_interp:
                # 直接剔除行
                continue

            # 牛顿插值：保留行但将野值字段置 None，后续插值填补
            row = dict(r)
            if cur_out:
                row['current'] = None
            if vol_out:
                row['voltage'] = None
            if tmp_out:
                row['actual_temperature'] = None
            result.append(row)

        if newton_interp:
            # 对各数值列独立做牛顿插值
            for col in numeric_cols:
                col_vals = [row.get(col) for row in result]
                filled = PackageManagementService._interpolate_column(col_vals)
                for i, row in enumerate(result):
                    row[col] = filled[i]

        return result

    @staticmethod
    def _interpolate_column(col: list) -> list:
        """用牛顿差商插值填补列中的 None 值（bisect 二分查找邻居，O(n log n)）。"""
        valid_pts = [(i, v) for i, v in enumerate(col) if v is not None]
        if not valid_pts:
            return col
        valid_indices = [p[0] for p in valid_pts]  # 有序索引列表，供二分查找
        result = list(col)
        for i, v in enumerate(col):
            if v is not None:
                continue
            # 二分定位：pos 是 i 在 valid_indices 中的插入位置
            pos = bisect.bisect_left(valid_indices, i)
            before = valid_pts[max(0, pos - 3):pos]
            after = valid_pts[pos:pos + 3]
            points = before + after
            if not points:
                continue
            if len(points) == 1:
                result[i] = points[0][1]
                continue
            xs = [float(p[0]) for p in points]
            ys = [float(p[1]) for p in points]
            result[i] = PackageManagementService._newton_eval(xs, ys, float(i))
        return result

    @staticmethod
    def _newton_eval(xs: list[float], ys: list[float], x: float) -> float:
        """牛顿差商表求插值多项式在 x 处的值。"""
        n = len(xs)
        dd = list(ys)
        coeffs = [dd[0]]
        for j in range(1, n):
            dd = [
                (dd[k + 1] - dd[k]) / (xs[k + j] - xs[k])
                for k in range(n - j)
            ]
            coeffs.append(dd[0])
        # 霍纳法求值
        val = coeffs[-1]
        for j in range(n - 2, -1, -1):
            val = val * (x - xs[j]) + coeffs[j]
        return round(val, 6)

    @staticmethod
    def _apply_smooth(records: list[dict[str, Any]], window: int) -> list[dict[str, Any]]:
        """滑动均值平滑（居中窗口），对数值列原地均值化。"""
        if window < 2 or not records:
            return records
        numeric_cols = ['current', 'voltage', 'set_temperature', 'actual_temperature']
        n = len(records)
        result = [dict(r) for r in records]
        half = window // 2
        for col in numeric_cols:
            vals = [r.get(col) for r in records]
            for i in range(n):
                start = max(0, i - half)
                end = min(n, start + window)
                if end - start < window:
                    start = max(0, end - window)
                window_vals = [v for v in vals[start:end] if v is not None]
                if window_vals:
                    result[i][col] = round(sum(window_vals) / len(window_vals), 6)
        return result

    def _pkg_to_dict(self, pkg: DataPackage) -> dict[str, Any]:
        return {
            'id': pkg.id,
            'name': pkg.name,
            'category_id': pkg.category_id,
            'remark': pkg.remark,
            'data_count': pkg.data_count,
            'row_start': pkg.row_start,
            'row_end': pkg.row_end,
            'stored_name': pkg.stored_name,
            'file_path': pkg.file_path,
            'created_at': pkg.created_at.strftime('%Y-%m-%d %H:%M:%S') if pkg.created_at else '',
            'updated_at': pkg.updated_at.strftime('%Y-%m-%d %H:%M:%S') if pkg.updated_at else '',
        }

    @staticmethod
    def _write_records_csv(records: list[dict[str, Any]], dest: Path) -> None:
        """将合并后的记录写入 CSV 文件（UTF-8 BOM，带英文表头）。"""
        with dest.open('w', newline='', encoding='utf-8-sig') as f:
            writer = csv.writer(f)
            writer.writerow(['time', 'current', 'voltage', 'set_temperature', 'actual_temperature'])
            for r in records:
                writer.writerow([
                    r.get('time', ''),
                    r.get('current', ''),
                    r.get('voltage', ''),
                    r.get('set_temperature', ''),
                    r.get('actual_temperature', ''),
                ])

    @staticmethod
    def _parse_int_id(value: str, label: str = 'ID') -> int:
        try:
            return int(value)
        except (ValueError, TypeError) as exc:
            raise ValueError(f'{label}格式错误') from exc
