# src/certflow/services/certificate_number_service.py
"""合格证自动编号服务（配置驱动）
后缀判定规则从 config.yaml → numbering.suffix_rules 读取，
可在不修改代码的情况下添加/修改后缀规则。
"""
from __future__ import annotations
import re
from datetime import date, datetime
from sqlalchemy.orm import Session
from certflow.models.sale_plan import SalePlan
from certflow.utils.logger import logger
[文档]
class CertificateNumberService:
"""合格证自动编号服务（配置驱动）
后缀判定规则按 numbering.suffix_rules 顺序匹配，命中即返回。
编号格式：V{YYMM}{起始序号:03d}{后缀}-{结束序号:03d}{后缀}
单台时： V{YYMM}{序号:03d}{后缀}
"""
def __init__(self, prefix_override: str | None = None) -> None:
"""初始化合格证自动编号服务
Args:
prefix_override: 自定义前缀覆盖（如 "2506"），None 则按当前年月生成
"""
self._prefix_override = prefix_override
# ============================================================
# 分隔符 / 前缀来源（配置驱动，§5.7.2-B/G）
# ============================================================
@staticmethod
def _cfg(key: str, default: str) -> str:
try:
from certflow.config.settings import cfg as _cfg
return str(_cfg(f"certificate.numbering.{key}", default) or default)
except Exception:
return default
[文档]
@classmethod
def stage_separator(cls) -> str:
"""编号阶段分隔符（默认 '---'）。"""
return cls._cfg("separator_stage", "---")
[文档]
@classmethod
def list_separator(cls) -> str:
"""清单保存分隔符（默认 '-'）。"""
return cls._cfg("separator_list", "-")
[文档]
@classmethod
def prefix_source(cls) -> str:
"""前缀(编号年月)来源：now / plan_date / shipped_date。"""
src = cls._cfg("prefix_source", "now")
return src if src in ("now", "plan_date", "shipped_date") else "now"
[文档]
@classmethod
def military_prefix(cls) -> str:
"""军工三位流水模式前缀（mode:7，无 V+YYMM，如 "JG"）。"""
return cls._cfg("military_prefix", "JG") or "JG"
@staticmethod
def _is_military_mode(record: SalePlan | None) -> bool:
"""记录是否走军工三位流水模式（number_model==7）。"""
if record is None:
return False
try:
from certflow.services.cert_numbering_policy import get_number_model
return get_number_model(record) == 7
except Exception:
return False
# ============================================================
# 公共方法
# ============================================================
[文档]
def generate_numbers(self, session: Session, sale_plan_ids: list[int]) -> int:
"""为指定销售计划记录批量生成产品编号
仅对 product_code 为空或以「【空白」开头的记录编号，按 sort_group/sort_order
顺序分配连续流水号，并写入 SalePlan。
Args:
session: SQLAlchemy 数据库会话对象
sale_plan_ids: 待编号的 SalePlan.id 列表
Returns:
int: 成功编号并更新入库的记录数
"""
records = (
session.query(SalePlan)
.filter(
SalePlan.id.in_(sale_plan_ids),
(
SalePlan.product_code.is_(None)
| (SalePlan.product_code == "")
| (SalePlan.product_code.like("【空白%"))
),
# H(维度16)：跳过不需要编号的记录（needs_numbering=False）；
# 历史 NULL 视为需要编号，兼容旧数据（避免全量跳过）。
(SalePlan.needs_numbering.is_(True) | (SalePlan.needs_numbering.is_(None))),
)
.order_by(SalePlan.sort_group, SalePlan.sort_order)
.all()
)
if not records:
return 0
prefix = self._get_prefix(records[0])
updated = 0
for record in records:
suffix = self._get_suffix(record)
quantity = record.quantity or 1
# 军工三位流水模式（mode:7，无 V+YYMM 前缀，逐台独立三位流水）
if self._is_military_mode(record):
mil_prefix = self._military_prefix()
# 军工计数器按军工前缀独立归组（不与 V+YYMM 混），三位流水连续
start_no = self._allocate_serial(session, mil_prefix, quantity)
end_no = start_no + quantity - 1
record.product_code = (
f"{mil_prefix}{start_no:03d}{suffix}-{mil_prefix}{end_no:03d}{suffix}"
)
logger.debug(
f"自动编号(军工) | {record.product_code} | 订单={record.sales_order_no}"
)
else:
start_no = self._allocate_serial(session, prefix, quantity)
end_no = start_no + quantity - 1
record.product_code = self._format_code(
prefix, start_no, end_no, suffix, separator=self.stage_separator()
)
logger.debug(
f"自动编号 | {record.product_code} | "
f"型号={record.product_model} | "
f"后缀={suffix or '(空)'} | "
f"订单={record.sales_order_no}"
)
updated += 1
session.commit()
logger.info(f"自动编号完成 | 数量={updated} | 前缀={prefix}")
return updated
[文档]
def generate_batch(
self,
session: Session,
sale_plan_id: int,
batch_qty: int,
ym: str | None = None,
start_override: int | None = None,
) -> dict:
"""按「本批台数」为单个销售计划生成一批产品编号（批次 0.5 / G4）
合同总数从 ``SalePlan.quantity`` 读取，用户只传 ``batch_qty``；
用 ``shipped_quantity``（已发累计）判定剩余与最后一批，从根上消除
``(4,1)`` 类参数歧义（见 OPERATOR_WORKFLOW §11.4.5）。生成的范围串
写入 ``SalePlan.product_code``（本批区间），并累加 ``shipped_quantity``；
发满则标记 ``shipping_status='已发货'``（与 B0-3 ``record_shipment`` 一致）。
编号前缀默认取当前月（``V+YYMM``）；传 ``ym``（如 ``"2604"``）可生成
**历史年月**批次，流水从该月已用最大流水续号（``peek_next_serial``）。
``start_override``（特殊插入）强制起始流水为指定值，并把计数器压到该起点。
Args:
session: SQLAlchemy 会话
sale_plan_id: 销售计划 ID
batch_qty: 本批发货台数（必须 > 0 且 <= 剩余可发）
ym: 目标编号年月（4 位 YYMM，如 ``2604``）；None 取当前月
start_override: 起始流水手填覆盖（特殊插入）；None 表示自动续号
Returns:
dict: 含 product_code / seq_start / seq_end / prefix / ym /
shipped_quantity / remaining / fully_shipped
Raises:
ValueError: 计划不存在 / batch_qty 非法 / 超过剩余可发
"""
plan = session.get(SalePlan, sale_plan_id)
if plan is None:
raise ValueError(f"SalePlan(id={sale_plan_id}) 不存在")
total = plan.quantity or 0
shipped = plan.shipped_quantity or 0
remaining = total - shipped
if batch_qty <= 0:
raise ValueError(f"本批台数必须 > 0，收到 {batch_qty}")
if batch_qty > remaining:
raise ValueError(
f"本批台数 {batch_qty} 超过剩余可发 {remaining}（合同 {total} / 已发 {shipped}）"
)
# 编号前缀：默认当前月（或 prefix_source 来源）；ym 覆盖为目标历史年月（如 2604 → V2604）
prefix = f"V{ym}" if ym else self._get_prefix(plan)
# 起始流水手填覆盖（特殊插入）：先把计数器压到 start-1，再分配即得指定起点
if start_override is not None:
self._seed_counter_to(session, prefix, int(start_override) - 1)
start_no = self._allocate_serial(session, prefix, batch_qty)
end_no = start_no + batch_qty - 1
# 无论自动/覆盖，写后把计数器推进到「已用最大流水」与本次结束号的较大者，
# 避免「起始流水手填覆盖」低于现有最大值时计数器回退，导致后续自动续号碰撞。
safe_end = max(self._max_serial_across(session, prefix), end_no)
self._bump_serial(session, prefix, safe_end)
suffix = self._get_suffix(plan)
range_str = self._format_code(
prefix, start_no, end_no, suffix, separator=self.stage_separator()
)
plan.product_code = range_str
plan.shipped_quantity = shipped + batch_qty
fully_shipped = False
if plan.shipped_quantity >= total:
plan.shipped_quantity = total
plan.shipping_status = "已发货"
fully_shipped = True
plan.fully_shipped = fully_shipped
session.commit()
logger.info(
f"批次编号完成 | id={plan.id} | {range_str} | "
f"已发={plan.shipped_quantity}/{total} | 满发={fully_shipped}"
)
return {
"sale_plan_id": sale_plan_id,
"product_code": range_str,
"prefix": prefix,
"ym": prefix[1:] if prefix.startswith("V") else prefix,
"seq_start": start_no,
"seq_end": end_no,
"shipped_quantity": plan.shipped_quantity,
"remaining": max(0, total - plan.shipped_quantity),
"fully_shipped": fully_shipped,
}
[文档]
def get_prefix(self) -> str:
"""获取当前编号前缀（V + YYYYMM 或覆盖前缀）
Returns:
str: 编号前缀字符串
"""
return self._get_prefix()
# ============================================================
# 前缀生成
# ============================================================
def _get_prefix(self, record: SalePlan | None = None) -> str:
"""计算编号前缀（V + 覆盖值或来源年月 YYMM）
来源由 ``certificate.numbering.prefix_source`` 决定（§5.7.2-G）：
- ``now``（默认）：实时年月；
- ``plan_date`` / ``shipped_date``：取记录对应日期的年月（贴近 VBA
「生产成功并发货时」语义）；记录缺失或日期为空时回退 ``now``。
``prefix_override`` 优先级最高（手动覆盖，如历史年月批次）。
Args:
record: 取日期来源的 SalePlan（plan_date/shipped_date）；批量同组
通常同月，取首条即可。
Returns:
str: 编号前缀
"""
if self._prefix_override:
return "V" + self._prefix_override
src = self.prefix_source()
if src in ("plan_date", "shipped_date") and record is not None:
date_val = getattr(record, src, None) or getattr(record, "plan_date", None)
if date_val:
try:
ym = date_val[:4] + date_val[5:7] if "-" in date_val else date_val[:6]
if len(ym) == 6 and ym.isdigit():
return "V" + ym[2:]
except Exception:
pass
return "V" + datetime.now().strftime("%y%m")
# ============================================================
# 后缀判定（配置驱动）
# ============================================================
# SalePlan 属性名 → 配置中 field 中文名的映射
# 仅保留 SalePlan 真实存在的列（注意：后缀规则作用于 SalePlan，
# 不能用 Certificate.supplier / 不存在的 color_mark）。
_FIELD_CN_TO_ATTR = {
"生产令号": "production_order_no",
"种类": "category",
"供货类型": "supply_type",
"产品名称": "product_name",
}
def _get_field_value(self, record: SalePlan, field_name: str) -> str:
"""获取 SalePlan 对象中指定字段的值
Args:
record: SalePlan ORM 对象
field_name: 配置中的中文字段名（映射到 ORM 属性）
Returns:
str: 字段值的去空格字符串，缺失返回空串
"""
attr = self._FIELD_CN_TO_ATTR.get(field_name, field_name)
value = getattr(record, attr, "") or ""
return str(value).strip()
def _get_suffix(self, record: SalePlan) -> str:
"""判定编号后缀（唯一真相源：config.numbering.suffix_rules）
规则按配置顺序匹配，命中即返回后缀。配置段为空或加载失败时返回空串
（不再回退硬编码规则，保证「配置驱动」语义不被旁路）。
Args:
record: SalePlan ORM 对象
Returns:
str: 匹配到的后缀字符串，无匹配返回空串
"""
try:
from certflow.config.settings import cfg
except ImportError:
logger.warning("无法加载配置，后缀置空（不再使用硬编码回退）")
return ""
rules = cfg("numbering.suffix_rules", [])
if not rules:
logger.warning("配置 numbering.suffix_rules 为空，后缀置空")
return ""
for rule in rules:
field = rule.get("field", "")
keywords = rule.get("keywords", [])
exclude = rule.get("exclude", [])
suffix = rule.get("suffix", "")
field_value = self._get_field_value(record, field) if field else ""
# 无 field 但有 keywords：遍历所有已知字段
if not field and keywords:
all_values = [
self._get_field_value(record, f) for f in self._FIELD_CN_TO_ATTR.values()
]
field_value = " ".join(v for v in all_values if v)
if not field_value:
continue
# exclude 检查
if exclude and field_value in exclude:
continue
# keywords 匹配
if keywords and any(kw in field_value for kw in keywords):
logger.debug(f"后缀匹配: {suffix} (字段={field}, 值={field_value[:30]})")
return suffix
return ""
# ============================================================
# 序列号管理
# ============================================================
def _get_max_serial_legacy(self, session: Session, prefix: str) -> int:
"""播种用：解析同前缀 SalePlan.product_code 的最大结束流水（一次性）。
仅看 SalePlan（产品编号源），不解析 Certificate.certificate_no
（其形如 ``CERT-YYYYMMDD-NNNN``，与产品编号 ``V...`` 是两个体系，
原分支恒不命中，见 BUG-004）。遍历全部同前缀码取 ``max``，规避
``order_by(desc).first()`` 在 ≥1000 时字符串序≠数值序的 BUG-001 风险。
Args:
session: SQLAlchemy 数据库会话对象
prefix: 编号前缀
Returns:
int: 最大结束流水号，无记录返回 0
"""
max_serial = 0
rows = (
session.query(SalePlan.product_code)
.filter(SalePlan.product_code.like(f"{prefix}%"))
.all()
)
for (code,) in rows:
if code:
max_serial = max(max_serial, self._extract_end_serial(code))
return max_serial
def _allocate_serial(self, session: Session, prefix: str, count: int) -> int:
"""从 ``AutoNumberCounter`` 原子分配 ``count`` 个连续流水号，返回起始流水号。
设计（BUG-005 / BUG-001 / BUG-004 统一修复）：
- **单一连续计数器**，按前缀（``V+YYMM``）分组 → 每月自动归零
（新前缀 = 新计数器行 = 从 1 起），无需 numbering_scheme 枚举。
- **不做后缀/类别隔离**：B 类等上千编号与日常共用同一流水，后续继续即可，
流水保持连续（用户明确：不引入内部/外部双字段，避免流水断续）。
- **异构/手动码不进计数器**：播种时仅按同前缀规范码取最大流水，
``A008`` 等乱序自由文本（不以 ``V+YYMM`` 开头）不命中前缀 → 不污染。
- 计数器行缺失时由既有 ``SalePlan.product_code`` 一次性播种，之后以计数器为准。
Args:
session: SQLAlchemy 会话
prefix: 编号前缀（如 ``V2607``）
count: 需连续分配的流水个数
Returns:
int: 起始流水号（下一个可用号）
"""
from certflow.models.auto_number import AutoNumberCounter
counter = (
session.query(AutoNumberCounter)
.filter(
AutoNumberCounter.rule_code == "product_code",
AutoNumberCounter.group_key == prefix,
)
.first()
)
if counter is None:
# 一次性播种：取同前缀既有 product_code 的最大流水（乱序码不命中前缀→排除）
seed = self._get_max_serial_legacy(session, prefix)
counter = AutoNumberCounter(
rule_code="product_code",
group_key=prefix,
current_serial=seed,
reset_date=datetime.now().strftime("%Y%m"),
)
session.add(counter)
session.flush()
start = counter.current_serial + 1
counter.current_serial = start + count - 1
return start
def _bump_serial(self, session: Session, prefix: str, end_serial: int) -> None:
"""手动编号后，确保计数器不低于 ``end_serial``。
避免「先自动编号（已建计数器行）→ 再手动编号到更大号 → 再自动编号」
时计数器回退导致流水碰撞。计数器行缺失时由既有 product_code 播种并与
``end_serial`` 取大。
Args:
session: SQLAlchemy 会话
prefix: 编号前缀（如 ``V2506``）
end_serial: 手动编号覆盖到的最大流水号
"""
from certflow.models.auto_number import AutoNumberCounter
counter = (
session.query(AutoNumberCounter)
.filter(
AutoNumberCounter.rule_code == "product_code",
AutoNumberCounter.group_key == prefix,
)
.first()
)
if counter is None:
seed = self._get_max_serial_legacy(session, prefix)
counter = AutoNumberCounter(
rule_code="product_code",
group_key=prefix,
current_serial=max(seed, end_serial),
reset_date=datetime.now().strftime("%Y%m"),
)
session.add(counter)
elif end_serial > counter.current_serial:
counter.current_serial = end_serial
session.flush()
def _extract_end_serial(self, code: str, prefix: str | None = None) -> int:
"""从编号字符串中提取结束流水号
Args:
code: 编号字符串（支持 "前缀-结束序号" 或 "前缀+序号" 形式）
prefix: 显式前缀（用于历史前缀提取，避免误用当前月前缀）。
为 None 时回退 ``_get_prefix()``（保持既有行为）。
Returns:
int: 提取到的结束流水号，未匹配返回 0
"""
match = re.search(r"-(\d+)", code)
if match:
return int(match.group(1))
pfx = prefix or self._get_prefix()
pattern = re.escape(pfx) + r"(\d+)"
match = re.search(pattern, code)
if match:
return int(match.group(1))
return 0
def _max_serial_across(self, session: Session, prefix: str) -> int:
"""跨 SalePlan.product_code 与 Certificate.product_code_range 取该前缀最大结束流水。
编号可能落在销售计划（自动/导入编号）或合格证（打印视图手动编）任一侧，
续号必须综合两表，否则手动编号不回灌会导致后续自动续号遗漏。
Args:
session: SQLAlchemy 会话
prefix: 编号前缀（如 ``V2604``）
Returns:
int: 该前缀最大结束流水号，无记录返回 0
"""
from certflow.models.certificate import Certificate
max_serial = 0
sp_rows = (
session.query(SalePlan.product_code)
.filter(SalePlan.product_code.like(f"{prefix}%"))
.all()
)
for (code,) in sp_rows:
if code:
max_serial = max(max_serial, self._extract_end_serial(code, prefix))
cert_rows = (
session.query(Certificate.product_code_range)
.filter(Certificate.product_code_range.like(f"{prefix}%"))
.all()
)
for (code,) in cert_rows:
if code:
max_serial = max(max_serial, self._extract_end_serial(code, prefix))
return max_serial
[文档]
def peek_next_serial(self, session: Session, prefix: str) -> int:
"""只读预瞄某前缀下一个可用流水号（不消耗、不建计数器）。
优先读 ``AutoNumberCounter``（已分配则 ``current_serial+1``）；
计数器缺失时回退跨表最大流水 + 1。供 UI 自动续号与「下一号」预览。
Args:
session: SQLAlchemy 会话
prefix: 编号前缀（如 ``V2604``）
Returns:
int: 下一可用流水号（从 1 起）
"""
from certflow.models.auto_number import AutoNumberCounter
counter = (
session.query(AutoNumberCounter)
.filter(
AutoNumberCounter.rule_code == "product_code",
AutoNumberCounter.group_key == prefix,
)
.first()
)
if counter is not None:
return counter.current_serial + 1
return self._max_serial_across(session, prefix) + 1
def _seed_counter_to(self, session: Session, prefix: str, serial: int) -> None:
"""将某前缀计数器行的 ``current_serial`` 强制设为 ``serial``（缺失则建行）。
用于「起始流水手填覆盖」：先把计数器压到 ``start-1``，随后 ``_allocate_serial``
即返回用户指定的起始号，且计数器随分配推进，保证后续自动续号不回退。
Args:
session: SQLAlchemy 会话
prefix: 编号前缀（如 ``V2604``）
serial: 目标 current_serial 值
"""
from certflow.models.auto_number import AutoNumberCounter
counter = (
session.query(AutoNumberCounter)
.filter(
AutoNumberCounter.rule_code == "product_code",
AutoNumberCounter.group_key == prefix,
)
.first()
)
if counter is None:
counter = AutoNumberCounter(
rule_code="product_code",
group_key=prefix,
current_serial=serial,
reset_date=datetime.now().strftime("%Y%m"),
)
session.add(counter)
else:
counter.current_serial = serial
session.flush()
# ============================================================
# 编号格式化
# ============================================================
@staticmethod
def _format_serial(serial: int) -> str:
"""格式化流水号（对齐VBA: <1000→3位, ≥1000→4位）"""
if serial < 1000:
return f"{serial:03d}"
return f"{serial:04d}"
def _format_code(
self, prefix: str, start: int, end: int, suffix: str, separator: str | None = None
) -> str:
"""拼接完整编号字符串
Args:
prefix: 编号前缀
start: 起始流水号
end: 结束流水号
suffix: 编号后缀
separator: 范围串分隔符（默认取 stage 分隔符 "---"）；清单保存用
``to_list_format`` 转 list 分隔符 "-"（§5.7.2-B）。
Returns:
str: 单台为 "前缀+序号+后缀"，多台为 "前缀+起始+分隔符+结束+后缀"
"""
if separator is None:
separator = self.stage_separator()
start_str = self._format_serial(start)
if start == end:
return f"{prefix}{start_str}{suffix}"
end_str = self._format_serial(end)
return f"{prefix}{start_str}{suffix}{separator}{end_str}{suffix}"
# ============================================================
# 军工三位流水模式（对齐VBA dd.vb:468-512）
# ============================================================
[文档]
@staticmethod
def generate_military_numbers(
prefix: str,
start_number: int,
quantity: int,
suffix: str = "",
) -> list[str]:
"""军工三位流水模式：前缀 + Format(i, "000") + 后缀
对齐 VBA dd.vb 第468-512行「军工 三位流水无V年月前缀」模式。
不包含 V+年月前缀，完全由用户手动指定。
Args:
prefix: 用户指定的前缀字符串（如 "JG"）
start_number: 起始编号
quantity: 打印份数
suffix: 编号后缀
Returns:
list[str]: 编号列表，如 ["JG001A", "JG002A", ...]
Examples:
>>> CertificateNumberService.generate_military_numbers("JG", 1, 3, "A")
['JG001A', 'JG002A', 'JG003A']
"""
return [f"{prefix}{i:03d}{suffix}" for i in range(start_number, start_number + quantity)]
# ============================================================
# 报告流水号分配（与合格证编号同源的 AutoNumberCounter 计数基础设施）
# ============================================================
[文档]
def allocate_report_serials(self, session: Session, make_date, count: int) -> list[int]:
"""为报告分配按「制作日期」归组的连续流水号（一页一个，返回 count 个）。
复用与合格证产品编号（``V+YYMM``）同源的 ``AutoNumberCounter`` 计数器，
rule_code 固定为 ``report_serial``，group_key 为制作日期 ``YYMMDD``；
按日归组、每日从 1 起连续递增，跨页/跨单整体续号，不依赖 JSON 文件或
``report_serial_seq`` 表。
编号渲染交由调用方（如 ``report_fill_service``）拼成
``C/1 编号：YYMMDDNNN`` 形态，本方法只负责分配底层连续流水。
Args:
session: SQLAlchemy 会话
make_date: 报告制作日期（``date`` 或 ISO 字符串）；流水按此日期归组。
count: 需分配的流水号个数（= 报告页数，每页一个编号）。
Returns:
list[int]: 长度 = count 的连续流水号（从当前计数器 +1 起）；
``count<=0`` 时返回空列表。
"""
from certflow.models.auto_number import AutoNumberCounter
if count <= 0:
return []
if isinstance(make_date, date):
key = make_date.strftime("%y%m%d")
reset = make_date.strftime("%Y%m%d")
else:
s = str(make_date)[:10]
key = s[2:].replace("-", "") if len(s) >= 10 else s
reset = s.replace("-", "")
counter = (
session.query(AutoNumberCounter)
.filter(
AutoNumberCounter.rule_code == "report_serial",
AutoNumberCounter.group_key == key,
)
.first()
)
if counter is None:
counter = AutoNumberCounter(
rule_code="report_serial",
group_key=key,
current_serial=0,
reset_date=reset,
)
session.add(counter)
session.flush()
start = counter.current_serial + 1
counter.current_serial = start + count - 1
session.commit()
return list(range(start, start + count))