guanlan-src/scripts/curate_oem_lexicon.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""OEM 词库筛选: 原始频次表 → 可用术语基准. 去模板/版权/公司名类噪音, 保留领域术语, 标注跨厂商共现 (共现 = 行业习惯用语的最强信号).
用法: python scripts/curate_oem_lexicon.py [--raw app_frontEnd/configs/terms/oem_lexicon.yaml] [--out app_frontEnd/configs/terms/terms_baseline.yaml]"""
from __future__ import annotations
import argparse, re, time
from pathlib import Path
import yaml
ROOT = Path(__file__).resolve().parents[1]
NOISE = re.compile("|".join([
    "版权", "保留所有权利", "商业机密", "印刷版本", "不受控制", "母版", "标题样式", "单击此处", "页码", "页眉", "页脚", "编辑",
    "有限公司", "分公司", "集团", "股份", "公司$", "电话", "传真", "邮箱", "地址", "网址", "联系",
    "第.页", "共.页", "附件", "附录", "目录", "修订", "编制", "审核人", "批准", "签字", "日期",
]))
# 领域轴: 词必须命中其一 (部件/状态/动作/量), 否则视为通用词
AXES = {
    "部件": "叶片 叶根 轮毂 变桨 桨距 主轴 轴承 齿轮箱 齿圈 联轴器 发电机 变流器 变频 偏航 塔筒 塔架 基础 机舱 液压 制动 刹车 润滑 油脂 滤芯 密封 螺栓 法兰 编码器 传感器 冷却 散热 风扇 加热 柜 泵 阀 电缆 滑环 集电环 变压器 开关 断路器",
    "状态": "故障 报警 停机 待机 并网 脱网 限功率 限电 过温 高温 低温 超速 振动 异响 泄漏 渗漏 磨损 点蚀 裂纹 腐蚀 松动 卡滞 堵塞 老化 退化 偏差 漂移 不平衡 对中 失效 损坏 异常 正常",
    "动作": "检查 更换 紧固 校准 标定 清洁 润滑 加注 排放 复位 复归 试验 测试 巡检 维护 检修 调试 拆卸 安装 修复 处理 监测 诊断 记录 观察 停运 隔离 挂牌",
    "量": "温度 压力 转速 扭矩 功率 电压 电流 频率 风速 风向 角度 位移 加速度 间隙 力矩 流量 液位 油位 时长 次数 效率 可利用率 发电量",
}
AXIS_WORDS = {k: set(v.split()) for k, v in AXES.items()}


def axis_of(w: str):
    hit = [k for k, ws in AXIS_WORDS.items() if any(x in w for x in ws)]
    return hit


def main():
    ap = argparse.ArgumentParser(); ap.add_argument("--raw", default="app_frontEnd/configs/terms/oem_lexicon.yaml"); ap.add_argument("--out", default="app_frontEnd/configs/terms/terms_baseline.yaml"); ap.add_argument("--min-n", type=int, default=3); a = ap.parse_args()
    raw = yaml.safe_load((ROOT / a.raw).read_text(encoding="utf-8"))
    keep, drop_noise, drop_generic = [], 0, 0
    for it in raw["items"]:
        w = it["term"]
        if NOISE.search(w): drop_noise += 1; continue
        ax = axis_of(w)
        if not ax: drop_generic += 1; continue
        if it["n"] < a.min_n: continue
        keep.append(dict(term=w, n=it["n"], axes=ax, oems=it["oems"], cross_vendor=len(it["oems"]) > 1, per_oem=it["per_oem"], sources=it["sources"][:2]))
    keep.sort(key=lambda x: (-int(x["cross_vendor"]), -x["n"], x["term"]))
    out = ROOT / a.out; out.parent.mkdir(parents=True, exist_ok=True)
    doc = dict(schema="terms_baseline/v0", built=time.strftime("%Y-%m-%d"),
               note="OEM 中文文档实证术语基准 (只存术语/频次/出处文件名, 无原文; D3)。cross_vendor=True 者 = 多家 OEM 共用 → 行业习惯用语, 优先采用。",
               source_stat=raw.get("stat"), n=len(keep), n_cross_vendor=sum(1 for x in keep if x["cross_vendor"]),
               dropped=dict(boilerplate=drop_noise, generic=drop_generic), axes=list(AXES), items=keep)
    out.write_text(yaml.safe_dump(doc, allow_unicode=True, sort_keys=False), encoding="utf-8")
    print(f"术语基准 {len(keep)} 条 (跨厂商 {doc['n_cross_vendor']}); 去模板 {drop_noise} / 去通用 {drop_generic} → {out}")
    for ax in AXES: print(f"  {ax}: {sum(1 for x in keep if ax in x['axes'])}")


if __name__ == "__main__":
    main()