#!/usr/bin/env python3 """ enrich_ads.py — расход рекламы и nmID-разбивка за target_date vs среднее за 7 дней. Источники WB Advert API: - /adv/v1/promotion/count — список кампаний и их состояния (активные нас интересуют) - /adv/v2/fullstats — детальная статистика расхода по nmID за период Помечает: - RK_off: расход вчера = 0, бейзлайн > 100 ₽/день - RK_dropped: расход < 10% от бейзлайна - RK_boosted: расход > 200% от бейзлайна - RK_ok: всё в норме - NO_DATA: нет скоупа или эндпоинт упал """ from __future__ import annotations import argparse import json import os import sys import time from datetime import date, timedelta from typing import Any import requests URL_COUNT = "https://advert-api.wildberries.ru/adv/v1/promotion/count" URL_FULLSTATS = "https://advert-api.wildberries.ru/adv/v3/fullstats" # GET, не POST! MAX_RETRIES = 3 PER_REQUEST_SLEEP_SECONDS = 65 # WB Advert global limit — ~1 req/min def log(msg: str) -> None: print(f"[ads] {msg}", file=sys.stderr) def get_jwt() -> str: jwt = os.environ.get("WB_ANALYTICS_JWT") if not jwt: log("ERROR: WB_ANALYTICS_JWT not set") sys.exit(3) return jwt def fetch_campaigns(jwt: str) -> list[int] | None: """Тянем только активные/паузные кампании, изменённые за последние 14 дней. Без фильтра у крупных селлеров бывает 2000+ исторических — fullstats не уложится.""" headers = {"Authorization": jwt} from datetime import datetime, timedelta, timezone cutoff = datetime.now(timezone.utc) - timedelta(days=14) # Статусы WB: 7=завершена, 8=выключено, 9=активна, 11=пауза, 4=отказ модератора # Берём только 9 (активна) и 11 (пауза) — реклама могла крутиться вчера ACTIVE_STATUSES = {9, 11} for attempt in range(MAX_RETRIES): try: r = requests.get(URL_COUNT, headers=headers, timeout=30) except requests.RequestException as exc: log(f" conn: {exc}") time.sleep(5); continue if r.status_code in (401, 403): return None if r.status_code == 429: time.sleep(60); continue if r.status_code != 200: log(f" count HTTP {r.status_code}") time.sleep(5); continue data = r.json() ids: list[int] = [] for group in data.get("adverts", []): if group.get("status") not in ACTIVE_STATUSES: continue for adv in group.get("advert_list", []): ad_id = adv.get("advertId") if isinstance(adv, dict) else None ct = adv.get("changeTime") if isinstance(adv, dict) else None if not ad_id: continue # Парсим changeTime if ct: try: change_dt = datetime.fromisoformat(ct.replace("Z", "+00:00")) if change_dt.tzinfo is None: change_dt = change_dt.replace(tzinfo=timezone.utc) if change_dt < cutoff: continue except ValueError: pass ids.append(ad_id) log(f" campaigns active+recent: {len(ids)}") return ids return None def fetch_fullstats(jwt: str, advert_ids: list[int], date_from: str, date_to: str, max_campaigns: int = 0) -> list[dict[str, Any]] | None: """fullstats v3 — это GET с параметрами id (один за раз), begin, end. Глобальный лимит WB advert ≈ 1 req/min. Поэтому если ad_ids > ~5, проход займёт много минут — обрезаем по max_campaigns. max_campaigns=0 означает не запрашивать вовсе (по умолчанию для утреннего отчёта).""" if max_campaigns <= 0: log(" fullstats per-nm disabled (set --max-campaigns to enable)") return None headers = {"Authorization": jwt} rows: list[dict[str, Any]] = [] for i, ad_id in enumerate(advert_ids[:max_campaigns]): for attempt in range(MAX_RETRIES): try: r = requests.get(URL_FULLSTATS, headers=headers, params={"id": ad_id, "begin": date_from, "end": date_to}, timeout=30) except requests.RequestException as exc: log(f" conn: {exc}") time.sleep(5); continue if r.status_code in (401, 403): return None if r.status_code == 429: log(f" 429 on advert {ad_id}; sleeping {PER_REQUEST_SLEEP_SECONDS}s") time.sleep(PER_REQUEST_SLEEP_SECONDS); continue if r.status_code == 400: log(f" 400 on advert {ad_id}: {r.text[:200]} — skip") break if r.status_code != 200: log(f" HTTP {r.status_code}: {r.text[:200]}") time.sleep(5 * (attempt + 1)); continue data = r.json() if isinstance(data, dict): rows.append(data) elif isinstance(data, list): rows.extend(data) break if i + 1 < min(max_campaigns, len(advert_ids)): time.sleep(PER_REQUEST_SLEEP_SECONDS) return rows def aggregate_spend_by_nm(rows: list[dict[str, Any]], target_date: str, baseline_dates: list[str]) -> dict[int, dict[str, float]]: """Из fullstats берём дневные nm-расходы (поле .days[].nm[]). Считаем target_spend и avg_baseline_spend.""" by_nm_today: dict[int, float] = {} by_nm_base_sum: dict[int, float] = {} base_set = set(baseline_dates) base_n = len(baseline_dates) for camp in rows or []: for day in camp.get("days", []): d_iso = (day.get("date") or "")[:10] for app in day.get("apps", []): for nm in app.get("nm", []): nm_id = nm.get("nmId") spend = nm.get("sum", 0) or 0 if not nm_id: continue if d_iso == target_date: by_nm_today[nm_id] = by_nm_today.get(nm_id, 0) + spend if d_iso in base_set: by_nm_base_sum[nm_id] = by_nm_base_sum.get(nm_id, 0) + spend avg = {nm: total / base_n for nm, total in by_nm_base_sum.items()} if base_n else {} out: dict[int, dict[str, float]] = {} keys = set(by_nm_today) | set(avg) for nm in keys: out[nm] = { "spent_today": by_nm_today.get(nm, 0), "spent_avg": avg.get(nm, 0), } return out def make_flag(s: dict[str, float]) -> str: today = s.get("spent_today", 0) avg = s.get("spent_avg", 0) if avg < 1: return "no_baseline" if today == 0 else "started" if today == 0 and avg > 100: return "RK_off" ratio = today / avg if avg else 0 if ratio < 0.1: return "RK_dropped" if ratio > 2.0: return "RK_boosted" return "RK_ok" def main() -> int: p = argparse.ArgumentParser() p.add_argument("--insights", required=True) p.add_argument("--max-campaigns", type=int, default=0, help="Сколько кампаний обходить через fullstats. 0 = пропустить per-nm разбивку " "(быстрый режим для утреннего отчёта). >0 — медленно, ~1 минута на кампанию.") args = p.parse_args() with open(args.insights, encoding="utf-8") as f: insights = json.load(f) target = insights["target_date"] baseline = insights["baseline_dates"] jwt = get_jwt() log("Fetching campaigns list…") campaigns = fetch_campaigns(jwt) if not campaigns: log("Skipped: no campaigns / no scope") for section in ("drops", "growths", "oos_candidates"): for s in insights.get(section, []): s.setdefault("enrichment", {})["ads"] = {"flag": "NO_DATA"} with open(args.insights, "w", encoding="utf-8") as f: json.dump(insights, f, ensure_ascii=False, indent=2) return 0 log(f" campaigns: {len(campaigns)}") rows = fetch_fullstats(jwt, campaigns, baseline[-1], target, max_campaigns=args.max_campaigns) if rows is None: log(" per-nm spend disabled — writing summary flag only") for section in ("drops", "growths", "oos_candidates"): for s in insights.get(section, []): s.setdefault("enrichment", {})["ads"] = { "flag": "SUMMARY_ONLY", "active_campaigns": len(campaigns), "note": "Детальный per-nm расход недоступен в быстром режиме (rate-limit WB ~1 req/min)", } with open(args.insights, "w", encoding="utf-8") as f: json.dump(insights, f, ensure_ascii=False, indent=2) return 0 log(f" fullstats rows: {len(rows)}") by_nm = aggregate_spend_by_nm(rows, target, baseline) for section in ("drops", "growths", "oos_candidates"): for s in insights.get(section, []): spend = by_nm.get(s["nmID"], {"spent_today": 0, "spent_avg": 0}) spend["flag"] = make_flag(spend) s.setdefault("enrichment", {})["ads"] = spend with open(args.insights, "w", encoding="utf-8") as f: json.dump(insights, f, ensure_ascii=False, indent=2) return 0 if __name__ == "__main__": sys.exit(main())