Upload 58 files

app/api/routes/dados_publicos.py

@@ -0,0 +1,155 @@
+"""
+Public Data API Routes - IBGE and TSE data access
+"""
+from fastapi import APIRouter, HTTPException, Query
+from pydantic import BaseModel, Field
+from typing import Optional, List, Dict, Any
+
+from app.services.ibge_api import (
+    listar_estados, 
+    listar_municipios, 
+    buscar_municipio,
+    enriquecer_localizacao
+)
+from app.services.tse_api import (
+    listar_eleicoes,
+    buscar_candidatos,
+    obter_candidato_detalhes,
+    buscar_politico
+)
+
+
+router = APIRouter(prefix="/dados", tags=["Public Data"])
+
+
+# ========== IBGE Endpoints ==========
+
+class EstadoResponse(BaseModel):
+    id: int
+    sigla: str
+    nome: str
+    regiao: str
+
+
+class MunicipioResponse(BaseModel):
+    id: int
+    nome: str
+    estado_sigla: str
+    estado_nome: str
+    regiao: str
+
+
+@router.get("/ibge/estados", response_model=List[EstadoResponse])
+async def get_estados():
+    """List all Brazilian states"""
+    estados = await listar_estados()
+    return [EstadoResponse(**e.__dict__) for e in estados]
+
+
+@router.get("/ibge/municipios/{uf}", response_model=List[MunicipioResponse])
+async def get_municipios(uf: str):
+    """List municipalities in a state"""
+    municipios = await listar_municipios(uf)
+    return [MunicipioResponse(**m.__dict__) for m in municipios]
+
+
+@router.get("/ibge/buscar")
+async def buscar_cidade(
+    nome: str = Query(..., min_length=2),
+    uf: Optional[str] = None
+):
+    """Search for a municipality by name"""
+    municipios = await buscar_municipio(nome, uf)
+    return [MunicipioResponse(**m.__dict__) for m in municipios]
+
+
+@router.get("/ibge/enriquecer")
+async def enriquecer_cidade(
+    cidade: str = Query(..., min_length=2),
+    uf: Optional[str] = None
+):
+    """Enrich a location name with IBGE data"""
+    return await enriquecer_localizacao(cidade, uf)
+
+
+# ========== TSE Endpoints ==========
+
+class EleicaoResponse(BaseModel):
+    id: int
+    ano: int
+    descricao: str
+    turno: int
+
+
+class CandidatoResponse(BaseModel):
+    id: int
+    nome: str
+    nome_urna: str
+    numero: str
+    cargo: str
+    partido_sigla: str
+    uf: str
+    municipio: str
+    situacao: str
+    total_bens: float
+
+
+class CandidatoDetalhadoResponse(BaseModel):
+    id: int
+    nome: str
+    nome_urna: str
+    numero: str
+    cargo: str
+    partido_sigla: str
+    partido_nome: str
+    uf: str
+    municipio: str
+    situacao: str
+    data_nascimento: str
+    genero: str
+    grau_instrucao: str
+    ocupacao: str
+    total_bens: float
+    bens: List[Dict[str, Any]]
+
+
+@router.get("/tse/eleicoes", response_model=List[EleicaoResponse])
+async def get_eleicoes():
+    """List available elections"""
+    eleicoes = await listar_eleicoes()
+    return [EleicaoResponse(**e.__dict__) for e in eleicoes]
+
+
+@router.get("/tse/candidatos")
+async def get_candidatos(
+    nome: str = Query(..., min_length=3),
+    ano: int = Query(default=2024),
+    uf: Optional[str] = None,
+    cargo: Optional[str] = None
+):
+    """Search for candidates by name"""
+    candidatos = await buscar_candidatos(nome, ano=ano, uf=uf, cargo=cargo)
+    return [CandidatoResponse(**c.__dict__) for c in candidatos]
+
+
+@router.get("/tse/candidato/{id_candidato}")
+async def get_candidato_detalhes(
+    id_candidato: int,
+    ano: int = Query(default=2024)
+):
+    """Get detailed candidate information including assets"""
+    candidato = await obter_candidato_detalhes(id_candidato, ano=ano)
+    
+    if not candidato:
+        raise HTTPException(status_code=404, detail="Candidato não encontrado")
+    
+    return CandidatoDetalhadoResponse(**candidato.__dict__)
+
+
+@router.get("/tse/politico")
+async def pesquisar_politico(nome: str = Query(..., min_length=3)):
+    """
+    Search for a politician across multiple elections.
+    Returns consolidated career information.
+    """
+    return await buscar_politico(nome)

app/main.py

@@ -8,7 +8,7 @@ from contextlib import asynccontextmanager
 
 from app.config import settings
 from app.core.database import init_db
-from app.api.routes import entities, relationships, events, search, ingest, analyze, graph, research, chat, investigate
+from app.api.routes import entities, relationships, events, search, ingest, analyze, graph, research, chat, investigate, dados_publicos
 
 
 @asynccontextmanager
@@ -60,6 +60,7 @@ app.include_router(graph.router, prefix="/api/v1")
 app.include_router(research.router, prefix="/api/v1")
 app.include_router(chat.router, prefix="/api/v1")
 app.include_router(investigate.router, prefix="/api/v1")
+app.include_router(dados_publicos.router, prefix="/api/v1")
 
 
 @app.get("/")

app/services/ibge_api.py

@@ -0,0 +1,192 @@
+"""
+IBGE API Service
+Access to Brazilian geographic and demographic data
+"""
+import httpx
+from typing import Optional, Dict, Any, List
+from dataclasses import dataclass
+
+
+IBGE_BASE_URL = "https://servicodados.ibge.gov.br/api/v1"
+
+
+@dataclass
+class Estado:
+    """Brazilian state data"""
+    id: int
+    sigla: str
+    nome: str
+    regiao: str
+
+
+@dataclass
+class Municipio:
+    """Brazilian municipality data"""
+    id: int
+    nome: str
+    estado_sigla: str
+    estado_nome: str
+    regiao: str
+    # Optional enriched data
+    populacao: Optional[int] = None
+    area_km2: Optional[float] = None
+
+
+async def listar_estados() -> List[Estado]:
+    """List all Brazilian states"""
+    try:
+        async with httpx.AsyncClient(timeout=15.0) as client:
+            response = await client.get(f"{IBGE_BASE_URL}/localidades/estados")
+            
+            if response.status_code != 200:
+                return []
+            
+            data = response.json()
+            estados = []
+            
+            for item in data:
+                estados.append(Estado(
+                    id=item["id"],
+                    sigla=item["sigla"],
+                    nome=item["nome"],
+                    regiao=item.get("regiao", {}).get("nome", "")
+                ))
+            
+            return sorted(estados, key=lambda x: x.nome)
+            
+    except Exception as e:
+        print(f"IBGE estados error: {e}")
+        return []
+
+
+async def listar_municipios(uf: str) -> List[Municipio]:
+    """List all municipalities in a state"""
+    try:
+        async with httpx.AsyncClient(timeout=15.0) as client:
+            response = await client.get(
+                f"{IBGE_BASE_URL}/localidades/estados/{uf}/municipios"
+            )
+            
+            if response.status_code != 200:
+                return []
+            
+            data = response.json()
+            municipios = []
+            
+            for item in data:
+                municipios.append(Municipio(
+                    id=item["id"],
+                    nome=item["nome"],
+                    estado_sigla=uf.upper(),
+                    estado_nome=item.get("microrregiao", {}).get("mesorregiao", {}).get("UF", {}).get("nome", ""),
+                    regiao=item.get("microrregiao", {}).get("mesorregiao", {}).get("UF", {}).get("regiao", {}).get("nome", "")
+                ))
+            
+            return sorted(municipios, key=lambda x: x.nome)
+            
+    except Exception as e:
+        print(f"IBGE municipios error: {e}")
+        return []
+
+
+async def buscar_municipio(nome: str, uf: Optional[str] = None) -> List[Municipio]:
+    """Search for municipalities by name"""
+    try:
+        # If UF provided, search only that state
+        if uf:
+            municipios = await listar_municipios(uf)
+            return [m for m in municipios if nome.lower() in m.nome.lower()]
+        
+        # Otherwise search all states (slower)
+        async with httpx.AsyncClient(timeout=30.0) as client:
+            response = await client.get(f"{IBGE_BASE_URL}/localidades/municipios")
+            
+            if response.status_code != 200:
+                return []
+            
+            data = response.json()
+            results = []
+            
+            for item in data:
+                if nome.lower() in item["nome"].lower():
+                    uf_info = item.get("microrregiao", {}).get("mesorregiao", {}).get("UF", {})
+                    results.append(Municipio(
+                        id=item["id"],
+                        nome=item["nome"],
+                        estado_sigla=uf_info.get("sigla", ""),
+                        estado_nome=uf_info.get("nome", ""),
+                        regiao=uf_info.get("regiao", {}).get("nome", "")
+                    ))
+            
+            return results[:20]  # Limit results
+            
+    except Exception as e:
+        print(f"IBGE search error: {e}")
+        return []
+
+
+async def obter_municipio_por_id(id_municipio: int) -> Optional[Municipio]:
+    """Get municipality by IBGE code"""
+    try:
+        async with httpx.AsyncClient(timeout=15.0) as client:
+            response = await client.get(
+                f"{IBGE_BASE_URL}/localidades/municipios/{id_municipio}"
+            )
+            
+            if response.status_code != 200:
+                return None
+            
+            item = response.json()
+            uf_info = item.get("microrregiao", {}).get("mesorregiao", {}).get("UF", {})
+            
+            return Municipio(
+                id=item["id"],
+                nome=item["nome"],
+                estado_sigla=uf_info.get("sigla", ""),
+                estado_nome=uf_info.get("nome", ""),
+                regiao=uf_info.get("regiao", {}).get("nome", "")
+            )
+            
+    except Exception as e:
+        print(f"IBGE municipio error: {e}")
+        return None
+
+
+async def enriquecer_localizacao(cidade: str, uf: Optional[str] = None) -> Dict[str, Any]:
+    """
+    Enrich a location name with IBGE data.
+    Useful for adding context to extracted locations.
+    """
+    resultado = {
+        "cidade_original": cidade,
+        "encontrado": False,
+        "ibge_codigo": None,
+        "cidade": None,
+        "estado": None,
+        "estado_sigla": None,
+        "regiao": None
+    }
+    
+    municipios = await buscar_municipio(cidade, uf)
+    
+    if municipios:
+        # Take best match (exact or first)
+        melhor = None
+        for m in municipios:
+            if m.nome.lower() == cidade.lower():
+                melhor = m
+                break
+        
+        if not melhor:
+            melhor = municipios[0]
+        
+        resultado.update({
+            "encontrado": True,
+            "ibge_codigo": melhor.id,
+            "cidade": melhor.nome,
+            "estado": melhor.estado_nome,
+            "estado_sigla": melhor.estado_sigla,
+            "regiao": melhor.regiao
+        })
+    
+    return resultado

app/services/tse_api.py

@@ -0,0 +1,264 @@
+"""
+TSE (Tribunal Superior Eleitoral) API Service
+Access to Brazilian electoral data - candidates, assets, donations
+"""
+import httpx
+from typing import Optional, Dict, Any, List
+from dataclasses import dataclass, field
+
+
+# DivulgaCand API (unofficial but functional)
+TSE_DIVULGACAND_URL = "https://divulgacandcontas.tse.jus.br/divulga/rest/v1"
+
+
+@dataclass
+class Candidato:
+    """Electoral candidate data"""
+    id: int
+    nome: str
+    nome_urna: str
+    cpf_parcial: str = ""  # TSE only shows partial
+    numero: str = ""
+    cargo: str = ""
+    partido_sigla: str = ""
+    partido_nome: str = ""
+    coligacao: str = ""
+    situacao: str = ""
+    
+    # Location
+    uf: str = ""
+    municipio: str = ""
+    
+    # Personal
+    data_nascimento: str = ""
+    genero: str = ""
+    grau_instrucao: str = ""
+    ocupacao: str = ""
+    
+    # Assets
+    total_bens: float = 0.0
+    bens: List[Dict[str, Any]] = field(default_factory=list)
+    
+    # Campaign
+    total_receitas: float = 0.0
+    total_despesas: float = 0.0
+
+
+@dataclass
+class Eleicao:
+    """Election metadata"""
+    id: int
+    ano: int
+    descricao: str
+    turno: int = 1
+
+
+async def listar_eleicoes() -> List[Eleicao]:
+    """List available elections"""
+    try:
+        async with httpx.AsyncClient(timeout=15.0) as client:
+            response = await client.get(f"{TSE_DIVULGACAND_URL}/eleicao/ordinarias")
+            
+            if response.status_code != 200:
+                return []
+            
+            data = response.json()
+            eleicoes = []
+            
+            for item in data:
+                eleicoes.append(Eleicao(
+                    id=item.get("id", 0),
+                    ano=item.get("ano", 0),
+                    descricao=item.get("descricaoEleicao", ""),
+                    turno=item.get("turno", 1)
+                ))
+            
+            return sorted(eleicoes, key=lambda x: x.ano, reverse=True)
+            
+    except Exception as e:
+        print(f"TSE eleicoes error: {e}")
+        return []
+
+
+async def buscar_candidatos(
+    nome: str,
+    ano: int = 2024,
+    uf: Optional[str] = None,
+    cargo: Optional[str] = None
+) -> List[Candidato]:
+    """
+    Search for candidates by name.
+    
+    Args:
+        nome: Candidate name to search
+        ano: Election year (default 2024)
+        uf: State filter (optional)
+        cargo: Position filter (optional)
+    """
+    try:
+        # First get the election ID for the year
+        eleicoes = await listar_eleicoes()
+        eleicao = next((e for e in eleicoes if e.ano == ano), None)
+        
+        if not eleicao:
+            # Try common election IDs
+            eleicao_id = {2024: 546, 2022: 544, 2020: 426, 2018: 295}.get(ano, 546)
+        else:
+            eleicao_id = eleicao.id
+        
+        # Build search URL
+        base_url = f"{TSE_DIVULGACAND_URL}/candidatura/listar/{ano}/{eleicao_id}"
+        
+        params = {"nomeCompleto": nome}
+        if uf:
+            params["uf"] = uf.upper()
+        if cargo:
+            params["cargo"] = cargo
+        
+        async with httpx.AsyncClient(timeout=30.0) as client:
+            response = await client.get(base_url, params=params)
+            
+            if response.status_code != 200:
+                return []
+            
+            data = response.json()
+            candidatos_data = data.get("candidatos", [])
+            
+            candidatos = []
+            for item in candidatos_data:
+                candidatos.append(Candidato(
+                    id=item.get("id", 0),
+                    nome=item.get("nomeCompleto", ""),
+                    nome_urna=item.get("nomeUrna", ""),
+                    cpf_parcial=item.get("cpf", "")[:3] + ".***.***-**" if item.get("cpf") else "",
+                    numero=str(item.get("numero", "")),
+                    cargo=item.get("cargo", {}).get("nome", "") if isinstance(item.get("cargo"), dict) else str(item.get("cargo", "")),
+                    partido_sigla=item.get("partido", {}).get("sigla", "") if isinstance(item.get("partido"), dict) else "",
+                    partido_nome=item.get("partido", {}).get("nome", "") if isinstance(item.get("partido"), dict) else "",
+                    uf=item.get("ufSigla", "") or item.get("uf", ""),
+                    municipio=item.get("municipio", {}).get("nome", "") if isinstance(item.get("municipio"), dict) else "",
+                    situacao=item.get("situacao", ""),
+                    total_bens=float(item.get("totalDeBens", 0) or 0)
+                ))
+            
+            return candidatos
+            
+    except Exception as e:
+        print(f"TSE search error: {e}")
+        return []
+
+
+async def obter_candidato_detalhes(
+    id_candidato: int,
+    ano: int = 2024,
+    eleicao_id: Optional[int] = None
+) -> Optional[Candidato]:
+    """Get detailed candidate information including assets"""
+    try:
+        if not eleicao_id:
+            eleicao_id = {2024: 546, 2022: 544, 2020: 426, 2018: 295}.get(ano, 546)
+        
+        async with httpx.AsyncClient(timeout=30.0) as client:
+            # Get candidate details
+            response = await client.get(
+                f"{TSE_DIVULGACAND_URL}/candidatura/buscar/{ano}/{eleicao_id}/candidato/{id_candidato}"
+            )
+            
+            if response.status_code != 200:
+                return None
+            
+            item = response.json()
+            
+            candidato = Candidato(
+                id=item.get("id", 0),
+                nome=item.get("nomeCompleto", ""),
+                nome_urna=item.get("nomeUrna", ""),
+                numero=str(item.get("numero", "")),
+                cargo=item.get("cargo", {}).get("nome", "") if isinstance(item.get("cargo"), dict) else "",
+                partido_sigla=item.get("partido", {}).get("sigla", "") if isinstance(item.get("partido"), dict) else "",
+                partido_nome=item.get("partido", {}).get("nome", "") if isinstance(item.get("partido"), dict) else "",
+                uf=item.get("ufSigla", ""),
+                municipio=item.get("localCandidatura", ""),
+                situacao=item.get("situacao", ""),
+                data_nascimento=item.get("dataNascimento", ""),
+                genero=item.get("genero", ""),
+                grau_instrucao=item.get("grauInstrucao", ""),
+                ocupacao=item.get("ocupacao", ""),
+                total_bens=float(item.get("totalDeBens", 0) or 0)
+            )
+            
+            # Try to get assets (bens)
+            try:
+                bens_response = await client.get(
+                    f"{TSE_DIVULGACAND_URL}/candidatura/buscar/{ano}/{eleicao_id}/candidato/{id_candidato}/bens"
+                )
+                if bens_response.status_code == 200:
+                    bens_data = bens_response.json()
+                    candidato.bens = [
+                        {
+                            "tipo": b.get("tipoBem", ""),
+                            "descricao": b.get("descricao", ""),
+                            "valor": float(b.get("valor", 0) or 0)
+                        }
+                        for b in bens_data
+                    ]
+            except:
+                pass
+            
+            return candidato
+            
+    except Exception as e:
+        print(f"TSE details error: {e}")
+        return None
+
+
+async def buscar_politico(nome: str) -> Dict[str, Any]:
+    """
+    Search for a politician across multiple elections.
+    Returns consolidated information.
+    """
+    resultado = {
+        "nome": nome,
+        "encontrado": False,
+        "candidaturas": [],
+        "ultimo_cargo": None,
+        "total_patrimonio": 0.0,
+        "partidos": set(),
+        "ufs": set()
+    }
+    
+    # Search in recent elections
+    for ano in [2024, 2022, 2020, 2018]:
+        candidatos = await buscar_candidatos(nome, ano=ano)
+        
+        for c in candidatos:
+            if nome.lower() in c.nome.lower():
+                resultado["encontrado"] = True
+                resultado["candidaturas"].append({
+                    "ano": ano,
+                    "cargo": c.cargo,
+                    "partido": c.partido_sigla,
+                    "uf": c.uf,
+                    "situacao": c.situacao,
+                    "patrimonio": c.total_bens
+                })
+                
+                if c.partido_sigla:
+                    resultado["partidos"].add(c.partido_sigla)
+                if c.uf:
+                    resultado["ufs"].add(c.uf)
+                
+                if c.total_bens > resultado["total_patrimonio"]:
+                    resultado["total_patrimonio"] = c.total_bens
+                
+                if not resultado["ultimo_cargo"]:
+                    resultado["ultimo_cargo"] = f"{c.cargo} ({ano})"
+        
+        if resultado["encontrado"]:
+            break  # Found in most recent election
+    
+    # Convert sets to lists for JSON
+    resultado["partidos"] = list(resultado["partidos"])
+    resultado["ufs"] = list(resultado["ufs"])
+    
+    return resultado