Como consultar a inscrição estadual de milhares de CNPJs em Python?

pip install fiscalapi-br
export FISCALAPI_API_KEY="fapi_sua_chave_aqui"

"""Consulta a IE de uma lista de CNPJs nos 27 estados, com retomada."""
import csv
import json
import re
import sys
import time
from datetime import datetime, timezone

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

ENTRADA = "cnpjs.csv"            # coluna "cnpj"
SAIDA = "resultado.jsonl"        # uma linha por CNPJ concluído (é também o checkpoint)
FALHAS = "falhas.jsonl"          # 5xx e falhas de rede: entram de novo na próxima rodada

# Segundos entre chamadas, pelo limite por minuto do plano.
PAUSA = {"lite": 12.0, "sem_plano": 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 cnpj_valido(cnpj: str) -> bool:
    """Dígitos verificadores, também do CNPJ alfanumérico. CNPJ com DV errado não
    volta erro da API: é consultado e cobrado. Melhor barrar antes."""
    if not re.fullmatch(r"[0-9A-Z]{12}[0-9]{2}", cnpj) or len(set(cnpj)) == 1:
        return False
    v = [ord(c) - 48 for c in cnpj]
    for n in (12, 13):
        pesos = list(range(n - 7, 1, -1)) + list(range(9, 1, -1))
        resto = sum(a * b for a, b in zip(v, pesos)) % 11
        if v[n] != (0 if resto < 2 else 11 - resto):
            return False
    return True


def ja_concluidos() -> set[str]:
    try:
        with open(SAIDA, encoding="utf-8") as f:
            return {json.loads(linha)["cnpj"] for linha in f if linha.strip()}
    except FileNotFoundError:
        return set()


def gravar(arquivo: str, registro: dict) -> None:
    registro["consultado_em"] = datetime.now(timezone.utc).isoformat()
    with open(arquivo, "a", encoding="utf-8") as f:
        f.write(json.dumps(registro, ensure_ascii=False) + "\n")


def main() -> None:
    api = FiscalAPI()  # lê FISCALAPI_API_KEY; já repete sozinho em 429 e 5xx
    pausa = PAUSA[PLANO]

    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")

    ultima = 0.0
    for i, cnpj in enumerate(pendentes, 1):
        if not cnpj_valido(cnpj):
            gravar(SAIDA, {"cnpj": cnpj, "erro": "CNPJ_INVALIDO"})  # não chama a API
            continue

        espera = pausa - (time.monotonic() - ultima)
        if espera > 0:
            time.sleep(espera)
        ultima = time.monotonic()

        try:
            resp = api.consultar_ie_todos(cnpj=cnpj)
        except InsufficientBalanceError as e:
            # 402: acabaram as consultas do mês e não há saldo de recarga. Repetir não adianta.
            print(f"Parei no {cnpj}: {e.message}. Rode de novo depois de recarregar.")
            sys.exit(2)
        except AuthenticationError as e:
            # 403: chave ausente, inválida ou revogada.
            print(f"Chave recusada: {e.message}")
            sys.exit(3)
        except RateLimitError as e:
            # 429 mesmo depois das tentativas do SDK: espera e deixa para a próxima rodada.
            print(f"Limite por minuto atingido; esperando {e.retry_after or 60} s")
            time.sleep(e.retry_after or 60)
            continue
        except BadRequestError as e:
            # 400: documento ausente ou sem dígitos. Grava como concluído.
            gravar(SAIDA, {"cnpj": cnpj, "erro": e.code, "mensagem": e.message})
            continue
        except (ServerError, APIConnectionError) as e:
            gravar(FALHAS, {"cnpj": cnpj, "erro": e.status_code, "mensagem": e.message})
            continue

        inscricoes = [
            {"uf": estado["uf"], "ie": ie["inscricao_estadual"], "situacao": ie["situacao_ie"]}
            for estado in resp["results"]
            for ie in estado["results"]
        ]
        gravar(SAIDA, {"cnpj": cnpj, "complete": resp.get("complete", True), "inscricoes": inscricoes})
        if i % 100 == 0:
            print(f"{i}/{len(pendentes)}")

    print("Fim. Os que caíram em falhas.jsonl entram de novo se você rodar outra vez.")


if __name__ == "__main__":
    main()