#!/usr/bin/env python3 """ Backfill CAPI — envia pro Meta as vendas APPROVED/COMPLETE da Hotmart num periodo, usando o mesmo pipeline do webhook (build_capi_event + send_to_meta). Uso: python backfill_capi.py --start 2026-06-25 --end 2026-08-25 [--dry-run] python backfill_capi.py --start 2026-06-25 --end 2026-08-25 --product-id 1437935 Padrao: - Periodo: ultimos 15 dias - --product-id: se omitido, puxa TODOS os produtos do produtor - Loop 2 status: APPROVED + COMPLETE - Dedupe local (SQLite do webhook) evita duplicata na Meta Dry-run: nao envia, mostra tabela agrupada por produto (contagem + valor total). """ from __future__ import annotations import argparse import os import sys import time from collections import defaultdict from datetime import datetime, timedelta, timezone from pathlib import Path import requests # Reusa infra do app sys.path.insert(0, str(Path(__file__).resolve().parent)) from app import ( # noqa: E402 HOTTOK, STUB_MODE, _get, already_processed, build_capi_event, log, mark_processed, send_to_meta, ) HOTMART_AUTH_URL = "https://api-sec-vlc.hotmart.com/security/oauth/token" HOTMART_BASE = "https://developers.hotmart.com/payments/api/v1" HOTMART_SALES_URL = f"{HOTMART_BASE}/sales/history" CLIENT_ID = os.getenv("HOTMART_CLIENT_ID", "").strip() CLIENT_SECRET = os.getenv("HOTMART_CLIENT_SECRET", "").strip() BASIC = os.getenv("HOTMART_BASIC", "").strip() DEFAULT_PRODUCT_ID = os.getenv("HOTMART_PRODUCT_ID", "").strip() # Status que o backfill deve varrer. Ambos viram Purchase (mesmo event_id = # transaction_id), entao o dedupe local garante que a Meta so recebe UMA vez. STATUSES_TO_BACKFILL = ["APPROVED", "COMPLETE"] def hotmart_token() -> str: log.info("Autenticando na Hotmart...") r = requests.post( HOTMART_AUTH_URL, headers={"Authorization": BASIC}, params={ "grant_type": "client_credentials", "client_id": CLIENT_ID, "client_secret": CLIENT_SECRET, }, timeout=30, ) r.raise_for_status() tok = r.json().get("access_token") if not tok: raise RuntimeError(f"Sem access_token: {r.text}") log.info("Auth Hotmart OK.") return tok def _to_ms(dt: datetime) -> int: return int(dt.replace(tzinfo=timezone.utc).timestamp() * 1000) def fetch_sales(token: str, start_dt: datetime, end_dt: datetime, status: str, product_id: str | None): """Itera pagina a pagina retornando itens do produto (ou de todos).""" headers = {"Authorization": f"Bearer {token}"} params: dict = { "transaction_status": status, "start_date": _to_ms(start_dt), "end_date": _to_ms(end_dt), "max_results": 500, } if product_id: params["product_id"] = product_id page = 0 while True: page += 1 r = requests.get(HOTMART_SALES_URL, headers=headers, params=params, timeout=60) if r.status_code != 200: log.error("Hotmart sales/history %s [%s]: %s", r.status_code, status, r.text[:300]) r.raise_for_status() data = r.json() items = data.get("items", []) or [] log.info("[%s] Pagina %d: %d itens", status, page, len(items)) for it in items: yield it page_info = data.get("page_info", {}) or {} next_token = page_info.get("next_page_token") if not next_token: break params = {"page_token": next_token} def item_to_data(item: dict) -> dict: """ O item de /sales/history vem no formato: { "product": {...}, "buyer": {...}, "purchase": {...} } exatamente o mesmo shape do 'data' do webhook 2.0 — reusa direto. """ return item def main(): parser = argparse.ArgumentParser() parser.add_argument("--start", default=None, help="Data inicio YYYY-MM-DD (padrao: 15 dias atras)") parser.add_argument("--end", default=None, help="Data fim YYYY-MM-DD (padrao: hoje)") parser.add_argument("--product-id", default=None, help=("Filtrar por 1 product_id. Se omitido, puxa " "todos os produtos do produtor.")) parser.add_argument("--dry-run", action="store_true", help="Nao envia pro Meta; mostra tabela agrupada por produto.") parser.add_argument("--force", action="store_true", help="Reenvia mesmo se ja processado no dedupe local.") args = parser.parse_args() end_dt = (datetime.strptime(args.end, "%Y-%m-%d") if args.end else datetime.utcnow()) start_dt = (datetime.strptime(args.start, "%Y-%m-%d") if args.start else end_dt - timedelta(days=15)) product_filter = (args.product_id or "").strip() or None log.info("=== Backfill CAPI Borrello ===") log.info("Produto: %s", product_filter or "(todos)") log.info("Periodo: %s -> %s", start_dt.date(), end_dt.date()) log.info("Statuses: %s", ", ".join(STATUSES_TO_BACKFILL)) log.info("STUB_MODE: %s DRY_RUN: %s FORCE: %s", STUB_MODE, args.dry_run, args.force) if not CLIENT_ID or not CLIENT_SECRET: log.error("HOTMART_CLIENT_ID/SECRET nao configurados.") sys.exit(1) token = hotmart_token() stats = {"fetched": 0, "sent_ok": 0, "sent_fail": 0, "deduped": 0, "skipped": 0} # Estruturas de agrupamento por produto by_product: dict[str, dict] = defaultdict( lambda: {"count": 0, "value_total": 0.0, "name": "", "sent_ok": 0, "sent_fail": 0, "deduped": 0}) seen_tx: set[str] = set() # dedupe intra-run (APPROVED + COMPLETE do mesmo tx) for status in STATUSES_TO_BACKFILL: log.info(">>> Buscando status=%s", status) try: iterator = fetch_sales(token, start_dt, end_dt, status, product_filter) for item in iterator: stats["fetched"] += 1 data = item_to_data(item) # Descobre product_id / name / value / tx logo pra agrupar pid = str(_get(data, "product.id", "prod", default="") or "").strip() or "unknown" pname = str(_get(data, "product.name", "prod_name", default="") or "").strip() tx = str(_get(data, "purchase.transaction", "transaction", default="") or "").strip() try: v = float(_get(data, "purchase.price.value", "purchase.full_price.value", "purchase.offer.price.value", "price", default=0) or 0) except Exception: v = 0.0 # Se o mesmo tx apareceu em APPROVED e agora em COMPLETE, # nao conta duas vezes no agrupamento. first_time_this_tx = tx and tx not in seen_tx if first_time_this_tx: seen_tx.add(tx) grp = by_product[pid] grp["count"] += 1 grp["value_total"] += v if pname and not grp["name"]: grp["name"] = pname # Roda o pipeline padrao event = build_capi_event(status, data) if not event: stats["skipped"] += 1 continue event_id = event["event_id"] if not args.force and already_processed(event_id): log.info("Dedupe local: %s ja enviado (status=%s tx=%s).", event_id, status, tx) stats["deduped"] += 1 if first_time_this_tx: by_product[pid]["deduped"] += 1 continue if args.dry_run: log.info("[dry-run] event_id=%s status=%s product=%s " "value=%s email_hash=%s", event_id, status, pid, event["custom_data"]["value"], event["user_data"].get("em", ["--"])[0][:12]) continue status_code, resp_text = send_to_meta(event) mark_processed(event_id, event["event_name"], tx or event_id, status_code, resp_text, item) if status_code == 200: stats["sent_ok"] += 1 by_product[pid]["sent_ok"] += 1 else: stats["sent_fail"] += 1 by_product[pid]["sent_fail"] += 1 # gentileza com a Meta time.sleep(0.25) except requests.HTTPError as exc: log.error("Erro fatal em status=%s: %s", status, exc) log.info("=== RESUMO GERAL ===") for k, v in stats.items(): log.info(" %s: %s", k, v) log.info("=== POR PRODUTO ===") header = f" {'product_id':<14} {'count':>6} {'valor_total':>14} " \ f"{'sent_ok':>8} {'sent_fail':>10} {'deduped':>8} name" log.info(header) log.info(" " + "-" * (len(header) - 2)) # ordena por count desc for pid, grp in sorted(by_product.items(), key=lambda kv: kv[1]["count"], reverse=True): log.info(" %-14s %6d %14.2f %8d %10d %8d %s", pid, grp["count"], grp["value_total"], grp["sent_ok"], grp["sent_fail"], grp["deduped"], grp["name"] or "-") if __name__ == "__main__": if not HOTTOK: log.warning("HOTMART_HOTTOK vazio — ok pra backfill (nao usa), " "mas o webhook nao vai aceitar events em prod.") main()