o2o-infinith-demo/scripts/monthly_ai_measure.py

#!/usr/bin/env python3
"""월간 AI 실측 체인: 질문 뱅크 러너 → 감성 판정 → 리포트 → supporter_builds.report.sentiment.

핸드오버 §2-4 의 "한 줄". 지금까지 사람이 네 개 명령을 순서대로 치던 것을 하나로 묶는다.
사람 게이트는 자동화하지 않는다. 보정(라벨링)은 여기 들어가지 않으며, 확정 여부는
calibration.confirmed 플래그로 리포트 카드에 그대로 전달한다.

  python3 scripts/monthly_ai_measure.py --clinic 뷰성형외과
  python3 scripts/monthly_ai_measure.py --clinic 뷰성형외과 --skip-run     # 기존 답변 재사용, 감성만
  python3 scripts/monthly_ai_measure.py --clinic 뷰성형외과 --build-id <uuid>  # Supabase 반영까지

한계: 질문 뱅크 러너(run_question_bank_openai.py)는 아직 뷰성형외과 전용이다
(scripts/question_bank_viewclinic.json 하드코딩). 다른 병원으로 넓히려면 러너를
먼저 병원별로 매개변수화해야 한다.
"""
import argparse, json, os, subprocess, sys, time, urllib.request, urllib.error, importlib.util

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
OUT = os.path.join(ROOT, "scripts", "out")
ENGINES = ("openai", "perplexity")

spec = importlib.util.spec_from_file_location("sqb", os.path.join(ROOT, "scripts", "sentiment_qb.py"))
sqb = importlib.util.module_from_spec(spec); spec.loader.exec_module(sqb)


def sh(argv):
    print("  $ " + " ".join(argv), flush=True)
    r = subprocess.run(argv, cwd=ROOT)
    if r.returncode != 0: sys.exit(f"실패: {' '.join(argv)}")


def patch_supabase(build_id, sentiment):
    """supporter_builds.report 에 sentiment 키만 병합한다. 다른 단계가 쓴 키를 지우지 않는다."""
    sqb.load_env()
    url = os.environ.get("VITE_SUPABASE_URL"); key = os.environ.get("SUPABASE_SERVICE_ROLE_KEY")
    if not (url and key): sys.exit("VITE_SUPABASE_URL / SUPABASE_SERVICE_ROLE_KEY 필요 (.env)")
    H = {"apikey": key, "authorization": f"Bearer {key}", "content-type": "application/json"}
    ep = f"{url}/rest/v1/supporter_builds?id=eq.{build_id}"
    req = urllib.request.Request(ep + "&select=report", headers=H)
    with urllib.request.urlopen(req, timeout=30, context=sqb.ssl_context()) as r:
        rows = json.loads(r.read().decode())
    if not rows: sys.exit(f"supporter_builds 에 {build_id} 없음")
    report = rows[0].get("report") or {}
    report["sentiment"] = sentiment
    body = json.dumps({"report": report, "updated_at": time.strftime("%Y-%m-%dT%H:%M:%S")}).encode()
    req = urllib.request.Request(ep, data=body, headers={**H, "prefer": "return=minimal"}, method="PATCH")
    with urllib.request.urlopen(req, timeout=30, context=sqb.ssl_context()) as r:
        print(f"  supporter_builds.report.sentiment 갱신 (HTTP {r.status})")


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--clinic", default="뷰성형외과")
    ap.add_argument("--alias", default="뷰성형외과의원,뷰 성형외과,View Plastic Surgery,viewclinic")
    ap.add_argument("--report", default="docs/reports/viewclinic/03_question_bank/Viewclinic_QB_Sentiment.md")
    ap.add_argument("--build-id", help="주면 supporter_builds.report.sentiment 에 반영한다")
    ap.add_argument("--skip-run", action="store_true", help="질문 뱅크 러너를 건너뛰고 기존 답변으로 감성만 다시 낸다")
    ap.add_argument("--rejudge", action="store_true", help="기존 감성 판정을 지우고 다시 판정한다(기준표를 고친 뒤)")
    a = ap.parse_args()

    if not a.skip_run:
        print("[1/3] 질문 뱅크 실측")
        for e in ENGINES:
            sh([sys.executable, "scripts/run_question_bank_openai.py", "--engine", e])
    else:
        print("[1/3] 질문 뱅크 실측 건너뜀 (--skip-run)")

    print("[2/3] 감성 판정")
    for e in ENGINES:
        src = os.path.join(OUT, f"qb_{e}_results.jsonl")
        dst = os.path.join(OUT, f"qb_sentiment_{e}.jsonl")
        if not os.path.exists(src): sys.exit(f"답변 파일 없음: {src}")
        if a.rejudge and os.path.exists(dst): os.remove(dst)
        sh([sys.executable, "scripts/sentiment_qb.py", "--in", src, "--out", dst,
            "--clinic", a.clinic, "--alias", a.alias])

    print("[3/3] 리포트")
    outs = [os.path.join(OUT, f"qb_sentiment_{e}.jsonl") for e in ENGINES]
    sh([sys.executable, "scripts/sentiment_qb.py", "--summary", *outs,
        "--report", a.report, "--clinic", a.clinic])

    rows = [json.loads(l) for p in outs for l in open(p, encoding="utf-8") if l.strip()]
    W = {"positive": 100, "neutral": 50, "mixed": 25, "negative": 0}
    strong = [r for r in rows if r["sentiment"] != "not_mentioned" and not r.get("weak_mention")]
    s = sqb.score(rows)
    sentiment = {
        "rubric": sqb.RUBRIC_VERSION, "model": rows[0].get("model"),
        "measuredAt": time.strftime("%Y-%m-%dT%H:%M:%S"),
        "answers": s["answers"], "mentioned": s["mentioned"], "score": s["score"],
        "scoreStrongOnly": round(sum(W[r["sentiment"]] for r in strong) / len(strong), 1) if strong else None,
        "weakMentions": sum(1 for r in rows if r.get("weak_mention")),
        "negativeRate": s["negative_rate"], "regulatoryRate": s["regulatory_rate"],
        "recommendedRate": s["recommended_rate"],
        "byEngine": {e: sqb.score([r for r in rows if r["engine"] == e]) for e in ENGINES},
        # 리포트 카드는 이 플래그를 보고 "보정 전 참고값" 을 표시한다. 2차 보정 전에는 확정이 아니다.
        "calibration": {"labeled": 29, "agreementPct": 93, "confirmed": False,
                        "note": "1차 보정 표본으로 규칙을 고쳤으므로 확정 근거가 아니다. 새 30건 2차 보정 필요."},
    }
    p = os.path.join(OUT, "sentiment_summary.json")
    open(p, "w", encoding="utf-8").write(json.dumps(sentiment, ensure_ascii=False, indent=2))
    print(f"  요약 → {p}")
    print(f"  언급 {s['mentioned']}/{s['answers']} · 감성 점수 {s['score']} · 약한 언급 제외 {sentiment['scoreStrongOnly']}")

    if a.build_id: patch_supabase(a.build_id, sentiment)
    else: print("  --build-id 없음. Supabase 반영은 건너뛴다.")


if __name__ == "__main__":
    main()