kangshifu-board/legacy/tools/enrich.py

"""enrich.py —— 补齐演示数据里"缺的部分"，让框架链路可被真实验证（不新增数据文件、不改业务字段）。

 1. 给每条过程层成果补出身：ownerSession（缺）、sources（原文出处，空数组 = 明确没有原文）
 2. 把 2 条不合格的成果标为 record 层（不合格/纯推测/重复），让记录层有真实样本
 3. 重建归纳：derivedFrom 改为指向真正由多人产生、且有原文的成果；
    交叉人数 = 被引用成果里不同成员数；evidenceQuoteCount = 带原文的成果数
 4. 同步 evidenceLevel 与原文是否一致（有原文→verified，无原文→speculative/supported）

运行：PARSE_MASTER_KEY=... python3 tools/enrich.py [--dry-run]

"""
import json
import os
import sys
import urllib.error
import urllib.request

BASE = "https://kangshifu.fmode.cn/parse"
APP_ID = "FKieRbIoGt_main"
MK = os.environ.get("PARSE_MASTER_KEY", "")
DRY = "--dry-run" in sys.argv
H = {"X-Parse-Application-Id": APP_ID, "Content-Type": "application/json"}
if MK:
    H["X-Parse-Master-Key"] = MK


def call(method, path, body=None):
    data = json.dumps(body, ensure_ascii=False).encode() if body is not None else None
    req = urllib.request.Request(BASE + path, data=data, headers=H, method=method)
    try:
        with urllib.request.urlopen(req, timeout=30) as r:
            return json.loads(r.read().decode() or "{}")
    except urllib.error.HTTPError as e:
        return {"_error": e.code, "_body": e.read().decode()[:400]}


def fetch(cls):
    return call("GET", f"/classes/{cls}?limit=1000&order=createdAt").get("results", [])


# 通用的"这条成果来自哪条真实评论"模板：只表示出处形态，不新增业务字段
QUOTE_TEMPLATES = [
    {"type": "consumer_review", "ref": "https://example.com/reviews/{n}", "quote": "{q}"},
]

INSIGHT_PLANS = {
    # (跨阶段理性) 让结论按阶段自然分开，避免互相抢同一批成果
    "拌面酱包口味单一被反复提及": (["S2", "S3"], ["张伟", "李娜", "王芳", "刘洋", "陈静", "赵磊"], 2),
    "夜宵场景对低卡有明确诉求": (["S3", "S4"], ["张伟", "李娜", "吴敏", "周涛", "郑凯"], 2),
    "学生党价格带敏感": (["S2", "S3"], ["王芳", "刘洋", "陈静", "赵磊"], 1),
    "露营场景渗透率低": (["S1", "S5"], ["孙悦", "周涛"], 0),
}


def stage_of(s):
    return (s or "")[:2].upper()


def quote_for(title, owner):
    for k, qs in QUOTES.items():
        if k in title:
            return qs[len(owner) % len(qs)]
    return ""


def quote_for(title, i):
    for k, qs in QUOTES.items():
        if k in title:
            return qs[i % len(qs)]
    return ""


def main():
    if not MK:
        sys.exit("PARSE_MASTER_KEY 未设置")
    if "results" not in call("GET", "/classes/KfArtifact?limit=1"):
        sys.exit("masterKey 验证失败")

    projects = {p["objectId"]: p for p in fetch("KfProject")}
    arts = fetch("KfArtifact")
    ins_all = fetch("KfInsight")
    # 只处理带演示标题的归纳
    ins = [i for i in ins_all if i.get("title") in INSIGHT_PLANS]

    # ---- 1. 补出身与 sources ----
    n = 0
    for a in arts:
        if " - " not in a.get("title", ""):
            continue  # 只处理演示数据，不动 upload skill 真实写入的记录
        p = projects.get(a.get("projectId"), {})
        upd = {}
        if not a.get("ownerSession"):
            upd["ownerSession"] = p.get("ownerSession") or f"sess-{a.get('ownerName', 'unknown')}"
        if "sources" not in a or a.get("sources"):
            # 演示数据本身不是真实评论：不伪造原文，统一置为空数组（= 明确没有原文）
            upd["sources"] = []
        if "layer" not in a:
            upd["layer"] = "process"
        # 证据级别与原文对齐：没有原文就不允许标"已验证"
        has_q = bool(a.get("sources")) and not ("sources" in upd and not upd["sources"])
        if has_q:
            want = "verified"
        elif a.get("evidenceLevel") == "verified":
            want = "supported"
        else:
            want = a.get("evidenceLevel")
        if want != a.get("evidenceLevel"):
            upd["evidenceLevel"] = want
        if upd:
            n += 1
            print("[dry]" if DRY else "upd", a["title"], upd)
            if not DRY:
                call("PUT", f"/classes/KfArtifact/{a['objectId']}", upd)
    print("updated artifacts:", n)

    # ---- 2. 记录层样本：没有原文且标为推���、同一个人重复的两条 ----
    spec = [a for a in arts if a.get("evidenceLevel") == "speculative" and not a.get("sources")]
    if not DRY:
        for a in spec[:2]:
            call("PUT", f"/classes/KfArtifact/{a['objectId']}", {"layer": "record", "meta": {"recordReason": "纯推测，无原文"}})
            print("record:", a["title"])
    print("record picks:", [a["title"] for a in spec[:2]])

    # ---- 3. 重建归纳 ----
    # 规则：每条归纳的支撑记录必须来自不同成员（这才叫"几人独立提到"）；
    #      需要原文的结论优先拿带 sources 的成果；成果被多条归纳共享时只计一次引用。
    fresh = fetch("KfArtifact")
    pool = [a for a in fresh if a.get("layer") != "record"]
    used = set()
    def stage_ok(a, stages):
        return any(stage_of(a.get("stage")).startswith(x) for x in stages)

    plans = [(i, INSIGHT_PLANS[i["title"]]) for i in ins]
    for i, (stages, members, want_quote) in plans:
        picks, owners = [], []
        for m in members:
            cand = [a for a in pool if a["objectId"] not in used and a.get("ownerName") == m
                    and stage_ok(a, stages) and not a.get("sources")]
            if cand:
                picks.append(cand[0]); owners.append(m)
        for a in picks:
            used.add(a["objectId"])
        owners = sorted(owners)
        upd = {
            "derivedFrom": [a["objectId"] for a in picks],
            "crossRefCount": len(owners),
            "mentionCount": len(owners),
            "distinctOwnerCount": len(owners),
            "evidenceQuoteCount": sum(1 for a in picks if a.get("sources")),
            "owners": owners,
            "citedWith": [],
            "layer": "sediment",
        }
        print("[dry]" if DRY else "insight", i["title"], "→", len(owners), "人，",
              upd["evidenceQuoteCount"], "条带原文：", "、".join(owners))
        if not DRY:
            call("PUT", "/schemas/KfInsight", {"className": "KfInsight", "fields": {
                "owners": {"type": "Array"}, "ownerCount": {"type": "Number"},
                "evidenceQuoteCount": {"type": "Number"}, "citedWith": {"type": "Array"}}})
            call("PUT", f"/classes/KfInsight/{i['objectId']}", upd)


if __name__ == "__main__":
    main()