#!/usr/bin/env python3 from __future__ import annotations import os, sys, json, time, math from datetime import datetime, timedelta, date from zoneinfo import ZoneInfo from collections import defaultdict import requests MSK=ZoneInfo('Europe/Moscow') TOKEN=os.environ.get('WB_ANALYTICS_TOKEN') or os.environ.get('WB_API_KEY') or os.environ.get('WB_TOKEN') or os.environ.get('WB_ANALYTICS_JWT') or os.environ.get('WB_JWT') if not TOKEN: print('ERROR: WB token env not found', file=sys.stderr); sys.exit(3) HEADERS={'Authorization': TOKEN, 'Content-Type':'application/json'} CONTENT_URL='https://content-api.wildberries.ru/content/v2/get/cards/list' FUNNEL_URL='https://seller-analytics-api.wildberries.ru/api/analytics/v3/sales-funnel/products/history' STOCKS_URL='https://statistics-api.wildberries.ru/api/v1/supplier/stocks' def log(s): print(f'[{datetime.now(MSK):%H:%M:%S}] {s}', file=sys.stderr, flush=True) def post_retry(url, body, timeout=60, tries=5, sleep429=60): last='' for a in range(tries): try: r=requests.post(url, headers=HEADERS, json=body, timeout=timeout) except Exception as e: last=f'connection {e}'; time.sleep(2*(a+1)); continue if r.status_code==429: wait=sleep429 if a>=1 else 20 log(f'429 {url}; sleep {wait}s'); time.sleep(wait); continue if r.status_code>=500: last=f'HTTP {r.status_code}: {r.text[:200]}'; time.sleep(3*(a+1)); continue return r raise RuntimeError(last or 'request failed') def fetch_cards(): cards=[]; cursor={'limit':100}; page=0 while True: body={'settings': {'cursor': cursor, 'filter': {'withPhoto': -1}}} r=post_retry(CONTENT_URL, body, timeout=30, sleep429=10) if r.status_code in (401,403): raise RuntimeError(f'Content auth {r.status_code}: {r.text[:200]}') if r.status_code!=200: raise RuntimeError(f'Content HTTP {r.status_code}: {r.text[:300]}') data=r.json() if data.get('error'): raise RuntimeError(f"Content soft error: {data.get('errorText')}") page_cards=data.get('cards') or [] for c in page_cards: nm=c.get('nmID') if nm: cards.append({'nmID': int(nm), 'title': c.get('title') or '', 'brand': c.get('brand') or '', 'subjectName': c.get('subjectName') or '', 'vendorCode': c.get('vendorCode') or '', 'createdAt': c.get('createdAt') or ''}) page+=1; log(f'Content page {page}: +{len(page_cards)} total={len(cards)}') if len(page_cards)<100: break cur=data.get('cursor') or {} if not cur.get('updatedAt') or not cur.get('nmID'): break cursor={'limit':100,'updatedAt':cur['updatedAt'],'nmID':cur['nmID']} return cards def fetch_stocks(): # current stock snapshot from Statistics API params={'dateFrom': datetime.now(MSK).date().isoformat()+'T00:00:00'} for a in range(5): r=requests.get(STOCKS_URL, headers={'Authorization': TOKEN}, params=params, timeout=120) if r.status_code==429: log('Stocks 429; sleep 60s'); time.sleep(60); continue if r.status_code in (401,403): log(f'Stocks no access {r.status_code}: {r.text[:160]}'); return None if r.status_code!=200: log(f'Stocks HTTP {r.status_code}: {r.text[:160]}'); time.sleep(5*(a+1)); continue rows=r.json(); break else: return None by=defaultdict(lambda: {'qty':0,'warehouses':defaultdict(int)}) for x in rows or []: nm=x.get('nmId') or x.get('nmID') if not nm: continue q=x.get('quantityFull') if q is None: q=(x.get('quantity') or 0)+(x.get('inWayToClient') or 0)+(x.get('inWayFromClient') or 0) q=int(q or 0) wh=x.get('warehouseName') or '—' by[int(nm)]['qty']+=q; by[int(nm)]['warehouses'][wh]+=q out={nm:{'qty':v['qty'],'top_wh': sorted(v['warehouses'].items(), key=lambda kv:-kv[1])[:3]} for nm,v in by.items()} log(f'Stocks rows={len(rows or [])}, nm_with_stock_rows={len(out)}') return out def fetch_funnel(nm_ids, start, end): by={} batches=[nm_ids[i:i+20] for i in range(0,len(nm_ids),20)] for bi,b in enumerate(batches,1): log(f'Funnel batch {bi}/{len(batches)} size={len(b)}') body={'nmIds': b, 'selectedPeriod': {'start': start, 'end': end}, 'timezone':'Europe/Moscow', 'aggregationLevel':'day'} r=post_retry(FUNNEL_URL, body, timeout=90, sleep429=60) if r.status_code in (401,403): raise RuntimeError(f'Funnel auth {r.status_code}: {r.text[:200]}') if r.status_code==400: raise RuntimeError(f'Funnel 400: {r.text[:500]}') if r.status_code!=200: raise RuntimeError(f'Funnel HTTP {r.status_code}: {r.text[:300]}') payload=r.json(); rows=payload.get('data') if isinstance(payload,dict) else payload for row in rows or []: product=row.get('product') or {} nm=product.get('nmId') or row.get('nmID') or row.get('nmId') if not nm: continue hist={} for e in row.get('history') or []: d=e.get('date') or e.get('dt') if d: hist[d[:10]]={ 'openCount': e.get('openCount',0) or 0, 'cartCount': e.get('cartCount',0) or 0, 'orderCount': e.get('orderCount',0) or 0, 'orderSum': e.get('orderSum',0) or 0, 'buyoutCount': e.get('buyoutCount',0) or 0, } by[int(nm)]=hist if bi < len(batches): time.sleep(20) return by def pct(today, base): if base==0: return None if today==0 else math.inf return (today-base)/base*100 def avg(hist, dates, key): vals=[(hist.get(d) or {}).get(key,0) or 0 for d in dates if d in hist] if not vals: return None return sum(vals)/len(vals) def main(): today=datetime.now(MSK).date() target=today-timedelta(days=1) # WB history endpoint max 7 days inclusive: target + previous 6 days. base_dates=[(target-timedelta(days=i)).isoformat() for i in range(1,7)] start=base_dates[-1]; end=target.isoformat() log(f'Period {start}..{end}; target={end}; baseline_days={base_dates}') cards=fetch_cards(); nm_ids=[c['nmID'] for c in cards] stocks=fetch_stocks() funnel=fetch_funnel(nm_ids, start, end) card_map={c['nmID']:c for c in cards} rows=[]; skipped_no_stock=0; skipped_no_base=0; skipped_no_target=0 for nm,hist in funnel.items(): st=(stocks or {}).get(nm, {'qty': None, 'top_wh': []}) qty=st.get('qty') if stocks is not None and (qty or 0) <= 0: skipped_no_stock+=1; continue target_day=hist.get(end) if not target_day: skipped_no_target+=1; continue base_orders=avg(hist, base_dates, 'orderCount') base_rev=avg(hist, base_dates, 'orderSum') base_opens=avg(hist, base_dates, 'openCount') base_carts=avg(hist, base_dates, 'cartCount') present_base_days=sum(1 for d in base_dates if d in hist) if present_base_days < 3 or ((base_orders or 0)<=0 and (base_rev or 0)<=0): skipped_no_base+=1; continue t_orders=target_day.get('orderCount',0) or 0 t_rev=target_day.get('orderSum',0) or 0 d_orders=pct(t_orders, base_orders or 0) d_rev=pct(t_rev, base_rev or 0) # Sales decrease = revenue or orders down by 20%+ vs previous 6-day average. if not ((d_orders is not None and d_orders <= -20) or (d_rev is not None and d_rev <= -20)): continue c=card_map.get(nm,{}) first='заказы' if (d_orders is not None and d_orders<=-20) else 'выручка' rows.append({ 'nmID':nm,'brand':c.get('brand',''),'title':c.get('title',''),'subjectName':c.get('subjectName',''),'vendorCode':c.get('vendorCode',''), 'stock_qty':qty,'top_wh':st.get('top_wh',[]), 'orders_yesterday':t_orders,'orders_base_avg':base_orders,'orders_delta_pct':d_orders, 'revenue_yesterday':t_rev,'revenue_base_avg':base_rev,'revenue_delta_pct':d_rev, 'opens_yesterday':target_day.get('openCount',0) or 0,'opens_base_avg':base_opens,'opens_delta_pct':pct(target_day.get('openCount',0) or 0, base_opens or 0), 'carts_yesterday':target_day.get('cartCount',0) or 0,'carts_base_avg':base_carts,'carts_delta_pct':pct(target_day.get('cartCount',0) or 0, base_carts or 0), 'first_signal':first,'base_days':present_base_days, 'wb_url':f'https://www.wildberries.ru/catalog/{nm}/detail.aspx' }) rows.sort(key=lambda r: (r['revenue_yesterday']-r['revenue_base_avg'])) account={ 'cards_total':len(cards),'funnel_skus':len(funnel),'target_date':end,'baseline_dates':base_dates, 'with_sales_drop_and_stock':len(rows),'skipped_no_stock':skipped_no_stock,'skipped_no_target':skipped_no_target,'skipped_no_base':skipped_no_base, 'stocks_available': stocks is not None, } out={'generated_at':datetime.now(MSK).isoformat(timespec='seconds'), 'account':account, 'rows':rows} os.makedirs('/var/www/hermes-artifacts/wb-drop-report', exist_ok=True) json_path='/var/www/hermes-artifacts/wb-drop-report/wb_sales_drop_with_stock.json' with open(json_path,'w',encoding='utf-8') as f: json.dump(out,f,ensure_ascii=False,indent=2) # CSV for Excel import import csv csv_path='/var/www/hermes-artifacts/wb-drop-report/wb_sales_drop_with_stock.csv' fields=['brand','nmID','vendorCode','title','subjectName','stock_qty','orders_yesterday','orders_base_avg','orders_delta_pct','revenue_yesterday','revenue_base_avg','revenue_delta_pct','opens_delta_pct','carts_delta_pct','first_signal','base_days','wb_url'] with open(csv_path,'w',encoding='utf-8-sig',newline='') as f: w=csv.DictWriter(f,fieldnames=fields); w.writeheader(); for r in rows: w.writerow({k:r.get(k) for k in fields}) print(json.dumps({'json':json_path,'csv':csv_path,'account':account,'top10':rows[:10]},ensure_ascii=False,indent=2)) if __name__=='__main__': main()