certflow.services.import_gate_service 源代码

# certflow/services/import_gate_service.py
"""#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 等「有单号但清单不全」的无效情形不自动检测，由人工在销售计划副本补录。）
"""

from __future__ import annotations

from typing import Any

from sqlalchemy.orm import Session

from certflow.config.settings import IMPORT_GATE




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

    # VBA 中「要货单号='假'」表示无效单号，等同缺失
    _INVALID_SENTINELS = {"假", "false", "none", "null", "na", "n/a"}

    @staticmethod
    def _is_missing(record: dict[str, Any], field: str) -> bool:
        """该字段在记录中是否视为缺失（空 / 空串 / VBA 无效哨兵）。"""
        v = record.get(field)
        if v is None:
            return True
        if isinstance(v, str):
            s = v.strip()
            if s == "":
                return True
            if s.lower() in ImportGateService._INVALID_SENTINELS:
                return True
        return False



    @classmethod
    def is_gated(cls, record: dict[str, Any], gate: dict[str, Any] | None = None) -> bool:
        """该记录是否触发门控（「全部订单身份字段均缺失」才判定为无单号无效行）。

        - order_identity_fields（现代：sales_order_no/production_order_no）
        - legacy_field（legacy：plan_no/计划单号4位，VBA 依此派生要货单号）
        任一非空即逃逸门控；全缺才门控。
        """
        gate = gate if gate is not None else IMPORT_GATE
        identity_fields = list(gate.get("order_identity_fields", []) or [])
        # 兼容旧配置名
        if not identity_fields:
            identity_fields = list(gate.get("skip_when_missing", []) or [])
        legacy = gate.get("legacy_field")
        if legacy:
            identity_fields = identity_fields + [legacy]
        if not identity_fields:
            return False
        # 全缺才门控（任一存在即视为有订单身份）
        return all(cls._is_missing(record, f) for f in identity_fields)




    @classmethod
    def apply(
        cls,
        records: list[dict[str, Any]],
        gate: dict[str, Any] | None = None,
    ) -> dict[str, Any]:
        """对记录列表应用门控。

        Returns:
            {
                "kept":   参与后续导入的记录（list[dict]），
                "gated":  触发门控的记录（用于隔离/统计），
                "mode":   实际生效的 mode（未启用则为 None），
                "applied": 是否实际执行了门控（enabled 开关），
            }
        """
        gate = gate if gate is not None else IMPORT_GATE
        if not gate.get("enabled", False):
            return {"kept": list(records), "gated": [], "mode": None, "applied": False}

        mode = (gate.get("mode") or "isolate").lower()
        gated = [r for r in records if cls.is_gated(r, gate)]
        kept = [r for r in records if not cls.is_gated(r, gate)]

        if mode == "warn":
            # 仍导入，但给触发行打黄标（非阻塞，等价于 VBA 标黄继续）
            for r in gated:
                r["flag"] = True
            kept = list(records)

        return {"kept": kept, "gated": gated, "mode": mode, "applied": True}




    @classmethod
    def persist_isolated(
        cls,
        gated: list[dict[str, Any]],
        meta: dict[str, Any] | None = None,
        session: Session | None = None,
    ) -> int:
        """将 isolate 模式门控的行持久化到隔离表（回收站）。

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

        Args:
            gated: ImportGateService.apply 返回的 gated 记录列表
            meta: 导入上下文（source_file/source_sheet/import_batch_id/reason）
            session: SQLAlchemy 会话（隔离行写入此会话并提交）

        Returns:
            int: 实际写入隔离表的行数
        """
        if not gated or session is None:
            return 0
        from certflow.models.quarantine_sale_plan import QuarantineSalePlan

        count = 0
        for rec in gated:
            session.add(QuarantineSalePlan.from_record(rec, meta))
            count += 1
        if count:
            session.commit()
        return count