#!/usr/bin/env python3 """ Enriquecimento de dados via Firecrawl Lê empresas do template, encontra websites e extrai contatos """ import json import csv import time import os import re from pathlib import Path from datetime import datetime from typing import List, Dict, Optional, Any import requests from dataclasses import dataclass, asdict import pandas as pd # ============ CONFIG ============ FIRECRAWL_API_KEY = "fc-c2836375e10d4bf5b73013cd4e67fa6f" FIRECRAWL_BASE = "https://api.firecrawl.dev/v1" # Arquivos de entrada/saída DATA_DIR = Path("/root/lelyon_prospecting/data") INPUT_JSON = DATA_DIR / "lelyon_prospects_latest.json" INPUT_EXCEL = DATA_DIR / "lelyon_prospects_20260904_225404.xlsx" OUTPUT_JSON = DATA_DIR / f"lelyon_enriched_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json" OUTPUT_CSV = DATA_DIR / f"lelyon_enriched_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv" OUTPUT_EXCEL = DATA_DIR / f"lelyon_enriched_{datetime.now().strftime('%Y%m%d_%H%M%S')}.xlsx" PROGRESS_FILE = DATA_DIR / "enrichment_progress.json" # Rate limiting DELAY_BETWEEN_REQUESTS = 3 # segundos MAX_RETRIES = 3 TIMEOUT = 60 # Schema para extração de contatos EXTRACT_SCHEMA = { "type": "object", "properties": { "emails": {"type": "array", "items": {"type": "string"}}, "phones": {"type": "array", "items": {"type": "string"}}, "whatsapps": {"type": "array", "items": {"type": "string"}}, "social_links": { "type": "object", "properties": { "linkedin": {"type": "string"}, "instagram": {"type": "string"}, "facebook": {"type": "string"}, "twitter": {"type": "string"}, "youtube": {"type": "string"} } }, "address": {"type": "string"}, "contact_names": {"type": "array", "items": {"type": "string"}}, "company_description": {"type": "string"} }, "required": ["emails", "phones", "whatsapps", "social_links", "address", "contact_names", "company_description"] } EXTRACT_PROMPT = """ Extraia TODAS as informações de contato desta empresa: - emails: todos os e-mails encontrados (contato, vendas, suporte, etc.) - phones: todos os telefones com DDD (fixo e celular) - whatsapps: números de WhatsApp (com ou sem link wa.me) - social_links: URLs de LinkedIn, Instagram, Facebook, Twitter/X, YouTube - address: endereço completo da empresa - contact_names: nomes de sócios, gerentes, responsáveis citados - company_description: breve descrição do que a empresa faz Ignore e-mails genéricos de provedores (gmail.com, hotmail.com, outlook.com) a menos que sejam o único contato. Priorize e-mails corporativos (@dominio.com.br). Retorne arrays vazios se não encontrar. """ headers = { "Authorization": f"Bearer {FIRECRAWL_API_KEY}", "Content-Type": "application/json" } session = requests.Session() session.headers.update(headers) # ============ HELPERS ============ def load_progress() -> Dict: """Carrega progresso salvo.""" if PROGRESS_FILE.exists(): with open(PROGRESS_FILE, "r", encoding="utf-8") as f: return json.load(f) return {"processed": [], "enriched": {}, "failed": [], "last_index": 0} def save_progress(progress: Dict): """Salva progresso incremental.""" with open(PROGRESS_FILE, "w", encoding="utf-8") as f: json.dump(progress, f, ensure_ascii=False, indent=2) def load_companies() -> List[Dict]: """Carrega empresas do JSON mais recente.""" if INPUT_JSON.exists(): with open(INPUT_JSON, "r", encoding="utf-8") as f: data = json.load(f) # Compatibilidade: pode ser dict com b2b/b2c ou lista direta if isinstance(data, dict): companies = data.get("b2b", []) + data.get("b2c", []) else: companies = data print(f"✅ Carregadas {len(companies)} empresas de {INPUT_JSON}") return companies # Fallback: Excel if INPUT_EXCEL.exists(): df = pd.read_excel(INPUT_EXCEL, sheet_name="Template_Enriquecimento") companies = df.to_dict("records") print(f"✅ Carregadas {len(companies)} empresas de {INPUT_EXCEL}") return companies raise FileNotFoundError("Nenhum arquivo de entrada encontrado") def guess_website(company: Dict) -> Optional[str]: """Tenta adivinhar o website da empresa.""" razao = company.get("razao_social", "").lower() fantasia = company.get("nome_fantasia", "").lower() cidade = company.get("cidade", "").lower() cnpj = company.get("cnpj", "").replace(".", "").replace("/", "").replace("-", "") # Limpar nome para busca nome_busca = fantasia or razao nome_busca = re.sub(r'[^\w\s]', '', nome_busca) nome_busca = re.sub(r'\b(ltda|me|epp|sa|eireli|mei)\b', '', nome_busca) nome_busca = ' '.join(nome_busca.split()[:4]) # primeiras 4 palavras # Estratégias de busca (ordem de prioridade) strategies = [ # 1. Busca direta por CNPJ em sites conhecidos f"https://cnpj.biz/{cnpj}", f"https://www.cnpj.io/{cnpj}", f"https://consultacnpj.com/cnpj/{cnpj}", # 2. Busca Google-style (seria via search API, mas Firecrawl não tem) # Vamos tentar padrões comuns de domínio ] # Gerar possíveis domínios baseados no nome slug = re.sub(r'[^a-z0-9]+', '', nome_busca.replace(' ', '')) if len(slug) > 3: for tld in ['.com.br', '.com', '.net.br', '.br']: strategies.append(f"https://{slug}{tld}") strategies.append(f"https://www.{slug}{tld}") # Adicionar cidade para desambiguação if cidade and len(slug) > 3: slug_city = slug + cidade.replace(' ', '').replace('-', '') for tld in ['.com.br', '.com']: strategies.append(f"https://{slug_city}{tld}") return strategies[:10] # Limitar tentativas def firecrawl_extract(url: str) -> Optional[Dict]: """Usa Firecrawl /extract para obter contatos estruturados.""" payload = { "urls": [url], "prompt": EXTRACT_PROMPT, "schema": EXTRACT_SCHEMA, "enableWebSearch": False } for attempt in range(MAX_RETRIES): try: resp = session.post(f"{FIRECRAWL_BASE}/extract", json=payload, timeout=TIMEOUT) if resp.status_code == 200: data = resp.json() if data.get("success") and data.get("data"): return data["data"] elif data.get("success") and not data.get("data"): return {} # Sucesso mas sem dados elif resp.status_code == 429: wait = 2 ** attempt * 5 print(f" ⏳ Rate limited, aguardando {wait}s...") time.sleep(wait) continue elif resp.status_code == 402: print(f" 💳 Créditos insuficientes ou plano excedido") return None else: print(f" ⚠️ Erro {resp.status_code}: {resp.text[:200]}") except requests.Timeout: print(f" ⏱️ Timeout na tentativa {attempt+1}") except Exception as e: print(f" ❌ Erro: {e}") if attempt < MAX_RETRIES - 1: time.sleep(2 ** attempt) return None def firecrawl_scrape(url: str) -> Optional[str]: """Scrape simples para obter HTML/Markdown (fallback).""" payload = { "url": url, "formats": ["markdown"], "onlyMainContent": True } try: resp = session.post(f"{FIRECRAWL_BASE}/scrape", json=payload, timeout=TIMEOUT) if resp.status_code == 200: data = resp.json() if data.get("success") and data.get("data", {}).get("markdown"): return data["data"]["markdown"] except Exception as e: print(f" ❌ Scrape erro: {e}") return None def extract_contacts_from_markdown(md: str) -> Dict: """Extrai contatos via regex do markdown (fallback se /extract falhar).""" result = { "emails": [], "phones": [], "whatsapps": [], "social_links": {}, "address": "", "contact_names": [], "company_description": "" } # Emails emails = re.findall(r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', md) # Filtrar genéricos corporate_emails = [e for e in emails if not any(p in e.lower() for p in ['gmail.com', 'hotmail.com', 'outlook.com', 'yahoo.com', 'icloud.com', 'bol.com.br', 'uol.com.br', 'terra.com.br'])] result["emails"] = list(dict.fromkeys(corporate_emails or emails))[:5] # Telefones brasileiros phones = re.findall(r'(\(?\d{2}\)?\s?)?(9?\d{4}[-\s]?\d{4})', md) clean_phones = [] for match in phones: ddd, num = match if ddd and num: clean = f"({ddd.strip('() ')}) {num.replace('-', '').replace(' ', '')}" clean_phones.append(clean) result["phones"] = list(dict.fromkeys(clean_phones))[:5] # WhatsApp wa_links = re.findall(r'(?:wa\.me|whatsapp\.com|api\.whatsapp\.com)/[\w\d]+', md, re.IGNORECASE) wa_nums = re.findall(r'(?:whatsapp|zap)[:\s]*(\+?55\s?\(?\d{2}\)?\s?9?\d{4}[-\s]?\d{4})', md, re.IGNORECASE) result["whatsapps"] = list(dict.fromkeys(wa_links + wa_nums))[:3] # Redes sociais social_patterns = { "linkedin": r'linkedin\.com/(?:in|company)/[\w\-]+', "instagram": r'instagram\.com/[\w\.\-]+', "facebook": r'facebook\.com/[\w\.\-]+', "twitter": r'(?:twitter|x)\.com/[\w\-]+', "youtube": r'youtube\.com/(?:c|channel|user)/[\w\-]+' } for platform, pattern in social_patterns.items(): matches = re.findall(pattern, md, re.IGNORECASE) if matches: result["social_links"][platform] = f"https://{matches[0]}" return result def merge_contacts(existing: Dict, new: Dict) -> Dict: """Mescla contatos novos com existentes, priorizando não-vazios.""" merged = existing.copy() for key in ["emails", "phones", "whatsapps", "contact_names"]: existing_vals = set(existing.get(key, [])) new_vals = set(new.get(key, [])) merged[key] = list(existing_vals | new_vals)[:5] for key in ["address", "company_description"]: if not merged.get(key) and new.get(key): merged[key] = new[key] if not merged.get("social_links"): merged["social_links"] = {} for platform, url in new.get("social_links", {}).items(): if url and not merged["social_links"].get(platform): merged["social_links"][platform] = url return merged def enrich_company(company: Dict) -> Dict: """Enriquece uma única empresa.""" cnpj = company.get("cnpj", "") nome = company.get("nome_fantasia") or company.get("razao_social", "") print(f"\n🔍 [{cnpj}] {nome}") # URLs para tentar urls = guess_website(company) enriched = { "emails": [], "phones": [], "whatsapps": [], "social_links": {}, "address": "", "contact_names": [], "company_description": "", "website_found": "", "website_status": "not_found", "enrichment_source": "firecrawl", "enrichment_date": datetime.now().isoformat() } for i, url in enumerate(urls): print(f" Tentativa {i+1}/{len(urls)}: {url}") # Tentar extract primeiro (mais estruturado) extracted = firecrawl_extract(url) if extracted is not None: enriched["website_found"] = url enriched["website_status"] = "active" enriched = merge_contacts(enriched, extracted) print(f" ✅ Extracted: {len(extracted.get('emails', []))} emails, {len(extracted.get('phones', []))} phones") break # Fallback: scrape + regex markdown = firecrawl_scrape(url) if markdown: enriched["website_found"] = url enriched["website_status"] = "active" regex_data = extract_contacts_from_markdown(markdown) enriched = merge_contacts(enriched, regex_data) print(f" 📄 Scraped + regex: {len(regex_data.get('emails', []))} emails, {len(regex_data.get('phones', []))} phones") break print(f" ❌ Falhou") time.sleep(DELAY_BETWEEN_REQUESTS) if not enriched["website_found"]: enriched["website_status"] = "not_found" print(f" 🚫 Nenhum site encontrado/acessível") # Atualizar empresa company.update({ "email": enriched["emails"][0] if enriched["emails"] else "", "telefone": enriched["phones"][0] if enriched["phones"] else "", "whatsapp": enriched["whatsapps"][0] if enriched["whatsapps"] else "", "website": enriched["website_found"] or "", "website_status": enriched["website_status"], "contato_nome": enriched["contact_names"][0] if enriched["contact_names"] else "", "linkedin": enriched["social_links"].get("linkedin", ""), "instagram": enriched["social_links"].get("instagram", ""), "facebook": enriched["social_links"].get("facebook", ""), "fonte_enriquecimento": "firecrawl", "data_enriquecimento": enriched["enrichment_date"], "observacoes": f"Emails: {len(enriched['emails'])} | Phones: {len(enriched['phones'])} | Site: {enriched['website_status']}" }) return company def save_results(companies: List[Dict]): """Salva resultados em JSON, CSV e Excel.""" timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') # JSON json_file = DATA_DIR / f"lelyon_enriched_{timestamp}.json" with open(json_file, "w", encoding="utf-8") as f: json.dump(companies, f, ensure_ascii=False, indent=2) # CSV csv_file = DATA_DIR / f"lelyon_enriched_{timestamp}.csv" fieldnames = [ "cnpj", "razao_social", "nome_fantasia", "cidade", "bairro", "cnae", "cnae_desc", "situacao", "porte", "capital_social", "data_abertura", "tipo_prospeccao", "b2c_categoria", "best_fit_service", "fit_score", "lelyon_fit", "email", "telefone", "whatsapp", "website", "website_status", "contato_nome", "contato_cargo", "linkedin", "instagram", "facebook", "fonte_enriquecimento", "data_enriquecimento", "observacoes" ] with open(csv_file, "w", encoding="utf-8-sig", newline="") as f: writer = csv.DictWriter(f, fieldnames=fieldnames) writer.writeheader() for c in companies: row = {k: c.get(k, "") for k in fieldnames} row["lelyon_fit"] = json.dumps(c.get("lelyon_fit", {}), ensure_ascii=False) writer.writerow(row) # Excel com abas excel_file = DATA_DIR / f"lelyon_enriched_{timestamp}.xlsx" with pd.ExcelWriter(excel_file, engine="openpyxl") as writer: df_all = pd.DataFrame(companies) df_all["lelyon_fit_str"] = df_all["lelyon_fit"].apply(lambda x: json.dumps(x, ensure_ascii=False) if isinstance(x, dict) else str(x)) df_b2b = df_all[df_all["tipo_prospeccao"] == "B2B"] df_b2c = df_all[df_all["tipo_prospeccao"] == "B2C"] df_all.to_excel(writer, sheet_name="Todos_Enriquecidos", index=False) df_b2b.to_excel(writer, sheet_name="B2B_Enriquecidos", index=False) df_b2c.to_excel(writer, sheet_name="B2C_Enriquecidos", index=False) # Resumo enriquecimento enriched_count = len([c for c in companies if c.get("email") or c.get("telefone") or c.get("website")]) stats = [ {"Métrica": "Total Empresas", "Valor": len(companies)}, {"Métrica": "Com Email", "Valor": len([c for c in companies if c.get("email")])}, {"Métrica": "Com Telefone", "Valor": len([c for c in companies if c.get("telefone")])}, {"Métrica": "Com WhatsApp", "Valor": len([c for c in companies if c.get("whatsapp")])}, {"Métrica": "Com Website", "Valor": len([c for c in companies if c.get("website")])}, {"Métrica": "Total Enriquecidas", "Valor": enriched_count}, {"Métrica": "Taxa Sucesso", "Valor": f"{enriched_count/len(companies)*100:.1f}%"}, ] pd.DataFrame(stats).to_excel(writer, sheet_name="Resumo_Enriquecimento", index=False) print(f"\n💾 Resultados salvos:") print(f" 📄 JSON: {json_file}") print(f" 📊 CSV: {csv_file}") print(f" 📈 Excel: {excel_file}") # Atualizar symlink latest for src, dst_name in [ (json_file, "lelyon_enriched_latest.json"), (csv_file, "lelyon_enriched_latest.csv"), (excel_file, "lelyon_enriched_latest.xlsx"), ]: dst = DATA_DIR / dst_name if dst.exists(): dst.unlink() dst.symlink_to(src.name) # ============ MAIN ============ def main(): print("=" * 60) print("FIRECRAWL ENRIQUECIMENTO - LELYON Prospects") print("=" * 60) print(f"API Key: {FIRECRAWL_API_KEY[:12]}...") print(f"Delay: {DELAY_BETWEEN_REQUESTS}s | Timeout: {TIMEOUT}s") print() # Carregar empresas companies = load_companies() if not companies: print("❌ Nenhuma empresa para enriquecer") return # Carregar progresso progress = load_progress() processed_cnpjs = set(progress.get("processed", [])) enriched_data = progress.get("enriched", {}) # Filtrar apenas não processadas pending = [c for c in companies if c.get("cnpj") not in processed_cnpjs] print(f"📋 Total: {len(companies)} | Já processadas: {len(processed_cnpjs)} | Pendentes: {len(pending)}") if not pending: print("✅ Todas já processadas!") # Reconstruir lista completa com dados enriquecidos final_companies = [] for c in companies: cnpj = c.get("cnpj") if cnpj in enriched_data: final_companies.append(enriched_data[cnpj]) else: final_companies.append(c) save_results(final_companies) return # Processar print(f"\n🚀 Iniciando enriquecimento de {len(pending)} empresas...") print(" (Ctrl+C para pausar - progresso é salvo automaticamente)\n") enriched_count = 0 failed_count = 0 try: for idx, company in enumerate(pending): cnpj = company.get("cnpj", "") # Enriquecer enriched = enrich_company(company) # Salvar progresso enriched_data[cnpj] = enriched processed_cnpjs.add(cnpj) progress["processed"] = list(processed_cnpjs) progress["enriched"] = enriched_data progress["last_index"] = idx save_progress(progress) if enriched.get("email") or enriched.get("telefone") or enriched.get("website"): enriched_count += 1 else: failed_count += 1 # Rate limiting if idx < len(pending) - 1: time.sleep(DELAY_BETWEEN_REQUESTS) except KeyboardInterrupt: print("\n\n⏸️ Interrompido pelo usuário. Progresso salvo.") # Reconstruir lista final completa print("\n📦 Consolidando resultados...") final_companies = [] for c in companies: cnpj = c.get("cnpj") if cnpj in enriched_data: final_companies.append(enriched_data[cnpj]) else: final_companies.append(c) save_results(final_companies) # Stats finais total_enriched = len([c for c in final_companies if c.get("email") or c.get("telefone") or c.get("website")]) print(f"\n✅ CONCLUÍDO!") print(f" Total processadas nesta sessão: {len(pending)}") print(f" Enriquecidas com sucesso: {enriched_count}") print(f" Sem dados encontrados: {failed_count}") print(f" Total enriquecidas (geral): {total_enriched}/{len(final_companies)} ({total_enriched/len(final_companies)*100:.1f}%)") # Limpar progresso se completou tudo if len(processed_cnpjs) >= len(companies): PROGRESS_FILE.unlink(missing_ok=True) print(" 🧹 Progresso limpo (todas processadas)") if __name__ == "__main__": main()