certflow.services.output_guard 源代码

# src/certflow/services/output_guard.py
"""统一输出护栏管道（print/scan 共用，配置驱动，与 UI 解耦）。

设计原则（docs/PRINT_BLUEPRINT.md §11.5 / §11.8 阶段1）：
- 本模块为**纯逻辑层**，不依赖 PySide6 / 任何 UI 框架，便于 pytest 无头测试。
- 所有「用户交互」（弹输入框补录、确认闸口）通过 :class:`GuardPrompt` 协议注入，
  view 层负责提供 PyQt 实现（如 ``print_view._QtGuardPrompt``），scan 可注入非交互实现。
- 历史差异比对（④ reconfirm_on_diff）依赖调用方传入 ``history_last``（由
  :class:`~certflow.services.print_history_service.HistoryBackfillService` 提供，
  读 ``PrintLog`` 聚合的「同型号上次实际值」）；历史源缺失时该字段自动跳过差异闸。

管道顺序（与旧 ``print_view._guard_before_print`` 等价并补 ④）：
  ② 必填补录 / ③ reconfirm（即使有值也确认）
  ④ 与历史差异确认（reconfirm_on_diff）
  ⑤ 确认闸口（confirmation.summary_fields 预览）
任一环节用户取消 → 返回 ``(False, data)``，调用方应中止本批（及多计划剩余批）。
"""

from __future__ import annotations

import re
from typing import Any, Protocol

from certflow.services.print_history_service import backfill as _backfill_history

# ============================================================
# 字段中文标签（合并 print + scan 需求，避免各 view 重复定义元组）
# ============================================================
FIELD_LABELS: dict[str, str] = {
    "product_name": "产品名称",
    "product_model": "产品型号",
    "product_spec": "规格",
    "test_standard": "检验标准",
    "check_standard": "检验标准",
    "inspector_id": "检验工号",
    "issue_date": "出厂日期",
    "manufacture_date": "出厂日期",
    "dn": "公称通径",
    "pn": "公称压力",
    "temperature": "工作温度",
    "medium": "适用介质",
    "serial_number": "产品编号",
    "pn_value": "公称压力值",
    "pn_unit": "压力单位",
}




def field_label(key: str) -> str:
    """字段 key → 中文标签（取不到回退 key 本身）。"""
    return FIELD_LABELS.get(key, key)



# ============================================================
# 交互协议（由 view 层注入 PyQt / 非交互实现）
# ============================================================


class GuardPrompt(Protocol):
    """护栏与用户交互的抽象接口。

    - ``prompt_field``：请求用户补全/确认某字段，返回字符串或 ``None``（用户取消）。
    - ``confirm``：确认闸口，返回是否继续。
    """



    def prompt_field(

        self, *, key: str, label: str, current: str, suggestion: str
    ) -> str | None: ...


    def confirm(self, *, message: str) -> bool: ...




# ============================================================
# 纯函数：完整性收集 + 差异计算（可独立测试）
# ============================================================


def collect_incomplete(data: dict[str, Any], fields: list[str]) -> list[str]:
    """非交互完整性闸门：返回 ``fields`` 中值为空的字段 key 列表。

    供 scan 视图在生成前检查合格证必填信息是否齐全（缺失即中止，不弹输入框）。
    """
    return [k for k in fields if not (data.get(k) or "").strip()]



# ============================================================
# 模板语言族 → 压力单位门限（中文底版预印单位，其余族需带单位）
# ============================================================


def family_requires_pn_unit(family: str) -> bool:
    """中文合格证底版已预印压力单位 → 不要求填单位；中英文/全英文/俄英文必须带单位。

    Args:
        family: 语言族（``全中文``/``中英文``/``全英文``/``俄英文``，来自 ``family_of``）。
    """
    return family != "全中文"





def scan_required_fields_for(family: str, base: list[str]) -> list[str]:
    """按模板语言族计算扫描件必填字段（配置 base + 压力字段门限）。

    - 公称压力数值（``pn_value``）任何族都必填；
    - 中文族：底版预印单位，不要求 ``pn_unit``；其余族必须带 ``pn_unit``。
    """
    req = list(base)
    if "pn_value" not in req:
        req.append("pn_value")
    if family_requires_pn_unit(family):
        if "pn_unit" not in req:
            req.append("pn_unit")
    else:
        req = [k for k in req if k != "pn_unit"]
    return req





def validate_pn_text(family: str, raw: str) -> str:
    """校验压力文本是否符合模板族的单位要求，返回错误提示（空串表示通过）。

    ``pn_display`` 为权威输入串（任意格式），本函数只做「门限」把关，不做改写：

    - 中文族：底板预印单位，**放行不设置门限**。用户输入数值（如 ``25.0``）保存时
      默认补 ``MPa``；若输入带单位（``32.0MPa`` 或英制 ``1500Lb``），渲染层按族处理
      （MPa 剥离、非 MPa 的 Lb 保留，见 ``format_pn``），本函数不拦截；
    - 其余族（中英文/全英文/俄英文）：**必须带单位**（如 ``2.5MPa`` / ``150Lb``），
      缺单位返回错误提示，阻断保存/入队以强制补齐（避免打印出无单位的压力文本）。
    """
    raw = (raw or "").strip()
    if not raw:
        return ""
    if family == "全中文":
        return ""
    if not re.search(r"(?i)(mpa|lb|kpa)", raw):
        return "中英文/全英文/俄英文模板合格证压力需带单位（如 2.5MPa 或 150Lb），请补充单位"
    return ""





def diff_fields(
    data: dict[str, Any],
    history_last: dict[str, str] | None,
    diff_keys: list[str],
) -> list[tuple[str, str, str]]:
    """计算与历史差异的项，返回 ``(key, current, history)`` 列表。

    仅当历史有值、当前有值、且两者不同才计入（空历史/空当前不触发差异闸）。
    """
    if not history_last:
        return []
    res: list[tuple[str, str, str]] = []
    for k in diff_keys:
        cur = (data.get(k) or "").strip()
        his = (history_last.get(k) or "").strip()
        if his and cur and cur != his:
            res.append((k, cur, his))
    return res



# ============================================================
# 统一护栏管道
# ============================================================
def _build_confirm_message(data: dict[str, Any], guard: dict[str, Any], *, multi: bool) -> str:
    confirmation = guard.get("confirmation", {}) or {}
    summary_fields = confirmation.get("summary_fields", []) or []
    lines = [f"· {field_label(k)} {data.get(k, '')}" for k in summary_fields]
    header = "即将批量发送以下合格证打印指令" if multi else "即将发送以下合格证打印指令"
    return header + "，请核对：\n\n" + "\n".join(lines) + "\n\n确认发送？"


def _prompt_required(
    data: dict[str, Any], guard: dict[str, Any], prompt: GuardPrompt
) -> tuple[bool, dict[str, Any]]:
    """② 必填补录 + ③ reconfirm（即使有值也确认）。返回 (ok, data)。"""
    data = dict(data)
    for f in guard.get("required_fields", []) or []:
        key = f.get("key")
        if not key:
            continue
        required = f.get("required", True)
        reconfirm = f.get("reconfirm", False)
        cur = (data.get(key) or "").strip()
        if cur and not reconfirm:
            continue
        if not cur and not required:
            continue
        text = prompt.prompt_field(key=key, label=field_label(key), current=cur, suggestion="")
        if text is None:
            return False, data
        text = text.strip()
        if not text and required:
            return False, data
        data[key] = text
    return True, data


def _prompt_diff(
    data: dict[str, Any],
    guard: dict[str, Any],
    prompt: GuardPrompt,
    history_last: dict[str, str] | None,
) -> tuple[bool, dict[str, Any]]:
    """④ 与历史差异确认（reconfirm_on_diff）。返回 (ok, data)。"""
    data = dict(data)
    for k in guard.get("reconfirm_on_diff", []) or []:
        for key, cur, his in diff_fields(data, history_last, [k]):
            text = prompt.prompt_field(key=key, label=field_label(key), current=cur, suggestion=his)
            if text is None:
                return False, data
            data[key] = text.strip() or cur
    return True, data


def _confirm_gate(
    data: dict[str, Any], guard: dict[str, Any], *, multi: bool, prompt: GuardPrompt
) -> bool:
    """⑤ 确认闸口：按 single/multi_batch 开关决定是否弹确认。"""
    confirmation = guard.get("confirmation", {}) or {}
    gate_open = confirmation.get("multi_batch" if multi else "single_batch", True)
    if not gate_open:
        return True
    return prompt.confirm(message=_build_confirm_message(data, guard, multi=multi))




def run_guard(
    data: dict[str, Any],
    *,
    guard: dict[str, Any],
    prompt: GuardPrompt,
    history_last: dict[str, str] | None = None,
    multi: bool = False,
) -> tuple[bool, dict[str, Any]]:
    """统一护栏管道。返回 ``(ok, updated_data)``。

    顺序：② 必填补录 / ③ reconfirm → ④ 差异确认 → ⑤ 确认闸口。
    任一环节用户取消 → 返回 ``(False, data)``，调用方应中止本批（及多计划剩余批）。

    Args:
        data: 当前合格证字段字典（不会被原地修改，返回副本）。
        guard: ``print.guard`` 段（来自 ``cfg("print.guard")``，dict 结构）。
        prompt: 交互实现（PyQt 或非交互）。
        history_last: 同型号上次实际值（Certificate 字段名 → 值），缺省跳过差异闸。
        multi: 是否多计划批量（影响确认闸口的 single/multi_batch 开关）。
    """
    # ① 历史自动补缺：用同型号上次实际值预填缺失字段（避免无谓弹框，且与 diff 闸不冲突）
    data = _backfill_history(dict(data), history_last)
    ok, data = _prompt_required(data, guard, prompt)
    if not ok:
        return False, data
    ok, data = _prompt_diff(data, guard, prompt, history_last)
    if not ok:
        return False, data
    if not _confirm_gate(data, guard, multi=multi, prompt=prompt):
        return False, data
    return True, data