"""生成 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()