certflow.handlers.save_handler module

数据库保存处理器模块

提供销售计划记录的数据转换和变更检测功能， ORM 操作通过回调委托给 Service 层。

certflow.handlers.save_handler.resolve_dual_fields(record, import_mode, pairs)

按 import_mode 计算 cert_* 合格证字段值（蓝图 §4.2.1）。

参数:

• record (dict[str, Any]) -- 已映射字段的导入记录（可能含 cert_* 键，仅当 Excel 提供对应列）

• import_mode (str) -- copy / override / lazy

• pairs (list[dict[str, str]]) -- 双字段映射对列表，每项 {contract, cert}

返回:

{cert_field: value}，仅含 pairs 中声明的 cert 字段

返回类型:

dict

规则：

• copy：cert_* = 合同值（杜绝"改合同静默跟随"风险）

• override：仅当源行含 cert 列（Excel 提供）时写 cert_*，否则留空回退

• ``lazy``（默认）：cert_* 留空，打印时回退合同值（保持现状）

class certflow.handlers.save_handler.SaveHandlerOrmDelegate(*args, **kwargs)

基类：Protocol

SaveHandler 的 ORM 委托协议

Service 层实现此协议，Handler 通过此接口间接操作数据库。

find_by_unique_key(unique_key)

按唯一键查找销售计划记录。

参数:

unique_key (str) -- 记录唯一键（由 IDGenerator 生成）。

返回:

匹配的销售计划记录；未找到时返回 None。

返回类型:

SalePlan | None

find_by_production_order_no(production_order_no)

按生产令号查找销售计划记录（老数据降级匹配）。

用于缺少唯一键的老数据模式，与生产令号精确匹配。

参数:

production_order_no (str) -- 生产令号。

返回:

匹配的销售计划记录；未找到时返回 None。

返回类型:

SalePlan | None

find_by_contract_and_model(contract_no, product_model)

按合同号 + 产品型号兜底匹配（跨来源补充场景）。

当 unique_key 与生产令号均不匹配时，用合同号与型号跨来源 补充缺失字段。

参数:

• contract_no (str) -- 合同号。

• product_model (str) -- 产品型号。

返回:

匹配的销售计划记录；未找到时返回 None。

返回类型:

SalePlan | None

add_sale_plan(sale_plan)

向会话新增一条销售计划记录（不自动提交）。

参数:

sale_plan (SalePlan) -- 待新增的 SalePlan 实例。

返回类型:

None

add_change(change)

向会话新增一条变更记录（不自动提交）。

参数:

change (SalePlanChange) -- 待新增的 SalePlanChange 实例。

返回类型:

None

flush_session()

刷新会话，将挂起的对象写入数据库（不提交事务）。

返回类型:

None

commit_session()

提交事务，持久化会话中所有挂起的变更。

返回类型:

None

class certflow.handlers.save_handler.SaveHandler(session, orm_delegate=None)

基类：object

数据库保存处理器

负责数据转换和变更检测，ORM 操作通过 orm_delegate 回调委托给 Service 层。

示例

>>> handler = SaveHandler(session, orm_delegate=service)
>>> imported, skipped = handler.save_styled_records(
...     sorted_records, file_path, sheet_name, batch_id, "2601", 2601
... )

参数:

• session (Session)

• orm_delegate (SaveHandlerOrmDelegate | None)

save_styled_records(sorted_records, file_path, sheet_name, import_batch_id, year_month_prefix, year_month_int, conflict_resolutions=None, collision_policy=None)

保存带格式的分组记录到数据库

参数:

• sorted_records (list[dict[str, Any]]) -- 已分组排序的记录列表

• file_path (str) -- 源文件路径

• sheet_name (str) -- 工作表名称

• import_batch_id (str) -- 导入批次ID

• year_month_prefix (str) -- 年月前缀字符串，如 "2601"

• year_month_int (int) -- 年月前缀整数，如 2601

• conflict_resolutions (dict[str, str] | None) -- db_conflict 决议（unique_key -> 动作）

• collision_policy (str | None) -- 批次内撞键处理策略（UI 选择，覆盖 ON_COLLISION 配置）： isolate（隔离，默认）/ first（取首行）/ discard（丢弃全部重复）

返回:

{'new_count': int, 'duplicate_count': int, 'changes_count': int, ...}

• new_count: 新增记录数

• duplicate_count: 跳过（已存在且内容无变更）记录数； 注意「变更」行不计入此处，避免与 changes_count 重叠

• changes_count: 已存在但内容发生字段变更的记录数

• change_categories / changed_fields: 变更按字段归类统计

返回类型:

dict

save_records_with_stats(records, source_file, source_sheet, import_batch_id, year_month_prefix, year_month_int, conflict_resolutions=None, collision_policy=None)

保存记录到数据库，并返回统计信息（不带格式）

参数:

• records (list[dict[str, Any]]) -- 待保存的记录字典列表

• source_file (str) -- 源文件路径

• source_sheet (str) -- 源工作表名称

• import_batch_id (str) -- 导入批次ID

• year_month_prefix (str) -- 年月前缀字符串

• year_month_int (int) -- 年月前缀整数

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

• collision_policy (str | None)

返回:

保存统计信息

返回类型:

Dict[str, Any]

CHANGE_CATEGORIES: dict[str, set[str]] = {'产品信息': {'product_model', 'product_name', 'product_spec'}, '供货方式': {'supply_type'}, '发货日期': {'contract_delivery_date', 'plan_date', 'planned_delivery_date'}, '客户项目': {'customer', 'project_name'}, '数量': {'quantity'}}