feat: Implementa job de ranking para 300k consultores
Backend: - Adiciona Scroll API no cliente Elasticsearch para processar todos os 300k+ consultores - Cria tabela TB_RANKING_CONSULTOR no Oracle para ranking pré-calculado - Implementa job de processamento com APScheduler (diário às 3h) - Adiciona endpoints: /ranking/paginado, /ranking/status, /ranking/processar, /ranking/estatisticas - Repository Oracle com paginação eficiente via ROW_NUMBER - Status do job com progresso em tempo real (polling) - Leitura automática de LOBs no OracleClient Frontend: - Componente RankingPaginado com paginação completa - Barra de progresso do job em tempo real - Botão para reprocessar ranking - Alternância entre Top N (rápido) e Ranking Completo (300k) Infraestrutura: - Docker compose com depends_on para garantir Oracle disponível - Schema SQL com procedure SP_ATUALIZAR_POSICOES - Índices otimizados para paginação
This commit is contained in:
Frederico Castro
162
backend/src/application/jobs/processar_ranking.py
Normal file
162
backend/src/application/jobs/processar_ranking.py
Normal file
@@ -0,0 +1,162 @@
|
||||
import json
|
||||
from datetime import datetime
|
||||
from typing import Optional, Dict, Any
|
||||
|
||||
from ...infrastructure.elasticsearch.client import ElasticsearchClient
|
||||
from ...infrastructure.oracle.client import OracleClient
|
||||
from ...infrastructure.oracle.ranking_repository import RankingOracleRepository
|
||||
from ...infrastructure.repositories.consultor_repository_impl import ConsultorRepositoryImpl
|
||||
from ...domain.services.calculador_pontuacao import CalculadorPontuacao
|
||||
from .job_status import job_status
|
||||
|
||||
|
||||
class ProcessarRankingJob:
|
||||
def __init__(
|
||||
self,
|
||||
es_client: ElasticsearchClient,
|
||||
oracle_client: OracleClient,
|
||||
ranking_repo: RankingOracleRepository,
|
||||
):
|
||||
self.es_client = es_client
|
||||
self.oracle_client = oracle_client
|
||||
self.ranking_repo = ranking_repo
|
||||
self.consultor_repo = ConsultorRepositoryImpl(es_client, oracle_client)
|
||||
self.calculador = CalculadorPontuacao()
|
||||
|
||||
async def executar(self, limpar_antes: bool = True) -> Dict[str, Any]:
|
||||
"""
|
||||
Executa o processamento completo do ranking:
|
||||
1. Limpa tabela (se solicitado)
|
||||
2. Scroll por todos os documentos ES
|
||||
3. Para cada batch: calcula pontuação e insere no Oracle
|
||||
4. Atualiza posições
|
||||
5. Retorna estatísticas
|
||||
"""
|
||||
if job_status.is_running:
|
||||
raise RuntimeError("Job já está em execução")
|
||||
|
||||
try:
|
||||
total = await self.es_client.contar_com_atuacoes()
|
||||
job_status.iniciar(total_esperado=total)
|
||||
|
||||
if limpar_antes:
|
||||
job_status.mensagem = "Limpando tabela de ranking..."
|
||||
self.ranking_repo.limpar_tabela()
|
||||
|
||||
job_status.mensagem = "Iniciando processamento via Scroll API..."
|
||||
|
||||
resultado = await self.es_client.buscar_todos_consultores(
|
||||
callback=self._processar_batch,
|
||||
batch_size=1000
|
||||
)
|
||||
|
||||
job_status.mensagem = "Atualizando posições no ranking..."
|
||||
self.ranking_repo.atualizar_posicoes()
|
||||
|
||||
estatisticas = self.ranking_repo.obter_estatisticas()
|
||||
|
||||
job_status.finalizar(sucesso=True)
|
||||
|
||||
return {
|
||||
"sucesso": True,
|
||||
"total_processados": resultado["processados"],
|
||||
"total_batches": resultado["batches"],
|
||||
"tempo_decorrido": job_status.tempo_decorrido,
|
||||
"estatisticas": estatisticas
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
job_status.finalizar(sucesso=False, erro=str(e))
|
||||
raise RuntimeError(f"Erro ao processar ranking: {e}")
|
||||
|
||||
async def _processar_batch(self, docs: list, progress: dict) -> None:
|
||||
"""
|
||||
Processa um batch de documentos:
|
||||
1. Constrói consultores
|
||||
2. Calcula pontuação
|
||||
3. Insere no Oracle
|
||||
4. Atualiza status
|
||||
"""
|
||||
consultores_para_inserir = []
|
||||
|
||||
for doc in docs:
|
||||
try:
|
||||
consultor = await self.consultor_repo._construir_consultor(doc)
|
||||
|
||||
consultor_dict = {
|
||||
"id_pessoa": consultor.id_pessoa,
|
||||
"nome": consultor.nome,
|
||||
"pontuacao_total": consultor.pontuacao_total,
|
||||
"componente_a": consultor.pontuacao.componente_a.total,
|
||||
"componente_b": consultor.pontuacao.componente_b.total,
|
||||
"componente_c": consultor.pontuacao.componente_c.total,
|
||||
"componente_d": consultor.pontuacao.componente_d.total,
|
||||
"ativo": consultor.ativo,
|
||||
"anos_atuacao": consultor.anos_atuacao,
|
||||
"detalhes": self._gerar_json_detalhes(consultor)
|
||||
}
|
||||
|
||||
consultores_para_inserir.append(consultor_dict)
|
||||
|
||||
except Exception as e:
|
||||
print(f"AVISO: Erro ao processar consultor {doc.get('id')}: {e}")
|
||||
continue
|
||||
|
||||
if consultores_para_inserir:
|
||||
self.ranking_repo.inserir_batch(consultores_para_inserir)
|
||||
|
||||
job_status.atualizar_progresso(
|
||||
processados=progress["processados"],
|
||||
batch_atual=progress["batch_atual"],
|
||||
mensagem=f"Processando batch {progress['batch_atual']} ({progress['percentual']}%)"
|
||||
)
|
||||
|
||||
def _gerar_json_detalhes(self, consultor) -> dict:
|
||||
"""
|
||||
Gera JSON com detalhes completos do consultor para armazenar no CLOB.
|
||||
"""
|
||||
return {
|
||||
"id_pessoa": consultor.id_pessoa,
|
||||
"nome": consultor.nome,
|
||||
"cpf": consultor.cpf,
|
||||
"coordenacoes_capes": [
|
||||
{
|
||||
"tipo": c.tipo,
|
||||
"area_avaliacao": c.area_avaliacao,
|
||||
"inicio": c.periodo.inicio.isoformat() if c.periodo.inicio else None,
|
||||
"fim": c.periodo.fim.isoformat() if c.periodo.fim else None,
|
||||
"ativo": c.periodo.ativo
|
||||
}
|
||||
for c in consultor.coordenacoes_capes
|
||||
],
|
||||
"coordenacoes_programas": [
|
||||
{
|
||||
"id_programa": c.id_programa,
|
||||
"nome_programa": c.nome_programa,
|
||||
"codigo_programa": c.codigo_programa,
|
||||
"nota_ppg": c.nota_ppg,
|
||||
"modalidade": c.modalidade,
|
||||
"area_avaliacao": c.area_avaliacao,
|
||||
"inicio": c.periodo.inicio.isoformat() if c.periodo.inicio else None,
|
||||
"fim": c.periodo.fim.isoformat() if c.periodo.fim else None
|
||||
}
|
||||
for c in consultor.coordenacoes_programas
|
||||
],
|
||||
"consultoria": {
|
||||
"total_eventos": consultor.consultoria.total_eventos,
|
||||
"eventos_recentes": consultor.consultoria.eventos_recentes,
|
||||
"situacao": consultor.consultoria.situacao,
|
||||
"anos_completos": consultor.consultoria.anos_completos,
|
||||
"areas": consultor.consultoria.areas
|
||||
} if consultor.consultoria else None,
|
||||
"premiacoes": [
|
||||
{
|
||||
"tipo": p.tipo,
|
||||
"nome_premio": p.nome_premio,
|
||||
"ano": p.ano,
|
||||
"pontos": p.pontos
|
||||
}
|
||||
for p in consultor.premiacoes
|
||||
],
|
||||
"pontuacao": consultor.pontuacao.detalhamento
|
||||
}
|
||||
Reference in New Issue
Block a user