o2o-infinith-demo/scripts/ga4_ai_traffic.py

#!/usr/bin/env python3
"""GA4 → AI 검색 유입 리포트. GA4 Data API(runReport)로 세션 출처·랜딩 페이지·키 이벤트를 받아 data/ai_channels.json 기준으로 AI 채널을 분류한다.

연동 방식(B2B): 병원(또는 서포터즈 사이트)의 GA4 속성에 우리 서비스 계정 이메일을 '뷰어'로 추가하면 끝. 코드 설치 없음.
  1) Google Cloud 프로젝트에서 Google Analytics Data API 켜기 → 서비스 계정 키(JSON) 발급
  2) GA4 관리 > 속성 액세스 관리 > 서비스 계정 이메일을 뷰어로 추가
  3) .env: GA4_SERVICE_ACCOUNT_JSON=<키 파일 경로>  (속성 ID는 --property 인자)

사용:
  python3 scripts/ga4_ai_traffic.py --property 123456789 --days 90 --clinic wonjin --out docs/reports/wonjin/05_ai_traffic
  python3 scripts/ga4_ai_traffic.py --from-csv <GA4 탐색 내보내기.csv> --clinic wonjin --out ...   # API 없이 (열: date, sessionSource, sessionMedium, landingPage, sessions, keyEvents)
  python3 scripts/ga4_ai_traffic.py --mock --clinic demo --out /tmp/ga4demo                        # 합성 데이터로 파이프라인 검증

산출: <out>/ai_traffic.json (일별·채널별·랜딩별 집계) + <out>/AI_Traffic_Report.md ("측정 가능 범위" 표 포함)
원칙: GA4 로 잴 수 없는 것(AI Overview·네이버 AI 브리핑·리퍼러 없는 유입·노출만 된 경우)은 숫자 없이 '측정 불가'로 적는다. 추정치를 만들지 않는다.
"""
import argparse, csv, datetime as dt, json, os, random, sys, collections

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
CHANNELS = json.load(open(os.path.join(ROOT, "data", "ai_channels.json"), encoding="utf-8"))

def load_env():
    p = os.path.join(ROOT, ".env")
    if not os.path.exists(p): return
    for line in open(p, encoding="utf-8"):
        if "=" in line and not line.strip().startswith("#"):
            k, v = line.split("=", 1); os.environ.setdefault(k.strip(), v.strip().strip('"').strip("'"))

# ---------- 분류 ----------
def classify(source: str, medium: str = "", utm_source: str = "") -> str | None:
    """세션 출처를 AI 채널 id 로. 아니면 None. 부분 문자열이 아니라 호스트 경계로 본다."""
    s = (source or "").lower().strip(); u = (utm_source or source or "").lower().strip()
    for ch in CHANNELS["channels"]:
        for host in ch["sources"]:
            h = host.lower()
            if s == h or s.endswith("." + h) or s.startswith(h + "/") or s == h.replace("www.", ""): return ch["id"]
        for us in ch["utm_sources"]:
            if u == us.lower(): return ch["id"]
    return None

LABEL = {c["id"]: c["label"] for c in CHANNELS["channels"]}
MEASURABLE = {c["id"]: c["measurable"] for c in CHANNELS["channels"]}

# ---------- 데이터 소스 ----------
def rows_from_api(property_id: str, days: int):
    try:
        from google.oauth2 import service_account
        from google.auth.transport.requests import AuthorizedSession
    except ImportError:
        sys.exit("google-auth 가 필요합니다: pip install google-auth requests")
    load_env()
    key_path = os.environ.get("GA4_SERVICE_ACCOUNT_JSON")
    if not key_path or not os.path.exists(key_path): sys.exit("GA4_SERVICE_ACCOUNT_JSON(서비스 계정 키 파일 경로) 필요 (.env)")
    creds = service_account.Credentials.from_service_account_file(key_path, scopes=["https://www.googleapis.com/auth/analytics.readonly"])
    sess = AuthorizedSession(creds)
    body = {"dateRanges": [{"startDate": f"{days}daysAgo", "endDate": "yesterday"}],
            "dimensions": [{"name": "date"}, {"name": "sessionSource"}, {"name": "sessionMedium"}, {"name": "landingPagePlusQueryString"}],
            "metrics": [{"name": "sessions"}, {"name": "keyEvents"}, {"name": "engagedSessions"}],
            "limit": 100000}
    r = sess.post(f"https://analyticsdata.googleapis.com/v1beta/properties/{property_id}:runReport", json=body, timeout=60)
    if r.status_code != 200: sys.exit(f"GA4 Data API {r.status_code}: {r.text[:300]}")
    j = r.json()
    for row in j.get("rows", []):
        d = [x["value"] for x in row["dimensionValues"]]; m = [x["value"] for x in row["metricValues"]]
        yield {"date": f"{d[0][:4]}-{d[0][4:6]}-{d[0][6:]}", "source": d[1], "medium": d[2], "landing": d[3].split("?")[0], "utm_source": _utm(d[3]),
               "sessions": int(float(m[0])), "key_events": int(float(m[1])), "engaged": int(float(m[2]))}

def _utm(landing_with_qs: str) -> str:
    if "utm_source=" not in landing_with_qs: return ""
    return landing_with_qs.split("utm_source=")[1].split("&")[0]

def rows_from_csv(path: str):
    with open(path, encoding="utf-8-sig") as f:
        for r in csv.DictReader(f):
            g = lambda *ks: next((r[k] for k in ks if k in r and r[k] not in (None, "")), "")
            yield {"date": g("date", "날짜"), "source": g("sessionSource", "세션 소스", "source"), "medium": g("sessionMedium", "세션 매체", "medium"),
                   "landing": g("landingPage", "landingPagePlusQueryString", "방문 페이지", "landing").split("?")[0], "utm_source": _utm(g("landingPagePlusQueryString", "landing")),
                   "sessions": int(float(g("sessions", "세션수", "세션") or 0)), "key_events": int(float(g("keyEvents", "주요 이벤트", "conversions") or 0)), "engaged": int(float(g("engagedSessions", "참여 세션수") or 0))}

def rows_mock(days: int = 60):
    random.seed(7); today = dt.date.today()
    srcs = [("google", "organic", 40), ("(direct)", "(none)", 18), ("chatgpt.com", "referral", 6), ("perplexity.ai", "referral", 2), ("gemini.google.com", "referral", 1), ("search.naver.com", "organic", 25), ("instagram.com", "referral", 4)]
    pages = ["/", "/posts/consult-questions", "/posts/procedure-eye", "/visit", "/videos", "/newsroom"]
    for i in range(days):
        d = (today - dt.timedelta(days=days - i)).isoformat()
        for s, m, w in srcs:
            for p in random.sample(pages, 3):
                n = max(0, int(random.gauss(w / 3, w / 6)))
                if n: yield {"date": d, "source": s, "medium": m, "landing": p, "utm_source": "chatgpt.com" if s == "chatgpt.com" else "", "sessions": n, "key_events": sum(1 for _ in range(n) if random.random() < 0.06), "engaged": int(n * 0.6)}

# ---------- 집계 ----------
def aggregate(rows):
    daily = collections.defaultdict(lambda: collections.Counter()); by_ch = collections.Counter(); ke_ch = collections.Counter()
    landing = collections.defaultdict(lambda: collections.Counter()); total = collections.Counter(); direct = 0; weeks = collections.defaultdict(lambda: collections.Counter())
    for r in rows:
        total["sessions"] += r["sessions"]; total["key_events"] += r["key_events"]
        ch = classify(r["source"], r["medium"], r.get("utm_source", ""))
        if r["source"] in ("(direct)", "direct"): direct += r["sessions"]
        if not ch: continue
        daily[r["date"]][ch] += r["sessions"]; by_ch[ch] += r["sessions"]; ke_ch[ch] += r["key_events"]
        landing[r["landing"]][ch] += r["sessions"]; landing[r["landing"]]["_ke"] += r["key_events"]
        y, w, _ = dt.date.fromisoformat(r["date"]).isocalendar(); weeks[f"{y}-W{w:02d}"][ch] += r["sessions"]
    ai_total = sum(by_ch.values())
    return {"total_sessions": total["sessions"], "total_key_events": total["key_events"], "direct_sessions": direct,
            "ai_sessions": ai_total, "ai_share_pct": round(ai_total / total["sessions"] * 100, 2) if total["sessions"] else 0,
            "by_channel": [{"id": c, "label": LABEL[c], "measurable": MEASURABLE[c], "sessions": n, "key_events": ke_ch[c], "cvr_pct": round(ke_ch[c] / n * 100, 1) if n else 0} for c, n in by_ch.most_common()],
            "weekly": {w: dict(v) for w, v in sorted(weeks.items())}, "daily": {d: dict(v) for d, v in sorted(daily.items())},
            "landing": sorted([{"page": p, "ai_sessions": sum(n for k, n in v.items() if k != "_ke"), "key_events": v["_ke"], "by": {k: n for k, n in v.items() if k != "_ke"}} for p, v in landing.items()], key=lambda x: -x["ai_sessions"])[:20]}

# ---------- 리포트 ----------
def scope_table_md():
    lines = ["| 채널 | GA4로 측정 | 비고 |", "|---|---|---|"]
    M = {"yes": "가능", "partial": "일부만", "no": "불가"}
    for c in CHANNELS["channels"]: lines.append(f"| {c['label']} | **{M[c['measurable']]}** | {c['note']} |")
    lines += ["", "GA4로 잴 수 없는 것:", ""] + [f"- {x}" for x in CHANNELS["not_measurable_by_ga4"]]
    return "\n".join(lines)

def report_md(agg, clinic, source_desc, days):
    L = [f"# AI 검색 유입 리포트 · {clinic}", "", f"기간 최근 {days}일 · 데이터 {source_desc} · 분류 기준 `data/ai_channels.json` v{CHANNELS['version']} · 작성 {dt.date.today()}", "",
         "## 1. 요약", "", "| 지표 | 값 |", "|---|---|",
         f"| 전체 세션 | {agg['total_sessions']:,} |", f"| AI 검색 유입 세션(측정 가능분) | {agg['ai_sessions']:,} ({agg['ai_share_pct']}%) |",
         f"| AI 유입의 키 이벤트(전화 탭·예약·홈페이지 이동) | {sum(c['key_events'] for c in agg['by_channel']):,} |", f"| direct 세션(리퍼러 없음, AI 유입 일부가 여기 섞임) | {agg['direct_sessions']:,} |", "",
         "## 2. 채널별", "", "| 채널 | 세션 | 키 이벤트 | 전환율 | 측정 |", "|---|---|---|---|---|"]
    M = {"yes": "가능", "partial": "일부만", "no": "불가"}
    for c in agg["by_channel"]: L.append(f"| {c['label']} | {c['sessions']:,} | {c['key_events']} | {c['cvr_pct']}% | {M[c['measurable']]} |")
    for c in CHANNELS["channels"]:
        if c["measurable"] == "no": L.append(f"| {c['label']} | 측정 불가 | | | 불가 |")
    L += ["", "## 3. AI 유입 랜딩 페이지 (상위)", "", "| 페이지 | AI 세션 | 키 이벤트 | 채널 |", "|---|---|---|---|"]
    for p in agg["landing"][:12]: L.append(f"| {p['page']} | {p['ai_sessions']} | {p['key_events']} | {', '.join(f'{LABEL[k]} {n}' for k, n in sorted(p['by'].items(), key=lambda x: -x[1]))} |")
    L += ["", "## 4. 주별 추이", "", "| 주 | " + " | ".join(LABEL[c["id"]] for c in agg["by_channel"]) + " | 합 |", "|---|" + "---|" * (len(agg["by_channel"]) + 1)]
    for w, v in agg["weekly"].items(): L.append(f"| {w} | " + " | ".join(str(v.get(c["id"], 0)) for c in agg["by_channel"]) + f" | {sum(v.values())} |")
    L += ["", "## 5. 측정 가능 범위", "", "이 리포트의 숫자는 GA4가 리퍼러나 utm으로 구분할 수 있는 유입만 셉니다. 아래 표의 '일부만'과 '불가'는 실제보다 적게 잡히거나 전혀 잡히지 않는 채널입니다. AI 답변에 병원이 얼마나 나오는지(노출)는 이 리포트가 아니라 답변엔진 실측(질문 뱅크)으로 잽니다. 두 숫자를 같은 시간축에 놓고 읽어야 합니다.", "", scope_table_md(), "",
          "## 6. 연동 방식", "", "GA4 속성 액세스 관리에서 INFINITH 서비스 계정 이메일을 '뷰어'로 추가하면 매일 자동으로 갱신됩니다. 코드 설치는 없습니다. 병원 사이트에 GA4가 없으면 태그 설치가 먼저이고, 서포터즈 사이트는 INFINITH가 직접 계측합니다."]
    return "\n".join(L) + "\n"

if __name__ == "__main__":
    ap = argparse.ArgumentParser()
    ap.add_argument("--property"); ap.add_argument("--days", type=int, default=90); ap.add_argument("--from-csv"); ap.add_argument("--mock", action="store_true")
    ap.add_argument("--clinic", required=True); ap.add_argument("--out", required=True)
    a = ap.parse_args()
    if a.mock: rows, desc = list(rows_mock(a.days if a.days < 90 else 60)), "합성 데이터(파이프라인 검증용)"
    elif a.from_csv: rows, desc = list(rows_from_csv(a.from_csv)), f"GA4 내보내기 {os.path.basename(a.from_csv)}"
    elif a.property: rows, desc = list(rows_from_api(a.property, a.days)), f"GA4 Data API 속성 {a.property}"
    else: sys.exit("--property 또는 --from-csv 또는 --mock 중 하나")
    agg = aggregate(rows)
    os.makedirs(a.out, exist_ok=True)
    json.dump({"clinic": a.clinic, "source": desc, "days": a.days, "generatedAt": dt.date.today().isoformat(), "channels_version": CHANNELS["version"], **agg}, open(os.path.join(a.out, "ai_traffic.json"), "w", encoding="utf-8"), ensure_ascii=False, indent=1)
    open(os.path.join(a.out, "AI_Traffic_Report.md"), "w", encoding="utf-8").write(report_md(agg, a.clinic, desc, a.days))
    print(f"세션 {agg['total_sessions']:,} · AI 유입 {agg['ai_sessions']:,} ({agg['ai_share_pct']}%) · 채널 {', '.join(f'{c['label']} {c['sessions']}' for c in agg['by_channel'])} → {a.out}")