Estabelecimentos

https://arquivos.receitafederal.gov.br/dados/cnpj/dados_abertos_cnpj/Estabelecimentos0.zip
https://arquivos.receitafederal.gov.br/dados/cnpj/dados_abertos_cnpj/Estabelecimentos1.zip
...
https://arquivos.receitafederal.gov.br/dados/cnpj/dados_abertos_cnpj/Estabelecimentos9.zip

import requests
import zipfile
import pandas as pd
from io import BytesIO
from pathlib import Path
 
# Colunas do arquivo de Estabelecimentos conforme layout da Receita Federal
COLUNAS_ESTABELECIMENTOS = [
    "cnpj_basico",
    "cnpj_ordem",
    "cnpj_dv",
    "identificador_matriz_filial",
    "nome_fantasia",
    "situacao_cadastral",
    "data_situacao_cadastral",
    "motivo_situacao_cadastral",
    "nome_cidade_exterior",
    "pais",
    "data_inicio_atividade",
    "cnae_fiscal",
    "cnae_fiscal_secundaria",
    "tipo_logradouro",
    "logradouro",
    "numero",
    "complemento",
    "bairro",
    "cep",
    "uf",
    "municipio",
    "ddd_1",
    "telefone_1",
    "ddd_2",
    "telefone_2",
    "ddd_fax",
    "fax",
    "correio_eletronico",
    "situacao_especial",
    "data_situacao_especial",
]
 
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 estabelecimentos
url = "https://arquivos.receitafederal.gov.br/dados/cnpj/dados_abertos_cnpj/Estabelecimentos0.zip"
caminho_csv = baixar_e_extrair(url, Path("./dados_rfb"))
 
# Ler o CSV (sem cabeçalho, separador ";", encoding Latin-1)
df_estab = pd.read_csv(
    caminho_csv,
    sep=";",
    header=None,
    names=COLUNAS_ESTABELECIMENTOS,
    dtype=str,
    encoding="latin-1",
)
 
print(f"Registros carregados: {len(df_estab):,}")
print(df_estab.head())

def montar_cnpj(row: pd.Series) -> str:
    """Monta o CNPJ completo (14 dígitos) a partir dos componentes."""
    return (
        row["cnpj_basico"].zfill(8)
        + row["cnpj_ordem"].zfill(4)
        + row["cnpj_dv"].zfill(2)
    )
 
 
# Adicionar coluna de CNPJ completo
df_estab["cnpj_completo"] = df_estab.apply(montar_cnpj, axis=1)
 
# Consultar estabelecimentos de uma empresa específica
cnpj_basico = "00000000"  # Banco do Brasil
estabelecimentos = df_estab[df_estab["cnpj_basico"] == cnpj_basico.zfill(8)]
 
print(f"Estabelecimentos encontrados: {len(estabelecimentos)}")
for _, est in estabelecimentos.head(5).iterrows():
    tipo = "Matriz" if est["identificador_matriz_filial"] == "1" else "Filial"
    print(f"  [{tipo}] CNPJ: {est['cnpj_completo']}")
    print(f"    {est['logradouro']}, {est['numero']} - {est['bairro']}")
    print(f"    {est['uf']} - CEP: {est['cep']}")
    print(f"    Tel: ({est['ddd_1']}) {est['telefone_1']}")
    print(f"    Email: {est['correio_eletronico']}")
    print()

def filtrar_por_uf_cnae(
    df: pd.DataFrame, uf: str, cnae: str, apenas_ativos: bool = True
) -> pd.DataFrame:
    """
    Filtra estabelecimentos por UF e CNAE fiscal.
 
    Args:
        df: DataFrame de estabelecimentos
        uf: Sigla da UF (ex: 'SP', 'RJ')
        cnae: Código CNAE fiscal (ex: '4711301' para comércio varejista)
        apenas_ativos: Se True, retorna apenas situação cadastral '02' (Ativo)
 
    Returns:
        DataFrame filtrado
    """
    mask = (df["uf"] == uf) & (df["cnae_fiscal"] == cnae)
    if apenas_ativos:
        mask = mask & (df["situacao_cadastral"] == "02")
    return df[mask]
 
 
# Exemplo: supermercados ativos em São Paulo
supermercados_sp = filtrar_por_uf_cnae(df_estab, "SP", "4711301")
print(f"Supermercados ativos em SP: {len(supermercados_sp):,}")

# Distribuição de estabelecimentos ativos por UF
ativos = df_estab[df_estab["situacao_cadastral"] == "02"]
distribuicao = (
    ativos.groupby("uf")
    .size()
    .reset_index(name="total")
    .sort_values("total", ascending=False)
)
 
print("Estabelecimentos ativos por UF (top 10):")
print(distribuicao.head(10).to_string(index=False))
 
# Distribuição por tipo (matriz vs filial)
tipo_map = {"1": "Matriz", "2": "Filial"}
ativos["tipo"] = ativos["identificador_matriz_filial"].map(tipo_map)
print(f"\nDistribuição Matriz/Filial:")
print(ativos["tipo"].value_counts())

# Baixar e ler tabela de CNAEs
url_cnaes = "https://arquivos.receitafederal.gov.br/dados/cnpj/dados_abertos_cnpj/Cnaes.zip"
caminho_cnaes = baixar_e_extrair(url_cnaes, Path("./dados_rfb"))
 
df_cnaes = pd.read_csv(
    caminho_cnaes,
    sep=";",
    header=None,
    names=["codigo", "descricao"],
    dtype=str,
    encoding="latin-1",
)
 
# Top 10 atividades econômicas mais frequentes
top_cnaes = (
    df_estab[df_estab["situacao_cadastral"] == "02"]
    .groupby("cnae_fiscal")
    .size()
    .reset_index(name="total")
    .sort_values("total", ascending=False)
    .head(10)
    .merge(df_cnaes, left_on="cnae_fiscal", right_on="codigo", how="left")
)
 
print("Top 10 atividades econômicas (CNAE):")
for _, row in top_cnaes.iterrows():
    print(f"  {row['cnae_fiscal']} - {row['descricao']}: {row['total']:,}")

CNPJ Completo = cnpj_basico (8) + cnpj_ordem (4) + cnpj_dv (2)
Exemplo:       00000000         + 0001         + 91
Resultado:     00000000000191

import pandas as pd
 
# Carregar dados (previamente baixados)
df_estab = pd.read_csv(
    "dados_rfb/Estabelecimentos0.csv",
    sep=";", header=None,
    names=["cnpj_basico", "cnpj_ordem", "cnpj_dv",
           "identificador_matriz_filial", "nome_fantasia",
           "situacao_cadastral", "data_situacao_cadastral",
           "motivo_situacao_cadastral", "nome_cidade_exterior",
           "pais", "data_inicio_atividade", "cnae_fiscal",
           "cnae_fiscal_secundaria", "tipo_logradouro", "logradouro",
           "numero", "complemento", "bairro", "cep", "uf", "municipio",
           "ddd_1", "telefone_1", "ddd_2", "telefone_2",
           "ddd_fax", "fax", "correio_eletronico",
           "situacao_especial", "data_situacao_especial"],
    dtype=str, encoding="latin-1",
)
 
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",
)
 
df_socios = pd.read_csv(
    "dados_rfb/Socios0.csv",
    sep=";", header=None,
    names=["cnpj_basico", "identificador_socio", "nome_socio_razao_social",
           "cpf_cnpj_socio", "qualificacao_socio", "data_entrada_sociedade",
           "pais", "representante_legal", "nome_representante",
           "qualificacao_representante_legal", "faixa_etaria"],
    dtype=str, encoding="latin-1",
)
 
# Filtrar matrizes ativas em uma UF
matrizes_sp = df_estab[
    (df_estab["uf"] == "SP") &
    (df_estab["identificador_matriz_filial"] == "1") &
    (df_estab["situacao_cadastral"] == "02")
].head(100)
 
# Enriquecer com dados cadastrais
matrizes_enriquecidas = matrizes_sp.merge(
    df_empresas[["cnpj_basico", "razao_social", "capital_social"]],
    on="cnpj_basico",
    how="left",
)
 
# Adicionar contagem de sócios
contagem_socios = (
    df_socios.groupby("cnpj_basico")
    .size()
    .reset_index(name="num_socios")
)
matrizes_enriquecidas = matrizes_enriquecidas.merge(
    contagem_socios, on="cnpj_basico", how="left"
)
 
print("Matrizes ativas em SP (com dados cadastrais e sócios):")
print(matrizes_enriquecidas[[
    "cnpj_basico", "razao_social", "cnae_fiscal",
    "bairro", "capital_social", "num_socios"
]].head(10))