"""状态推断模块
从自由文本 execution_status 中推断标准化的 production_status，
并支持提取日期信息。
规则设计原则：
- 优先级从高到低：发货 > 完成 > 生产中 > 待生产
- 日期格式自动识别：2024/03/15已发、2024-03-20发完 等
- 默认返回"待生产"
配置驱动：
- 状态枚举、推断规则、查询关键词均从 config.yaml 的 production_status 节点加载
- 调用 init_from_config() 后，所有硬编码回退值被配置值覆盖
"""
from __future__ import annotations
import re
from pathlib import Path
from typing import Any, Literal
# ============================================================
# 状态常量与类型定义（默认值，会被配置覆盖）
# ============================================================
ProductionStatus = Literal["待生产", "生产中", "已完成", "已发货", "部分发货"]
STATUS_VALUES: list[str] = [
"待生产",
"生产中",
"已完成",
"已发货",
"部分发货",
]
STATUS_DEFAULT: str = "待生产"
[文档]
class StatusInference:
"""从自由文本执行情况中推断生产状态
execution_status 是历史自由文本字段，内容格式不统一：
- "2024/03/15已发"
- "已发货"
- "2024-03-20发完"
- "生产中"
- "完成"
- ""（空值）
通过正则匹配将自由文本映射到标准化的 production_status。
配置驱动：调用 init_from_config() 从 config.yaml 加载推断规则和查询关键词。
Examples:
>>> StatusInference.infer("2024/03/15已发")
'已发货'
>>> StatusInference.infer("生产中")
'生产中'
>>> StatusInference.infer(None)
'待生产'
>>> StatusInference.extract_date("2024/03/15已发")
'2024-03-15'
"""
_initialized: bool = False
# 状态映射规则（按优先级排序，先匹配的先生效）—— 默认值，会被配置覆盖
STATUS_RULES: list[tuple[str, str]] = [
# 部分发货（必须在发货之前，因为"部分发货"包含"发货"）
(r"(部分发|分批发|已发部分)", "部分发货"),
# 发货相关
(r"(已发|发完|发货|发运|已交付|已发出)", "已发货"),
(r"\d{4}[/-]\d{1,2}[/-]\d{1,2}\s*(已发|发货|发运)", "已发货"),
# 完成相关
(r"(完成|完工|已完|制作完成)", "已完成"),
# 生产中相关
(r"(生产中|加工中|制作中)", "生产中"),
# 待生产/未开始
(r"(待生产|未开始|未生产|未加工)", "待生产"),
]
# 查询关键词映射 —— 默认值，会被配置覆盖
QUERY_KEYWORDS: dict[str, str] = {
"已发货": "已发",
"部分发货": "部分",
"已完成": "完成",
"生产中": "生产",
}
@classmethod
def _load_config_from_yaml(cls, config_path: str | None = None) -> dict[str, Any]:
"""尝试通过项目配置系统或直接解析 YAML 加载配置"""
# 尝试通过项目配置系统加载（支持 !include 等自定义标签）
try:
from certflow.config.settings import cfg as settings_cfg
return settings_cfg("production_status", {})
except Exception:
pass
# 回退：直接解析 YAML
import yaml
if config_path is None:
config_path = str(
Path(__file__).resolve().parent.parent.parent.parent / "config" / "config.yaml"
)
try:
with open(config_path, encoding="utf-8") as f:
full_config = yaml.safe_load(f)
return full_config.get("production_status", {}) if full_config else {}
except (FileNotFoundError, Exception):
return {}
[文档]
@classmethod
def init_from_config(cls, config_path: str | None = None) -> None:
"""从配置文件初始化状态推断模块
在应用启动时调用一次，从 config.yaml 的 production_status 节点加载：
- 状态枚举值 (values)
- 默认状态 (default)
- 推断规则 (inference_rules)
- 查询关键词 (query_keywords)
优先使用项目 ConfigLoader（支持 !include 语法），
回退到标准 yaml.safe_load（仅当配置不含自定义标签时）。
Args:
config_path: 配置文件路径，为 None 时自动查找
"""
if cls._initialized:
return
config = cls._load_config_from_yaml(config_path)
if not config:
cls._initialized = True
return
# 加载状态枚举值
values: list[str] = config.get("values", [])
if values:
global STATUS_VALUES
STATUS_VALUES = values
# 加载默认状态
default: str | None = config.get("default")
if default:
global STATUS_DEFAULT
STATUS_DEFAULT = default
# 加载推断规则
rules: list[dict] = config.get("inference_rules", [])
if rules:
cls.STATUS_RULES = [("|".join(r["patterns"]), r["set_to"]) for r in rules]
# 加载查询关键词
keywords: dict = config.get("query_keywords", {})
if keywords:
cls.QUERY_KEYWORDS = keywords
cls._initialized = True
[文档]
@classmethod
def infer(cls, execution_status: str | None) -> ProductionStatus:
"""从执行情况推断生产状态
Args:
execution_status: 执行情况的原始文本，如 "2024/03/15已发"
Returns:
推断的状态，如 "已发货"，无匹配时返回默认状态
"""
if not execution_status or not isinstance(execution_status, str):
return STATUS_DEFAULT # type: ignore[return-value]
text = str(execution_status).strip()
if not text:
return STATUS_DEFAULT # type: ignore[return-value]
for pattern, status in cls.STATUS_RULES:
if re.search(pattern, text):
return status # type: ignore[return-value]
return STATUS_DEFAULT # type: ignore[return-value]
[文档]
@classmethod
def get_query_keyword(cls, status: str) -> str:
"""获取状态对应的查询关键词（用于模糊匹配 execution_status）
Args:
status: 标准化状态值，如 "已发货"
Returns:
查询关键词，如 "已发"，未配置时返回原状态值
"""
if not cls._initialized:
cls.init_from_config()
return cls.QUERY_KEYWORDS.get(status, status)
[文档]
@classmethod
def load_rules_from_config(cls, config: dict[str, Any]) -> None:
"""从配置字典加载规则（向后兼容旧接口）
Args:
config: 配置字典，支持两种格式：
- 旧格式：{"status_rules": [(pattern, status), ...]}
- 新格式：{"inference_rules": [{"patterns": [...], "set_to": "..."}, ...]}
Example:
>>> config = {
... "status_rules": [
... ("已发|发完", "已发货"),
... ("完成", "已完成"),
... ]
... }
>>> StatusInference.load_rules_from_config(config)
"""
if "inference_rules" in config:
cls.STATUS_RULES = [
("|".join(r["patterns"]), r["set_to"]) for r in config["inference_rules"]
]
elif "status_rules" in config:
cls.STATUS_RULES = [(pattern, status) for pattern, status in config["status_rules"]]
# ============================================================
# 自测
# ============================================================
if __name__ == "__main__":
test_cases: list[tuple[str | None, str, str | None]] = [
("2024/03/15已发", "已发货", "2024-03-15"),
("已发货", "已发货", None),
("2024-3-20发完", "已发货", "2024-03-20"),
("2026/1/9已发", "已发货", "2026-01-09"),
("生产中", "生产中", None),
("已完成", "已完成", None),
("部分发货", "部分发货", None),
("待生产", "待生产", None),
("2024年03月15日已发货", "已发货", "2024-03-15"),
("2024-03-15发完", "已发货", "2024-03-15"),
("加工中", "生产中", None),
("已完工", "已完成", None),
("2024/05/20已交付", "已发货", "2024-05-20"),
("分批发", "部分发货", None),
("未开始", "待生产", None),
("", "待生产", None),
(None, "待生产", None),
("无意义文本", "待生产", None),
]
all_passed = True
for text, expected_status, expected_date in test_cases:
status = StatusInference.infer(text)
date = StatusInference.extract_date(text)
status_ok = status == expected_status
date_ok = date == expected_date
if not status_ok or not date_ok:
all_passed = False
print(
f"FAIL: text={text!r} | "
f"status: got={status!r}, expected={expected_status!r} | "
f"date: got={date!r}, expected={expected_date!r}"
)
if all_passed:
print("所有测试通过！")
else:
print("存在测试失败，请检查。")