zsjz-ai/ai-server/scripts/gen_field_labels.py

# -*- coding: utf-8 -*-
"""
生成字段中文名字典 ai-server/src/main/resources/agent/field-labels.json。

数据源（按优先级）：
  1. 字段上的 @ExcelProperty("中文名")（导出表头用，最权威）
  2. 紧邻字段的 javadoc 首行（/** 本方姓名 */）

产物两段：
  classes —— 类全限定名 → {字段名: 中文名}，用于「按行的实际类型精确取名」（如 CallNightDTO.time = 时间段标识）
  "*"     —— 字段名 → 中文名 的兜底并集，仅收实体/DTO 的票且多数派占比 ≥ 0.6，
             用于 SQL 列别名 / 手拼 Map 行这类拿不到类型的场景；拿不准的字段留英文不猜

重新生成：python ai-server/scripts/gen_field_labels.py
"""
import json
import pathlib
import re
import sys

ROOT = pathlib.Path(__file__).resolve().parents[1] / "src/main/java"
OUT = pathlib.Path(__file__).resolve().parents[1] / "src/main/resources/agent/field-labels.json"

# 只扫业务数据类所在包：实体 / DTO / VO / 查询对象（服务类字段没有业务含义）
SCAN_PREFIXES = ("com.zsjz.ai.common.model.", "com.zsjz.ai.module.")

PACKAGE_RE = re.compile(r"^\s*package\s+([\w.]+)\s*;")
CLASS_RE = re.compile(r"^\s*(?:@\w+(?:\([^)]*\))?\s*)*(?:public\s+|final\s+|abstract\s+)*"
                      r"(?:class|record|enum)\s+(\w+)")
# 字段声明：私有/受保护/公有 + 非 static/transient + 类型 + 名字 + 分号（方法带括号，天然排除）
FIELD_RE = re.compile(r"^\s*(?:private|protected|public)\s+(?!static\b)(?!transient\b)"
                      r"([\w.$?<>,\[\]\s]+?)\s+(\w+)\s*(?:=\s*[^;]+)?;\s*$")
EXCEL_RE = re.compile(r'@ExcelProperty\s*\(([^;]*?)\)\s*$')
QUOTED_RE = re.compile(r'"([^"]+)"')
# 历史代码里有一批「//对方名称\")」形态的残缺行注释（Javadoc 被工具改坏了），同样当列名用
BROKEN_COMMENT_RE = re.compile(r'^//\s*"?(.+?)"?\)\s*$')
# javadoc 行内标记 {@link Xxx} / {@code xxx}：取标记里的文字，别让花括号把整条注释判废
INLINE_TAG_RE = re.compile(r"\{@\w+\s+([^}]*)\}")


def clean(text: str) -> str:
    """注释行 → 可用的中文列名：去 * 前缀、摊平行内 javadoc 标记、去 HTML 标签与引号，截到第一个句读"""
    text = re.sub(r"^[\s*]+|[\s*]+$", "", text).strip()
    text = INLINE_TAG_RE.sub(lambda m: m.group(1), text)
    text = re.sub(r"</?[a-zA-Z][^>]*>", "", text)  # <p> <b> 之类
    # {@link #sampleText} 展开后是自引用「#sampleText 的字符数」——去掉引用，别让表头带上字段名
    text = re.sub(r"#\w+\s*", "", text).lstrip("的").strip()
    text = text.strip('"').strip()
    if not text or text.startswith("@") or text.startswith("{@") or text.startswith("<"):
        return ""
    # 「借贷类型全部='',收入=收入」这类取值域说明不是列名
    if any(ch in text for ch in "=;{}"):
        return ""
    # 「失败原因（FAIL / SKIP 时才有值），供排查」→ 取**位置最早**的句读截断
    # （按分隔符优先级顺序找会漏掉更靠前的括号，必须比位置）
    cuts = [i for i in (text.find(sep) for sep in ("。", "：", "，", ",", "（", "("))
            if 0 < i <= 30]
    if cuts:
        text = text[:min(cuts)]
    return text[:30].strip()


def scan(path: pathlib.Path):
    """返回 (类全限定名, {字段: 中文名})；无可用中文名的类返回 None"""
    lines = path.read_text(encoding="utf-8", errors="ignore").splitlines()
    package = ""
    cls = ""
    fields = {}
    i = 0
    while i < len(lines):
        line = lines[i]
        if not cls:
            m = PACKAGE_RE.match(line)
            if m:
                package = m.group(1)
            m = CLASS_RE.match(line)
            if m and package:
                cls = m.group(1)
            i += 1
            continue

        m = FIELD_RE.match(line)
        if not m:
            i += 1
            continue
        name = m.group(2)
        if name == "serialVersionUID" or name in fields:
            i += 1
            continue

        # 回溯：跳过注解行（可能多行），再找紧邻的 javadoc
        excel = ""
        j = i - 1
        while j >= 0 and lines[j].strip().startswith("@"):
            em = EXCEL_RE.match(lines[j].strip())
            if em:
                quoted = QUOTED_RE.findall(em.group(1))
                if quoted:
                    excel = quoted[-1]  # 多级表头取最后一级 = 该列标题
            j -= 1
        label = clean(excel)
        if not label and j >= 0:
            broken = BROKEN_COMMENT_RE.match(lines[j].strip())
            if broken:
                label = clean(broken.group(1))
        if not label and j >= 0 and lines[j].strip().endswith("*/"):
            k = j
            while k >= 0 and "/**" not in lines[k]:
                k -= 1
            if k >= 0:
                block = []
                for raw in lines[k:j + 1]:
                    body = raw.strip()
                    body = body[3:] if body.startswith("/**") else body
                    body = body[:-2] if body.endswith("*/") else body
                    block.append(body.lstrip("*").strip())
                for row in block:
                    if not row or row.startswith("@"):
                        continue
                    label = clean(row)
                    if label:
                        break
        if label:
            fields[name] = label
        i += 1

    if not fields:
        return None
    return f"{package}.{cls}", fields


def main() -> int:
    classes = {}
    for path in sorted(ROOT.rglob("*.java")):
        rel = str(path.relative_to(ROOT)).replace("\\", "/")
        if not rel.replace("/", ".").startswith(SCAN_PREFIXES):
            continue
        found = scan(path)
        if found:
            classes[found[0]] = found[1]

    # 兜底并集（JSON 的 "*" 段）：只收「实体 / DTO」这类数据类的票，
    # 且要求多数派占比 >= 0.6 —— 拿不准的字段宁可留英文也不猜。
    # 同名字段跨域含义不同（personName 在话单是「本方姓名」、在机票是「旅客姓名」；
    # count 在统计行是「份数」、在轨迹是「同行次数」），猜错比不翻更误导。
    # 行本身带类型（classes 段）或 table_field 里有的字段，不依赖这段兜底。
    votes: dict[str, dict[str, int]] = {}
    for fqn in sorted(classes):
        if ".entity." not in fqn and ".dto." not in fqn:
            continue
        weight = 2 if ".entity." in fqn else 1
        for key, cn in classes[fqn].items():
            bucket = votes.setdefault(key, {})
            bucket[cn] = bucket.get(cn, 0) + weight
    fallback = {}
    for key, bucket in votes.items():
        ranked = sorted(bucket.items(), key=lambda kv: (-kv[1], kv[0]))
        if ranked[0][1] / sum(bucket.values()) >= 0.6:
            fallback[key] = ranked[0][0]

    payload = {
        "_comment": "由 ai-server/scripts/gen_field_labels.py 生成，勿手改；"
                    "重新生成后按需提交。classes=按类精确取名，*=SQL 别名/Map 行的兜底。",
        "classes": classes,
        "*": fallback,
    }
    OUT.parent.mkdir(parents=True, exist_ok=True)
    OUT.write_text(json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True) + "\n",
                   encoding="utf-8")
    print(f"classes={len(classes)} fields={sum(len(v) for v in classes.values())} "
          f"fallback={len(fallback)} -> {OUT}")
    return 0


if __name__ == "__main__":
    sys.exit(main())