kg-agent/scripts/build_field_association.py

"""生成 模板-源数据-DMS 三方字段关联 CSV。

输出：
  output/模板_源数据_DMS字段关联.csv
"""

from __future__ import annotations

import ast
import csv
import importlib.util
import json
import re
import sys
from pathlib import Path


sys.stdout.reconfigure(encoding="utf-8")
ROOT = Path(r"E:\CODE\knowledge_agent")
OUT = ROOT / "output" / "模板_源数据_DMS字段关联.csv"
DATA_FILES = [
    p.name for p in (ROOT / "data").rglob("*")
    if p.is_file() and p.suffix.lower() in {".xlsx", ".xls", ".csv", ".docx", ".doc", ".pdf"}
]
DMS_ALIASES: dict[str, str] = {}
try:
    raw_dms = json.loads((ROOT / "output" / "dms_scoped_models_fields.json").read_text(encoding="utf-8-sig"))
    DMS_ALIASES = {
        k: (v.get("modelAlias") or v.get("path") or k)
        for k, v in raw_dms.items()
        if isinstance(v, dict)
    }
except Exception:  # noqa: BLE001
    pass


def load_module(name: str, path: Path):
    spec = importlib.util.spec_from_file_location(name, path)
    mod = importlib.util.module_from_spec(spec)
    assert spec and spec.loader
    spec.loader.exec_module(mod)
    return mod


dms_mod = load_module("dms_supplement", ROOT / "scripts" / "generate_dms_supplement.py")
tpl_mod = load_module("template_gen", ROOT / "scripts" / "generate_templates.py")


def norm_header(h: str) -> str:
    h = str(h or "").strip()
    h = re.sub(r"\*$", "", h)
    h = re.sub(r"（[^）]*）$", "", h)
    return re.sub(r"\s+", "", h)


def parse_field_sources() -> dict[str, list[tuple[str, str]]]:
    tree = ast.parse(Path(tpl_mod.__file__).read_text(encoding="utf-8"))
    namespace = {k: v for k, v in tpl_mod.__dict__.items() if not k.startswith("__")}
    result: dict[str, list[tuple[str, str]]] = {}
    for node in ast.walk(tree):
        if not isinstance(node, ast.Call):
            continue
        if not isinstance(node.func, ast.Name) or node.func.id != "_make_file":
            continue
        if len(node.args) < 3:
            continue
        filename_node = node.args[0]
        headers_node = node.args[2]
        try:
            filename = ast.literal_eval(filename_node)
            headers = ast.literal_eval(headers_node)
        except Exception:  # noqa: BLE001
            continue
        if not isinstance(filename, str) or not isinstance(headers, list):
            continue
        field_sources = []
        for kw in node.keywords:
            if kw.arg != "field_sources":
                continue
            try:
                value = eval(compile(ast.Expression(body=kw.value), "<tpl>", "eval"),
                             namespace, namespace)
            except Exception:  # noqa: BLE001
                value = []
            if isinstance(value, list):
                field_sources = [
                    (str(a), str(b))
                    for a, b in value
                    if isinstance(a, str) and isinstance(b, str)
                ]
        stem = Path(filename).stem
        result[stem] = field_sources
    return result


def _split_source(desc: str) -> tuple[str, str]:
    desc = desc.strip()
    m = re.match(r"(.+?)\s*→\s*(.+)", desc, re.S)
    if not m:
        return desc, desc
    return m.group(1).strip(), m.group(2).strip()


def _file_names(src: str) -> list[str]:
    names: list[str] = []
    for part in re.split(r"[；;]", src):
        part = part.strip()
        if not part:
            continue
        if "新增（融合）" in part or "新增（分类）" in part:
            continue
        if "关联「" in part or part in {"—", "-"}:
            continue
        name = re.split(r"[\\/]", part)[-1].strip()
        name = re.sub(r"[（(][^）)]*[）)]$", "", name).strip()
        if re.search(r"\.(xlsx|xls|csv|docx|doc|pdf)$", name, re.I):
            names.append(name)
            continue
        # 字段来源里可能是“登记表名称”而非完整文件名，尝试在工作区数据文件中匹配
        hit = next((f for f in DATA_FILES if name and name in f), None)
        if hit:
            names.append(hit)
    return names


def parse_source(desc: str) -> tuple[str, str]:
    src, field = _split_source(desc)
    names = _file_names(src)
    src_out = "；".join(names)
    if "新增（融合）" in desc or "新增（分类）" in desc:
        return "", field
    if not names:
        return "", field
    quoted = re.findall(r"「([^」]+)」", field)
    if quoted:
        field = " / ".join(quoted)
    return src_out, field


def split_fields(label: str) -> list[str]:
    return [re.sub(r"\s+", "", x) for x in re.split(r"[/、]", label) if re.sub(r"\s+", "", x)]


def main() -> None:
    sources = parse_field_sources()
    rows: list[dict] = []
    last_file: dict[str, str] = {}
    for tpl, fields in dms_mod.TEMPLATES.items():
        tpl_sources = sources.get(tpl, [])
        for field in fields:
            dms_source = dms_mod.DMS_SOURCES.get((tpl, field), "")
            dms_model = dms_field = ""
            if dms_source:
                m = re.match(r"^([^.]+)\.(.*)$", dms_source)
                if m:
                    dms_model, dms_field = m.group(1), m.group(2)
            dms_model_display = dms_model
            if dms_model and dms_model in DMS_ALIASES:
                dms_model_display = f"{dms_model}（{DMS_ALIASES[dms_model]}）"
            meta_file = meta_field = ""
            matched_desc = ""
            for label, desc in tpl_sources:
                for lf in split_fields(label):
                    if norm_header(field) == norm_header(lf) or (
                        len(norm_header(lf)) >= 2 and norm_header(lf) in norm_header(field)
                    ):
                        matched_desc = desc
                        break
                if matched_desc:
                    break
            if matched_desc:
                raw_src, _ = _split_source(matched_desc)
                if raw_src.strip() in {"同一登记表", "同上", "同文件", "同一文件"}:
                    meta_file = last_file.get(tpl, "")
                    meta_field = parse_source(matched_desc)[1]
                else:
                    meta_file, meta_field = parse_source(matched_desc)
                    if meta_file:
                        last_file[tpl] = meta_file
            if not meta_file:
                meta_field = ""
            rows.append({
                "模板文件名称": f"{tpl}.csv",
                "模板字段": field,
                "元数据文件": meta_file,
                "元数据字段名称": meta_field,
                "dms模型": dms_model_display,
                "dms字段名称": dms_field,
                "是否需要人工增加": "否" if dms_source else "是",
            })

    OUT.parent.mkdir(parents=True, exist_ok=True)
    with OUT.open("w", newline="", encoding="utf-8-sig") as f:
        w = csv.writer(f)
        w.writerow(["模板文件名称", "模板字段", "元数据文件", "元数据字段名称",
                    "dms模型", "dms字段名称", "是否需要人工增加"])
        for r in rows:
            w.writerow([r["模板文件名称"], r["模板字段"], r["元数据文件"],
                        r["元数据字段名称"], r["dms模型"], r["dms字段名称"],
                        r["是否需要人工增加"]])
    print("输出:", OUT)
    print("字段行数:", len(rows))


if __name__ == "__main__":
    main()