#!/usr/bin/env python3 """ Script: Importar campo CNPJ nos contatos Santa Fe — CRM Linkia PX3 Executa em 2026-08-07 após reset do rate limit diário (~08:15 UTC) Estratégia: 1. Criar custom field CNPJ (se não existir) 2. Para cada linha do CSV: a. Se contactId já conhecido (duplicados de ontem): atualiza direto b. Se contato criado ontem: busca por email/telefone c. Se foi 429 ontem: tenta criar agora d. Se foi linha vazia/tel quebrado: pula 3. Valida 3 amostras aleatórias Histórico de ontem (2026-08-06): - 82 contatos criados - 27 duplicados com contactId conhecido - 12 erros irrecuperáveis (linhas vazias, telefone inválido) """ import csv import json import time import random import re import sys import urllib.request import urllib.error import urllib.parse from io import StringIO # ============================================================ # CONFIGURAÇÃO # ============================================================ TOKEN = "pit-25f78b4f-4cc8-48f2-b8dc-73a98f851eaf" LOCATION_ID = "W7PGxpfbsFaEEUoQOtUb" BASE_URL = "https://services.leadconnectorhq.com" CSV_PATH = "/opt/mia/workspace/clientes/px3lab/santa_fe_planilha.csv" RESULT_PATH = "/opt/mia/workspace/clientes/px3lab/cnpj_import_result.json" HEADERS = { "Authorization": f"Bearer {TOKEN}", "Version": "2021-07-28", "Content-Type": "application/json" } # Mapa linha (1-based, excluindo header) -> contactId já conhecido dos duplicados de ontem # Baseado no resultado de /tmp/import_santafe_resultado.json KNOWN_CONTACT_IDS = { 1: "G4dZHiJc6fWMjDCtV3kl", # 14.864.212 ADRIANA BIEMBENGUT 4: "qxD5D1uL3D9j9sZGQL5r", # 35.796.980 REGIS RODRIGO 17: "6uxZWUqCLqWr4oSHOHNV", # 58.854.537 FERNANDO HENRIQUE 41: "cKhxgF5LQdMXYfMgzAJF", # CLEBER CRISTIANO IZEPON 50: "Gv6AsR1hkHppcaaBcz7w", # F.J. CAMILO & E.M. SILVA 53: "Aehqhys8zyK7B8jrQ3PF", # GAMA FORMATURAS 56: "1B1YzDzorjINpvGVwivt", # GLOBAL FORMATURAS 64: "yXkKdATOQejnW3ouyYtY", # JMA PRODUCOES FOTOGRAFICAS 65: "1B1YzDzorjINpvGVwivt", # JN DE MORAES FORMATURAS 67: "H6ATpajUlL8iPiUCtdwH", # JN FORMATURAS LTDA 68: "2Hfh5eH6yjrD8UhkKOgz", # JOAO BORGES LTDA 69: "76dmu3Z7eYCzNxbL2rLb", # JOAO LUIZ MAURO 72: "qBSOuvhrOEvw0G3BB82l", # KAIROS FORMATURAS 75: "5RWyaTgXWmcgbBaKNcpK", # LUIZ HENRIQUE DO NASCIMENTO 82: "2Hfh5eH6yjrD8UhkKOgz", # MLFA EVENTOS FOTOGRAFICOS 92: "htNDbE2wVuFZTvWolOys", # RAFAEL COLETTO LOPES 93: "qBSOuvhrOEvw0G3BB82l", # REVER FORMATURAS 98: "htNDbE2wVuFZTvWolOys", # SANTA FE FOTO & VIDEO 99: "qBSOuvhrOEvw0G3BB82l", # SANTOS & PEDRAZZANI 103: "2Hfh5eH6yjrD8UhkKOgz", # SEVEN FORMATURAS 104: "1B1YzDzorjINpvGVwivt", # SOLENE FORMATURAS 106: "1B1YzDzorjINpvGVwivt", # STUDIO MAX 107: "qBSOuvhrOEvw0G3BB82l", # SUPER LOJA SANTA FE 111: "h0XTgKwVAZJlsfhEd7xf", # V M VENDAS & FORMATURAS 115: "1B1YzDzorjINpvGVwivt", # VALLE FORMATURAS 118: "Pv9woHDbEmxZdBj2bZ8A", # VENSE VENDAS ESPECIAIS 120: "2Hfh5eH6yjrD8UhkKOgz", # VIP FORMATURAS } # Linhas que devem ser puladas (irrecuperáveis) SKIP_LINES = { 12: "linha_vazia", 32: "telefone_longo", 35: "linha_vazia", 54: "telefone_longo", 61: "linha_vazia", 84: "linha_vazia", 85: "telefone_longo", 110: "telefone_longo", } # Linhas que falharam com 429 ontem - precisam ser criadas agora RATE_LIMITED_LINES = {94, 95, 96, 97} # ============================================================ # HELPERS # ============================================================ def api(method, path, body=None, retries=6): url = BASE_URL + path data = json.dumps(body).encode() if body else None req = urllib.request.Request(url, data=data, headers=HEADERS, method=method) for attempt in range(retries): try: with urllib.request.urlopen(req, timeout=30) as resp: return json.loads(resp.read()) except urllib.error.HTTPError as e: resp_body = e.read().decode() if e.code == 429: wait = 30 * (attempt + 1) print(f" [RATE LIMIT] aguardando {wait}s... (tentativa {attempt+1}/{retries})", flush=True) time.sleep(wait) continue print(f" [HTTP {e.code}] {path} => {resp_body[:200]}", flush=True) return {"error": e.code, "message": resp_body} except Exception as ex: print(f" [ERR] {ex}", flush=True) time.sleep(5) return {"error": "max_retries"} def normalize_phone(phone): digits = re.sub(r'\D', '', phone) if not digits: return None if len(digits) == 11 and digits.startswith('0'): digits = digits[1:] if len(digits) in (10, 11): digits = "55" + digits if len(digits) >= 12: return "+" + digits return None def normalize_cnpj(cnpj_raw): if not cnpj_raw or not cnpj_raw.strip(): return None digits = re.sub(r'\D', '', cnpj_raw.strip()) if len(digits) == 14: return f"{digits[:2]}.{digits[2:5]}.{digits[5:8]}/{digits[8:12]}-{digits[12:]}" return cnpj_raw.strip() # ============================================================ # PASSO 1: Criar ou reusar campo CNPJ # ============================================================ def get_or_create_cnpj_field(): print("\n=== PASSO 1: Custom Field CNPJ ===", flush=True) resp = api("GET", f"/locations/{LOCATION_ID}/customFields") if resp.get("error"): print(f" ERRO ao listar campos: {resp}", flush=True) return None, None fields = resp.get("customFields", []) for f in fields: name = f.get("name", "").lower() key = f.get("fieldKey", "").lower() if "cnpj" in name or "cnpj" in key: print(f" Campo CNPJ JA EXISTE: id={f['id']} key={f.get('fieldKey')} name={f.get('name')}", flush=True) return f["id"], f.get("fieldKey") print(" Campo CNPJ não encontrado. Criando...", flush=True) payload = {"name": "CNPJ", "dataType": "TEXT", "model": "contact"} result = api("POST", f"/locations/{LOCATION_ID}/customFields", body=payload) field = result.get("customField") or result if field.get("id"): print(f" CRIADO: id={field['id']} key={field.get('fieldKey')}", flush=True) return field["id"], field.get("fieldKey") print(f" ERRO ao criar campo: {json.dumps(result)}", flush=True) return None, None # ============================================================ # PASSO 2: Ler CSV # ============================================================ def load_csv(): print("\n=== PASSO 2: Lendo CSV ===", flush=True) with open(CSV_PATH, encoding="utf-8") as f: content = f.read() reader = csv.DictReader(StringIO(content)) rows = list(reader) print(f" Total de linhas: {len(rows)}", flush=True) return rows # ============================================================ # Buscar contato por query (email ou telefone) # ============================================================ def find_contact_by_query(query): if not query or not query.strip(): return None resp = api("GET", f"/contacts/search?locationId={LOCATION_ID}&query={urllib.parse.quote(query.strip())}&limit=5") contacts = resp.get("contacts", []) return contacts[0]["id"] if contacts else None def create_contact(row, cnpj, cnpj_field_id): razao = row.get("Razao Social", row.get("Razão Social", "")).strip() email = row.get("E-mail", "").strip() celular = row.get("Celular", "").strip() telefone = row.get("Telefone", "").strip() phone = normalize_phone(celular) or normalize_phone(telefone) name = re.sub(r'^\d{2}\.\d{3}\.\d{3}\s+', '', razao).strip() parts = name.split() first = parts[0].capitalize() if parts else name last = " ".join(p.capitalize() for p in parts[1:]) if len(parts) > 1 else "" payload = { "firstName": first, "lastName": last, "companyName": name, "locationId": LOCATION_ID, "source": "Import planilha Santa Fé", "tags": ["origem:import-planilha", "evento:santa-fe-ggv", "import:2026-08-06"], "customFields": [{"id": cnpj_field_id, "value": cnpj}] } if email and "@" in email: payload["email"] = email if phone: payload["phone"] = phone resp = api("POST", "/contacts/", body=payload) contact = resp.get("contact", resp) if contact.get("id"): return contact["id"], "created" return None, f"error_{resp}" # ============================================================ # PASSO 3: Processar cada linha # ============================================================ def process_rows(rows, cnpj_field_id): print(f"\n=== PASSO 3: Processando {len(rows)} linhas ===", flush=True) stats = { "updated_known_id": 0, "updated_found": 0, "created_new": 0, "skipped_empty": 0, "skipped_no_cnpj": 0, "skipped_tel_longo": 0, "skipped_no_identifier": 0, "errors": 0, "skip_details": [] } results = [] for i, row in enumerate(rows, 1): razao = row.get("Razão Social", row.get("Razao Social", "")).strip() cnpj_raw = row.get("CNPJ", "").strip() email = row.get("E-mail", "").strip() celular = row.get("Celular", "").strip() telefone = row.get("Telefone", "").strip() # Linha totalmente vazia if not razao and not cnpj_raw and not email: print(f" [{i:03d}] Linha vazia - pulando", flush=True) stats["skipped_empty"] += 1 stats["skip_details"].append({"linha": i, "motivo": "linha_vazia"}) continue # Linhas irrecuperáveis conhecidas if i in SKIP_LINES: motivo = SKIP_LINES[i] print(f" [{i:03d}] {razao[:40]} - pulando ({motivo})", flush=True) if motivo == "linha_vazia": stats["skipped_empty"] += 1 elif motivo == "telefone_longo": stats["skipped_tel_longo"] += 1 stats["skip_details"].append({"linha": i, "razao": razao, "motivo": motivo}) continue # CNPJ vazio cnpj = normalize_cnpj(cnpj_raw) if not cnpj: print(f" [{i:03d}] {razao[:40]} - CNPJ vazio, pulando", flush=True) stats["skipped_no_cnpj"] += 1 stats["skip_details"].append({"linha": i, "razao": razao, "motivo": "cnpj_vazio"}) continue print(f" [{i:03d}] {razao[:45]} | CNPJ={cnpj}", flush=True) # CASO A: contactId já conhecido (duplicados de ontem) if i in KNOWN_CONTACT_IDS: contact_id = KNOWN_CONTACT_IDS[i] payload = {"customFields": [{"id": cnpj_field_id, "value": cnpj}]} resp = api("PUT", f"/contacts/{contact_id}", body=payload) if resp.get("contact") or resp.get("id") or resp.get("succeded"): print(f" -> ATUALIZADO (known_id) {contact_id}", flush=True) stats["updated_known_id"] += 1 results.append({"linha": i, "razao": razao, "status": "updated_known_id", "cnpj": cnpj, "contact_id": contact_id}) else: print(f" -> ERRO (known_id) {contact_id}: {resp}", flush=True) stats["errors"] += 1 results.append({"linha": i, "razao": razao, "status": "error", "cnpj": cnpj, "contact_id": contact_id, "error": str(resp)}) time.sleep(0.4) continue # CASO C: linhas com 429 ontem - criar agora if i in RATE_LIMITED_LINES: contact_id, action = create_contact(row, cnpj, cnpj_field_id) if contact_id: print(f" -> CRIADO (rate_limited ontem) {contact_id}", flush=True) stats["created_new"] += 1 results.append({"linha": i, "razao": razao, "status": "created", "cnpj": cnpj, "contact_id": contact_id}) else: print(f" -> ERRO ao criar: {action}", flush=True) stats["errors"] += 1 results.append({"linha": i, "razao": razao, "status": "error", "cnpj": cnpj, "contact_id": None, "error": str(action)}) time.sleep(0.4) continue # CASO B: contatos criados ontem - buscar por email/tel phone = normalize_phone(celular) or normalize_phone(telefone) contact_id = None if email and "@" in email: contact_id = find_contact_by_query(email) time.sleep(0.3) if not contact_id and phone: contact_id = find_contact_by_query(phone) time.sleep(0.3) if contact_id: payload = {"customFields": [{"id": cnpj_field_id, "value": cnpj}]} resp = api("PUT", f"/contacts/{contact_id}", body=payload) if resp.get("contact") or resp.get("id") or resp.get("succeded"): print(f" -> ATUALIZADO (found) {contact_id}", flush=True) stats["updated_found"] += 1 results.append({"linha": i, "razao": razao, "status": "updated_found", "cnpj": cnpj, "contact_id": contact_id}) else: print(f" -> ERRO (found) {contact_id}: {resp}", flush=True) stats["errors"] += 1 results.append({"linha": i, "razao": razao, "status": "error", "cnpj": cnpj, "contact_id": contact_id, "error": str(resp)}) elif not email and not phone: print(f" -> Sem identificador válido - pulando", flush=True) stats["skipped_no_identifier"] += 1 stats["skip_details"].append({"linha": i, "razao": razao, "motivo": "sem_identificador"}) else: # Contato não encontrado mesmo tendo email/telefone - tenta criar contact_id, action = create_contact(row, cnpj, cnpj_field_id) if contact_id: print(f" -> CRIADO (nao encontrado) {contact_id}", flush=True) stats["created_new"] += 1 results.append({"linha": i, "razao": razao, "status": "created", "cnpj": cnpj, "contact_id": contact_id}) else: print(f" -> ERRO ao criar: {action}", flush=True) stats["errors"] += 1 results.append({"linha": i, "razao": razao, "status": "error", "cnpj": cnpj, "contact_id": None, "error": str(action)}) time.sleep(0.4) return stats, results # ============================================================ # PASSO 4: Validar 3 amostras # ============================================================ def validate_samples(results, cnpj_field_id): print("\n=== PASSO 4: Validando amostras ===", flush=True) success = [r for r in results if r["status"] in ("updated_known_id", "updated_found", "created") and r.get("contact_id")] if not success: print(" Nenhum contato com sucesso para validar", flush=True) return [] samples = random.sample(success, min(3, len(success))) validations = [] for s in samples: contact_id = s["contact_id"] expected_cnpj = s["cnpj"] resp = api("GET", f"/contacts/{contact_id}") contact = resp.get("contact", resp) found_cnpj = None for cf in contact.get("customFields", []): if cf.get("id") == cnpj_field_id: found_cnpj = cf.get("value") break ok = found_cnpj == expected_cnpj status_str = "OK" if ok else f"DIVERGENTE (esperado={expected_cnpj}, encontrado={found_cnpj})" print(f" {contact_id} | {s['razao'][:40]} | {status_str}", flush=True) validations.append({ "contact_id": contact_id, "razao": s["razao"], "expected_cnpj": expected_cnpj, "found_cnpj": found_cnpj, "ok": ok }) time.sleep(0.5) return validations # ============================================================ # MAIN # ============================================================ if __name__ == "__main__": print("=" * 60, flush=True) print("IMPORTAÇÃO CNPJ — CRM LINKIA PX3 — 2026-08-07", flush=True) print("=" * 60, flush=True) cnpj_field_id, cnpj_field_key = get_or_create_cnpj_field() if not cnpj_field_id: print("ERRO FATAL: Não foi possível criar/obter o custom field CNPJ. Abortando.", flush=True) sys.exit(1) print(f"\n Custom Field CNPJ: id={cnpj_field_id} key={cnpj_field_key}", flush=True) rows = load_csv() stats, results = process_rows(rows, cnpj_field_id) validations = validate_samples(results, cnpj_field_id) total_cnpj = stats["updated_known_id"] + stats["updated_found"] + stats["created_new"] output = { "cnpj_field_id": cnpj_field_id, "cnpj_field_key": cnpj_field_key, "stats": stats, "total_com_cnpj_preenchido": total_cnpj, "validations": validations, "results": results } with open(RESULT_PATH, "w") as f: json.dump(output, f, indent=2, ensure_ascii=False) print("\n" + "=" * 60, flush=True) print("RELATÓRIO FINAL", flush=True) print("=" * 60, flush=True) print(f"Custom Field CNPJ ID: {cnpj_field_id}", flush=True) print(f"Custom Field CNPJ Key: {cnpj_field_key}", flush=True) print(f"", flush=True) print(f"Atualizados (ID já known): {stats['updated_known_id']}", flush=True) print(f"Atualizados (busca GHL): {stats['updated_found']}", flush=True) print(f"Criados agora: {stats['created_new']}", flush=True) print(f"TOTAL com CNPJ preenchido: {total_cnpj}", flush=True) print(f"", flush=True) print(f"Ignorados - linha vazia: {stats['skipped_empty']}", flush=True) print(f"Ignorados - CNPJ vazio: {stats['skipped_no_cnpj']}", flush=True) print(f"Ignorados - tel longo: {stats['skipped_tel_longo']}", flush=True) print(f"Ignorados - sem identif.: {stats['skipped_no_identifier']}", flush=True) print(f"Erros: {stats['errors']}", flush=True) print(f"", flush=True) print("Amostras validadas:", flush=True) for v in validations: print(f" {v['contact_id']} | {'OK' if v['ok'] else 'FALHOU'} | {v['razao'][:40]}", flush=True) print(f"\nResultado completo: {RESULT_PATH}", flush=True)