#!/usr/bin/env python3 """ Sweet Angels - SDR Médio Porte Sorocaba Setup: custom fields + importação de 300 leads no CRM Linkia """ import json import time import openpyxl import requests from datetime import datetime, date # --- Config --- TOKEN = "pit-5c81b5cf-19ef-4b6d-8d90-2df3a3e054b7" LOC = "4z6Fjpgj2soAQZpu8Ddk" BASE = "https://services.leadconnectorhq.com" HEADERS = { "Authorization": f"Bearer {TOKEN}", "Version": "2021-07-28", "Content-Type": "application/json" } PIPELINE_ID = "KMQE42Xf8IQ0WoNxM77r" ENTRY_STAGE = "74908ed8-4fbf-447f-9341-b8e8f72feeae" # NOVO LEAD XLSX = "/opt/mia/workspace/clientes/sweet_angels/sdr_medio_porte/dados/base_prospeccao_300.xlsx" FIELD_MAP_PATH = "/opt/mia/workspace/clientes/sweet_angels/sdr_medio_porte/field_map.json" RESULT_PATH = "/opt/mia/workspace/clientes/sweet_angels/sdr_medio_porte/medio_porte_envio_linkia.json" # --- Definição dos 27 custom fields desejados --- # (nome_planilha, nome_crm, dataType, [picklistOptions]) FIELDS_SPEC = [ ("ID", "ID_Prospeccao", "NUMBER", None), ("Empresa", "Empresa", "TEXT", None), ("Razão social", "Razao_Social", "TEXT", None), ("CNPJ", "CNPJ", "TEXT", None), # já existe ("Segmento", "Segmento", "TEXT", None), ("Cidade", "Cidade_Prospeccao", "TEXT", None), ("UF", "UF", "TEXT", None), ("Faixa de porte", "Faixa_de_Porte", "TEXT", None), ("Porte Receita", "Porte_Receita", "TEXT", None), ("Capital social", "Capital_Social", "NUMBER", None), ("Data de abertura", "Data_de_Abertura", "DATE", None), ("Telefone", "Telefone_Prospeccao", "TEXT", None), ("Telefone 2", "Telefone_2", "TEXT", None), ("E-mail", "Email_Prospeccao", "TEXT", None), ("Site", "Site", "TEXT", None), # já existe ("Evidência", "Evidencia", "TEXT", None), ("Sinal de investimento", "Sinal_Investimento", "TEXT", None), ("Prioridade", "Prioridade_Lead", "TEXT", None), ("Decisor sugerido", "Decisor_Sugerido", "TEXT", None), ("Fonte empresarial", "Fonte_Empresarial", "TEXT", None), ("Fonte cadastral", "Fonte_Cadastral", "TEXT", None), ("Data da validação", "Data_da_Validacao", "DATE", None), ("Status CRM", "Status_CRM", "TEXT", None), ("Responsável", "Responsavel", "TEXT", None), ("Próxima ação", "Proxima_Acao", "TEXT", None), ("Data próxima ação", "Data_Proxima_Acao", "DATE", None), ("Observações", "Observacoes", "TEXT", None), ] # Mapeamento manual de campos que já existem (nome_crm -> id existente) EXISTING_FIELD_IDS = { "CNPJ": "OXuuJ2qaOS3Rft8yYYbM", "Site": "HTynXy6CmQBknlpaGxBH", } def api_get(path): r = requests.get(f"{BASE}{path}", headers=HEADERS) r.raise_for_status() return r.json() def api_post(path, body): r = requests.post(f"{BASE}{path}", headers=HEADERS, json=body) body_text = r.text # consume body try: data = r.json() except Exception: data = {"raw": body_text} return r.status_code, data def api_put(path, body): r = requests.put(f"{BASE}{path}", headers=HEADERS, json=body) body_text = r.text try: data = r.json() except Exception: data = {"raw": body_text} return r.status_code, data # ============================================================ # FASE 1 — Mapeamento de custom fields # ============================================================ def build_field_map(): print("\n=== FASE 1: Mapeamento de custom fields ===") # Busca todos os custom fields existentes existing = api_get(f"/locations/{LOC}/customFields") existing_fields = existing.get("customFields", []) # Índice por nome e por fieldKey by_name = {f["name"].strip().lower(): f for f in existing_fields} by_key = {f["fieldKey"].lower(): f for f in existing_fields} field_map = {} # col_planilha -> {field_id, field_name, field_key, status} for (col_planilha, nome_crm, dtype, picklist) in FIELDS_SPEC: # Verifica se já existe pelo ID manual if nome_crm in EXISTING_FIELD_IDS: fid = EXISTING_FIELD_IDS[nome_crm] field_obj = next((f for f in existing_fields if f["id"] == fid), None) field_map[col_planilha] = { "field_id": fid, "field_name": field_obj["name"] if field_obj else nome_crm, "field_key": field_obj["fieldKey"] if field_obj else f"contact.{nome_crm.lower()}", "status": "reutilizado" } print(f" [REUSO] {col_planilha} -> {fid} ({nome_crm})") continue # Verifica por nome key_lower = nome_crm.replace("_", " ").lower() alt_lower = col_planilha.strip().lower() found = by_name.get(key_lower) or by_name.get(alt_lower) if found: field_map[col_planilha] = { "field_id": found["id"], "field_name": found["name"], "field_key": found["fieldKey"], "status": "reutilizado" } print(f" [REUSO] {col_planilha} -> {found['id']} ({found['name']})") continue # Cria novo campo payload = { "name": nome_crm, "dataType": dtype, "model": "contact" } if picklist: payload["picklistOptions"] = picklist status_code, resp = api_post(f"/locations/{LOC}/customFields", payload) time.sleep(0.4) if status_code in (200, 201): cf = resp.get("customField", resp) fid = cf.get("id", "") fkey = cf.get("fieldKey", "") field_map[col_planilha] = { "field_id": fid, "field_name": nome_crm, "field_key": fkey, "status": "criado" } print(f" [CRIADO] {col_planilha} -> {fid} ({nome_crm})") else: print(f" [ERRO] {col_planilha}: {status_code} {resp}") field_map[col_planilha] = { "field_id": None, "field_name": nome_crm, "field_key": None, "status": f"erro_{status_code}" } with open(FIELD_MAP_PATH, "w") as f: json.dump(field_map, f, ensure_ascii=False, indent=2) print(f"\nField map salvo em {FIELD_MAP_PATH}") return field_map # ============================================================ # FASE 2 — Importação de leads # ============================================================ def serialize_value(val): """Converte valores da planilha para string compatível com o CRM.""" if val is None: return None if isinstance(val, (datetime, date)): if hasattr(val, 'strftime'): return val.strftime("%Y-%m-%d") if isinstance(val, float): if val == int(val): return str(int(val)) return str(val) if isinstance(val, int): return str(val) return str(val).strip() or None def load_xlsx(): wb = openpyxl.load_workbook(XLSX) # Leads_Região — todos os 300 ws_main = wb["Leads_Região"] headers = [cell.value for cell in ws_main[1]] leads = [] for row in ws_main.iter_rows(min_row=2, values_only=True): if not any(row): continue leads.append(dict(zip(headers, row))) # Prioridade_A — IDs das 29 empresas prioridade ws_prio = wb["Prioridade_A"] prio_ids = set() for row in ws_prio.iter_rows(min_row=2, values_only=True): if row[0] is not None: prio_ids.add(row[0]) # coluna ID # Ordena: Prioridade_A primeiro leads_sorted = sorted(leads, key=lambda x: (0 if x.get("ID") in prio_ids else 1, x.get("ID") or 9999)) return leads_sorted, prio_ids def import_leads(field_map, prio_ids): print("\n=== FASE 2: Importação de leads ===") leads_data, _ = load_xlsx() sucessos = [] falhas = [] total = len(leads_data) for i, lead in enumerate(leads_data, 1): empresa = lead.get("Empresa") or f"Lead_{i}" cnpj = lead.get("CNPJ") tel = lead.get("Telefone") tel2 = lead.get("Telefone 2") email = lead.get("E-mail") is_prio = lead.get("ID") in prio_ids print(f" [{i}/{total}] {empresa} {'[PRIO-A]' if is_prio else ''}") # --- Monta custom fields --- custom_fields = [] for col_planilha, finfo in field_map.items(): if not finfo.get("field_id"): continue raw_val = lead.get(col_planilha) val = serialize_value(raw_val) if val is not None: custom_fields.append({ "id": finfo["field_id"], "value": val }) # --- Cria / atualiza Contact --- contact_payload = { "locationId": LOC, "name": empresa, "companyName": lead.get("Razão social") or empresa, "customFields": custom_fields, "tags": ["medio-porte-sorocaba"] + (["Prioridade_A"] if is_prio else []) } if tel: contact_payload["phone"] = str(tel) if email: contact_payload["email"] = str(email) status_code, resp = api_post("/contacts/", contact_payload) if status_code == 400 and "contactId" in str(resp): # Lead já existe - pega o ID do erro contact_id = (resp.get("meta") or {}).get("contactId") or resp.get("contactId") if not contact_id: # tenta no message import re match = re.search(r'[0-9a-zA-Z]{20}', str(resp)) contact_id = match.group(0) if match else None print(f" -> Contato já existe, usando {contact_id}") elif status_code in (200, 201): contact_id = (resp.get("contact") or {}).get("id") or resp.get("id") else: print(f" [ERRO contato] {status_code}: {resp}") falhas.append({"empresa": empresa, "motivo": f"contact_{status_code}", "detalhe": str(resp)[:200]}) continue if not contact_id: print(f" [ERRO] Não conseguiu extrair contact_id: {resp}") falhas.append({"empresa": empresa, "motivo": "sem_contact_id", "detalhe": str(resp)[:200]}) continue # Delay pós-criação de contato (indexação GHL) time.sleep(1.0) # --- Cria Opportunity --- opp_payload = { "pipelineId": PIPELINE_ID, "locationId": LOC, "name": empresa, "pipelineStageId": ENTRY_STAGE, "status": "open", "contactId": contact_id, } status_opp, resp_opp = api_post("/opportunities/", opp_payload) if status_opp in (200, 201): opp_id = (resp_opp.get("opportunity") or {}).get("id") or resp_opp.get("id") sucessos.append({ "empresa": empresa, "contact_id": contact_id, "opportunity_id": opp_id, "prioridade_a": is_prio }) print(f" -> OK | contact={contact_id} | opp={opp_id}") else: print(f" [ERRO opportunity] {status_opp}: {resp_opp}") # Contato criado mas opp falhou - registra como sucesso parcial sucessos.append({ "empresa": empresa, "contact_id": contact_id, "opportunity_id": None, "prioridade_a": is_prio, "opp_erro": f"{status_opp}" }) falhas.append({"empresa": empresa, "motivo": f"opp_{status_opp}", "detalhe": str(resp_opp)[:200]}) # Throttle gentil time.sleep(0.3) result = { "gerado_em": datetime.now().isoformat(), "total": total, "sucessos_count": len([s for s in sucessos if s.get("opportunity_id")]), "falhas_count": len(falhas), "sucessos": sucessos, "falhas": falhas } with open(RESULT_PATH, "w") as f: json.dump(result, f, ensure_ascii=False, indent=2) print(f"\n=== RESULTADO ===") print(f" Total processado: {total}") print(f" Contatos + Opportunities OK: {result['sucessos_count']}") print(f" Falhas: {result['falhas_count']}") print(f" Resultado em: {RESULT_PATH}") return result # ============================================================ # FASE 3 — Verificação spot-check (1 contato) # ============================================================ def spot_check(sucessos): if not sucessos: print("\n[SKIP] Nenhum sucesso para verificar.") return # Pega primeiro que tem contact_id first = next((s for s in sucessos if s.get("contact_id")), None) if not first: return cid = first["contact_id"] print(f"\n=== FASE 3: Spot-check contato {cid} ({first['empresa']}) ===") r = requests.get(f"{BASE}/contacts/{cid}", headers=HEADERS) r.text # consume if r.status_code == 200: data = r.json() contact = data.get("contact", data) cfs = contact.get("customFields", []) print(f" Custom fields preenchidos: {len(cfs)}") for cf in cfs[:10]: print(f" id={cf.get('id')} | value={cf.get('value')}") tags = contact.get("tags", []) print(f" Tags: {tags}") else: print(f" Erro ao buscar contato: {r.status_code}") # ============================================================ # MAIN # ============================================================ if __name__ == "__main__": print("Sweet Angels - SDR Médio Porte Sorocaba - Setup Import") print(f"Location: {LOC} | Pipeline: {PIPELINE_ID}") print(f"Entry stage: NOVO LEAD ({ENTRY_STAGE})") print("=" * 60) leads_data, prio_ids = load_xlsx() print(f"\nPlanilha carregada: {len(leads_data)} leads | {len(prio_ids)} Prioridade_A") field_map = build_field_map() result = import_leads(field_map, prio_ids) spot_check(result["sucessos"]) print("\nConcluído.")