cnpj-processor 4.0.5

Sistema de Processamento de Dados CNPJ da Receita Federal do Brasil

CNPJ Processor 🏢

Sistema profissional de processamento de dados públicos CNPJ da Receita Federal do Brasil

Automatize o download, processamento e análise dos dados públicos de CNPJ com performance excepcional e API simplificada.

🚀 Instalação

pip install cnpj-processor

⚡ Início Rápido

Via Linha de Comando (CLI)

# Pipeline completo (download + processamento + banco de dados)
cnpj-processor

# Download apenas
cnpj-processor --step download --tipos empresas estabelecimentos

# Processar painel consolidado por UF
cnpj-processor --step painel --painel-uf GO --painel-situacao 2

Via API Python

from cnpj_processor import CNPJProcessor

# Criar processador
processor = CNPJProcessor()

# Pipeline completo
success, folder = processor.run()

# Painel de empresas ativas em São Paulo
success, folder = processor.run(
    step='painel',
    painel_uf='GO',
    painel_situacao=2  # Ativas
)

🎯 Principais Funcionalidades

📥 Download Inteligente

• Download assíncrono de alta performance
• Retomada automática em caso de falha
• Verificação de integridade de arquivos
• Cache inteligente para evitar downloads duplicados

⚙️ Processamento Otimizado

• Pipeline paralelo: download e processamento simultâneos
• Até 70% mais rápido que processamento sequencial
• Suporte a múltiplos tipos: empresas, estabelecimentos, sócios, simples
• Exportação para Parquet com compressão eficiente

💾 Gestão Inteligente de Espaço

• Limpeza automática: Remove ZIPs e arquivos temporários por padrão
• Banco opcional: DuckDB criado apenas quando solicitado
• Múltiplas estratégias: De 15 GB (apenas banco) até 93 GB (tudo)
• Controle total: Flags para customizar retenção de artefatos

🎨 Painel Consolidado

• Combinação inteligente de dados de múltiplas fontes
• Filtros avançados: UF, situação cadastral, Simples Nacional
• Ideal para análises e dashboards
• Formato otimizado para BI tools

💾 Banco de Dados

• Geração automática de banco DuckDB
• Queries SQL de alta performance
• Integração perfeita com ferramentas de análise

📚 API Simplificada

A API do cnpj-processor foi projetada para ser simples, poderosa e intuitiva.

Métodos Principais

run() - Método Universal

Execute qualquer operação com um único método:

processor = CNPJProcessor()

# Pipeline completo
processor.run()

# Download específico
processor.run(step='download', tipos=['empresas'], remote_folder='2026-01')

# Processamento com economia de espaço
processor.run(
    step='all',
    delete_zips_after_extract=True,
    cleanup_all_after_db=True
)

# Painel customizado
processor.run(
    step='painel',
    painel_uf='GO',
    painel_situacao=2,
    output_subfolder='painel_go_ativas'
)

Parâmetros do run():

Parâmetro	Tipo	Descrição
step	str	Etapa: 'download', 'process', 'database', 'painel', 'all'
tipos	list	Tipos a processar: ['empresas', 'estabelecimentos', 'simples', 'socios']
remote_folder	str	Pasta remota (formato AAAA-MM)
output_subfolder	str	Subpasta de saída
force_download	bool	Forçar re-download
keep_artifacts	bool	Manter ZIPs e arquivos temporários (padrão: False)
create_database	bool	Criar banco DuckDB (padrão: False)
cleanup_after_db	bool	Remover parquets após criar banco
keep_parquet_after_db	bool	Manter parquets após criar banco
processar_painel	bool	Processar painel consolidado
painel_uf	str	Filtrar painel por UF
painel_situacao	int	Filtrar por situação (1=Nula, 2=Ativa, 3=Suspensa, 4=Inapta, 8=Baixada)
criar_empresa_privada	bool	Criar subset de empresas privadas
criar_subset_uf	str	Criar subset por UF
quiet	bool	Modo silencioso
log_level	str	Nível de log ('DEBUG', 'INFO', 'WARNING', 'ERROR')

get_latest_folder() - Consultar Pasta Mais Recente

processor = CNPJProcessor()
latest = processor.get_latest_folder()
print(f"Pasta mais recente: {latest}")  # '2026-01'

get_available_folders() - Listar Pastas Disponíveis

processor = CNPJProcessor()
folders = processor.get_available_folders()
print(f"Disponíveis: {folders}")  # ['2026-01', '2025-12', ...]

💡 Exemplos Práticos

Exemplo 1: Pipeline Completo

from cnpj_processor import CNPJProcessor

processor = CNPJProcessor()
success, folder = processor.run()

if success:
    print(f"✅ Dados processados em: {folder}")

Exemplo 2: Download Seletivo

# Baixar apenas empresas e estabelecimentos
processor = CNPJProcessor()
success, folder = processor.run(
    step='download',
    tipos=['empresas', 'estabelecimentos'],
    remote_folder='2026-01'
)

Exemplo 3: Processamento com Economia de Espaço

# Padrão: Remove ZIPs e temporários automaticamente, mantém apenas parquets
processor = CNPJProcessor()
success, folder = processor.run()  # ~20 GB

# Máxima economia: Criar banco e remover parquets
success, folder = processor.run(
    create_database=True,    # Cria banco DuckDB
    cleanup_after_db=True    # Remove parquets
)  # ~15 GB

Exemplo 4: Painel Analítico Customizado

# Painel apenas de empresas ativas de Goiás
processor = CNPJProcessor()
success, folder = processor.run(
    step='painel',
    painel_uf='GO',
    painel_situacao=2,  # Ativas
    output_subfolder='painel_go_ativas'
)

Exemplo 5: Processar Múltiplos Períodos

processor = CNPJProcessor()
pastas = ['2025-12', '2026-01']

for pasta in pastas:
    print(f"Processando {pasta}...")
    success, folder = processor.run(
        step='all',
        remote_folder=pasta,
        output_subfolder=f'dados_{pasta.replace("-", "_")}'
    )
    print(f"{'✅' if success else '❌'} {pasta}")

Exemplo 6: Subset Especializado

# Apenas empresas privadas
processor = CNPJProcessor()
success, folder = processor.run(
    step='all',
    tipos=['empresas'],
    criar_empresa_privada=True,
    output_subfolder='empresas_privadas'
)

# Apenas estabelecimentos de uma UF
success, folder = processor.run(
    step='all',
    tipos=['estabelecimentos'],
    criar_subset_uf='GO',
    output_subfolder='estabelecimentos_sp'
)

Exemplo 7: Estratégias de Espaço em Disco

processor = CNPJProcessor()

# Estratégia 1: Análise de dados (padrão)
success, folder = processor.run()
# Espaço: ~20 GB (apenas parquets)

# Estratégia 2: Com banco de dados
success, folder = processor.run(create_database=True)
# Espaço: ~35 GB (parquets + banco)

# Estratégia 3: Máxima economia
success, folder = processor.run(
    create_database=True,
    cleanup_after_db=True
)
# Espaço: ~15 GB (apenas banco)

# Estratégia 4: Manter tudo (desenvolvimento)
success, folder = processor.run(
    keep_artifacts=True,
    create_database=True,
    keep_parquet_after_db=True
)
# Espaço: ~93 GB (ZIPs + temporários + parquets + banco)

🔧 Uso via CLI

O cnpj-processor também oferece interface completa de linha de comando:

# Pipeline completo
cnpj-processor

# Download de pasta específica
cnpj-processor --step download --remote-folder 2026-01

# Processar apenas estabelecimentos
cnpj-processor --tipos estabelecimentos

# Painel filtrado
cnpj-processor --step painel --painel-uf GO --painel-situacao 2

# Criar banco de dados (opcional)
cnpj-processor --create-database

# Máxima economia de espaço
cnpj-processor --create-database --cleanup-after-db

# Ver pasta mais recente disponível
cnpj-processor --show-latest-folder

# Ver versão
cnpj-processor --version

# Ajuda completa
cnpj-processor --help

Atalhos de CLI

Interface otimizada com atalhos intuitivos:

# Equivalentes (forma completa vs. atalho)
cnpj-processor --tipos empresas --step download --remote-folder 2026-01
cnpj-processor -t empresas -s download -r 2026-01

# Criar banco com economia de espaço
cnpj-processor --create-database --cleanup-after-db --quiet
cnpj-processor -D -c -q

# Manter todos os artefatos
cnpj-processor --keep-artifacts --create-database --keep-parquet-after-db
cnpj-processor -k -D -K

# Painel filtrado
cnpj-processor --step painel --painel-uf GO --painel-situacao 2
cnpj-processor -s painel --painel-uf GO --painel-situacao 2

📊 Estrutura de Dados

Arquivos Gerados

parquet/
├── 2026-01/                    # Pasta por período
│   ├── empresa/               # Dados de empresas
│   ├── estabelecimento/       # Dados de estabelecimentos
│   ├── simples/              # Dados do Simples Nacional
│   ├── socio/                # Dados de sócios
│   ├── painel_dados.parquet  # Painel consolidado
│   └── cnpj.duckdb          # Banco de dados

Formato Painel

O painel consolidado combina dados de três fontes:

• Estabelecimento: CNPJ, razão social, endereço, situação
• Empresa: Nome fantasia, capital social, porte
• Simples: Opção pelo Simples Nacional, data de inclusão

Campos principais:

• cnpj_basico: CNPJ raiz (8 dígitos)
• cnpj_completo: CNPJ completo (14 dígitos)
• razao_social: Nome empresarial
• nome_fantasia: Nome fantasia
• uf: Unidade Federativa
• municipio: Município
• situacao_cadastral: Situação (Ativa, Baixada, etc.)
• opcao_simples: Se optante pelo Simples
• capital_social: Capital social da empresa
• porte: Porte da empresa

🎯 Casos de Uso

1. Análise de Mercado

# Obter painel de empresas ativas por estado
processor = CNPJProcessor()
success, folder = processor.run(
    step='painel',
    painel_uf='GO',
    painel_situacao=2
)

2. Compliance e Due Diligence

# Download completo para análise interna
processor = CNPJProcessor()
success, folder = processor.run(
    step='all',
    tipos=['empresas', 'estabelecimentos', 'socios']
)

3. Data Science / ML

# Preparar dados para modelos
processor = CNPJProcessor()
success, folder = processor.run(
    step='all',
    cleanup_after_db=True  # Mantém apenas banco final
)

4. Dashboards BI

# Gerar painel para PowerBI/Tableau
processor = CNPJProcessor()
success, folder = processor.run(
    step='painel',
    processar_painel=True
)

🔍 Requisitos do Sistema

• Python: 3.9 ou superior
• Sistema Operacional: Windows, Linux, macOS
• Espaço em Disco: Mínimo 50GB recomendado
• Memória RAM: Mínimo 4GB, recomendado 8GB+
• Conexão Internet: Necessária para download

🛡️ Tratamento de Erros

from cnpj_processor import CNPJProcessor

processor = CNPJProcessor()

try:
    success, folder = processor.run(
        step='all',
        tipos=['empresas']
    )
    
    if success:
        print(f"✅ Sucesso! Dados em: {folder}")
    else:
        print("⚠️ Concluído com avisos. Verifique os logs.")
        
except KeyboardInterrupt:
    print("\n🛑 Processamento interrompido pelo usuário")
except Exception as e:
    print(f"❌ Erro: {e}")

📈 Performance

Benchmarks

• Pipeline Otimizado: 70% mais rápido que processamento sequencial
• Download Assíncrono: Múltiplos arquivos simultâneos
• Processamento Paralelo: Utilização eficiente de múltiplos cores
• Compressão Inteligente: Arquivos Parquet com zstd

Tempos Típicos

Operação	Tempo Estimado
Download completo	5-15 minutos
Processamento (todos os tipos)	10-30 minutos
Geração de banco	2-5 minutos
Painel consolidado	5-10 minutos

Tempos variam conforme hardware e conexão de rede

🤝 Contribuindo

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

1. Fork o repositório
2. Crie uma branch para sua feature (git checkout -b feature/AmazingFeature)
3. Commit suas mudanças (git commit -m 'Add some AmazingFeature')
4. Push para a branch (git push origin feature/AmazingFeature)
5. Abra um Pull Request

📝 Licença

Este projeto está licenciado sob a Licença MIT - veja o arquivo LICENSE para detalhes.

🔗 Links Úteis

• PyPI: https://pypi.org/project/cnpj-processor/
• Documentação Completa: Ver pasta docs/ no repositório
• Issues: Reporte bugs e sugira melhorias
• Dados CNPJ: Receita Federal - Dados Públicos

🙏 Agradecimentos

• Receita Federal do Brasil pela disponibilização dos dados públicos
• Comunidade Python pelo ecossistema de ferramentas excepcionais
• Todos os contribuidores do projeto