certflow.controllers.import_controller 源代码

# certflow/controllers/import_controller.py
"""导入控制器

桥接导入界面（View）与导入服务（Service）：
- 写：执行销售计划导入（经 ``SalePlanService.import_from_excel_with_config``）；
- 读（预览）：导入范围筛选（经 ``SalePlanService.apply_range_filter``）。

架构层次（蓝图 M5「架构统一」目标 —— 补齐导入链路缺失的 Controller 层）：
    View (ImportDialog / ImportConfirmDialog)
      → Controller (ImportController)
      → Service (SalePlanService)
      → Handler (ExcelHandler / DataCleaner / Sorter / SaveHandler)
      → Model (SalePlan / QuarantineSalePlan)
"""

from __future__ import annotations

from pathlib import Path
from typing import Any

from certflow.controllers.base_controller import BaseController
from certflow.services.sale_plan_service import SalePlanService
from certflow.utils.id_generator import IDGenerator
from certflow.utils.logger import logger




class ImportController(BaseController):
    """导入控制器 — 导入界面的业务调度层（薄层）"""

    def __init__(self, db_session=None, db_manager=None, use_shared: bool = True):
        super().__init__(db_session=db_session, db_manager=db_manager, use_shared=use_shared)
        self._service = self.get_service(SalePlanService)

    # ============================================================
    # 写操作：执行导入
    # ============================================================



    def run_import(self, config_params: dict) -> dict:
        """执行销售计划导入，返回统计 dict。

        仅做「请求透传 + 调 Service + 收结果」，事务由 Service 内部持有
        （与历史 ``run_import`` 行为一致，Controller 不额外 commit/rollback）。
        """
        config = dict(config_params)
        resolutions = config_params.get("conflict_resolutions")
        if resolutions:
            config["conflict_resolutions"] = resolutions
        try:
            return self._service.import_from_excel_with_config(config)
        except Exception as e:
            logger.error(f"导入销售计划失败: {e}")
            raise


    # ============================================================
    # 读操作（预览）：导入范围筛选
    # ============================================================



    @staticmethod
    def apply_range_filter(
        records: list[dict[str, Any]],
        range_filter: dict[str, str] | None,
        selected_rows: list[int] | None = None,
    ) -> list[dict[str, Any]]:
        """导入范围筛选（预览用，不落库）。

        委托 ``SalePlanService.apply_range_filter``（M5 已提升为公开静态方法）。
        """
        return SalePlanService.apply_range_filter(records, range_filter, selected_rows)


    # ============================================================
    # 静态门面：收口 View 对 Handler / Utils / Model 的散落直连
    # （即使 ImportController 未注入实例，View 仍可经类名调用，向后兼容）
    # ============================================================



    @staticmethod
    def read_and_clean_sheet(
        file_path: str,
        sheet_name: str,
        header_row: int,
        skip_rows: int,
        column_mapping: dict[str, str] | None = None,
    ) -> list[dict[str, Any]]:
        """读取工作表并经清洗管线转为记录列表（与真实导入同款读取）。

        委托 ``ExcelHandler.read_excel`` + ``DataCleaner.clean_sale_plan``；
        pandas 在 ``header=`` 下自动对重名列加 ``.1``/``.2`` 后缀去重，避免
        ``df[col]`` 因重名列返回 DataFrame。
        """
        df = SalePlanService.read_excel(
            file_path=Path(file_path),
            sheet_name=sheet_name,
            header_row=header_row,
            skiprows=skip_rows,
        )
        if column_mapping:
            existing = [c for c in column_mapping if c in df.columns]
            df = df[existing].rename(columns=column_mapping)
        return SalePlanService.clean_sale_plan(df).to_dict("records")




    @staticmethod
    def detect_header_row(df_raw, keywords: list[str], max_rows: int = 20) -> int | None:
        """自动检测表头行（委托 ``SalePlanService.detect_header_row``）。"""
        return SalePlanService.detect_header_row(df_raw, keywords, max_rows=max_rows)




    @staticmethod
    def extract_year_from_filename(file_path: str) -> int | None:
        """从工作簿文件名提取年份后两位（委托 ``SalePlanService.extract_year_from_filename``）。"""
        return SalePlanService.extract_year_from_filename(file_path)




    @staticmethod
    def normalize_record_for_key(rec: dict[str, Any], session: Any) -> dict[str, Any]:
        """预览/扫描阶段与入库同款清洗：口径/压力规范化后返回新 dict（不污染入参）。

        委托 ``IDGenerator.normalize_spec`` / ``IDGenerator.normalize_pn``。
        """
        rec = dict(rec)
        rec = IDGenerator.normalize_spec(rec, session)
        return IDGenerator.normalize_pn(rec, session)




    @staticmethod
    def generate_unique_key(rec: dict[str, Any]) -> str:
        """生成记录唯一键（委托 ``IDGenerator.generate_unique_key``）。"""
        return IDGenerator.generate_unique_key(rec)




    @staticmethod
    def generate_change_signature(rec: dict[str, Any]) -> str:
        """生成记录变更签名（委托 ``IDGenerator.generate_change_signature``）。"""
        return IDGenerator.generate_change_signature(rec)




    @staticmethod
    def find_existing_by_unique_key(session: Any, unique_key: str) -> Any:
        """按唯一键查询已存在记录（委托 ``SalePlanService.find_by_unique_key``）。"""
        return SalePlanService(session).find_by_unique_key(unique_key)