certflow.services.cert_numbering_policy 源代码

# src/certflow/services/cert_numbering_policy.py
"""合格证编号策略（配置驱动）——步骤 6：编号时机与编码模式。

对应蓝图 §4.4：把「编号时机」与「编码模式判定」从硬编码抽到 ``config.yaml``：

- ``certificate.numbering.timing``：``on_demand``（默认，查询页手动触发）
  / ``on_import``（写入 Certificate 时对未编号记录自动编号）。
- ``certificate.numbering.modes``：按 ``when_field`` 的关键词匹配 → ``number_model``，
  替代原先硬编码的 ``number_model = 0``（复用 ``printer.numbering_modes`` 数值）。
- ``certificate.numbering.range_regex``：范围编码解析正则（如
  ``V2603121XL---125XL`` → 起止流水），供逐台展开 / 校验。

本模块的核心判定函数为**纯函数**（``resolve_number_model`` / ``parse_number_range``），
不依赖配置加载与 Qt，可独立单测；``get_*`` 系列为读取配置的薄封装。

设计原则（与总原则一致）：默认配置保持现状语义——``timing`` 默认 ``on_demand``、
``modes`` 默认为空（一律回退 ``default_mode=0``），即与既有行为零差异；
启用只需改 YAML，无需改代码。
"""

from __future__ import annotations

import re
from typing import Any


def _field_value(record: Any, field: str) -> str:
    """取记录中某字段的去空格字符串值（兼容 ORM 对象与 dict）。"""
    value = record.get(field, "") if isinstance(record, dict) else getattr(record, field, "")
    return str(value or "").strip()




def resolve_number_model(record: Any, modes: list[dict] | None, default: int = 0) -> int:
    """按配置的编码模式规则判定 ``number_model``（纯函数）。

    规则按顺序匹配：某规则的 ``when_field``（或兼容键 ``field``）对应字段值包含
    ``keywords`` 中任一关键词时，返回该规则的 ``mode``；全部未命中返回 ``default``。

    Args:
        record: SalePlan / Certificate ORM 对象，或字段字典。
        modes: 模式规则列表（每项含 ``when_field`` / ``keywords`` / ``mode``）。
        default: 未命中任何规则时的默认编码模式。

    Returns:
        int: 判定出的 ``number_model`` 数值。
    """
    for rule in modes or []:
        field = rule.get("when_field") or rule.get("field")
        keywords = rule.get("keywords") or []
        if not field or not keywords:
            continue
        value = _field_value(record, field)
        if value and any(str(kw) in value for kw in keywords):
            try:
                return int(rule.get("mode", default))
            except (TypeError, ValueError):
                return default
    return default





def resolve_sn_kks_position(record: Any, modes: list[dict] | None, default: str = "") -> str:
    """按配置的编码模式规则判定双编码位置约定（纯函数，对齐 VBA ar(2,8)）。

    与 ``resolve_number_model`` 共用命中规则：某规则的 ``when_field`` 对应字段值
    含 ``keywords`` 中任一关键词时，返回该规则的 ``sn_kks_position``；全部未命中
    返回 ``default``（空串=默认不互换）。

    Args:
        record: SalePlan / Certificate ORM 对象，或字段字典。
        modes: 模式规则列表（每项含 ``when_field`` / ``keywords`` / ``sn_kks_position``）。
        default: 未命中时的默认位置约定（空串）。

    Returns:
        str: 双编码位置约定（"B12单元格为产品编号，B13单元格为KKS编码" 等，
             或空串）。
    """
    for rule in modes or []:
        field = rule.get("when_field") or rule.get("field")
        keywords = rule.get("keywords") or []
        if not field or not keywords:
            continue
        value = _field_value(record, field)
        if value and any(str(kw) in value for kw in keywords):
            return str(rule.get("sn_kks_position", default) or default)
    return default



# 范围编码默认匹配：起始数字串 + 后缀 + 一个及以上连字符 + 结束数字串
# 说明：起始串常含 V+YYMM 前缀（如 "2603121"），结束串仅为流水（如 "125"）；
# 故起始流水按结束串位数从末尾截取（"2603121"[-3:]="121"），剥离 YYMM。
_DEFAULT_RANGE_RE = re.compile(r"(\d+)(\D*)-+(\d+)")




def parse_number_range(code: str, regex: str | None = None) -> tuple[int, int] | None:
    """解析范围编码的起止流水（纯函数）。

    如 ``V2603121XL---125XL`` → ``(121, 125)``、``V2604520Y-524Y`` → ``(520, 524)``；
    非范围编码返回 ``None``。

    Args:
        code: 完整编码字符串。
        regex: 自定义解析正则（需含命名组 ``start`` / ``end``）；``None`` 时用内置逻辑。

    Returns:
        tuple[int, int] | None: ``(起始流水, 结束流水)``；无法解析返回 ``None``。
    """
    if not code:
        return None
    text = str(code).strip()

    # 自定义正则：要求命名组 start / end，直接取用
    if regex:
        try:
            match = re.match(regex, text)
        except re.error:
            return None
        if not match:
            return None
        try:
            return (int(match.group("start")), int(match.group("end")))
        except (IndexError, ValueError):
            return None

    # 内置逻辑：起始串按结束串位数从末尾截取，剥离 V+YYMM 前缀
    match = _DEFAULT_RANGE_RE.search(text)
    if not match:
        return None
    start_full, _suffix, end_str = match.group(1), match.group(2), match.group(3)
    try:
        end = int(end_str)
        n = len(end_str)
        start = int(start_full[-n:]) if len(start_full) > n else int(start_full)
    except ValueError:
        return None
    return (start, end)



# 产品编号元信息解析：V{YYMM}{起始流水}{后缀}[-{结束流水}{后缀}]
# 前缀为非数字字母串（如 "V"），其后 4 位为年月，再后为起始流水，流水与
# 结束流水之间可含后缀与一个或多个连字符（-+，对齐 _format_code 的 ---）。
_PRODUCT_CODE_META_RE = re.compile(
    r"^(?P<prefix>[A-Za-z]+?)(?P<ym>\d{4})(?P<start>\d+)(?P<suffix>.*?)(?:-+(?P<end>\d+).*)?$"
)




def get_military_prefix() -> str:
    """读取军工模式前缀（mode:7，无 V+YYMM 的军工前缀，如 "JG"）。

    默认 "JG"；军工三位流水模式自动编号时用此前缀 + 三位流水。
    """
    value = _cfg_get("certificate.numbering.military_prefix", "JG")
    return str(value or "JG").strip()



# 军工三位流水区间串解析：前缀(非V字母) + 三位流水 + 后缀，单 '-' 连接起止，
# 如 "JG001A-JG100A" → 起止流水 1/100、后缀 "A"、台数 100。
# 与标准模式（含 V+YYMM）区分：军工前缀后的"年月"位缺失，流水直接三位起。
_MILITARY_META_RE = re.compile(
    r"^(?P<prefix>[A-Za-z]+?)(?P<start>\d{3,})(?P<suffix>[A-Za-z]*?)"
    r"(?:-(?P<prefix2>[A-Za-z]+?)(?P<end>\d{3,})(?P<suffix2>[A-Za-z]*?))?$"
)




def parse_military_code_meta(code: str) -> dict | None:
    """解析军工三位流水区间串（mode:7，无 V+YYMM 前缀）。

    如 ``JG001A-JG100A`` →
    ``{"prefix": "JG", "seq_start": 1, "seq_end": 100, "suffix": "A",
       "quantity": 100}``；单台 ``JG001A`` → seq_end==seq_start。

    军工码特征：前缀非 ``V`` 且其后直接是 3 位以上流水（无 4 位年月），
    与标准 ``V{YYMM}{流水}`` 区分。供 G2 逐台展开写入 Certificate。

    Args:
        code: 军工产品编号区间串（SalePlan.product_code / Certificate.product_code_range）。

    Returns:
        dict | None: 含 prefix/seq_start/seq_end/suffix/quantity；无法解析返回 None。
    """
    if not code:
        return None
    text = str(code).strip()
    m = _MILITARY_META_RE.match(text)
    if not m:
        return None
    # 起止前缀须一致（军工区间串首尾同前缀）
    if m.group("prefix2") and m.group("prefix2") != m.group("prefix"):
        return None
    seq_start = int(m.group("start"))
    end_str = m.group("end")
    seq_end = int(end_str) if end_str else seq_start
    suffix = m.group("suffix") or ""
    suffix2 = m.group("suffix2") or ""
    if suffix2 and suffix2 != suffix:
        return None
    quantity = seq_end - seq_start + 1
    if quantity < 1:
        return None
    return {
        "prefix": m.group("prefix"),
        "seq_start": seq_start,
        "seq_end": seq_end,
        "suffix": suffix,
        "quantity": quantity,
    }





def parse_product_code_meta(code: str) -> dict | None:
    """解析产品编号区间串的元信息（前缀/年月/起止流水/台数）。

    如 ``V2604520Y---524Y`` →
    ``{"prefix": "V", "ym": "2604", "seq_start": 520, "seq_end": 524,
       "suffix": "Y", "quantity": 5}``；单台 ``V2604520Y`` → seq_end==seq_start。

    供 G2 在 ``create_certificates`` 中将区间串拆填入
    ``Certificate.product_code_range`` / ``product_code_prefix`` /
    ``product_code_ym`` / ``product_code_seq_start`` / ``product_code_seq_end`` /
    ``quantity``（本批台数 = seq_end - seq_start + 1）。

    Args:
        code: 产品编号区间串（SalePlan.product_code / Certificate.product_code_range）。

    Returns:
        dict | None: 含 prefix/ym/seq_start/seq_end/suffix/quantity；无法解析返回 None。
    """
    if not code:
        return None
    text = str(code).strip()
    m = _PRODUCT_CODE_META_RE.match(text)
    if not m:
        return None
    seq_start = int(m.group("start"))
    end_str = m.group("end")
    seq_end = int(end_str) if end_str else seq_start
    quantity = seq_end - seq_start + 1
    if quantity < 1:
        return None
    return {
        "prefix": m.group("prefix"),
        "ym": m.group("ym"),
        "seq_start": seq_start,
        "seq_end": seq_end,
        "suffix": m.group("suffix"),
        "quantity": quantity,
    }



def _cfg_get(key: str, default: Any) -> Any:
    """读取配置（隔离导入失败场景，配置不可用时回退默认）。"""
    try:
        from certflow.config.settings import cfg
    except ImportError:
        return default
    return cfg(key, default)




def get_numbering_timing() -> str:
    """读取编号时机：``on_demand``（默认）或 ``on_import``。"""
    value = _cfg_get("certificate.numbering.timing", "on_demand")
    return str(value or "on_demand")





def get_issue_date_source() -> str:
    """读取出厂日期年月来源（BUG-006 J，配置驱动）。

    - ``sync_numbering_ym``（默认）：出厂日期年月同步产品编号年月
      （``product_code_ym``，如 ``V2607…`` → ``2026.07``）。
    - ``current_ym``：出厂日期取当前年月（``datetime.now()``）。

    Returns:
        str: ``sync_numbering_ym`` 或 ``current_ym``（未知值回退默认）。
    """
    value = _cfg_get("certificate.issue_date_source", "sync_numbering_ym")
    return str(value or "sync_numbering_ym")





def get_number_model(record: Any) -> int:
    """按 ``certificate.numbering.modes`` 判定记录的 ``number_model``。

    默认（modes 为空）回退 ``certificate.numbering.default_mode``（默认 0），
    与既有硬编码 ``number_model = 0`` 行为一致。
    """
    modes = _cfg_get("certificate.numbering.modes", []) or []
    default = _cfg_get("certificate.numbering.default_mode", 0) or 0
    try:
        default = int(default)
    except (TypeError, ValueError):
        default = 0
    return resolve_number_model(record, modes, default)





def get_sn_kks_position(record: Any) -> str:
    """按 ``certificate.numbering.modes`` 判定记录的 ``sn_kks_position``。

    默认（modes 为空）回退空串（默认不互换）。与 ``get_number_model`` 共用命中
    规则，使双编码位置约定随客户/关键词一并确定（对齐 VBA ar(2,8)）。

    Args:
        record: SalePlan / Certificate ORM 对象，或字段字典。

    Returns:
        str: 双编码位置约定（"B12单元格为产品编号，B13单元格为KKS编码" 等，
             或空串）。
    """
    modes = _cfg_get("certificate.numbering.modes", []) or []
    return resolve_sn_kks_position(record, modes, "")





def get_range_regex() -> str | None:
    """读取范围编码解析正则（未配置返回 ``None`` → 用内置默认）。"""
    value = _cfg_get("certificate.numbering.range_regex", None)
    return str(value) if value else None





def get_no_cert_categories() -> list[str]:
    """读取免编号品类关键词列表（默认空列表）。"""
    value = _cfg_get("numbering.no_cert_categories", []) or []
    return [str(v) for v in value]





def is_auto_calc_dn_pn() -> bool:
    """自动计算口径压力开关（§5.7.2-E / 维度8，对齐 VBA G11）。

    默认 True（现状：打印前自动从型号解析口径/压力并预填）；
    ``certificate.numbering.auto_calc_dn_pn: false`` 时关闭自动推算，
    仅用已存值、不就口径/压力缺失弹补充对话框。

    Returns:
        bool: 是否启用自动计算口径压力。
    """
    return bool(_cfg_get("certificate.numbering.auto_calc_dn_pn", True))





def is_yellow_skip_enabled() -> bool:
    """读取是否按源表黄底(65535)跳过编号（默认 False）。"""
    return bool(_cfg_get("numbering.yellow_skip_enabled", False))





def derive_needs_numbering(record: Any) -> bool:
    """导入期派生是否需要编号（对齐 VBA 黄底跳过语义，维度16/H）。

    返回 ``False``（不需要编号）当满足任一条件：
    - 品类/产品名/型号命中 ``no_cert_categories`` 关键词（螺栓/密封圈等免出证铭牌件）；
    - ``yellow_skip_enabled`` 且背景色含黄底(65535)。

    其余返回 ``True``；历史 ``NULL`` 视为需要编号（兼容旧数据）。

    Args:
        record: SalePlan ORM 对象或字段字典（需含 ``category`` / ``product_name`` /
                ``product_model`` / ``_background_colors``）。

    Returns:
        bool: 是否需要编号。
    """
    text = " ".join(_field_value(record, f) for f in ("category", "product_name", "product_model"))
    for kw in get_no_cert_categories():
        if kw and kw in text:
            return False
    if is_yellow_skip_enabled():
        bg = record.get("_background_colors", {}) if isinstance(record, dict) else {}
        for v in bg.values():
            if v == 65535 or str(v) == "65535":
                return False
    return True