Base CNPJ Completa

https://arquivos.receitafederal.gov.br/dados/cnpj/dados_abertos_cnpj/Empresas0.zip
https://arquivos.receitafederal.gov.br/dados/cnpj/dados_abertos_cnpj/Empresas1.zip
...
https://arquivos.receitafederal.gov.br/dados/cnpj/dados_abertos_cnpj/Empresas9.zip

import requests
import zipfile
import pandas as pd
from io import BytesIO
from pathlib import Path
 
# Colunas do arquivo de Empresas conforme layout da Receita Federal
COLUNAS_EMPRESAS = [
    "cnpj_basico",
    "razao_social",
    "natureza_juridica",
    "qualificacao_responsavel",
    "capital_social",
    "porte_empresa",
    "ente_federativo_responsavel",
]
 
def baixar_e_extrair(url: str, destino: Path) -> Path:
    """Baixa um arquivo ZIP e extrai o CSV contido nele."""
    print(f"Baixando {url}...")
    response = requests.get(url, stream=True)
    response.raise_for_status()
 
    with zipfile.ZipFile(BytesIO(response.content)) as zf:
        nome_csv = zf.namelist()[0]
        zf.extract(nome_csv, destino)
        print(f"Extraído: {destino / nome_csv}")
        return destino / nome_csv
 
 
# Baixar o primeiro arquivo de empresas (Empresas0.zip)
url = "https://arquivos.receitafederal.gov.br/dados/cnpj/dados_abertos_cnpj/Empresas0.zip"
caminho_csv = baixar_e_extrair(url, Path("./dados_rfb"))
 
# Ler o CSV (sem cabeçalho, separador ";", encoding Latin-1)
df = pd.read_csv(
    caminho_csv,
    sep=";",
    header=None,
    names=COLUNAS_EMPRESAS,
    dtype=str,
    encoding="latin-1",
)
 
print(f"Registros carregados: {len(df):,}")
print(df.head())

def consultar_empresa(df: pd.DataFrame, cnpj_basico: str) -> pd.DataFrame:
    """
    Consulta dados de uma empresa pelo CNPJ básico (8 dígitos).
 
    Args:
        df: DataFrame com os dados de empresas
        cnpj_basico: 8 primeiros dígitos do CNPJ (ex: '00000000')
 
    Returns:
        DataFrame filtrado com os dados da empresa
    """
    return df[df["cnpj_basico"] == cnpj_basico.zfill(8)]
 
 
# Exemplo: consultar Banco do Brasil (CNPJ básico: 00000000)
resultado = consultar_empresa(df, "00000000")
if not resultado.empty:
    print(f"Razão Social: {resultado.iloc[0]['razao_social']}")
    print(f"Natureza Jurídica: {resultado.iloc[0]['natureza_juridica']}")
    print(f"Capital Social: R$ {resultado.iloc[0]['capital_social']}")

from pathlib import Path
 
def carregar_todas_empresas(diretorio: Path) -> pd.DataFrame:
    """
    Carrega todos os arquivos de empresas (Empresas0 a Empresas9) em um único DataFrame.
 
    Atenção: requer ~5 GB de RAM. Para máquinas com menos memória,
    considere usar Dask ou processar arquivo por arquivo.
    """
    dfs = []
    for i in range(10):
        url = f"https://arquivos.receitafederal.gov.br/dados/cnpj/dados_abertos_cnpj/Empresas{i}.zip"
        caminho = baixar_e_extrair(url, diretorio)
        df_parte = pd.read_csv(
            caminho,
            sep=";",
            header=None,
            names=COLUNAS_EMPRESAS,
            dtype=str,
            encoding="latin-1",
        )
        dfs.append(df_parte)
        print(f"Empresas{i}: {len(df_parte):,} registros")
 
    df_completo = pd.concat(dfs, ignore_index=True)
    print(f"\nTotal de empresas: {len(df_completo):,}")
    return df_completo

# Baixar e ler tabela de naturezas jurídicas
url_naturezas = "https://arquivos.receitafederal.gov.br/dados/cnpj/dados_abertos_cnpj/Naturezas.zip"
caminho_nat = baixar_e_extrair(url_naturezas, Path("./dados_rfb"))
 
df_naturezas = pd.read_csv(
    caminho_nat,
    sep=";",
    header=None,
    names=["codigo", "descricao"],
    dtype=str,
    encoding="latin-1",
)
 
# Juntar com dados de empresas para obter descrição da natureza jurídica
df_empresas = df.merge(
    df_naturezas,
    left_on="natureza_juridica",
    right_on="codigo",
    how="left",
    suffixes=("", "_natureza"),
)
 
print(df_empresas[["razao_social", "descricao"]].head(10))

import pandas as pd
import requests
 
API_KEY = "SEU_TOKEN_AQUI"
BASE_URL = "https://api.portaldatransparencia.gov.br/api-de-dados"
headers = {"chave-api-dados": API_KEY, "Accept": "application/json"}
 
# 1. Carregar base de empresas (previamente baixada)
df_empresas = pd.read_csv(
    "dados_rfb/Empresas0.csv",
    sep=";",
    header=None,
    names=["cnpj_basico", "razao_social", "natureza_juridica",
           "qualificacao_responsavel", "capital_social",
           "porte_empresa", "ente_federativo_responsavel"],
    dtype=str,
    encoding="latin-1",
)
 
# 2. Consultar contratos de uma empresa específica no Portal da Transparência
cnpj_completo = "00000000000191"  # Banco do Brasil
resp = requests.get(
    f"{BASE_URL}/contratos",
    headers=headers,
    params={"cnpjContratado": cnpj_completo, "pagina": 1},
)
contratos = pd.DataFrame(resp.json())
 
# 3. Enriquecer contratos com dados cadastrais da empresa
cnpj_basico = cnpj_completo[:8]
dados_empresa = df_empresas[df_empresas["cnpj_basico"] == cnpj_basico]
 
if not dados_empresa.empty and not contratos.empty:
    empresa = dados_empresa.iloc[0]
    print(f"Empresa: {empresa['razao_social']}")
    print(f"Natureza Jurídica: {empresa['natureza_juridica']}")
    print(f"Capital Social: R$ {empresa['capital_social']}")
    print(f"Contratos encontrados: {len(contratos)}")