5faf867b05
네이버 AI 브리핑은 공식 API가 없어 지금까지 답변 노출을 재지 못했다. SerpApi naver_ai_overview(제3자 SERP API)로 조회하는 엔진을 붙였다(haewon 결정 2026-09-15). search.naver.com robots.txt 가 수집을 금지하므로 고객 리포트·랜딩에는 넣지 않고 내부 측정까지만 반영했다. 고객 노출은 법무 확인 뒤에 따로 한다. - run_question_bank_openai.py: --engine naver_briefing, --device, --no-cache(반복 측정 시 1시간 캐시 무시). 브리핑이 없는 문항은 answer 를 비우고 briefing=false 로 기록. 인용 출처(sources) 보관 - monthly_ai_measure.py: 기본 엔진 3개(키 없으면 네이버 건너뜀), supporter_builds.report.naverBriefing(노출률·언급·출처 상위, legalReviewed=false) - sentiment_qb.py: 빈 답변은 판정 API를 부르지 않고 미언급 처리 - merge_qb_results.py: NAVER 열·브리핑 없음 표시, 측정일을 생성일이 아닌 엔진별 실제 날짜로 - export_qb_excel.py: 네이버 열·NAVER상세 시트, --out 으로 기존 파일 보존. TOP50 점수는 기존 정의(ChatGPT+Perplexity, 만점 840) 유지 실측(뷰성형외과, 모바일, 2026-09-15): 브리핑 노출 119/120, 브랜드 질문 언급 36/36, 논브랜드 1/84. 인용 출처는 네이버 블로그 87·카페 51·굿닥 23·뷰 공식 20 순. SerpApi 123회, 감성 판정 $0.37. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
356 lines
15 KiB
Python
356 lines
15 KiB
Python
#!/usr/bin/env python3
|
|
"""뷰성형외과 Question Bank 120문항 답변엔진 실측 러너 (OpenAI / Perplexity / 네이버 AI 브리핑).
|
|
|
|
측정 방식:
|
|
- openai: Responses API + web_search 도구 (서울 위치 고정)
|
|
- perplexity: chat/completions, model=sonar (자체 웹검색 내장)
|
|
- naver_briefing: SerpApi naver_ai_overview (제3자 SERP API). 네이버는 공식 API가 없다.
|
|
ChatGPT 소비자 UI 자동 조회는 약관 위반이므로 쓰지 않는다.
|
|
|
|
naver_briefing 은 haewon 결정(2026-09-15)으로 넣었다. search.naver.com robots.txt 가 모든 봇과
|
|
AI·RAG 목적 수집을 금지하고 있어, 결과를 고객 리포트에 쓰기 전에 법무 확인이 필요하다.
|
|
브리핑이 뜨지 않는 질문은 answer 를 비우고 briefing=false 로 기록한다(그것도 결과다).
|
|
SerpApi 는 같은 파라미터를 1시간 캐시한다. 반복 측정은 --no-cache 로 새로 조회한다(검색 1회씩 과금).
|
|
|
|
각 질문을 원문 그대로 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
|
|
SERPAPI_API_KEY=... python3 scripts/run_question_bank_openai.py --engine naver_briefing --device mobile --limit 5
|
|
"""
|
|
import argparse
|
|
import json
|
|
import os
|
|
import re
|
|
import ssl
|
|
import sys
|
|
import time
|
|
import urllib.error
|
|
import urllib.parse
|
|
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 get_retry(url, tries=4, timeout=120):
|
|
for i in range(tries):
|
|
try:
|
|
with urllib.request.urlopen(urllib.request.Request(url), timeout=timeout, context=SSL_CTX) as r:
|
|
return json.loads(r.read())
|
|
except urllib.error.HTTPError as e:
|
|
body = e.read().decode(errors="replace")
|
|
if e.code in (429, 500, 502, 503) and i < tries - 1:
|
|
time.sleep(15 * (i + 1))
|
|
continue
|
|
# 결과 없음도 4xx 본문으로 올 수 있어 JSON 이면 그대로 돌려준다
|
|
try:
|
|
return json.loads(body)
|
|
except json.JSONDecodeError:
|
|
raise RuntimeError(f"HTTP {e.code}: {body[:500]}") from e
|
|
except (urllib.error.URLError, TimeoutError):
|
|
if i < tries - 1:
|
|
time.sleep(10)
|
|
continue
|
|
raise
|
|
|
|
|
|
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
|
|
|
|
|
|
class NaverBriefingEngine:
|
|
"""SerpApi naver_ai_overview. 반환 (answer, urls). 브리핑이 없으면 ("", []) 와 self.last_meta.briefing=False."""
|
|
name = "naver_briefing"
|
|
|
|
def __init__(self, device, no_cache):
|
|
self.key = os.environ.get("SERPAPI_API_KEY")
|
|
if not self.key:
|
|
sys.exit("SERPAPI_API_KEY가 필요합니다 (.env). serpapi.com 가입 후 대시보드의 API Key")
|
|
self.device = device
|
|
self.no_cache = no_cache
|
|
self.model = f"serpapi:naver_ai_overview:{device}"
|
|
self.last_meta = {}
|
|
|
|
def ask(self, question):
|
|
params = {"engine": "naver_ai_overview", "query": question, "device": self.device, "api_key": self.key}
|
|
if self.no_cache:
|
|
params["no_cache"] = "true"
|
|
resp = get_retry("https://serpapi.com/search.json?" + urllib.parse.urlencode(params))
|
|
ov = resp.get("ai_overview") if isinstance(resp.get("ai_overview"), dict) else resp
|
|
text = (ov.get("markdown") or "").strip()
|
|
if not text:
|
|
text = "\n".join(b.get("snippet", "") for b in ov.get("text_blocks") or [] if isinstance(b, dict)).strip()
|
|
refs = [r for r in ov.get("references") or [] if isinstance(r, dict)]
|
|
urls = [r["link"] for r in refs if r.get("link")]
|
|
self.last_meta = {
|
|
"briefing": bool(text),
|
|
"device": self.device,
|
|
"sources": [{"title": r.get("title"), "source": r.get("source"), "link": r.get("link")} for r in refs][:10],
|
|
"serpapi_id": (resp.get("search_metadata") or {}).get("id"),
|
|
}
|
|
if not text and resp.get("error"):
|
|
self.last_meta["no_briefing_reason"] = str(resp["error"])[:200]
|
|
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)",
|
|
"naver_briefing": "네이버 AI 브리핑 (SerpApi naver_ai_overview, 제3자 SERP API)"}[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)}%)",
|
|
*([f"- AI 브리핑 노출: {sum(1 for r in done.values() if r.get('briefing'))}/{len(done)}문항 "
|
|
"(브리핑이 없으면 언급 없음으로 센다)"] if engine == "naver_briefing" else []),
|
|
"", "### 병원별 언급 빈도 (상위 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", "viewclinic", "03_question_bank", 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", "naver_briefing"], default="openai")
|
|
ap.add_argument("--device", choices=["mobile", "desktop"], default="mobile", help="naver_briefing 전용")
|
|
ap.add_argument("--no-cache", action="store_true", help="naver_briefing 반복 측정 시 SerpApi 캐시 무시")
|
|
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)
|
|
elif args.engine == "naver_briefing":
|
|
eng = NaverBriefingEngine(args.device, args.no_cache)
|
|
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,
|
|
**getattr(eng, "last_meta", {}),
|
|
"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()
|