#!/usr/bin/env python3 """외국인 관점 영문 Question Bank 답변엔진 실측 러너 (OpenAI / Perplexity). run_question_bank_openai.py의 영문·해외 위치 변형. 차이점: - 질문 파일·출력 폴더를 인자로 받는다 (기본: docs/reports/viewclinic/05_global_en) - OpenAI web_search user_location을 --country/--city로 지정 (기본 US / New York) - 병원명 추출 프롬프트가 영문 표기 변형(VIEW Plastic Surgery, View Clinic, VIEW Seoul)을 같은 병원으로 본다 - 인용 URL을 도메인별로 집계해 "AI가 어떤 표면을 근거로 쓰는가"를 기록한다 사용: python3 scripts/run_question_bank_en.py --engine perplexity python3 scripts/run_question_bank_en.py --engine openai --country US --city "New York" python3 scripts/run_question_bank_en.py --engine openai --summarize """ import argparse, json, os, re, sys, time from urllib.parse import urlparse sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) import run_question_bank_openai as base # noqa: E402 ROOT = base.ROOT DEFAULT_DIR = os.path.join(ROOT, "docs", "reports", "viewclinic", "05_global_en") EXTRACT_PROMPT = """Below is an answer from an AI search engine. Extract the names of plastic surgery clinics / hospitals mentioned, in order of appearance, up to 8. - Clinic names only (exclude platforms, apps, media, agencies: e.g. RealSelf, Reddit, Gangnam Unni, Bookimed, Whatclinic are NOT clinics) - "VIEW Plastic Surgery" also appears as View Clinic, VIEW Clinic Korea, VIEW Seoul, VIEW Plastic Surgery Clinic, 뷰성형외과. Treat all as the same clinic and normalize to "VIEW Plastic Surgery". - Output JSON only: {"clinics": ["Clinic 1", ...], "view_mentioned": true/false, "view_rank": position or null, "view_sentiment": "positive"|"neutral"|"negative"|null} - view_sentiment: how the answer characterizes VIEW (null if not mentioned) Answer: """ class OpenAIEngineLoc(base.OpenAIEngine): def __init__(self, model, search_tool, country, city, tz): super().__init__(model, search_tool) self.loc = {"type": "approximate", "country": country, "city": city, "timezone": tz} def ask(self, question): tool = {"type": self.search_tool, "user_location": self.loc} try: resp = base.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): self.search_tool = "web_search_preview" return self.ask(question) raise return base.openai_answer_text(resp) def extract_clinics(answer): okey = os.environ.get("OPENAI_API_KEY") if not okey: sys.exit("OPENAI_API_KEY needed for extraction") resp = base.post_retry("https://api.openai.com/v1/responses", {"model": "gpt-4o-mini", "input": EXTRACT_PROMPT + answer[:9000]}, {"Authorization": f"Bearer {okey}"}) text, _ = base.openai_answer_text(resp) m = re.search(r"\{.*\}", text or "", re.S) try: return json.loads(m.group(0)) if m else {"clinics": [], "view_mentioned": False, "view_rank": None, "view_sentiment": None} except json.JSONDecodeError: return {"clinics": [], "view_mentioned": False, "view_rank": None, "view_sentiment": None, "parse_error": text[:200]} def domain(u): try: d = urlparse(u).netloc.lower() return d[4:] if d.startswith("www.") else d except Exception: return u def out_path(d, engine): return os.path.join(d, f"qb_en_{engine}_results.jsonl") def load_done(d, engine): done = {} p = out_path(d, 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(d, engine, qfile): done = load_done(d, engine) qs = json.load(open(qfile)) cats = {"A": "Brand", "B": "Travel & logistics", "C": "Recommendation & comparison", "D": "Procedure info", "E": "Pricing", "F": "Safety & trust", "G": "Aftercare & remote follow-up"} freq, dom_freq, view_dom = {}, {}, {} per_cat = {} lines = [f"# VIEW Plastic Surgery EN Question Bank results: {engine}", "", f"Measured {time.strftime('%Y-%m-%d')} / {len(done)} questions", "", "| # | Question | VIEW mentioned | Rank | Sentiment | Clinics (in order) | Cited domains |", "|---|---|---|---|---|---|---|"] for q in qs: r = done.get(q["id"]) if not r: continue c = per_cat.setdefault(q["cat"], [0, 0, 0]); c[0] += 1 clinics = [x for x in r.get("clinics", []) if isinstance(x, str)][:8] for i, x in enumerate(clinics): freq.setdefault(x, [0, 0]); freq[x][0] += 1 if i == 0: freq[x][1] += 1 doms = sorted({domain(u) for u in r.get("urls", [])}) for dm in doms: dom_freq[dm] = dom_freq.get(dm, 0) + 1 vm = bool(r.get("view_mentioned")) if vm: c[1] += 1 for dm in doms: view_dom[dm] = view_dom.get(dm, 0) + 1 if any("viewplasticsurgery.com" in u or "viewclinic" in u for u in r.get("urls", [])): c[2] += 1 lines.append(f"| {q['id']} | {q['question']} | {'O' if vm else 'X'} | {r.get('view_rank') or '-'} | {r.get('view_sentiment') or '-'} | " f"{', '.join(clinics) if clinics else '(none)'} | {', '.join(doms[:6])} |") mentioned = sum(v[1] for v in per_cat.values()) lines += ["", "## Summary", "", f"- VIEW mentioned: {mentioned}/{len(done)} ({mentioned*100//max(len(done),1)}%)", "", "| Category | Questions | VIEW mentioned | viewplasticsurgery.com cited |", "|---|---|---|---|"] for k in sorted(per_cat): n, m, cited = per_cat[k] lines.append(f"| {k} {cats.get(k,'')} | {n} | {m} ({m*100//n}%) | {cited} ({cited*100//n}%) |") lines += ["", "### Clinic mention frequency (top 20)", "", "| Clinic | Questions | First position |", "|---|---|---|"] for name, (cnt, first) in sorted(freq.items(), key=lambda x: -x[1][0])[:20]: lines.append(f"| {name} | {cnt} | {first} |") lines += ["", "### Cited domains (top 25)", "", "| Domain | Questions cited | Cited when VIEW mentioned |", "|---|---|---|"] for dm, cnt in sorted(dom_freq.items(), key=lambda x: -x[1])[:25]: lines.append(f"| {dm} | {cnt} | {view_dom.get(dm, 0)} |") out_md = os.path.join(d, f"QB_en_{engine}_results.md") open(out_md, "w").write("\n".join(lines) + "\n") json.dump({"per_cat": per_cat, "freq": freq, "dom_freq": dom_freq, "view_dom": view_dom, "n": len(done), "mentioned": mentioned}, open(os.path.join(d, f"QB_en_{engine}_summary.json"), "w"), ensure_ascii=False, indent=1) print(f"saved: {out_md}") def main(): ap = argparse.ArgumentParser() ap.add_argument("--engine", choices=["openai", "perplexity"], default="perplexity") ap.add_argument("--model", default=None) ap.add_argument("--search-tool", default="web_search") ap.add_argument("--country", default="US"); ap.add_argument("--city", default="New York"); ap.add_argument("--tz", default="America/New_York") ap.add_argument("--dir", default=DEFAULT_DIR) ap.add_argument("--questions", default=os.path.join(DEFAULT_DIR, "question_bank_en.json")) ap.add_argument("--limit", type=int, default=0); ap.add_argument("--sleep", type=float, default=1.0) ap.add_argument("--summarize", action="store_true") a = ap.parse_args() base.load_dotenv() if a.summarize: summarize(a.dir, a.engine, a.questions); return eng = OpenAIEngineLoc(a.model or "gpt-4o", a.search_tool, a.country, a.city, a.tz) if a.engine == "openai" \ else base.PerplexityEngine(a.model or "sonar") os.makedirs(a.dir, exist_ok=True) qs = json.load(open(a.questions)); done = load_done(a.dir, a.engine) todo = [q for q in qs if q["id"] not in done] if a.limit: todo = todo[:a.limit] print(f"{eng.name}: {len(todo)} to run ({len(done)} done)", flush=True) with open(out_path(a.dir, a.engine), "a") as f: for i, q in enumerate(todo, 1): t0 = time.time() try: answer, urls = eng.ask(q["question"]) ext = extract_clinics(answer) except Exception as e: print(f" [{i}/{len(todo)}] {q['id']} ERROR {e}", flush=True); time.sleep(5); continue row = {"id": q["id"], "cat": q["cat"], "question": q["question"], "engine": eng.name, "answer": answer, "urls": urls, "measured_at": time.strftime("%Y-%m-%dT%H:%M:%S"), **ext} f.write(json.dumps(row, ensure_ascii=False) + "\n"); f.flush() print(f" [{i}/{len(todo)}] {q['id']} view={'O' if ext.get('view_mentioned') else 'X'} rank={ext.get('view_rank')} " f"clinics={ext.get('clinics', [])[:4]} urls={len(urls)} ({time.time()-t0:.0f}s)", flush=True) time.sleep(a.sleep) summarize(a.dir, a.engine, a.questions) if __name__ == "__main__": main()