APIs Gratuitas de CPF: Quando Elas Deixam de Ser Viáveis?

import requests
import os
from datetime import datetime, timedelta

class MonitorLimite:
    def __init__(self):
    self.api_key = os.environ.get("CPFHUB_API_KEY", "")
    self.base_url = "https://api.cpfhub.io/cpf"
    self.historico = []

    def registrar_consulta(self, status_code: int):
    self.historico.append({
    "timestamp": datetime.now(),
    "status": status_code,
    "excedeu_limite": status_code == 402  # CPFHub.io cobra excedente, não bloqueia
    })

    def analisar_periodo(self, dias: int = 30) -> dict:
    corte = datetime.now() - timedelta(days=dias)
    periodo = [h for h in self.historico if h["timestamp"] > corte]

    if not periodo:
    return {"erro": "Sem dados no período"}

    total = len(periodo)
    excedentes = sum(1 for h in periodo if h["excedeu_limite"])
    taxa_excedente = (excedentes / total * 100) if total > 0 else 0

    # Análise por semana para identificar tendência
    semanas = {}
    for h in periodo:
    semana = h["timestamp"].isocalendar()[1]
    if semana not in semanas:
    semanas[semana] = {"total": 0, "excedentes": 0}
    semanas[semana]["total"] += 1
    semanas[semana]["excedentes"] += 1 if h["excedeu_limite"] else 0

    tendencia = "estável"
    taxas_semanais = [
    s["excedentes"] / max(s["total"], 1) * 100
    for s in semanas.values()
    ]
    if len(taxas_semanais) >= 2:
    if taxas_semanais[-1] > taxas_semanais[0] * 1.5:
    tendencia = "crescente"
    elif taxas_semanais[-1] < taxas_semanais[0] * 0.5:
    tendencia = "decrescente"

    return {
    "total_consultas": total,
    "consultas_excedentes": excedentes,
    "taxa_excedente": f"{taxa_excedente:.1f}%",
    "tendencia": tendencia,
    "viavel": taxa_excedente < 1.0,
    "recomendacao": "manter" if taxa_excedente < 1.0 else
    "monitorar" if taxa_excedente < 5.0 else "migrar"
    }

class CorrelacaoImpacto:
    def __init__(self):
    self.dados_diarios = []

    def adicionar_dia(self, data: str, api_disponivel_pct: float,
    cadastros_completados: int, cadastros_abandonados: int):
    total = cadastros_completados + cadastros_abandonados
    taxa_conversao = (cadastros_completados / total * 100) if total > 0 else 0

    self.dados_diarios.append({
    "data": data,
    "disponibilidade_api": api_disponivel_pct,
    "cadastros_ok": cadastros_completados,
    "cadastros_abandonados": cadastros_abandonados,
    "taxa_conversao": round(taxa_conversao, 1)
    })

    def correlacao(self) -> dict:
    if len(self.dados_diarios) < 7:
    return {"erro": "Mínimo de 7 dias de dados necessário"}

    dias_bons = [d for d in self.dados_diarios if d["disponibilidade_api"] > 99]
    dias_ruins = [d for d in self.dados_diarios if d["disponibilidade_api"] <= 99]

    conversao_dias_bons = (
    sum(d["taxa_conversao"] for d in dias_bons) / len(dias_bons)
    if dias_bons else 0
    )
    conversao_dias_ruins = (
    sum(d["taxa_conversao"] for d in dias_ruins) / len(dias_ruins)
    if dias_ruins else 0
    )

    impacto = conversao_dias_bons - conversao_dias_ruins

    return {
    "dias_analisados": len(self.dados_diarios),
    "dias_com_boa_disponibilidade": len(dias_bons),
    "dias_com_problemas": len(dias_ruins),
    "conversao_media_dias_bons": f"{conversao_dias_bons:.1f}%",
    "conversao_media_dias_ruins": f"{conversao_dias_ruins:.1f}%",
    "impacto_na_conversao": f"{impacto:.1f} pontos percentuais",
    "justifica_upgrade": impacto > 2.0
    }

def avaliar_necessidades_tecnicas(requisitos: dict) -> dict:
    """
    requisitos: dict com True/False para cada necessidade
    """
    necessidades_premium = {
    "sla_contratual": {
    "peso": 5,
    "descricao": "Garantia formal de disponibilidade"
    },
    "suporte_dedicado": {
    "peso": 3,
    "descricao": "Atendimento técnico com SLA"
    },
    "volume_ampliado": {
    "peso": 4,
    "descricao": "Mais requisições por mês"
    },
    "dados_adicionais": {
    "peso": 2,
    "descricao": "Campos extras na resposta"
    },
    "prioridade_servidor": {
    "peso": 3,
    "descricao": "Menor latência e maior estabilidade"
    }
    }

    score = 0
    score_max = 0
    detalhes = []

    for necessidade, config in necessidades_premium.items():
    precisa = requisitos.get(necessidade, False)
    score_max += config["peso"]
    if precisa:
    score += config["peso"]
    detalhes.append(f"Precisa: {config['descricao']} (peso {config['peso']})")

    percentual = (score / score_max * 100) if score_max > 0 else 0

    return {
    "score": f"{score}/{score_max}",
    "percentual": f"{percentual:.0f}%",
    "recomendacao": "migrar" if percentual >= 60 else
    "avaliar" if percentual >= 30 else "manter gratuito",
    "detalhes": detalhes
    }

# Exemplo: empresa com exigência regulatória
resultado = avaliar_necessidades_tecnicas({
    "sla_contratual": True,
    "suporte_dedicado": False,
    "volume_ampliado": True,
    "dados_adicionais": False,
    "prioridade_servidor": True
})

class CustoOportunidade:
    def __init__(self, custo_hora_dev: float = 150.0):
    self.custo_hora = custo_hora_dev
    self.atividades = []

    def registrar(self, descricao: str, horas_mes: float):
    self.atividades.append({
    "descricao": descricao,
    "horas_mes": horas_mes,
    "custo_mes": horas_mes * self.custo_hora
    })

    def calcular_total(self, custo_plano_pago: float) -> dict:
    custo_total_contornos = sum(a["custo_mes"] for a in self.atividades)
    horas_totais = sum(a["horas_mes"] for a in self.atividades)
    economia_potencial = custo_total_contornos - custo_plano_pago

    return {
    "horas_gastas_contornos": round(horas_totais, 1),
    "custo_contornos": f"R$ {custo_total_contornos:.2f}",
    "custo_plano_pago": f"R$ {custo_plano_pago:.2f}",
    "economia_migrando": f"R$ {economia_potencial:.2f}",
    "vale_migrar": economia_potencial > 0,
    "atividades": self.atividades
    }

# Exemplo real
custo = CustoOportunidade(custo_hora_dev=150.0)
custo.registrar("Investigar e corrigir falhas de disponibilidade", 4.0)
custo.registrar("Manter e atualizar lógica de fallback", 2.0)
custo.registrar("Responder tickets de suporte por falhas na API", 3.0)
custo.registrar("Monitorar disponibilidade manualmente", 1.0)

resultado = custo.calcular_total(custo_plano_pago=149.00)
# Custo dos contornos: R$ 1500.00/mês
# Custo do plano pago: R$ 149.00/mês
# Economia migrando: R$ 1351.00/mês