Blame for scripts/classify_skill_usage.py

# -*- coding: utf-8 -*-
"""B3 用途二次分类：1229 条技能 → K/M/T/X/U 五值 → data/skill_usage_labels.csv

用法：
  python scripts/classify_skill_usage.py            # 生成 CSV + 分布回执
  python scripts/classify_skill_usage.py --check    # 复跑字节一致性

判定口径（确定性，无随机、无时间戳）：
  stage1 逐类打分：score(c) = Σ(关键词权重 × 命中) + 0.6 × 2~4 字滑窗重叠率
         —— 逐条对各标签打分再取最大，绝不用"整类文本聚合"打分（那样所有条目会塌到同一类）。
         argmax 且 max >= TH 才算判定；否则进 stage2。
  stage2 兜底：按索引侧 domain / mechanism 查表定性；仍无依据则标 U（待判）。
  U 上限：判据要求待判 <= 20，超出即视为分类器失效而不是"数据太怪"。

标签语义：
  K 知识源：提供内容/资料/知识结构，可被检索与引用
  M 方法学：提供方法/模型/算法/流程/范式，可被调用为"怎么做"
  T 工具链：提供可执行的命令、脚本、部署与平台能力
  X 无关：与中西医融合康复域无关，明确不纳入
  U 待判：证据不足，挂账不猜
"""
import csv
import hashlib
import io
import json
import os
import sys

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
INDEX = os.path.join(ROOT, "data", "skill_matrix_index.json")
CSV_OUT = os.path.join(ROOT, "data", "skill_usage_labels.csv")
HEADER = ["name", "domain", "mechanism", "lane", "label", "score", "stage", "reason"]

TH = 0.80
TH_WEAK = 0.35   # 弱证据档：仅用于索引域未归域的条目，低于此判 U 挂账

# 关键词权重：域名词面证据，不做同义词扩张，避免误伤
KW = {
    "K": {"知识": 1.2, "导学": 1.0, "学派": 0.9, "医案": 0.9, "文献": 1.0, "教材": 1.0,
          "经典": 1.0, "讲稿": 0.9, "讲义": 0.9, "传承": 0.8, "综述": 0.8, "词条": 0.8,
          "方剂": 0.7, "本草": 0.7, "内经": 0.8, "伤寒": 0.8, "百科": 0.8, "典籍": 0.8,
          "ontology": 1.0, "知识库": 1.2, "索引": 0.6, "图谱": 0.8, "概览": 0.8,
          "archive": 0.8, "corpus": 0.8, "timeline": 0.6, "规范": 0.7},
    "M": {"方法": 1.2, "方法论": 1.2, "框架": 1.0, "矩阵": 0.9, "体系": 0.9, "机制": 1.0,
          "模型": 1.0, "算法": 1.0, "算子": 1.0, "流程": 1.0, "sop": 1.0, "协议": 0.9,
          "策略": 0.9, "推理": 0.9, "范式": 1.0, "拆解": 0.8, "定位": 0.6, "解析": 0.7,
          "判定": 0.8, "分类": 0.6, "建模": 0.8, "调度": 0.8, "校准": 0.7, "打分": 0.7,
          "design": 0.6, "analysis": 0.6, "pipeline": 0.9, "迭代": 0.7, "打磨": 0.6,
          "车间": 1.0, "匹配": 0.7, "收敛": 0.8, "调度": 0.8},
    "T": {"工具": 1.2, "脚本": 1.2, "cli": 1.2, "命令": 1.0, "操作": 1.0, "基准": 1.2,
          "引导": 0.9, "适配": 1.0, "安装": 1.0, "部署": 1.2, "路由": 1.0, "api": 1.0,
          "mcp": 1.0, "打包": 0.9, "上传": 0.8, "下载": 0.8, "渲染": 0.8, "docker": 1.2,
          "向量库": 0.9, "bridge": 0.9, "热插拔": 0.9, "网关": 0.9, "codec": 0.9,
          "sidecar": 0.9, "harness": 0.9, "质检": 1.2, "检测": 0.6, "读取": 0.7, "编辑": 0.7,
          "交由": 0.8, "自动化": 0.6},
    "X": {"娱乐": 1.0, "星座": 1.0, "彩票": 1.0, "游戏攻略": 1.0, "八卦": 1.0, "影视": 0.8,
          "小说": 0.8, "美食": 0.8, "旅游": 0.8, "穿搭": 0.8, "美妆": 0.8, "宠物": 0.6,
          "股票": 0.6, "量化交易": 0.8, "招聘": 0.6, "房产": 0.6,
          # 农业/养殖/园艺类：与中西医融合康复域无关，属明确不纳入（写进票据，不藏着）
          "养殖": 1.0, "鱼塘": 1.0, "鹅": 0.8, "藻": 0.8, "朴门": 1.0, "果园": 0.8,
          "化肥": 0.8, "农药": 0.8, "蚯蚓": 0.8, "梯级水系": 0.8,
          # 纯想象/猎奇场景：不属于康复域
          "虫洞": 1.0, "星际": 0.8, "探险": 0.8, "趣味": 0.6},
}

# stage2 兜底：索引侧既有字段 → 标签（依据单一，不叠加猜测）
DOMAIN_FALLBACK = {
    "literature": "K", "gaoyi": "K",
    "research": "M", "optimization": "M", "quant": "M",
    "data": "T", "growth": "T",
    "writing": "X", "business": "X",
    "unmapped": "U",
}
MECH_FALLBACK = {
    "meridian": "M", "cytoskeleton": "M",
    "semantic_bus": "T", "skin_v6": "M", "holographic_heartbeat": "M",
    "hebbian_field": "M", "unmapped": "U",
}


def shingles(text, lo=2, hi=4):
    out = set()
    for i in range(len(text)):
        for n in range(lo, hi + 1):
            if i + n <= len(text):
                out.add(text[i:i + n])
    return out


def overlap_rate(a, b):
    if not a or not b:
        return 0.0
    return len(a & b) / float(len(a))


def score_all(text, ref_shingles):
    sc = {}
    low = text.lower()
    for label, kws in KW.items():
        s = 0.0
        for w, weight in kws.items():
            if w in low:
                s += weight
        s += 0.6 * overlap_rate(shingles(text), ref_shingles[label])
        sc[label] = round(s, 4)
    return sc


def refs_by_label():
    return {label: shingles("".join(kws.keys())) for label, kws in KW.items()}


def classify(rec, ref):
    text = (rec.get("name", "") + " " + rec.get("description", "")).strip()
    sc = score_all(text, ref)
    order = ["K", "M", "T", "X"]
    best = max(order, key=lambda k: (sc[k], -order.index(k)))
    if sc[best] >= TH:
        return best, sc[best], "stage1", "kw+shingle argmax={} score={}".format(best, sc[best])
    d = DOMAIN_FALLBACK.get(rec.get("domain"), "U")
    has_domain_evidence = (rec.get("domain_score") or 0) > 0 and rec.get("domain") != "unmapped"
    if d != "U" and has_domain_evidence:
        return d, sc[best], "stage2", "domain 定性={}（索引域有证据，score={}）".format(
            d, rec.get("domain_score"))
    if sc[best] >= TH_WEAK:
        return best, sc[best], "stage2-weak", "弱证据 argmax={} score={}（域未归域，仅词面）".format(
            best, sc[best])
    return "U", sc[best], "undecided", "domain={} mechanism={} 证据不足（score={}<{}）".format(
        rec.get("domain"), rec.get("mechanism"), sc[best], TH_WEAK)


def build():
    with open(INDEX, encoding="utf-8") as f:
        idx = json.load(f)
    ref = refs_by_label()
    rows = []
    for rec in idx["skills"]:
        label, score, stage, reason = classify(rec, ref)
        rows.append({"name": rec.get("name", ""), "domain": rec.get("domain", ""),
                     "mechanism": rec.get("mechanism", ""), "lane": rec.get("lane", ""),
                     "label": label, "score": "{:.4f}".format(score), "stage": stage,
                     "reason": reason})
    counts = {"K": 0, "M": 0, "T": 0, "X": 0, "U": 0}
    for r in rows:
        counts[r["label"]] += 1
    digest = hashlib.sha256(
        json.dumps(rows, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8")
    ).hexdigest()[:16]
    return rows, counts, digest


def csv_text(rows):
    buf = io.StringIO()
    w = csv.DictWriter(buf, fieldnames=HEADER, lineterminator="\n")
    w.writeheader()
    for r in rows:
        w.writerow(r)
    return buf.getvalue()


def main(argv):
    rows, counts, digest = build()
    text = csv_text(rows)
    if "--check" in argv:
        if not os.path.isfile(CSV_OUT):
            print(json.dumps({"verdict": "FAIL", "reason": "missing"}, ensure_ascii=False))
            return 1
        with open(CSV_OUT, encoding="utf-8") as f:
            same = f.read() == text
        print(json.dumps({"verdict": "PASS" if same else "FAIL", "byte_consistent": same,
                          "digest": digest, "counts": counts}, ensure_ascii=False))
        return 0 if same else 1
    with open(CSV_OUT, "w", encoding="utf-8", newline="\n") as f:
        f.write(text)
    print(json.dumps({"out": CSV_OUT, "rows": len(rows), "counts": counts,
                      "digest": digest, "sha256": hashlib.sha256(text.encode("utf-8")).hexdigest()},
                     ensure_ascii=False))
    return 0


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