#!/usr/bin/env python3 """ enrich_stocks.py — добавляет к insights.json колонки по остаткам и OOS-флаги. Делает ОДИН запрос к supplier/stocks за сегодня. Если в кэше есть вчерашний снимок остатков — детектит выпадение конкретного склада (квази-dropoff). Сегодняшний снимок сохраняется в кэш для следующего прогона. Флаги: - OOS_full: суммарный остаток = 0 - OOS_warehouse_dropoff: остаток на складе упал с >0 до 0 (между вчера и сегодня) - OK: иначе Stocks API имеет жёсткий rate-limit 1 req/min, поэтому второй запрос мы не делаем. """ 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 typing import Any import requests URL = "https://statistics-api.wildberries.ru/api/v1/supplier/stocks" MAX_RETRIES = 5 def log(msg: str) -> None: print(f"[stocks] {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_stocks(jwt: str, date_from: str) -> list[dict[str, Any]] | None: headers = {"Authorization": jwt} params = {"dateFrom": f"{date_from}T00:00:00"} for attempt in range(MAX_RETRIES): try: r = requests.get(URL, headers=headers, params=params, timeout=120) except requests.RequestException as exc: log(f" connection: {exc}; retry {attempt + 1}") time.sleep(5 * (attempt + 1)) continue if r.status_code in (401, 403): log(f" {r.status_code} no access to stocks; skipping") return None if r.status_code == 429: log(f" 429; sleeping 60s") time.sleep(60) continue if r.status_code != 200: log(f" HTTP {r.status_code}: {r.text[:200]}") time.sleep(5 * (attempt + 1)) continue try: return r.json() except ValueError: log(" invalid JSON") return None return None def group_by_nm_warehouse(rows: list[dict[str, Any]]) -> dict[int, dict[str, int]]: grouped: dict[int, dict[str, int]] = {} for r in rows or []: nm = r.get("nmId") wh = r.get("warehouseName") or "—" q = r.get("quantityFull", 0) or 0 grouped.setdefault(nm, {}).setdefault(wh, 0) grouped[nm][wh] += q return grouped def make_flag(nm_id: int, today_map: dict[int, dict[str, int]], yest_map: dict[int, dict[str, int]]) -> dict[str, Any]: """Компактный вывод: только агрегаты + dropoff (или единственный склад в случае OOS). by_warehouse_today целиком НЕ возвращаем — это 200-800 байт на SKU, балласт для отчёта.""" today_wh = today_map.get(nm_id, {}) yest_wh = yest_map.get(nm_id, {}) total_today = sum(today_wh.values()) total_yest = sum(yest_wh.values()) flag = "OK" dropoff_warehouse = None if total_today == 0: flag = "OOS_full" else: for wh, q_yest in yest_wh.items(): if q_yest > 0 and today_wh.get(wh, 0) == 0: flag = "OOS_warehouse_dropoff" dropoff_warehouse = wh break out: dict[str, Any] = { "total_today": total_today, "total_yesterday": total_yest, "flag": flag, "dropoff_warehouse": dropoff_warehouse, } # Включаем разбивку по складам только в крайних случаях, когда она информативна if flag in ("OOS_full", "OOS_warehouse_dropoff"): # Топ-3 склада сегодня — для контекста top_wh = sorted(today_wh.items(), key=lambda kv: -kv[1])[:3] out["top_warehouses"] = {k: v for k, v in top_wh if v > 0} return out def main() -> int: p = argparse.ArgumentParser() p.add_argument("--insights", required=True, help="Path to insights.json (will be updated in place)") p.add_argument("--cache-dir", default=".wb-daily-drops.cache") args = p.parse_args() with open(args.insights, encoding="utf-8") as f: insights = json.load(f) target_date = date.fromisoformat(insights["target_date"]) today = target_date + timedelta(days=1) # «сегодня» — день, в который запускаем скилл cache_dir = Path(args.cache_dir) cache_dir.mkdir(parents=True, exist_ok=True) today_cache = cache_dir / f"stocks_{today.isoformat()}.json" prev_cache = cache_dir / f"stocks_{target_date.isoformat()}.json" log(f"Fetching stocks today={today.isoformat()}") jwt = get_jwt() today_stocks = fetch_stocks(jwt, today.isoformat()) if today_stocks is None: log("Skipped: stocks unavailable (no scope or hard error)") for section in ("drops", "growths", "oos_candidates"): for s in insights.get(section, []): s.setdefault("enrichment", {})["stocks"] = {"flag": "NO_DATA"} with open(args.insights, "w", encoding="utf-8") as f: json.dump(insights, f, ensure_ascii=False, indent=2) return 0 # Сохраняем сегодняшний снимок в кэш для следующего прогона try: with open(today_cache, "w", encoding="utf-8") as f: json.dump(today_stocks, f, ensure_ascii=False) except OSError as e: log(f" WARN: failed to cache stocks: {e}") # Ищем вчерашний снимок в кэше (не делаем второй API-запрос — rate-limit 1/min) yest_stocks: list[dict[str, Any]] = [] if prev_cache.exists(): try: with open(prev_cache, encoding="utf-8") as f: yest_stocks = json.load(f) log(f" found cached yesterday stocks: {len(yest_stocks)} rows") except (OSError, ValueError) as e: log(f" WARN: failed to load yesterday cache: {e}") else: log(" no yesterday cache — dropoff detection will be unavailable until next run") today_map = group_by_nm_warehouse(today_stocks) yest_map = group_by_nm_warehouse(yest_stocks) enriched_count = 0 for section in ("drops", "growths", "oos_candidates"): for s in insights.get(section, []): nm = s["nmID"] flag_info = make_flag(nm, today_map, yest_map) if not yest_stocks: flag_info["dropoff_warehouse"] = None # OOS_full всё ещё считается, но OOS_warehouse_dropoff невозможен if flag_info["flag"] == "OOS_warehouse_dropoff": flag_info["flag"] = "OK" # ложный без вчерашнего снимка s.setdefault("enrichment", {})["stocks"] = flag_info enriched_count += 1 with open(args.insights, "w", encoding="utf-8") as f: json.dump(insights, f, ensure_ascii=False, indent=2) log(f" enriched {enriched_count} SKUs with stocks data") return 0 if __name__ == "__main__": sys.exit(main())