#!/usr/bin/env python3 """ Baixa e processa o dump mensal oficial CNPJ da Receita Federal (dados.gov.br) Filtra: UF=SP, 8 cidades Grande ABC, CNAEs alvo B2B/B2C Gera base real para prospecção LELYON """ import requests import zipfile import csv import io import os import time from pathlib import Path from datetime import datetime from typing import List, Dict, Set, Optional import pandas as pd # ============ CONFIG ============ DATA_DIR = Path("/root/lelyon_prospecting/data") DATA_DIR.mkdir(exist_ok=True) # 8 cidades do Grande ABC + SP (códigos IBGE) TARGET_CITIES = { "3550308": "São Paulo", "3547000": "São Bernardo do Campo", "3548701": "Santo André", "3548800": "São Caetano do Sul", "3513606": "Diadema", "3529305": "Mauá", "3542607": "Ribeirão Pires", "3543506": "Rio Grande da Serra", } # CNAEs alvo B2B (serviços LELYON) B2B_CNAES = { "6201501", "6201502", "6202300", "6203100", "6204000", "6209100", "6311900", "6312200", "6201500", "6202301", "7311400", "7319002", "7319003", "7319004", "7319099", "7320300", "7020400", "8211300", "8299799", "7319000", "4791302", "4791303", "4789099", "8211300", "8299799" } # CNAEs alvo B2C (varejo/serviços locais) B2C_CNAES = { "4711302", "4712100", "4721102", "4722100", "4723700", "4730201", "4741300", "4742201", "4744001", "4744002", "4751201", "4752100", "4753100", "4754701", "4755501", "4761001", "4762100", "4763800", "4771701", "4772500", "4781400", "4782201", "4783100", "4784900", "4789099", "4791302", "4791303", "5611201", "5611203", "5612100", "5620101", "5630900", "9602501", "9602502", "9609202", "9609299", "9601000", "9603300", "9609201", "9609299" } ALL_TARGET_CNAES = B2B_CNAES | B2C_CNAES # URLs conhecidas do dump RF (padrão mensal) # O dump fica em: https://dadosabertos.rfb.gov.br/CNPJ/ ou mirrors DUMP_URLS = [ "https://dadosabertos.rfb.gov.br/CNPJ/DADOS_ABERTOS_CNPJ_202410.zip", # exemplo "https://receita.economia.gov.br/orientacao/tributaria/cadastros/cadastro-nacional-de-pessoas-juridicas-cnpj/dados-publicos-cnpj", ] # ============ HELPERS ============ def download_file(url: str, dest: Path, chunk_size: int = 8192) -> bool: """Download com progresso.""" try: print(f"⬇️ Baixando: {url}") resp = requests.get(url, stream=True, timeout=60) resp.raise_for_status() total = int(resp.headers.get('content-length', 0)) downloaded = 0 with open(dest, 'wb') as f: for chunk in resp.iter_content(chunk_size=chunk_size): if chunk: f.write(chunk) downloaded += len(chunk) if total > 0: pct = downloaded / total * 100 print(f"\r Progresso: {pct:.1f}% ({downloaded/1024/1024:.1f}/{total/1024/1024:.1f} MB)", end="") print() return True except Exception as e: print(f"\n❌ Erro no download: {e}") return False def find_latest_dump_url() -> Optional[str]: """Tenta encontrar a URL do dump mais recente.""" # URLs padrão conhecidas - tenta várias base_urls = [ "https://dadosabertos.rfb.gov.br/CNPJ/", "https://dadosabertos.rfb.gov.br/dataset/cnpj", ] for base in base_urls: try: resp = requests.get(base, timeout=30) if resp.status_code == 200: # Procura links .zip import re zips = re.findall(r'href=["\']([^"\']*\.zip)["\']', resp.text) for zip_url in zips: if "DADOS_ABERTOS_CNPJ" in zip_url.upper() or "cnpj" in zip_url.lower(): if not zip_url.startswith("http"): from urllib.parse import urljoin zip_url = urljoin(base, zip_url) print(f"🔗 Encontrado dump: {zip_url}") return zip_url except Exception as e: print(f"⚠️ Erro ao buscar em {base}: {e}") return None def process_zip(zip_path: Path) -> Dict: """Processa o ZIP do dump RF e extrai empresas alvo.""" print(f"📦 Processando: {zip_path}") companies_b2b = [] companies_b2c = [] stats = {"total": 0, "matched": 0, "b2b": 0, "b2c": 0, "errors": 0} with zipfile.ZipFile(zip_path, 'r') as z: # Lista arquivos no ZIP csv_files = [f for f in z.namelist() if f.endswith('.csv')] print(f" Arquivos CSV no ZIP: {len(csv_files)}") # O arquivo de empresas costuma ser o maior for csv_name in csv_files: if "EMPRECSV" in csv_name.upper() or "EMPRESA" in csv_name.upper(): print(f" 📄 Processando: {csv_name}") with z.open(csv_name) as f: # Decodifica como latin-1 (padrão RF) text = io.TextIOWrapper(f, encoding='latin-1') reader = csv.DictReader(text, delimiter=';') for row in reader: stats["total"] += 1 # Filtrar UF=SP uf = row.get("uf", "").strip().upper() if uf != "SP": continue # Filtrar município (código IBGE) municipio = row.get("municipio", "").strip() if municipio not in TARGET_CITIES: continue # Filtrar CNAE principal cnae = row.get("cnae_fiscal", "").strip() if not cnae or cnae not in ALL_TARGET_CNAES: continue stats["matched"] += 1 # Classificar B2B/B2C category = "B2B" if cnae in B2B_CNAES else "B2C" if category == "B2B": stats["b2b"] += 1 else: stats["b2c"] += 1 # Construir registro company = { "cnpj": format_cnpj(row.get("cnpj_basico", "") + row.get("cnpj_ordem", "") + row.get("cnpj_dv", "")), "razao_social": row.get("razao_social", "").strip(), "nome_fantasia": row.get("nome_fantasia", "").strip(), "cidade": TARGET_CITIES.get(municipio, municipio), "bairro": row.get("bairro", "").strip(), "cnae": cnae, "cnae_desc": row.get("cnae_fiscal_descricao", "").strip(), "situacao": map_situacao(row.get("situacao_cadastral", "")), "porte": map_porte(row.get("porte", "")), "capital_social": parse_capital(row.get("capital_social", "")), "data_abertura": row.get("data_inicio_atividade", "").strip(), "tipo_prospeccao": category, "b2c_categoria": "N/A" if category == "B2B" else guess_b2c_category(cnae), "best_fit_service": map_cnae_to_service(cnae), "fit_score": calculate_fit_score(cnae, category), "lelyon_fit": calculate_lelyon_fit(cnae), "email": "", # RF não tem email "telefone": format_phone_rf(row.get("ddd_telefone_1", ""), row.get("telefone_1", "")), "whatsapp": "", "website": "", "website_status": "pending", "contato_nome": "", "contato_cargo": "", "linkedin": "", "instagram": "", "facebook": "", "fonte_enriquecimento": "rf_dump", "data_enriquecimento": datetime.now().isoformat(), "observacoes": f"RF Dump Oficial | CNAE: {cnae} | Cidade: {TARGET_CITIES.get(municipio)}", "natureza_juridica": row.get("natureza_juridica", ""), "simples_nacional": "Sim" if row.get("opcao_pelo_simples") == "S" else "Não", "data_situacao": row.get("data_situacao_cadastral", ""), "cnaes_secundarios": row.get("cnae_fiscal_secundaria", ""), } if category == "B2B": companies_b2b.append(company) else: companies_b2c.append(company) if stats["matched"] % 1000 == 0: print(f" 📊 Encontradas: {stats['matched']} (B2B: {stats['b2b']}, B2C: {stats['b2c']})") print(f"\n✅ Processamento concluído!") print(f" Total lidas: {stats['total']:,}") print(f" Match SP + Cidades + CNAE: {stats['matched']:,}") print(f" B2B: {stats['b2b']:,} | B2C: {stats['b2c']:,}") return {"b2b": companies_b2b, "b2c": companies_b2c, "stats": stats} def format_cnpj(basico: str, ordem: str, dv: str) -> str: cnpj = f"{basico}{ordem}{dv}" if len(cnpj) == 14: return f"{cnpj[:2]}.{cnpj[2:5]}.{cnpj[5:8]}/{cnpj[8:12]}-{cnpj[12:]}" return cnpj def map_situacao(code: str) -> str: mapa = {"01": "Ativa", "02": "Baixada", "03": "Inapta", "04": "Suspensa", "08": "Nula"} return mapa.get(code.strip(), code.strip()) def map_porte(code: str) -> str: mapa = {"01": "Não Informado", "02": "MEI", "03": "ME", "04": "EPP", "05": "Grande"} return mapa.get(code.strip(), code.strip()) def parse_capital(val: str) -> float: if not val: return 0.0 try: return float(val.replace(",", ".")) except: return 0.0 def format_phone_rf(ddd: str, telefone: str) -> str: ddd = ddd.strip() tel = telefone.strip() if ddd and tel: clean = tel.replace("-", "").replace(" ", "") return f"({ddd}) {clean[:4]}-{clean[4:]}" if len(clean) == 8 else f"({ddd}) {clean}" return "" def guess_b2c_category(cnae: str) -> str: categorias = { "47": "Varejo & Comércio", "56": "Alimentação & Bebidas", "96": "Beleza & Estética", "45": "Serviços Automotivos", "85": "Educação & Treinamento", "86": "Saúde & Bem-estar", "55": "Hospedagem & Turismo", "90": "Artes & Cultura", "93": "Lazer & Esportes", "97": "Serviços Domésticos" } return categorias.get(cnae[:2], "Outros") def map_cnae_to_service(cnae: str) -> str: mapa = { "6201501": "PWA/App Development", "6201502": "PWA/App Development", "6202300": "PWA/App Development", "6203100": "PWA/App Development", "6204000": "Agente IA/Automação", "6209100": "Agente IA/Automação", "6311900": "Agente IA/Automação", "6312200": "Agente IA/Automação", "4791302": "Loja WordPress/WooCommerce", "4791303": "Loja WordPress/WooCommerce", "4789099": "Loja WordPress/WooCommerce", "7020400": "CRM & Estoque QR-Code", "7311400": "Marketing Digital/Ads", "7319002": "Marketing Digital/Ads", "7319003": "Marketing Digital/Ads", "7319004": "Marketing Digital/Ads", "7319099": "Marketing Digital/Ads", "7320300": "Marketing Digital/Ads", "8211300": "CRM & Estoque QR-Code", "8299799": "CRM & Estoque QR-Code", "6209100": "Segurança + SEO", } return mapa.get(cnae, "N/A") def calculate_fit_score(cnae: str, category: str) -> int: if category == "B2B": high = {"6201501", "6201502", "6202300", "6203100", "6204000", "6209100", "6311900", "4791302", "7020400", "7311400"} med = {"6201500", "6202300", "6203100", "6311900", "6312200", "8211300", "8299799", "7311400", "7319002", "7319003"} if cnae in high: return 90 elif cnae in med: return 70 return 40 else: high = {"4791302", "5611201", "5611203", "9602501", "9602502", "4781400", "4771701", "4772500", "4711302"} med = {"4712100", "4721102", "4741300", "4751201", "4752100", "4753100", "4754701", "4782201", "4783100", "5620101"} if cnae in high: return 85 elif cnae in med: return 60 return 35 def calculate_lelyon_fit(cnae: str) -> Dict: fits = {"PWA/App Development": 0, "Agente IA/Automação": 0, "Loja WordPress/WooCommerce": 0, "CRM & Estoque QR-Code": 0, "Segurança + SEO": 0, "Marketing Digital/Ads": 0} cnae_fits = { "6201501": {"PWA/App Development": 95, "Agente IA/Automação": 80}, "6201502": {"PWA/App Development": 90, "Agente IA/Automação": 75}, "6202300": {"PWA/App Development": 85, "Agente IA/Automação": 70}, "6203100": {"PWA/App Development": 80, "Agente IA/Automação": 65}, "6204000": {"Agente IA/Automação": 95, "PWA/App Development": 70, "Segurança + SEO": 60}, "6209100": {"Agente IA/Automação": 85, "Segurança + SEO": 70, "CRM & Estoque QR-Code": 50}, "6311900": {"Agente IA/Automação": 80, "PWA/App Development": 60}, "4791302": {"Loja WordPress/WooCommerce": 95, "CRM & Estoque QR-Code": 70}, "7020400": {"CRM & Estoque QR-Code": 90, "Agente IA/Automação": 60}, "7311400": {"Marketing Digital/Ads": 95, "PWA/App Development": 50}, "7319002": {"Marketing Digital/Ads": 90}, "7319003": {"Marketing Digital/Ads": 85}, "7319004": {"Marketing Digital/Ads": 85}, "7319099": {"Marketing Digital/Ads": 80}, "7320300": {"Marketing Digital/Ads": 70, "Agente IA/Automação": 50}, "8211300": {"CRM & Estoque QR-Code": 85}, "8299799": {"CRM & Estoque QR-Code": 80}, "6209100": {"Segurança + SEO": 75, "Agente IA/Automação": 60}, "4791302": {"Loja WordPress/WooCommerce": 90, "Marketing Digital/Ads": 60}, "5611201": {"Loja WordPress/WooCommerce": 70, "CRM & Estoque QR-Code": 60}, "5611203": {"Loja WordPress/WooCommerce": 70, "CRM & Estoque QR-Code": 55}, "9602501": {"Loja WordPress/WooCommerce": 60, "Marketing Digital/Ads": 65}, "9602502": {"Loja WordPress/WooCommerce": 60, "Marketing Digital/Ads": 70}, "4781400": {"Loja WordPress/WooCommerce": 75, "Marketing Digital/Ads": 60}, "4771701": {"Loja WordPress/WooCommerce": 80, "CRM & Estoque QR-Code": 65}, "4772500": {"Loja WordPress/WooCommerce": 75, "Marketing Digital/Ads": 65}, "4711302": {"Loja WordPress/WooCommerce": 85, "CRM & Estoque QR-Code": 80}, } return cnae_fits.get(cnae, fits) # ============ MAIN ============ def main(): print("=" * 70) print("DOWNLOAD E PROCESSAMENTO DUMP RF CNPJ OFICIAL") print("=" * 70) print(f"Cidades alvo: {list(TARGET_CITIES.values())}") print(f"CNAEs B2B: {len(B2B_CNAES)} | B2C: {len(B2C_CNAES)}") print() # 1. Tentar encontrar URL do dump dump_url = find_latest_dump_url() if not dump_url: print("⚠️ Não foi possível encontrar URL do dump automaticamente.") print("\n📥 Baixando dump conhecido (Out/2024)...") dump_url = "https://dadosabertos.rfb.gov.br/CNPJ/DADOS_ABERTOS_CNPJ_202410.zip" # 2. Baixar zip_path = DATA_DIR / f"rf_dump_{datetime.now().strftime('%Y%m')}.zip" if not zip_path.exists(): if not download_file(dump_url, zip_path): print("❌ Falha no download. Tente manualmente:") print(f" wget {dump_url}") print(f" mv arquivo.zip {zip_path}") return else: print(f"✅ Dump já existe: {zip_path} ({zip_path.stat().st_size/1024/1024:.1f} MB)") # 3. Processar result = process_zip(zip_path) # 4. Salvar timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') combined = {"b2b": result["b2b"], "b2c": result["b2c"]} # JSON json_file = DATA_DIR / f"lelyon_real_rf_{timestamp}.json" with open(json_file, "w", encoding="utf-8") as f: json.dump(combined, f, ensure_ascii=False, indent=2) # CSV csv_file = DATA_DIR / f"lelyon_real_rf_{timestamp}.csv" all_companies = result["b2b"] + result["b2c"] if all_companies: fieldnames = list(all_companies[0].keys()) with open(csv_file, "w", encoding="utf-8-sig", newline="") as f: writer = csv.DictWriter(f, fieldnames=fieldnames) writer.writeheader() for c in all_companies: writer.writerow(c) # Excel excel_file = DATA_DIR / f"lelyon_real_rf_{timestamp}.xlsx" with pd.ExcelWriter(excel_file, engine="openpyxl") as writer: df_all = pd.DataFrame(all_companies) 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", index=False) df_b2b.to_excel(writer, sheet_name="B2B", index=False) df_b2c.to_excel(writer, sheet_name="B2C", index=False) stats_df = pd.DataFrame([result["stats"]]) stats_df.to_excel(writer, sheet_name="Estatisticas", index=False) # Symlinks for src, dst_name in [ (json_file, "lelyon_real_rf_latest.json"), (csv_file, "lelyon_real_rf_latest.csv"), (excel_file, "lelyon_real_rf_latest.xlsx"), ]: dst = DATA_DIR / dst_name if dst.exists(): dst.unlink() dst.symlink_to(src.name) print(f"\n💾 Arquivos salvos:") print(f" 📄 JSON: {json_file}") print(f" 📊 CSV: {csv_file}") print(f" 📈 Excel: {excel_file}") print(f" 🔗 Symlinks: *_latest.* atualizados") if __name__ == "__main__": main()