Como fazer a transição de uma API de CPF gratuita para uma API premium

import os
import json

class InventarioIntegracao:
    def __init__(self):
    self.pontos_integracao = []

    def registrar(self, servico: str, arquivo: str, tipo: str):
    self.pontos_integracao.append({
    "servico": servico,
    "arquivo": arquivo,
    "tipo": tipo,
    "usa_variavel_ambiente": True,
    "tem_fallback": False,
    "tem_testes": False
    })

    def gerar_checklist(self) -> dict:
    total = len(self.pontos_integracao)
    com_env = sum(1 for p in self.pontos_integracao if p["usa_variavel_ambiente"])
    com_fallback = sum(1 for p in self.pontos_integracao if p["tem_fallback"])
    com_testes = sum(1 for p in self.pontos_integracao if p["tem_testes"])

    return {
    "total_pontos": total,
    "usando_variavel_ambiente": f"{com_env}/{total}",
    "com_fallback": f"{com_fallback}/{total}",
    "com_testes": f"{com_testes}/{total}",
    "pronto_para_migrar": com_env == total,
    "pontos": self.pontos_integracao
    }

# Levantamento
inventario = InventarioIntegracao()
inventario.registrar("API de cadastro", "services/cpf_service.py", "backend")
inventario.registrar("Validação no checkout", "services/checkout.py", "backend")
inventario.registrar("Importação CSV", "jobs/importar_clientes.py", "batch")
print(json.dumps(inventario.gerar_checklist(), indent=2))

import os
import requests
from enum import Enum

class APITier(Enum):
    GRATUITO = "gratuito"
    PREMIUM = "premium"

class CPFServiceMigravel:
    def __init__(self):
    self.configs = {
    APITier.GRATUITO: {
    "api_key": os.environ.get("CPFHUB_API_KEY_FREE", ""),
    "base_url": "https://api.cpfhub.io/cpf",
    },
    APITier.PREMIUM: {
    "api_key": os.environ.get("CPFHUB_API_KEY_PREMIUM", ""),
    "base_url": "https://api.cpfhub.io/cpf",
    }
    }

    tier_str = os.environ.get("CPFHUB_TIER", "gratuito")
    self.tier_ativo = APITier(tier_str)
    self.session = requests.Session()

    def consultar(self, cpf: str) -> dict:
    config = self.configs[self.tier_ativo]
    cpf_limpo = "".join(c for c in cpf if c.isdigit())

    response = self.session.get(
    f"{config['base_url']}/{cpf_limpo}",
    headers={"x-api-key": config["api_key"]},
    timeout=10
    )
    response.raise_for_status()

    resultado = response.json()
    resultado["_tier"] = self.tier_ativo.value
    return resultado

    def alternar_tier(self, novo_tier: APITier):
    self.tier_ativo = novo_tier

# Migração controlada por variável de ambiente:
# CPFHUB_TIER=gratuito (padrão)
# CPFHUB_TIER=premium (após migração)

# Shadow mode: comparar respostas entre gratuito e premium
class ShadowComparator:
    def __init__(self):
    self.service_free = CPFServiceMigravel()
    self.service_free.tier_ativo = APITier.GRATUITO

    self.service_premium = CPFServiceMigravel()
    self.service_premium.tier_ativo = APITier.PREMIUM

    self.divergencias = []

    def comparar(self, cpf: str) -> dict:
    resultado_free = self.service_free.consultar(cpf)
    resultado_premium = self.service_premium.consultar(cpf)

    campos_comparar = ["cpf", "name", "birthDate", "gender"]
    iguais = True

    for campo in campos_comparar:
    val_free = resultado_free.get("data", {}).get(campo)
    val_premium = resultado_premium.get("data", {}).get(campo)

    if val_free != val_premium:
    iguais = False
    self.divergencias.append({
    "cpf": cpf,
    "campo": campo,
    "free": val_free,
    "premium": val_premium
    })

    return {
    "cpf": cpf,
    "paridade": iguais,
    "divergencias": len(self.divergencias)
    }

class ValidacaoPosMigracao:
    def __init__(self, service: CPFServiceMigravel):
    self.service = service
    self.resultados = []

    def executar_suite(self, cpfs_teste: list) -> dict:
    for cpf in cpfs_teste:
    try:
    resultado = self.service.consultar(cpf)
    self.resultados.append({
    "cpf": cpf,
    "sucesso": resultado.get("success", False),
    "tier": resultado.get("_tier"),
    "status": "ok"
    })
    except Exception as e:
    self.resultados.append({
    "cpf": cpf,
    "sucesso": False,
    "tier": self.service.tier_ativo.value,
    "status": f"erro: {str(e)}"
    })

    total = len(self.resultados)
    sucessos = sum(1 for r in self.resultados if r["sucesso"])

    return {
    "total_testes": total,
    "sucessos": sucessos,
    "falhas": total - sucessos,
    "taxa_sucesso": f"{sucessos/total*100:.1f}%",
    "tier_ativo": self.service.tier_ativo.value,
    "pronto_para_producao": sucessos == total
    }

# Executar validação
service = CPFServiceMigravel()
validador = ValidacaoPosMigracao(service)
resultado = validador.executar_suite([
    "12345678909",
    "98765432100",
    "11122233344",
])
print(json.dumps(resultado, indent=2))

class OtimizadorConsumo:
    def __init__(self, limite_mensal: int, custo_mensal: float):
    self.limite = limite_mensal
    self.custo = custo_mensal
    self.consumo_diario = []

    def registrar_dia(self, requisicoes: int):
    self.consumo_diario.append(requisicoes)

    def projecao_mensal(self) -> dict:
    if not self.consumo_diario:
    return {"erro": "Sem dados"}

    media_diaria = sum(self.consumo_diario) / len(self.consumo_diario)
    projecao = media_diaria * 30
    utilizacao = (projecao / self.limite) * 100
    custo_por_req = self.custo / max(projecao, 1)

    return {
    "media_diaria": round(media_diaria),
    "projecao_mensal": round(projecao),
    "limite_mensal": self.limite,
    "utilizacao": f"{utilizacao:.1f}%",
    "custo_por_requisicao": f"R$ {custo_por_req:.4f}",
    "alerta": "sub-utilizado" if utilizacao < 30 else
    "adequado" if utilizacao < 80 else "próximo do limite"
    }