Pular para o conteúdo principal

🏛️ Government APIs Integration

📋 Visão Geral

O Cidadão.AI integra 30+ APIs governamentais brasileiras para análise de transparência pública. O sistema utiliza o TransparencyOrchestrator para gerenciar múltiplas fontes de dados com estratégias inteligentes de fallback e agregação.

Cobertura de APIs

  • 8 APIs Federais (IBGE, DataSUS, INEP, PNCP, etc.)
  • 11 TCEs Estaduais (Tribunais de Contas)
  • Portal da Transparência (com workaround para 78% de taxa de erro)

🇧🇷 APIs Federais (8)

1. Portal da Transparência

Endpoint: https://api.portaldatransparencia.gov.br/api-de-dados/ Status: ⚠️ 78% dos endpoints retornam 403 Forbidden

# Endpoints disponíveis (22% funcionam)
GET /api/v1/transparency/contracts
GET /api/v1/transparency/expenses
GET /api/v1/transparency/agreements

Workaround Implementado:

  • Estratégia FALLBACK para TCEs estaduais
  • Cache agressivo (24h) para endpoints funcionais
  • Retry com exponential backoff
  • Timeout de 30s para evitar travamento

2. IBGE - Instituto Brasileiro de Geografia e Estatística

Endpoint: https://servicodados.ibge.gov.br/api/v1/ Status: ✅ 100% funcional

# Demographics
GET /api/v1/federal/ibge/demographics
GET /api/v1/federal/ibge/population/{city_code}
GET /api/v1/federal/ibge/municipalities

# Examples
{
"city_code": "3106200", # Belo Horizonte
"population": 2521564,
"area_km2": 331.401,
"density": 7611.02
}

3. DataSUS - Sistema Único de Saúde

Endpoint: http://tabnet.datasus.gov.br/cgi/deftohtm.exe Status: ✅ 90% funcional

GET /api/v1/federal/datasus/health-metrics
GET /api/v1/federal/datasus/hospitals
GET /api/v1/federal/datasus/mortality-data

# Health Metrics Example
{
"state": "MG",
"hospitals": 752,
"beds": 45238,
"coverage": 0.89,
"mortality_rate": 5.2
}

4. INEP - Instituto Nacional de Estudos e Pesquisas

Endpoint: https://dadosabertos.inep.gov.br/ Status: ✅ 95% funcional

GET /api/v1/federal/inep/education-stats
GET /api/v1/federal/inep/schools
GET /api/v1/federal/inep/enrollment

# Education Stats Example
{
"city": "Belo Horizonte",
"schools": 892,
"students": 432108,
"teachers": 21453,
"ideb_score": 6.2
}

5. PNCP - Portal Nacional de Contratações Públicas

Endpoint: https://pncp.gov.br/api/ Status: ✅ 85% funcional

GET /api/v1/federal/pncp/contracts
GET /api/v1/federal/pncp/bids
GET /api/v1/federal/pncp/suppliers

# Contract Example
{
"contract_id": "62023000001",
"supplier": "Empresa XYZ Ltda",
"value": 1500000.00,
"object": "Fornecimento de equipamentos",
"status": "vigente"
}

6-8. Outras APIs Federais

  • BCB - Banco Central (taxas, inflação)
  • ANS - Agência Nacional de Saúde (planos de saúde)
  • ANEEL - Agência Nacional de Energia (tarifas)

📊 TCEs - Tribunais de Contas Estaduais (11)

TCE-CE - Ceará

Endpoint: https://api.tce.ce.gov.br/ Status: ✅ Funcional Dados: Contratos, despesas, folha de pagamento

TCE-PE - Pernambuco

Endpoint: https://sistemas.tce.pe.gov.br/ Status: ✅ Funcional Dados: Licitações, contratos, prestação de contas

TCE-MG - Minas Gerais

Endpoint: https://dadosabertos.tce.mg.gov.br/ Status: ✅ Funcional Dados: Contratos, despesas, pessoal

TCE-SP - São Paulo

Endpoint: https://transparencia.tce.sp.gov.br/api/ Status: ✅ Funcional Dados: Contratos, convênios, obras

TCE-RJ - Rio de Janeiro

Endpoint: https://www.tcerj.tc.br/ Status: ⚠️ Parcial Dados: Contratos, despesas

TCEs Adicionais (6)

  • TCE-BA - Bahia
  • TCE-RS - Rio Grande do Sul
  • TCE-PR - Paraná
  • TCE-SC - Santa Catarina
  • TCE-GO - Goiás
  • TCE-DF - Distrito Federal

🔄 TransparencyOrchestrator

Estratégias de Consulta

from enum import Enum

class QueryStrategy(Enum):
FALLBACK = "fallback" # Tenta em ordem de prioridade
AGGREGATE = "aggregate" # Combina todas as fontes
FASTEST = "fastest" # Primeira resposta válida
PARALLEL = "parallel" # Todas em paralelo

Fluxo de Orquestração

Implementação

class TransparencyOrchestrator:
"""Orquestrador de múltiplas APIs governamentais"""

async def query(
self,
query_type: str,
filters: dict,
strategy: QueryStrategy = QueryStrategy.FALLBACK
):
"""
Executa query com estratégia definida

Args:
query_type: "contracts", "expenses", "demographics"
filters: {"year": 2024, "state": "MG"}
strategy: FALLBACK | AGGREGATE | FASTEST | PARALLEL
"""

if strategy == QueryStrategy.FALLBACK:
return await self._fallback_query(query_type, filters)

elif strategy == QueryStrategy.PARALLEL:
return await self._parallel_query(query_type, filters)

elif strategy == QueryStrategy.FASTEST:
return await self._fastest_query(query_type, filters)

elif strategy == QueryStrategy.AGGREGATE:
return await self._aggregate_query(query_type, filters)

💾 Cache Multi-Layer

Estratégia de Cache

CamadaTTLHit RateUso
L1: Memory5 min45%Queries repetidas
L2: Redis1 hora32%Dados compartilhados
L3: PostgreSQL24 horas10%Dados históricos
Total-87%Cache combinado

Implementação

class CacheService:
async def get_or_fetch(
self,
key: str,
fetch_function: Callable,
ttl: int = 3600
):
# L1: Memory cache
if key in self.memory_cache:
return self.memory_cache[key]

# L2: Redis cache
cached = await self.redis.get(key)
if cached:
self.memory_cache[key] = cached
return cached

# L3: Database cache
db_cached = await self.db.get_cache(key)
if db_cached and not db_cached.expired:
await self.redis.set(key, db_cached.value, ex=ttl)
self.memory_cache[key] = db_cached.value
return db_cached.value

# Cache MISS - fetch fresh data
data = await fetch_function()

# Store in all layers
self.memory_cache[key] = data
await self.redis.set(key, data, ex=ttl)
await self.db.store_cache(key, data, ttl)

return data

🔍 Intelligent Routing

Federal vs State Detection

def route_query(query: str, filters: dict) -> List[str]:
"""Detecta se é consulta federal ou estadual"""

# Federal indicators
if any(kw in query.lower() for kw in ["federal", "país", "brasil", "nacional"]):
return ["portal_transparencia", "pncp", "ibge"]

# State indicators
if "state" in filters or "uf" in filters:
state = filters.get("state") or filters.get("uf")
return [f"tce_{state.lower()}", "portal_transparencia"]

# City indicators
if "city" in filters or "município" in query.lower():
city_state = get_state_from_city(filters["city"])
return [f"tce_{city_state}", "ibge"]

# Default: try all
return ["portal_transparencia", "tce_ce", "tce_pe", "tce_mg"]

🛡️ Error Handling

Circuit Breaker Pattern

class CircuitBreaker:
def __init__(
self,
failure_threshold: int = 3,
timeout: float = 60.0
):
self.failure_count = 0
self.failure_threshold = failure_threshold
self.timeout = timeout
self.state = "CLOSED" # CLOSED | OPEN | HALF_OPEN

async def call(self, func: Callable):
if self.state == "OPEN":
if time.time() - self.last_failure > self.timeout:
self.state = "HALF_OPEN"
else:
raise CircuitBreakerOpen("Circuit breaker is OPEN")

try:
result = await func()
if self.state == "HALF_OPEN":
self.state = "CLOSED"
self.failure_count = 0
return result

except Exception as e:
self.failure_count += 1
if self.failure_count >= self.failure_threshold:
self.state = "OPEN"
self.last_failure = time.time()
raise

📊 Performance Metrics

API Response Times

APIp50p95p99Success Rate
IBGE120ms250ms500ms99.5%
DataSUS450ms1200ms2500ms90.2%
INEP200ms600ms1200ms95.8%
PNCP350ms900ms1800ms85.3%
Portal Transparência---22.0% ⚠️
TCE-CE280ms700ms1500ms88.5%
TCE-MG310ms800ms1600ms91.2%

📚 Exemplos de Uso

Buscar Contratos (Fallback Strategy)

orchestrator = TransparencyOrchestrator()

result = await orchestrator.query(
query_type="contracts",
filters={
"year": 2024,
"state": "MG",
"min_value": 100000
},
strategy=QueryStrategy.FALLBACK
)

# Result
{
"source": "tce_mg",
"total_contracts": 1234,
"total_value": 450000000.00,
"contracts": [...]
}

Agregar Dados de Múltiplas Fontes

result = await orchestrator.query(
query_type="demographics",
filters={"city": "Belo Horizonte"},
strategy=QueryStrategy.AGGREGATE
)

# Result agregado
{
"sources": ["ibge", "datasus", "inep"],
"population": 2521564, # IBGE
"hospitals": 89, # DataSUS
"schools": 892, # INEP
"aggregated_at": "2025-01-22T10:00:00Z"
}

🔗 Referências

Documentação Relacionada

APIs Oficiais


Última Atualização: 2025-01-22 Status: ✅ 30+ APIs Integradas Autor: Anderson Henrique da Silva

💡 Destaque: 87% de cache hit rate graças à estratégia multi-layer! 🚀