#!/usr/bin/env python3 """뷰성형외과 Question Bank 120문항 답변엔진 실측 러너 (OpenAI / Perplexity). 측정 방식: - openai: Responses API + web_search 도구 (서울 위치 고정) - perplexity: chat/completions, model=sonar (자체 웹검색 내장) ChatGPT 소비자 UI 자동 조회는 약관 위반이므로 쓰지 않는다. 각 질문을 원문 그대로 1회 질의하고, 답변에서 언급된 병원명 상위 5개를 후처리 모델(OpenAI gpt-4o-mini, 없으면 Gemini 2.5 Flash)로 추출해 JSONL 기록. 재실행 시 완료된 문항은 건너뛴다. 사용: python3 scripts/run_question_bank_openai.py --engine perplexity OPENAI_API_KEY=... python3 scripts/run_question_bank_openai.py --engine openai python3 scripts/run_question_bank_openai.py --engine openai --summarize """ import argparse import json import os import re import ssl import sys import time import urllib.error import urllib.request try: import certifi SSL_CTX = ssl.create_default_context(cafile=certifi.where()) except ImportError: SSL_CTX = ssl.create_default_context() ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) QUESTIONS = os.path.join(ROOT, "scripts", "question_bank_viewclinic.json") OUT_DIR = os.path.join(ROOT, "scripts", "out") EXTRACT_PROMPT = """아래는 AI 검색의 답변이다. 이 답변에서 언급된 성형외과·의원·클리닉 이름을 등장 순서대로 최대 5개 추출하라. - 병원 이름만 (플랫폼·앱·언론사 제외. 예: 강남언니, 바비톡, 네이버는 병원이 아님) - "뷰성형외과"는 VIEW성형외과, View Plastic Surgery 등 표기 변형도 동일 병원으로 본다 - JSON만 출력: {"clinics": ["병원1", ...], "view_mentioned": true/false, "view_rank": 순번 또는 null} 답변: """ def load_dotenv(): path = os.path.join(ROOT, ".env") if not os.path.exists(path): return for line in open(path): line = line.strip() if line and not line.startswith("#") and "=" in line: k, v = line.split("=", 1) os.environ.setdefault(k.strip(), v.strip().strip('"').strip("'")) def post(url, payload, headers, timeout=180): req = urllib.request.Request(url, data=json.dumps(payload).encode(), headers={"Content-Type": "application/json", **headers}) with urllib.request.urlopen(req, timeout=timeout, context=SSL_CTX) as r: return json.loads(r.read()) def post_retry(url, payload, headers, tries=4): for i in range(tries): try: return post(url, payload, headers) except urllib.error.HTTPError as e: body = e.read().decode(errors="replace") if e.code in (429, 500, 502, 503) and i < tries - 1: wait = 15 * (i + 1) print(f" HTTP {e.code}, {wait}s 대기 후 재시도", flush=True) time.sleep(wait) continue raise RuntimeError(f"HTTP {e.code}: {body[:500]}") from e except (urllib.error.URLError, TimeoutError): if i < tries - 1: time.sleep(10) continue raise # ---------- engines ---------- def openai_answer_text(resp): texts, urls = [], [] for item in resp.get("output", []): if item.get("type") != "message": continue for c in item.get("content", []): if c.get("type") == "output_text": texts.append(c.get("text", "")) for a in c.get("annotations", []): if a.get("type") == "url_citation" and a.get("url"): urls.append(a["url"]) return "\n".join(texts).strip(), urls class OpenAIEngine: name = "openai" def __init__(self, model, search_tool): self.key = os.environ.get("OPENAI_API_KEY") if not self.key: sys.exit("OPENAI_API_KEY가 필요합니다 (.env 또는 환경변수)") self.model = model self.search_tool = search_tool def ask(self, question): tool = {"type": self.search_tool, "user_location": {"type": "approximate", "country": "KR", "city": "Seoul", "timezone": "Asia/Seoul"}} try: resp = post_retry("https://api.openai.com/v1/responses", {"model": self.model, "tools": [tool], "input": question}, {"Authorization": f"Bearer {self.key}"}) except RuntimeError as e: if self.search_tool == "web_search" and "web_search" in str(e): print(" web_search 거부됨, web_search_preview로 폴백", flush=True) self.search_tool = "web_search_preview" return self.ask(question) raise return openai_answer_text(resp) class PerplexityEngine: name = "perplexity" def __init__(self, model): self.key = os.environ.get("PERPLEXITY_API_KEY") if not self.key: sys.exit("PERPLEXITY_API_KEY가 필요합니다 (.env)") self.model = model def ask(self, question): resp = post_retry("https://api.perplexity.ai/chat/completions", {"model": self.model, "messages": [{"role": "user", "content": question}]}, {"Authorization": f"Bearer {self.key}"}) text = resp.get("choices", [{}])[0].get("message", {}).get("content", "").strip() urls = list(resp.get("citations") or []) for s in resp.get("search_results") or []: if s.get("url"): urls.append(s["url"]) return text, urls # ---------- extraction ---------- def extract_clinics(answer): prompt = EXTRACT_PROMPT + answer[:8000] text = None okey = os.environ.get("OPENAI_API_KEY") if okey: resp = post_retry("https://api.openai.com/v1/responses", {"model": "gpt-4o-mini", "input": prompt}, {"Authorization": f"Bearer {okey}"}) text, _ = openai_answer_text(resp) else: gkey = os.environ.get("GEMINI_API_KEY") if not gkey: sys.exit("추출용으로 OPENAI_API_KEY 또는 GEMINI_API_KEY가 필요합니다") resp = post_retry( "https://generativelanguage.googleapis.com/v1beta/models/gemini-2.5-flash:generateContent?key=" + gkey, {"contents": [{"parts": [{"text": prompt}]}]}, {}) parts = resp.get("candidates", [{}])[0].get("content", {}).get("parts", []) text = "\n".join(p.get("text", "") for p in parts) m = re.search(r"\{.*\}", text or "", re.S) if not m: return {"clinics": [], "view_mentioned": False, "view_rank": None, "parse_error": (text or "")[:200]} try: return json.loads(m.group(0)) except json.JSONDecodeError: return {"clinics": [], "view_mentioned": False, "view_rank": None, "parse_error": text[:200]} # ---------- io / summary ---------- def out_path(engine): return os.path.join(OUT_DIR, f"qb_{engine}_results.jsonl") def load_done(engine): done = {} p = out_path(engine) if os.path.exists(p): for line in open(p): if line.strip(): row = json.loads(line) done[row["id"]] = row return done def summarize(engine): done = load_done(engine) if not done: print(f"{engine}: 결과 없음") return label = {"openai": "OpenAI 웹검색(Responses API + web_search, 서울 고정)", "perplexity": "Perplexity API (sonar)"}[engine] qs = json.load(open(QUESTIONS)) freq, mentioned = {}, 0 lines = [f"# 뷰성형외과 Question Bank 실측 결과: {engine}", "", f"측정일: {time.strftime('%Y-%m-%d')} / 방식: {label} / {len(done)}문항 완료", "", "| # | 질문 | 사전판정 | 뷰 언급 | 뷰 순위 | 답변 내 상위 병원 (등장순) |", "|---|---|---|---|---|---|"] for q in qs: r = done.get(q["id"]) if not r: continue clinics = [c for c in r.get("clinics", []) if isinstance(c, str)][:5] for i, c in enumerate(clinics): freq.setdefault(c, [0, 0]) freq[c][0] += 1 if i == 0: freq[c][1] += 1 vm = r.get("view_mentioned") if vm: mentioned += 1 rank = r.get("view_rank") lines.append(f"| {q['id']} | {q['question']} | {q['prior']} | {'O' if vm else 'X'} | " f"{rank if rank else '-'} | {', '.join(clinics) if clinics else '(병원 언급 없음)'} |") lines += ["", "## 집계", "", f"- 뷰성형외과 언급: {mentioned}/{len(done)}문항 ({mentioned*100//max(len(done),1)}%)", "", "### 병원별 언급 빈도 (상위 15)", "", "| 병원 | 언급 질문 수 | 1순위 등장 수 |", "|---|---|---|"] for name, (cnt, first) in sorted(freq.items(), key=lambda x: -x[1][0])[:15]: lines.append(f"| {name} | {cnt} | {first} |") out_md = os.path.join(ROOT, "docs", "reports", f"Viewclinic_QB_{engine}_results.md") open(out_md, "w").write("\n".join(lines) + "\n") print(f"저장: {out_md}") def main(): ap = argparse.ArgumentParser() ap.add_argument("--engine", choices=["openai", "perplexity"], default="openai") ap.add_argument("--model", default=None) ap.add_argument("--search-tool", default="web_search") ap.add_argument("--limit", type=int, default=0) ap.add_argument("--sleep", type=float, default=1.5) ap.add_argument("--summarize", action="store_true") args = ap.parse_args() load_dotenv() if args.summarize: summarize(args.engine) return if args.engine == "openai": eng = OpenAIEngine(args.model or "gpt-4o", args.search_tool) else: eng = PerplexityEngine(args.model or "sonar") os.makedirs(OUT_DIR, exist_ok=True) qs = json.load(open(QUESTIONS)) done = load_done(args.engine) todo = [q for q in qs if q["id"] not in done] if args.limit: todo = todo[: args.limit] print(f"[{args.engine}] 전체 {len(qs)} / 완료 {len(done)} / 이번 실행 {len(todo)}") with open(out_path(args.engine), "a") as f: for n, q in enumerate(todo, 1): print(f"[{n}/{len(todo)}] {q['id']} {q['question'][:30]}", flush=True) answer, urls = eng.ask(q["question"]) ext = extract_clinics(answer) if answer else { "clinics": [], "view_mentioned": False, "view_rank": None} row = {"id": q["id"], "question": q["question"], "prior": q["prior"], "engine": args.engine, "model": eng.model, "answer": answer, "citations": urls[:10], **ext, "ts": time.strftime("%Y-%m-%dT%H:%M:%S")} f.write(json.dumps(row, ensure_ascii=False) + "\n") f.flush() time.sleep(args.sleep) summarize(args.engine) if __name__ == "__main__": main()