kangshifu-board/legacy/tools/setup_cloud.py

#!/usr/bin/env python3
"""云端准备脚本（开发工具，不属于前端交付物）。

做的事：
  1. 补齐字段：KfArtifact.layer（四层归属）、KfProject.layer、KfArtifact.sourceIdx（原材料层样例指针）
  2. 补数：残留的 KfProbe 探针 -> 改名为标准的 KfSource（原材料层样例）；
     原文证据为空、但 evidenceLevel 标为 verified 的成果 -> 降为 supported（避免"无原文却称已验证"）
  3. 重建归纳：按 derivedFrom（-card + source 混合引用）重新计算 derivedArtifacts / derivedSources /
     owners / citedWith / evidenceQuotes / evidenceUnaligned / mentionCount
  4. 验证 PARSE_MASTER_KEY（环境变量）有效

运行时：PARSE_MASTER_KEY=... python3 tools/setup_cloud.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", [])


def ensure_fields():
    for cls, fields in [
        ("KfArtifact", {"layer": {"type": "String"}, "sourceIdx": {"type": "Number"}}),
        ("KfProject", {"layer": {"type": "String"}}),
    ]:
        if DRY:
            print("[dry] schema", cls, list(fields))
            continue
        r = call("PUT", f"/schemas/{cls}", {"className": cls, "fields": fields})
        print("schema", cls, "ok" if "fields" in r else r)


def cleanup_probes():
    probes = fetch("KfProbe")
    for p in probes:
        title = "原材料层样例 · 探针记录"
        body = {
            "layer": "raw",
            "sourceType": "probe",
            "title": title,
            "summary": "开发期连通性探针，已转为标准原材料层样例记录。",
            "rawText": p.get("note", ""),
            "sources": [],
            "tags": [],
            "payload": {},
            "meta": {"migratedFrom": "KfProbe", "kfProbeId": p["objectId"]},
        }
        if DRY:
            print("[dry] probe -> KfSource", p["objectId"])
            continue
        r = call("POST", "/classes/KfSource", body)
        if "objectId" in r:
            call("DELETE", f"/classes/KfProbe/{p['objectId']}")
            print("probe", p["objectId"], "->", r["objectId"])


def fix_evidence_levels():
    for a in fetch("KfArtifact"):
        if not a.get("sources") and a.get("evidenceLevel") == "verified":
            if DRY:
                print("[dry] downgrade", a["title"])
                continue
            call("PUT", f"/classes/KfArtifact/{a['objectId']}", {"evidenceLevel": "supported"})
            print("downgraded verified->supported:", a["title"])


def rebuild_insights():
    arts = {a["objectId"]: a for a in fetch("KfArtifact")}
    srcs = {s["objectId"]: s for s in fetch("KfSource")}
    for ins in fetch("KfInsight"):
        pointed = ins.get("derivedFrom") or []
        art_ids = [i for i in pointed if i in arts]
        src_ids = [i for i in pointed if i in srcs]
        owners = sorted({arts[i].get("ownerName") for i in art_ids if arts[i].get("ownerName")})
        lvls = sorted({arts[i].get("evidenceLevel") for i in art_ids if arts[i].get("evidenceLevel")})
        cited_with = sorted({ins["objectId"] for a in arts.values()
                             if set(a.get("sources") or []) & set(art_ids)} - {ins["objectId"]})
        upd = {
            "layer": "sediment",
            "derivedArtifacts": art_ids,
            "derivedSources": src_ids,
            "owners": owners,
            "ownerCount": len(owners),
            "crossRefCount": ins.get("crossRefCount", 0),
            "mentionCount": ins.get("crossRefCount", 0),
            "evidenceQuoteCount": len(art_ids),
            "evidenceUnaligned": "verified" not in lvls,
            "citedWith": cited_with,
        }
        if DRY:
            print("[dry] insight", ins["title"], upd)
            continue
        r = call("PUT", f"/classes/KfInsight/{ins['objectId']}", upd)
        print("insight", ins["title"], "ok" if "updatedAt" in r else r)


if __name__ == "__main__":
    if not MK:
        sys.exit("PARSE_MASTER_KEY 未设置")
    print("masterKey auth:", "ok" if "results" in call("GET", "/classes/KfProject?limit=1") else "FAILED")
    ensure_fields()
    cleanup_probes()
    fix_evidence_levels()
    rebuild_insights()