# -*- coding: utf-8 -*- """按原 xlsx 格式补全业务表字段类型与描述，输出到桌面与 docs。""" import json from openpyxl import Workbook, load_workbook from openpyxl.styles import Font, Alignment, Border, Side, PatternFill from openpyxl.utils import get_column_letter from pathlib import Path ROOT = Path(r"c:\Users\Liumouren\Desktop\临时文件\元以科技\松江\松江农民退养补贴") SRC = ROOT / "docs" / "松江农民土地退补系统字段名称.xlsx" FRESH = ROOT / "docs" / "_dms_fields_fresh_20260904.json" OUT_DESKTOP = Path(r"c:\Users\Liumouren\Desktop") / "松江农民土地退补系统字段名称.xlsx" OUT_DOCS = ROOT / "docs" / "松江农民土地退补系统字段名称.xlsx" models = json.loads(FRESH.read_text(encoding="utf-8")) TYPE_CN = { "text": "文本", "double": "数值(小数)", "integer": "数值(整数)", "timestamp": "日期时间", "long": "长整型", "float": "数值(小数)", } FRONT_CN = { "content": "文本框", "float_num": "小数", "int_num": "整数", "date_time": "日期时间", "date": "日期", } def type_label(f): t = f.get("type") or "" ft = f.get("frontType") or "" base = TYPE_CN.get(t, t or "") if ft and ft in FRONT_CN: return f"{base}/{FRONT_CN[ft]}" if base else FRONT_CN[ft] return base index = [] for key, m in models.items(): for f in m["fields"]: index.append((key, f)) def find_field(name_hints, prefer_keys=None): hints = [h for h in name_hints if h] prefer_keys = prefer_keys or [] candidates = [] for key, f in index: alias = (f.get("alias") or "").strip() name = (f.get("name") or "").strip() short = name[2:] if name.startswith("c_") else name for h in hints: h = str(h).strip() if not h: continue target = h if h.startswith("c_") else ("c_" + h) if alias == h or name == h or short == h or name == target: candidates.append((key, f, 0)) elif h in alias or (alias and alias in h): candidates.append((key, f, 1)) if prefer_keys: for pk in prefer_keys: for c in candidates: if c[0] == pk: return c return candidates[0] if candidates else None EXPLICIT = { "人员姓名": (["personnel"], ["姓名", "c_name"]), "人员身份证号码": (["personnel"], ["身份证号", "c_id_number"]), "社会保险登记号": (["personnel"], ["个人社会保险登记码", "c_social_insurance_code"]), "手机号": (["personnel"], ["手机号", "c_phone_number"]), "联系地址": (["personnel"], ["家庭住址", "c_address"]), "户籍性质": (["personnel"], ["户籍性质", "c_household_type"]), "街镇": (["personnel"], ["街镇ID", "c_town_id"]), "村居": (["personnel"], ["村居ID", "c_village_id"]), "村委代码": (["personnel"], ["村委代码", "c_village_code"]), "委托方案代码": (["personnel", "region"], ["委托方案代码", "c_entrust_plan_code"]), "委托方案名称": (["personnel"], ["委托方案名称", "c_entrust_plan_name"]), "收款人名称": (["personnel"], ["收款人名称", "c_payee_name"]), "银行卡号": (["personnel"], ["银行卡号", "c_bank_card_number"]), "发放银行": (["personnel", "batchStat", "batch"], ["开户行", "c_bank_name", "发放银行"]), "收款银行行号": (["personnel"], ["收款银行行号", "c_bank_branch_code"]), "退养资格类型": (["personnel"], ["退养资格类型", "c_retire_qualify_type"]), "居保状态": (["personnel"], ["居保养老状态", "c_insurance_status"]), "死亡状态": (["personnel"], ["死亡状态", "c_death_status"]), "死亡日期": (["personnel"], ["死亡日期", "c_death_date"]), "继承人姓名": (["personnel"], ["继承人姓名", "c_heir_name"]), "继承人与逝者关系": (["personnel"], ["继承人关系", "c_heir_relation"]), "继承人联系方式": (["personnel"], ["继承人电话", "c_heir_phone"]), "继承人身份证": (["personnel"], ["继承人身份证", "c_heir_id_number"]), "继承人银行卡": (["personnel"], ["继承人银行卡", "c_heir_bank_card"]), "丧葬费补差额": (["personnel"], ["丧葬费补差额", "c_funeral_diff_amount"]), "丧葬是否已发放": (["personnel", "payment"], ["c_remarks"]), "补贴金额": (["standard"], ["金额", "c_amount"]), "丧葬金额": (["standard"], ["丧葬金额", "c_funeral_amount"]), "生效时间": (["standard"], ["生效时间", "c_date_time"]), "失效时间": (["standard"], ["失效时间", "c_expire_time"]), "标准类型": (["standard"], ["标准类型", "c_standard_type"]), "是否当前有效": (["standard"], ["是否当前有效", "c_current_flag"]), "调整原因": (["standard"], ["调整原因", "c_adjust_reason"]), "享受起始年月": (["personnel"], ["享受起始年月", "代发起始年月", "c_enjoy_start_month"]), "代发截止年月": (["personnel"], ["代发截止年月", "c_enjoy_end_month"]), "补发开始年月": (["personnel"], ["补发开始年月", "c_supplement_start_month"]), "下次发放起始年月": (["personnel"], ["下次发放起始年月", "c_next_pay_start_month"]), "区级资金承担比例": (["region"], ["区级资金承担比例", "c_district_share_percent"]), "镇级资金承担比例": (["region"], ["镇级资金承担比例", "c_town_share_percent"]), "是否已出盘": (["batch", "payment"], ["是否已出盘", "出盘", "c_disk_out"]), "实发金额": (["payment"], ["实发金额", "c_actual_amount"]), "回盘是否成功": (["payment"], ["回盘是否成功", "c_return_success"]), "回盘结果": (["payment"], ["回盘结果", "c_return_result"]), "调标前月补": (["payment"], ["调标前月补", "c_old_monthly_amount"]), "调标后月补": (["payment"], ["调标后月补", "c_new_monthly_amount"]), "调标应发月数": (["payment"], ["调标应发月数", "c_adjust_months"]), "告警等级": (["alert"], ["告警等级", "c_level"]), "告警类型": (["alert"], ["告警类型", "c_type"]), "告警内容": (["alert"], ["告警内容", "c_content"]), "失败原因码": (["alert", "payment"], ["失败原因码", "c_fail_code"]), "处理状态": (["alert", "process"], ["处理状态", "c_status"]), "处理结果": (["alert", "process"], ["处理结果", "c_result"]), "发起人": (["alert", "process"], ["发起人", "c_initiator"]), "接收人": (["alert", "process"], ["接收人", "c_receiver"]), "流程id": (["process", "alert"], ["流程ID", "流程id", "c_process_id"]), "处理人": (["alert", "process"], ["处理人", "c_handler"]), "处理时间": (["alert", "process"], ["处理时间", "c_handle_time"]), "文件地址": (["alert", "batch"], ["文件地址", "c_file_url"]), "备注描述": (["alert", "process"], ["备注", "c_remarks"]), "关联批次ID": (["batchStat", "payment", "batch"], ["关联批次ID", "批次ID", "c_batch_id"]), "批次号": (["batch", "batchStat", "payment"], ["批次号", "c_batch_number"]), "汇总表编号": (["batchStat"], ["汇总表编号", "c_summary_no"]), "发放年月": (["batch", "batchStat", "payment"], ["发放年月", "c_pay_month"]), "镇id": (["batchStat", "batch", "personnel"], ["街镇ID", "镇ID", "c_town_id"]), "村id": (["batchStat", "batch", "personnel"], ["村居ID", "村ID", "c_village_id"]), "统计粒度": (["batchStat"], ["统计粒度", "c_stat_level"]), "应支付人数": (["batchStat", "batch"], ["应支付人数", "c_payable_count"]), "应支付总额": (["batchStat", "batch"], ["应支付总额", "c_payable_amount"]), "实支付人数": (["batchStat", "batch"], ["实支付人数", "c_paid_count"]), "实支付总额": (["batchStat", "batch"], ["实支付总额", "c_paid_amount"]), "支付未成功人数": (["batchStat", "batch"], ["支付未成功人数", "失败人数", "c_fail_count"]), "支付未成功总额": (["batchStat", "batch"], ["支付未成功总额", "c_fail_amount"]), "标准金额": (["batchStat", "payment"], ["标准金额", "月补金额", "c_standard_amount"]), "死亡金额": (["batchStat", "payment"], ["死亡金额", "丧葬金额", "c_funeral_amount"]), "补充金额": (["batchStat", "payment"], ["补充金额", "c_extra_amount"]), "补发金额": (["batchStat", "payment"], ["补发金额", "c_supplement_amount"]), "回盘日期": (["batch", "batchStat", "payment"], ["回盘日期", "c_return_date"]), "经办人": (["batch", "batchStat"], ["经办人", "c_operator"]), "复核人": (["batch", "batchStat"], ["复核人", "c_reviewer"]), "备注": (["batchStat", "batch", "personnel"], ["备注", "c_remarks"]), "人数": (["batch", "batchStat"], ["人数", "c_person_count"]), } CAT_DEFAULT = { "人员基本信息": ("村合作社填报", "随时更新"), "区划与委托": ("系统维护/村合作社填报", "按需更新"), "银行信息": ("村合作社填报", "随时更新"), "退养资格": ("镇经发填报", "随时更新"), "居保匹配": ("大数据匹配", "随时更新"), "死亡状态匹配": ("大数据匹配", "随时更新"), "死亡与继承": ("镇社区事务受理中心/系统计算", "随时更新"), "金额标准": ("政策文件/系统计算", "一年一次或按业务"), "告警": ("系统计算", "每月更新"), "批次统计": ("系统计算", "每月更新"), } TABLE_META = { "personnel": ("人员基本信息表", "村合作社/镇社区/系统", "随时更新"), "region": ("区划表", "系统维护", "按需更新"), "standard": ("金额标准表", "政策文件/镇经发", "一年一次或调标时"), "alert": ("告警表", "系统计算", "随时更新"), "process": ("流程/特殊业务表", "业务部门填报", "随时更新"), "batch": ("批次表", "系统计算", "每月更新"), "batchStat": ("批次统计表", "系统计算", "每月更新"), "payment": ("发放明细表", "系统计算/回盘导入", "每月更新"), } SPECIAL_NOTES = { "丧葬是否已发放": "当前多用人员备注 FUNERAL_PAID 标记，非独立字段", "生效时间": "对应标准表 c_date_time", "是否当前有效": "对应标准表 c_current_flag", "死亡状态": "取值：正常/已死亡；与业务态 c_biz_status=FUNERAL 并存", "街镇": "存业务ID，界面展示中文名", "村居": "存业务ID，界面展示中文名", "镇id": "存业务ID，界面展示中文名", "村id": "存业务ID，界面展示中文名", "发放银行": "人员表对应开户行 c_bank_name", } wb0 = load_workbook(SRC) ws0 = wb0.active orig = [] for i, row in enumerate(ws0.iter_rows(values_only=True), 1): if i == 1: continue vals = list(row) while len(vals) < 7: vals.append(None) cat, name, typ, src, freq, note, must = vals[:7] orig.append( { "cat": cat, "name": name, "type": typ, "source": src, "freq": freq, "note": note, "must": must, } ) cur_cat = None for r in orig: if r["cat"]: cur_cat = r["cat"] else: r["cat"] = cur_cat enriched = [] used_fields = set() for r in orig: name = r["name"] prefer, hints = EXPLICIT.get(name, (None, [name])) hit = find_field(hints, prefer) if prefer is not None else None if hit is None: hit = find_field([name] + list(hints or []), prefer) note_parts = [] if r["note"]: note_parts.append(str(r["note"])) typ = r["type"] must = r["must"] src = r["source"] freq = r["freq"] if hit: key, f, _ = hit used_fields.add((key, f["name"])) typ = type_label(f) if must is None and f.get("must"): must = "是" m = models[key] note_parts.append(f"DMS:{m['label']} {f['name']}") desc = f.get("describe") or "" if desc and desc not in ("格式为富文本", "number", "时间戳"): note_parts.append(str(desc)) if name in SPECIAL_NOTES: note_parts.append(SPECIAL_NOTES[name]) else: note_parts.append("未在DMS单字段精确匹配，请核对业务口径") if name in SPECIAL_NOTES: note_parts.append(SPECIAL_NOTES[name]) if not src or not freq: ds, df = CAT_DEFAULT.get(r["cat"], ("业务填报", "按需更新")) src = src or ds freq = freq or df enriched.append( [ r["cat"], name, typ, src, freq, "；".join([p for p in note_parts if p]), must or "", ] ) rows_sheet1 = [] prev = None for row in enriched: cat = row[0] show_cat = cat if cat != prev else None prev = cat rows_sheet1.append([show_cat, row[1], row[2], row[3], row[4], row[5], row[6]]) # 人员表其余未出现字段 extra_rows = [] append_cat = "人员表补充字段" for f in models["personnel"]["fields"]: if ("personnel", f["name"]) in used_fields: continue used_fields.add(("personnel", f["name"])) note = f"DMS:人员基本信息表 {f['name']}" if f["name"] == "c_death_status": note += "；取值正常/已死亡；与 c_biz_status=FUNERAL 并存" extra_rows.append( [ None, f.get("alias") or f["name"], type_label(f), "系统/业务填报", "随时更新", note, "是" if f.get("must") else "", ] ) if extra_rows: extra_rows[0][0] = append_cat thin = Border( left=Side(style="thin", color="B0B0B0"), right=Side(style="thin", color="B0B0B0"), top=Side(style="thin", color="B0B0B0"), bottom=Side(style="thin", color="B0B0B0"), ) header_fill = PatternFill("solid", fgColor="1F4E79") header_font = Font(color="FFFFFF", bold=True, name="微软雅黑", size=11) cat_fill = PatternFill("solid", fgColor="D6E3F0") body_font = Font(name="微软雅黑", size=10) wrap = Alignment(wrap_text=True, vertical="center") HEADERS7 = ["类别", "字段名", "字段类型", "数据来源", "更新频率", "备注", "是否必须满足"] HEADERS_FULL = HEADERS7 + ["字段编码", "业务表", "栏目ID", "模型ID"] def style_header(ws, headers): for col, h in enumerate(headers, 1): cell = ws.cell(1, col, h) cell.fill = header_fill cell.font = header_font cell.alignment = Alignment(horizontal="center", vertical="center") cell.border = thin ws.row_dimensions[1].height = 22 ws.freeze_panes = "A2" ws.auto_filter.ref = f"A1:{get_column_letter(len(headers))}1" def autosize(ws, widths): for i, w in enumerate(widths, 1): ws.column_dimensions[get_column_letter(i)].width = w def write_rows(ws, rows, headers): style_header(ws, headers) for r_i, row in enumerate(rows, 2): for c_i, val in enumerate(row, 1): cell = ws.cell(r_i, c_i, "" if val is None else val) cell.font = body_font cell.alignment = wrap cell.border = thin if c_i == 1 and val: cell.fill = cat_fill cell.font = Font(name="微软雅黑", size=10, bold=True) wb = Workbook() ws1 = wb.active ws1.title = "业务字段清单" write_rows(ws1, rows_sheet1, HEADERS7) autosize(ws1, [16, 18, 16, 22, 12, 55, 12]) if extra_rows: r0 = len(rows_sheet1) + 3 tip = ws1.cell(r0 - 1, 1, "—— 以下为对照 DMS 后补充的人员表字段 ——") tip.font = Font(name="微软雅黑", bold=True, color="C00000") for i, row in enumerate(extra_rows): for c_i, val in enumerate(row, 1): cell = ws1.cell(r0 + i, c_i, "" if val is None else val) cell.font = body_font cell.alignment = wrap cell.border = thin if c_i == 1 and val: cell.fill = cat_fill cell.font = Font(name="微软雅黑", size=10, bold=True) ws2 = wb.create_sheet("业务表字段全量") full_rows = [] for key in ["personnel", "region", "standard", "payment", "batch", "batchStat", "alert", "process"]: m = models[key] tname, src, freq = TABLE_META[key] first = True for f in m["fields"]: note = f"字段编码 {f['name']}" desc = f.get("describe") or "" if desc and desc not in ("格式为富文本", "number", "时间戳"): note += "；" + str(desc) if f["name"] == "c_death_status": note += "；取值正常/已死亡；与 c_biz_status=FUNERAL 并存" if f["name"] == "c_return_success": note += "；回盘失败以 false + 备注回盘失败|RETURN_FAIL 为准" full_rows.append( [ tname if first else None, f.get("alias") or f["name"], type_label(f), src, freq, note, "是" if f.get("must") else "", f["name"], m["tableCode"], m["columnId"], m["modelId"], ] ) first = False write_rows(ws2, full_rows, HEADERS_FULL) autosize(ws2, [18, 20, 16, 22, 14, 50, 12, 28, 18, 10, 10]) ws3 = wb.create_sheet("业务表索引") style_header(ws3, ["业务表", "表编码", "栏目ID", "模型ID", "字段数", "说明"]) idx_rows = [ ("人员基本信息表", "sjnmtybt_ryjbxx", 1847, 1970, len(models["personnel"]["fields"]), "人员底数、资格、居保、死亡继承等"), ("区划表", "sjnmtybt_xzqh", 1842, 1965, len(models["region"]["fields"]), "区/镇/村及资金比例、委托方案"), ("金额标准表", "sjnmtybt_jebz", 1843, 1966, len(models["standard"]["fields"]), "月补/丧葬标准、调标"), ("发放明细表", "sjnmtybt_ffmx", 1849, 1974, len(models["payment"]["fields"]), "应发实发、回盘、调标差额明细"), ("批次表", "sjnmtybt_pcb", 1845, 1968, len(models["batch"]["fields"]), "全区/村月批、出盘回盘状态"), ("批次统计表", "sjnmtybt_pctj", 1846, 1969, len(models["batchStat"]["fields"]), "镇/村汇总统计"), ("告警表", "sjnmtybt_alart", 1841, 1964, len(models["alert"]["fields"]), "死亡/回盘失败等告警"), ("流程/特殊业务表", "sjnmtybt_lcb", 1844, 1967, len(models["process"]["fields"]), "特殊业务审批与流程"), ] for i, row in enumerate(idx_rows, 2): for c, v in enumerate(row, 1): cell = ws3.cell(i, c, v) cell.font = body_font cell.border = thin cell.alignment = wrap autosize(ws3, [18, 22, 10, 10, 10, 40]) ws4 = wb.create_sheet("填写说明") notes = [ "本文件在《松江农民土地退补系统字段名称》原表格式基础上补全：字段类型、备注（含DMS字段编码/表归属）、以及业务表全量字段。", "原表列：类别、字段名、字段类型、数据来源、更新频率、备注、是否必须满足。", "「业务字段清单」：保留原业务口径字段顺序并补类型/备注；文末追加人员表其余未列出字段。", "「业务表字段全量」：8 张核心业务表全部自定义字段（同原列格式，并增加字段编码/业务表/栏目ID/模型ID）。", "字段类型示例：文本/文本框、数值(小数)/小数、数值(整数)/整数、日期时间/日期时间。", "死亡状态 c_death_status 与业务态 c_biz_status=FUNERAL 并存；回盘失败以 c_return_success=false + 备注为准。", "生成日期：2026-09-04；数据来源：生产 DMS 模型 getModelById。", ] for i, t in enumerate(notes, 1): cell = ws4.cell(i, 1, t) cell.font = body_font cell.alignment = wrap ws4.column_dimensions["A"].width = 120 wb.save(OUT_DESKTOP) wb.save(OUT_DOCS) print("saved", OUT_DESKTOP) print("saved", OUT_DOCS) print("sheet1_orig", len(rows_sheet1), "personnel_extra", len(extra_rows), "full", len(full_rows))