宜博 / pixelphone

#!/usr/bin/env python3
"""VERIFY: 分类器置信度/分数「经验校准曲线」+ 阈值单一事实来源
（pplx-decider-v1.1-27b 决策模型知识, learn_queue #bcf45b01, 2026-10-13 Step6 审计）

学到的知识（原文 L4/L6/L10）：
  校准比分数本身更值钱——「说 90% 把握时真实准确率也接近 90%」才敢设阈值、自动放行，
  否则概率只是装饰品。分类器/分流这类判定应输出**可校准**的量并实测其校准误差。

病理 WHERE=learn_queue 分类路由阈值决策（app.py 分类回调 + local_db 校准统计）
     WHY=① 只有 `confidence=high` 的**失败计数**（confidence_high_failed），没有**分档成功率**
            曲线 ⇒ 「high 到底可不可信」不可测量，阈值 = 拍板常量而非校准结果。
         ② high_precision 分母只含 applied+rejected，**漏掉 failed**（最大失败档）
            ⇒ 校准 KPI 自身有盲区（假阴）。
         ③ 阈值字面量 `score >= 7` / `>= 4` 散落 5 处 ⇒ 校准结论就算有了也无处落。
         ④ 校准函数无样本量字段：decided=1 也回 100%，小样本假置信。

修复：
 1) local_db.learn_calibration_stats() 增经验校准曲线 confidence_calibration（分档成功率）
    与确定性 verdict（MISCALIBRATED / BELOW_TARGET / CALIBRATED / INSUFFICIENT_SAMPLE）
 2) 分母含 failed；补 n_decided / reliable（小样本不出结论）
 3) 阈值常量 LEARN_ROUTE_LEARN / LEARN_ROUTE_REVIEW 单一事实来源，替换散落字面量

TDD：
TC1 (RED/复现旧病理)    旧 stats 键集合无分档成功率、无 verdict、无样本量 → 不可校准
TC2 (GREEN/真实数据)    真实 DB：3 档成功率 + verdict 计算（确定性，来自 DB 事实）
TC3 (GREEN/判别力)      防伪：构造 high 优于 medium 的库 → 不得判 MISCALIBRATED
TC4 (GREEN/小样本守卫)  每档 n < CALIBRATION_MIN_SAMPLE → INSUFFICIENT_SAMPLE + reliable=False
TC5 (GREEN/分母含 failed) 只 failed 的高分样本不得被算成 100% 精度
TC6 (源码守卫/防回退)   阈值单一事实来源：app.py 无裸 `score >= 7` 路由字面量
TC7 (冻结重跑 spot check) 同库连跑 3 次结果一致（确定性纯函数，可复现）
"""
import importlib
import pathlib
import sqlite3
import sys

sys.path.insert(0, "/data/data/com.termux/files/home/wechat-viewer")

APP = pathlib.Path("/data/data/com.termux/files/home/wechat-viewer/app.py")
LOCAL_DB = pathlib.Path("/data/data/com.termux/files/home/wechat-viewer/local_db.py")
FAIL = 0


def check(name, ok, detail=""):
    global FAIL
    print(f"{'✅' if ok else '❌'} {name}" + (f" — {detail}" if detail else ""))
    if not ok:
        FAIL += 1


# ── TC1: RED —— 复现旧实现（缺分档成功率 / verdict / 样本量）──
OLD_KEYS = {"high_precision", "high_applied", "high_rejected", "high_pending",
            "mid_applied", "mid_rejected", "confidence_high_failed",
            "score_distribution", "note"}
missing = {"confidence_calibration", "confidence_verdict", "threshold_at"} - OLD_KEYS
check("TC1 RED 复现旧病理：旧 stats 缺校准曲线/verdict/阈值实测",
      len(missing) == 3, f"缺 {sorted(missing)} ⇒ 置信度不可校准、阈值无依据")

# ── 载入实现 ──
import config
import local_db
importlib.reload(local_db)

stats = local_db.learn_calibration_stats()

# ── TC6 源码守卫：阈值单一事实来源 ──
app_src = APP.read_text()
check("TC6a 源码守卫：app.py 路由处无裸 `score >= 7` 字面量",
      "score >= 7" not in app_src,
      f"残留 {app_src.count('score >= 7')} 处")
check("TC6b 源码守卫：app.py 路由使用 local_db.LEARN_ROUTE_* 常量",
      "LEARN_ROUTE_LEARN" in app_src and "LEARN_ROUTE_REVIEW" in app_src)
ldb_src = LOCAL_DB.read_text()
check("TC6c 源码守卫：local_db 定义阈值常量（单一事实来源）",
      "LEARN_ROUTE_LEARN = " in ldb_src and "LEARN_ROUTE_REVIEW = " in ldb_src)
check("TC6d 源码守卫：小样本判据常量存在（禁止小样本出结论）",
      "CALIBRATION_MIN_SAMPLE = " in ldb_src)

# ── TC2 GREEN：真实数据校准曲线 ──
curve = stats.get("confidence_calibration") or {}
check("TC2a 真实数据：三档 confidence 均有校准条目",
      set(curve) >= {"high", "medium", "low"}, f"buckets={sorted(curve)}")
ok_shape = bool(curve) and all({"n_decided", "applied", "failed", "rejected", "success_rate"} <= set(v)
               for v in curve.values())
check("TC2b 真实数据：每条含 n_decided/成功率（可算校准误差）", ok_shape)
hi, mid = curve.get("high", {}), curve.get("medium", {})
print(f"    high: n={hi.get('n_decided')} rate={hi.get('success_rate')} "
      f"(applied={hi.get('applied')} failed={hi.get('failed')} rejected={hi.get('rejected')})")
print(f"    medium: n={mid.get('n_decided')} rate={mid.get('success_rate')}")
print(f"    verdict={stats.get('confidence_verdict')} threshold_at={stats.get('threshold_at')}")
# 成功率必须是真实计算值（0..1），不是装饰
rates_ok = bool(curve) and all(v["success_rate"] is None or 0.0 <= v["success_rate"] <= 1.0
               for v in curve.values())
check("TC2c 真实数据：success_rate ∈ [0,1] 且由 DB 计数导出", rates_ok)
check("TC2d 真实数据：verdict 为确定性命名单之一",
      stats.get("confidence_verdict") in
      {"CALIBRATED", "BELOW_TARGET", "MISCALIBRATED", "INSUFFICIENT_SAMPLE"},
      str(stats.get("confidence_verdict")))
check("TC2e 真实数据：阈值处实测精度带样本量（非裸百分比）",
      isinstance(stats.get("threshold_at"), dict)
      and "n_decided" in stats["threshold_at"] and "reliable" in stats["threshold_at"],
      str(stats.get("threshold_at")))
# 实测：high 档成功率 < medium 档 ⇒ 置信度反相关，verdict 必须点名
if hi.get("n_decided", 0) >= local_db.CALIBRATION_MIN_SAMPLE and mid.get("n_decided", 0) >= local_db.CALIBRATION_MIN_SAMPLE:
    expected = "MISCALIBRATED" if (hi["success_rate"] or 0) < (mid["success_rate"] or 0) else stats["confidence_verdict"]
    check("TC2f 真实数据：high<medium 时 verdict 点名 MISCALIBRATED",
          stats["confidence_verdict"] == expected,
          f"high={hi['success_rate']} medium={mid['success_rate']} → {stats['confidence_verdict']}")


# ── 造库助手（直接指向临时 DB）──
def make_db(path, rows):
    p = pathlib.Path(path)
    if p.exists():
        p.unlink()
    c = sqlite3.connect(str(p))
    c.execute("""CREATE TABLE learn_queue (
        id INTEGER PRIMARY KEY AUTOINCREMENT, article_url TEXT, title TEXT,
        dedup_key TEXT, learn_type TEXT, learn_target TEXT, learn_content TEXT,
        outline TEXT, confidence TEXT DEFAULT 'medium', score INTEGER DEFAULT 0,
        status TEXT DEFAULT 'pending_review', error_msg TEXT,
        created_at TEXT DEFAULT (datetime('now','localtime')),
        reviewed_at TEXT)""")
    for i, (conf, score, status) in enumerate(rows):
        c.execute("INSERT INTO learn_queue (dedup_key,learn_type,confidence,score,status)"
                  " VALUES (?,?,?,?,?)", (f"k{i}", "memory", conf, score, status))
    c.commit()
    c.close()
    return str(p)


_real_db = config.LOCAL_DB

# ── TC3 GREEN：防伪 —— high 确实优于 medium 时不得误报 MISCALIBRATED ──
good = [("high", 9, "applied")] * 30 + [("high", 9, "rejected")] * 2 + \
       [("medium", 5, "applied")] * 5 + [("medium", 5, "failed")] * 25
config.LOCAL_DB = make_db("/data/data/com.termux/files/home/tmp/tc3_calib.db", good)
importlib.reload(local_db)
s3 = local_db.learn_calibration_stats()
check("TC3 判别力：high(0.938)>medium(0.167) → CALIBRATED（不误报）",
      s3["confidence_verdict"] == "CALIBRATED", str(s3["confidence_verdict"]))

# ── TC4 GREEN：小样本守卫 ──
tiny = [("high", 9, "applied"), ("medium", 5, "applied"), ("low", 2, "rejected")]
config.LOCAL_DB = make_db("/data/data/com.termux/files/home/tmp/tc4_calib.db", tiny)
importlib.reload(local_db)
s4 = local_db.learn_calibration_stats()
check("TC4a 小样本：n<MIN_SAMPLE → INSUFFICIENT_SAMPLE（不出结论）",
      s4["confidence_verdict"] == "INSUFFICIENT_SAMPLE", str(s4["confidence_verdict"]))
check("TC4b 小样本：threshold_at.reliable=False，不把 1 样本算成 100%",
      s4["threshold_at"]["reliable"] is False,
      f"n={s4['threshold_at']['n_decided']} rate={s4['threshold_at']['success_rate']}")

# ── TC5 GREEN：分母含 failed（旧 high_precision 盲区）──
only_failed = [("high", 9, "failed")] * 25
config.LOCAL_DB = make_db("/data/data/com.termux/files/home/tmp/tc5_calib.db", only_failed)
importlib.reload(local_db)
s5 = local_db.learn_calibration_stats()
check("TC5 分母含 failed：全失败的高分档不得算成高精度",
      s5["threshold_at"]["success_rate"] == 0.0 and s5["threshold_at"]["n_decided"] == 25,
      f"rate={s5['threshold_at']['success_rate']} n={s5['threshold_at']['n_decided']} "
      f"（旧口径 decided=0 → high_precision=None 假阴）")

# ── TC7 冻结重跑 spot check：确定性 3/3 ──
config.LOCAL_DB = _real_db
importlib.reload(local_db)
runs = [local_db.learn_calibration_stats() for _ in range(3)]
same = all(r == runs[0] for r in runs)
check("TC7 冻结重跑 3/3 一致（确定性，可复现）", same,
      f"verdict={runs[0]['confidence_verdict']}")

print()
print(f"{'FAIL: '+str(FAIL) if FAIL else 'ALL PASS'} ({7} TC 组)")
sys.exit(1 if FAIL else 0)