"""按最新 DMS 范围重新比对 15 类模板字段，生成补充数据清单。 最新 DMS 范围： 1. 申勤物业数字化/申勤物业项目清单、申勤员工 2. AMS枢元/申勤数字化改造（全部模型） 输出： - data/manual_fill/DMS补数字段清单.csv - output/dms_scoped_template_mapping.csv - docs/DMS字段映射_最新范围.md """ from __future__ import annotations import csv import json 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") DMS_JSON = ROOT / "output" / "dms_scoped_models_fields.json" OUT_XLSX = ROOT / "data" / "manual_fill" / "DMS补数字段清单.csv" OUT_CSV = ROOT / "output" / "dms_scoped_template_mapping.csv" OUT_DOC = ROOT / "docs" / "DMS字段映射_最新范围.md" TEMPLATES: dict[str, list[str]] = { "项目信息": ["项目编号", "项目名称", "项目简称", "上级项目编号", "续签前项目编号", "项目起止时间", "省份", "市（区）", "细分业态", "汇总业态", "委托方式", "计费方式", "合同面积", "合同金额", "年化合同额", "合同额具体信息", "服务期限", "合同状况", "首次服务日期", "年化合同收入", "客户满意度", "客户投诉", "甲方名称", "项目地址", "项目负责人", "项目负责人工号", "实际毛利率", "服务状态", "备注"], "项目月度财务": ["项目编号", "项目名称", "年月", "开票金额", "收款金额", "口径说明", "财务经办人工号", "财务经办人", "备注"], "月度快照": ["项目编号", "项目名称", "年月", "在岗人数", "排班人次", "考勤人次", "开票金额", "收款金额", "备注"], "科目余额": ["项目编号", "项目名称", "期间", "科目编码", "科目名称", "借方发生额", "贷方发生额", "期末余额", "备注"], "投标记录": ["项目编号", "项目名称", "年份", "投标类型", "中标结果", "中标日期", "投标金额", "文档路径", "投标经办人工号", "投标经办人", "备注"], "人员信息": ["姓名", "工号", "组织名称", "岗位名称", "当前服务项目编号", "服务起始年月", "服务结束年月", "职级", "员工层级", "所属组织名称", "入职日期", "离职日期", "司龄", "性别", "出生日期", "政治面貌", "联系电话", "备注"], "考勤与人员财务": ["姓名", "工号", "项目编号", "项目名称", "年月", "1-31日", "上班天数", "平时加班小时", "国定加班小时", "餐费补助金额", "加班超时费金额", "国定加班费金额", "值班费金额", "税后工资", "备注"], "排班月报": ["姓名", "工号", "项目编号", "项目名称", "年月", "1-31日班次", "备注"], "岗位编制": ["项目编号", "项目名称", "岗位", "预算编制", "项目编制", "标准工时人数", "在岗人数", "备注"], "设备信息": ["项目编号", "项目名称", "设备编号", "设备类型", "名称", "规格型号参数", "制造厂商", "安装位置", "出厂日期", "启用日期", "入项目日期", "退项目日期", "出厂编号", "完好状况", "原值", "设备责任人工号", "设备责任人", "备注"], "采购与维保": ["记录类型", "项目编号", "项目名称", "供应商名称", "供应商编号", "类别", "名称/合同名称", "数量", "单位", "单价", "金额", "合同年限", "月份/日期", "当前状态", "经办人工号", "经办人", "备注"], "检查记录": ["项目编号", "项目名称", "检查类型", "检查日期", "检查得分", "问题描述", "整改情况", "整改完成日期", "检查人工号", "检查人", "备注"], "片区信息": ["片区编号", "片区名称", "片区负责人工号", "片区负责人", "备注"], "项目片区关系": ["项目编号", "项目名称", "片区编号", "片区名称", "备注"], "人员证书": ["工号", "姓名", "专业类别", "证书类别", "证书名称", "备注"], } # 仅使用最新范围内的 DMS 模型。key = (模板, 模板字段) DMS_SOURCES: dict[tuple[str, str], str] = { ("项目信息", "项目编号"): "shenqin_project_annual_stat.c_kingdee_code（金蝶编码） / sq_shenqin_project.c_c_project_code（项目编号）", ("项目信息", "项目名称"): "shenqin_project_annual_stat.c_project_name（项目名称） / ams_sq_project.c_xmmc（项目名称）", ("项目信息", "项目简称"): "shenqin_project_annual_stat.c_project_short_name（项目简称）", ("项目信息", "项目起止时间"): "shenqin_project_annual_stat.c_contract_period（合同起止日期） / sq_project_period.c_c_date_range_text（日期区间）", ("项目信息", "省份"): "ams_sq_project.c_sfzxs（省份/直辖市）", ("项目信息", "市（区）"): "ams_sq_project.c_sq（市/区）", ("项目信息", "细分业态"): "ams_sq_project.c_xfyt（细分业态）", ("项目信息", "汇总业态"): "ams_sq_project.c_hzyt（汇总业态）", ("项目信息", "委托方式"): "ams_sq_project.c_wtfs（委托方式）", ("项目信息", "计费方式"): "ams_sq_project.c_jffs（计费方式）", ("项目信息", "合同面积"): "shenqin_project_annual_stat.c_building_area_sqm（建筑面积） / ams_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 / c_stock_amount / c_new_amount / c_estimated_amount", ("项目信息", "服务期限"): "shenqin_project_annual_stat.c_service_period（服务期限）", ("项目信息", "合同状况"): "shenqin_project_annual_stat.c_contract_status_online / c_contract_status_offline", ("项目信息", "首次服务日期"): "ams_sq_project.c_scfwrq（首次服务日期）", ("项目信息", "年化合同收入"): "ams_sq_project.c_nhhtsr（年化合同收入）", ("项目信息", "甲方名称"): "shenqin_project_annual_stat.c_client_name（甲方名称）", ("项目信息", "项目地址"): "shenqin_project_annual_stat.c_project_address（项目地址） / ams_sq_project.c_xmdz（项目地址）", ("项目信息", "项目负责人"): "ams_sq_project.c_xmdzfzr（项目点状负责人）", ("项目信息", "实际毛利率"): "ams_sq_project.c_xmdnsjmll（项目当年实际毛利率）", ("项目月度财务", "项目编号"): "shenqin_project_annual_stat.c_kingdee_code（金蝶编码） / sq_shenqin_project.c_c_project_code（项目编号）", ("项目月度财务", "项目名称"): "shenqin_project_annual_stat.c_project_name（项目名称） / ams_sq_project.c_xmmc（项目名称）", ("月度快照", "项目编号"): "shenqin_project_annual_stat.c_kingdee_code（金蝶编码） / sq_shenqin_project.c_c_project_code（项目编号）", ("月度快照", "项目名称"): "shenqin_project_annual_stat.c_project_name（项目名称） / ams_sq_project.c_xmmc（项目名称）", ("科目余额", "项目编号"): "shenqin_project_annual_stat.c_kingdee_code（金蝶编码） / sq_shenqin_project.c_c_project_code（项目编号）", ("科目余额", "项目名称"): "shenqin_project_annual_stat.c_project_name（项目名称） / ams_sq_project.c_xmmc（项目名称）", ("投标记录", "项目编号"): "shenqin_project_annual_stat.c_kingdee_code（金蝶编码） / sq_shenqin_project.c_c_project_code（项目编号）", ("投标记录", "项目名称"): "shenqin_project_annual_stat.c_project_name（项目名称） / ams_sq_project.c_xmmc（项目名称）", ("投标记录", "中标结果"): "ams_sq_project.c_zbjgrzbdrhxsj（中标结果）", ("投标记录", "中标日期"): "ams_sq_project.c_xmzbrq（项目中标日期）", ("投标记录", "文档路径"): "sq_project_file.c_c_abs_path（绝对路径） / sq_bid_archive_scan.c_c_abs_path（标书扫描绝对路径）", ("人员信息", "姓名"): "sq_employee.c_xm（姓名）", ("人员信息", "工号"): "sq_employee.c_ygbm（员工编码）", ("人员信息", "组织名称"): "sq_employee.c_sszzmc（所属组织名称）", ("人员信息", "岗位名称"): "sq_employee.c_gwmc（岗位名称）", ("人员信息", "服务起始年月"): "sq_employee.c_htksrq（合同开始日期，口径需确认）", ("人员信息", "服务结束年月"): "sq_employee.c_htjsrq（合同结束日期，口径需确认）", ("人员信息", "职级"): "sq_employee.c_zj（职级）", ("人员信息", "员工层级"): "sq_employee.c_ygcj（员工层级）", ("人员信息", "所属组织名称"): "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_bz（备注）", ("考勤与人员财务", "姓名"): "sq_employee.c_xm（姓名）", ("考勤与人员财务", "工号"): "sq_employee.c_ygbm（员工编码）", ("考勤与人员财务", "项目编号"): "shenqin_project_annual_stat.c_kingdee_code（金蝶编码） / sq_shenqin_project.c_c_project_code（项目编号）", ("考勤与人员财务", "项目名称"): "shenqin_project_annual_stat.c_project_name（项目名称） / sq_employee.c_xmmc（项目名称）", ("排班月报", "姓名"): "sq_employee.c_xm（姓名）", ("排班月报", "工号"): "sq_employee.c_ygbm（员工编码）", ("排班月报", "项目编号"): "shenqin_project_annual_stat.c_kingdee_code（金蝶编码） / sq_shenqin_project.c_c_project_code（项目编号）", ("排班月报", "项目名称"): "shenqin_project_annual_stat.c_project_name（项目名称） / sq_employee.c_xmmc（项目名称）", ("岗位编制", "项目编号"): "shenqin_project_annual_stat.c_kingdee_code（金蝶编码） / sq_shenqin_project.c_c_project_code（项目编号）", ("岗位编制", "项目名称"): "shenqin_project_annual_stat.c_project_name（项目名称） / ams_sq_project.c_xmmc（项目名称）", ("岗位编制", "岗位"): "sq_position_salary.c_c_position_name（岗位名称） / sq_employee.c_gwmc（岗位名称）", ("岗位编制", "在岗人数"): "ams_sq_project.c_xmygrs（项目员工人数，需按岗位拆分）", ("设备信息", "项目编号"): "shenqin_project_annual_stat.c_kingdee_code（金蝶编码） / sq_shenqin_project.c_c_project_code（项目编号）", ("设备信息", "项目名称"): "shenqin_project_annual_stat.c_project_name（项目名称） / ams_sq_project.c_xmmc（项目名称）", ("采购与维保", "项目编号"): "shenqin_project_annual_stat.c_kingdee_code（金蝶编码） / sq_shenqin_project.c_c_project_code（项目编号）", ("采购与维保", "项目名称"): "shenqin_project_annual_stat.c_project_name（项目名称） / ams_sq_project.c_xmmc（项目名称）", ("检查记录", "项目编号"): "shenqin_project_annual_stat.c_kingdee_code（金蝶编码） / sq_shenqin_project.c_c_project_code（项目编号）", ("检查记录", "项目名称"): "shenqin_project_annual_stat.c_project_name（项目名称） / ams_sq_project.c_xmmc（项目名称）", ("片区信息", "片区名称"): "ams_sq_project.c_glqynbkj（管理区域内部口径） / sq_shenqin_project.c_c_region（区域）", ("片区信息", "片区负责人"): "ams_sq_project.c_pjfzr（片区负责人）", ("项目片区关系", "项目编号"): "shenqin_project_annual_stat.c_kingdee_code（金蝶编码） / sq_shenqin_project.c_c_project_code（项目编号）", ("项目片区关系", "项目名称"): "shenqin_project_annual_stat.c_project_name（项目名称） / ams_sq_project.c_xmmc（项目名称）", ("项目片区关系", "片区名称"): "ams_sq_project.c_glqynbkj（管理区域内部口径） / sq_shenqin_project.c_c_region（区域）", ("人员证书", "工号"): "sq_personnel_certificate.c_c_person_id（人员ID，需与 sq_employee.c_ygbm 核对映射）", ("人员证书", "姓名"): "sq_personnel_material.c_c_name（姓名） / sq_employee.c_xm（姓名）", ("人员证书", "证书类别"): "sq_personnel_certificate.c_c_cert_type（证书类型）", } # 与 DMS/其他补充表对齐的主连接键。即使 DMS 已有，也必须在补充 Excel 中保留。 JOIN_KEYS: dict[str, set[str]] = { "项目信息": {"项目编号"}, "项目月度财务": {"项目编号", "年月"}, "月度快照": {"项目编号", "年月"}, "科目余额": {"项目编号", "期间", "科目编码"}, "投标记录": {"项目编号", "年份"}, "人员信息": {"工号"}, "考勤与人员财务": {"工号", "项目编号", "年月"}, "排班月报": {"工号", "项目编号", "年月"}, "岗位编制": {"项目编号", "岗位"}, "设备信息": {"项目编号", "设备编号"}, "采购与维保": {"项目编号", "名称/合同名称", "月份/日期"}, "检查记录": {"项目编号", "检查日期"}, "片区信息": {"片区编号"}, "项目片区关系": {"项目编号", "片区编号"}, "人员证书": {"工号", "证书名称"}, } # 非 DMS 字段中，可从已有源 Excel 搬运的字段 SOURCE_EXCEL_FIELDS: set[tuple[str, str]] = { ("项目信息", "上级项目编号"), ("项目信息", "续签前项目编号"), ("项目信息", "客户满意度"), ("项目信息", "客户投诉"), ("项目月度财务", "年月"), ("项目月度财务", "开票金额"), ("项目月度财务", "收款金额"), ("月度快照", "年月"), ("月度快照", "在岗人数"), ("月度快照", "排班人次"), ("月度快照", "考勤人次"), ("月度快照", "开票金额"), ("月度快照", "收款金额"), ("科目余额", "期间"), ("科目余额", "科目编码"), ("科目余额", "科目名称"), ("科目余额", "借方发生额"), ("科目余额", "贷方发生额"), ("科目余额", "期末余额"), ("投标记录", "年份"), ("投标记录", "投标金额"), ("人员信息", "离职日期"), ("考勤与人员财务", "年月"), ("考勤与人员财务", "1-31日"), ("考勤与人员财务", "上班天数"), ("考勤与人员财务", "平时加班小时"), ("考勤与人员财务", "国定加班小时"), ("考勤与人员财务", "餐费补助金额"), ("考勤与人员财务", "加班超时费金额"), ("考勤与人员财务", "国定加班费金额"), ("考勤与人员财务", "值班费金额"), ("考勤与人员财务", "税后工资"), ("考勤与人员财务", "备注"), ("排班月报", "年月"), ("排班月报", "1-31日班次"), ("岗位编制", "预算编制"), ("岗位编制", "项目编制"), ("岗位编制", "标准工时人数"), ("设备信息", "设备编号"), ("设备信息", "设备类型"), ("设备信息", "名称"), ("设备信息", "规格型号参数"), ("设备信息", "制造厂商"), ("设备信息", "安装位置"), ("设备信息", "出厂日期"), ("设备信息", "启用日期"), ("设备信息", "出厂编号"), ("设备信息", "完好状况"), ("设备信息", "原值"), ("采购与维保", "供应商名称"), ("采购与维保", "类别"), ("采购与维保", "名称/合同名称"), ("采购与维保", "数量"), ("采购与维保", "单位"), ("采购与维保", "单价"), ("采购与维保", "金额"), ("采购与维保", "合同年限"), ("采购与维保", "月份/日期"), ("采购与维保", "当前状态"), ("检查记录", "检查类型"), ("检查记录", "检查日期"), ("检查记录", "检查得分"), ("检查记录", "问题描述"), ("检查记录", "整改情况"), ("检查记录", "整改完成日期"), ("人员证书", "专业类别"), ("人员证书", "证书名称"), } DEFAULT_SOURCES: dict[str, str] = { "项目信息": "市场部项目管理表 / 业务大表", "项目月度财务": "财务部收费进程表（1-12月物业费/期内开票收款）", "月度快照": "由考勤/排班/项目月度财务汇总生成（可参考 scripts/sample_snapshot.py）", "科目余额": "财务部科目余额表（科目编码/名称/借贷发生额/期末余额）", "投标记录": "业务大表中标情况 / 档案目录", "人员信息": "人事部员工花名册", "考勤与人员财务": "人事部考勤表（13份考勤 xls/xlsx）", "排班月报": "人事部排班文件（每日班次代码）", "岗位编制": "运营部 9-项目岗位编制汇总.xlsx", "设备信息": "运营部设备台账（青浦工业园区/徐汇环境监测中心）", "采购与维保": "采购部耗材汇总/维保合同台账/固定资产台账", "检查记录": "运营部品质巡检及整改记录台账", "片区信息": "业务大表「管理区域」/ 8-项目信息汇总表", "项目片区关系": "业务大表「管理区域」推导", "人员证书": "合景悠活职称及资格证书管理登记表", } SPECIAL_NOTES: dict[tuple[str, str], str] = { ("项目信息", "项目负责人工号"): "DMS 有项目负责人姓名；工号需按姓名在员工花名册中匹配", ("项目信息", "上级项目编号"): "可由业务大表层级/委托方式推导后回填", ("项目信息", "续签前项目编号"): "可由续签链/档案目录推导后回填", ("人员信息", "当前服务项目编号"): "DMS sq_employee 只有项目名称/GUID，编号需按项目名称映射", ("人员信息", "离职日期"): "DMS 只有合同结束日期，不是离职日期，需人工补", ("月度快照", "在岗人数"): "优先按考勤明细不同工号数统计，并与岗位编制在岗人数核对", ("月度快照", "排班人次"): "排班月报按项目+年月汇总行数", ("月度快照", "考勤人次"): "考勤明细按项目+年月汇总行数", ("科目余额", "科目编码"): "财务科目编码，同一项目同一期间内唯一", ("科目余额", "科目名称"): "与科目编码配套的中文科目名称", ("考勤与人员财务", "税后工资"): "源考勤表无该列名，部分文件金额填在表头为「员工签字」的列", ("考勤与人员财务", "1-31日"): "值为 常/日/夜/日1 等班次标记，原样搬运", ("排班月报", "1-31日班次"): "值为班次代码，如 OFF/A003/SQ-104/M11-033；OFF=休息", ("设备信息", "设备责任人工号"): "模板必填；源台账可能有责任人姓名，工号需匹配花名册", ("采购与维保", "供应商编号"): "用于关联供应商主数据；DMS 范围内无供应商模型，需新编或另建", ("片区信息", "片区编号"): "最新 DMS 范围内没有 xzbm 片区模型，需人工统一编码", ("项目片区关系", "片区编号"): "同上，需与片区信息表保持一致", ("人员证书", "证书名称"): "DMS 证书表只有证书类型/证书号掩码，证书全称需补充", } def norm_header(h: str, tpl: str) -> str: h = str(h or "").strip() h = re.sub(r"\*$", "", h) h = re.sub(r"（[^）]*）$", "", h) if re.fullmatch(r"\d{1,2}日", h): return "1-31日班次" if tpl == "排班月报" else "1-31日" return h def read_required() -> dict[str, set[str]]: required: dict[str, set[str]] = {} for f in sorted((ROOT / "data" / "templates").glob("*.csv")): tpl = f.stem req: set[str] = set() with f.open("r", encoding="utf-8-sig", newline="") as fh: reader = csv.reader(fh) header = next(reader, []) for h in header: if h and "*" in str(h): req.add(norm_header(str(h), tpl)) required[tpl] = req return required def build_rows() -> list[dict]: required = read_required() rows: list[dict] = [] for tpl, fields in TEMPLATES.items(): req = required.get(tpl, set()) for i, fld in enumerate(fields, start=1): has_dms = (tpl, fld) in DMS_SOURCES is_join = fld in JOIN_KEYS.get(tpl, set()) include = (not has_dms) or is_join if has_dms and not is_join: kind = "DMS已有" elif has_dms and is_join: kind = "连接字段（DMS已有，Excel需带）" elif is_join: kind = "连接字段（需补充）" elif (tpl, fld) in SOURCE_EXCEL_FIELDS: kind = "业务字段（源Excel补充）" else: kind = "业务字段（新增/人工确认）" note = SPECIAL_NOTES.get((tpl, fld), "") if not note and not has_dms and (tpl, fld) in SOURCE_EXCEL_FIELDS: note = DEFAULT_SOURCES.get(tpl, "") elif not note and not has_dms: note = "DMS 范围内无对应字段，需人工确认" elif not note and has_dms: note = DMS_SOURCES[(tpl, fld)] rows.append({ "template": tpl, "seq": i, "field": fld, "has_dms": has_dms, "dms_source": DMS_SOURCES.get((tpl, fld), ""), "kind": kind, "is_join": is_join, "include": include, "required": "是" if fld in req else "否", "note": note, }) return rows def write_xlsx(rows: list[dict]) -> None: wb = Workbook() header_fill = PatternFill("solid", fgColor="2F5597") header_font = Font(name="微软雅黑", size=10, bold=True, color="FFFFFF") body_font = Font(name="微软雅黑", size=10) thin = Side(style="thin", color="BFBFBF") border = Border(left=thin, right=thin, top=thin, bottom=thin) # Sheet 1：全量字段比对 ws = wb.active ws.title = "字段比对总表" headers = ["模板", "序号", "模板字段", "DMS是否有", "DMS来源(模型.字段)", "补充类型", "是否连接字段", "是否需放入补充Excel", "模板必填", "数据来源/填报说明"] widths = [16, 6, 18, 9, 46, 24, 11, 16, 8, 52] 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 r = 2 for row in rows: vals = [ row["template"], row["seq"], row["field"], "是" if row["has_dms"] else "否", row["dms_source"], row["kind"], "是" if row["is_join"] else "否", "是" if row["include"] else "否", row["required"], row["note"], ] for c, v in enumerate(vals, start=1): cell = ws.cell(row=r, column=c, value=v) cell.font = body_font cell.border = border cell.alignment = Alignment(vertical="top", wrap_text=True) if row["kind"].startswith("连接字段"): ws.cell(row=r, column=6).fill = PatternFill("solid", fgColor="FFF2CC") elif row["has_dms"]: ws.cell(row=r, column=6).fill = PatternFill("solid", fgColor="EDEDED") else: ws.cell(row=r, column=6).fill = PatternFill("solid", fgColor="FCE4D6") if row["required"] == "是": ws.cell(row=r, column=9).fill = PatternFill("solid", fgColor="FCE4D6") r += 1 ws.freeze_panes = "A2" ws.auto_filter.ref = f"A1:J{r - 1}" # Sheet 2：待补充字段清单 ws2 = wb.create_sheet("待补充字段清单") for c, (h, w) in enumerate(zip(headers, widths), start=1): cell = ws2.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 ws2.column_dimensions[get_column_letter(c)].width = w ws2.row_dimensions[1].height = 28 r2 = 2 for row in rows: if not row["include"]: continue vals = [ row["template"], row["seq"], row["field"], "是" if row["has_dms"] else "否", row["dms_source"], row["kind"], "是" if row["is_join"] else "否", "是", row["required"], row["note"], ] for c, v in enumerate(vals, start=1): cell = ws2.cell(row=r2, column=c, value=v) cell.font = body_font cell.border = border cell.alignment = Alignment(vertical="top", wrap_text=True) if row["kind"].startswith("连接字段"): ws2.cell(row=r2, column=6).fill = PatternFill("solid", fgColor="FFF2CC") else: ws2.cell(row=r2, column=6).fill = PatternFill("solid", fgColor="FCE4D6") r2 += 1 ws2.freeze_panes = "A2" ws2.auto_filter.ref = f"A1:J{r2 - 1}" # Sheet 3：连接字段清单 ws3 = wb.create_sheet("连接字段清单") h3 = ["模板", "连接字段", "DMS是否有", "DMS来源", "连接说明"] w3 = [16, 20, 9, 46, 46] for c, (h, w) in enumerate(zip(h3, w3), start=1): cell = ws3.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 ws3.column_dimensions[get_column_letter(c)].width = w ws3.row_dimensions[1].height = 24 r3 = 2 for tpl, keys in JOIN_KEYS.items(): for k in sorted(keys): has_dms = (tpl, k) in DMS_SOURCES src = DMS_SOURCES.get((tpl, k), "") note = f"{tpl} 补充表与 DMS 数据按该字段对齐" vals = [tpl, k, "是" if has_dms else "否", src, note] for c, v in enumerate(vals, start=1): cell = ws3.cell(row=r3, column=c, value=v) cell.font = body_font cell.border = border cell.alignment = Alignment(vertical="top", wrap_text=True) if not has_dms: ws3.cell(row=r3, column=3).fill = PatternFill("solid", fgColor="FCE4D6") r3 += 1 ws3.freeze_panes = "A2" ws3.auto_filter.ref = f"A1:E{r3 - 1}" # Sheet 4：统计汇总 ws4 = wb.create_sheet("统计汇总") ws4.column_dimensions["A"].width = 34 ws4.column_dimensions["B"].width = 12 ws4.column_dimensions["C"].width = 52 total = len(rows) dms_count = sum(1 for x in rows if x["has_dms"]) missing_count = total - dms_count source_excel_count = sum(1 for x in rows if (not x["has_dms"]) and (x["template"], x["field"]) in SOURCE_EXCEL_FIELDS) new_count = missing_count - source_excel_count join_count = sum(1 for x in rows if x["is_join"]) join_dms = sum(1 for x in rows if x["is_join"] and x["has_dms"]) join_missing = join_count - join_dms supplement_count = missing_count + join_dms summary = [ ("统计项", "数值", "说明"), ("最新范围 DMS 模型数", 24, "申勤物业项目清单/申勤员工 2 个 + AMS枢元/申勤数字化改造 22 个"), ("15 类模板字段总数", total, "1-31日 按 1 个字段统计（实际是 31 列）"), ("DMS 已有字段数", dms_count, "可从本次范围内 DMS 模型自动获取"), ("DMS 无字段数（需业务补充）", missing_count, "DMS 范围内无对应字段"), ("其中：源Excel可搬运", source_excel_count, "已有源 Excel，但 DMS 未建模型，需搬运"), ("其中：需新增/人工确认", new_count, "DMS 与源 Excel 均无，需人工确认/新建"), ("连接字段数", join_count, "用于补充 Excel 与 DMS/其他表对齐"), ("其中：DMS已有但Excel需带", join_dms, "如项目编号、工号；补充表也必须保留"), ("其中：连接字段需补充", join_missing, "如年月、设备编号、片区编号等"), ("补充 Excel 需包含字段总数", supplement_count, "= DMS 无字段数 + DMS 已有的连接字段数"), ] for r, row in enumerate(summary, start=1): for c, v in enumerate(row, start=1): cell = ws4.cell(row=r, column=c, value=v) cell.border = border cell.alignment = Alignment(vertical="top", wrap_text=True) if r == 1: cell.font = header_font cell.fill = header_fill else: cell.font = body_font if c == 1: cell.font = Font(name="微软雅黑", size=10, bold=True) ws4.freeze_panes = "A2" wb.save(OUT_XLSX) def write_csv(rows: list[dict]) -> None: with OUT_CSV.open("w", newline="", encoding="utf-8-sig") as f: w = csv.writer(f) w.writerow(["模板", "序号", "模板字段", "DMS是否有", "DMS来源(模型.字段)", "补充类型", "是否连接字段", "是否需放入补充Excel", "模板必填", "数据来源/填报说明"]) for row in rows: w.writerow([ row["template"], row["seq"], row["field"], "是" if row["has_dms"] else "否", row["dms_source"], row["kind"], "是" if row["is_join"] else "否", "是" if row["include"] else "否", row["required"], row["note"], ]) def write_markdown(rows: list[dict]) -> None: total = len(rows) dms_count = sum(1 for x in rows if x["has_dms"]) missing_count = total - dms_count source_excel_count = sum(1 for x in rows if (not x["has_dms"]) and (x["template"], x["field"]) in SOURCE_EXCEL_FIELDS) new_count = missing_count - source_excel_count join_count = sum(1 for x in rows if x["is_join"]) join_dms = sum(1 for x in rows if x["is_join"] and x["has_dms"]) join_missing = join_count - join_dms supplement_count = missing_count + join_dms lines = [ "# DMS 数据库字段映射（最新范围）", "", "> 依据本次指定的 DMS 数据范围重新生成：", "> 1. 申勤物业数字化：申勤物业项目清单、申勤员工", "> 2. AMS枢元 -- 申勤数字化改造（全部模型）", "", "## 总览", "", f"- 范围内 DMS 模型数：24", f"- 15 类模板字段总数：{total}", f"- DMS 已有字段：{dms_count}", f"- DMS 无字段（需业务补充）：{missing_count}（源Excel可搬运 {source_excel_count} / 需新增或人工确认 {new_count}）", f"- 连接字段：{join_count}（DMS已有但 Excel 需带 {join_dms} / 需补充 {join_missing}）", f"- 补充 Excel 需包含字段总数：{supplement_count}", "", "## 覆盖概览", "", "| 模板 | 字段数 | DMS已有 | DMS无 | 需放入补充Excel |", "|---|---:|---:|---:|---:|", ] for tpl, fields in TEMPLATES.items(): trows = [x for x in rows if x["template"] == tpl] has = sum(1 for x in trows if x["has_dms"]) inc = sum(1 for x in trows if x["include"]) lines.append(f"| {tpl} | {len(fields)} | {has} | {len(fields) - has} | {inc} |") lines += ["", "## 字段级映射", ""] for tpl, fields in TEMPLATES.items(): lines.append(f"### {tpl}") lines.append("| 模板字段 | DMS是否有 | DMS来源 | 补充类型 | 连接字段 | 说明 |") lines.append("|---|---|---|---|---|---|") for x in rows: if x["template"] != tpl: continue lines.append( f"| {x['field']} | {'是' if x['has_dms'] else '否'} | " f"{x['dms_source'] or '—'} | {x['kind']} | " f"{'是' if x['is_join'] else '否'} | {x['note']} |" ) lines.append("") OUT_DOC.write_text("\n".join(lines), encoding="utf-8") def main() -> None: rows = build_rows() OUT_XLSX.parent.mkdir(parents=True, exist_ok=True) write_csv(rows) OUT_XLSX.write_bytes(OUT_CSV.read_bytes()) write_markdown(rows) print("输出:", OUT_XLSX) print("输出:", OUT_CSV) print("输出:", OUT_DOC) if __name__ == "__main__": main()