sj_nmtybt_server/scripts/build-field-xlsx-20260904.py

# -*- 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))