#!/usr/bin/env python3
"""
backfill_purchases.py — Retroativo de Purchases PX3 Lab
=======================================================

Reprocessa compradores historicos:
  1. Aplica tag comprou-<produto> no contato Linkia (upsert por email)
  2. Envia Purchase pro CAPI PX3 (/purchase) - dedupe garantido por event_id

Fontes suportadas (--source):
  csv          -> Le CSV exportado da planilha VENDAS (Google Sheets)
                  Header esperado: PESSOA, CNPJ, RAZAO_SOCIAL, CEP, RUA, ...,
                  EMAIL_REP, TELEFONE_REP, DATA E HORA, SOFTWARE,
                  TOTAL DE CRÉDITO, TIPO DE CRÉDITO, VENDEDOR, PREÇO,
                  CIDADE_REP, ESTADO_REP, CEP_REP, NOME_COMPLETO_REP
                  (mesmas colunas do node "Append row in sheet1" do workflow VRA)
  n8n_replay   -> Puxa execucoes historicas do workflow VRA (id fixo) via
                  N8N API e reprocessa. Cobre so ate onde N8N retem
                  executions (tipicamente 30-90 dias).

Uso:
  # Dry-run (default: 90 dias atras ate hoje)
  python3 backfill_purchases.py --source csv --csv-path /tmp/vendas.csv --dry-run

  # Testar 5 registros com test-event-code do Meta
  python3 backfill_purchases.py --source csv --csv-path /tmp/vendas.csv \\
      --limit 5 --test-code TEST_BACKFILL_2026

  # Producao completa (batch de 10 com pause 500ms)
  python3 backfill_purchases.py --source csv --csv-path /tmp/vendas.csv \\
      --batch-size 10

  # So aplicar tag, sem enviar CAPI
  python3 backfill_purchases.py --source csv --csv-path /tmp/vendas.csv \\
      --only-tag

  # Reprocessar so os ultimos 30 dias via N8N replay
  python3 backfill_purchases.py --source n8n_replay --start 2026-07-25

Autor: Paulo (agencia climb) - 2026-08-24
"""

from __future__ import annotations

import argparse
import csv
import json
import logging
import os
import re
import sys
import time
import unicodedata
from datetime import datetime, timedelta, timezone
from logging.handlers import RotatingFileHandler
from pathlib import Path
from typing import Any

import requests

# ---------------------------------------------------------------------------
# Config
# ---------------------------------------------------------------------------
BASE_DIR = Path(__file__).resolve().parent
PROJECT_DIR = BASE_DIR.parent  # /opt/mia/workspace/clientes/px3lab

CAPI_ENDPOINT = "https://track-capi-px3.agentesclimb.us/purchase"
LINKIA_ENDPOINT = "https://services.leadconnectorhq.com/contacts/upsert"

# Carrega PX3_TOKEN e PX3_LOCATION_ID do config_ghl.py
sys.path.insert(0, str(PROJECT_DIR))
try:
    from config_ghl import PX3_TOKEN, PX3_LOCATION_ID  # type: ignore
except ImportError as e:
    print(f"[FATAL] Nao consegui importar config_ghl.py: {e}", file=sys.stderr)
    sys.exit(1)

# N8N (opcional, so pra --source n8n_replay)
N8N_ENV = Path("/opt/mia/config/n8n_px3.env")
N8N_URL = ""
N8N_TOKEN = ""
if N8N_ENV.exists():
    for raw in N8N_ENV.read_text(encoding="utf-8").splitlines():
        line = raw.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        k, v = line.split("=", 1)
        if k.strip() == "N8N_PX3_URL":
            N8N_URL = v.strip()
        elif k.strip() == "N8N_PX3_TOKEN":
            N8N_TOKEN = v.strip()

VRA_WORKFLOW_ID = "QcJuvqhV98vIYtA6"

# Mapeamento SOFTWARE -> tag slug Linkia
# NOTA: variantes conhecidas no CSV VRA:
#   - "FOTODRIVE" e "FOTO DRIVE" aparecem como sinonimos de "PHOTO DRIVE"
PRODUCT_TAG_MAP = {
    "CLOUD PHOTORF":    "comprou-claud-photorf",
    "FASTALBUM":        "comprou-fast-album",
    "PHOTO DRIVE":      "comprou-photo-drive",
    "FOTO DRIVE":       "comprou-photo-drive",
    "FOTODRIVE":        "comprou-photo-drive",
    "E-PHOTORF":        "comprou-e-photorf",
    "CHECKOUT PHOTORF": "comprou-checkout-photorf",
}

TRIAL_MARKER = "PERÍODO DE TESTE"  # valor do campo TIPO DE CRÉDITO
REPOSICAO_MARKER = "REPOSIÇÃO"  # crédito cortesia / substituição - nao e venda nova
REGULAR_MARKER = "REGULAR"      # venda real -> processa
BONUS_MARKER = "BÕNUS"          # venda com bonus - processa como Purchase normal


# ---------------------------------------------------------------------------
# Tabela de preco por faixa (aplicada quando VALOR nao vem preenchido no CSV)
# Aprovada pelo Renato em 2026-08-18.
# ---------------------------------------------------------------------------
def preco_por_faixa(creditos: int) -> float:
    """Retorna o preco unitario (R$/credito) baseado no volume de creditos."""
    if creditos <= 0:
        return 0.0
    if creditos <= 10_000:
        return 0.090
    if creditos <= 50_000:
        return 0.035
    if creditos <= 200_000:
        return 0.027
    return 0.0268

# ---------------------------------------------------------------------------
# Logging
# ---------------------------------------------------------------------------
LOG_FILE = Path("/opt/mia/logs/backfill_purchases_px3.log")
LOG_FILE.parent.mkdir(parents=True, exist_ok=True)

_fmt = logging.Formatter("%(asctime)s [%(levelname)s] %(message)s")
_fh = RotatingFileHandler(LOG_FILE, maxBytes=10 * 1024 * 1024, backupCount=5, encoding="utf-8")
_fh.setFormatter(_fmt)
_sh = logging.StreamHandler(sys.stdout)
_sh.setFormatter(_fmt)
logging.basicConfig(level=logging.INFO, handlers=[_fh, _sh])
log = logging.getLogger("backfill_px3")


# ---------------------------------------------------------------------------
# Helpers de normalizacao
# ---------------------------------------------------------------------------
def parse_price_br(raw: str) -> float:
    """Converte '0,065' ou '0.065' ou 'R$ 0,065' em float."""
    if raw is None:
        return 0.0
    s = str(raw).strip()
    if not s:
        return 0.0
    s = re.sub(r"[^\d,\.\-]", "", s)
    # Se tem virgula E ponto, ponto e milhar
    if "," in s and "." in s:
        s = s.replace(".", "").replace(",", ".")
    elif "," in s:
        s = s.replace(",", ".")
    try:
        return float(s)
    except ValueError:
        return 0.0


def parse_int(raw: str) -> int:
    if raw is None:
        return 0
    s = re.sub(r"[^\d\-]", "", str(raw))
    try:
        return int(s) if s else 0
    except ValueError:
        return 0


def parse_date_br(raw: str) -> datetime | None:
    """Aceita 'dd/mm/yyyy HH:MM:SS' ou 'dd/mm/yyyy' ou ISO."""
    if not raw:
        return None
    raw = str(raw).strip()
    for fmt in (
        "%d/%m/%Y %H:%M:%S",
        "%d/%m/%Y %H:%M",
        "%d/%m/%Y",
        "%Y-%m-%dT%H:%M:%S",
        "%Y-%m-%dT%H:%M:%SZ",
        "%Y-%m-%d %H:%M:%S",
        "%Y-%m-%d",
    ):
        try:
            return datetime.strptime(raw[:19], fmt)
        except ValueError:
            continue
    return None


def split_name(nome: str) -> tuple[str, str]:
    parts = (nome or "").strip().split()
    if not parts:
        return "", ""
    if len(parts) == 1:
        return parts[0], ""
    return parts[0], " ".join(parts[1:])


def normalize_phone_br(raw: str) -> str:
    """Retorna telefone so com digitos, com prefixo 55 se BR."""
    d = re.sub(r"\D", "", str(raw or ""))
    if not d:
        return ""
    if len(d) == 11:  # DDD + celular sem 55
        return "55" + d
    if len(d) == 10:  # fixo BR sem 55
        return "55" + d
    return d


def mask(v: str, keep: int = 4) -> str:
    if not v:
        return ""
    s = str(v)
    if len(s) <= keep + 2:
        return "***" + s[-2:]
    return s[:2] + "***" + s[-keep:]


# ---------------------------------------------------------------------------
# Modelo comum: Sale
# ---------------------------------------------------------------------------
class Sale:
    __slots__ = (
        "row_number", "date", "software", "tipo_credito", "email", "phone",
        "first_name", "last_name", "city", "state", "zip_code",
        "unit_price", "num_credits", "value", "razao_social",
        "vendedor", "value_from_csv", "raw_source", "parse_errors",
        "price_source", "applied_unit_price",
    )

    def __init__(
        self,
        row_number: int,
        date: datetime | None,
        software: str,
        tipo_credito: str,
        email: str,
        phone: str,
        first_name: str,
        last_name: str,
        city: str,
        state: str,
        zip_code: str,
        unit_price: float,
        num_credits: int,
        razao_social: str = "",
        vendedor: str = "",
        value_from_csv: float = 0.0,
        raw_source: str = "",
        parse_errors: list[str] | None = None,
    ) -> None:
        self.row_number = row_number
        self.date = date
        self.software = software.strip().upper() if software else ""
        self.tipo_credito = (tipo_credito or "").strip().upper()
        self.email = (email or "").strip()
        self.phone = normalize_phone_br(phone)
        self.first_name = first_name
        self.last_name = last_name
        self.city = (city or "").strip()
        self.state = (state or "").strip()
        self.zip_code = (zip_code or "").strip()
        self.unit_price = unit_price
        self.num_credits = num_credits
        self.vendedor = (vendedor or "").strip()
        self.value_from_csv = value_from_csv
        # REGRA APROVADA 2026-08-18:
        # 1) Se VALOR do CSV veio > 0 -> usa direto (nao recalcula).
        # 2) Se VALOR vazio ou 0 -> aplica preco por faixa (preco_por_faixa) sobre TOTAL_CREDITO.
        # 3) unit_price da coluna PREÇO fica so como referencia informativa (nao entra mais no calculo).
        if value_from_csv and value_from_csv > 0:
            self.value = round(value_from_csv, 2)
            self.price_source = "csv_valor"
            self.applied_unit_price = (value_from_csv / num_credits) if num_credits > 0 else 0.0
        else:
            faixa = preco_por_faixa(num_credits)
            self.value = round(faixa * num_credits, 2)
            self.price_source = "faixa_auto"
            self.applied_unit_price = faixa
        self.razao_social = razao_social
        self.raw_source = raw_source
        self.parse_errors = parse_errors or []

    @property
    def is_trial(self) -> bool:
        return "TESTE" in self.tipo_credito or self.tipo_credito == TRIAL_MARKER.upper()

    @property
    def is_reposicao(self) -> bool:
        # normalizado sem acento pra ficar robusto
        norm = _strip_accents(self.tipo_credito).upper()
        return "REPOSICAO" in norm

    @property
    def is_regular(self) -> bool:
        return self.tipo_credito == REGULAR_MARKER

    @property
    def is_bonus(self) -> bool:
        # Normalizado sem acento pra pegar "BONUS", "BÔNUS", "BÕNUS" etc.
        norm = _strip_accents(self.tipo_credito).upper()
        return norm == "BONUS"

    @property
    def is_sale(self) -> bool:
        """Novo (aprovado 2026-08-18): REGULAR ou BONUS conta como venda."""
        return self.is_regular or self.is_bonus

    @property
    def has_email(self) -> bool:
        return bool(self.email and "@" in self.email)

    @property
    def is_valid(self) -> bool:
        """Compat: valid = email OK e software conhecido (usado pelo fluxo antigo)."""
        return self.has_email and self.software in PRODUCT_TAG_MAP

    @property
    def is_processable(self) -> bool:
        """Regra atualizada 2026-08-18: (REGULAR ou BONUS) + email OK + software conhecido."""
        return self.is_sale and self.has_email and self.software in PRODUCT_TAG_MAP

    @property
    def skip_reason(self) -> str:
        if not self.has_email:
            return "sem_email"
        if self.is_trial:
            return "trial"
        if self.is_reposicao:
            return "reposicao"
        if not self.is_sale:
            return f"tipo_credito_desconhecido:{self.tipo_credito or '(vazio)'}"
        if self.software not in PRODUCT_TAG_MAP:
            return f"software_desconhecido:{self.software or '(vazio)'}"
        return ""

    @property
    def tag_slug(self) -> str:
        return PRODUCT_TAG_MAP.get(self.software, "")

    def event_id(self, prefix: str = "backfill-vra") -> str:
        d = self.date.strftime("%Y%m%d") if self.date else "nodate"
        return f"{prefix}-r{self.row_number}-{d}"


def _strip_accents(s: str) -> str:
    if not s:
        return ""
    return "".join(
        c for c in unicodedata.normalize("NFD", s)
        if unicodedata.category(c) != "Mn"
    )


# ---------------------------------------------------------------------------
# Fonte: CSV
# ---------------------------------------------------------------------------
CSV_COLUMN_MAP = {
    # canonico       -> possiveis nomes no header
    "date":          ["DATA E HORA", "DATA_E_HORA", "DATA", "data"],
    "software":      ["SOFTWARE", "software"],
    "tipo_credito":  ["TIPO DE CRÉDITO", "TIPO DE CREDITO", "TIPO_DE_CREDITO", "TIPO_DE_DE_CREDITO"],
    "email":         ["EMAIL_REP", "E-MAIL", "EMAIL", "email"],
    "phone":         ["TELEFONE_REP", "TELEFONE", "phone"],
    "nome":          ["NOME_COMPLETO_REP", "NOME_COMPLETO", "NOME"],
    "city":          ["CIDADE_REP", "CIDADE"],
    "state":         ["ESTADO_REP", "ESTADO"],
    "zip":           ["CEP_REP", "CEP"],
    "price":         ["PREÇO", "PRECO", "PREÇO_UNITARIO", "PRICE"],
    "credits":       ["TOTAL DE CRÉDITO", "TOTAL_DE_CREDITO", "TOTAL DE CREDITO"],
    "razao_social":  ["RAZAO_SOCIAL", "RAZÃO SOCIAL", "EMPRESA"],
    "vendedor":      ["VENDEDOR", "vendedor"],
    "valor":         ["VALOR", "valor", "VALOR_TOTAL"],
}


def _pick(row: dict, canonical: str) -> str:
    for name in CSV_COLUMN_MAP[canonical]:
        if name in row and row[name] not in (None, ""):
            return str(row[name])
    return ""


def load_from_csv(path: str) -> list[Sale]:
    p = Path(path)
    if not p.exists():
        log.error("CSV nao encontrado: %s", path)
        return []

    sales: list[Sale] = []
    with p.open("r", encoding="utf-8-sig", newline="") as f:
        reader = csv.DictReader(f)
        for i, row in enumerate(reader, start=2):  # linha 1 = header
            errors: list[str] = []
            raw_date = _pick(row, "date")
            date = parse_date_br(raw_date)
            if raw_date and date is None:
                errors.append(f"data_invalida:{raw_date!r}")

            software = _pick(row, "software")
            tipo = _pick(row, "tipo_credito")
            email = _pick(row, "email")
            phone = _pick(row, "phone")
            nome = _pick(row, "nome")
            first, last = split_name(nome)
            city = _pick(row, "city")
            state = _pick(row, "state")
            zip_c = _pick(row, "zip")
            unit_price = parse_price_br(_pick(row, "price"))
            num = parse_int(_pick(row, "credits"))
            razao = _pick(row, "razao_social")
            vendedor = _pick(row, "vendedor")
            valor_csv = parse_price_br(_pick(row, "valor"))

            sales.append(Sale(
                row_number=i,
                date=date,
                software=software,
                tipo_credito=tipo,
                email=email,
                phone=phone,
                first_name=first,
                last_name=last,
                city=city,
                state=state,
                zip_code=zip_c,
                unit_price=unit_price,
                num_credits=num,
                razao_social=razao,
                vendedor=vendedor,
                value_from_csv=valor_csv,
                raw_source="csv",
                parse_errors=errors,
            ))
    return sales


# ---------------------------------------------------------------------------
# Fonte: N8N replay
# ---------------------------------------------------------------------------
def load_from_n8n(start: datetime, end: datetime) -> list[Sale]:
    """Puxa executions do workflow VRA e extrai body do webhook."""
    if not N8N_URL or not N8N_TOKEN:
        log.error("N8N_PX3_URL/N8N_PX3_TOKEN nao configurado em /opt/mia/config/n8n_px3.env")
        return []

    sales: list[Sale] = []
    seen_ids: set[str] = set()
    cursor: str | None = None
    while True:
        params = {
            "workflowId": VRA_WORKFLOW_ID,
            "limit": 250,
            "status": "success",
            "includeData": "true",
        }
        if cursor:
            params["cursor"] = cursor
        r = requests.get(
            f"{N8N_URL}/api/v1/executions",
            headers={"X-N8N-API-KEY": N8N_TOKEN},
            params=params,
            timeout=30,
        )
        if r.status_code != 200:
            log.error("N8N executions error %s: %s", r.status_code, r.text[:200])
            break
        data = r.json()
        for execu in data.get("data", []):
            eid = execu.get("id")
            if eid in seen_ids:
                continue
            seen_ids.add(eid)
            started = execu.get("startedAt", "")
            try:
                dt = datetime.strptime(started[:19], "%Y-%m-%dT%H:%M:%S")
            except ValueError:
                dt = None
            if dt and (dt < start or dt > end):
                continue

            # extrair body do webhook PLANILHA DE VENDAS
            try:
                run_data = execu["data"]["resultData"]["runData"]
                pv_node = run_data.get("PLANILHA DE VENDAS", [])
                if not pv_node:
                    continue
                body = pv_node[0]["data"]["main"][0][0]["json"]["body"]
            except (KeyError, IndexError, TypeError):
                continue

            software = (body.get("DADOS_VENDA", {}) or {}).get("SOFTWARE", "")
            tipo = (body.get("DADOS_VENDA", {}) or {}).get("TIPO_DE_DE_CREDITO", "")
            price = (body.get("DADOS_VENDA", {}) or {}).get("PREÇO", "")
            credits = (body.get("DADOS_VENDA", {}) or {}).get("TOTAL_DE_CREDITO", "")
            rep = body.get("REPRESENTANTE", {}) or {}
            nome = rep.get("NOME_COMPLETO", "")
            first, last = split_name(nome)
            email = rep.get("E-MAIL", "")
            phone = rep.get("TELEFONE", "")
            city = rep.get("CIDADE", "")
            state = rep.get("ESTADO", "")
            zip_c = rep.get("CEP", "")
            razao = (body.get("DADOS_DA_EMPRESA", {}) or {}).get("RAZAO_SOCIAL", "")

            sales.append(Sale(
                row_number=int(eid),
                date=dt,
                software=software,
                tipo_credito=tipo,
                email=email,
                phone=phone,
                first_name=first,
                last_name=last,
                city=city,
                state=state,
                zip_code=zip_c,
                unit_price=parse_price_br(price),
                num_credits=parse_int(credits),
                razao_social=razao,
                raw_source=f"n8n_exec_{eid}",
            ))

        cursor = data.get("nextCursor")
        if not cursor:
            break
    return sales


# ---------------------------------------------------------------------------
# Acoes: Linkia tag / CAPI purchase
# ---------------------------------------------------------------------------
def apply_linkia_tag(sale: Sale, dry_run: bool = False) -> tuple[bool, str]:
    if dry_run:
        return True, "DRY"
    body = {
        "locationId": PX3_LOCATION_ID,
        "email": sale.email,
        "phone": ("+" + sale.phone) if sale.phone else "",
        "firstName": sale.first_name,
        "lastName": sale.last_name,
        "tags": [sale.tag_slug],
    }
    if sale.razao_social:
        body["companyName"] = sale.razao_social
    try:
        r = requests.post(
            LINKIA_ENDPOINT,
            headers={
                "Authorization": f"Bearer {PX3_TOKEN}",
                "Version": "2021-07-28",
                "Content-Type": "application/json",
            },
            json=body,
            timeout=15,
        )
        # 200/201 = ok, 400 com contactId no erro = existente (bug conhecido do GHL)
        if r.status_code in (200, 201):
            return True, f"OK {r.status_code}"
        return False, f"ERR {r.status_code} {r.text[:200]}"
    except requests.RequestException as e:
        return False, f"REQ_ERR {e}"


def send_capi_purchase(sale: Sale, test_code: str = "", dry_run: bool = False) -> tuple[bool, str]:
    if dry_run:
        return True, "DRY"

    body = {
        "email": sale.email,
        "phone": sale.phone,
        "first_name": sale.first_name,
        "last_name": sale.last_name,
        "city": sale.city,
        "state": sale.state,
        "zip": sale.zip_code,
        "country": "BR",
        "value": sale.value,
        "currency": "BRL",
        "content_name": sale.software,
        "content_ids": [sale.software],
        "num_items": sale.num_credits,
        "event_id": sale.event_id(),
    }
    # Override do test_event_code por request (suporte adicionado no app.py em 2026-08-24).
    # Se o caller passou --test-code, envia junto no body. Sem afetar o env do CAPI (N8N segue com TEST_VRA_2026).
    if test_code:
        body["test_event_code"] = test_code
    try:
        r = requests.post(CAPI_ENDPOINT, json=body, timeout=15)
        if r.status_code == 200:
            j = r.json()
            if j.get("dedupe"):
                return True, f"DEDUPE {sale.event_id()}"
            # O CAPI devolve 'resp' como string com o JSON bruto do Meta.
            # Ex: {"events_received":1,"messages":[],"fbtrace_id":"..."}
            events_received = "?"
            fbtrace = "-"
            resp_raw = j.get("resp", "")
            if resp_raw:
                try:
                    meta_j = json.loads(resp_raw)
                    events_received = meta_j.get("events_received", "?")
                    fbtrace = meta_j.get("fbtrace_id", "-")
                except (json.JSONDecodeError, TypeError):
                    pass
            return True, (
                f"OK meta={j.get('meta_status')} events_received={events_received} "
                f"fbtrace={fbtrace} event_id={sale.event_id()}"
            )
        return False, f"ERR {r.status_code} {r.text[:200]}"
    except requests.RequestException as e:
        return False, f"REQ_ERR {e}"


# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------
def summarize(sales: list[Sale]) -> dict:
    """Retorna resumo simples (compat com fluxo antigo, usado no --dry-run tambem)."""
    total = len(sales)
    valid = [s for s in sales if s.is_valid]
    non_trial = [s for s in valid if not s.is_trial]
    by_product: dict[str, int] = {}
    total_value = 0.0
    for s in non_trial:
        by_product[s.software] = by_product.get(s.software, 0) + 1
        total_value += s.value
    return {
        "total_rows": total,
        "valid": len(valid),
        "invalid": total - len(valid),
        "trials_skipped": len(valid) - len(non_trial),
        "purchases_to_process": len(non_trial),
        "total_value_brl": round(total_value, 2),
        "by_product": by_product,
    }


def _fmt_brl(v: float) -> str:
    # 1.234.567,89
    s = f"{v:,.2f}"
    return s.replace(",", "X").replace(".", ",").replace("X", ".")


def dryrun_report(sales: list[Sale], source_label: str) -> None:
    """Relatorio completo pro dry-run (formato pedido pela Mia)."""
    total = len(sales)

    # buckets por TIPO DE CRÉDITO
    by_tipo: dict[str, int] = {}
    for s in sales:
        key = s.tipo_credito or "(vazio)"
        by_tipo[key] = by_tipo.get(key, 0) + 1

    regular = [s for s in sales if s.is_regular]
    bonus = [s for s in sales if s.is_bonus]
    trials = [s for s in sales if s.is_trial]
    reposicao = [s for s in sales if s.is_reposicao]
    outros = [s for s in sales
              if not s.is_regular and not s.is_bonus and not s.is_trial and not s.is_reposicao]

    # processavel: (REGULAR ou BONUS) + email OK + software conhecido
    processable = [s for s in sales if s.is_processable]

    # skippados = qualquer coisa que nao seja processavel
    skipped = [s for s in sales if not s.is_processable]

    # motivos de skip
    skip_reasons: dict[str, int] = {}
    for s in skipped:
        r = s.skip_reason or "outro"
        skip_reasons[r] = skip_reasons.get(r, 0) + 1

    # separar processaveis:
    #  - com VALOR do CSV (usa VALOR direto)
    #  - sem VALOR (aplica faixa automatica em TOTAL_CREDITO)
    proc_com_valor = [s for s in processable if s.price_source == "csv_valor"]
    proc_sem_valor = [s for s in processable if s.price_source == "faixa_auto"]

    # distribuicao SOFTWARE (em processaveis)
    by_soft_count: dict[str, int] = {}
    by_soft_value: dict[str, float] = {}
    for s in processable:
        by_soft_count[s.software] = by_soft_count.get(s.software, 0) + 1
        by_soft_value[s.software] = by_soft_value.get(s.software, 0.0) + s.value

    # distribuicao SOFTWARE de TUDO (pra ver o mix bruto tambem)
    by_soft_all: dict[str, int] = {}
    for s in sales:
        key = s.software or "(vazio)"
        by_soft_all[key] = by_soft_all.get(key, 0) + 1

    # vendedores (em processaveis)
    by_vend: dict[str, int] = {}
    by_vend_value: dict[str, float] = {}
    for s in processable:
        vk = s.vendedor or "(sem vendedor)"
        by_vend[vk] = by_vend.get(vk, 0) + 1
        by_vend_value[vk] = by_vend_value.get(vk, 0.0) + s.value

    top_vend = sorted(by_vend.items(), key=lambda x: x[1], reverse=True)

    # datas (em processaveis)
    dates = [s.date for s in processable if s.date]
    dt_min = min(dates) if dates else None
    dt_max = max(dates) if dates else None

    total_value = sum(s.value for s in processable)

    # parse errors (linha X, motivo Y)
    parse_broken = [s for s in sales if s.parse_errors]

    # -----------------------------------------------------------------------
    print("\n" + "=" * 70)
    print(f"DRY-RUN BACKFILL PX3 - source: {source_label}")
    print(f"Total de linhas: {total}")
    print("=" * 70)

    print("\n[DISTRIBUICAO POR TIPO DE CREDITO]")
    for k, v in sorted(by_tipo.items(), key=lambda x: x[1], reverse=True):
        print(f"  - {k:30} {v:>4}")
    print(f"  Resumo: REGULAR={len(regular)}  BONUS={len(bonus)}  TESTE={len(trials)}  "
          f"REPOSICAO={len(reposicao)}  outros/vazio={len(outros)}")

    print("\n[DISTRIBUICAO POR SOFTWARE - todas as linhas]")
    for k, v in sorted(by_soft_all.items(), key=lambda x: x[1], reverse=True):
        print(f"  - {k:25} {v:>4}")

    print("\n[PROCESSAVEIS] ((REGULAR ou BONUS) + email OK + software conhecido)")
    print(f"  Total: {len(processable)}  (REGULAR={sum(1 for s in processable if s.is_regular)} "
          f"BONUS={sum(1 for s in processable if s.is_bonus)})")
    print(f"    - com VALOR preenchido no CSV (usa direto):      {len(proc_com_valor)}")
    print(f"    - sem VALOR (aplica faixa automatica em creditos): {len(proc_sem_valor)}")
    valor_confiavel = sum(s.value for s in proc_com_valor)
    valor_calc = sum(s.value for s in proc_sem_valor)
    print(f"  Valor total (soma tudo):        R$ {_fmt_brl(total_value)}")
    print(f"    subtotal VALOR CSV direto:    R$ {_fmt_brl(valor_confiavel)}")
    print(f"    subtotal faixa auto (calc):   R$ {_fmt_brl(valor_calc)}")
    if dt_min and dt_max:
        print(f"  Periodo: {dt_min.strftime('%d/%m/%Y')} -> {dt_max.strftime('%d/%m/%Y')}")
    else:
        print("  Periodo: (sem datas validas)")

    print("\n[TABELA DE FAIXA APLICADA - regra aprovada 2026-08-18]")
    print("  ate 10.000 creditos          -> R$ 0,0900 / credito")
    print("  10.001 a 50.000 creditos     -> R$ 0,0350 / credito")
    print("  50.001 a 200.000 creditos    -> R$ 0,0270 / credito")
    print("  acima de 200.000 creditos    -> R$ 0,0268 / credito")

    if proc_sem_valor:
        print(f"\n[AMOSTRA FAIXA AUTO] {len(proc_sem_valor)} linhas sem VALOR - mostrando ate 10:")
        print(f"  {'row':>4}  {'tipo':8}  {'soft':16}  {'credits':>8}  {'faixa':>10}  {'valor_final':>16}")
        for s in proc_sem_valor[:10]:
            print(f"  {s.row_number:>4}  {s.tipo_credito[:8]:8}  {s.software[:16]:16}  "
                  f"{s.num_credits:>8}  R$ {s.applied_unit_price:>7.4f}  "
                  f"R$ {_fmt_brl(s.value):>13}")
        if len(proc_sem_valor) > 10:
            print(f"  ... +{len(proc_sem_valor) - 10} linhas")

    print("\n[POR SOFTWARE - processaveis]")
    for k in sorted(by_soft_count.keys()):
        c = by_soft_count[k]
        v = by_soft_value[k]
        print(f"  - {k:25} {c:>4}   R$ {_fmt_brl(v)}")

    print("\n[TOP VENDEDORES - processaveis]")
    for k, c in top_vend[:5]:
        v = by_vend_value[k]
        print(f"  - {k:25} {c:>4} vendas   R$ {_fmt_brl(v)}")

    print("\n[SKIPPADOS]")
    print(f"  Total: {len(skipped)}")
    for r, c in sorted(skip_reasons.items(), key=lambda x: x[1], reverse=True):
        print(f"  - {r:40} {c:>4}")

    # amostras
    print("\n[AMOSTRA - 5 PROCESSAVEIS mascaradas + calculo do valor]")
    for s in processable[:5]:
        nome = (s.first_name + " " + s.last_name).strip()
        src_tag = "CSV_VALOR" if s.price_source == "csv_valor" else "FAIXA_AUTO"
        print(
            f"  row={s.row_number:>3} "
            f"date={s.date.strftime('%d/%m/%Y') if s.date else 'None':10} "
            f"soft={s.software:15} "
            f"tipo={s.tipo_credito:8} "
            f"vend={(s.vendedor or '-')[:15]:15} "
            f"email={mask(s.email, 6):22} "
            f"credits={s.num_credits:>7} "
            f"preco_aplic=R${s.applied_unit_price:.4f} "
            f"src={src_tag:10} "
            f"val=R$ {_fmt_brl(s.value):>12}"
        )

    print("\n[AMOSTRA - 3 SKIPPADOS mascarados]")
    shown = 0
    for s in skipped:
        nome = (s.first_name + " " + s.last_name).strip()
        print(
            f"  row={s.row_number:>3} "
            f"reason={s.skip_reason:30} "
            f"soft={(s.software or '-')[:15]:15} "
            f"tipo={(s.tipo_credito or '-')[:10]:10} "
            f"email={mask(s.email, 6) if s.email else '(vazio)':20} "
            f"nome={mask(nome, 3) if nome else '(vazio)':18}"
        )
        shown += 1
        if shown >= 3:
            break

    if parse_broken:
        print(f"\n[LINHAS COM ERRO DE PARSE] {len(parse_broken)} linhas")
        for s in parse_broken[:10]:
            print(f"  row={s.row_number} errors={s.parse_errors}")
    else:
        print("\n[LINHAS COM ERRO DE PARSE] nenhuma")

    print("=" * 70)


def show_sample(sales: list[Sale], n: int = 5) -> None:
    print("\n--- Amostra (mascarada) ---")
    shown = 0
    for s in sales:
        if not s.is_valid or s.is_trial:
            continue
        print(
            f"  row={s.row_number} date={s.date.strftime('%Y-%m-%d') if s.date else 'None'} "
            f"soft={s.software:16} tipo={s.tipo_credito:15} "
            f"email={mask(s.email, 6):20} phone={mask(s.phone, 4):15} "
            f"nome={mask(s.first_name + ' ' + s.last_name, 3):20} "
            f"val=R${s.value:>10.2f} ({s.num_credits}x{s.unit_price}) "
            f"city={s.city[:15]:15}/{s.state[:2]:2}"
        )
        shown += 1
        if shown >= n:
            break
    print()


def main() -> int:
    parser = argparse.ArgumentParser(description="Backfill Purchases PX3")
    parser.add_argument("--source", choices=["csv", "n8n_replay"], required=True)
    parser.add_argument("--csv-path", default="/tmp/vendas.csv",
                        help="Caminho do CSV (so pra --source csv)")
    parser.add_argument("--start", default=None,
                        help="Data inicial YYYY-MM-DD (default: 90d atras)")
    parser.add_argument("--end", default=None,
                        help="Data final YYYY-MM-DD (default: hoje)")
    parser.add_argument("--dry-run", action="store_true")
    parser.add_argument("--limit", type=int, default=0,
                        help="Maximo de eventos (0 = sem limite)")
    parser.add_argument("--test-code", default="",
                        help="test_event_code do Meta - informativo (o CAPI aplica pelo env)")
    parser.add_argument("--batch-size", type=int, default=10,
                        help="Pausa 500ms a cada N")
    parser.add_argument("--only-tag", action="store_true",
                        help="So aplica tag Linkia (nao envia CAPI)")
    parser.add_argument("--only-capi", action="store_true",
                        help="So envia CAPI (nao aplica tag)")
    args = parser.parse_args()

    if args.only_tag and args.only_capi:
        log.error("--only-tag e --only-capi sao mutuamente exclusivos")
        return 2

    # Datas
    now = datetime.now()
    if args.start:
        start = datetime.strptime(args.start, "%Y-%m-%d")
    else:
        # dry-run: janela larga (10 anos) pra ver tudo. producao: 90d.
        start = (now - timedelta(days=3650)) if args.dry_run else (now - timedelta(days=90))
    if args.end:
        end = datetime.strptime(args.end, "%Y-%m-%d").replace(hour=23, minute=59, second=59)
    else:
        end = now + timedelta(days=1)  # inclui hoje inteiro

    log.info(
        "backfill start | source=%s start=%s end=%s dry_run=%s limit=%s test_code=%s batch=%s only_tag=%s only_capi=%s",
        args.source, start.date(), end.date(), args.dry_run, args.limit,
        args.test_code or "(none)", args.batch_size, args.only_tag, args.only_capi,
    )

    # Carrega
    if args.source == "csv":
        all_sales = load_from_csv(args.csv_path)
        # filtra por data
        sales = [s for s in all_sales if (s.date is None or start <= s.date <= end)]
    else:
        sales = load_from_n8n(start, end)

    if args.dry_run:
        label = f"{args.source} janela={start.date()}..{end.date()}"
        if args.source == "csv":
            label += f" file={args.csv_path}"
        dryrun_report(sales, label)
        log.info("DRY-RUN concluido. Nada foi enviado.")
        return 0

    stats = summarize(sales)
    print("\n" + "=" * 60)
    print(f"SOURCE: {args.source}")
    print(f"JANELA: {start.date()} .. {end.date()}")
    print(f"Total linhas carregadas: {stats['total_rows']}")
    print(f"  invalidas (sem email/software desconhecido): {stats['invalid']}")
    print(f"  trials pulados: {stats['trials_skipped']}")
    print(f"  compras a processar: {stats['purchases_to_process']}")
    print(f"  valor total processavel: R$ {stats['total_value_brl']:,.2f}")
    print(f"  distribuicao por produto:")
    for k, v in sorted(stats["by_product"].items()):
        print(f"    - {k}: {v}")
    print("=" * 60)
    show_sample(sales)

    if args.test_code:
        # avisa que o test_code deve estar no env do servico CAPI
        curr = os.environ.get("META_TEST_EVENT_CODE", "")
        log.warning(
            "test_code=%s informado. Confirme que META_TEST_EVENT_CODE do CAPI esta ativo. "
            "Env atual visto no processo do backfill: %r (informativo, o CAPI usa o seu proprio env).",
            args.test_code, curr,
        )

    # Executa (so REGULAR + email OK + software conhecido)
    ok_tag = err_tag = ok_capi = err_capi = skipped = 0
    processed = 0
    for s in sales:
        if not s.is_processable:
            skipped += 1
            continue
        if args.limit and processed >= args.limit:
            break

        # Tag Linkia
        if not args.only_capi:
            ok, msg = apply_linkia_tag(s, dry_run=False)
            if ok:
                ok_tag += 1
            else:
                err_tag += 1
            log.info("TAG row=%d email=%s soft=%s slug=%s -> %s",
                     s.row_number, mask(s.email, 6), s.software, s.tag_slug, msg)

        # CAPI purchase
        if not args.only_tag:
            ok, msg = send_capi_purchase(s, test_code=args.test_code, dry_run=False)
            if ok:
                ok_capi += 1
            else:
                err_capi += 1
            log.info("CAPI row=%d email=%s soft=%s value=%.2f -> %s",
                     s.row_number, mask(s.email, 6), s.software, s.value, msg)

        processed += 1
        if processed % args.batch_size == 0:
            time.sleep(0.5)

    print("\n" + "=" * 60)
    print("RESULTADO")
    print(f"  processados: {processed}")
    print(f"  pulados (trial/invalido): {skipped}")
    print(f"  tag ok/err: {ok_tag}/{err_tag}")
    print(f"  capi ok/err: {ok_capi}/{err_capi}")
    print("=" * 60)
    log.info("backfill DONE processed=%d ok_tag=%d err_tag=%d ok_capi=%d err_capi=%d skipped=%d",
             processed, ok_tag, err_tag, ok_capi, err_capi, skipped)
    return 0


if __name__ == "__main__":
    sys.exit(main())