AiDOPWarehouse/doc/_gen_flow_diagrams.py

# -*- coding: utf-8 -*-
"""生成蓝图用业务流程图（每功能 1 张 + 每模块总体 1 张），S1 蓝图风格：横向流程框图。

输出：doc/_tmp_docx_out/flows/{module}_{key}.png
"""
from __future__ import annotations

import sys
from pathlib import Path

import matplotlib

matplotlib.use("Agg")
import matplotlib.pyplot as plt
from matplotlib.patches import FancyArrowPatch, FancyBboxPatch

HERE = Path(__file__).resolve().parent
sys.path.insert(0, str(HERE))

from _gen_s9_modules_data import MODULES  # noqa: E402

OUT = HERE / "_tmp_docx_out" / "flows"
OUT.mkdir(parents=True, exist_ok=True)

plt.rcParams["font.family"] = ["Microsoft YaHei"]
plt.rcParams["axes.unicode_minus"] = False

# 模块总体流程（手工概括，与文档 1.2 总体业务流程图呼应）
OVERALL = {
    "系统集成": ["第三方系统(ERP/MES/WMS)", "数据源登记/标准API推数(方式丙)", "同步任务执行与清洗标准化", "贴源/标准/指标层入仓", "业务模块与看板消费", "出站回写第三方系统"],
    "S9": ["运营指标主数据", "指标层日批计算", "九宫格总览(红黄绿预警)", "格子下钻模块看板", "联动智慧诊断"],
    "运营诊断": ["看板未达标指标", "智慧诊断环节识别", "七维下钻根因溯源", "诊断结论与报告", "转改善闭环"],
    "运营改善": ["诊断问题建档", "审批派单到责任人", "行动项执行跟踪", "有效性验证(前后对比)", "复盘闭环与经验固化"],
    "ChatBI": ["看板/九宫格上下文", "自然语言提问", "意图解析与指标聚合", "结论与建议输出", "跳转看板或诊断"],
}

MODULE_PREFIX = {"系统集成": "int", "S9": "s9", "运营诊断": "od", "运营改善": "oi", "ChatBI": "cb"}


def draw_flow(title: str, steps: list[tuple[str, str]], out: Path):
    """steps: [(活动名称, 执行角色)]，横向框图 + 箭头。"""
    n = len(steps)
    w = max(8.0, 1.9 * n + 1.2)
    fig, ax = plt.subplots(figsize=(w, 1.9), dpi=200)
    ax.set_xlim(0, n * 1.9 + 0.4)
    ax.set_ylim(0, 1.9)
    ax.axis("off")
    ax.text(n * 0.95 + 0.2, 1.68, title, ha="center", va="center", fontsize=12, fontweight="bold", color="#1F3864")
    box_w, box_h, y0 = 1.55, 0.78, 0.42
    for i, (name, role) in enumerate(steps):
        x0 = 0.2 + i * 1.9
        box = FancyBboxPatch((x0, y0), box_w, box_h, boxstyle="round,pad=0.06",
                             fc="#E8F1FA", ec="#2E75B6", lw=1.4)
        ax.add_patch(box)
        ax.text(x0 + box_w / 2, y0 + box_h * 0.62, name, ha="center", va="center",
                fontsize=10.5, fontweight="bold", color="#1F3864", wrap=True)
        ax.text(x0 + box_w / 2, y0 + box_h * 0.24, role, ha="center", va="center",
                fontsize=8.5, color="#595959")
        if i < n - 1:
            ar = FancyArrowPatch((x0 + box_w + 0.07, y0 + box_h / 2), (x0 + 1.9 + 0.13, y0 + box_h / 2),
                                 arrowstyle="-|>", mutation_scale=16, color="#2E75B6", lw=1.6)
            ax.add_patch(ar)
    fig.savefig(out, bbox_inches="tight", facecolor="white")
    plt.close(fig)
    print("FLOW", out.name)


def main():
    for mod_key, mod in MODULES.items():
        prefix = MODULE_PREFIX[mod_key]
        draw_flow(f"{mod['module_title']}·总体业务流程", [(s, "") for s in OVERALL[mod_key]],
                  OUT / f"{prefix}_overall.png")
        for feat in mod["features"]:
            if feat.get("impl_status") == "未实现":
                continue
            steps = [(name, role) for _no, name, role, _desc, _inp, _out in feat["flow_steps"]]
            draw_flow(f"{feat['name']}（{feat['func']}）业务流程", steps,
                      OUT / f"{prefix}_{feat['func'].lower().replace('-', '_')}.png")


if __name__ == "__main__":
    main()