#!/usr/bin/env python3 """ Mapeamento de leads PX3 LAB para lançamento Cloud PhotoRF 2.0 (07/10/2026) Buckets: quente / morno / frio / gelado / descarte """ import json import time import csv import sys import os import requests from datetime import datetime, timezone, timedelta from pathlib import Path from collections import defaultdict # ────────────────────────────────────────────── # CONFIG # ────────────────────────────────────────────── TOKEN = "pit-cb07507e-3712-4f3e-8c24-fafc8a8430ba" LOCATION = "W7PGxpfbsFaEEUoQOtUb" BASE_URL = "https://services.leadconnectorhq.com" HEADERS = { "Authorization": f"Bearer {TOKEN}", "Version": "2021-07-28", "Content-Type": "application/json", "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/126.0 Safari/537.36", } NOW = datetime.now(timezone.utc) OUT_DIR = Path("/opt/mia/workspace/clientes/px3lab/mapeamento_leads_lancamento_1007") # ────────────────────────────────────────────── # CUSTOM FIELD IDs relevantes # ────────────────────────────────────────────── UTM_SOURCE_ID = "ysHWPzyR0uZ2nd8D9sbi" UTM_MEDIUM_ID = "Hiabk9sp0zIkgLjUYJjR" UTM_CAMPAIGN_ID = "H16g1x7tRA8OoHnqiTdp" UTM_CONTENT_ID = "ppQzEkQdPyQPMlUdxCdp" UTM_TERM_ID = "bNblpNN7VTrRJ6CgBytE" SDR_TEMPERATURA_ID = "lKiTBLRCMYLT2XmzcbFg" SDR_RESULTADO_ID = "8ekpHqngQqNFZT6JphdK" SDR_PRODUTO_INT_ID = "mqOExdhEAtRPd5rOl664" SDR_CLASSIF_ID = "jEHY43NsCA2OzGaPP4GQ" SDR_STATUS_ATEND_ID = "1dWMDmqGAbU5hyCv1kz0" # Pipeline IDs PIPELINE_SDR = "iV13JEVbMgfcad9Xw6gz" PIPELINE_PADRAO = "4AEDs93pm2FLfDSbSZYM" PIPELINE_PROSP = "Pj7dm7v7AJqUPiG7UESq" PIPELINE_ODOO = "ISpWA1XXfYHCpModwGyJ" PIPELINE_CONAFOR = "6YjM6IvNbe5IWyD9JEqN" # Stages avançados (quentes) por pipeline SDR_STAGES_QUENTES = { "70733ae5-a5d0-4ec5-8235-97859b8ef10f", # EM_NEGOCIACAO "faf155d0-a694-435c-9015-ccc974f72167", # ENCAMINHADO "1dddadaa-6c50-423d-a2fa-f2487e0a6baa", # DEMO_AGENDADA "bfca2016-a09c-42e5-86ef-a78836b67b48", # QUALIFICADO SDR } SDR_STAGES_GANHAMOS = {"72587990-1a07-4196-9582-117deebf2ae3"} SDR_STAGES_PERDEMOS = {"07ab9b2a-e5d4-4313-9c69-6a593fb7b193"} PROSP_STAGES_QUENTES = { "c6f8cce8-338d-43f5-bb13-5e98ecd08b64", # PROPOSTA "e573ee69-bc9f-4656-b1d4-2b91f155f1b6", # QUALIFICADO } PROSP_STAGES_GANHAMOS = {"4566e2f6-fd0b-4cb5-bc1a-f52239564ebd"} PROSP_STAGES_PERDEMOS = {"d1269476-f46f-4d2c-b81b-2418017d2067"} # Tags que indicam descarte TAGS_DESCARTE = { "no-contact", "bounce", "unsubscribe", "descadastrado", "dnd", "invalido", "spam", "blacklist", "opt-out", "cancelado", "blocked", } # Tags que indicam calor TAGS_WAITLIST = { "waitlist-photorf2", "waitlist_photorf2", "waitlist-cloudphotorf", "photorf2", "cloud-photorf", "cloudphotorf2", "lista-espera", "lista_espera_photorf", "photorf-2", } TAGS_CLIENTE_ATIVO = { "cliente-ativo", "cliente_ativo", "cliente ativo", "comprou", "cliente", "assinante", "renovacao", "assinatura-ativa", } TAGS_INTERESSE_PX3 = { "interesse-px3", "px3lab", "fotografia", "fotografo", "cloudphotorf", "photorf", "cloud-photorf", } # Campanhas quentes no UTM CAMPANHAS_QUENTES = { "cloudphotorf", "photorf", "px3-lancamento", "lancamento-photorf", "cloud_photorf", "photorf2", "cloudphotorf2", "photorf_2", } # ────────────────────────────────────────────── # PAGINAÇÃO DE CONTATOS # ────────────────────────────────────────────── def fetch_all_contacts(): contacts = [] url = f"{BASE_URL}/contacts/" params = {"locationId": LOCATION, "limit": 100} page = 0 while True: page += 1 try: r = requests.get(url, headers=HEADERS, params=params, timeout=30) r.raise_for_status() data = r.json() except Exception as e: print(f"[ERRO página {page}]: {e}") time.sleep(5) continue batch = data.get("contacts", []) contacts.extend(batch) print(f" Página {page}: +{len(batch)} contatos (total até agora: {len(contacts)})", flush=True) meta = data.get("meta", {}) next_url = meta.get("nextPageUrl") if not next_url or len(batch) < 100: break # Paginação: extrair startAfter e startAfterId start_after = meta.get("startAfter") start_after_id = meta.get("startAfterId") if not start_after or not start_after_id: break params = { "locationId": LOCATION, "limit": 100, "startAfter": start_after, "startAfterId": start_after_id, } time.sleep(0.5) # rate limit gentil return contacts # ────────────────────────────────────────────── # OPORTUNIDADES — mapear contato → stage/pipeline # ────────────────────────────────────────────── def fetch_opportunities(): """Retorna dict: contact_id -> lista de opps relevantes""" opp_map = defaultdict(list) url = f"{BASE_URL}/opportunities/search" params = { "location_id": LOCATION, "status": "open", "limit": 100, } page = 0 while True: page += 1 try: r = requests.get(url, headers=HEADERS, params=params, timeout=30) r.raise_for_status() data = r.json() except Exception as e: print(f"[ERRO opps página {page}]: {e}") break batch = data.get("opportunities", []) for opp in batch: cid = opp.get("contact", {}).get("id") or opp.get("contactId") if cid: opp_map[cid].append(opp) print(f" Opps página {page}: +{len(batch)} (total contatos com opp: {len(opp_map)})", flush=True) meta = data.get("meta", {}) start_after = meta.get("startAfter") start_after_id = meta.get("startAfterId") if not start_after or not start_after_id or len(batch) < 100: break params["startAfter"] = start_after params["startAfterId"] = start_after_id time.sleep(0.5) return dict(opp_map) # ────────────────────────────────────────────── # EXTRAÇÃO DE CUSTOM FIELDS # ────────────────────────────────────────────── def get_cf(custom_fields, field_id): for cf in (custom_fields or []): if cf.get("id") == field_id: return (cf.get("value") or "").strip() return "" # ────────────────────────────────────────────── # SCORING # ────────────────────────────────────────────── def days_since(date_str): """Retorna dias desde a data; None se inválido""" if not date_str: return None try: dt = datetime.fromisoformat(date_str.replace("Z", "+00:00")) return (NOW - dt).days except Exception: return None def score_contact(c, opps): """ Retorna (score, bucket, motivos) score é numérico (maior = mais quente) bucket: 'quente' | 'morno' | 'frio' | 'gelado' | 'descarte' """ score = 0 motivos = [] tags_lower = {t.lower() for t in (c.get("tags") or [])} email = (c.get("email") or "").strip().lower() dnd = c.get("dnd", False) custom_fields = c.get("customFields") or [] utm_source = get_cf(custom_fields, UTM_SOURCE_ID).lower() utm_campaign = get_cf(custom_fields, UTM_CAMPAIGN_ID).lower() utm_medium = get_cf(custom_fields, UTM_MEDIUM_ID).lower() sdr_temp = get_cf(custom_fields, SDR_TEMPERATURA_ID).lower() sdr_result = get_cf(custom_fields, SDR_RESULTADO_ID).lower() sdr_produto = get_cf(custom_fields, SDR_PRODUTO_INT_ID).lower() # ── DESCARTE IMEDIATO ── if dnd: return -999, "descarte", ["DND ativo"] if tags_lower & TAGS_DESCARTE: matched = tags_lower & TAGS_DESCARTE return -999, "descarte", [f"Tag de descarte: {', '.join(matched)}"] if not email: # Sem email não dá pra disparar - descarte para fins de email marketing return -998, "descarte", ["Sem email cadastrado"] # ── SIGNALS FORTES (quente) ── # Waitlist PhotoRF 2.0 if tags_lower & TAGS_WAITLIST: score += 50 motivos.append("tag waitlist PhotoRF 2.0") # Campanha UTM recente de PhotoRF for camp in CAMPANHAS_QUENTES: if camp in utm_campaign: score += 40 motivos.append(f"utm_campaign={utm_campaign}") break # SDR temperatura campo "quente" ou "muito quente" if "quente" in sdr_temp: score += 35 motivos.append(f"sdr_temperatura={sdr_temp}") # SDR resultado positivo if any(x in sdr_result for x in ["ganho", "won", "fechou", "comprou", "contratou"]): score += 45 motivos.append(f"sdr_resultado={sdr_result}") # Cliente ativo por tag if tags_lower & TAGS_CLIENTE_ATIVO: matched = tags_lower & TAGS_CLIENTE_ATIVO score += 45 motivos.append(f"tag cliente ativo: {', '.join(matched)}") # Tags de interesse PX3 if tags_lower & TAGS_INTERESSE_PX3: matched = tags_lower & TAGS_INTERESSE_PX3 score += 20 motivos.append(f"tag interesse PX3: {', '.join(matched)}") # Produto de interesse SDR = photorf ou cloud if any(x in sdr_produto for x in ["photorf", "cloud", "px3"]): score += 30 motivos.append(f"sdr_produto_interesse={sdr_produto}") # ── OPORTUNIDADES ── won_opps = [] hot_opps = [] pipeline_info = [] for opp in opps: stage_id = opp.get("pipelineStageId") or opp.get("stageId", "") pipeline_id = opp.get("pipelineId", "") stage_name = opp.get("stage", {}).get("name", "") if isinstance(opp.get("stage"), dict) else "" opp_name = opp.get("name", "") status = (opp.get("status") or "").lower() if stage_id in SDR_STAGES_GANHAMOS or stage_id in PROSP_STAGES_GANHAMOS or status == "won": score += 50 won_opps.append(f"WON: {opp_name or stage_name}") motivos.append(f"opp WON: {opp_name or 'sem nome'}") elif stage_id in SDR_STAGES_QUENTES or stage_id in PROSP_STAGES_QUENTES: score += 35 hot_opps.append(f"stage avancado: {stage_name or stage_id[:8]}") motivos.append(f"stage avançado: {stage_name or stage_id[:8]}") elif stage_id in SDR_STAGES_PERDEMOS or stage_id in PROSP_STAGES_PERDEMOS or status == "lost": score -= 10 motivos.append("opp perdida") else: score += 10 pipeline_info.append(f"pipeline ativo: {pipeline_id[:8]}") # ── RECÊNCIA DE INTERAÇÃO ── date_updated = c.get("dateUpdated") date_added = c.get("dateAdded") dias_update = days_since(date_updated) dias_added = days_since(date_added) # Usar o mais recente entre dateUpdated e dateAdded dias_interacao = dias_update if dias_interacao is None: dias_interacao = dias_added if dias_interacao is not None: if dias_interacao <= 7: score += 30 motivos.append(f"interação há {dias_interacao}d (muito recente)") elif dias_interacao <= 30: score += 20 motivos.append(f"interação há {dias_interacao}d") elif dias_interacao <= 60: score += 10 motivos.append(f"interação há {dias_interacao}d (morno)") elif dias_interacao <= 90: score += 5 motivos.append(f"interação há {dias_interacao}d") elif dias_interacao <= 180: score -= 5 motivos.append(f"interação há {dias_interacao}d (frio)") else: score -= 20 motivos.append(f"interação há {dias_interacao}d (gelado)") # ── UTM Source ── if utm_source in ("meta", "google", "facebook", "instagram"): score += 5 motivos.append(f"utm_source={utm_source}") # ── GELADO — sem email + sem interação recente ── if dias_interacao is not None and dias_interacao > 365: score -= 30 motivos.append("sem interação há +1 ano") # ── BUCKET FINAL ── if score >= 60: bucket = "quente" elif score >= 25: bucket = "morno" elif score >= 0: bucket = "frio" elif score >= -20: bucket = "gelado" else: bucket = "descarte" # Forçar descarte se sem email (já tratado acima, mas double-check) if not email and bucket not in ("descarte",): bucket = "descarte" motivos.append("sem email - inviável para email marketing") return score, bucket, motivos # ────────────────────────────────────────────── # CSV WRITER # ────────────────────────────────────────────── def write_csv(path, rows, fieldnames): with open(path, "w", newline="", encoding="utf-8") as f: writer = csv.DictWriter(f, fieldnames=fieldnames, extrasaction="ignore") writer.writeheader() writer.writerows(rows) print(f" Salvo: {path} ({len(rows)} linhas)") # ────────────────────────────────────────────── # MAIN # ────────────────────────────────────────────── def main(): print("=" * 60) print("MAPEAMENTO LEADS PX3 — Cloud PhotoRF 2.0 (07/10/2026)") print("=" * 60) print("\n[1/3] Buscando contatos...") contacts = fetch_all_contacts() print(f"Total de contatos carregados: {len(contacts)}") print("\n[2/3] Buscando oportunidades...") opp_map = fetch_opportunities() print(f"Contatos com oportunidade(s): {len(opp_map)}") print("\n[3/3] Scoring e segmentação...") buckets = {"quente": [], "morno": [], "frio": [], "gelado": [], "descarte": []} stats_tags = defaultdict(int) stats_utm_campaign = defaultdict(int) for c in contacts: cid = c.get("id", "") opps = opp_map.get(cid, []) score, bucket, motivos = score_contact(c, opps) custom_fields = c.get("customFields") or [] tags = c.get("tags") or [] for t in tags: stats_tags[t] += 1 utm_campaign = get_cf(custom_fields, UTM_CAMPAIGN_ID) if utm_campaign: stats_utm_campaign[utm_campaign] += 1 # Data de última interação date_updated = c.get("dateUpdated") or c.get("dateAdded") or "" dias = days_since(date_updated) # Stage atual (pegar o mais relevante) stage_atual = "" pipeline_atual = "" if opps: opp = opps[0] stage_info = opp.get("stage") or {} stage_atual = stage_info.get("name", "") if isinstance(stage_info, dict) else "" pipeline_atual = opp.get("pipelineId", "")[:8] row = { "email": c.get("email") or "", "nome": c.get("contactName") or f"{c.get('firstName','')} {c.get('lastName','')}".strip(), "telefone": c.get("phone") or "", "score": score, "bucket": bucket, "stage_atual": stage_atual, "pipeline": pipeline_atual, "ultima_interacao": date_updated[:10] if date_updated else "", "dias_desde_interacao": dias if dias is not None else "", "tags": "; ".join(tags), "utm_source": get_cf(custom_fields, UTM_SOURCE_ID), "utm_medium": get_cf(custom_fields, UTM_MEDIUM_ID), "utm_campaign": get_cf(custom_fields, UTM_CAMPAIGN_ID), "sdr_temperatura": get_cf(custom_fields, SDR_TEMPERATURA_ID), "sdr_resultado": get_cf(custom_fields, SDR_RESULTADO_ID), "sdr_produto_interesse": get_cf(custom_fields, SDR_PRODUTO_INT_ID), "dnd": c.get("dnd", False), "source": c.get("source") or "", "motivo_score": " | ".join(motivos) if motivos else "sem sinais", } buckets[bucket].append(row) # ── RELATÓRIO DE CONTAGEM ── print("\n" + "=" * 60) print("RESULTADO DA SEGMENTAÇÃO") print("=" * 60) for b, rows in buckets.items(): print(f" {b.upper():10s}: {len(rows):>5} leads") total_ativo = len(buckets["quente"]) + len(buckets["morno"]) + len(buckets["frio"]) total_gelado = len(buckets["gelado"]) + len(buckets["descarte"]) print(f"\n TOTAL ATIVO (Q+M+F): {total_ativo}") print(f" DESCARTE+GELADO: {total_gelado}") print(f" TOTAL GERAL: {len(contacts)}") print("\n Top 10 tags da base:") for tag, count in sorted(stats_tags.items(), key=lambda x: -x[1])[:10]: print(f" {tag}: {count}") print("\n Top 10 utm_campaign:") for camp, count in sorted(stats_utm_campaign.items(), key=lambda x: -x[1])[:10]: print(f" {camp}: {count}") # ── SALVAR CSVs ── print("\nSalvando CSVs...") FIELDNAMES = [ "email", "nome", "telefone", "score", "bucket", "stage_atual", "pipeline", "ultima_interacao", "dias_desde_interacao", "tags", "utm_source", "utm_medium", "utm_campaign", "sdr_temperatura", "sdr_resultado", "sdr_produto_interesse", "dnd", "source", "motivo_score" ] # Ordenar por score desc dentro de cada bucket for b in buckets: buckets[b].sort(key=lambda r: -r["score"]) write_csv(OUT_DIR / "leads_quente.csv", buckets["quente"], FIELDNAMES) write_csv(OUT_DIR / "leads_morno.csv", buckets["morno"], FIELDNAMES) write_csv(OUT_DIR / "leads_frio.csv", buckets["frio"], FIELDNAMES) # Gelado + descarte juntos no descarte descarte_total = buckets["gelado"] + buckets["descarte"] descarte_total.sort(key=lambda r: -r["score"]) write_csv(OUT_DIR / "leads_descarte.csv", descarte_total, FIELDNAMES) # ── RETORNO ESTRUTURADO ── result = { "total": len(contacts), "buckets": {b: len(rows) for b, rows in buckets.items()}, "total_ativo": total_ativo, "stats_tags": dict(sorted(stats_tags.items(), key=lambda x: -x[1])[:20]), "stats_utm_campaign": dict(sorted(stats_utm_campaign.items(), key=lambda x: -x[1])[:20]), "sample_quente": buckets["quente"][:5] if buckets["quente"] else [], } with open(OUT_DIR / "stats.json", "w") as f: json.dump(result, f, indent=2, default=str, ensure_ascii=False) print(f"\nstats.json salvo em {OUT_DIR}") print("CONCLUÍDO.") return result if __name__ == "__main__": result = main()