certflow.services.import_gate_service module

#30 P1 导入门控:yaml 驱动跳过/隔离无单号行

替代 VBA「要货单号='假'/空 → 整行跳过」逻辑,全配置驱动。 门控在读表得到 records 之后、进入 save_handler 之前过滤,不改动源文件

配置(config.yaml sales_plan.import_gate): - enabled: 总开关 - order_identity_fields: 现代要货单号来源字段(2025+ 行凭此逃逸门控) - legacy_field: legacy 要货单号来源字段(pre-2025 行凭 plan_no 逃逸门控) - mode:

  • skip = 直接丢弃(不导入、不隔离)

  • isolate = 隔离到待复核(返回 gated 列表,由调用方决定落库/展示)

  • warn = 仍导入但给该行打 flag=True(黄标,等价于非阻塞标黄)

  • match_keys: 门控匹配维度(隔离时与既有行归并的键,预留)

门控语义(业务正确,替代 VBA「要货单号='假'/空→整行跳过」): 仅当「全部订单身份字段(order_identity_fields + legacy_field)均缺失」时, 该行才被判定为「无单号」无效行并触发门控;任一身份字段非空即逃逸。 —— 因此 pre-2025 行(凭 plan_no/计划单号4位,VBA 即依此派生要货单号)不会被误门控;

post-2025 行(凭 sales_order_no/production_order_no)也不会; 仅真正全缺的行才被门控。

(single-SO=batch 等「有单号但清单不全」的无效情形不自动检测,由人工在销售计划副本补录。)

class certflow.services.import_gate_service.ImportGateService[源代码]

基类:object

导入门控:按 config.yaml sales_plan.import_gate 过滤无单号行。

classmethod is_gated(record, gate=None)[源代码]

该记录是否触发门控(「全部订单身份字段均缺失」才判定为无单号无效行)。

  • order_identity_fields(现代:sales_order_no/production_order_no)

  • legacy_field(legacy:plan_no/计划单号4位,VBA 依此派生要货单号)

任一非空即逃逸门控;全缺才门控。

参数:
返回类型:

bool

classmethod apply(records, gate=None)[源代码]

对记录列表应用门控。

返回:

{

"kept": 参与后续导入的记录(list[dict]), "gated": 触发门控的记录(用于隔离/统计), "mode": 实际生效的 mode(未启用则为 None), "applied": 是否实际执行了门控(enabled 开关),

}

参数:
返回类型:

dict[str, Any]

classmethod persist_isolated(gated, meta=None, session=None)[源代码]

将 isolate 模式门控的行持久化到隔离表(回收站)。

isolate 模式语义:被门控的无效行不进 sale_plans,而是隔离到 sale_plans_quarantine,供「校正队列」人工复核 / 释放 / 补录字典自学习。

参数:
  • gated (list[dict[str, Any]]) -- ImportGateService.apply 返回的 gated 记录列表

  • meta (dict[str, Any] | None) -- 导入上下文(source_file/source_sheet/import_batch_id/reason)

  • session (Session | None) -- SQLAlchemy 会话(隔离行写入此会话并提交)

返回:

实际写入隔离表的行数

返回类型:

int