AiDOPWarehouse/doc/_add_req_func_codes.py

# -*- coding: utf-8 -*-
"""Add REQ/FUNC codes from triple对照表 into existing requirement/blueprint docx."""
from __future__ import annotations

import re
import shutil
import tempfile
from pathlib import Path

import openpyxl
from docx import Document

BASE = Path(r"C:\Users\skygu\OneDrive\Projects\AIDOP\项目\项目管理\产互联项目管理\交付文档")

DOCS = {
    "S0": {
        "req": BASE / "S0" / "S0-基础主数据-业务需求描述-V0.5.docx",
        "blue": BASE / "S0" / "S0-基础主数据建模-业务蓝图设计方案-V0.5.docx",
        "triple": BASE / "S0" / "S0-三联对照表.xlsx",
    },
    "S1": {
        "req": BASE / "S1" / "S1产销协同业务需求描述.docx",
        "blue": BASE / "S1" / "S1产销协同模块蓝图设计方案.docx",
        "triple": BASE / "S1" / "S1-三联对照表.xlsx",
    },
    "S2": {
        "req": BASE / "S2" / "S2制造协同业务需求描述.docx",
        "blue": BASE / "S2" / "S2制造协同模块蓝图设计方案.docx",
        "triple": BASE / "S2" / "S2-三联对照表.xlsx",
    },
    "S3": {
        "req": BASE / "S3" / "S3供应协同业务需求描述.docx",
        "blue": BASE / "S3" / "S3供应协同模块蓝图设计方案.docx",
        "triple": BASE / "S3" / "S3-三联对照表.xlsx",
    },
    "S4": {
        "req": BASE / "S4" / "S4采购执行业务需求描述.docx",
        "blue": BASE / "S4" / "S4采购执行模块蓝图设计方案.docx",
        "triple": BASE / "S4" / "S4-三联对照表.xlsx",
    },
}

# S3/S4 交付目录暂无 xlsx 时，与 aidopFuncCodes.ts 及 S1 编号规则对齐
BUILTIN_TRIPLE: dict[str, list[tuple[str, str, str, str]]] = {
    "S3": [
        ("REQ-S3-001", "物料需求计划", "FUNC-S3-001", "物料需求计划"),
        ("REQ-S3-002", "物料交货计划", "FUNC-S3-002", "物料交货计划"),
        ("REQ-S3-003", "交货单异常记录", "FUNC-S3-003", "交货单异常记录"),
        ("REQ-S3-004", "物料采购申请", "FUNC-S3-004", "物料采购申请"),
        ("REQ-S3-005", "要货令", "FUNC-S3-005", "要货令"),
        ("REQ-S3-006", "物料采购订单", "FUNC-S3-006", "物料采购订单"),
        ("REQ-S3-007", "委外加工订单", "FUNC-S3-007", "委外加工订单"),
        ("REQ-S3-008", "工序外协订单", "FUNC-S3-008", "工序外协订单"),
        ("REQ-S3-009", "供应协同看板", "FUNC-S3-009", "供应协同看板"),
        ("REQ-S3-010", "工单物料齐套上线看板", "FUNC-S3-010", "工单物料齐套上线看板"),
        ("REQ-S3-011", "MDP运行监控", "FUNC-S3-011", "MDP运行监控"),
        ("REQ-S3-012", "委外采购明细", "FUNC-S3-012", "委外采购明细"),
    ],
    "S4": [
        ("REQ-S4-001", "供应商交货管理", "FUNC-S4-001", "供应商交货管理"),
        ("REQ-S4-002", "供应商发货单", "FUNC-S4-002", "供应商发货单"),
        ("REQ-S4-003", "采购退货单", "FUNC-S4-003", "采购退货单"),
        ("REQ-S4-004", "IQC退货查询", "FUNC-S4-004", "IQC退货查询"),
        ("REQ-S4-005", "采购执行看板主页", "FUNC-S4-005", "采购执行看板主页"),
        ("REQ-S4-006", "供应商欠料看板", "FUNC-S4-006", "供应商欠料看板"),
    ],
}

# Document title aliases -> triple 需求名称 key
S0_ALIASES = {
    "物料维护": "物料管理",
    "客户信息维护": "客户管理",
    "订单优先级配置": "订单优先级规则管理",
    "合同评审周期": "合同评审周期规则管理",
    "产品设计周期": "产品设计周期规则管理",
    "订单评审周期": "订单评审周期规则管理",
    "供应商维护": "供应商管理",
    "货源清单": "采购分类与货源管理",
    "品类采购前置期": "品类交货周期管理",
    "物料计划周期": "物料计划周期管理",
    "物料计划周期标准": "物料计划周期管理",
    "生产线维护": "产线管理",
    "生产线": "产线管理",
    "工作中心": "工作中心管理",
    "产线岗位维护": "产线工位管理",
    "生产线物料（线边物料）": "产线物料管理",
    "线边物料": "产线物料管理",
    "人员技能维护": "人员技能分配",
    "标准工序": "标准工序管理",
    "标准工艺路线": "标准工艺路线管理",
    "工序流转卡": "工艺流程卡管理",
    "标准 BOM": "标准 BOM 管理",
    "SOP 维护": "SOP 维护管理",
    "SOP 文档维护": "SOP 维护管理",
    "SOP 文件类型": "SOP 文件类型管理",
    "前处理要素": "前置工序要素管理",
    "物料前处理要素": "前置工序要素管理",
    "前处理要素参数": "前置工序要素参数管理",
    "生产要素参数": "生产要素参数管理",
    "物料替代": "物料替代明细管理",
    "物料替代关系": "物料替代明细管理",
    "物料替代（制造入口）": "物料替代明细管理（制造域）",
    "选择替代方案": "物料代替方案管理",
    "选择替代方案（制造入口）": "物料代替方案管理",
    "工单控制参数": "工单控制参数管理",
    "订单排程周期": "订单排程周期管理",
    "检验项目": "检验项目管理",
    "检验方法": "检验方法管理",
    "检验依据": "检验基准管理",
    "检验标准": "检验标准管理",
    "检验频率": "检验频次管理",
    "检验仪器": "检验仪器管理",
    "抽样方案": "抽样方案管理",
    "原材料检验规范（IQC）": "来料检验规范（IQC）",
    "过程检验规范（IPQC）": "过程检验规范（IPQC）",
    "FQC 检验规范": "成品检验规范（OQC/FQC）",
    "OQC 检验规范": "成品检验规范（OQC/FQC）",
    "原材料白名单": "来料白名单管理",
    "质量基础": "质量字典管理",
    "质量基础字典": "质量字典管理",
    "库位维护": "库位管理",
    "货架列表": "库位货架管理",
    "部门维护": "部门管理",
    "雇员列表": "员工管理",
    "物料职责维护": "员工工作职责管理",
    "成本中心": "成本中心管理",
    "零件包装规格": "物料包装规格管理",
    "物料状态任务指派": "任务分配管理",
    "人员技能": "人员技能管理",
    "标签格式": "标签类型管理",
    "单号类型": "批次类型管理",
    "单号规则维护": "批次号控制管理",
    "单号规则": "批次号控制管理",
}

S1_ALIASES = {}
S2_ALIASES = {
    "生产排程": "生产排程",
    "排产异常记录": "排产异常记录",
    "排产异常": "排产异常记录",
    "排产数据维护": "排产异常记录",
    "可执行日计划": "可执行日计划",
    "产线工作日历管理": "产线工作日历管理",
    "产线工作日历": "产线工作日历管理",
    "产线休息时间管理": "产线休息时间管理",
    "产线休息时间": "产线休息时间管理",
    "产线节假日管理": "产线节假日管理",
    "产线节假日": "产线节假日管理",
    "产线加班管理": "产线加班管理",
    "产线加班": "产线加班管理",
    "工单执行进度看板": "工单执行进度看板",
    "工单执行进度": "工单执行进度看板",
    "工单执行追踪": "工单执行进度看板",
    "S2制造协同看板": "S2制造协同看板（KPI）",
    "制造协同看板与KPI": "S2制造协同看板（KPI）",
    "制造协同看板": "S2制造协同看板（KPI）",
    "作业计划": "可执行日计划",
}

S1_REQ_SECTION = {
    "产品设计管理": "REQ-S1-001",
    "合同评审管理": "REQ-S1-002",
    "订单评审管理": "REQ-S1-003",
    "订单交付管理": "REQ-S1-004",
    "订单发货管理": "REQ-S1-005",
    "工单下达": "REQ-S1-006",
    "计划联动看板": "REQ-S1-007",
    "需求明细核验": "REQ-S1-008",
    "产销协同看板": "REQ-S1-009",
}

S1_FUNC_SECTION = {k: f"FUNC-S1-{i:03d}" for i, k in enumerate(
    ["产品设计管理", "订单评审管理", "订单交付管理", "订单发货管理", "合同评审管理",
     "计划联动看板", "需求明细核验", "工单下达", "产销协同看板"], 1)}

# S1 blueprint chapter order differs from REQ order; map by title
S1_FUNC_BY_TITLE = {
    "产品设计管理": "FUNC-S1-001",
    "订单评审管理": "FUNC-S1-003",
    "订单交付管理": "FUNC-S1-004",
    "订单发货管理": "FUNC-S1-005",
    "合同评审管理": "FUNC-S1-002",
    "计划联动看板": "FUNC-S1-007",
    "需求明细核验": "FUNC-S1-008",
    "工单下达": "FUNC-S1-006",
    "产销协同看板": "FUNC-S1-009",
}

S3_REQ_SECTION = {name: req for req, name, _, _ in BUILTIN_TRIPLE["S3"]}
S3_FUNC_BY_TITLE = {
    title: func
    for _, name, func, fname in BUILTIN_TRIPLE["S3"]
    for title in (name, fname)
}
S4_REQ_SECTION = {name: req for req, name, _, _ in BUILTIN_TRIPLE["S4"]}
S4_FUNC_BY_TITLE = {
    title: func
    for _, name, func, fname in BUILTIN_TRIPLE["S4"]
    for title in (name, fname)
}

S3_ALIASES: dict[str, str] = {}
S4_ALIASES: dict[str, str] = {}


def load_triple_from_rows(rows: list[tuple[str, str, str, str]]) -> dict[str, tuple[str, str]]:
    by_name: dict[str, tuple[str, str]] = {}
    for req, req_name, func, func_name in rows:
        by_name[req_name] = (req, func)
        by_name[func_name] = (req, func)
    return by_name


def load_triple(path: Path, module: str) -> dict[str, tuple[str, str]]:
    if module in BUILTIN_TRIPLE and not path.exists():
        return load_triple_from_rows(BUILTIN_TRIPLE[module])
    tmp = Path(tempfile.gettempdir()) / f"{module}_triple.xlsx"
    shutil.copy2(path, tmp)
    wb = openpyxl.load_workbook(tmp, data_only=True)
    ws = wb.active
    by_name: dict[str, tuple[str, str]] = {}
    if module == "S2":
        for row in ws.iter_rows(min_row=2, max_row=ws.max_row, values_only=True):
            if not row or len(row) < 5:
                continue
            func, req = str(row[3]).strip(), str(row[4]).strip()
            if not req.startswith("REQ-"):
                continue
            menu = str(row[6] or "")
            name = menu.split("→")[-1].strip() if "→" in menu else func
            by_name[name] = (req, func)
        # also map by standard func names from repo
        names = [
            "生产排程", "排产异常记录", "可执行日计划", "产线工作日历管理", "产线休息时间管理",
            "产线节假日管理", "产线加班管理", "工单执行进度看板", "S2制造协同看板（KPI）",
        ]
        reqs = [f"REQ-S2-{i:03d}" for i in range(1, 10)]
        funcs = [f"FUNC-S2-{i:03d}" for i in range(1, 10)]
        for n, r, f in zip(names, reqs, funcs):
            by_name.setdefault(n, (r, f))
        wb.close()
        return by_name

    for row in ws.iter_rows(min_row=2, max_row=ws.max_row, values_only=True):
        if not row or not row[0]:
            continue
        req, req_name, func, func_name = [str(x).strip() for x in row[:4]]
        by_name[req_name] = (req, func)
        by_name[func_name] = (req, func)
    wb.close()
    return by_name


def build_title_map(module: str, by_name: dict[str, tuple[str, str]]) -> dict[str, tuple[str, str]]:
    aliases = {"S0": S0_ALIASES, "S1": S1_ALIASES, "S2": S2_ALIASES, "S3": S3_ALIASES, "S4": S4_ALIASES}[module]
    title_map: dict[str, tuple[str, str]] = {}
    for name, codes in by_name.items():
        title_map[name] = codes
    for alias, target in aliases.items():
        if target in by_name:
            title_map[alias] = by_name[target]
        elif target in title_map:
            title_map[alias] = title_map[target]
    return title_map


def normalize_title(text: str) -> str:
    text = text.strip()
    text = re.sub(r"^（[一二三四五六七八九十]+）", "", text)
    text = re.sub(r"^\d+(?:\.\d+)*\s+", "", text)
    text = re.sub(r"\d+$", "", text).strip()
    text = re.sub(r"\s*——.*$", "", text)
    text = re.sub(r"\s*—\s*.*$", "", text)
    text = re.sub(r"（FUNC-[^）]+）", "", text)
    text = re.sub(r"（REQ-[^）]+）", "", text)
    text = re.sub(r"\s*\(REQ-[^)]+\)", "", text)
    text = re.sub(r"\s*\(FUNC-[^)]+\)", "", text)
    return text.strip()


def extract_heading_title(text: str) -> str | None:
    text = text.strip()
    if not text:
        return None
    if text in {"功能说明", "业务描述", "目录", "目录2"}:
        return None
    if re.match(r"^[一二三四五六七八九十]+、", text):
        return None
    if "  —  " in text or " — " in text:
        return normalize_title(text.split("—", 1)[0])
    if len(text) > 80:
        return None
    m = re.match(r"^（[一二三四五六七八九十]+）(.+)$", text)
    if m:
        return normalize_title(m.group(1))
    m = re.match(r"^(\d+(?:\.\d+)*)\s+(.+)$", text)
    if m:
        return normalize_title(m.group(2))
    if len(text) <= 40:
        return normalize_title(text)
    return None


def has_code(text: str, code: str) -> bool:
    return code in text


def set_paragraph_text(paragraph, new_text: str) -> None:
    if paragraph.runs:
        paragraph.runs[0].text = new_text
        for r in paragraph.runs[1:]:
            r.text = ""
    else:
        paragraph.add_run(new_text)


def strip_existing_codes(text: str) -> str:
    return re.sub(r"（(?:REQ|FUNC)-S\d+-\d+）", "", text)


def append_code(paragraph, code: str) -> bool:
    text = paragraph.text
    if has_code(text, code):
        return False
    suffix = f"（{code}）"
    set_paragraph_text(paragraph, text.rstrip() + suffix)
    return True


def match_codes(module: str, title: str, title_map: dict[str, tuple[str, str]], doc_kind: str) -> str | None:
    if module == "S1" and doc_kind == "req":
        return S1_REQ_SECTION.get(title)
    if module == "S1" and doc_kind == "blue":
        return S1_FUNC_BY_TITLE.get(title)
    if module == "S3" and doc_kind == "req":
        return S3_REQ_SECTION.get(title)
    if module == "S3" and doc_kind == "blue":
        return S3_FUNC_BY_TITLE.get(title)
    if module == "S4" and doc_kind == "req":
        return S4_REQ_SECTION.get(title)
    if module == "S4" and doc_kind == "blue":
        return S4_FUNC_BY_TITLE.get(title)
    if title in title_map:
        req, func = title_map[title]
        return req if doc_kind == "req" else func
    return None


def iter_paragraphs(doc: Document):
    for p in doc.paragraphs:
        yield p
    for table in doc.tables:
        for row in table.rows:
            for cell in row.cells:
                for p in cell.paragraphs:
                    yield p


S2_BLUE_EXACT_ROWS = {
    "生产排程": "FUNC-S2-001",
    "排产异常记录": "FUNC-S2-002",
    "排产数据维护": "FUNC-S2-002",
    "可执行日计划": "FUNC-S2-003",
    "日计划列表查询": "FUNC-S2-003",
    "日计划下达": "FUNC-S2-003",
    "产线工作日历管理": "FUNC-S2-004",
    "产线休息时间管理": "FUNC-S2-005",
    "产线节假日管理": "FUNC-S2-006",
    "产线加班管理": "FUNC-S2-007",
    "工单执行进度看板": "FUNC-S2-008",
    "工单执行追踪": "FUNC-S2-008",
}


def s2_req_inline_match(text: str, key: str) -> bool:
    if key == "排产异常":
        return text.startswith("排产异常")
    if key == "可执行日计划":
        return text.startswith("展示面向车间产线的可执行日计划") or text.startswith('计划员/调度人员进入"可执行日计划"')
    if key == "产线工作日历":
        return text.startswith("产线工作日历")
    if key == "产线休息时间":
        return text.startswith("产线休息时间")
    if key == "产线节假日":
        return text.startswith("产线节假日")
    if key == "产线加班":
        return text.startswith("产线加班")
    if key == "工单执行进度":
        return text.startswith("工单执行进度") or text.startswith("管理层/计划员进入\"工单执行进度看板\"")
    return False


def process_doc(
    path: Path,
    module: str,
    title_map: dict[str, tuple[str, str]],
    doc_kind: str,
    *,
    refresh: bool = False,
) -> int:
    doc = Document(path)
    changed = 0
    if refresh:
        for p in iter_paragraphs(doc):
            stripped = strip_existing_codes(p.text)
            if stripped != p.text:
                set_paragraph_text(p, stripped)
                changed += 1
    for p in iter_paragraphs(doc):
        text = p.text.strip()
        if not text:
            continue
        if module == "S2" and doc_kind == "blue" and text in S2_BLUE_EXACT_ROWS:
            if append_code(p, S2_BLUE_EXACT_ROWS[text]):
                changed += 1
            continue
        title = extract_heading_title(p.text)
        if not title:
            continue
        code = match_codes(module, title, title_map, doc_kind)
        if not code:
            continue
        if append_code(p, code):
            changed += 1
    if module == "S2" and doc_kind == "req":
        inline_keys = [
            ("排产异常", "REQ-S2-002"),
            ("可执行日计划", "REQ-S2-003"),
            ("产线工作日历", "REQ-S2-004"),
            ("产线休息时间", "REQ-S2-005"),
            ("产线节假日", "REQ-S2-006"),
            ("产线加班", "REQ-S2-007"),
            ("工单执行进度", "REQ-S2-008"),
        ]
        for p in iter_paragraphs(doc):
            t = p.text.strip()
            if not t or len(t) > 200:
                continue
            for key, code in inline_keys:
                if s2_req_inline_match(t, key) and not has_code(t, code):
                    if append_code(p, code):
                        changed += 1
                    break
    doc.save(path)
    return changed


def main():
    import sys

    refresh = "--refresh" in sys.argv
    only = None
    for arg in sys.argv[1:]:
        if arg.startswith("--only="):
            only = {x.strip().upper() for x in arg.split("=", 1)[1].split(",") if x.strip()}
    modules = [m for m in DOCS if not only or m in only]
    for module in modules:
        cfg = DOCS[module]
        by_name = load_triple(cfg["triple"], module)
        title_map = build_title_map(module, by_name)
        n1 = process_doc(cfg["req"], module, title_map, "req", refresh=refresh)
        n2 = process_doc(cfg["blue"], module, title_map, "blue", refresh=refresh)
        print(f"{module}: req +{n1}, blue +{n2}, mapped titles {len(title_map)}")


if __name__ == "__main__":
    main()