kg-agent/scripts/generate_manual_fill_workbook.py

"""生成「人工补填清单」Excel：列出所有非 DMS 来源的模板字段（需人工搬运/填写）。

来源分类沿用 scripts/generate_dms_mapping.py：
- DMS来源            —— 自动从 DMS 模型拉取，无需人工
- 源Excel搬运（DMS未建模型）—— 源 Excel 有、DMS 无模型，需人工从源文件搬运
- 新增融合            —— DMS 与源 Excel 均无，为图谱建模/关联添加，需人工确认填写

输出：data/manual_fill/人工补填清单.xlsx
"""

from __future__ import annotations

import importlib.util
import re
import sys
from pathlib import Path

from openpyxl import Workbook, load_workbook
from openpyxl.styles import Alignment, Border, Font, PatternFill, Side
from openpyxl.utils import get_column_letter


sys.stdout.reconfigure(encoding="utf-8")
ROOT = Path(r"E:\CODE\knowledge_agent")

# 载入字段分类模块（复用 TEMPLATES / FUSION_FIELDS / SOURCE_EXCEL_NOTES / OVERRIDES / EXCLUDE_DMS）
_spec = importlib.util.spec_from_file_location(
    "dms_map", ROOT / "scripts" / "generate_dms_mapping.py"
)
dms_map = importlib.util.module_from_spec(_spec)
assert _spec and _spec.loader
_spec.loader.exec_module(dms_map)

TEMPLATES = dms_map.TEMPLATES
FUSION = dms_map.FUSION_FIELDS
SRC_NOTES = dms_map.SOURCE_EXCEL_NOTES
DEFAULT_SOURCES = dms_map.DEFAULT_SOURCES
OVERRIDES = dms_map.OVERRIDES
EXCLUDE_DMS = dms_map.EXCLUDE_DMS
NOTES = dms_map.NOTES
ALIASES = dms_map.load_dms()

# 每个模板与 DMS 数据合并时的对齐键（决定 DMS 行与人工补填行如何 merge）
MERGE_KEYS: dict[str, str] = {
    "项目信息": "项目编号",
    "项目月度财务": "项目编号 + 年月",
    "投标记录": "项目编号 + 年份",
    "人员信息": "工号",
    "考勤与人员财务": "工号 + 项目编号 + 年月",
    "排班明细": "工号 + 项目编号 + 年月",
    "岗位编制": "项目编号 + 岗位",
    "设备信息": "项目编号 + 设备编号",
    "采购与维保": "项目编号 + 名称/合同名称 + 月份/日期",
    "检查记录": "项目编号 + 检查日期",
    "片区信息": "片区编号",
    "项目片区关系": "项目编号 + 片区编号",
    "人员证书": "工号 + 证书名称",
}

# 特殊备注（源数据坑 / 填写口径提醒）
SPECIAL_NOTES: dict[tuple[str, str], str] = {
    ("考勤与人员财务", "税后工资"): "源表无「税后工资」列名：赵巷公园等文件该金额填在表头为「员工签字」的列（如 4896.25）",
    ("考勤与人员财务", "1-31日"): "值为 常/日/夜/日1 等班次标记，原样搬运",
    ("排班明细", "1-31日班次"): "值为班次代码，如 OFF/A003/SQ-104/M11-033；OFF=休息",
    ("人员证书", "姓名"): "可读性核对列，由工号解析；登记表申勤 sheet 缺工号，需按姓名在花名册补全",
    ("人员信息", "当前服务项目编号"): "DMS sq_employee 仅有项目名称(五级公司)，编号需人工映射为项目编号",
    ("项目信息", "项目负责人工号"): "DMS 有项目负责人姓名，工号需在员工花名册按姓名匹配",
    ("设备信息", "设备责任人工号"): "模板必填；源设备台账可能有责任人姓名，工号需匹配花名册",
    ("项目信息", "续签前项目编号"): "可从 scripts/analyze_renewal_folders.py 生成的 output/续签链-档案目录.csv 回填",
    ("项目信息", "上级项目编号"): "由业务大表层级序号/委托方式推导父项目",
    ("项目月度财务", "财务经办人工号"): "源收费进程表有经办人姓名，工号需匹配花名册",
}


def norm_header(h: str, tpl: str) -> str:
    """模板表头 → 规范字段名（与 generate_dms_mapping.TEMPLATES 对齐）。"""
    h = h.strip()
    h = re.sub(r"\*$", "", h)
    h = re.sub(r"（选填）$", "", h)
    m = re.match(r"^(\d{1,2})日$", h)
    if m:
        return "1-31日班次" if tpl == "排班明细" else "1-31日"
    h = re.sub(r"（[^）]*）$", "", h)
    return h


def dms_src(tpl: str, f: str) -> str:
    """判断字段是否有 DMS 来源（与 generate_dms_mapping 同逻辑）。"""
    ov = OVERRIDES.get((tpl, f))
    if ov:
        return f"{ov[0]}.{ov[1]}"
    if (tpl, f) in EXCLUDE_DMS:
        return ""
    for alias, desc in ALIASES.items():
        nf = dms_map.norm(f)
        if nf == alias or (len(nf) >= 2 and nf in alias):
            return desc
    return ""


def read_template_meta() -> dict[str, set[str]]:
    """读取 data/templates/*.xlsx：返回 {模板: 必填字段集合}。"""
    required: dict[str, set[str]] = {}
    for f in sorted((ROOT / "data" / "templates").glob("*.xlsx")):
        tpl = f.stem
        wb = load_workbook(f, read_only=True)
        ws = wb.worksheets[0]
        row = next(ws.iter_rows(min_row=1, max_row=1))
        req: set[str] = set()
        for c in row:
            try:
                rgb = str(c.fill.fgColor.rgb) if c.fill and c.fill.fgColor else ""
            except Exception:  # noqa: BLE001
                rgb = ""
            if "C00000" in rgb and c.value:
                req.add(norm_header(str(c.value), tpl))
        required[tpl] = req
        wb.close()
    return required


HEADER_FILL = PatternFill("solid", fgColor="2F5597")
SRC_FILL = PatternFill("solid", fgColor="EDEDED")
FUSION_FILL = PatternFill("solid", fgColor="FFF2CC")
REQ_FILL = PatternFill("solid", fgColor="FCE4D6")
HEADER_FONT = Font(name="微软雅黑", size=10, bold=True, color="FFFFFF")
BODY_FONT = Font(name="微软雅黑", size=10)
NOTE_FONT = Font(name="微软雅黑", size=9, color="595959")
THIN = Side(style="thin", color="BFBFBF")
BORDER = Border(left=THIN, right=THIN, top=THIN, bottom=THIN)


def main() -> None:
    required = read_template_meta()
    out_dir = ROOT / "data" / "manual_fill"
    out_dir.mkdir(parents=True, exist_ok=True)
    out_path = out_dir / "人工补填清单.xlsx"

    wb = Workbook()
    ws = wb.active
    ws.title = "补填清单"
    headers = ["模板", "序号", "字段", "补填类型", "来源与填写说明", "与DMS对齐键", "必填", "备注/示例"]
    widths = [16, 6, 20, 16, 70, 26, 6, 46]
    for c, (h, w) in enumerate(zip(headers, widths), start=1):
        cell = ws.cell(row=1, column=c, value=h)
        cell.font = HEADER_FONT
        cell.fill = HEADER_FILL
        cell.alignment = Alignment(horizontal="center", vertical="center", wrap_text=True)
        cell.border = BORDER
        ws.column_dimensions[get_column_letter(c)].width = w
    ws.row_dimensions[1].height = 28

    total_manual = 0
    total_fusion = 0
    total_src = 0
    r = 2
    per_tpl_summary: list[tuple[str, int, int, int]] = []
    for tpl, fields in TEMPLATES.items():
        req_set = required.get(tpl, set())
        manual = []
        for f in fields:
            if not dms_src(tpl, f):
                manual.append(f)
        n_fusion = sum(1 for f in manual if f in FUSION.get(tpl, set()))
        n_src = len(manual) - n_fusion
        total_manual += len(manual)
        total_fusion += n_fusion
        total_src += n_src
        per_tpl_summary.append((tpl, len(fields), len(manual), n_fusion))
        for i, f in enumerate(manual, start=1):
            if f in FUSION.get(tpl, set()):
                kind = "新增融合"
                note = "DMS 与源 Excel 均无，需人工确认填写；用途：图谱关联/可读性核对"
            else:
                kind = "源Excel搬运"
                note = SRC_NOTES.get((tpl, f)) or DEFAULT_SOURCES.get(tpl, "")
            ws.cell(row=r, column=1, value=tpl).font = BODY_FONT
            ws.cell(row=r, column=2, value=i).font = BODY_FONT
            ws.cell(row=r, column=3, value=f).font = BODY_FONT
            ws.cell(row=r, column=4, value=kind).font = BODY_FONT
            ws.cell(row=r, column=5, value=note).font = BODY_FONT
            ws.cell(row=r, column=6, value=MERGE_KEYS.get(tpl, "")).font = BODY_FONT
            is_req = "是" if f in req_set else "否"
            ws.cell(row=r, column=7, value=is_req).font = BODY_FONT
            ws.cell(row=r, column=8, value=SPECIAL_NOTES.get((tpl, f), "")).font = BODY_FONT
            ws.cell(row=r, column=4).fill = FUSION_FILL if kind == "新增融合" else SRC_FILL
            if is_req == "是":
                ws.cell(row=r, column=7).fill = REQ_FILL
            for c in range(1, 9):
                ws.cell(row=r, column=c).border = BORDER
                ws.cell(row=r, column=c).alignment = Alignment(vertical="top", wrap_text=True)
            r += 1
    ws.freeze_panes = "A2"
    ws.auto_filter.ref = f"A1:H{r - 1}"

    # ---- 使用说明 sheet ----
    ns = wb.create_sheet("使用说明")
    ns.column_dimensions["A"].width = 118
    lines = [
        "【用途】DMS 数据 + 人工补填 → 13 类模板样式 → 知识图谱构建",
        "",
        "一、字段来源三类",
        "  ① DMS来源：DMS 模型中有对应字段（见 docs/DMS字段映射.md），由系统自动拉取，无需人工填写。",
        "  ② 源Excel搬运：源 Excel 中有、但 DMS 未建模型（考勤/排班/设备/检查/月度财务等），需人工从源文件搬运到模板。",
        "  ③ 新增融合：DMS 与源 Excel 均无，为图谱关联/可读性添加，需人工确认填写（工号、项目编号、片区编号等）。",
        "",
        "二、合并流程（如何从 DMS + 人工数据得到模板）",
        "  1. DMS 拉取：按 docs/DMS字段映射.md 的映射，从 16 个申勤模型拉取 DMS来源 字段（scripts/fetch_dms_fields.py 已打通模型定义；",
        "     拉数需有效 token，token 过期时从 DMS 控制台重新复制 vuejs_token）。",
        "  2. 人工补填：按本清单「补填清单」sheet 逐字段处理——源Excel搬运 = 从源文件复制对应列；新增融合 = 人工确认填写。",
        "     实际数据填到 data/templates 下 13 份模板文件（每份含填写说明）；也可按本清单建立录入任务分发给各部门。",
        "  3. 合并：以「与DMS对齐键」为键，将 DMS 数据行与人工补填数据行合并成完整模板行（键相同的行合并，人工列缺失则报错）。",
        "  4. 校验：graph/reader 字段名写死校验 + 必填校验 + 引用存在性校验（缺项目/人员/片区/经办人/责任人会给出警告清单）。",
        "  5. 构建：scripts/build_graph.py 读取模板文件 → Neo4j 两阶段构建（先节点后关系）。",
        "",
        "三、各模板对齐键",
    ]
    for tpl, _, _, _ in per_tpl_summary:
        lines.append(f"  {tpl}：{MERGE_KEYS[tpl]}")
    lines += [
        "",
        "四、本次需要人工补填的规模",
        f"  13 类模板共 {sum(n for _, n, _, _ in per_tpl_summary)} 个字段；DMS 自动提供 {sum(n for _, n, m, _ in per_tpl_summary) - total_manual} 个；",
        f"  需人工补填 {total_manual} 个（其中 源Excel搬运 {total_src} 个、新增融合 {total_fusion} 个）。",
        "",
        "五、特殊说明",
        "  · 考勤「税后工资」：源考勤表没有该列名，赵巷公园等文件的月度工资额填在表头为「员工签字」的列（如 4896.25），按此口径搬运。",
        "  · 考勤/排班「备注」：考勤备注来自考勤表「备注」列；排班文件无备注列，标为新增融合。",
        "  · 「1-31日」「1-31日班次」在模板中是 31 列（1日~31日），本清单合并为一行表示。",
        "  · 凡涉及人的字段必须补工号（项目负责人/检查人/投标/采购/财务经办人/设备责任人/片区负责人），避免重名。",
        "  · 续签前项目编号、上级项目编号可由 scripts/analyze_renewal_folders.py、业务大表推导后回填，人工只需核对。",
        "",
        "六、相关文件",
        "  字段映射（含 DMS 字段）：docs/DMS字段映射.md、output/dms_template_mapping.csv",
        "  13 份数据模板：data/templates/*.xlsx",
        "  本清单：data/manual_fill/人工补填清单.xlsx",
    ]
    ns.cell(row=1, column=1, value=lines[0]).font = Font(name="微软雅黑", size=12, bold=True)
    for i, n in enumerate(lines[1:], start=2):
        cell = ns.cell(row=i, column=1, value=n)
        cell.font = NOTE_FONT
        cell.alignment = Alignment(wrap_text=True, vertical="top")

    wb.save(out_path)
    print("输出:", out_path)
    print(f"需人工补填字段总数: {total_manual}（源Excel搬运 {total_src} / 新增融合 {total_fusion}）")
    for tpl, n, m, nf in per_tpl_summary:
        print(f"  {tpl}: 共{n}字段, 需人工{m}（搬运{m - nf}/新增{nf}）")


if __name__ == "__main__":
    main()