kg-agent/scripts/generate_dms_mapping.py

"""生成 13 类模板字段 ↔ DMS 模型字段 映射文档。

来源分三类：
1. DMS来源                 —— DMS 模型中有对应字段（模型.字段）
2. 源Excel来源（DMS未建模型） —— 源 Excel 中确实存在，但 DMS 没有建模型，
                               需人工从源 Excel 填入模板（标注具体文件+列）
3. 新增融合（图谱关联字段）    —— DMS 与源 Excel 均无，为图谱建模/关联额外添加
"""

from __future__ import annotations

import json
import re
import sys
from pathlib import Path


sys.stdout.reconfigure(encoding="utf-8")
OUT = Path(r"E:\CODE\knowledge_agent\output\dms_models_fields.json")
DOC = Path(r"E:\CODE\knowledge_agent\docs\DMS字段映射.md")


def norm(s: str) -> str:
    return re.sub(r"[\s（）()/*、，,：:元万㎡条%]", "", s or "")


TEMPLATES: dict[str, list[str]] = {
    "项目信息": ["项目编号", "项目名称", "项目简称", "上级项目编号", "续签前项目编号", "项目起止时间",
               "省份", "市（区）", "细分业态", "汇总业态", "委托方式", "计费方式", "合同面积",
               "合同金额", "年化合同额", "合同额具体信息", "服务期限", "合同状况", "首次服务日期",
               "年化合同收入", "客户满意度", "客户投诉", "甲方名称", "项目地址", "项目负责人",
               "项目负责人工号", "实际毛利率", "服务状态", "备注"],
    "项目月度财务": ["项目编号", "项目名称", "年月", "开票金额", "收款金额", "口径说明",
                  "财务经办人工号", "财务经办人", "备注"],
    "投标记录": ["项目编号", "项目名称", "年份", "投标类型", "中标结果", "中标日期", "投标金额",
               "文档路径", "投标经办人工号", "投标经办人", "备注"],
    "人员信息": ["姓名", "工号", "组织名称", "岗位名称", "当前服务项目编号", "服务起始年月",
               "服务结束年月", "职级", "员工层级", "所属组织名称", "入职日期", "离职日期", "司龄",
               "性别", "出生日期", "政治面貌", "联系电话", "备注"],
    "考勤与人员财务": ["姓名", "工号", "项目编号", "项目名称", "年月", "1-31日", "上班天数",
                     "平时加班小时", "国定加班小时", "餐费补助金额", "加班超时费金额",
                     "国定加班费金额", "值班费金额", "税后工资", "备注"],
    "排班明细": ["姓名", "工号", "项目编号", "项目名称", "年月", "1-31日班次", "备注"],
    "岗位编制": ["项目编号", "项目名称", "岗位", "预算编制", "项目编制", "标准工时人数", "在岗人数", "备注"],
    "设备信息": ["项目编号", "项目名称", "设备编号", "设备类型", "名称", "规格型号参数", "制造厂商",
               "安装位置", "出厂日期", "启用日期", "入项目日期", "退项目日期", "出厂编号", "完好状况",
               "原值", "设备责任人工号", "设备责任人", "备注"],
    "采购与维保": ["记录类型", "项目编号", "项目名称", "供应商名称", "供应商编号", "类别",
                 "名称/合同名称", "数量", "单位", "单价", "金额", "合同年限", "月份/日期",
                 "当前状态", "经办人工号", "经办人", "备注"],
    "检查记录": ["项目编号", "项目名称", "检查类型", "检查日期", "检查得分", "问题描述",
               "整改情况", "整改完成日期", "检查人工号", "检查人", "备注"],
    "片区信息": ["片区编号", "片区名称", "片区负责人工号", "片区负责人", "备注"],
    "项目片区关系": ["项目编号", "项目名称", "片区编号", "片区名称", "备注"],
    "人员证书": ["工号", "姓名", "专业类别", "证书类别", "证书名称", "备注"],
}

NOTES: dict[str, str] = {
    "项目信息": "专用模型：sq_project(申勤项目数据) + shenqin_project_annual_stat(年化统计) + sq_project_period(项目期次) + sq_xq(续签表)",
    "人员信息": "专用模型：sq_employee(申勤员工)",
    "投标记录": "专用模型：shenqin_txb(投续标表) + shenqin_zbb(招标表) + sq_tender_intent/notice/result(招标意向/公告/结果)",
    "片区信息": "专用模型：xzbm(行政编码及片区)",
    "项目片区关系": "无专用关联表：可基于 sq_project.管理区域 / xzbm 行政编码 推导",
    "人员证书": "专用模型：sq_personnel_certificate(AMS枢元/人员证书)",
    "考勤与人员财务": "DMS 中未发现考勤专用模型（姓名/工号等为跨模型公共字段）",
    "排班明细": "DMS 中未发现排班专用模型",
    "项目月度财务": "DMS 无月度开票/收款明细（仅年化/合同金额）",
    "岗位编制": "DMS 无编制/在岗模型（sq_project 仅有定岗人数，为预算口径）",
    "设备信息": "DMS 中未发现设备台账模型",
    "检查记录": "DMS 中未发现品质/安全检查模型",
    "采购与维保": "DMS 中未发现采购/维保模型",
}

# 有专用 DMS 模型的模板（其余模板按"源Excel来源"处理）
MODELED = {"项目信息", "人员信息", "投标记录", "片区信息", "人员证书"}

# 字段的源 Excel 说明（DMS 未建模型时）
DEFAULT_SOURCES: dict[str, str] = {
    "项目信息": "市场部项目管理表 + 业务大表（表1）",
    "项目月度财务": "财务部收费进程表（1-12月物业费/期内开票收款）",
    "投标记录": "业务大表表2 中标情况 + 档案目录（中标通知书等）",
    "人员信息": "人事部员工信息.xlsx（花名册）",
    "考勤与人员财务": "人事部考勤表（如 青浦-赵巷公园-员工岗位考勤表.xls）",
    "排班明细": "人事部排班文件（每日班次代码）",
    "岗位编制": "运营部 9-项目岗位编制汇总.xlsx",
    "设备信息": "运营部设备台账（青浦工业园区/徐汇环境监测中心）",
    "采购与维保": "采购部 耗材汇总/维保合同台账/固定资产台账",
    "检查记录": "运营部品质巡检及整改记录台账",
    "片区信息": "业务大表「管理区域」/ 8-项目信息汇总表",
    "项目片区关系": "业务大表「管理区域」推导（M:N）",
    "人员证书": "合景悠活职称及资格证书管理登记表",
}

# 源 Excel 字段级说明（DMS 未建模型时，标注到具体文件+列）
SOURCE_EXCEL_NOTES: dict[tuple[str, str], str] = {
    ("考勤与人员财务", "工号"): "人事部考勤表「员工编码」列（13 份考勤 xls/xlsx）",
    ("考勤与人员财务", "年月"): "考勤表工作表名/表头（如 青浦-赵巷公园-员工岗位考勤表2026年1月-4月员工岗位考勤表.xls →「2026年×月」）",
    ("考勤与人员财务", "1-31日"): "考勤表 1~31 日列，班次标记（常/日/夜/日1 等）",
    ("考勤与人员财务", "上班天数"): "考勤表「上班天数」列",
    ("考勤与人员财务", "平时加班小时"): "考勤表「平时加班小时」列",
    ("考勤与人员财务", "国定加班小时"): "考勤表「国定加班小时」列",
    ("考勤与人员财务", "餐费补助金额"): "考勤表「餐费补助金额」列",
    ("考勤与人员财务", "加班超时费金额"): "考勤表「加班超时费金额」列",
    ("考勤与人员财务", "国定加班费金额"): "考勤表「国定加班费金额」列",
    ("考勤与人员财务", "值班费金额"): "考勤表「值班费金额」列",
    ("考勤与人员财务", "税后工资"): "考勤表末尾金额列（表头为「员工签字」，实际填月度税后工资，如 4896.25）",
    ("考勤与人员财务", "备注"): "考勤表「备注」列",
    ("排班明细", "工号"): "人事部排班 xlsx「工号(Employee ID)」列",
    ("排班明细", "年月"): "排班 xlsx 日期列（2026-01-01 … 2026-01-31），如 排班：青浦-赵巷公园、赵巷文体中心、赵巷镇政府、赵巷品牌公司系统排班（2026年1-4月）.xlsx",
    ("排班明细", "1-31日班次"): "排班 xlsx 1~31 日「班次代码(Shift Code)」列（OFF/A003/SQ-104/M11-033 等）",
    ("岗位编制", "岗位"): "运营部 9-项目岗位编制汇总.xlsx",
    ("设备信息", "设备编号"): "运营部设备台账（如 青浦工业园区/徐汇环境监测中心）",
    ("设备信息", "设备类型"): "运营部设备台账（如 青浦工业园区/徐汇环境监测中心）",
    ("设备信息", "规格型号参数"): "运营部设备台账（如 青浦工业园区/徐汇环境监测中心）",
    ("设备信息", "制造厂商"): "运营部设备台账（如 青浦工业园区/徐汇环境监测中心）",
    ("设备信息", "安装位置"): "运营部设备台账（如 青浦工业园区/徐汇环境监测中心）",
    ("设备信息", "出厂日期"): "运营部设备台账（如 青浦工业园区/徐汇环境监测中心）",
    ("设备信息", "启用日期"): "运营部设备台账（如 青浦工业园区/徐汇环境监测中心）",
    ("设备信息", "出厂编号"): "运营部设备台账（如 青浦工业园区/徐汇环境监测中心）",
    ("设备信息", "完好状况"): "运营部设备台账（如 青浦工业园区/徐汇环境监测中心）",
    ("设备信息", "原值"): "运营部设备台账（如 青浦工业园区/徐汇环境监测中心）",
    ("检查记录", "检查类型"): "运营部品质巡检及整改记录台账",
    ("检查记录", "检查日期"): "运营部品质巡检及整改记录台账",
    ("检查记录", "检查得分"): "运营部品质巡检及整改记录台账",
    ("检查记录", "问题描述"): "运营部品质巡检及整改记录台账",
    ("检查记录", "整改情况"): "运营部品质巡检及整改记录台账",
    ("检查记录", "整改完成日期"): "运营部品质巡检及整改记录台账",
    ("项目月度财务", "年月"): "财务部收费进程表（1-12月物业费/期内开票收款）",
    ("项目月度财务", "开票金额"): "财务部收费进程表（1-12月物业费/期内开票收款）",
    ("项目月度财务", "收款金额"): "财务部收费进程表（1-12月物业费/期内开票收款）",
    ("项目信息", "客户满意度"): "运营部\\03 客户满意度及投诉记录\\客户满意度\\申勤物业服务满意度评价问卷报告（2025年度）.docx → 满意度得分（聚合为项目属性）",
    ("项目信息", "客户投诉"): "运营部\\03 客户满意度及投诉记录\\客户投诉\\SHSQ-JGJL-02-0301客户投诉受理单.doc 等 → 投诉条数",
}

# 真正新增融合的字段（DMS 与源 Excel 均无，为图谱建模/关联添加）
FUSION_FIELDS: dict[str, set[str]] = {
    "项目信息": {"项目负责人工号", "服务状态", "续签前项目编号"},
    "项目月度财务": {"项目名称", "口径说明", "财务经办人工号", "财务经办人"},
    "投标记录": {"项目名称", "投标类型", "投标经办人工号", "投标经办人", "中标日期", "文档路径"},
    "人员信息": {"当前服务项目编号", "服务起始年月", "服务结束年月", "离职日期"},
    "考勤与人员财务": {"项目名称"},
    "排班明细": {"项目名称", "备注"},
    "岗位编制": {"项目名称"},
    "设备信息": {"项目名称", "入项目日期", "退项目日期", "设备责任人工号", "设备责任人"},
    "采购与维保": {"记录类型", "项目名称", "供应商编号", "经办人工号", "经办人"},
    "检查记录": {"项目名称", "检查人工号", "检查人"},
    "片区信息": {"片区编号", "片区负责人工号", "片区负责人"},
    "项目片区关系": {"项目名称", "片区名称"},
    "人员证书": {"姓名", "备注"},
}

# 不参与 DMS 别名匹配的字段（避免被误匹配到跨模型公共字段，如 sq_employee.c_bz）
EXCLUDE_DMS: set[tuple[str, str]] = {
    ("考勤与人员财务", "备注"),
    ("排班明细", "备注"),
}

# 语义映射覆盖：模板字段 -> (模型, DMS字段)
OVERRIDES: dict[tuple[str, str], tuple[str, str]] = {
    ("人员信息", "工号"): ("sq_employee", "c_ygbm 员工编码"),
    ("人员信息", "姓名"): ("sq_employee", "c_xm 姓名"),
    ("人员信息", "岗位名称"): ("sq_employee", "c_gwmc 岗位名称"),
    ("人员信息", "员工层级"): ("sq_employee", "c_ygcj 员工层级"),
    ("人员信息", "职级"): ("sq_employee", "c_zj 职级"),
    ("人员信息", "所属组织名称"): ("sq_employee", "c_sszzmc 所属组织名称"),
    ("人员信息", "入职日期"): ("sq_employee", "c_rzrq 入职日期"),
    ("人员信息", "司龄"): ("sq_employee", "c_sl 司龄"),
    ("人员信息", "性别"): ("sq_employee", "c_xb 性别"),
    ("人员信息", "出生日期"): ("sq_employee", "c_csrq 出生日期"),
    ("人员信息", "政治面貌"): ("sq_employee", "c_zzmm 政治面貌"),
    ("人员信息", "联系电话"): ("sq_employee", "c_sjhm 手机号码"),
    ("人员信息", "当前服务项目编号"): ("sq_employee", "c_xmmc 项目名称(五级公司)"),
    ("项目信息", "项目编号"): ("sq_project", "c_jdbm 金蝶编码"),
    ("项目信息", "项目名称"): ("sq_project", "c_xmmc 项目名称"),
    ("项目信息", "上级项目编号"): ("sq_project", "c_father_code 父项目编码"),
    ("项目信息", "省份"): ("sq_project", "c_sfzxs 省份（直辖市）"),
    ("项目信息", "市（区）"): ("sq_project", "c_sq 市（区）"),
    ("项目信息", "管理区域"): ("sq_project", "c_glqynbkj 管理区域（内部口径）"),
    ("项目信息", "细分业态"): ("sq_project", "c_xfyt 细分业态"),
    ("项目信息", "汇总业态"): ("sq_project", "c_hzyt 汇总业态"),
    ("项目信息", "委托方式"): ("sq_project", "c_wtfs 委托方式"),
    ("项目信息", "计费方式"): ("sq_project", "c_jffs 计费方式"),
    ("项目信息", "合同面积"): ("sq_project", "c_htmj 合同面积（平方米）"),
    ("项目信息", "合同金额"): ("shenqin_project_annual_stat", "c_contract_amount 合同金额"),
    ("项目信息", "年化合同额"): ("shenqin_project_annual_stat", "c_annualized_amount 年化合同额"),
    ("项目信息", "合同额具体信息"): ("shenqin_project_annual_stat", "c_current_period_amount/预估/存量/新增"),
    ("项目信息", "服务期限"): ("shenqin_project_annual_stat", "c_service_period 服务期限"),
    ("项目信息", "合同状况"): ("shenqin_project_annual_stat", "c_contract_status_online/offline"),
    ("项目信息", "首次服务日期"): ("sq_project", "c_scfwrq 首次服务日期"),
    ("项目信息", "年化合同收入"): ("sq_project", "c_nhhtsr 年化合同收入"),
    ("项目信息", "甲方名称"): ("shenqin_project_annual_stat", "c_client_name 甲方名称"),
    ("项目信息", "项目地址"): ("shenqin_project_annual_stat", "c_project_address 项目地址"),
    ("项目信息", "项目负责人"): ("sq_project", "c_xmdzfzr 项目点状负责人"),
    ("项目信息", "实际毛利率"): ("sq_project", "c_xmdnsjmll 项目当年实际毛利率"),
    ("项目信息", "项目起止时间"): ("sq_project_period", "c_c_date_range_text 日期区间"),
    ("项目信息", "项目简称"): ("shenqin_project_annual_stat", "c_project_short_name 项目简称"),
    ("投标记录", "项目编号"): ("shenqin_txb", "c_jdbm 金蝶编码"),
    ("投标记录", "投标金额"): ("shenqin_txb", "tbbj 投标报价"),
    ("投标记录", "中标结果"): ("sq_tender_result", "c_winning_suppliers 中标供应商名称"),
    ("投标记录", "年份"): ("sq_tender_notice", "c_publish_time 发布时间"),
    ("片区信息", "片区编号"): ("xzbm", "c_part_code 片区编码"),
    ("片区信息", "片区名称"): ("xzbm", "c_part_name 片区名称"),
    ("人员证书", "工号"): ("sq_personnel_certificate", "c_c_person_id 人员ID"),
    ("人员证书", "证书类别"): ("sq_personnel_certificate", "c_c_cert_type 证书类型"),
}


def load_dms() -> dict[str, dict[str, str]]:
    data = json.loads(OUT.read_text(encoding="utf-8"))
    aliases: dict[str, str] = {}
    for model, info in data.items():
        for fname, alias in info.get("fields", []):
            key = norm(alias)
            if key:
                aliases.setdefault(key, f"{model}.{fname}（{alias}）")
    return aliases


def main() -> None:
    aliases = load_dms()
    lines = ["# DMS 数据库字段映射（13 类模板）", "",
             "> 依据 DMS 模型元数据生成（模型：申勤物业数字化 / AMS枢元 相关）。"
             "来源分三类：① DMS来源 = 模型.字段（别名）；② 源Excel来源（DMS未建模型）"
             "= 源 Excel 中确有、但 DMS 无对应模型，需人工从源 Excel 填入；"
             "③ 新增融合 = DMS 与源 Excel 均无，为图谱建模/关联额外添加。", ""]
    summary = []
    for tpl, fields in TEMPLATES.items():
        lines.append(f"## {tpl}")
        lines.append(f"**说明**：{NOTES.get(tpl, '')}")
        lines.append("| 模板字段 | 来源（DMS 模型.字段 / 源Excel文件+列） | 类型 |")
        lines.append("|---|---|---|")
        found = 0
        for f in fields:
            ov = OVERRIDES.get((tpl, f))
            src = ""
            note = ""
            if ov:
                src = f"{ov[0]}.{ov[1]}"
                found += 1
            else:
                if (tpl, f) not in EXCLUDE_DMS:
                    for alias, desc in aliases.items():
                        if norm(f) == alias or (len(norm(f)) >= 2 and norm(f) in alias):
                            src = desc
                            found += 1
                            break
            if src:
                kind = "DMS来源"
            elif f in FUSION_FIELDS.get(tpl, set()):
                kind = "新增融合（图谱关联字段）"
                note = "DMS 与源 Excel 均无，图谱建模/关联用"
            else:
                kind = "源Excel来源（DMS未建模型）"
                note = SOURCE_EXCEL_NOTES.get((tpl, f)) or DEFAULT_SOURCES.get(tpl, "")
            if src:
                display = src
            elif kind.startswith("源Excel"):
                display = f"源Excel：{note}" if note else "—"
            else:
                display = note or "—"
            lines.append(f"| {f} | {display} | {kind} |")
        summary.append((tpl, len(fields), found))
        lines.append("")
    lines.insert(3, "## 覆盖概览")
    lines.insert(4, "| 模板 | 字段数 | 有DMS来源 |")
    lines.insert(5, "|---|---|---|")
    for t, n, f in summary:
        lines.insert(6, f"| {t} | {n} | {f} |")
    DOC.write_text("\n".join(lines), encoding="utf-8")

    import csv
    csv_path = Path(r"E:\CODE\knowledge_agent\output\dms_template_mapping.csv")
    with csv_path.open("w", newline="", encoding="utf-8-sig") as f:
        w = csv.writer(f)
        w.writerow(["模板", "模板字段", "DMS来源(模型.字段)", "类型", "说明"])
        for tpl, fields in TEMPLATES.items():
            for fld in fields:
                ov = OVERRIDES.get((tpl, fld))
                src = ""
                if ov:
                    src = f"{ov[0]}.{ov[1]}"
                else:
                    if (tpl, fld) not in EXCLUDE_DMS:
                        for alias, desc in aliases.items():
                            if norm(fld) == alias or (len(norm(fld)) >= 2 and norm(fld) in alias):
                                src = desc
                                break
                if src:
                    kind = "DMS来源"
                    note = NOTES.get(tpl, "")
                elif fld in FUSION_FIELDS.get(tpl, set()):
                    kind = "新增融合"
                    note = "图谱建模/关联字段"
                else:
                    kind = "源Excel来源（DMS未建模型）"
                    note = SOURCE_EXCEL_NOTES.get((tpl, fld)) or DEFAULT_SOURCES.get(tpl, "")
                w.writerow([tpl, fld, src, kind, note])
    print("覆盖概览：")
    for t, n, f in summary:
        print(f"  {t}: {f}/{n}")
    print("文档输出:", DOC)
    print("CSV 输出:", csv_path)


if __name__ == "__main__":
    main()