#!/usr/bin/env python3 """ fetch_funnel.py — собирает воронку WB по выборке SKU. ВАЖНО (проверено боевым прогоном 27.05.2026): - Эндпоинт: POST /api/analytics/v3/sales-funnel/products/history - Лимит периода: 7 дней (D-6 … D), то есть target + 6 предыдущих. Это даёт бейзлайн на 6 дней (не 7, как было в первоначальном ТЗ). - Поля запроса: nmIds (lower-case), selectedPeriod (не period). - Поля ответа в .history[]: openCount, cartCount, orderCount, orderSum, buyoutCount, buyoutSum, buyoutPercent, addToCartConversion, cartToOrderConversion, addToWishlistCount. Поля clickCount НЕТ — отдельного этапа «клик» в воронке нет. Скилл оперирует моделью «показы → корзина → заказ → выкуп». Output: { target_date, baseline_dates, by_nm: {nmID: {card, dates: {YYYY-MM-DD: {...}}}}, errors } Exit codes: 0 — успех 2 — нет данных за target_date (WB ещё не закрыл день) 3 — критическая ошибка (401, нет SKU и т.п.) """ from __future__ import annotations import argparse import json import os import sys import time from datetime import date, datetime, timedelta from pathlib import Path from zoneinfo import ZoneInfo from typing import Any import requests WB_CONTENT_URL = "https://content-api.wildberries.ru/content/v2/get/cards/list" WB_FUNNEL_URL = "https://seller-analytics-api.wildberries.ru/api/analytics/v3/sales-funnel/products/history" BATCH_SIZE = 20 # ПОДТВЕРЖДЕНО боевым запросом 27.05.2026: WB analytics возвращает # 400 "too many nmIds is expected - max 20" на больший батч. # Изменять этот размер нельзя — это hard-limit на стороне WB. MAX_RETRIES = 5 RATE_LIMIT_SLEEP_SECONDS = 20 # 3 req/min на seller-analytics PERIOD_MAX_DAYS = 7 # ВНУТРЕННИЙ ЛИМИТ WB: окно до 7 дней включительно (target + 6) MSK = ZoneInfo("Europe/Moscow") def log(msg: str) -> None: print(f"[{datetime.now(MSK):%H:%M:%S}] {msg}", file=sys.stderr) def get_jwt() -> str: jwt = os.environ.get("WB_ANALYTICS_JWT") or os.environ.get("WB_JWT") if not jwt: log("ERROR: WB_ANALYTICS_JWT not set") sys.exit(3) return jwt def fetch_cards_cached(jwt: str, cache_dir: Path | None, max_pages: int | None = None) -> list[dict[str, Any]]: """Обёртка с однодневным кэшем: за один день карточки кабинета меняются редко, нет смысла дёргать Content API на каждом запуске.""" if cache_dir is not None: cache_dir.mkdir(parents=True, exist_ok=True) today_iso = date.today().isoformat() cache_file = cache_dir / f"cards_{today_iso}.json" if cache_file.exists(): try: with open(cache_file, encoding="utf-8") as f: cards = json.load(f) log(f" Content cache hit: {len(cards)} cards (from {cache_file.name})") return cards except (OSError, ValueError) as e: log(f" Content cache read failed: {e}; refetching") cards = _fetch_cards_uncached(jwt, max_pages) if cache_dir is not None and cards: try: with open(cache_dir / f"cards_{date.today().isoformat()}.json", "w", encoding="utf-8") as f: json.dump(cards, f, ensure_ascii=False) except OSError as e: log(f" Content cache write failed: {e}") return cards def _fetch_cards_uncached(jwt: str, max_pages: int | None = None) -> list[dict[str, Any]]: """Тянет карточки кабинета. Content API склонен к 500-кам на пагинации — при сбое страницы делаем 5 ретраев с прогрессивной паузой, если всё равно не отдаёт — возвращаем уже накопленное (best-effort) и идём дальше.""" cards: list[dict[str, Any]] = [] cursor: dict[str, Any] = {"limit": 100} headers = {"Authorization": jwt, "Content-Type": "application/json"} page = 0 while True: if max_pages is not None and page >= max_pages: log(f" Content: reached max_pages={max_pages}, stopping pagination") return cards body = {"settings": {"cursor": cursor, "filter": {"withPhoto": -1}}} page_ok = False for attempt in range(MAX_RETRIES): try: r = requests.post(WB_CONTENT_URL, json=body, headers=headers, timeout=30) except requests.RequestException as exc: log(f" Content connection: {exc}; retry {attempt + 1}") time.sleep(3 * (attempt + 1)); continue if r.status_code == 401: log("ERROR: 401 Unauthorized on Content API") sys.exit(3) if r.status_code == 429: log(f" Content 429; sleep {3 * (attempt + 1)}s") time.sleep(3 * (attempt + 1)); continue if r.status_code != 200: log(f" Content HTTP {r.status_code} (attempt {attempt + 1}/{MAX_RETRIES}): {r.text[:200]}") time.sleep(3 * (attempt + 1)); continue try: data = r.json() except ValueError: log(" Content: invalid JSON, retry") time.sleep(3 * (attempt + 1)); continue if data.get("error"): # WB иногда отдаёт 200 + {"error": true} — то же, что 500 log(f" Content soft-error: {data.get('errorText')}; retry {attempt + 1}") time.sleep(3 * (attempt + 1)); continue page_cards = data.get("cards", []) for c in page_cards: cards.append({ "nmID": c.get("nmID"), "title": c.get("title") or "", "brand": c.get("brand") or "", "subjectName": c.get("subjectName") or "", "createdAt": c.get("createdAt") or "", }) page += 1 log(f" Content page {page}: +{len(page_cards)} (total {len(cards)})") if len(page_cards) < 100: return cards try: cursor = { "limit": 100, "updatedAt": data["cursor"]["updatedAt"], "nmID": data["cursor"]["nmID"], } except KeyError: log(" Content: no cursor in response, stopping pagination") return cards page_ok = True break if not page_ok: # Все ретраи страницы упали — best-effort, возвращаем накопленное log(f" Content: page failed after {MAX_RETRIES} retries; proceeding with {len(cards)} cards already loaded") return cards def fetch_funnel_batch( jwt: str, nm_ids: list[int], date_from: str, date_to: str ) -> tuple[list[dict[str, Any]] | None, str | None]: headers = {"Authorization": jwt, "Content-Type": "application/json"} body = { "nmIds": nm_ids, "selectedPeriod": {"start": date_from, "end": date_to}, "timezone": "Europe/Moscow", "aggregationLevel": "day", } last_err: str | None = None for attempt in range(MAX_RETRIES): try: r = requests.post(WB_FUNNEL_URL, json=body, headers=headers, timeout=60) except requests.RequestException as exc: last_err = f"Connection: {exc}" time.sleep(2 ** attempt) continue if r.status_code == 401: return None, "401 Unauthorized — JWT invalid or missing 'Аналитика'" if r.status_code == 403: return None, "403 Forbidden — JWT missing 'Аналитика'" if r.status_code == 429: sleep_for = (2 ** attempt) * 10 log(f" 429 rate-limit; sleeping {sleep_for}s") time.sleep(sleep_for); continue if r.status_code == 400: return None, f"400 Bad Request: {r.text[:300]}" if r.status_code != 200: last_err = f"HTTP {r.status_code}: {r.text[:200]}" time.sleep(2 ** attempt); continue try: payload = r.json() except ValueError: last_err = "Invalid JSON"; time.sleep(2 ** attempt); continue if isinstance(payload, list): return payload, None if isinstance(payload, dict) and "data" in payload: return payload["data"], None last_err = f"Unknown payload shape: {str(payload)[:200]}" break return None, last_err or "Unknown error" def select_nm_ids(cards: list[dict[str, Any]], mode: str, top_n: int | None, sku_list: list[int] | None, category: str | None) -> list[dict[str, Any]]: if mode == "all": return cards if mode == "sku_list": sku_set = set(sku_list or []) return [c for c in cards if c["nmID"] in sku_set] if mode == "category": return [c for c in cards if c["subjectName"] == category] if mode == "top_revenue": # Сначала тянем всех; обрежем по выручке после получения воронки. return cards raise ValueError(f"Unknown selection mode: {mode}") def normalize_history_entry(entry: dict[str, Any]) -> dict[str, Any]: """Приводим WB-имена полей к каноническим в нашем коде.""" return { # абсолютные счётчики "openCount": entry.get("openCount", 0) or 0, "cartCount": entry.get("cartCount", 0) or 0, "orderCount": entry.get("orderCount", 0) or 0, "orderSum": entry.get("orderSum", 0) or 0, "buyoutCount": entry.get("buyoutCount", 0) or 0, "buyoutSum": entry.get("buyoutSum", 0) or 0, "addToWishlistCount": entry.get("addToWishlistCount", 0) or 0, # производные (целые проценты от WB) "addToCartConversion": entry.get("addToCartConversion", 0) or 0, "cartToOrderConversion": entry.get("cartToOrderConversion", 0) or 0, "buyoutPercent": entry.get("buyoutPercent", 0) or 0, } def main() -> int: p = argparse.ArgumentParser() p.add_argument("--target-date", required=False) p.add_argument("--baseline-days", type=int, default=6, help="Сколько дней бейзлайна. Лимит WB: вместе с target ≤ 7 ⇒ макс 6.") p.add_argument("--selection", choices=["all", "top_revenue", "sku_list", "category"], default="top_revenue") p.add_argument("--top-n", type=int, default=200) p.add_argument("--sku-list", type=str, default="") p.add_argument("--category", type=str, default="") p.add_argument("--output", required=True) p.add_argument("--cache-dir", default=".wb-daily-drops.cache") args = p.parse_args() if args.baseline_days > 6: log(f"WARN: baseline-days={args.baseline_days} > 6, WB API ограничивает периодом 7 дней; clamping to 6") args.baseline_days = 6 jwt = get_jwt() today_msk = datetime.now(MSK).date() target_date = date.fromisoformat(args.target_date) if args.target_date else (today_msk - timedelta(days=1)) begin_date = target_date - timedelta(days=args.baseline_days) log(f"Period: {begin_date} → {target_date} (target={target_date}, baseline {args.baseline_days}d, MSK)") log("Step 1: fetch cards (Content API, w/ daily cache)") cards = fetch_cards_cached(jwt, Path(args.cache_dir)) log(f" total active cards: {len(cards)}") if not cards: return 3 sku_list_parsed: list[int] = [] if args.sku_list: sku_list_parsed = [int(x.strip()) for x in args.sku_list.split(",") if x.strip()] selected = select_nm_ids(cards, args.selection, args.top_n, sku_list_parsed, args.category or None) log(f" selected for funnel: {len(selected)}") if not selected: return 3 log("Step 2: fetch funnel data") by_nm: dict[int, dict[str, Any]] = {} errors: list[str] = [] nm_to_card = {c["nmID"]: c for c in selected} nm_ids = [c["nmID"] for c in selected if c["nmID"]] for i in range(0, len(nm_ids), BATCH_SIZE): batch = nm_ids[i : i + BATCH_SIZE] log(f" batch {i // BATCH_SIZE + 1}/{(len(nm_ids) + BATCH_SIZE - 1) // BATCH_SIZE} (size {len(batch)})") rows, err = fetch_funnel_batch(jwt, batch, begin_date.isoformat(), target_date.isoformat()) if err: log(f" batch failed: {err}"); errors.append(f"batch {i}: {err}") if rows: for row in rows: product = row.get("product") or {} nm_id = product.get("nmId") or row.get("nmID") if not nm_id: continue history = row.get("history") or [] if nm_id not in by_nm: by_nm[nm_id] = {"card": nm_to_card.get(nm_id, {}), "dates": {}} for entry in history: d = entry.get("date") or entry.get("dt") if not d: continue by_nm[nm_id]["dates"][d] = normalize_history_entry(entry) if i + BATCH_SIZE < len(nm_ids): time.sleep(RATE_LIMIT_SLEEP_SECONDS) if args.selection == "top_revenue": log(f" applying top-{args.top_n} by revenue cutoff") baseline_set = {(target_date - timedelta(days=i)).isoformat() for i in range(1, args.baseline_days + 1)} def base_rev(v: dict[str, Any]) -> float: return sum(d.get("orderSum", 0) for date_iso, d in v.get("dates", {}).items() if date_iso in baseline_set) ranked = sorted(by_nm.items(), key=lambda kv: base_rev(kv[1]), reverse=True) kept = {nm_id for nm_id, _ in ranked[: args.top_n]} by_nm = {nm_id: v for nm_id, v in by_nm.items() if nm_id in kept} log(f" kept {len(by_nm)} SKUs after top-{args.top_n}") target_iso = target_date.isoformat() rows_at_target = sum(1 for v in by_nm.values() if target_iso in v.get("dates", {})) log(f" rows at target_date={target_iso}: {rows_at_target}") if rows_at_target == 0: log("CRITICAL: no data for target date") with open(args.output, "w", encoding="utf-8") as f: json.dump({"target_date": target_iso, "status": "no_data_yet", "errors": errors}, f, ensure_ascii=False, indent=2) return 2 out = { "target_date": target_iso, "baseline_dates": [(target_date - timedelta(days=i)).isoformat() for i in range(1, args.baseline_days + 1)], "by_nm": {str(k): v for k, v in by_nm.items()}, "errors": errors, "selection": { "mode": args.selection, "top_n": args.top_n, "sku_list": sku_list_parsed, "category": args.category, }, } with open(args.output, "w", encoding="utf-8") as f: json.dump(out, f, ensure_ascii=False, indent=2) log(f"Saved {args.output}: {len(by_nm)} SKUs, {len(errors)} batch errors") return 0 if __name__ == "__main__": sys.exit(main())