#!/usr/bin/env python3 import os, json, time, math, html, csv from pathlib import Path from datetime import date, timedelta import requests from openpyxl import Workbook from openpyxl.styles import Font, PatternFill, Alignment from openpyxl.utils import get_column_letter OUTDIR = Path('/var/www/hermes-artifacts/wb_lyskogor_sales_drops_2026-08-25') OUTDIR.mkdir(parents=True, exist_ok=True) TODAY = date(2026, 8, 25) END = TODAY - timedelta(days=1) # последняя полная дата WB START = END - timedelta(days=6) PAST_END = START - timedelta(days=1) PAST_START = PAST_END - timedelta(days=6) URL = 'https://seller-analytics-api.wildberries.ru/api/analytics/v3/sales-funnel/products' def load_env(): env = {} for p in ['/root/.hermes/.env', '/root/Hermes_doc/tools/marketplaces/api_accounts.env']: pp = Path(p) if not pp.exists(): continue for line in pp.read_text(encoding='utf-8', errors='ignore').splitlines(): s = line.strip() if s and not s.startswith('#') and '=' in s: k, v = s.split('=', 1) env[k.strip()] = v.strip().strip('"').strip("'") return env def get_token(): env = load_env() for k in ['WB_ANALYTICS_TOKEN', 'WB_MAIN_ANALYTICS_TOKEN', 'WB_API_KEY', 'WB_MAIN_API_KEY']: if env.get(k): return env[k], k raise SystemExit('WB token not found') def post_retry(token, payload, label): headers = {'Authorization': token, 'Content-Type': 'application/json'} for attempt in range(6): r = requests.post(URL, json=payload, headers=headers, timeout=60) if r.status_code == 429: wait = int(r.headers.get('X-RateLimit-Retry', '60') or '60') time.sleep(max(20, min(wait, 90))) continue if r.status_code >= 500: time.sleep(5 * (attempt + 1)); continue if r.status_code != 200: raise RuntimeError(f'{label}: HTTP {r.status_code}: {r.text[:800]}') return r.json() raise RuntimeError(f'{label}: failed after retries') def fetch_all(token): # Один запрос limit=1000 обычно покрывает весь кабинет; пагинация оставлена на случай расширения ассортимента. out = [] offset = 0 limit = 1000 while True: payload = { 'selectedPeriod': {'start': START.isoformat(), 'end': END.isoformat()}, 'pastPeriod': {'start': PAST_START.isoformat(), 'end': PAST_END.isoformat()}, 'limit': limit, 'offset': offset, 'orderBy': {'field': 'orderSum', 'mode': 'desc'}, } data = post_retry(token, payload, f'offset={offset}') products = ((data.get('data') or {}).get('products') or []) out.extend(products) if len(products) < limit: break offset += limit time.sleep(20) return out def metric(product_row, period, key, default=0): return (((product_row.get('statistic') or {}).get(period) or {}).get(key, default) or 0) def conv(product_row, period, key): return (((((product_row.get('statistic') or {}).get(period) or {}).get('conversions') or {}).get(key, 0)) or 0) def pct(new, old): if old == 0: if new == 0: return None return math.inf return (new - old) / old * 100 def fmt_pct(x): if x is None: return '—' if x == math.inf: return '+∞' return f'{x:+.1f}%' def fmt_num(x): return f'{x:,.0f}'.replace(',', ' ') def fmt_money(x): return f'{x:,.0f} ₽'.replace(',', ' ') def reason(row): reasons=[] if row['open_pct'] is not None and row['open_pct'] <= -20: reasons.append('упали показы') if row['cart_pct'] is not None and row['cart_pct'] <= -20: reasons.append('упали корзины') if row['atc_pp'] <= -2: reasons.append('хуже показ→корзина') if row['cto_pp'] <= -5: reasons.append('хуже корзина→заказ') if row['selected_orders'] == 0 and row['past_orders'] > 0: reasons.append('0 заказов за 7д') if not reasons: reasons.append('продажи ниже без явного слома воронки') return ', '.join(reasons[:3]) def main(): token, token_key = get_token() products = fetch_all(token) rows=[] totals={k:0 for k in ['selected_revenue','past_revenue','selected_orders','past_orders','selected_opens','past_opens','selected_carts','past_carts']} for pr in products: p=pr.get('product') or {} st_sel=(pr.get('statistic') or {}).get('selected') or {} st_past=(pr.get('statistic') or {}).get('past') or {} stocks=p.get('stocks') or {} row={ 'nmID': p.get('nmId'), 'vendorCode': p.get('vendorCode') or 'нет данных', 'title': p.get('title') or '', 'brand': p.get('brandName') or '', 'subject': p.get('subjectName') or '', 'rating': p.get('feedbackRating') or 0, 'productRating': p.get('productRating') or 0, 'stock_wb': stocks.get('wb') or 0, 'stock_mp': stocks.get('mp') or 0, 'stock_total': (stocks.get('wb') or 0) + (stocks.get('mp') or 0), 'stock_sum': stocks.get('balanceSum') or 0, 'selected_opens': st_sel.get('openCount') or 0, 'past_opens': st_past.get('openCount') or 0, 'selected_carts': st_sel.get('cartCount') or 0, 'past_carts': st_past.get('cartCount') or 0, 'selected_orders': st_sel.get('orderCount') or 0, 'past_orders': st_past.get('orderCount') or 0, 'selected_revenue': st_sel.get('orderSum') or 0, 'past_revenue': st_past.get('orderSum') or 0, 'selected_avg_price': st_sel.get('avgPrice') or 0, 'past_avg_price': st_past.get('avgPrice') or 0, 'selected_atc': conv(pr,'selected','addToCartPercent'), 'past_atc': conv(pr,'past','addToCartPercent'), 'selected_cto': conv(pr,'selected','cartToOrderPercent'), 'past_cto': conv(pr,'past','cartToOrderPercent'), 'selected_buyout': conv(pr,'selected','buyoutPercent'), 'past_buyout': conv(pr,'past','buyoutPercent'), 'selected_localization': st_sel.get('localizationPercent') or 0, 'past_localization': st_past.get('localizationPercent') or 0, } row.update({ 'revenue_abs': row['selected_revenue'] - row['past_revenue'], 'orders_abs': row['selected_orders'] - row['past_orders'], 'open_abs': row['selected_opens'] - row['past_opens'], 'cart_abs': row['selected_carts'] - row['past_carts'], 'revenue_pct': pct(row['selected_revenue'], row['past_revenue']), 'orders_pct': pct(row['selected_orders'], row['past_orders']), 'open_pct': pct(row['selected_opens'], row['past_opens']), 'cart_pct': pct(row['selected_carts'], row['past_carts']), 'atc_pp': row['selected_atc'] - row['past_atc'], 'cto_pp': row['selected_cto'] - row['past_cto'], 'buyout_pp': row['selected_buyout'] - row['past_buyout'], }) row['reason'] = reason(row) for k in totals: totals[k] += row[k] rows.append(row) # Significant drops: past had real volume, current lower by revenue/orders. drops=[r for r in rows if (r['past_orders']>=2 or r['past_revenue']>=3000) and (r['revenue_abs']<0 or r['orders_abs']<0)] drops.sort(key=lambda r: (r['revenue_abs'], r['orders_abs'])) severe=[r for r in drops if (r['revenue_pct'] is not None and r['revenue_pct'] <= -30 and r['revenue_abs'] <= -1000) or (r['selected_orders']==0 and r['past_orders']>=2)] growth=[r for r in rows if r['revenue_abs']>0 and (r['selected_orders']>=2 or r['selected_revenue']>=3000)] growth.sort(key=lambda r: r['revenue_abs'], reverse=True) # Excel xlsx=OUTDIR/'WB_Лысогор_просадки_продаж_7д_18-24_авг_2026.xlsx' wb=Workbook() ws=wb.active; ws.title='Просадки продаж' headers=['nmID / vendorCode','brand','title','subject','reason','past_revenue','selected_revenue','revenue_abs','revenue_pct','past_orders','selected_orders','orders_abs','orders_pct','past_opens','selected_opens','open_pct','past_carts','selected_carts','cart_pct','past_atc','selected_atc','atc_pp','past_cto','selected_cto','cto_pp','stock_wb','stock_mp','stock_total','stock_sum','rating','productRating'] ws.append(headers) for r in drops: vals=[f"{r['nmID']} / {r['vendorCode']}", r['brand'], r['title'], r['subject'], r['reason']] vals += [r[h] for h in headers[5:]] ws.append(vals) for c in ws[1]: c.fill=PatternFill('solid', fgColor='1F2937'); c.font=Font(color='FFFFFF', bold=True); c.alignment=Alignment(wrap_text=True) for row in ws.iter_rows(min_row=2): row[4].fill=PatternFill('solid', fgColor='FEF3C7') for cell in row: cell.alignment=Alignment(vertical='top', wrap_text=True) ws.freeze_panes='A2'; ws.auto_filter.ref=ws.dimensions widths=[32,12,42,22,34,14,14,14,12,10,10,10,12,12,12,12,12,12,12,10,10,10,10,10,10,10,10,10,14,8,8] for i,w in enumerate(widths,1): ws.column_dimensions[get_column_letter(i)].width=w ws2=wb.create_sheet('Сводка') summary_rows=[ ['Период текущий', f'{START}—{END}'], ['Период сравнения', f'{PAST_START}—{PAST_END}'], ['Всего карточек в API', len(rows)], ['Карточек с просадкой', len(drops)], ['Сильных просадок', len(severe)], ['Выручка текущие 7д', totals['selected_revenue']], ['Выручка предыдущие 7д', totals['past_revenue']], ['Дельта выручки', totals['selected_revenue']-totals['past_revenue'], pct(totals['selected_revenue'], totals['past_revenue'])], ['Заказы текущие 7д', totals['selected_orders']], ['Заказы предыдущие 7д', totals['past_orders']], ['Дельта заказов', totals['selected_orders']-totals['past_orders'], pct(totals['selected_orders'], totals['past_orders'])], ['Показы текущие 7д', totals['selected_opens']], ['Показы предыдущие 7д', totals['past_opens'], pct(totals['selected_opens'], totals['past_opens'])], ['Корзины текущие 7д', totals['selected_carts']], ['Корзины предыдущие 7д', totals['past_carts'], pct(totals['selected_carts'], totals['past_carts'])], ] for row in summary_rows: ws2.append(row) ws3=wb.create_sheet('Росты') ws3.append(headers) for r in growth: vals=[f"{r['nmID']} / {r['vendorCode']}", r['brand'], r['title'], r['subject'], r['reason']] vals += [r[h] for h in headers[5:]] ws3.append(vals) wb.save(xlsx) # CSV all drops csvp=OUTDIR/'WB_Лысогор_просадки_продаж_7д.csv' with open(csvp,'w',encoding='utf-8-sig',newline='') as f: fieldnames=list(drops[0].keys()) if drops else [] w=csv.DictWriter(f, fieldnames=fieldnames); w.writeheader(); w.writerows(drops) # HTML def tr(r): return f"{r['nmID']} / {html.escape(str(r['vendorCode']))}
{html.escape(r['brand'])}{html.escape(r['title'])}
{html.escape(r['subject'])}{html.escape(r['reason'])}{fmt_money(r['past_revenue'])} → {fmt_money(r['selected_revenue'])}
{fmt_money(r['revenue_abs'])} / {fmt_pct(r['revenue_pct'])}{fmt_num(r['past_orders'])} → {fmt_num(r['selected_orders'])}
{fmt_num(r['orders_abs'])} / {fmt_pct(r['orders_pct'])}{fmt_num(r['past_opens'])} → {fmt_num(r['selected_opens'])}
{fmt_pct(r['open_pct'])}{fmt_num(r['past_carts'])} → {fmt_num(r['selected_carts'])}
{fmt_pct(r['cart_pct'])}WB {fmt_num(r['stock_wb'])}
FBS {fmt_num(r['stock_mp'])}
Всего {fmt_num(r['stock_total'])}" htmlp=OUTDIR/'WB_Лысогор_просадки_продаж_7д_18-24_авг_2026.html' cards=f"""
Карточек в API
{len(rows)}
Проседают
{len(drops)}
с объёмом в прошлом периоде
Сильные просадки
{len(severe)}
-30%+ и -1000 ₽ / 0 заказов
Итог выручки 7д
{fmt_pct(pct(totals['selected_revenue'], totals['past_revenue']))}
{fmt_money(totals['past_revenue'])} → {fmt_money(totals['selected_revenue'])}
""" html_doc=f"""

WB Лысогор — карточки, проседающие в продажах за последние 7 дней

Текущий период: {START}—{END}. Сравнение: {PAST_START}—{PAST_END}. Формат ID: nmID / артикул продавца.
{cards}

Топ сильных просадок

{''.join(tr(r) for r in severe[:30])}
nmID / vendorCodeТоварСигналВыручкаЗаказыПоказыКорзиныОстаток

Все просадки с объёмом

{''.join(tr(r) for r in drops)}
nmID / vendorCodeТоварСигналВыручкаЗаказыПоказыКорзиныОстаток

Полная таблица и росты — в Excel: {xlsx.name}

""" htmlp.write_text(html_doc, encoding='utf-8') result={ 'period': {'selected':[START.isoformat(),END.isoformat()], 'past':[PAST_START.isoformat(),PAST_END.isoformat()]}, 'token_key_used': token_key, 'counts': {'api_products':len(rows), 'drops':len(drops), 'severe':len(severe), 'growth':len(growth)}, 'totals': totals | {'revenue_pct': pct(totals['selected_revenue'], totals['past_revenue']), 'orders_pct': pct(totals['selected_orders'], totals['past_orders']), 'opens_pct': pct(totals['selected_opens'], totals['past_opens']), 'carts_pct': pct(totals['selected_carts'], totals['past_carts'])}, 'severe': severe[:20], 'drops': drops[:20], 'growth': growth[:10], 'files': {'html':str(htmlp), 'xlsx':str(xlsx), 'csv':str(csvp)}, } (OUTDIR/'summary.json').write_text(json.dumps(result, ensure_ascii=False, indent=2, default=str), encoding='utf-8') print(json.dumps(result, ensure_ascii=False, indent=2, default=str)) if __name__ == '__main__': main()