Como consultar o Simples Nacional de uma lista de CNPJs?

pip install fiscalapi-br
export FISCALAPI_API_KEY="fapi_sua_chave_aqui"

"""Consulta o Simples Nacional de uma lista de CNPJs, com retomada."""
import csv
import os
import sys
import time

from fiscalapi_br import (
    APIConnectionError,
    AuthenticationError,
    BadRequestError,
    FiscalAPI,
    InsufficientBalanceError,
    RateLimitError,
    ServerError,
)

ENTRADA = "cnpjs.csv"   # coluna "cnpj"
SAIDA = "simples.csv"   # resultado; é também o checkpoint
COLUNAS = [
    "cnpj", "nome_empresarial", "optante_simples", "data_opcao_simples",
    "enquadrado_simei", "regime_regular_ibs", "regime_regular_cbs",
    "data_consulta", "erro",
]

# Segundos entre chamadas, pelo limite por minuto do plano.
PAUSA = {"lite": 12.0, "starter": 3.0, "pro": 1.0, "enterprise": 0.5}
PLANO = "starter"


def limpar(valor: str) -> str:
    return "".join(c for c in valor if c.isalnum()).upper()


def ja_concluidos() -> set[str]:
    if not os.path.exists(SAIDA):
        return set()
    with open(SAIDA, encoding="utf-8", newline="") as f:
        return {linha["cnpj"] for linha in csv.DictReader(f)}


def gravar(linha: dict) -> None:
    novo = not os.path.exists(SAIDA)
    with open(SAIDA, "a", encoding="utf-8", newline="") as f:
        w = csv.DictWriter(f, fieldnames=COLUNAS, extrasaction="ignore")
        if novo:
            w.writeheader()
        w.writerow(linha)


def main() -> None:
    api = FiscalAPI()  # lê FISCALAPI_API_KEY; repete sozinho em 429 e 5xx
    with open(ENTRADA, encoding="utf-8", newline="") as f:
        cnpjs = [limpar(linha["cnpj"]) for linha in csv.DictReader(f)]
    feitos = ja_concluidos()
    pendentes = list(dict.fromkeys(c for c in cnpjs if c and c not in feitos))
    print(f"{len(feitos)} já concluídos, {len(pendentes)} pendentes")

    falhas = 0
    for i, cnpj in enumerate(pendentes, 1):
        inicio = time.monotonic()
        try:
            r = api.consultar_simples_nacional(cnpj)
            gravar({**r, "cnpj": cnpj})
        except InsufficientBalanceError as e:
            print(f"Parei no {cnpj}: {e.message}. Rode de novo depois de recarregar.")
            sys.exit(2)
        except AuthenticationError as e:
            print(f"Chave recusada: {e.message}")
            sys.exit(3)
        except BadRequestError as e:
            # 400 CNPJ_INVALIDO ou 404 CNPJ_NAO_ENCONTRADO: grava e não repete.
            gravar({"cnpj": cnpj, "erro": e.code})
        except RateLimitError as e:
            time.sleep(e.retry_after or 60)   # fica para a próxima rodada
            falhas += 1
        except (ServerError, APIConnectionError):
            falhas += 1                        # fica para a próxima rodada
        if i % 100 == 0:
            print(f"{i}/{len(pendentes)}")
        espera = PAUSA[PLANO] - (time.monotonic() - inicio)
        if espera > 0:
            time.sleep(espera)

    print(f"Fim. {falhas} ficaram para a próxima rodada: rode de novo para consultá-los.")


if __name__ == "__main__":
    main()