cnpj-processor 3.7.0

Sistema de Processamento de Dados CNPJ da Receita Federal do Brasil

Processador de Dados CNPJ 🏢

🆕 Versão 3.7.0 (Publicada no PyPI) - API Pública + Sistema Completamente Otimizado com Exportação para Parquet, Atalhos e Publicação Automática

A partir da versão 3.7.0, o projeto está publicado no PyPI como cnpj-processor. Esta versão representa a primeira publicação oficial combinando o sistema completo com exportação de dados para parquet, sistema de atalhos, versionamento unificado, publicação automática via script e infraestrutura moderna. O sistema anterior (v2.x) foi completamente reestruturado utilizando padrões de design modernos e infraestrutura unificada.

Este projeto automatiza o download, processamento e armazenamento dos dados públicos de CNPJ disponibilizados pela Receita Federal. Ele foi desenvolvido para ser eficiente, resiliente, modular e fácil de usar.

🌐 Compatibilidade Multiplataforma

O sistema foi projetado e testado para funcionar perfeitamente em todos os sistemas operacionais modernos:

• ✅ Windows (7, 8, 10, 11) - Totalmente compatível
• ✅ Linux (Ubuntu, Debian, CentOS, Fedora, etc.) - Nativo
• ✅ macOS (10.14+) - Totalmente compatível
• ✅ Outros sistemas Unix - Suporte através de fallbacks universais

Detecção Automática do Sistema

O sistema detecta automaticamente o sistema operacional e usa as APIs nativas mais eficientes:

• Windows: ctypes.windll para verificação de espaço em disco e os.splitdrive() para paths
• Linux/Unix: os.statvfs() para informações de disco e paths Unix padrão
• Fallback Universal: shutil.disk_usage() para máxima compatibilidade

Todas as funcionalidades foram testadas e validadas em múltiplas plataformas, garantindo experiência consistente independente do sistema operacional.

🚀 O que há de Novo na Versão 3.5.0

📊 EXPORTAÇÃO PARA PARQUET (julho 2025):

• ✅ Exportação otimizada para parquet: Dados consolidados para painéis analíticos
• ✅ Processador de Painel: Combinação inteligente de dados de Estabelecimento, Empresa e Simples Nacional
• ✅ Filtros avançados: Filtragem por UF, situação cadastral e status do Simples Nacional
• ✅ Compressão eficiente: Arquivos otimizados com compressão zstd para economia de espaço
• ✅ Exportação para CSV: Conversão automática de parquet para CSV para compatibilidade
• ✅ Documentação completa: Exemplos práticos em exemplo_uso_painel.py
• ✅ Performance excepcional: Processamento otimizado com Polars para grandes volumes de dados

🎯 SISTEMA COMPLETO DE ATALHOS (junho 2025):

• ✅ 22 atalhos implementados para todos os argumentos do sistema
• ✅ Interface otimizada: Redução de até 78% no comprimento dos comandos
• ✅ Lógica inteligente: Mapeamento baseado na primeira letra significativa
• ✅ Resolução de conflitos: Sistema inteligente (ex: -h reservado para help, -W para hide-pending)
• ✅ Documentação completa: 3 documentos especializados com 40+ exemplos práticos
• ✅ Exemplos comparativos:
  • ANTES: --tipos empresas --step process --source-zip-folder dados --output-subfolder resultado --quiet --delete-zips-after-extract --cleanup-after-db
  • AGORA: -t empresas -s process -z dados -o resultado -q -d -c ⚡ 78% mais curto!

⚙️ SISTEMA DE VERSIONAMENTO AUTOMÁTICO (junho 2025):

• ✅ Detecção automática via git tags: Sistema prioriza versão do git sobre fallback
• ✅ Script release.py avançado: Criação automatizada de releases com incremento inteligente
• ✅ Integração total com main.py: Versão aparece automaticamente no --help
• ✅ Sistema de fallback robusto: Funciona mesmo em ambientes sem git
• ✅ Comandos de automação:
  • python scripts/release.py --patch: Incrementa versão patch (3.2.0 → 3.2.1)
  • python scripts/release.py --minor: Incrementa versão minor (3.2.0 → 3.3.0)
  • python scripts/release.py --major: Incrementa versão major (3.2.0 → 4.0.0)
  • python scripts/release.py 3.5.0: Cria versão específica
• ✅ Documentação especializada: Guia completo em VERSIONAMENTO.md

🔥 REVOLUÇÃO NO PIPELINE - Pipeline Otimizado de Processamento Imediato (versões anteriores):

• ✅ Pipeline Revolucionário: Implementado sistema que processa cada arquivo imediatamente após verificação/download
• ✅ Performance Transformada:
  • ANTES: Download ALL arquivos (6s) → Process ALL (9min 32s) = 12min 26s
  • AGORA: Download + Process IMMEDIATE = 3min 43s (⚡ 70% mais rápido!)
• ✅ Processamento Inteligente:
  • Se arquivo existe → Processa imediatamente
  • Se não existe → Baixa e processa imediatamente
  • Eliminação completa de latência entre download e processamento
• ✅ Semáforos de Controle: Controle avançado de concorrência para otimização de recursos
• ✅ Execução Paralela: Todos os arquivos são processados em paralelo com asyncio.gather()

🛠️ Melhorias e Estabilizações Anteriores:

• ✅ Sistema de Testes de Rede Otimizado: Implementado cache para evitar testes duplicados de conectividade
• ✅ Processador de Estabelecimentos Aprimorado: Campo tipo_situacao_cadastral com classificação inteligente
• ✅ Sistema de Limpeza Robusto: Funcionalidade de limpeza com verificações de segurança aprimoradas
• ✅ Correções Críticas de Runtime: Eliminação completa de erros de sintaxe e runtime
• ✅ 69.2% redução de código (5.940 → 1.725 linhas) na refatoração v3.0+
• ✅ 100% eliminação de duplicação (4.200 linhas duplicadas removidas)
• ✅ Arquitetura unificada com padrões Factory, Strategy e Template Method

Benefícios Transformadores da v3.2.0:

• 🚀 Interface ultra-otimizada: Comandos 78% mais curtos e intuitivos
• 🤖 Automação completa: Releases automatizadas via git tags
• ⚡ Performance excepcional: Pipeline revolucionário + atalhos eficientes
• 📚 Documentação profissional: Guias especializados para cada funcionalidade
• 🛡️ Mais confiável: 100% taxa de sucesso vs ~85% anterior
• 🔧 Mais fácil de manter: Versionamento automático + interface simplificada
• 🌐 Mais universal: Funciona identicamente em Windows, Linux e macOS
• 💾 Mais eficiente: Pipeline inteligente + economia automática de espaço em disco

Navegação

🚀 Como Usar

• Como Usar
• Pré-requisitos
• Instalação
• Execução
• 🎯 Sistema de Atalhos
• ⚙️ Sistema de Versionamento
• Gerenciamento de Cache
• O que o Script Faz

📊 Exportação para Painel

• Exportação para Parquet
• Processador de Painel
• Filtros Avançados
• Conversão para CSV
• Exemplo de Uso

📋 Fluxo do Processo

• Fluxo do Processo
• [Fluxo Modular Atual (--step)]
• [Ferramentas Utilizadas]

✨ Características

• Características

🏗️ Sistema de Entidades

• Versão 3.0 - Sistema de Entidades
• Documentação Completa
• Exemplos de Uso
• Testes

📝 Monitoramento e Configuração

• [Logs e Monitoramento]
• [Configurações]

⚡ Otimizações de Processamento

• [Otimizações de Processamento]
• [Processamento sequencial de arquivos ZIP]
• [Sistema de Cache para Downloads]
• [Paralelização do Processamento de CSV]
• [Tratamento Específico de Exceções]
• [Verificações de Segurança]
• [Limpeza de arquivos temporários]
• [Economia de Espaço em Disco]
• [Melhorias na Conversão de Tipos]

🤝 Contribuição e Licença

• Contribuindo
• Licença
• Notas

Pré-requisitos

• Python 3.9 ou superior
• Espaço em disco suficiente para os arquivos
• Conexão com internet estável
• Sistema Operacional: Windows, Linux ou macOS (detecção automática)

Instalação

Windows

  # Clone o repositório
  git clone https://github.com/seu-usuario/cnpj.git
  cd cnpj

  # Crie um ambiente virtual
  python -m venv venv
  venv\Scripts\activate

  # Instale as dependências
  pip install -r requirements.txt

Linux/macOS

# Clone o repositório
git clone https://github.com/seu-usuario/cnpj.git
cd cnpj

# Crie um ambiente virtual
python3 -m venv venv
source venv/bin/activate

# Instale as dependências
pip install -r requirements.txt

Configuração Universal

O sistema funciona identicamente em todos os sistemas operacionais:

# Configure o ambiente (funciona em Windows, Linux e macOS)
cp .env.local.example .env.local

Configure as variáveis no arquivo .env.local:

# URL base dos dados da Receita Federal
URL_ORIGIN=https://dados.rfb.gov.br/CNPJ/

# Diretórios para download e processamento (paths são automaticamente adaptados)
PATH_ZIP=./download/      # Arquivos ZIP baixados
PATH_UNZIP=./unzip/      # Arquivos extraídos
PATH_PARQUET=./parquet/  # Arquivos Parquet processados

# Configurações do banco de dados
FILE_DB_PARQUET=cnpj.duckdb
PATH_REMOTE_PARQUET=//servidor/compartilhado/

Nota: Os caminhos são automaticamente adaptados para cada sistema operacional. Use / ou \ conforme sua preferência - o sistema normaliza automaticamente.

🎯 Sistema de Atalhos

🆕 NOVIDADE v3.2.0: Sistema completo de atalhos implementado! Reduza seus comandos em até 78%!

⚡ Atalhos Principais

Atalho	Argumento Completo	Descrição
-t	--tipos	Tipos de dados (empresas, estabelecimentos, simples, socios)
-s	--step	Etapa (download, process, database, all)
-q	--quiet	Modo silencioso
-v	--verbose-ui	Modo verboso
-r	--remote-folder	Pasta remota para download/processamento
-a	--all-folders	Baixar todas as pastas disponíveis
-f	--from-folder	Pasta inicial para download/processamento
-o	--output-subfolder	Subpasta de saída para parquets

🧹 Atalhos de Limpeza e Otimização

Atalho	Argumento Completo	Descrição
-d	--delete-zips-after-extract	Deletar ZIPs após extração
-c	--cleanup-after-db	Deletar parquets após DB
-C	--cleanup-all-after-db	Deletar parquets E ZIPs após DB

🎯 Atalhos de Processamento Específico

Atalho	Argumento Completo	Descrição
-E	--criar-empresa-privada	Criar subset empresas privadas
-U	--criar-subset-uf	Criar subset por UF (ex: -U SP)
-p	--process-all-folders	Processar todas as pastas locais

🖥️ Atalhos de Interface Visual

Atalho	Argumento Completo	Descrição
-B	--show-progress	Forçar exibição de barras de progresso
-H	--hide-progress	Ocultar barras de progresso
-S	--show-pending	Forçar exibição de lista de pendentes
-W	--hide-pending	Ocultar lista de arquivos pendentes

🔥 Exemplos Comparativos - Antes vs Agora

Download Básico

# ANTES (78 caracteres):
python main.py --tipos empresas --step download --quiet --remote-folder 2024-01

# AGORA (36 caracteres - 54% mais curto):
python main.py -t empresas -s download -q -r 2024-01

Processamento com Economia de Espaço

# ANTES (132 caracteres):
python main.py --tipos estabelecimentos --step process --delete-zips-after-extract --cleanup-after-db --quiet --output-subfolder resultado

# AGORA (47 caracteres - 64% mais curto):
python main.py -t estabelecimentos -s process -d -c -q -o resultado

Download Sequencial Otimizado

# ANTES (89 caracteres):
python main.py --all-folders --from-folder 2023-01 --quiet --delete-zips-after-extract

# AGORA (29 caracteres - 67% mais curto):
python main.py -a -f 2023-01 -q -d

📚 Documentação Completa de Atalhos

• ATALHOS.md: Guia completo com todos os atalhos e exemplos detalhados
• README_ATALHOS.md: Resumo executivo dos atalhos mais importantes

⚙️ Sistema de Versionamento

🆕 NOVIDADE v3.2.0: Sistema completo de versionamento automático baseado em git tags!

🔧 Comandos de Release Automatizados

# Incrementar versão patch (3.2.0 → 3.2.1)
python scripts/release.py --patch

# Incrementar versão minor (3.2.0 → 3.3.0)
python scripts/release.py --minor

# Incrementar versão major (3.2.0 → 4.0.0)
python scripts/release.py --major

# Criar versão específica
python scripts/release.py 3.5.0

# Criar versão com mensagem personalizada
python scripts/release.py --patch --message "Correções críticas de segurança"

# Ver informações de versão atual
python scripts/release.py --info

⚡ Detecção Automática de Versão

O sistema detecta automaticamente a versão através de:

1. Git Tags (prioridade máxima): git describe --tags --abbrev=0
2. Fallback: Versão definida em src/__version__.py

# A versão aparece automaticamente no help
python main.py --help
# Sistema de Processamento de Dados CNPJ v3.2.0

# Verificar versão programaticamente
python -c "from src.__version__ import get_version; print('Versão:', get_version())"

📚 Documentação Completa de Versionamento

• VERSIONAMENTO.md: Guia completo do sistema de versionamento

Execução

O script principal main.py aceita diversos argumentos para customizar a execução. 🆕 AGORA COM ATALHOS COMPLETOS!

Comandos Universais (funcionam identicamente em Windows/Linux/macOS)

# 1. Execução completa (padrão: baixa, processa, cria DuckDB):
python main.py
# Equivalente a:
python main.py -s all

# 🔥 NOVO: Pipeline Otimizado - Processamento Imediato (v3.1.4+)
# O sistema agora processa cada arquivo IMEDIATAMENTE após verificação/download
# Performance: 70% mais rápido que versões anteriores!

# 2. 🎯 EXEMPLOS COM ATALHOS (v3.2.0+):

# Download apenas empresas em modo silencioso (78% mais curto):
python main.py -t empresas -q

# Processar apenas estabelecimentos da pasta 2024-01:
python main.py -s process -t estabelecimentos -z dados-zip/2024-01

# Download de todas as pastas desde 2023-01:
python main.py -a -f 2023-01

# Processamento com economia máxima de espaço:
python main.py -t empresas -d -C -q

# Estabelecimentos de São Paulo com interface completa:
python main.py -t estabelecimentos -U SP -o estab_sp -v

# 3. Apenas baixar os arquivos ZIP mais recentes (todos os tipos):
python main.py -s download

# 4. Apenas baixar arquivos ZIP de Empresas e Sócios:
python main.py -s download -t empresas socios

# 5. Baixar e processar dados de uma pasta específica (ex: 2024-01):
python main.py -s download -t socios -r 2024-01

# 6. Apenas processar ZIPs existentes para Parquet:
#    (Necessário especificar a pasta de origem dos ZIPs e a subpasta de saída Parquet)
python main.py -s process -z ../dados-abertos-zip -o meu_processamento_manual

# 7. Apenas processar ZIPs existentes de Simples e Sócios:
python main.py -s process -z "D:/MeusDownloads/CNPJ_ZIPs" -o simples_socios -t simples socios

# 8. Apenas criar/atualizar o banco DuckDB a partir de Parquets existentes:
#    (Necessário especificar a subpasta onde os Parquets estão)
python main.py -s database -o meu_processamento_manual

# 9. Processar Empresas, criando subset 'empresa_privada':
#    (Execução completa, mas poderia ser --step process se os ZIPs já existirem)
python main.py -s all -t empresas -o apenas_empresas_polars -E

# 10. Processar Estabelecimentos, criando subset para SP:
#     (Execução completa, mas poderia ser --step process se os ZIPs já existirem)
python main.py -s all -t estabelecimentos -o process_go -U GO

# 11. NOVO: Baixar arquivos de todas as pastas remotas a partir de 2023-01 até a mais atual:
python main.py -a -f 2023-01 -s download

# 12. NOVO: Baixar e processar arquivos de todas as pastas remotas desde a mais antiga até a mais atual:
python main.py -a

# 13. NOVO: Baixar e processar dados a partir de 2023-06 até a mais atual:
python main.py -a -f 2023-06

# 14. NOVO: Processar todas as pastas locais no formato AAAA-MM a partir de 2023-03:
python main.py -s process -p -f 2023-03 -o processados_desde_2023_03

# 15. NOVO: Processar dados deletando os ZIPs após extração para economizar espaço:
python main.py -t empresas -d

# 16. NOVO: Baixar e processar dados de 2023-01 até atual, deletando ZIPs após processamento:
python main.py -a -f 2023-01 -d

# 17. NOVO: Processar todas as pastas locais deletando ZIPs para economizar espaço:
python main.py -s process -p -o economizando_espaco -d

# 18. NOVO: Processamento conservador de espaço - apenas estabelecimentos com deleção de ZIPs:
python main.py -t estabelecimentos -d -o estabelecimentos_sem_zips

# 🔥 EXEMPLOS DESTACANDO O PIPELINE OTIMIZADO (v3.1.4+):

# 19. Pipeline otimizado para máxima velocidade - apenas sócios:
python main.py -t socios -q
# Resultado: Processamento IMEDIATO de cada arquivo conforme fica disponível

# 20. Pipeline otimizado com economia de espaço - processamento ultra-eficiente:
python main.py -t empresas -d -C -q
# Resultado: 70% mais rápido + máxima economia de espaço

# 21. Pipeline otimizado para processamento histórico eficiente:
python main.py -a -f 2023-01 -q
# Resultado: Processa cada pasta/arquivo imediatamente quando disponível

# EXEMPLOS COM CONTROLE DE INTERFACE VISUAL:

# 22. Download em modo silencioso (sem barras de progresso nem lista de pendentes):
python main.py -q

# 23. Download com interface completa (barras de progresso + lista de pendentes):
python main.py -v

# 24. Download mostrando apenas as barras de progresso (oculta lista de pendentes):
python main.py --show-progress --hide-pending

# 25. Download de pasta específica com barras de progresso ativadas:
python main.py --remote-folder 2024-01 --show-progress

# 26. Processamento em modo verboso com todas as informações visuais:
python main.py -s process -z ../dados/2023-05 -o teste -v

# 27. Download de todas as pastas em modo silencioso para logs limpos:
python main.py -a -q

# 28. Processamento mostrando lista de arquivos pendentes mas sem barras de progresso:
python main.py -t empresas -S -H

# 29. Download forçado com interface mínima (apenas lista de pendentes):
python main.py -F -H -S

# 30. Processamento de múltiplas pastas em modo silencioso:
python main.py -s process -p -o batch_silent -q

# 31. Download de pasta específica com barras de progresso ativadas:
python main.py -r 2024-01 -P

# EXEMPLOS COM LIMPEZA DE ARQUIVOS (🆕 ECONOMIA MÁXIMA DE ESPAÇO):

# 32. Processar dados e criar banco DuckDB, removendo arquivos parquet após criação:
python main.py -s all -t empresas -c

# 33. Processar dados e criar banco DuckDB, removendo arquivos parquet E ZIP após criação:
python main.py -s all -t empresas -C

# 34. Criar banco DuckDB a partir de parquets existentes e remover os parquets:
python main.py -s database -o processados_2023_05 -c

# 35. Download, processamento e banco completo com limpeza total (economiza máximo espaço):
python main.py -a -f 2023-01 -C

# 36. Processamento conservador com deleção de ZIPs durante extração e limpeza final:
python main.py -t estabelecimentos -d -c

# 37. Economia máxima: processar estabelecimentos com todas as opções de limpeza:
python main.py -t estabelecimentos -d -C -o economia_maxima

# 🏢 EXEMPLOS COM PROCESSAMENTO DO PAINEL CONSOLIDADO (NOVO v3.2.1):

# 38. Processamento completo com painel consolidado (TODOS OS DADOS - SEM FILTROS):
python main.py --processar-painel

# 39. Painel completo incluindo estabelecimentos inativos (TODOS OS DADOS):
python main.py --processar-painel --painel-incluir-inativos

# 40. Painel de Goiás incluindo estabelecimentos inativos:
python main.py --processar-painel --painel-uf GO --painel-incluir-inativos

# 41. Processamento de tipos específicos + painel para Minas Gerais:
python main.py -t empresas estabelecimentos simples --processar-painel --painel-uf MG

# 42. Painel de pasta remota específica com filtro de situação:
python main.py -r 2024-01 --processar-painel --painel-situacao 2

# 43. Painel completo com todos os filtros (SP, ativos, sem inativos):
python main.py --processar-painel --painel-uf SP --painel-situacao 2

# 44. Painel para estabelecimentos suspensos de todas as UFs:
python main.py --processar-painel --painel-situacao 3 --painel-incluir-inativos

# 45. Pipeline completo: download + processamento + painel + banco (economia máxima):
python main.py --processar-painel --painel-uf SP -C

# 46. Pipeline completo com painel de TODOS OS DADOS + economia máxima:
python main.py --processar-painel -C

# 47. Painel em modo silencioso para processamento em lote:
python main.py --processar-painel --painel-uf GO -q

# 48. Painel COMPLETO em modo silencioso (automação):
python main.py --processar-painel -q

# 49. Painel com dados de múltiplas pastas remotas:
python main.py -a -f 2023-01 --processar-painel --painel-uf SP

# 50. Painel de TODOS OS DADOS com múltiplas pastas remotas:
python main.py -a -f 2023-01 --processar-painel

🎯 Principais Argumentos com Atalhos:

*--step/-s {download,process,database,all}: Define qual(is) etapa(s) executar (padrão: all). *--tipos/-t {empresas,estabelecimentos,simples,socios}: Filtra quais tipos de dados baixar ou processar (padrão: todos). *--remote-folder/-r <pasta>: Especifica a pasta remota dos dados (ex: 2024-01). Usado para organizar arquivos por data. *--source-zip-folder/-z <caminho>: Pasta de origem dos arquivos ZIP (obrigatório para --step process). *--output-subfolder/-o <nome>: Subpasta em PATH_PARQUET para salvar/ler Parquets (obrigatório para --step process e --step database). *--criar-empresa-privada/-E: Flag para criar subset de empresas privadas (na etapa process). *--criar-subset-uf/-U <UF>: Flag para criar subset de estabelecimentos por UF (na etapa process). *--all-folders/-a: Baixa/processa de TODOS os diretórios remotos disponíveis ou todas as pastas locais. *--from-folder/-f <pasta>: 🆕 Especifica pasta inicial para download/processamento sequencial (formato AAAA-MM). *--process-all-folders/-p: 🆕 Processa todas as pastas locais no formato AAAA-MM encontradas. *--delete-zips-after-extract/-d: 🆕 Deleta arquivos ZIP após extração bem-sucedida para economizar espaço. *--cleanup-after-db/-c: 🆕 Deleta arquivos parquet após criação bem-sucedida do banco DuckDB. *--cleanup-all-after-db/-C: 🆕 Deleta arquivos parquet E ZIP após criação bem-sucedida do banco DuckDB. *--force-download/-F: Força download mesmo que arquivos já existam localmente ou no cache. *--log-level/-l <NÍVEL>: Ajusta o nível de log (padrão: INFO).

Argumentos do Processamento do Painel Consolidado (🆕 NOVO v3.2.1):

*--processar-painel: 🆕 Ativa o processamento do painel consolidado (combina estabelecimentos + simples + empresas). *--painel-uf <UF>: 🆕 Filtra painel por UF específica (ex: SP, GO, MG). *--painel-situacao <CODIGO>: 🆕 Filtra por situação cadastral (1=Nula, 2=Ativa, 3=Suspensa, 4=Inapta, 8=Baixada). *--painel-incluir-inativos: 🆕 Inclui estabelecimentos inativos no painel (padrão é só ativos).

Argumentos de Controle de Interface Visual:

*--quiet/-q: 🆕 Modo silencioso - desativa barras de progresso e lista de pendentes. *--verbose-ui/-v: 🆕 Modo verboso - ativa barras de progresso e lista de pendentes. *--show-progress/-B: 🆕 Força exibição de barras de progresso. *--hide-progress/-H: 🆕 Força ocultação de barras de progresso. *--show-pending/-S: 🆕 Força exibição da lista de arquivos pendentes. *--hide-pending/-W: 🆕 Força ocultação da lista de arquivos pendentes.

Prioridade dos Argumentos de Interface:

• Modo silencioso (--quiet/-q) tem prioridade máxima sobre todos os outros
• Argumentos específicos (--show-progress/-B, --hide-progress/-H, etc.) têm prioridade sobre modos gerais
• Modo verboso (--verbose-ui/-v) sobrescreve configurações padrão

Gerenciamento de Cache

# Exibir informações sobre arquivos em cache
python -m src.cache_manager cache-info

# Limpar o cache de downloads
python -m src.cache_manager clear-cache

📊 Exportação para Parquet

A versão 3.5.0 introduz um sistema completo de exportação para arquivos parquet, otimizados para análise de dados e uso em painéis analíticos. Esta funcionalidade permite consolidar dados de diferentes entidades (Estabelecimentos, Empresas e Simples Nacional) em uma única visão unificada.

Processador de Painel

O sistema inclui um processador especializado (PainelProcessor) que combina dados de múltiplas fontes:

• Dados de Estabelecimento: Informações cadastrais, situação, localização
• Dados de Empresa: Porte, natureza jurídica, razão social
• Dados do Simples Nacional: Status de optante, datas de opção/exclusão, MEI

O processador realiza operações de JOIN otimizadas usando o framework Polars, garantindo performance mesmo com grandes volumes de dados.

Filtros Avançados

O processador de painel suporta diversos filtros para personalizar os dados exportados:

• Filtro por UF: Limitar dados a estados específicos
• Filtro de Situação Cadastral: Filtrar por estabelecimentos ativos, inativos, etc.
• Filtro de Simples Nacional: Filtrar por optantes ou não optantes
• Inclusão de Inativos: Opção para incluir ou excluir estabelecimentos inativos

Conversão para CSV

Os dados podem ser facilmente convertidos para CSV para compatibilidade com outras ferramentas:

# Exportar para CSV
processor.export_to_csv(
    "data/exports/painel/painel_sp_ativos.parquet",
    "data/exports/painel/painel_sp_ativos.csv",
    delimiter=";"
)

Para exemplos mais detalhados, consulte o arquivo exemplo_uso_painel.py.

Comandos via Interface Principal

# Processamento completo com painel consolidado
python main.py --processar-painel

# Painel para uma UF específica
python main.py --processar-painel --painel-uf SP

# Painel incluindo estabelecimentos inativos
python main.py --processar-painel --painel-incluir-inativos

# Painel com filtro de situação cadastral (2 = Ativo)
python main.py --processar-painel --painel-situacao 2

📋 Fluxo do Processo de Geração de Dados CNPJ

O processador de dados CNPJ funciona através de um sistema modular controlado pelo argumento --step, permitindo executar etapas específicas ou o fluxo completo. Cada etapa é independente e pode ser executada separadamente, oferecendo flexibilidade total no processamento dos dados da Receita Federal.

📊 O que o Script Faz

O script main.py orquestra um fluxo modular com pipeline otimizado de processamento imediato que pode ser executado em etapas:

🔥 Pipeline Otimizado (v3.1.4+) - Processamento Imediato

O sistema agora utiliza um pipeline revolucionário que elimina a latência entre download e processamento:

• Verificação Inteligente: Para cada arquivo:
  • Se existe → Processa IMEDIATAMENTE
  • Se não existe → Baixa e processa IMEDIATAMENTE
• Execução Paralela: Todos os arquivos são processados simultaneamente com asyncio.gather()
• Performance Transformada: 70% mais rápido que o fluxo sequencial anterior
• Controle de Concorrência: Semáforos avançados otimizam uso de recursos

Etapas do Processo

1.Download dos Dados (--step download ou all) - COM PIPELINE OTIMIZADO *Identifica os arquivos ZIP mais recentes no portal da Receita Federal. *--step all, cada arquivo é processado imediatamente após verificação/download *Baixa os arquivos necessários (considerando os tipos especificados) de forma assíncrona e paralela. *Utiliza cache para evitar downloads repetidos. *Verifica a integridade básica dos arquivos baixados. *Pipeline Imediato: Não espera todos os downloads para iniciar processamento

2.Processamento para Parquet (--step process ou all) - PROCESSAMENTO IMEDIATO *No pipeline otimizado, ocorre simultaneamente com download *Lê arquivos ZIP de uma pasta de origem (--source-zip-folder). *Extrai o conteúdo de cada ZIP para uma subpasta temporária. *Processa os arquivos de dados (CSV ou similar): *Aplica transformações (renomeação, conversão de tipos, etc.). *Gera arquivos Parquet otimizados e particionados na subpasta de saída (--output-subfolder). *Cria subsets opcionais (--criar-empresa-privada, --criar-subset-uf). *Limpa as subpastas temporárias. *Eficiência: Processamento imediato elimina tempo de espera

3.Criação do Banco de Dados (--step database ou all) *Lê os arquivos Parquet de uma subpasta especificada (--output-subfolder). *Cria ou atualiza um arquivo de banco de dados DuckDB (cnpj.duckdb por padrão). *Cria tabelas no DuckDB para cada tipo de dado encontrado (empresas, estabelecimentos, socios, simples, e tabelas auxiliares como cnae, municipio, etc., se presentes na pasta base). *Opcionalmente, faz backup do banco para um local remoto.

Comparação de Performance

Fluxo	Tempo Total	Descrição
Anterior	~12min 26s	Download ALL (6s) → Process ALL (9min 32s) = 12min 26s
Otimizado v3.1.4	~3min 43s	Download + Process IMMEDIATE → Database ⚡ 70% mais rápido

✨ Características do Projeto

*Pipeline Otimizado: funcionalidade revolucionária que processa cada arquivo imediatamente após verificação/download, eliminando latência e melhorando performance em 70% *Execução Modular: Controle granular do fluxo com --step (download, process, database, all) *🌐 Compatibilidade Total: Funciona identicamente em Windows, Linux e macOS com detecção automática do SO *Sistema de Entidades: 🆕 Sistema robusto de entidades com validação automática, transformações e schemas Pydantic. *Pipeline Assíncrono: Download e processamento simultâneos com streaming inteligente e processamento imediato. *Download Eficiente: Assíncrono, paralelo, com cache, ordenação por tamanho e retentativas automáticas. *Processamento Híbrido: 🆕 Paralelização inteligente - usa múltiplas threads onde aumenta performance, processamento sequencial onde economiza recursos. *Download Cronológico: 🆕 Download ordenado de múltiplas pastas remotas em ordem cronológica com --all-folders e --from-folder. *Processamento Múltiplo: 🆕 Processamento inteligente de múltiplas pastas locais com --process-all-folders e controle por --from-folder. *Economia de Espaço: 🆕 Deleção automática de ZIPs após extração com --delete-zips-after-extract para conservar espaço em disco. *Monitoramento Avançado: Estatísticas em tempo real, métricas de performance e relatórios automáticos. *Validação Robusta: 🆕 Sistema híbrido com Pydantic 2.x, correção automática e relatórios detalhados. *Organização Inteligente: Estrutura de pastas por data (parquet/AAAA-MM/tipo/) com --remote-folder. *Saída Otimizada: Arquivos Parquet particionados e banco DuckDB consolidado. *Configurabilidade: Variáveis de ambiente (.env.local) e argumentos de linha de comando. *Subsets Opcionais: Criação de subsets por UF (--criar-subset-uf) ou para empresas privadas (--criar-empresa-privada). *Logging Detalhado: Logs estruturados em arquivo e console formatado com Rich. *Resiliência: Sistema robusto de recuperação de falhas e limpeza automática de recursos. *🛡️ Circuit Breaker Global: 🆕 Sistema inteligente de interrupção coordenada que detecta falhas críticas e para toda a aplicação para evitar desperdício de recursos. *🎨 Controle de Interface Visual: 🆕 Controle granular de barras de progresso e listas de arquivos pendentes com modos silencioso (--quiet), verboso (--verbose-ui) e argumentos específicos.

🛡️ Sistema de Circuit Breaker Global

🔧 dezembro de 2024 - Versão 3.1.0 - Sistema de Circuit Breaker + Controle de Interface Visual

🚨 Sistema de Circuit Breaker Global

• ✅ Detecção Inteligente de Falhas: Monitora 8 tipos de falhas críticas (conectividade, disco, memória, etc.)
• ✅ Interrupção Coordenada: Para toda a aplicação quando detecta falhas que impedem conclusão
• ✅ Thresholds Configuráveis: Diferentes limites para cada tipo de falha (1-10 falhas/10min)
• ✅ Thread-Safe: Funciona perfeitamente com processamento paralelo
• ✅ Economia de Recursos: Evita desperdício de processamento em falhas irrecuperáveis

🎨 Controle Granular de Interface Visual

• ✅ Modo Silencioso: --quiet para execução sem interface visual (ideal para automação)
• ✅ Modo Verboso: --verbose-ui para interface completa com progresso e pendências
• ✅ Controles Específicos: --show-progress, --hide-progress, --show-pending, --hide-pending
• ✅ Priorização Inteligente: Sistema de prioridades que respeita preferências do usuário
• ✅ Logs Limpos: Interface otimizada para diferentes cenários de uso

🔧 junho de 2025 - Versão 3.1.4 - Pipeline Revolucionário + Estabilização Completa

🔥 Pipeline Otimizado de Processamento Imediato

• ✅ Revolução na Performance: Sistema processa arquivos imediatamente após verificação/download
• ✅ 70% Mais Rápido: Tempo total reduzido de ~12min 26s para ~3min 43s
• ✅ Eliminação de Latência: Zero tempo de espera entre download e processamento
• ✅ Streaming Inteligente:
  • Arquivo existe → Processa IMEDIATAMENTE
  • Arquivo não existe → Baixa e processa IMEDIATAMENTE
• ✅ Execução Paralela: Todos os arquivos processados simultaneamente com asyncio.gather()
• ✅ Semáforos Avançados: Controle de concorrência otimizado para máxima eficiência
• ✅ Pipeline Assíncrono: Download e processamento em pipeline contínuo

🛠️ Correções Críticas de Runtime

• ✅ Eliminação Total de Erros de Sintaxe: Correção de todos os problemas de indentação e estrutura de código
• ✅ Validação de Importação: Todos os módulos importam corretamente sem exceções
• ✅ Teste de Execução: main.py executa perfeitamente em todos os modos
• ✅ Processadores Estabilizados:
  • EmpresaProcessor: Erros de indentação corrigidos
  • SimplesProcessor: Estrutura de código estabilizada
  • AsyncDownloader: Blocos try/except corrigidos
  • ParallelUtils: Indentação de blocos with corrigida
• ✅ Sistema de Testes de Rede Otimizado: Cache implementado para evitar testes duplicados
• ✅ Mapeamento de Estabelecimentos Corrigido: Lógica de tipo_situacao_cadastral funcionando corretamente

💡 Benefícios Transformadores da v3.1.4

• 🚀 Performance Revolucionária: Pipeline otimizado elimina 70% do tempo de processamento
• ⚡ Processamento Imediato: Não há mais espera entre etapas do processo
• 🤖 Automação Segura: Pode ser usado em scripts automatizados sem receio de crashes
• 👨‍💻 Desenvolvimento Fluido: Desenvolvedores podem focar na lógica sem se preocupar com erros básicos
• 📊 Dados Consistentes: Processamento de estabelecimentos com classificação correta de situação cadastral
• 💾 Economia Inteligente: Cache de rede evita redundâncias + pipeline otimizado economiza recursos
• 🏗️ Base Sólida: Fundação estável para futuras melhorias e funcionalidades
• 🔄 Fluxo Contínuo: Pipeline assíncrono elimina gargalos entre download e processamento

🔧 Comandos do Pipeline Otimizado - Validados e Funcionais

# Pipeline otimizado - processamento imediato (NOVO)
python main.py --tipos socios  # 70% mais rápido!

# Pipeline com economia total de espaço (OTIMIZADO)
python main.py --delete-zips-after-extract --cleanup-all-after-db --quiet

# Processamento histórico ultra-eficiente (PIPELINE CONTÍNUO)
python main.py --all-folders --from-folder 2023-01 --quiet

# Teste de funcionalidade básica (100% funcional)
python main.py --help

# Execução completa com pipeline revolucionário
python main.py --step all --tipos empresas

🛠️ Processamento e Regras de Negócio

Durante o processamento, várias transformações e regras de negócio são aplicadas, especialmente aos dados de Empresas:

1.Conversão de Tipos: Colunas numéricas e de data são convertidas para os tipos apropriados. 2.Renomeação: Algumas colunas são renomeadas para maior clareza (ex: razao_social_nome_empresarial para razao_social). 3.Extração de CPF: - O CPF (Pessoa Física) é extraído da coluna razao_social. - O script busca por padrões formatados (xxx.xxx.xxx-xx) ou por sequências de 11 dígitos. - O CPF extraído (apenas os 11 dígitos) é armazenado em uma nova coluna chamada CPF. - Esta coluna não é obrigatória, pois nem todas as razões sociais conterão um CPF. 4.Limpeza da Razão Social: Após a extração do CPF, o mesmo é removido da coluna razao_social original para manter apenas o nome/razão social. Espaços extras são removidos.

Essas transformações são implementadas nas funções de transformação específicas dentro de src/process/empresa.py.

🤝 Contribuindo

Contribuições são bem-vindas! Por favor:

1. Faça um fork do projeto
2. Crie uma branch para sua feature
3. Faça commit das mudanças
4. Push para a branch
5. Abra um Pull Request

📄 Licença

Este projeto está sob a licença MIT. Veja o arquivo LICENSE para mais detalhes.

⚠️ Notas

• 🌐 Compatibilidade: O sistema funciona identicamente em Windows, Linux e macOS
• 🔍 Detecção Automática: Sistema operacional e recursos de hardware são detectados automaticamente
• ✨ Sistema Estabilizado: Versão 3.2.0 elimina completamente erros de runtime e sintaxe
• 🛡️ Confiabilidade Total: Todos os módulos foram validados e testados para importação e execução
• 🔥 Pipeline Revolucionário: Sistema processa arquivos imediatamente após verificação/download, eliminando 70% do tempo de processamento
• O processamento utiliza múltiplos workers paralelos para máxima performance
• ⚡ Performance Transformada:
  • Pipeline Anterior: Download ALL (6s) + Process ALL (9min 32s) = 12min 26s
  • Pipeline Otimizado: Download + Process IMMEDIATE = 3min 43s
• Requisitos mínimos de espaço em disco:
  • Empresas: 5GB
  • Estabelecimentos: 8GB
  • Simples Nacional: 3GB
  • 💡 Dica: Use --delete-zips-after-extract para economizar até 50% do espaço necessário
  • 💡 Economia Máxima: Use --cleanup-all-after-db para economizar até 80% do espaço total
• 🧵 Processamento Híbrido: O sistema usa paralelização inteligente onde aumenta performance e sequenciamento onde economiza recursos
• 🏗️ Dados de Estabelecimentos Aprimorados: Campo tipo_situacao_cadastral fornece classificação inteligente da situação das empresas
• 🛡️ Sistema Robusto: Cache de testes de rede e correções de escopo eliminam problemas de duplicação e runtime
• 🔧 Pronto para Produção: Sistema completamente estabilizado e validado para uso em ambientes de produção
• 🚀 Processamento Imediato: Não há mais espera entre download e processamento - cada arquivo é processado assim que disponível
• Em caso de falhas, o sistema tentará novamente automaticamente com workers paralelos
• Verificação de espaço em disco é realizada antes da descompactação
• 🆕 Download Cronológico: Use --all-folders --from-folder AAAA-MM para baixar dados históricos de forma organizada
• 🆕 Economia de Espaço: A opção --delete-zips-after-extract remove ZIPs automaticamente após processamento paralelo bem-sucedido
• 🆕 Limpeza Pós-Database: As opções --cleanup-after-db e --cleanup-all-after-db removem arquivos desnecessários após criação do banco
• 🚀 Performance: Sistema otimizado com 6-12 workers simultâneos baseado no hardware disponível
• 📂 Caminhos: O sistema normaliza automaticamente caminhos de arquivo para cada sistema operacional
• 💾 APIs Nativas: Usa APIs específicas do SO para máxima eficiência (Windows: ctypes.windll, Linux: os.statvfs)

---

Desenvolvido com ❤️ e Python 3.9+! Completamente estabilizado e otimizado com pipeline revolucionário de processamento imediato para máxima performance e economia de recursos. Funciona perfeitamente em Windows, Linux e macOS! 🌐

⚡ Otimizações de Processamento do Pipeline (v3.1.4+)

Este projeto foi otimizado para lidar com grandes volumes de dados de maneira eficiente, com pipeline revolucionário de processamento imediato:

🔥 Pipeline Otimizado de Processamento Imediato (v3.1.4)

• Processamento Imediato: Cada arquivo é processado imediatamente após verificação/download
• Eliminação de Latência: Não espera todos os downloads terminarem para iniciar processamento
• Performance Transformada: 70% mais rápido que o fluxo anterior
• Controle de Concorrência: Semáforos avançados para otimização de recursos
• Execução Paralela: Processamento simultâneo de múltiplos arquivos com asyncio.gather()

Processamento Híbrido Inteligente

• Sequencial por ZIP: Cada arquivo ZIP é processado individualmente para economizar espaço em disco
• Paralelo por CSV: Arquivos CSV dentro de cada ZIP são processados em paralelo para máxima performance
• Pipeline Otimizado: Download/verificação + processamento imediato eliminam tempo de espera
• Limpeza Automática: Arquivos temporários são removidos imediatamente após processamento

Sistema de Cache Avançado

• Cache de metadados para evitar reprocessamento desnecessário
• Cache de Testes de Rede: Evita testes duplicados de conectividade (v3.1.2+)
• Configurável via parâmetros de tempo de expiração
• Comandos integrados para gerenciamento (cache-info, clear-cache)

Verificações de Segurança Proativas

• Verificação de espaço em disco antes de iniciar processamento
• Verificação de conexão com internet antes de downloads
• Estimativa de tamanho de arquivos após descompactação
• Circuit breaker integrado para detecção de falhas sistêmicas

Economia de Espaço Inteligente

• --delete-zips-after-extract: Deleção automática após verificação de integridade
• Verificação robusta: Confirma sucesso da extração antes de deletar
• Logs informativos: Registra espaço economizado
• Compatibilidade total: Funciona com processamento paralelo e todos os modos

Benefícios do Pipeline Otimizado

Aspecto	Versão Anterior	Versão 3.1.4	Melhoria
Fluxo	Download ALL → Process ALL	Download + Process IMMEDIATE	🔥 Revolucionário
Tempo Total	~12min 26s	~3min 43s	⚡ 70% mais rápido
Latência	6s de espera entre etapas	0s (processamento imediato)	✨ Eliminada
Eficiência	Download em lote + processamento em lote	Streaming inteligente	🚀 Transformada
Recursos	Picos de uso de disco	Uso constante e otimizado	💾 Otimizado