Como enriquecer uma planilha de CNPJs com o telefone e o e-mail dos sócios

empresa,cnpj
Exemplo Tecnologia,11.222.333/0001-81
Exemplo Comércio,44.555.666/0001-99

# No Windows (PowerShell)
$env:DATAX_KEY = "SUA_CHAVE"
python enriquecer.py

# No macOS ou Linux
export DATAX_KEY="SUA_CHAVE"
python3 enriquecer.py

"""Completa uma lista de CNPJs com o telefone e o e-mail dos sócios.

Lê empresas.csv (com uma coluna "cnpj") e grava socios.csv, uma linha por sócio.
Se parar no meio, rode de novo: os CNPJs já consultados são pulados.
"""
import csv
import os
import re
import sys
import time
from datetime import date
from pathlib import Path

import requests

API = "https://api.datax.plus"
HEADERS = {"api-key": os.environ["DATAX_KEY"]}
ENTRADA = Path("empresas.csv")
SAIDA = Path("socios.csv")
PAUSA = 0.5  # 120 chamadas por minuto, o limite do plano Intermediário
COLUNAS = ["cnpj", "nome", "qualificacao", "telefones", "emails", "consultado_em", "observacao"]


def ler_cnpjs():
    with ENTRADA.open(encoding="utf-8-sig", newline="") as f:
        inicio = f.read(2048)
        f.seek(0)
        separador = ";" if inicio.count(";") > inicio.count(",") else ","
        linhas = csv.reader(f, delimiter=separador)
        cabecalho = [c.strip().lower() for c in next(linhas)]
        if "cnpj" not in cabecalho:
            sys.exit('O arquivo empresas.csv precisa de uma coluna chamada "cnpj".')
        coluna = cabecalho.index("cnpj")
        cnpjs = {}
        for linha in linhas:
            valor = linha[coluna] if len(linha) > coluna else ""
            limpo = re.sub(r"[^0-9A-Za-z]", "", valor).upper()
            if limpo:
                cnpjs[limpo.zfill(14)] = True
    return list(cnpjs)


def ja_consultados():
    if not SAIDA.exists():
        return set()
    with SAIDA.open(encoding="utf-8-sig", newline="") as f:
        return {linha["cnpj"] for linha in csv.DictReader(f, delimiter=";")}


def main():
    r = requests.get(f"{API}/v1/plano/limite", headers=HEADERS, timeout=30)
    r.raise_for_status()
    plano = r.json()
    print(f"Plano {plano['planoNome']}: {plano['consumoDia']} de {plano['limiteDiario']} usadas hoje")

    feitos = ja_consultados()
    pendentes = [c for c in ler_cnpjs() if c not in feitos]
    print(f"{len(pendentes)} CNPJs para consultar, {len(feitos)} já consultados")

    arquivo_novo = not SAIDA.exists()
    consumo = 0
    with SAIDA.open("a", encoding="utf-8-sig", newline="") as f:
        saida = csv.writer(f, delimiter=";")
        if arquivo_novo:
            saida.writerow(COLUNAS)
        hoje = date.today().isoformat()

        for cnpj in pendentes:
            r = requests.post(f"{API}/v1/empresas/{cnpj}/socios", headers=HEADERS, timeout=60)

            if r.status_code in (404, 422):
                saida.writerow([cnpj, "", "", "", "", hoje, f"recusado pela API ({r.status_code})"])
            elif r.status_code == 200:
                dados = r.json()
                consumo += dados.get("unidadesConsumidas", 0)
                socios = dados.get("socios") or []
                if not socios:
                    saida.writerow([cnpj, "", "", "", "", hoje, "nenhum sócio retornado"])
                for s in socios:
                    saida.writerow([
                        cnpj,
                        s.get("nome"),
                        s.get("qualificacao"),
                        " | ".join(s.get("telefones") or []),
                        " | ".join(s.get("emails") or []),
                        hoje,
                        "",
                    ])
            else:
                print(f"Parando em {cnpj}: {r.status_code} {r.text[:200]}")
                print(f"Consumo desta rodada: {consumo}. Rode de novo depois de resolver.")
                sys.exit(1)

            f.flush()
            time.sleep(PAUSA)

    print(f"Pronto. Consumo desta rodada: {consumo}.")


if __name__ == "__main__":
    main()