#!/usr/bin/env python3 import argparse import json import os import sys from datetime import datetime, timedelta, timezone if "/home/jay/workspace" not in sys.path: sys.path.insert(0, "/home/jay/workspace") from sa_api_client import NaverSAClient # type: ignore[import-not-found] from utils.logger import get_logger # type: ignore[import-not-found] logger = get_logger(__name__) KST = timezone(timedelta(hours=9)) _OUTPUT_PATH = "/home/jay/workspace/dashboard/data/naver-sa-stats.json" _DAILY_BUDGET = 21000 _GROUP_RULES = [ ("F", "보험영업", lambda name: "보험영업" in name), ("G", "개인사업/지점장", lambda name: "개인사업" in name or "지점장" in name), ("H", "무자본창업", lambda name: "무자본" in name or "창업" in name), ] def _empty_summary() -> dict: return {"impressions": 0, "clicks": 0, "cost": 0, "ctr": 0.0, "cpc": 0, "conversions": 0} def _empty_output(since: str, until: str) -> dict: now = datetime.now(KST).isoformat(timespec="seconds") days = (datetime.fromisoformat(until) - datetime.fromisoformat(since)).days + 1 daily_trend = [] for i in range(days): d = (datetime.fromisoformat(since) + timedelta(days=i)).strftime("%Y-%m-%d") daily_trend.append({"date": d, "impressions": 0, "clicks": 0, "cost": 0, "conversions": 0}) return { "last_collected": now, "summary": { "today": _empty_summary(), "yesterday": _empty_summary(), }, "keyword_groups": [ {"id": gid, "name": gname, "impressions": 0, "clicks": 0, "cost": 0, "ctr": 0.0, "cpc": 0} for gid, gname, _ in _GROUP_RULES ], "daily_trend": daily_trend, "budget": {"daily": _DAILY_BUDGET, "spent_today": 0, "burn_rate": 0.0}, "campaigns": [], } def collect(date: str, days: int) -> None: until_dt = datetime.fromisoformat(date) since_dt = until_dt - timedelta(days=days - 1) since = since_dt.strftime("%Y-%m-%d") until = until_dt.strftime("%Y-%m-%d") today_str = until yesterday_str = (until_dt - timedelta(days=1)).strftime("%Y-%m-%d") try: client = NaverSAClient() except Exception as e: logger.error("NaverSAClient 초기화 실패: %s", e) _save(_empty_output(since, until)) return # 1. 캠페인 목록 try: campaigns = client.get_campaigns() except Exception as e: logger.error("캠페인 조회 실패: %s", e) _save(_empty_output(since, until)) return campaign_ids = [c.get("nccCampaignId") or c.get("id") for c in campaigns if c] campaign_ids = [cid for cid in campaign_ids if cid] # 2. 광고그룹 + 키워드 수집 (그룹 매핑용) adgroup_group_map: dict[str, str] = {} for cid in campaign_ids: try: adgroups = client.get_adgroups(campaign_id=cid) except Exception as e: logger.warning("광고그룹 조회 실패 (campaign=%s): %s", cid, e) continue for ag in adgroups: ag_id = ag.get("nccAdgroupId") or ag.get("id", "") ag_name = ag.get("name", "") for gid, _, matcher in _GROUP_RULES: if matcher(ag_name): adgroup_group_map[ag_id] = gid break # 3. 캠페인 stats (전체 기간) daily_map: dict[str, dict] = {} group_agg: dict[str, dict] = {gid: {"imp": 0, "clk": 0, "cost": 0} for gid, _, _ in _GROUP_RULES} today_stats = _empty_summary() yesterday_stats = _empty_summary() campaign_results = [] if campaign_ids: try: stats = client.get_stats( ids=campaign_ids, time_range={"since": since, "until": until}, ) except Exception as e: logger.error("stats 조회 실패: %s", e) stats = [] for record in stats: stat_date = record.get("date", "") imp = record.get("impCnt", 0) clk = record.get("clkCnt", 0) cost = record.get("salesAmt", 0) conv = record.get("ccnt", 0) # daily_trend 집계 if stat_date not in daily_map: daily_map[stat_date] = {"date": stat_date, "impressions": 0, "clicks": 0, "cost": 0, "conversions": 0} daily_map[stat_date]["impressions"] += imp daily_map[stat_date]["clicks"] += clk daily_map[stat_date]["cost"] += cost daily_map[stat_date]["conversions"] += conv if stat_date == today_str: today_stats["impressions"] += imp today_stats["clicks"] += clk today_stats["cost"] += cost today_stats["conversions"] += conv if stat_date == yesterday_str: yesterday_stats["impressions"] += imp yesterday_stats["clicks"] += clk yesterday_stats["cost"] += cost yesterday_stats["conversions"] += conv # today ctr/cpc if today_stats["impressions"]: today_stats["ctr"] = round(today_stats["clicks"] / today_stats["impressions"] * 100, 2) if today_stats["clicks"]: today_stats["cpc"] = round(today_stats["cost"] / today_stats["clicks"]) if yesterday_stats["impressions"]: yesterday_stats["ctr"] = round(yesterday_stats["clicks"] / yesterday_stats["impressions"] * 100, 2) if yesterday_stats["clicks"]: yesterday_stats["cpc"] = round(yesterday_stats["cost"] / yesterday_stats["clicks"]) # 캠페인별 결과 for c in campaigns: cid = c.get("nccCampaignId") or c.get("id", "") campaign_results.append( { "id": cid, "name": c.get("name", ""), "status": c.get("status", ""), } ) # daily_trend 정렬 (누락 날짜 채우기) daily_trend = [] for i in range(days): d = (since_dt + timedelta(days=i)).strftime("%Y-%m-%d") daily_trend.append(daily_map.get(d, {"date": d, "impressions": 0, "clicks": 0, "cost": 0, "conversions": 0})) # keyword_groups keyword_groups = [] for gid, gname, _ in _GROUP_RULES: agg = group_agg.get(gid, {"imp": 0, "clk": 0, "cost": 0}) imp = agg["imp"] clk = agg["clk"] cost = agg["cost"] keyword_groups.append( { "id": gid, "name": gname, "impressions": imp, "clicks": clk, "cost": cost, "ctr": round(clk / imp * 100, 2) if imp else 0.0, "cpc": round(cost / clk) if clk else 0, } ) spent_today = today_stats["cost"] burn_rate = round(spent_today / _DAILY_BUDGET, 4) if _DAILY_BUDGET else 0.0 output = { "last_collected": datetime.now(KST).isoformat(timespec="seconds"), "summary": { "today": today_stats, "yesterday": yesterday_stats, }, "keyword_groups": keyword_groups, "daily_trend": daily_trend, "budget": { "daily": _DAILY_BUDGET, "spent_today": spent_today, "burn_rate": burn_rate, }, "campaigns": campaign_results, } _save(output) def _save(data: dict) -> None: os.makedirs(os.path.dirname(_OUTPUT_PATH), exist_ok=True) tmp = _OUTPUT_PATH + ".tmp" with open(tmp, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) os.replace(tmp, _OUTPUT_PATH) logger.info("naver-sa-stats.json 저장 완료: %s", _OUTPUT_PATH) def main() -> None: parser = argparse.ArgumentParser(description="네이버 검색광고 통계 수집") parser.add_argument("--date", default=datetime.now(KST).strftime("%Y-%m-%d"), help="기준 날짜 (YYYY-MM-DD)") parser.add_argument("--days", type=int, default=7, help="수집 기간 (일수)") args = parser.parse_args() collect(args.date, args.days) if __name__ == "__main__": main()