QSA — Quadro Societário

https://arquivos.receitafederal.gov.br/dados/cnpj/dados_abertos_cnpj/Socios0.zip
https://arquivos.receitafederal.gov.br/dados/cnpj/dados_abertos_cnpj/Socios1.zip
...
https://arquivos.receitafederal.gov.br/dados/cnpj/dados_abertos_cnpj/Socios9.zip
 
# alternativa prática (espelho)
https://data.brasil.io/dataset/socios-brasil/socios.csv.gz

import requests
import zipfile
import pandas as pd
from io import BytesIO
from pathlib import Path
 
# Colunas do arquivo de Sócios conforme layout da Receita Federal
COLUNAS_SOCIOS = [
    "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",
]
 
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 sócios (Socios0.zip)
url = "https://arquivos.receitafederal.gov.br/dados/cnpj/dados_abertos_cnpj/Socios0.zip"
caminho_csv = baixar_e_extrair(url, Path("./dados_rfb"))
 
# Ler o CSV (sem cabeçalho, separador ";", encoding Latin-1)
df_socios = pd.read_csv(
    caminho_csv,
    sep=";",
    header=None,
    names=COLUNAS_SOCIOS,
    dtype=str,
    encoding="latin-1",
)
 
print(f"Registros carregados: {len(df_socios):,}")
print(df_socios.head())

import pandas as pd
 
url = "https://data.brasil.io/dataset/socios-brasil/socios.csv.gz"
 
df_socios = pd.read_csv(
    url,
    compression="gzip",
    dtype=str,
    low_memory=False,
)
 
print(f"Registros carregados: {len(df_socios):,}")
print(df_socios.columns.tolist())
print(df_socios.head())

def consultar_socios(df: pd.DataFrame, cnpj_basico: str) -> pd.DataFrame:
    """
    Retorna os sócios e administradores de uma empresa.
 
    Args:
        df: DataFrame com os dados de sócios
        cnpj_basico: 8 primeiros dígitos do CNPJ
 
    Returns:
        DataFrame com os sócios da empresa
    """
    return df[df["cnpj_basico"] == cnpj_basico.zfill(8)]
 
 
# Exemplo: consultar sócios de uma empresa
socios = consultar_socios(df_socios, "00000000")
for _, socio in socios.iterrows():
    tipo = "PJ" if socio["identificador_socio"] == "1" else "PF"
    print(f"  [{tipo}] {socio['nome_socio_razao_social']} "
          f"(Qualif: {socio['qualificacao_socio']}, "
          f"Entrada: {socio['data_entrada_sociedade']})")

def encontrar_empresas_por_socio(df: pd.DataFrame, nome_socio: str) -> pd.DataFrame:
    """
    Encontra todas as empresas em que uma pessoa aparece como sócia.
    Útil para análise de redes de relacionamento empresarial.
 
    Args:
        df: DataFrame com os dados de sócios
        nome_socio: Nome (ou parte do nome) do sócio
 
    Returns:
        DataFrame com as empresas encontradas
    """
    mask = df["nome_socio_razao_social"].str.contains(
        nome_socio, case=False, na=False
    )
    return df[mask][["cnpj_basico", "nome_socio_razao_social",
                      "qualificacao_socio", "data_entrada_sociedade"]]
 
 
# Encontrar todas as empresas de um determinado sócio
empresas = encontrar_empresas_por_socio(df_socios, "NOME DO SOCIO")
print(f"Empresas encontradas: {len(empresas)}")
print(empresas.head(10))

# Baixar tabela de qualificações
url_qualif = "https://arquivos.receitafederal.gov.br/dados/cnpj/dados_abertos_cnpj/Qualificacoes.zip"
caminho_qualif = baixar_e_extrair(url_qualif, Path("./dados_rfb"))
 
df_qualificacoes = pd.read_csv(
    caminho_qualif,
    sep=";",
    header=None,
    names=["codigo", "descricao"],
    dtype=str,
    encoding="latin-1",
)
 
# Juntar para obter descrição da qualificação
df_socios_decodificado = df_socios.merge(
    df_qualificacoes,
    left_on="qualificacao_socio",
    right_on="codigo",
    how="left",
    suffixes=("", "_qualif"),
)
 
print(df_socios_decodificado[
    ["cnpj_basico", "nome_socio_razao_social", "descricao"]
].head(10))

import pandas as pd
 
# Carregar dados de sócios (previamente baixados)
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",
)
 
# Lista de CNPJs de empresas contratadas pelo governo (obtida do Portal da Transparência)
cnpjs_contratados = ["12345678", "87654321", "11223344"]
 
# Filtrar sócios das empresas contratadas
socios_contratados = df_socios[
    df_socios["cnpj_basico"].isin(cnpjs_contratados)
]
 
# Encontrar sócios em comum (mesma pessoa em múltiplas empresas contratadas)
socios_duplicados = (
    socios_contratados
    .groupby("nome_socio_razao_social")["cnpj_basico"]
    .nunique()
    .reset_index()
    .rename(columns={"cnpj_basico": "num_empresas"})
)
 
socios_em_comum = socios_duplicados[socios_duplicados["num_empresas"] > 1]
print("Sócios presentes em múltiplas empresas contratadas:")
print(socios_em_comum.sort_values("num_empresas", ascending=False))