"""工作流服务 - 管理文档完成状态
提供标记文档完成、自动更新进度百分比、自动更新生产状态等功能。
自动升级规则由 config.yaml production_status.auto_upgrade 驱动。
"""
from __future__ import annotations
from datetime import datetime
from typing import Any
from sqlalchemy.orm import Session
from certflow.models.sale_plan import SalePlan
from certflow.utils.logger import logger
def _load_auto_upgrade_rules() -> list[dict[str, Any]]:
"""从 config.yaml 加载自动升级规则"""
try:
from certflow.config.settings import cfg
return cfg("production_status.auto_upgrade", [])
except Exception:
return []
[文档]
class WorkflowService:
"""工作流服务 - 管理文档完成状态
自动升级规则由 config.yaml production_status.auto_upgrade 配置驱动。
新增升级规则只需修改 YAML，无需改动代码。
"""
# 文档类型 → (完成标记字段, 时间字段)
_DOC_FIELD_MAP = {
"certificate": ("certificate_done", "certificate_time"),
"nameplate": ("nameplate_done", "nameplate_time"),
"test_report": ("test_report_done", "test_report_time"),
"warranty": ("warranty_done", "warranty_time"),
"scan": ("cert_scan_done", "cert_scan_time"),
}
# 总共 5 个文档步骤
_TOTAL_STEPS = 5
def __init__(self, session: Session):
"""初始化工作流服务
Args:
session: SQLAlchemy 数据库会话
"""
self.session = session
self._auto_upgrade_rules = _load_auto_upgrade_rules()
[文档]
def mark_document_done(
self, sale_plan_id: int, doc_type: str, file_path: str | None = None
) -> SalePlan | None:
"""标记文档完成
Args:
sale_plan_id: 销售计划 ID
doc_type: 文档类型 (certificate/nameplate/test_report/warranty/scan)
file_path: 文件路径（可选，预留扩展）
Returns:
更新后的 SalePlan 对象，未找到则返回 None
"""
if doc_type not in self._DOC_FIELD_MAP:
logger.warning(f"未知文档类型: {doc_type}")
return None
plan = self.session.query(SalePlan).get(sale_plan_id)
if plan is None:
logger.warning(f"未找到 SalePlan id={sale_plan_id}")
return None
done_field, time_field = self._DOC_FIELD_MAP[doc_type]
setattr(plan, done_field, True)
setattr(plan, time_field, datetime.now())
# 自动更新进度百分比
self._update_progress(plan)
# 自动更新生产状态（由配置驱动）
self._update_status(plan)
self.session.commit()
logger.info(f"标记完成: SalePlan#{sale_plan_id} {doc_type}")
return plan
[文档]
def mark_document_undone(self, sale_plan_id: int, doc_type: str) -> SalePlan | None:
"""撤销文档完成标记
Args:
sale_plan_id: 销售计划 ID
doc_type: 文档类型
Returns:
更新后的 SalePlan 对象，未找到则返回 None
"""
if doc_type not in self._DOC_FIELD_MAP:
logger.warning(f"未知文档类型: {doc_type}")
return None
plan = self.session.query(SalePlan).get(sale_plan_id)
if plan is None:
logger.warning(f"未找到 SalePlan id={sale_plan_id}")
return None
done_field, time_field = self._DOC_FIELD_MAP[doc_type]
setattr(plan, done_field, False)
setattr(plan, time_field, None)
self._update_progress(plan)
self._update_status(plan)
self.session.commit()
logger.info(f"撤销完成: SalePlan#{sale_plan_id} {doc_type}")
return plan
[文档]
def batch_mark_done(self, sale_plan_ids: list[int], doc_type: str) -> dict[str, int]:
"""批量标记文档完成
Args:
sale_plan_ids: 销售计划 ID 列表
doc_type: 文档类型
Returns:
{"success": N, "failed": M}
"""
success = 0
failed = 0
for pid in sale_plan_ids:
result = self.mark_document_done(pid, doc_type)
if result is not None:
success += 1
else:
failed += 1
return {"success": success, "failed": failed}
def _update_progress(self, plan: SalePlan) -> None:
"""根据各文档完成情况更新进度百分比"""
done_count = sum(
[
plan.certificate_done or False,
plan.nameplate_done or False,
plan.test_report_done or False,
plan.warranty_done or False,
plan.cert_scan_done or False,
]
)
plan.progress_percent = int(done_count / self._TOTAL_STEPS * 100)
def _update_status(self, plan: SalePlan) -> None:
"""根据进度自动更新生产状态（由 config.yaml auto_upgrade 规则驱动）
配置格式见 config.yaml production_status.auto_upgrade:
- condition: Python 表达式（可用 progress_percent, is_ready_to_ship, production_status）
- set_to: 目标状态
- set_all_docs_done: 是否标记全部文档完成
"""
# 构建求值上下文
context = {
"progress_percent": plan.progress_percent,
"is_ready_to_ship": plan.is_ready_to_ship,
"production_status": plan.production_status,
}
for rule in self._auto_upgrade_rules:
condition = rule.get("condition", "")
if not condition:
continue
try:
matched = eval(condition, {"__builtins__": {}}, context)
except Exception:
logger.warning(f"自动升级规则求值失败: {condition}")
continue
if not matched:
continue
# 匹配成功，执行规则
if rule.get("set_all_docs_done"):
plan.certificate_done = True
plan.nameplate_done = True
plan.test_report_done = True
plan.warranty_done = True
plan.cert_scan_done = True
plan.progress_percent = 100
logger.info(f"已发货自动标记全部完成: SalePlan#{plan.id}")
return
target = rule.get("set_to")
if target:
plan.production_status = target
logger.debug(f"自动升级状态: SalePlan#{plan.id} → {target}")
return