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")
ALL_ENGINES = ("openai", "perplexity", "naver_briefing")
# naver_briefing 은 SerpApi(제3자 SERP API)로 네이버 AI 브리핑을 조회한다(haewon 결정 2026-09-15).
# SERPAPI_API_KEY 가 없으면 건너뛴다. 결과를 고객 리포트에 쓰기 전 법무 확인이 필요하다(run_question_bank_openai.py 머리말).

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 naver_briefing_summary(path):
    """네이버 AI 브리핑 노출률·브리핑 내 언급률·인용 출처 상위. 브리핑이 없는 질문과 언급 없음을 나눠 센다."""
    rows = [json.loads(l) for l in open(path, encoding="utf-8") if l.strip()]
    shown = [r for r in rows if r.get("briefing")]
    src = {}
    for r in shown:
        for s in {(x.get("source") or x.get("link") or "").strip() for x in r.get("sources") or []}:
            if s: src[s] = src.get(s, 0) + 1
    brand = lambda r: "뷰성형외과" in r.get("question", "")
    nb_shown = [r for r in shown if not brand(r)]
    return {
        "source": "SerpApi naver_ai_overview", "device": rows[0].get("device") if rows else None,
        "measuredAt": time.strftime("%Y-%m-%dT%H:%M:%S"),
        "questions": len(rows), "briefingShown": len(shown),
        "mentionedInBriefing": sum(1 for r in shown if r.get("view_mentioned")),
        "nonBrandBriefingShown": len(nb_shown),
        "nonBrandMentioned": sum(1 for r in nb_shown if r.get("view_mentioned")),
        "topSources": [{"source": k, "answers": v} for k, v in sorted(src.items(), key=lambda x: -x[1])[:10]],
        "legalReviewed": False,
    }


def patch_supabase(build_id, sentiment, extra=None):
    """supporter_builds.report 에 sentiment(와 extra 키)만 병합한다. 다른 단계가 쓴 키를 지우지 않는다."""
    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
    for k, v in (extra or {}).items():
        report[k] = v
    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{''.join(' · ' + k for k in (extra or {}))} (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="기존 감성 판정을 지우고 다시 판정한다(기준표를 고친 뒤)")
    ap.add_argument("--engines", default=",".join(ALL_ENGINES), help="쉼표 구분. 기본: openai,perplexity,naver_briefing")
    a = ap.parse_args()
    sqb.load_env()
    ENGINES = [e for e in a.engines.split(",") if e in ALL_ENGINES]
    if "naver_briefing" in ENGINES and not os.environ.get("SERPAPI_API_KEY"):
        print("  SERPAPI_API_KEY 없음. naver_briefing 은 건너뛴다.")
        ENGINES.remove("naver_briefing")

    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']}")

    extra = {}
    nb_path = os.path.join(OUT, "qb_naver_briefing_results.jsonl")
    if "naver_briefing" in ENGINES and os.path.exists(nb_path):
        extra["naverBriefing"] = nb = naver_briefing_summary(nb_path)
        open(os.path.join(OUT, "naver_briefing_summary.json"), "w", encoding="utf-8").write(json.dumps(nb, ensure_ascii=False, indent=2))
        print(f"  네이버 AI 브리핑 노출 {nb['briefingShown']}/{nb['questions']} · 브리핑 내 뷰 언급 {nb['mentionedInBriefing']} · 논브랜드 {nb['nonBrandMentioned']}/{nb['nonBrandBriefingShown']}")

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


if __name__ == "__main__":
    main()