certflow.controllers.import_controller module

导入控制器

桥接导入界面（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)

class certflow.controllers.import_controller.ImportController(db_session=None, db_manager=None, use_shared=True)

基类：BaseController

导入控制器 — 导入界面的业务调度层（薄层）

参数:

use_shared (bool)

run_import(config_params)

执行销售计划导入，返回统计 dict。

仅做「请求透传 + 调 Service + 收结果」，事务由 Service 内部持有 （与历史 run_import 行为一致，Controller 不额外 commit/rollback）。

参数:

config_params (dict)

返回类型:

dict

static apply_range_filter(records, range_filter, selected_rows=None)

导入范围筛选（预览用，不落库）。

委托 ``SalePlanService.apply_range_filter``（M5 已提升为公开静态方法）。

参数:

• records (list[dict[str, Any]])

• range_filter (dict[str, str] | None)

• selected_rows (list[int] | None)

返回类型:

list[dict[str, Any]]

static read_and_clean_sheet(file_path, sheet_name, header_row, skip_rows, column_mapping=None)

读取工作表并经清洗管线转为记录列表（与真实导入同款读取）。

委托 ExcelHandler.read_excel + DataCleaner.clean_sale_plan； pandas 在 header= 下自动对重名列加 .1/.2 后缀去重，避免 df[col] 因重名列返回 DataFrame。

参数:

• file_path (str)

• sheet_name (str)

• header_row (int)

• skip_rows (int)

• column_mapping (dict[str, str] | None)

返回类型:

list[dict[str, Any]]

static detect_header_row(df_raw, keywords, max_rows=20)

自动检测表头行（委托 SalePlanService.detect_header_row）。

参数:

• keywords (list[str])

• max_rows (int)

返回类型:

int | None

static extract_year_from_filename(file_path)

从工作簿文件名提取年份后两位（委托 SalePlanService.extract_year_from_filename）。

参数:

file_path (str)

返回类型:

int | None

static normalize_record_for_key(rec, session)

预览/扫描阶段与入库同款清洗：口径/压力规范化后返回新 dict（不污染入参）。

委托 IDGenerator.normalize_spec / IDGenerator.normalize_pn。

参数:

• rec (dict[str, Any])

• session (Any)

返回类型:

dict[str, Any]

static generate_unique_key(rec)

生成记录唯一键（委托 IDGenerator.generate_unique_key）。

参数:

rec (dict[str, Any])

返回类型:

str

static generate_change_signature(rec)

生成记录变更签名（委托 IDGenerator.generate_change_signature）。

参数:

rec (dict[str, Any])

返回类型:

str

static find_existing_by_unique_key(session, unique_key)

按唯一键查询已存在记录（委托 SalePlanService.find_by_unique_key）。

参数:

• session (Any)

• unique_key (str)

返回类型:

Any