kg-agent/scripts/fetch_dms_fields.py

"""读取 DMS 申勤相关模型的字段定义（只读），输出到 output/dms_models_fields.json。"""

from __future__ import annotations

import json
import sys
import urllib.request


sys.stdout.reconfigure(encoding="utf-8")
TOKEN = ("eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9."
         "eyJyb2xlSWQiOiIwLDEsMiwxMiIsImV4cCI6MTc4NjU2NjU0NCwidXNlcklkIjoyNzQs"
         "InVzZXJuYW1lIjoidXNlcl93YW5neGkifQ."
         "FHjT1Bi_gYW3wAfdqTXpMFfmzW2L8V-gYJ0On1PnWq0")
BASE = "http://121.43.55.7:10081/dms/model/getModelById"

IDS = {
    "sq_employee(申勤员工)": 1793,
    "sq_project(申勤项目数据)": 1794,
    "file1da77(申勤物业项目清单)": 1795,
    "sq_xq(续签表)": 1950,
    "xzbm(行政编码及片区)": 1902,
    "sq_personnel_certificate(人员证书)": 1890,
    "sq_position_salary(岗位区域薪资)": 1900,
    "shenqin_zbb(招标表)": 1853,
    "shenqin_txb(投续标表)": 1852,
    "shenqin_project_annual_stat(年化数据统计)": 1836,
    "sq_project_period(项目期次)": 1876,
    "sq_shenqin_project(项目主数据)": 1896,
    "sq_yt_column(申勤业态表)": 1909,
    "sq_tender_intent(招标意向)": 1954,
    "sq_tender_notice(招标公告)": 1955,
    "sq_tender_result(结果公告)": 1956,
}


def fetch(mid: int) -> dict:
    url = BASE + "?modelId=" + str(mid)
    req = urllib.request.Request(
        url, data=b"{}",
        headers={"token": TOKEN, "Content-Type": "application/json"})
    with urllib.request.urlopen(req, timeout=30) as r:
        return json.loads(r.read().decode("utf-8", "replace"))


def main() -> None:
    out = {}
    for name, mid in IDS.items():
        try:
            data = fetch(mid)
            content = data.get("content") or {}
            fields = content.get("fieldList") or "{}"
            if isinstance(fields, str):
                fields = json.loads(fields)
            names = []
            if isinstance(fields, dict):
                for k, v in fields.items():
                    alias = v.get("alias") if isinstance(v, dict) else (v if isinstance(v, str) else str(v))
                    names.append((k, alias))
            elif isinstance(fields, list):
                for v in fields:
                    if isinstance(v, dict):
                        names.append((v.get("name"), v.get("alias")))
                    else:
                        names.append((str(v), str(v)))
            out[name] = {"modelId": mid, "modelName": content.get("modelName"),
                         "fields": names}
            print(f"== {name} [{content.get('modelName')}] 字段数={len(names)}")
            for n, a in names:
                print(f"    {n} | {a}")
        except Exception as e:  # noqa: BLE001
            print(f"== {name} ERR {e}")
    with open(r"E:\CODE\knowledge_agent\output\dms_models_fields.json", "w", encoding="utf-8") as f:
        json.dump(out, f, ensure_ascii=False, indent=1)


if __name__ == "__main__":
    main()