"""AIO Tracker: AI 검색엔진별 유입 트래픽 Before/After 변화 자동 계산.""" from __future__ import annotations import argparse import csv from datetime import date, timedelta from pathlib import Path from typing import Dict, List, Optional from config import classify_ai_source, is_ga4_configured # type: ignore[import-not-found] # GA4 API 옵셔널 임포트 try: from google.analytics.data_v1beta import BetaAnalyticsDataClient # type: ignore[import-not-found] from google.analytics.data_v1beta.types import ( # type: ignore[import-not-found] DateRange, Dimension, Metric, RunReportRequest, ) _GA4_AVAILABLE = True except ImportError: _GA4_AVAILABLE = False def parse_csv(path: Path) -> Dict[str, int]: """CSV를 파싱해 {referrer: sessions} 반환. 컬럼: 'referrer' 또는 'source'. Raises: FileNotFoundError: 파일 미존재 ValueError: 필수 컬럼 없음 """ path = Path(path) if not path.exists(): raise FileNotFoundError(f"CSV 파일을 찾을 수 없습니다: {path}") result: Dict[str, int] = {} with path.open(encoding="utf-8", newline="") as f: reader = csv.DictReader(f) fieldnames = reader.fieldnames or [] ref_col = next((c for c in fieldnames if c.strip().lower() in ("referrer", "source")), None) if ref_col is None: raise ValueError(f"CSV에 'referrer' 또는 'source' 컬럼이 없습니다: {path}") for row in reader: key = row[ref_col].strip() sessions = int(row["sessions"].strip()) if key: result[key] = result.get(key, 0) + sessions return result def calculate_change_rate(before: int, after: int) -> Optional[float]: """변화율 계산: (after - before) / before * 100. before=0이면 None(NEW).""" if before == 0: return None return (after - before) / before * 100 def generate_markdown_report(summary: List[Dict], date_label: str = "") -> str: """AI 소스 요약을 마크다운 리포트 문자열로 변환한다.""" title_date = f" ({date_label})" if date_label else "" lines = [ f"# AIO 성과 리포트{title_date}", "", "## AI 유입 트래픽 변화", "| 소스 | Before | After | 변화율 |", "|------|--------|-------|--------|", ] for row in summary: rate = row["change_rate"] if rate is None: rate_str = "NEW" elif rate >= 0: rate_str = f"+{rate:.1f}%" else: rate_str = f"{rate:.1f}%" lines.append(f"| {row['source']} | {row['before']} | {row['after']} | {rate_str} |") return "\n".join(lines) class AioTracker: """AI 검색엔진 유입 트래픽 Before/After 분석기.""" def __init__( self, before_csv: Optional[Path] = None, after_csv: Optional[Path] = None, property_id: str = "", before_range: Optional[tuple] = None, after_range: Optional[tuple] = None, ): self.property_id = property_id if before_csv and after_csv: self.before_data = parse_csv(before_csv) self.after_data = parse_csv(after_csv) elif property_id and _GA4_AVAILABLE and is_ga4_configured(): b_start, b_end = before_range or _default_before_range() a_start, a_end = after_range or _default_after_range() self.before_data = _fetch_ga4_referrers(property_id, b_start, b_end) self.after_data = _fetch_ga4_referrers(property_id, a_start, a_end) else: self.before_data = {} self.after_data = {} def compute_ai_summary(self) -> List[Dict]: """Before/After를 AI 소스별로 집계. 기타(비-AI) 소스 제외.""" ai_before: Dict[str, int] = {} ai_after: Dict[str, int] = {} for ref, s in self.before_data.items(): src = classify_ai_source(ref) if src != "기타": ai_before[src] = ai_before.get(src, 0) + s for ref, s in self.after_data.items(): src = classify_ai_source(ref) if src != "기타": ai_after[src] = ai_after.get(src, 0) + s result = [] for source in sorted(set(ai_before) | set(ai_after)): before = ai_before.get(source, 0) after = ai_after.get(source, 0) result.append( { "source": source, "before": before, "after": after, "change_rate": calculate_change_rate(before, after), } ) return result def _default_before_range() -> tuple: today = date.today() end = today - timedelta(days=today.weekday() + 1) return (end - timedelta(days=6)).isoformat(), end.isoformat() def _default_after_range() -> tuple: today = date.today() end = today - timedelta(days=today.weekday() + 1) start = end + timedelta(days=1) return start.isoformat(), (start + timedelta(days=6)).isoformat() def _fetch_ga4_referrers(property_id: str, start_date: str, end_date: str) -> Dict[str, int]: """GA4 Data API로 소스별 세션 수 조회.""" if not _GA4_AVAILABLE: return {} client = BetaAnalyticsDataClient() # type: ignore[possibly-undefined] request = RunReportRequest( # type: ignore[possibly-undefined] property=f"properties/{property_id}", dimensions=[Dimension(name="sessionSource")], # type: ignore[possibly-undefined] metrics=[Metric(name="sessions")], # type: ignore[possibly-undefined] date_ranges=[DateRange(start_date=start_date, end_date=end_date)], # type: ignore[possibly-undefined] ) response = client.run_report(request) result: Dict[str, int] = {} for row in response.rows: src = row.dimension_values[0].value result[src] = result.get(src, 0) + int(row.metric_values[0].value) return result def _parse_args(argv=None): parser = argparse.ArgumentParser(description="AI 검색엔진 유입 트래픽 Before/After 분석") parser.add_argument("--property-id", default="") parser.add_argument("--before", default="", help="YYYY-MM-DD:YYYY-MM-DD") parser.add_argument("--after", default="", help="YYYY-MM-DD:YYYY-MM-DD") parser.add_argument("--before-csv", default="") parser.add_argument("--after-csv", default="") parser.add_argument("--output", default="") return parser.parse_args(argv) def main(argv=None): args = _parse_args(argv) before_csv = Path(args.before_csv) if args.before_csv else None after_csv = Path(args.after_csv) if args.after_csv else None before_range = tuple(args.before.split(":")) if args.before else None after_range = tuple(args.after.split(":")) if args.after else None tracker = AioTracker( before_csv=before_csv, after_csv=after_csv, property_id=args.property_id, before_range=before_range, after_range=after_range, ) summary = tracker.compute_ai_summary() report = generate_markdown_report(summary, date_label=date.today().isoformat()) if args.output: out = Path(args.output) out.parent.mkdir(parents=True, exist_ok=True) out.write_text(report, encoding="utf-8") print(f"리포트 저장 완료: {out}") else: print(report) if __name__ == "__main__": main()