certflow.services.print_history_service 源代码

# src/certflow/services/print_history_service.py
"""历史回填兜底服务（docs/PRINT_BLUEPRINT.md §11.2 / §11.8 阶段1·2）。

阶段1（本文件当前范围）：仅实现 :meth:`HistoryBackfillService.candidates`，
读 ``PrintLog`` 聚合「同型号上次实际打印值」，供统一护栏的 ④ 差异确认闸
（``reconfirm_on_diff``）使用。复用 PrintLog（51 字段/每产品一条）作历史源，零新表。

阶段2 待扩展：``backfill``（仅补缺、易变字段不兜底）+ ``record_after_generate``
（打印成功后写入 PrintLog，§10.8 不可事后改写）。

注意：``check_standard``（检验标准）不在 PrintLog 的 51 字段中，故差异闸对该字段
暂取不到历史值、自动跳过差异确认（仍走 required/reconfirm 逻辑）。后续如需覆盖，
可在阶段2 扩展其他历史源。
"""

from __future__ import annotations

from datetime import datetime
from typing import Any

from sqlalchemy.orm import Session

from certflow.models.print_log import PrintLog
from certflow.utils.database import DatabaseManager
from certflow.utils.logger import logger

# PrintLog 列 -> Certificate 字段名（用于 diff 历史比对；check_standard 不在 PrintLog，略过）
PRINTLOG_TO_CERT_FIELD: dict[str, str] = {
    "ShiYongWenDu": "temperature",
    "ShiYongJieZhi": "medium",
    "PN": "pn",
    "DN": "dn",
}

# 历史自动补缺（阶段2·A backfill）允许的字段白名单：仅易变、可衍生字段，
# 名称/型号/规格/日期/编号等「易变且不兜底」字段绝不自动回填（§11.8 阶段2）。
BACKFILL_FIELDS: tuple[str, ...] = ("temperature", "medium", "pn", "dn", "check_standard")




def backfill(
    data: dict[str, Any],
    history_last: dict[str, str] | None,
    fields: tuple[str, ...] | None = None,
) -> dict[str, Any]:
    """历史自动补缺（护栏 stage ①）：用同型号上次实际值预填 ``data`` 中**缺失**的字段。

    仅对 ``fields`` 白名单内、且当前为空的字段生效；用户已填值、历史缺失、或不在
    白名单的字段一律不动。返回更新后的 ``data``（不污染入参）。

    Args:
        data: 当前合格证字段字典。
        history_last: 同型号上次实际值（``cert_field -> value``），缺省跳过。
        fields: 允许补缺的字段白名单，缺省用 :data:`BACKFILL_FIELDS`。
    """
    if not history_last:
        return data
    fields = fields or BACKFILL_FIELDS
    out = dict(data)
    for f in fields:
        if f not in history_last:
            continue
        if (out.get(f) or "").strip():
            continue  # 用户已填，不覆盖
        out[f] = history_last[f]
    return out





class HistoryBackfillService:
    """历史回填兜底服务：读 PrintLog 聚合历史 / 构造并写入 PrintLog（§11.8 阶段1·2）。"""

    def __init__(self, session: Session) -> None:
        self.session = session



    def candidates(self, model: str | None, spec: str | None = None) -> dict[str, str] | None:
        """按产品型号取最近一条成功打印记录的实际值（作历史候选）。

        Args:
            model: 产品型号（``product_model``）。
            spec: 规格（预留；§11.2 归并键 ``key: [product_model, product_spec]``，
                  阶段2 起按 型号+规格 进一步归并，当前仅按型号）。

        Returns:
            字典 ``{cert_field: value}``（如 ``{"temperature": "≤200℃", ...}``）；
            无历史记录返回 ``None``。
        """
        if not model:
            return None
        model = model.strip()
        if not model:
            return None
        try:
            log = (
                self.session.query(PrintLog)
                .filter(PrintLog.ChanPinXingHao == model)
                .filter(PrintLog.status == "success")
                .order_by(PrintLog.id.desc())
                .first()
            )
        except Exception as exc:  # noqa: BLE001
            logger.debug(f"[HistoryBackfill] 查询 PrintLog 失败({model}): {exc}")
            return None
        if log is None:
            return None
        res: dict[str, str] = {}
        for col, cert_field in PRINTLOG_TO_CERT_FIELD.items():
            val = getattr(log, col, None)
            if val:
                res[cert_field] = str(val)
        return res or None


    # ---- 阶段2·B：PrintLog 统一写点（消除 cert_log_service / certificate_print_service 两处构造） ----



    def assemble_print_log(self, *, certificate_id: int, fields: dict[str, Any]) -> PrintLog:
        """单一 PrintLog 构造器：由调用方解析好的字段字典组装 ``PrintLog``（不提交）。

        各打印路径（GUI / CLI-batch）保留各自的 SN/KKS/温度/PN 派生逻辑，
        仅在此处统一落库构造，杜绝运行期重复写与字段不一致。

        Args:
            certificate_id: 关联合格证批次 ID。
            fields: 已解析的 PrintLog 字段字典（键为 PrintLog 列名）。
        """
        return PrintLog(certificate_id=certificate_id, **fields)




    def record_after_generate(
        self,
        *,
        certificate_id: int,
        fields: dict[str, Any],
        status: str = "success",
        error_message: str | None = None,
        content_summary: str | None = None,
        now: datetime | None = None,
    ) -> PrintLog | None:
        """打印成功后写入一条 PrintLog（统一写点，逐条提交）。

        取代旧 ``CertificatePrintService.record_print_log`` 的内联构造+提交；
        CLI/batch 路径经此单点写库。GUI 路径因批量单提交语义走
        :meth:`assemble_print_log` + 外部统一提交，不调用本方法。

        §10.8：PrintLog 不可事后改写，故本方法仅新增、不更新既有记录。

        Returns:
            成功返回 ``PrintLog``，异常返回 ``None``。
        """
        try:
            now = now or datetime.now()
            f = dict(fields)
            f.setdefault("hgz_PrintTime", now)
            f["status"] = status
            if error_message is not None:
                f["error_message"] = error_message
            if content_summary is not None:
                f["content_summary"] = content_summary
            log = self.assemble_print_log(certificate_id=certificate_id, fields=f)
            self.session.add(log)
            DatabaseManager.commit_with_retry(self.session)
            return log
        except Exception as e:  # noqa: BLE001
            logger.error(f"[HistoryBackfill] 写入 PrintLog 失败(cert={certificate_id}): {e}")
            self.session.rollback()
            return None