CNPJ Data Pipeline

Baixa e processa dados de empresas brasileiras da Receita Federal para PostgreSQL.
Parte do cnpj.chat — dados públicos de empresas, acessíveis para todos.

[!IMPORTANT] Desde v1.3.2 — A Receita Federal migrou os arquivos CNPJ para um novo repositório Nextcloud. Esta versão já suporta a nova URL e realiza downloads via WebDAV automaticamente. Nenhuma configuração adicional necessária.

[!TIP] Novo — Estratégia de carga configurável. Use LOADING_STRATEGY=replace para carga completa mais rápida (TRUNCATE + INSERT) ou upsert (default) para manter disponibilidade durante a carga.

Requisitos

• uv — gerencia pacotes e versões do Python. Substitui pip e virtualenv
• just — roda os comandos do projeto. Substitui Makefile
• Docker — roda o PostgreSQL sem instalar banco local
• Python 3.11+ — uv instala automaticamente se necessário

Instalação

macOS (Homebrew):

brew install uv just

Linux / Windows: veja a instalação do uv e do just.

uv e just são binários prontos — não precisa de Rust ou compilação.

Início Rápido

cp .env.example .env
just up      # Iniciar PostgreSQL
just run     # Executar pipeline

Via Docker

Imagem pronta publicada a cada release no GitHub Container Registry. Não precisa clonar o repositório.

# Listar meses disponíveis
docker run --rm ghcr.io/caiopizzol/cnpj-data-pipeline --list

# Processar um mês em um Postgres seu
docker run --rm \
  -e DATABASE_URL=postgres://user:pass@host:5432/cnpj \
  ghcr.io/caiopizzol/cnpj-data-pipeline --month 2024-11

# Exportar para Parquet (sem banco)
docker run --rm \
  -e OUTPUT_FORMAT=parquet \
  -v $(pwd)/parquet:/app/parquet \
  ghcr.io/caiopizzol/cnpj-data-pipeline

O schema é aplicado automaticamente na primeira execução. Para rodar com o Postgres do projeto, use docker compose run --rm pipeline <args>.

Comandos

just install # Instalar dependências
just up      # Iniciar PostgreSQL
just down    # Parar PostgreSQL
just db      # Entrar no banco (psql)
just run     # Executar pipeline
just reset   # Limpar e reiniciar banco
just lint    # Verificar código
just format  # Formatar código
just test    # Rodar testes
just check   # Rodar todos (lint, format, test)

Uso

just run                          # Processar mês mais recente
just run --list                   # Listar meses disponíveis
just run --month 2024-11          # Processar mês específico
just run --month 2024-11 --force  # Forçar reprocessamento

Configuração

DATABASE_URL=postgres://postgres:postgres@localhost:5435/cnpj
BATCH_SIZE=500000
TEMP_DIR=./temp
DOWNLOAD_WORKERS=4
RETRY_ATTEMPTS=3
RETRY_DELAY=5
CONNECT_TIMEOUT=30
READ_TIMEOUT=300
KEEP_DOWNLOADED_FILES=false
LOADING_STRATEGY=upsert  # "upsert" ou "replace"
OUTPUT_FORMAT=postgres   # "postgres" ou "parquet"
PARQUET_OUTPUT_DIR=./parquet
PARQUET_TYPED_OUTPUT=false  # Quando true, datas e numéricos saem tipados (Date, Float64, Int32)
PROCESS_WORKERS=1        # Arquivos do mesmo grupo em paralelo (ex: 4)

DATABASE_URL: parâmetros libpq

O DATABASE_URL é repassado direto ao driver, então qualquer parâmetro suportado pelo libpq funciona via query string. Útil para Postgres gerenciado (Railway, RDS, Supabase, Neon) e para isolar o pipeline em um schema próprio.

Sempre coloque o valor entre aspas no shell: o & da query string é metacaractere e, sem aspas, faz o bash colocar o comando em background.

# SSL obrigatório
DATABASE_URL='postgres://user:pass@host:5432/cnpj?sslmode=require'

# Pipeline em schema separado
DATABASE_URL='postgres://user:pass@host:5432/cnpj?options=-c%20search_path%3Dcnpj'

# Combinado
DATABASE_URL='postgres://user:pass@host:5432/cnpj?sslmode=require&options=-c%20search_path%3Dcnpj'

Sobre search_path: o schema precisa existir antes (CREATE SCHEMA cnpj;), o libpq não cria. Se você incluir public como fallback (search_path=cnpj,public) e o public já tiver tabelas do pipeline de uma execução anterior, o bootstrap (ensure_schema) detecta as tabelas no public e não cria nada no cnpj. Para isolamento estrito, deixe só cnpj no search_path, ou use um banco novo.

Estratégia de carga (PostgreSQL)

Formato de saída

Parquet

Com OUTPUT_FORMAT=parquet, o pipeline exporta direto para arquivos Parquet com compressão ZSTD. Sem necessidade de PostgreSQL.

OUTPUT_FORMAT=parquet just run

Saída (~6GB a partir de ~85GB de CSVs):

parquet/
  cnaes.parquet
  motivos.parquet
  municipios.parquet
  naturezas_juridicas.parquet
  paises.parquet
  qualificacoes_socios.parquet
  empresas.parquet
  estabelecimentos.parquet
  socios.parquet
  dados_simples.parquet
  manifest.json

Consulte com DuckDB:

SELECT * FROM 'parquet/empresas.parquet' WHERE cnpj_basico = '00000000';
SELECT COUNT(*) FROM 'parquet/estabelecimentos.parquet' WHERE uf = 'SP';

Schema

Documentação completa: docs/data-schema.md

Sobre normalização e tabelas derivadas: docs/post-processing.md · docs/data-audit.md

EMPRESAS (1) ─── (N) ESTABELECIMENTOS
         ├─── (N) SOCIOS
         └─── (1) DADOS_SIMPLES

[!IMPORTANT] Se você já carregou dados antes da mudança para socios.socio_id, recrie socios e as receitas derivadas (socios_quality_flags, socios_clean) para preservar sócios com CPF mascarado ou documento ausente.

Veja o passo a passo em docs/upgrading.md.

Fonte de Dados

Estes dados são públicos e oficiais, disponibilizados pela própria Receita Federal do Brasil.

O conjunto de dados CNPJ no Portal de Dados Abertos contém três recursos:

A Coordenação-Geral de Gestão de Cadastros e Benefícios Fiscais (Cocad) classifica estes dados como dados públicos, de livre acesso a qualquer interessado (NT 47/2024, item 10). CPFs de sócios são mascarados conforme art. 198 da Lei 5.172/1966 (CTN).

Estes dados não são vazados, obtidos ilegalmente, ou protegidos por sigilo fiscal.

Contribuidores