fred/ranking
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Reimplementa sistema de ranking com novos critérios V2

Browse Source 
Mudanças principais:
- Substitui 4 Componentes (A,B,C,D) por 3 Blocos (A,C,D)
- Remove Componente B (Coordenação PPG = 0 pts no V1)
- Adiciona novos tipos de atuação do Elasticsearch
- Implementa critérios de pontuação com tetos individuais

Bloco A - Coordenação CAPES:
- CA (max 450), CAJ (max 370), CAJ_MP (max 315), CAM (max 280)
- Calcula base + tempo + bônus atualidade + bônus retorno

Bloco C - Consultoria:
- CONS_ATIVO (base 150), CONS_HIST (base 100), CONS_FALECIDO (base 100)
- Bônus continuidade: 3anos=+5, 5anos=+10, 8anos=+15
- Bônus retorno: +15

Bloco D - Premiações/Avaliações:
- Inscrições (INSC_AUTOR, INSC_INST)
- Avaliações (AVAL_COMIS_PREMIO, AVAL_COMIS_GP)
- Coordenações (COORD_COMIS_PREMIO, COORD_COMIS_GP)
- Premiações (PREMIACAO, PREMIACAO_GP, MENCAO)
- Bolsas CNPQ, Participações, Orientações, Membros de Banca

Frontend:
- Header, ConsultorCard, CompararModal atualizados para 3 blocos
- API service atualizado para nova estrutura de dados
This commit is contained in:
Frederico Castro
2025-12-13 16:41:55 -03:00
parent 97cd328415
commit 2d4e93f82a
15 changed files with 1517 additions and 1001 deletions
Download Patch File Download Diff File Expand all files Collapse all files

103
backend/src/application/dtos/consultor_dto.py
View File

@@ -1,6 +1,5 @@
from dataclasses import dataclass, asdict
from typing import List, Optional, Dict, Any
from datetime import datetime
@dataclass
@@ -13,6 +12,7 @@ class PeriodoDTO:
@dataclass
class CoordenacaoCapesDTO:
codigo: str
tipo: str
area_avaliacao: str
periodo: PeriodoDTO
@@ -21,55 +21,94 @@ class CoordenacaoCapesDTO:
@dataclass
class CoordenacaoProgramaDTO:
id_programa: int
nome_programa: str
codigo_programa: str
nota_ppg: str
modalidade: str
area_avaliacao: str
class ConsultoriaDTO:
codigo: str
situacao: str
periodo: PeriodoDTO
areas: List[str]
anos_consecutivos: int
retornos: int
@dataclass
class ConsultoriaDTO:
total_eventos: int
eventos_recentes: int
primeiro_evento: str
ultimo_evento: str
continuidade: int
areas: List[str]
class InscricaoDTO:
codigo: str
tipo: str
premio: str
ano: int
situacao: str
anos_completos: int
anos_consecutivos: int
retornos: int
vezes_responsavel: int
@dataclass
class AvaliacaoComissaoDTO:
codigo: str
tipo: str
premio: str
ano: int
comissao_tipo: str
@dataclass
class PremiacaoDTO:
codigo: str
tipo: str
nome_premio: str
ano: int
pontos: int
@dataclass
class ComponentePontuacaoDTO:
class BolsaCNPQDTO:
codigo: str
nivel: str
area: str
@dataclass
class ParticipacaoDTO:
codigo: str
tipo: str
descricao: str
ano: Optional[int]
@dataclass
class OrientacaoDTO:
codigo: str
tipo: str
nivel: str
ano: Optional[int]
@dataclass
class MembroBancaDTO:
codigo: str
tipo: str
nivel: str
ano: Optional[int]
@dataclass
class PontuacaoAtuacaoDTO:
codigo: str
base: int
tempo: int
extras: int
bonus: int
retorno: int
total: int
quantidade: int
@dataclass
class PontuacaoBlocoDTO:
bloco: str
total: int
atuacoes: List[PontuacaoAtuacaoDTO]
@dataclass
class PontuacaoCompletaDTO:
componente_a: ComponentePontuacaoDTO
componente_b: ComponentePontuacaoDTO
componente_c: ComponentePontuacaoDTO
componente_d: ComponentePontuacaoDTO
bloco_a: PontuacaoBlocoDTO
bloco_c: PontuacaoBlocoDTO
bloco_d: PontuacaoBlocoDTO
pontuacao_total: int
@@ -81,6 +120,9 @@ class ConsultorResumoDTO:
ativo: bool
veterano: bool
pontuacao_total: int
bloco_a: int
bloco_c: int
bloco_d: int
rank: Optional[int] = None
@@ -93,9 +135,14 @@ class ConsultorDetalhadoDTO:
ativo: bool
veterano: bool
coordenacoes_capes: List[CoordenacaoCapesDTO]
coordenacoes_programas: List[CoordenacaoProgramaDTO]
consultoria: Optional[ConsultoriaDTO]
inscricoes: List[InscricaoDTO]
avaliacoes_comissao: List[AvaliacaoComissaoDTO]
premiacoes: List[PremiacaoDTO]
bolsas_cnpq: List[BolsaCNPQDTO]
participacoes: List[ParticipacaoDTO]
orientacoes: List[OrientacaoDTO]
membros_banca: List[MembroBancaDTO]
pontuacao: PontuacaoCompletaDTO
rank: Optional[int] = None

113
backend/src/application/jobs/processar_ranking.py
View File

@@ -22,19 +22,10 @@ class ProcessarRankingJob:
self.oracle_remote_client = oracle_remote_client
self.oracle_local_client = oracle_local_client
self.ranking_repo = ranking_repo
# Para acelerar a carga principal, não buscamos PPG aqui (Componente B vem depois)
self.consultor_repo = ConsultorRepositoryImpl(es_client, oracle_client=None)
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")
@@ -73,13 +64,6 @@ class ProcessarRankingJob:
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:
@@ -90,10 +74,10 @@ class ProcessarRankingJob:
"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,
"componente_a": consultor.pontuacao_bloco_a,
"componente_b": 0,
"componente_c": consultor.pontuacao_bloco_c,
"componente_d": consultor.pontuacao_bloco_d,
"ativo": consultor.ativo,
"anos_atuacao": consultor.anos_atuacao,
"detalhes": self._gerar_json_detalhes(consultor)
@@ -117,15 +101,13 @@ class ProcessarRankingJob:
)
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": [
{
"codigo": c.codigo,
"tipo": c.tipo,
"area_avaliacao": c.area_avaliacao,
"inicio": c.periodo.inicio.isoformat() if c.periodo.inicio else None,
@@ -134,37 +116,78 @@ class ProcessarRankingJob:
}
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,
"continuidade": consultor.consultoria.continuidade,
"anos_consecutivos": consultor.consultoria.anos_consecutivos,
"codigo": consultor.consultoria.codigo,
"situacao": consultor.consultoria.situacao,
"anos_completos": consultor.consultoria.anos_completos,
"inicio": consultor.consultoria.periodo.inicio.isoformat() if consultor.consultoria.periodo.inicio else None,
"fim": consultor.consultoria.periodo.fim.isoformat() if consultor.consultoria.periodo.fim else None,
"areas": consultor.consultoria.areas,
"vezes_responsavel": consultor.consultoria.vezes_responsavel
"anos_consecutivos": consultor.consultoria.anos_consecutivos,
"retornos": consultor.consultoria.retornos
} if consultor.consultoria else None,
"inscricoes": [
{
"codigo": i.codigo,
"tipo": i.tipo,
"premio": i.premio,
"ano": i.ano,
"situacao": i.situacao
}
for i in consultor.inscricoes
],
"avaliacoes_comissao": [
{
"codigo": a.codigo,
"tipo": a.tipo,
"premio": a.premio,
"ano": a.ano,
"comissao_tipo": a.comissao_tipo
}
for a in consultor.avaliacoes_comissao
],
"premiacoes": [
{
"codigo": p.codigo,
"tipo": p.tipo,
"nome_premio": p.nome_premio,
"ano": p.ano,
"pontos": p.pontos
"ano": p.ano
}
for p in consultor.premiacoes
],
"pontuacao": consultor.pontuacao.detalhamento
"bolsas_cnpq": [
{
"codigo": b.codigo,
"nivel": b.nivel,
"area": b.area
}
for b in consultor.bolsas_cnpq
],
"participacoes": [
{
"codigo": p.codigo,
"tipo": p.tipo,
"descricao": p.descricao,
"ano": p.ano
}
for p in consultor.participacoes
],
"orientacoes": [
{
"codigo": o.codigo,
"tipo": o.tipo,
"nivel": o.nivel,
"ano": o.ano
}
for o in consultor.orientacoes
],
"membros_banca": [
{
"codigo": m.codigo,
"tipo": m.tipo,
"nivel": m.nivel,
"ano": m.ano
}
for m in consultor.membros_banca
],
"pontuacao": consultor.pontuacao.to_dict() if consultor.pontuacao else None
}

183
backend/src/application/use_cases/obter_ranking.py
View File

@@ -1,5 +1,4 @@
from typing import List, Optional
from datetime import datetime
from ...domain.repositories.consultor_repository import ConsultorRepository
from ...domain.entities.consultor import Consultor
@@ -8,10 +7,16 @@ from ..dtos.consultor_dto import (
ConsultorDetalhadoDTO,
PeriodoDTO,
CoordenacaoCapesDTO,
CoordenacaoProgramaDTO,
ConsultoriaDTO,
InscricaoDTO,
AvaliacaoComissaoDTO,
PremiacaoDTO,
ComponentePontuacaoDTO,
BolsaCNPQDTO,
ParticipacaoDTO,
OrientacaoDTO,
MembroBancaDTO,
PontuacaoAtuacaoDTO,
PontuacaoBlocoDTO,
PontuacaoCompletaDTO,
)
@@ -33,6 +38,9 @@ class ObterRankingUseCase:
ativo=c.ativo,
veterano=c.veterano,
pontuacao_total=c.pontuacao_total,
bloco_a=c.pontuacao_bloco_a,
bloco_c=c.pontuacao_bloco_c,
bloco_d=c.pontuacao_bloco_d,
rank=idx + 1,
)
for idx, c in enumerate(consultores)
@@ -42,7 +50,6 @@ class ObterRankingUseCase:
self, limite: int = 100, componente: Optional[str] = None
) -> List[ConsultorDetalhadoDTO]:
consultores = await self.repository.buscar_ranking(limite=limite, componente=componente)
return [self._converter_para_dto_detalhado(c, idx + 1) for idx, c in enumerate(consultores)]
def _converter_para_dto_detalhado(
@@ -57,10 +64,11 @@ class ObterRankingUseCase:
veterano=consultor.veterano,
coordenacoes_capes=[
CoordenacaoCapesDTO(
codigo=cc.codigo,
tipo=cc.tipo,
area_avaliacao=cc.area_avaliacao,
periodo=PeriodoDTO(
inicio=cc.periodo.inicio.isoformat(),
inicio=cc.periodo.inicio.isoformat() if cc.periodo.inicio else "",
fim=cc.periodo.fim.isoformat() if cc.periodo.fim else None,
ativo=cc.periodo.ativo,
anos_decorridos=cc.periodo.anos_decorridos,
@@ -70,79 +78,128 @@ class ObterRankingUseCase:
)
for cc in consultor.coordenacoes_capes
],
coordenacoes_programas=[
CoordenacaoProgramaDTO(
id_programa=cp.id_programa,
nome_programa=cp.nome_programa,
codigo_programa=cp.codigo_programa,
nota_ppg=cp.nota_ppg,
modalidade=cp.modalidade,
area_avaliacao=cp.area_avaliacao,
periodo=PeriodoDTO(
inicio=cp.periodo.inicio.isoformat(),
fim=cp.periodo.fim.isoformat() if cp.periodo.fim else None,
ativo=cp.periodo.ativo,
anos_decorridos=cp.periodo.anos_decorridos,
),
)
for cp in consultor.coordenacoes_programas
],
consultoria=ConsultoriaDTO(
total_eventos=consultor.consultoria.total_eventos,
eventos_recentes=consultor.consultoria.eventos_recentes,
primeiro_evento=consultor.consultoria.primeiro_evento.isoformat(),
ultimo_evento=consultor.consultoria.ultimo_evento.isoformat(),
continuidade=consultor.consultoria.continuidade,
areas=consultor.consultoria.areas,
codigo=consultor.consultoria.codigo,
situacao=consultor.consultoria.situacao,
anos_completos=consultor.consultoria.anos_completos,
periodo=PeriodoDTO(
inicio=consultor.consultoria.periodo.inicio.isoformat() if consultor.consultoria.periodo.inicio else "",
fim=consultor.consultoria.periodo.fim.isoformat() if consultor.consultoria.periodo.fim else None,
ativo=consultor.consultoria.periodo.ativo,
anos_decorridos=consultor.consultoria.periodo.anos_decorridos,
),
areas=consultor.consultoria.areas,
anos_consecutivos=consultor.consultoria.anos_consecutivos,
retornos=consultor.consultoria.retornos,
vezes_responsavel=consultor.consultoria.vezes_responsavel,
)
if consultor.consultoria
else None,
) if consultor.consultoria else None,
inscricoes=[
InscricaoDTO(
codigo=i.codigo,
tipo=i.tipo,
premio=i.premio,
ano=i.ano,
situacao=i.situacao,
)
for i in consultor.inscricoes
],
avaliacoes_comissao=[
AvaliacaoComissaoDTO(
codigo=a.codigo,
tipo=a.tipo,
premio=a.premio,
ano=a.ano,
comissao_tipo=a.comissao_tipo,
)
for a in consultor.avaliacoes_comissao
],
premiacoes=[
PremiacaoDTO(
codigo=p.codigo,
tipo=p.tipo,
nome_premio=p.nome_premio,
ano=p.ano,
pontos=p.pontos,
)
for p in consultor.premiacoes
],
bolsas_cnpq=[
BolsaCNPQDTO(
codigo=b.codigo,
nivel=b.nivel,
area=b.area,
)
for b in consultor.bolsas_cnpq
],
participacoes=[
ParticipacaoDTO(
codigo=p.codigo,
tipo=p.tipo,
descricao=p.descricao,
ano=p.ano,
)
for p in consultor.participacoes
],
orientacoes=[
OrientacaoDTO(
codigo=o.codigo,
tipo=o.tipo,
nivel=o.nivel,
ano=o.ano,
)
for o in consultor.orientacoes
],
membros_banca=[
MembroBancaDTO(
codigo=m.codigo,
tipo=m.tipo,
nivel=m.nivel,
ano=m.ano,
)
for m in consultor.membros_banca
],
pontuacao=PontuacaoCompletaDTO(
componente_a=ComponentePontuacaoDTO(
base=consultor.pontuacao.componente_a.base,
tempo=consultor.pontuacao.componente_a.tempo,
extras=consultor.pontuacao.componente_a.extras,
bonus=consultor.pontuacao.componente_a.bonus,
retorno=consultor.pontuacao.componente_a.retorno,
total=consultor.pontuacao.componente_a.total,
bloco_a=PontuacaoBlocoDTO(
bloco="A",
total=consultor.pontuacao.bloco_a.total,
atuacoes=[
PontuacaoAtuacaoDTO(
codigo=a.codigo,
base=a.base,
tempo=a.tempo,
bonus=a.bonus,
total=a.total,
quantidade=a.quantidade,
)
for a in consultor.pontuacao.bloco_a.atuacoes
],
),
componente_b=ComponentePontuacaoDTO(
base=consultor.pontuacao.componente_b.base,
tempo=consultor.pontuacao.componente_b.tempo,
extras=consultor.pontuacao.componente_b.extras,
bonus=consultor.pontuacao.componente_b.bonus,
retorno=0,
total=consultor.pontuacao.componente_b.total,
bloco_c=PontuacaoBlocoDTO(
bloco="C",
total=consultor.pontuacao.bloco_c.total,
atuacoes=[
PontuacaoAtuacaoDTO(
codigo=a.codigo,
base=a.base,
tempo=a.tempo,
bonus=a.bonus,
total=a.total,
quantidade=a.quantidade,
)
for a in consultor.pontuacao.bloco_c.atuacoes
],
),
componente_c=ComponentePontuacaoDTO(
base=consultor.pontuacao.componente_c.base,
tempo=consultor.pontuacao.componente_c.tempo,
extras=consultor.pontuacao.componente_c.extras,
bonus=consultor.pontuacao.componente_c.bonus,
retorno=0,
total=consultor.pontuacao.componente_c.total,
),
componente_d=ComponentePontuacaoDTO(
base=consultor.pontuacao.componente_d.base,
tempo=consultor.pontuacao.componente_d.tempo,
extras=consultor.pontuacao.componente_d.extras,
bonus=consultor.pontuacao.componente_d.bonus,
retorno=0,
total=consultor.pontuacao.componente_d.total,
bloco_d=PontuacaoBlocoDTO(
bloco="D",
total=consultor.pontuacao.bloco_d.total,
atuacoes=[
PontuacaoAtuacaoDTO(
codigo=a.codigo,
base=a.base,
tempo=a.tempo,
bonus=a.bonus,
total=a.total,
quantidade=a.quantidade,
)
for a in consultor.pontuacao.bloco_d.atuacoes
],
),
pontuacao_total=consultor.pontuacao.total,
),

134
backend/src/domain/entities/consultor.py
View File

@@ -1,13 +1,34 @@
from dataclasses import dataclass, field
from typing import List, Optional
from typing import List, Optional, Dict, Any
from datetime import datetime
from ..value_objects.periodo import Periodo
from ..value_objects.pontuacao import PontuacaoCompleta
@dataclass
class Atuacao:
codigo: str
tipo_es: str
inicio: Optional[datetime] = None
fim: Optional[datetime] = None
dados: Dict[str, Any] = field(default_factory=dict)
@property
def ativo(self) -> bool:
return self.fim is None
@property
def anos_completos(self) -> int:
if not self.inicio:
return 0
fim = self.fim or datetime.now()
return int((fim - self.inicio).days // 365)
@dataclass
class CoordenacaoCapes:
codigo: str
tipo: str
area_avaliacao: str
periodo: Periodo
@@ -16,46 +37,70 @@ class CoordenacaoCapes:
@dataclass
class CoordenacaoPrograma:
id_programa: int
nome_programa: str
codigo_programa: str
nota_ppg: str
modalidade: str
area_avaliacao: str
class Consultoria:
codigo: str
situacao: str
periodo: Periodo
areas: List[str] = field(default_factory=list)
anos_consecutivos: int = 0
retornos: int = 0
@dataclass
class Consultoria:
total_eventos: int
eventos_recentes: int
primeiro_evento: datetime
ultimo_evento: datetime
areas: List[str] = field(default_factory=list)
situacao: str = "N/A"
anos_completos: int = 0
anos_consecutivos: int = 0
retornos: int = 0
vezes_responsavel: int = 0
class Inscricao:
codigo: str
tipo: str
premio: str
ano: int
situacao: str = ""
@property
def continuidade(self) -> int:
if self.anos_consecutivos >= 8:
return 15
elif self.anos_consecutivos >= 5:
return 10
elif self.anos_consecutivos >= 3:
return 5
return 0
@dataclass
class AvaliacaoComissao:
codigo: str
tipo: str
premio: str
ano: int
comissao_tipo: str = ""
@dataclass
class Premiacao:
codigo: str
tipo: str
nome_premio: str
ano: int
pontos: int
@dataclass
class BolsaCNPQ:
codigo: str
nivel: str
area: str = ""
@dataclass
class Participacao:
codigo: str
tipo: str
descricao: str = ""
ano: Optional[int] = None
@dataclass
class Orientacao:
codigo: str
tipo: str
nivel: str
ano: Optional[int] = None
@dataclass
class MembroBanca:
codigo: str
tipo: str
nivel: str
ano: Optional[int] = None
@dataclass
@@ -64,28 +109,29 @@ class Consultor:
nome: str
cpf: Optional[str] = None
coordenacoes_capes: List[CoordenacaoCapes] = field(default_factory=list)
coordenacoes_programas: List[CoordenacaoPrograma] = field(default_factory=list)
consultoria: Optional[Consultoria] = None
inscricoes: List[Inscricao] = field(default_factory=list)
avaliacoes_comissao: List[AvaliacaoComissao] = field(default_factory=list)
premiacoes: List[Premiacao] = field(default_factory=list)
bolsas_cnpq: List[BolsaCNPQ] = field(default_factory=list)
participacoes: List[Participacao] = field(default_factory=list)
orientacoes: List[Orientacao] = field(default_factory=list)
membros_banca: List[MembroBanca] = field(default_factory=list)
pontuacao: Optional[PontuacaoCompleta] = None
atuacoes_raw: List[Atuacao] = field(default_factory=list)
@property
def anos_atuacao(self) -> float:
if not self.consultoria:
if not self.consultoria or not self.consultoria.periodo.inicio:
return 0.0
dias = (datetime.now() - self.consultoria.primeiro_evento).days
dias = (datetime.now() - self.consultoria.periodo.inicio).days
return round(dias / 365.25, 1)
@property
def ativo(self) -> bool:
if not self.consultoria:
return False
situacao = (self.consultoria.situacao or "").lower()
if "atividade" in situacao:
return True
if "inativ" in situacao:
return False
return self.consultoria.eventos_recentes > 0
return self.consultoria.codigo == "CONS_ATIVO"
@property
def veterano(self) -> bool:
@@ -94,3 +140,15 @@ class Consultor:
@property
def pontuacao_total(self) -> int:
return self.pontuacao.total if self.pontuacao else 0
@property
def pontuacao_bloco_a(self) -> int:
return self.pontuacao.bloco_a.total if self.pontuacao else 0
@property
def pontuacao_bloco_c(self) -> int:
return self.pontuacao.bloco_c.total if self.pontuacao else 0
@property
def pontuacao_bloco_d(self) -> int:
return self.pontuacao.bloco_d.total if self.pontuacao else 0

313
backend/src/domain/services/calculador_pontuacao.py
View File

@@ -1,44 +1,44 @@
from datetime import datetime
from typing import List
from typing import List, Dict
from collections import defaultdict
from ..entities.consultor import (
Consultor,
CoordenacaoCapes,
CoordenacaoPrograma,
Consultoria,
Inscricao,
AvaliacaoComissao,
Premiacao,
BolsaCNPQ,
Participacao,
Orientacao,
MembroBanca,
)
from ..value_objects.pontuacao import ComponentePontuacao, PontuacaoCompleta
from ..value_objects.pontuacao import PontuacaoAtuacao, PontuacaoBloco, PontuacaoCompleta
from ..value_objects.criterios_pontuacao import CRITERIOS, get_criterio, Bloco
from ..value_objects.periodo import Periodo
class CalculadorPontuacao:
@staticmethod
def _mesclar_periodos(periodos: List[Periodo]) -> List[Periodo]:
"""
Mescla períodos sobrepostos/contíguos para evitar contagem dupla.
"""
if not periodos:
return []
periodos_ordenados = sorted(periodos, key=lambda p: p.inicio)
periodos_ordenados = sorted(periodos, key=lambda p: p.inicio if p.inicio else datetime.min)
mesclados: List[Periodo] = []
for p in periodos_ordenados:
if not mesclados:
mesclados.append(p)
continue
ultimo = mesclados[-1]
ultimo_fim = ultimo.fim or datetime.now()
atual_fim = p.fim or datetime.now()
if p.inicio <= ultimo_fim:
if p.inicio and p.inicio <= ultimo_fim:
novo_fim = max(ultimo_fim, atual_fim)
mesclados[-1] = Periodo(inicio=ultimo.inicio, fim=novo_fim if not ultimo.ativo else None)
else:
mesclados.append(p)
return mesclados
@staticmethod
@@ -47,164 +47,183 @@ class CalculadorPontuacao:
return sum(p.anos_completos(ref) for p in periodos)
@staticmethod
def calcular_componente_a(coordenacoes: List[CoordenacaoCapes]) -> ComponentePontuacao:
def calcular_bloco_a(coordenacoes: List[CoordenacaoCapes]) -> PontuacaoBloco:
if not coordenacoes:
return ComponentePontuacao(base=0, tempo=0, extras=0, bonus=0, retorno=0)
return PontuacaoBloco(bloco="A", atuacoes=[])
base_map = {"CA": 200, "CAJ": 150, "CAJ-MP": 120, "CAM": 100}
tempo_max_map = {"CA": 100, "CAJ": 80, "CAJ-MP": 60, "CAM": 50}
bonus_atual_map = {"CA": 30, "CAJ": 20, "CAJ-MP": 15, "CAM": 10}
mult_tempo_map = {"CA": 10, "CAJ": 8, "CAJ-MP": 6, "CAM": 5}
# Agrupa por tipo de coordenação e considera o melhor tipo (hierarquia)
tipos_ordenados = ["CA", "CAJ", "CAJ-MP", "CAM"]
coord_por_tipo = {t: [] for t in tipos_ordenados}
tipos_ordenados = ["CA", "CAJ", "CAJ_MP", "CAM"]
coord_por_tipo: Dict[str, List[CoordenacaoCapes]] = defaultdict(list)
for c in coordenacoes:
coord_por_tipo.setdefault(c.tipo, []).append(c)
codigo = c.codigo.replace("-", "_")
coord_por_tipo[codigo].append(c)
coord_escolhida_tipo = None
for t in tipos_ordenados:
if coord_por_tipo.get(t):
coord_escolhida_tipo = t
break
if not coord_escolhida_tipo:
return ComponentePontuacao(base=0, tempo=0, extras=0, bonus=0, retorno=0)
coord_do_tipo = coord_por_tipo.get(coord_escolhida_tipo, [])
coord_por_area = {}
for c in coord_do_tipo:
area = c.area_avaliacao or "N/A"
coord_por_area.setdefault(area, []).append(c.periodo)
anos_total = 0
ativo = False
retornos_encontrados = 0
for _, periodos_area in coord_por_area.items():
mesclados = CalculadorPontuacao._mesclar_periodos(periodos_area)
anos_total += CalculadorPontuacao._anos_completos_periodos(mesclados)
ativo = ativo or any(p.ativo for p in periodos_area)
if len(mesclados) > 1:
retornos_encontrados += 1
base = base_map.get(coord_escolhida_tipo, 0)
tempo = min(int(anos_total * mult_tempo_map.get(coord_escolhida_tipo, 0)), tempo_max_map.get(coord_escolhida_tipo, 0))
extras = 0
areas_distintas = list(coord_por_area.keys())
if len(areas_distintas) > 1:
extras = min((len(areas_distintas) - 1) * 20, 100)
bonus = bonus_atual_map.get(coord_escolhida_tipo, 0) if ativo else 0
retorno = 20 if retornos_encontrados > 0 else 0
return ComponentePontuacao(base=base, tempo=tempo, extras=extras, bonus=bonus, retorno=retorno, teto=450)
@staticmethod
def calcular_componente_b(coordenacoes: List[CoordenacaoPrograma]) -> ComponentePontuacao:
"""
Calcula pontuação do Componente B (Coordenação de Programa PPG).
Regras (máximo 180 pts):
- Base: 70 pts por ser coordenador
- Tempo: 5 pts/ano (máx 50)
- Programas adicionais: 20 pts/programa (máx 40)
- Nota do PPG: usar MAIOR nota (7=20, 6=15, 5=10, 4=5, 3=0)
"""
if not coordenacoes:
return ComponentePontuacao(base=0, tempo=0, extras=0, bonus=0)
# Base: 70 pts por ser coordenador
base = 70
# Tempo: 5 pts/ano (máx 50)
anos_totais = sum(c.periodo.anos_completos(datetime.now()) for c in coordenacoes)
tempo = min(int(anos_totais * 5), 50)
# Programas adicionais: 20 pts/programa extra (máx 40)
programas_distintos = len({c.id_programa for c in coordenacoes})
extras = min((programas_distintos - 1) * 20, 40) if programas_distintos > 1 else 0
# Nota do PPG: usar MAIOR nota entre os programas
# Escala: 7=20, 6=15, 5=10, 4=5, 3=0
maior_nota = 0
for c in coordenacoes:
try:
nota_str = str(c.nota_ppg).strip()
# Trata notas válidas (3-7 ou A para programas novos)
if nota_str in ['7']:
maior_nota = max(maior_nota, 7)
elif nota_str in ['6']:
maior_nota = max(maior_nota, 6)
elif nota_str in ['5']:
maior_nota = max(maior_nota, 5)
elif nota_str in ['4']:
maior_nota = max(maior_nota, 4)
elif nota_str in ['3']:
maior_nota = max(maior_nota, 3)
except Exception:
atuacoes = []
for tipo in tipos_ordenados:
if tipo not in coord_por_tipo:
continue
# Mapeia nota para pontos
mapa_nota = {7: 20, 6: 15, 5: 10, 4: 5, 3: 0}
bonus = mapa_nota.get(maior_nota, 0)
criterio = get_criterio(tipo)
if not criterio:
continue
return ComponentePontuacao(base=base, tempo=tempo, extras=extras, bonus=bonus, teto=180)
coords = coord_por_tipo[tipo]
periodos = [c.periodo for c in coords]
mesclados = CalculadorPontuacao._mesclar_periodos(periodos)
anos_total = CalculadorPontuacao._anos_completos_periodos(mesclados)
ativo = any(c.periodo.ativo for c in coords)
tem_retorno = len(mesclados) > 1
base = criterio.base
tempo = min(anos_total * criterio.multiplicador_tempo, criterio.teto_tempo)
bonus_atualidade = criterio.bonus_atualidade if ativo else 0
bonus_retorno = criterio.bonus_retorno if tem_retorno else 0
bonus = bonus_atualidade + bonus_retorno
total_bruto = base + tempo + bonus
total = min(total_bruto, criterio.teto)
atuacoes.append(PontuacaoAtuacao(
codigo=tipo,
base=base,
tempo=tempo,
bonus=bonus,
total=total,
quantidade=len(coords),
))
return PontuacaoBloco(bloco="A", atuacoes=atuacoes)
@staticmethod
def calcular_componente_c(consultoria: Consultoria) -> ComponentePontuacao:
def calcular_bloco_c(consultoria: Consultoria) -> PontuacaoBloco:
if not consultoria:
return ComponentePontuacao(base=0, tempo=0, extras=0, bonus=0)
return PontuacaoBloco(bloco="C", atuacoes=[])
base = 150 if consultoria.eventos_recentes > 0 else 100
codigo = consultoria.codigo
criterio = get_criterio(codigo)
if not criterio:
return PontuacaoBloco(bloco="C", atuacoes=[])
anos = consultoria.anos_completos if consultoria.anos_completos else int(
((datetime.now() - consultoria.primeiro_evento).days) // 365
)
tempo = min(int(anos * 5), 50)
base = criterio.base
extras_eventos = min(consultoria.total_eventos * 2, 20)
extras_responsavel = min(consultoria.vezes_responsavel * 5, 25)
extras_areas = min((len(consultoria.areas) - 1) * 10, 30) if len(consultoria.areas) > 1 else 0
extras = extras_eventos + extras_responsavel + extras_areas
tempo = 0
if criterio.pontua_tempo and consultoria.periodo.inicio:
anos = consultoria.periodo.anos_completos(datetime.now())
tempo = min(anos * criterio.multiplicador_tempo, criterio.teto_tempo)
continuidade = consultoria.anos_consecutivos
if continuidade >= 8:
bonus_continuidade = 15
elif continuidade >= 5:
bonus_continuidade = 10
elif continuidade >= 3:
bonus_continuidade = 5
else:
bonus_continuidade = 0
bonus = 0
if codigo == "CONS_ATIVO":
if consultoria.anos_consecutivos >= 8:
bonus += criterio.bonus_continuidade_8anos
elif consultoria.anos_consecutivos >= 5:
bonus += criterio.bonus_continuidade_5anos
elif consultoria.anos_consecutivos >= 3:
bonus += criterio.bonus_continuidade_3anos
if consultoria.retornos > 0:
bonus += criterio.bonus_retorno
retorno_bonus = 15 if consultoria.retornos > 0 else 0
bonus = bonus_continuidade + retorno_bonus
total_bruto = base + tempo + bonus
total = min(total_bruto, criterio.teto)
return ComponentePontuacao(base=base, tempo=tempo, extras=extras, bonus=bonus, teto=230)
atuacoes = [PontuacaoAtuacao(
codigo=codigo,
base=base,
tempo=tempo,
bonus=bonus,
total=total,
quantidade=1,
)]
return PontuacaoBloco(bloco="C", atuacoes=atuacoes)
@staticmethod
def calcular_componente_d(premiacoes: List[Premiacao]) -> ComponentePontuacao:
if not premiacoes:
return ComponentePontuacao(base=0, tempo=0, extras=0, bonus=0)
def calcular_bloco_d(
inscricoes: List[Inscricao],
avaliacoes: List[AvaliacaoComissao],
premiacoes: List[Premiacao],
bolsas: List[BolsaCNPQ],
participacoes: List[Participacao],
orientacoes: List[Orientacao],
membros_banca: List[MembroBanca],
) -> PontuacaoBloco:
atuacoes = []
totais_por_codigo: Dict[str, Dict] = defaultdict(lambda: {"base": 0, "qtd": 0})
avaliador = [p for p in premiacoes if "avaliador" in (p.tipo or "").lower()]
outros = [p for p in premiacoes if p not in avaliador]
for insc in inscricoes:
criterio = get_criterio(insc.codigo)
if criterio:
totais_por_codigo[insc.codigo]["base"] += criterio.base
totais_por_codigo[insc.codigo]["qtd"] += 1
pontos_avaliador = min(sum(p.pontos for p in avaliador), 20)
total_pontos = pontos_avaliador + sum(p.pontos for p in outros)
total_pontos = min(total_pontos, 180)
for aval in avaliacoes:
criterio = get_criterio(aval.codigo)
if criterio:
totais_por_codigo[aval.codigo]["base"] += criterio.base
totais_por_codigo[aval.codigo]["qtd"] += 1
return ComponentePontuacao(base=total_pontos, tempo=0, extras=0, bonus=0, teto=180)
for prem in premiacoes:
criterio = get_criterio(prem.codigo)
if criterio:
totais_por_codigo[prem.codigo]["base"] += criterio.base
totais_por_codigo[prem.codigo]["qtd"] += 1
for bolsa in bolsas:
criterio = get_criterio(bolsa.codigo)
if criterio:
totais_por_codigo[bolsa.codigo]["base"] += criterio.base
totais_por_codigo[bolsa.codigo]["qtd"] += 1
for part in participacoes:
criterio = get_criterio(part.codigo)
if criterio:
totais_por_codigo[part.codigo]["base"] += criterio.base
totais_por_codigo[part.codigo]["qtd"] += 1
for orient in orientacoes:
criterio = get_criterio(orient.codigo)
if criterio:
totais_por_codigo[orient.codigo]["base"] += criterio.base
totais_por_codigo[orient.codigo]["qtd"] += 1
for mb in membros_banca:
criterio = get_criterio(mb.codigo)
if criterio:
totais_por_codigo[mb.codigo]["base"] += criterio.base
totais_por_codigo[mb.codigo]["qtd"] += 1
for codigo, dados in totais_por_codigo.items():
criterio = get_criterio(codigo)
if not criterio:
continue
total = min(dados["base"], criterio.teto)
atuacoes.append(PontuacaoAtuacao(
codigo=codigo,
base=dados["base"],
tempo=0,
bonus=0,
total=total,
quantidade=dados["qtd"],
))
return PontuacaoBloco(bloco="D", atuacoes=atuacoes)
@classmethod
def calcular_pontuacao_completa(cls, consultor: Consultor) -> PontuacaoCompleta:
comp_a = cls.calcular_componente_a(consultor.coordenacoes_capes)
comp_b = cls.calcular_componente_b(consultor.coordenacoes_programas)
comp_c = cls.calcular_componente_c(consultor.consultoria)
comp_d = cls.calcular_componente_d(consultor.premiacoes)
bloco_a = cls.calcular_bloco_a(consultor.coordenacoes_capes)
bloco_c = cls.calcular_bloco_c(consultor.consultoria)
bloco_d = cls.calcular_bloco_d(
inscricoes=consultor.inscricoes,
avaliacoes=consultor.avaliacoes_comissao,
premiacoes=consultor.premiacoes,
bolsas=consultor.bolsas_cnpq,
participacoes=consultor.participacoes,
orientacoes=consultor.orientacoes,
membros_banca=consultor.membros_banca,
)
return PontuacaoCompleta(
componente_a=comp_a, componente_b=comp_b, componente_c=comp_c, componente_d=comp_d
bloco_a=bloco_a,
bloco_c=bloco_c,
bloco_d=bloco_d,
)

289
backend/src/domain/value_objects/criterios_pontuacao.py Normal file
View File

@@ -0,0 +1,289 @@
from dataclasses import dataclass
from typing import Dict, Optional
from enum import Enum
class Bloco(Enum):
A = "A"
C = "C"
D = "D"
class TipoAtuacao(Enum):
FUNCAO = "Função"
RESULTADO = "Resultado"
PAPEL = "Papel"
PARTICIPACAO = "Participação"
@dataclass(frozen=True)
class CriterioPontuacao:
codigo: str
bloco: Bloco
tipo: TipoAtuacao
base: int
teto: int
pontua_tempo: bool = False
multiplicador_tempo: int = 0
teto_tempo: int = 0
bonus_atualidade: int = 0
bonus_retorno: int = 0
bonus_continuidade_3anos: int = 0
bonus_continuidade_5anos: int = 0
bonus_continuidade_8anos: int = 0
CRITERIOS: Dict[str, CriterioPontuacao] = {
"CA": CriterioPontuacao(
codigo="CA",
bloco=Bloco.A,
tipo=TipoAtuacao.FUNCAO,
base=200,
teto=450,
pontua_tempo=True,
multiplicador_tempo=10,
teto_tempo=100,
bonus_atualidade=30,
bonus_retorno=20,
),
"CAJ": CriterioPontuacao(
codigo="CAJ",
bloco=Bloco.A,
tipo=TipoAtuacao.FUNCAO,
base=150,
teto=370,
pontua_tempo=True,
multiplicador_tempo=8,
teto_tempo=80,
bonus_atualidade=20,
bonus_retorno=20,
),
"CAJ_MP": CriterioPontuacao(
codigo="CAJ_MP",
bloco=Bloco.A,
tipo=TipoAtuacao.FUNCAO,
base=120,
teto=315,
pontua_tempo=True,
multiplicador_tempo=6,
teto_tempo=60,
bonus_atualidade=15,
bonus_retorno=20,
),
"CAM": CriterioPontuacao(
codigo="CAM",
bloco=Bloco.A,
tipo=TipoAtuacao.FUNCAO,
base=100,
teto=280,
pontua_tempo=True,
multiplicador_tempo=5,
teto_tempo=50,
bonus_atualidade=10,
bonus_retorno=20,
),
"PPG_COORD": CriterioPontuacao(
codigo="PPG_COORD",
bloco=Bloco.A,
tipo=TipoAtuacao.FUNCAO,
base=0,
teto=0,
pontua_tempo=True,
multiplicador_tempo=0,
teto_tempo=0,
),
"CONS_ATIVO": CriterioPontuacao(
codigo="CONS_ATIVO",
bloco=Bloco.C,
tipo=TipoAtuacao.FUNCAO,
base=150,
teto=230,
pontua_tempo=True,
multiplicador_tempo=5,
teto_tempo=50,
bonus_continuidade_3anos=5,
bonus_continuidade_5anos=10,
bonus_continuidade_8anos=15,
bonus_retorno=15,
),
"CONS_HIST": CriterioPontuacao(
codigo="CONS_HIST",
bloco=Bloco.C,
tipo=TipoAtuacao.FUNCAO,
base=100,
teto=230,
pontua_tempo=True,
multiplicador_tempo=5,
teto_tempo=50,
),
"CONS_FALECIDO": CriterioPontuacao(
codigo="CONS_FALECIDO",
bloco=Bloco.C,
tipo=TipoAtuacao.FUNCAO,
base=100,
teto=230,
pontua_tempo=False,
),
"INSC_AUTOR": CriterioPontuacao(
codigo="INSC_AUTOR",
bloco=Bloco.D,
tipo=TipoAtuacao.PAPEL,
base=10,
teto=20,
),
"INSC_INST": CriterioPontuacao(
codigo="INSC_INST",
bloco=Bloco.D,
tipo=TipoAtuacao.PAPEL,
base=30,
teto=60,
),
"AVAL_COMIS_PREMIO": CriterioPontuacao(
codigo="AVAL_COMIS_PREMIO",
bloco=Bloco.D,
tipo=TipoAtuacao.FUNCAO,
base=30,
teto=60,
),
"AVAL_COMIS_GP": CriterioPontuacao(
codigo="AVAL_COMIS_GP",
bloco=Bloco.D,
tipo=TipoAtuacao.FUNCAO,
base=50,
teto=100,
),
"COORD_COMIS_PREMIO": CriterioPontuacao(
codigo="COORD_COMIS_PREMIO",
bloco=Bloco.D,
tipo=TipoAtuacao.FUNCAO,
base=50,
teto=100,
),
"COORD_COMIS_GP": CriterioPontuacao(
codigo="COORD_COMIS_GP",
bloco=Bloco.D,
tipo=TipoAtuacao.FUNCAO,
base=60,
teto=120,
),
"BOL_BPQ_SUPERIOR": CriterioPontuacao(
codigo="BOL_BPQ_SUPERIOR",
bloco=Bloco.D,
tipo=TipoAtuacao.RESULTADO,
base=30,
teto=60,
),
"BOL_BPQ_INTERMEDIARIO": CriterioPontuacao(
codigo="BOL_BPQ_INTERMEDIARIO",
bloco=Bloco.D,
tipo=TipoAtuacao.RESULTADO,
base=50,
teto=100,
),
"PREMIACAO": CriterioPontuacao(
codigo="PREMIACAO",
bloco=Bloco.D,
tipo=TipoAtuacao.RESULTADO,
base=150,
teto=180,
),
"PREMIACAO_GP": CriterioPontuacao(
codigo="PREMIACAO_GP",
bloco=Bloco.D,
tipo=TipoAtuacao.RESULTADO,
base=30,
teto=60,
),
"MENCAO": CriterioPontuacao(
codigo="MENCAO",
bloco=Bloco.D,
tipo=TipoAtuacao.RESULTADO,
base=10,
teto=20,
),
"EVENTO": CriterioPontuacao(
codigo="EVENTO",
bloco=Bloco.D,
tipo=TipoAtuacao.PARTICIPACAO,
base=1,
teto=5,
),
"PROJ": CriterioPontuacao(
codigo="PROJ",
bloco=Bloco.D,
tipo=TipoAtuacao.PARTICIPACAO,
base=10,
teto=40,
),
"ORIENT_POS_DOC": CriterioPontuacao(
codigo="ORIENT_POS_DOC",
bloco=Bloco.D,
tipo=TipoAtuacao.PARTICIPACAO,
base=15,
teto=100,
),
"ORIENT_TESE": CriterioPontuacao(
codigo="ORIENT_TESE",
bloco=Bloco.D,
tipo=TipoAtuacao.PARTICIPACAO,
base=10,
teto=50,
),
"ORIENT_DISS": CriterioPontuacao(
codigo="ORIENT_DISS",
bloco=Bloco.D,
tipo=TipoAtuacao.PARTICIPACAO,
base=5,
teto=25,
),
"CO_ORIENT_POS_DOC": CriterioPontuacao(
codigo="CO_ORIENT_POS_DOC",
bloco=Bloco.D,
tipo=TipoAtuacao.PARTICIPACAO,
base=7,
teto=35,
),
"CO_ORIENT_TESE": CriterioPontuacao(
codigo="CO_ORIENT_TESE",
bloco=Bloco.D,
tipo=TipoAtuacao.PARTICIPACAO,
base=5,
teto=25,
),
"CO_ORIENT_DISS": CriterioPontuacao(
codigo="CO_ORIENT_DISS",
bloco=Bloco.D,
tipo=TipoAtuacao.PARTICIPACAO,
base=3,
teto=15,
),
"MB_BANCA_POS_DOC": CriterioPontuacao(
codigo="MB_BANCA_POS_DOC",
bloco=Bloco.D,
tipo=TipoAtuacao.PARTICIPACAO,
base=3,
teto=15,
),
"MB_BANCA_TESE": CriterioPontuacao(
codigo="MB_BANCA_TESE",
bloco=Bloco.D,
tipo=TipoAtuacao.PARTICIPACAO,
base=3,
teto=15,
),
"MB_BANCA_DISS": CriterioPontuacao(
codigo="MB_BANCA_DISS",
bloco=Bloco.D,
tipo=TipoAtuacao.PARTICIPACAO,
base=2,
teto=10,
),
}
def get_criterio(codigo: str) -> Optional[CriterioPontuacao]:
return CRITERIOS.get(codigo)
def get_criterios_bloco(bloco: Bloco) -> Dict[str, CriterioPontuacao]:
return {k: v for k, v in CRITERIOS.items() if v.bloco == bloco}

93
backend/src/domain/value_objects/pontuacao.py
View File

@@ -1,71 +1,62 @@
from dataclasses import dataclass
from typing import Dict
from dataclasses import dataclass, field
from typing import Dict, List
@dataclass(frozen=True)
class ComponentePontuacao:
class PontuacaoAtuacao:
codigo: str
base: int
tempo: int
extras: int
bonus: int
retorno: int = 0
teto: int = 0
total: int
quantidade: int = 1
@dataclass(frozen=True)
class PontuacaoBloco:
bloco: str
atuacoes: List[PontuacaoAtuacao] = field(default_factory=list)
@property
def total(self) -> int:
soma = self.base + self.tempo + self.extras + self.bonus + self.retorno
if self.teto > 0:
return min(soma, self.teto)
return soma
return sum(a.total for a in self.atuacoes)
def to_dict(self) -> Dict:
return {
"bloco": self.bloco,
"total": self.total,
"atuacoes": [
{
"codigo": a.codigo,
"base": a.base,
"tempo": a.tempo,
"bonus": a.bonus,
"total": a.total,
"quantidade": a.quantidade,
}
for a in self.atuacoes
],
}
@dataclass(frozen=True)
class PontuacaoCompleta:
componente_a: ComponentePontuacao
componente_b: ComponentePontuacao
componente_c: ComponentePontuacao
componente_d: ComponentePontuacao
bloco_a: PontuacaoBloco
bloco_c: PontuacaoBloco
bloco_d: PontuacaoBloco
@property
def total(self) -> int:
return (
self.componente_a.total
+ self.componente_b.total
+ self.componente_c.total
+ self.componente_d.total
)
return self.bloco_a.total + self.bloco_c.total + self.bloco_d.total
@property
def detalhamento(self) -> Dict[str, Dict[str, int]]:
def to_dict(self) -> Dict:
return {
"componente_a": {
"base": self.componente_a.base,
"tempo": self.componente_a.tempo,
"extras": self.componente_a.extras,
"bonus": self.componente_a.bonus,
"retorno": self.componente_a.retorno,
"total": self.componente_a.total,
},
"componente_b": {
"base": self.componente_b.base,
"tempo": self.componente_b.tempo,
"extras": self.componente_b.extras,
"bonus": self.componente_b.bonus,
"total": self.componente_b.total,
},
"componente_c": {
"base": self.componente_c.base,
"tempo": self.componente_c.tempo,
"extras": self.componente_c.extras,
"bonus": self.componente_c.bonus,
"total": self.componente_c.total,
},
"componente_d": {
"base": self.componente_d.base,
"tempo": self.componente_d.tempo,
"extras": self.componente_d.extras,
"bonus": self.componente_d.bonus,
"total": self.componente_d.total,
},
"bloco_a": self.bloco_a.to_dict(),
"bloco_c": self.bloco_c.to_dict(),
"bloco_d": self.bloco_d.to_dict(),
"pontuacao_total": self.total,
}
@property
def detalhamento(self) -> Dict:
return self.to_dict()

589
backend/src/infrastructure/repositories/consultor_repository_impl.py
View File

@@ -6,9 +6,14 @@ import asyncio
from ...domain.entities.consultor import (
Consultor,
CoordenacaoCapes,
CoordenacaoPrograma,
Consultoria,
Inscricao,
AvaliacaoComissao,
Premiacao,
BolsaCNPQ,
Participacao,
Orientacao,
MembroBanca,
)
from ...domain.repositories.consultor_repository import ConsultorRepository
from ...domain.services.calculador_pontuacao import CalculadorPontuacao
@@ -42,36 +47,12 @@ _ranking_cache = RankingCache(ttl_seconds=300)
class ConsultorRepositoryImpl(ConsultorRepository):
def __init__(self, es_client: ElasticsearchClient, oracle_client: OracleClient):
def __init__(self, es_client: ElasticsearchClient, oracle_client: OracleClient = None):
self.es_client = es_client
self.oracle_client = oracle_client
self.calculador = CalculadorPontuacao()
self.es_disponivel = True
def _mesclar_periodos(self, periodos: List[Periodo]) -> List[Periodo]:
"""
Mescla períodos sobrepostos/contíguos para evitar contagem dupla de tempo.
"""
if not periodos:
return []
periodos = sorted(periodos, key=lambda p: p.inicio)
mesclados: List[Periodo] = []
for p in periodos:
if not mesclados:
mesclados.append(p)
continue
ultimo = mesclados[-1]
fim_ultimo = ultimo.fim or datetime.now()
fim_atual = p.fim or datetime.now()
if p.inicio <= fim_ultimo:
novo_fim = max(fim_ultimo, fim_atual)
mesclados[-1] = Periodo(inicio=ultimo.inicio, fim=novo_fim if not ultimo.ativo else None)
else:
mesclados.append(p)
return mesclados
def _parse_date(self, date_str: Optional[str]) -> Optional[datetime]:
if not date_str:
return None
@@ -80,10 +61,87 @@ class ConsultorRepositoryImpl(ConsultorRepository):
except:
return None
def _extrair_consultoria(self, atuacoes: List[Dict[str, Any]]) -> Optional[Consultoria]:
consultorias = [
a for a in atuacoes if a.get("tipo") in ["Consultor", "Histórico de Consultoria"]
def _mesclar_periodos(self, periodos: List[Periodo]) -> List[Periodo]:
if not periodos:
return []
periodos = sorted(periodos, key=lambda p: p.inicio if p.inicio else datetime.min)
mesclados: List[Periodo] = []
for p in periodos:
if not mesclados:
mesclados.append(p)
continue
ultimo = mesclados[-1]
fim_ultimo = ultimo.fim or datetime.now()
fim_atual = p.fim or datetime.now()
if p.inicio and p.inicio <= fim_ultimo:
novo_fim = max(fim_ultimo, fim_atual)
mesclados[-1] = Periodo(inicio=ultimo.inicio, fim=novo_fim if not ultimo.ativo else None)
else:
mesclados.append(p)
return mesclados
def _inferir_tipo_coordenacao(self, coord: Dict[str, Any]) -> str:
dados_coord = coord.get("dadosCoordenacaoArea", {}) or {}
tipo_coord = dados_coord.get("tipo", "").lower()
if "câmara" in tipo_coord or "camara" in tipo_coord:
return "CAM"
elif "adjunt" in tipo_coord:
if "profissional" in tipo_coord or "mestrado" in tipo_coord:
return "CAJ_MP"
return "CAJ"
elif "coordenador de área" in tipo_coord:
return "CA"
descricao = coord.get("descricao", "").lower()
nome = coord.get("nome", "").lower()
texto = f"{descricao} {nome}"
if "câmara" in texto or "camara" in texto:
return "CAM"
elif "mestrado profissional" in texto:
return "CAJ_MP"
elif "adjunt" in texto:
return "CAJ"
return "CA"
def _extrair_coordenacoes_capes(self, atuacoes: List[Dict[str, Any]]) -> List[CoordenacaoCapes]:
coordenacoes = [
a for a in atuacoes
if a.get("tipo") in ["Coordenação de Área de Avaliação", "Histórico de Coordenação de Área de Avaliação"]
]
resultado = []
for coord in coordenacoes:
dados_coord = coord.get("dadosCoordenacaoArea", {}) or {}
inicio = self._parse_date(dados_coord.get("inicioVinculacao")) or self._parse_date(coord.get("inicio"))
if not inicio:
continue
codigo = self._inferir_tipo_coordenacao(coord)
fim = self._parse_date(dados_coord.get("fimVinculacao")) or self._parse_date(coord.get("fim"))
if inicio and fim and fim < inicio:
fim = None
area_avaliacao_obj = dados_coord.get("areaAvaliacao", {}) or {}
area_avaliacao = area_avaliacao_obj.get("nome") if isinstance(area_avaliacao_obj, dict) else coord.get("areaAvaliacao", "N/A")
if not area_avaliacao:
area_avaliacao = coord.get("descricao", "N/A").split(" - ")[0] if coord.get("descricao") else "N/A"
resultado.append(CoordenacaoCapes(
codigo=codigo,
tipo=codigo,
area_avaliacao=area_avaliacao,
periodo=Periodo(inicio=inicio, fim=fim),
areas_adicionais=[],
ja_coordenou_antes=len(resultado) > 0,
))
return resultado
def _extrair_consultoria(self, atuacoes: List[Dict[str, Any]]) -> Optional[Consultoria]:
consultorias = [a for a in atuacoes if a.get("tipo") in ["Consultor", "Histórico de Consultoria"]]
if not consultorias:
return None
@@ -109,7 +167,7 @@ class ConsultorRepositoryImpl(ConsultorRepository):
)
if inicio and fim and fim < inicio:
fim = None # dados inconsistentes: trata como em aberto
fim = None
if inicio:
try:
periodos.append(Periodo(inicio=inicio, fim=fim))
@@ -125,283 +183,302 @@ class ConsultorRepositoryImpl(ConsultorRepository):
return None
mesclados = self._mesclar_periodos(periodos)
anos_total = sum(p.anos_completos(datetime.now()) for p in mesclados)
anos_consecutivos = max((p.anos_completos(datetime.now()) for p in mesclados), default=0)
retornos = max(0, len(mesclados) - 1)
ativo = any(p.ativo for p in periodos)
eventos_sae = [a for a in atuacoes if a.get("tipo") == "Evento"]
total_eventos = len(eventos_sae)
limite_recente = datetime.now() - timedelta(days=730)
eventos_recentes = 0
vezes_responsavel = 0
for ev in eventos_sae:
data_fim = self._parse_date(ev.get("fim")) or self._parse_date(ev.get("inicio"))
if data_fim and data_fim >= limite_recente:
eventos_recentes += 1
dados_evento = ev.get("dadosEvento", {}) or {}
if dados_evento.get("consultorResponsavel") == "Sim":
vezes_responsavel += 1
situacao_final = situacoes[0] if situacoes else "N/A"
is_ativo = ativo or "atividade" in situacao_final.lower() or "ativo" in situacao_final.lower()
is_falecido = "falecido" in situacao_final.lower()
if is_falecido:
codigo = "CONS_FALECIDO"
elif is_ativo:
codigo = "CONS_ATIVO"
else:
codigo = "CONS_HIST"
primeiro_evento = min(p.inicio for p in periodos)
ultimo_evento = max((p.fim or datetime.now()) for p in periodos) if not ativo else datetime.now()
areas = list(set(areas)) if areas else ["N/A"]
situacao_final = situacoes[0] if situacoes else "N/A"
return Consultoria(
total_eventos=total_eventos,
eventos_recentes=eventos_recentes,
primeiro_evento=primeiro_evento,
ultimo_evento=ultimo_evento,
areas=areas,
codigo=codigo,
situacao=situacao_final,
anos_completos=anos_total,
periodo=Periodo(inicio=primeiro_evento, fim=None if ativo else datetime.now()),
areas=areas,
anos_consecutivos=anos_consecutivos,
retornos=retornos,
vezes_responsavel=vezes_responsavel,
)
def _extrair_coordenacoes_capes(
self, atuacoes: List[Dict[str, Any]]
) -> List[CoordenacaoCapes]:
coordenacoes = [
a
for a in atuacoes
if a.get("tipo")
in [
"Coordenação de Área de Avaliação",
"Histórico de Coordenação de Área de Avaliação",
]
]
resultado = []
for coord in coordenacoes:
dados_coord = coord.get("dadosCoordenacaoArea", {}) or {}
inicio = (
self._parse_date(dados_coord.get("inicioVinculacao"))
or self._parse_date(coord.get("inicio"))
)
if not inicio:
def _extrair_inscricoes(self, atuacoes: List[Dict[str, Any]]) -> List[Inscricao]:
inscricoes = []
for a in atuacoes:
if a.get("tipo") != "Inscrição Prêmio":
continue
tipo = self._inferir_tipo_coordenacao(coord)
fim = (
self._parse_date(dados_coord.get("fimVinculacao"))
or self._parse_date(coord.get("fim"))
)
dados = a.get("dadosParticipacaoInscricaoPremio", {}) or {}
tipo_part = dados.get("tipo", "")
nome_premio = dados.get("nomePremio") or dados.get("premio") or a.get("descricao", "")
ano = dados.get("ano")
if not ano:
inicio = self._parse_date(a.get("inicio"))
ano = inicio.year if inicio else datetime.now().year
if inicio and fim and fim < inicio:
fim = None # ignora fins inconsistentes para não quebrar cálculo
is_institucional = "coordenador" in tipo_part.lower() or "ppg" in tipo_part.lower()
codigo = "INSC_INST" if is_institucional else "INSC_AUTOR"
area_avaliacao_obj = dados_coord.get("areaAvaliacao", {}) or {}
area_avaliacao = area_avaliacao_obj.get("nome") if isinstance(area_avaliacao_obj, dict) else coord.get("areaAvaliacao", "N/A")
if not area_avaliacao:
area_avaliacao = coord.get("descricao", "N/A").split(" - ")[0] if coord.get("descricao") else "N/A"
inscricoes.append(Inscricao(
codigo=codigo,
tipo=tipo_part,
premio=nome_premio,
ano=ano,
situacao=dados.get("situacao", ""),
))
resultado.append(
CoordenacaoCapes(
tipo=tipo,
area_avaliacao=area_avaliacao,
periodo=Periodo(inicio=inicio, fim=fim),
areas_adicionais=[],
ja_coordenou_antes=len(resultado) > 0,
)
)
return inscricoes
return resultado
def _extrair_avaliacoes_comissao(self, atuacoes: List[Dict[str, Any]]) -> List[AvaliacaoComissao]:
avaliacoes = []
for a in atuacoes:
if a.get("tipo") != "Avaliação Prêmio":
continue
def _inferir_tipo_coordenacao(self, coord: Dict[str, Any]) -> str:
dados_coord = coord.get("dadosCoordenacaoArea", {}) or {}
tipo_coord = dados_coord.get("tipo", "").lower()
dados = a.get("dadosParticipacaoPremio", {}) or {}
tipo_part = dados.get("tipo", "")
nome_premio = dados.get("nomePremio") or dados.get("premio") or a.get("descricao", "")
ano = dados.get("ano")
if not ano:
inicio = self._parse_date(a.get("inicio"))
ano = inicio.year if inicio else datetime.now().year
if "câmara" in tipo_coord or "camara" in tipo_coord:
return "CAM"
elif "adjunt" in tipo_coord:
if "profissional" in tipo_coord or "mestrado" in tipo_coord:
return "CAJ-MP"
return "CAJ"
elif "coordenador de área" in tipo_coord:
return "CA"
comissao = dados.get("comissao", {}) or {}
comissao_tipo = comissao.get("tipo", "") if isinstance(comissao, dict) else ""
descricao = coord.get("descricao", "").lower()
nome = coord.get("nome", "").lower()
texto = f"{descricao} {nome}"
is_grande_premio = "grande" in nome_premio.lower()
is_coordenador = "coordenador" in tipo_part.lower() or "presidente" in tipo_part.lower()
if "câmara" in texto or "camara" in texto:
return "CAM"
elif "mestrado profissional" in texto:
return "CAJ-MP"
elif "adjunt" in texto:
return "CAJ"
else:
return "CA"
def _classificar_nivel_premio(self, nome: str) -> str:
nome = (nome or "").lower()
if "grande prêmio capes de tese" in nome or "grande premio capes de tese" in nome:
return "nivel1_grande"
if "prêmio capes de tese" in nome or "premio capes de tese" in nome:
return "nivel1_pct"
if "interfarma" in nome or "vale-capes" in nome or "vale capes" in nome:
return "nivel2"
if nome:
return "nivel3"
return "desconhecido"
def _pontuar_premiacao_recebida(self, nivel: str, tipo_premiacao: str) -> int:
tipo = (tipo_premiacao or "").lower()
if nivel == "nivel1_grande":
base = 150
extra = 50 if "grande" in tipo else 0
return min(base + extra, 180)
if nivel == "nivel1_pct":
base = 100
if "mencao" in tipo:
extra = 15
elif "premio" in tipo:
extra = 25
if is_coordenador:
codigo = "COORD_COMIS_GP" if is_grande_premio else "COORD_COMIS_PREMIO"
else:
extra = 0
return min(base + extra, 150)
if nivel == "nivel2":
base = 30
if "premio" in tipo:
extra = 20
elif "mencao" in tipo:
extra = 10
else:
extra = 0
return min(base + extra, 60)
# nivel3 e fallback
base = 10
if "premio" in tipo:
extra = 5
elif "mencao" in tipo:
extra = 3
else:
extra = 0
return min(base + extra, 20)
codigo = "AVAL_COMIS_GP" if is_grande_premio else "AVAL_COMIS_PREMIO"
def _pontuar_participacao_premio(self, nivel: str, tipo_participacao: str) -> int:
tipo = (tipo_participacao or "").lower()
if "avaliador" in tipo or "banca" in tipo:
return 2 # teto final tratado em componente D
if "coordenador" in tipo or "comissao" in tipo or "comissão" in tipo:
if nivel == "nivel1_grande":
return 115 # valor máximo já com peso
if nivel == "nivel1_pct":
return 115 # aproximação segura para teto
if nivel == "nivel2":
return 80
return 40
if "inscricao" in tipo or "inscrição" in tipo:
if nivel in ["nivel1_grande", "nivel1_pct"]:
return 2
if nivel == "nivel2":
return 1
return 1
return 0
avaliacoes.append(AvaliacaoComissao(
codigo=codigo,
tipo=tipo_part,
premio=nome_premio,
ano=ano,
comissao_tipo=comissao_tipo,
))
return avaliacoes
def _extrair_premiacoes(self, atuacoes: List[Dict[str, Any]]) -> List[Premiacao]:
premiacoes = []
for a in atuacoes:
tipo_atuacao = a.get("tipo", "")
dados_premiacao = a.get("dadosPremiacaoPremio") or a.get("dadosPremio") or {}
dados_participacao = (
a.get("dadosParticipacaoPremio")
or a.get("dadosParticipacaoInscricaoPremio")
or {}
)
# Premiações recebidas
if dados_premiacao:
nome_premio = dados_premiacao.get("nomePremio") or a.get("descricao", "N/A")
tipo_premiacao = dados_premiacao.get("tipoPremiacao") or dados_premiacao.get("tipo", "")
ano = dados_premiacao.get("ano") or a.get("ano")
if not ano:
inicio = self._parse_date(a.get("inicio"))
ano = inicio.year if inicio else datetime.now().year
nivel = self._classificar_nivel_premio(nome_premio)
pontos = self._pontuar_premiacao_recebida(nivel, tipo_premiacao)
premiacoes.append(
Premiacao(
tipo=tipo_premiacao or tipo_atuacao or "Premiação",
nome_premio=nome_premio,
ano=ano or datetime.now().year,
pontos=int(pontos),
)
)
if a.get("tipo") != "Premiação Prêmio":
continue
# Participações (inscrição/avaliação/coordenação)
if dados_participacao:
tipo_part = (
dados_participacao.get("tipoParticipacao")
or dados_participacao.get("tipo")
or tipo_atuacao
)
nome_premio = dados_participacao.get("nomePremio") or a.get("descricao", "N/A")
ano = dados_participacao.get("ano") or a.get("ano")
if not ano:
inicio = self._parse_date(a.get("inicio"))
ano = inicio.year if inicio else datetime.now().year
nivel = self._classificar_nivel_premio(nome_premio)
pontos = self._pontuar_participacao_premio(nivel, tipo_part)
premiacoes.append(
Premiacao(
tipo=tipo_part or "Participação Prêmio",
nome_premio=nome_premio,
ano=ano or datetime.now().year,
pontos=int(pontos),
)
)
continue
# Fallback para tipos antigos
if tipo_atuacao in [
"Premiação Prêmio",
"Premiação",
"Avaliação Prêmio",
"Inscrição Prêmio",
]:
pontos = self._pontuar_participacao_premio("nivel3", tipo_atuacao)
dados = a.get("dadosPremiacaoPremio", {}) or a.get("dadosPremio", {}) or {}
tipo_premiacao = dados.get("tipoPremiacao") or dados.get("premiacao") or ""
nome_premio = dados.get("nomePremio") or dados.get("evento") or a.get("descricao", "")
ano = dados.get("ano")
if not ano:
inicio = self._parse_date(a.get("inicio"))
ano = inicio.year if inicio else datetime.now().year
premiacoes.append(
Premiacao(
tipo=tipo_atuacao,
nome_premio=a.get("descricao", "N/A"),
ano=ano,
pontos=int(pontos),
)
)
tipo_lower = tipo_premiacao.lower()
nome_lower = nome_premio.lower()
if "grande" in nome_lower or "grande" in tipo_lower:
codigo = "PREMIACAO"
elif "menção" in tipo_lower or "mencao" in tipo_lower or "honrosa" in tipo_lower:
codigo = "MENCAO"
else:
codigo = "PREMIACAO_GP"
premiacoes.append(Premiacao(
codigo=codigo,
tipo=tipo_premiacao,
nome_premio=nome_premio,
ano=ano,
))
return premiacoes
def _extrair_bolsas_cnpq(self, atuacoes: List[Dict[str, Any]]) -> List[BolsaCNPQ]:
bolsas = []
for a in atuacoes:
if a.get("tipo") != "Bolsa CNPQ" and "bolsa" not in a.get("tipo", "").lower():
continue
dados = a.get("dadosBolsa", {}) or {}
nivel = dados.get("nivel", "") or dados.get("categoria", "") or ""
area = dados.get("areaConhecimento", "") or ""
nivel_lower = nivel.lower()
if "1a" in nivel_lower or "1b" in nivel_lower or "1c" in nivel_lower or "1d" in nivel_lower:
codigo = "BOL_BPQ_SUPERIOR"
else:
codigo = "BOL_BPQ_INTERMEDIARIO"
bolsas.append(BolsaCNPQ(
codigo=codigo,
nivel=nivel,
area=area,
))
return bolsas
def _extrair_participacoes(self, atuacoes: List[Dict[str, Any]]) -> List[Participacao]:
participacoes = []
for a in atuacoes:
tipo = a.get("tipo", "")
if tipo == "Evento":
participacoes.append(Participacao(
codigo="EVENTO",
tipo="Evento",
descricao=a.get("descricao", ""),
ano=self._parse_date(a.get("inicio")).year if self._parse_date(a.get("inicio")) else None,
))
elif tipo == "Projeto" or "projeto" in tipo.lower():
participacoes.append(Participacao(
codigo="PROJ",
tipo="Projeto",
descricao=a.get("descricao", ""),
ano=self._parse_date(a.get("inicio")).year if self._parse_date(a.get("inicio")) else None,
))
return participacoes
def _extrair_orientacoes(self, atuacoes: List[Dict[str, Any]]) -> List[Orientacao]:
orientacoes = []
for a in atuacoes:
tipo = a.get("tipo", "").lower()
if "orientação" not in tipo and "orientacao" not in tipo:
continue
if "co-orientação" in tipo or "coorientação" in tipo or "co_orient" in tipo:
continue
dados = a.get("dadosOrientacao", {}) or {}
nivel = dados.get("nivel", "") or dados.get("tipo", "") or ""
ano = dados.get("ano")
if not ano:
inicio = self._parse_date(a.get("inicio"))
ano = inicio.year if inicio else None
nivel_lower = nivel.lower()
if "pós-doc" in nivel_lower or "pos-doc" in nivel_lower or "posdoc" in nivel_lower:
codigo = "ORIENT_POS_DOC"
elif "tese" in nivel_lower or "doutorado" in nivel_lower:
codigo = "ORIENT_TESE"
else:
codigo = "ORIENT_DISS"
orientacoes.append(Orientacao(
codigo=codigo,
tipo=tipo,
nivel=nivel,
ano=ano,
))
return orientacoes
def _extrair_coorientacoes(self, atuacoes: List[Dict[str, Any]]) -> List[Orientacao]:
coorientacoes = []
for a in atuacoes:
tipo = a.get("tipo", "").lower()
if "co-orientação" not in tipo and "coorientação" not in tipo and "co_orient" not in tipo:
continue
dados = a.get("dadosOrientacao", {}) or a.get("dadosCoorientacao", {}) or {}
nivel = dados.get("nivel", "") or dados.get("tipo", "") or ""
ano = dados.get("ano")
if not ano:
inicio = self._parse_date(a.get("inicio"))
ano = inicio.year if inicio else None
nivel_lower = nivel.lower()
if "pós-doc" in nivel_lower or "pos-doc" in nivel_lower or "posdoc" in nivel_lower:
codigo = "CO_ORIENT_POS_DOC"
elif "tese" in nivel_lower or "doutorado" in nivel_lower:
codigo = "CO_ORIENT_TESE"
else:
codigo = "CO_ORIENT_DISS"
coorientacoes.append(Orientacao(
codigo=codigo,
tipo=tipo,
nivel=nivel,
ano=ano,
))
return coorientacoes
def _extrair_membros_banca(self, atuacoes: List[Dict[str, Any]]) -> List[MembroBanca]:
membros = []
for a in atuacoes:
tipo = a.get("tipo", "").lower()
if "banca" not in tipo:
continue
dados = a.get("dadosBanca", {}) or {}
nivel = dados.get("nivel", "") or dados.get("tipo", "") or ""
ano = dados.get("ano")
if not ano:
inicio = self._parse_date(a.get("inicio"))
ano = inicio.year if inicio else None
nivel_lower = nivel.lower()
if "pós-doc" in nivel_lower or "pos-doc" in nivel_lower or "posdoc" in nivel_lower:
codigo = "MB_BANCA_POS_DOC"
elif "tese" in nivel_lower or "doutorado" in nivel_lower:
codigo = "MB_BANCA_TESE"
else:
codigo = "MB_BANCA_DISS"
membros.append(MembroBanca(
codigo=codigo,
tipo=tipo,
nivel=nivel,
ano=ano,
))
return membros
async def _construir_consultor(self, doc: Dict[str, Any]) -> Consultor:
id_pessoa = doc["id"]
dados_pessoais = doc.get("dadosPessoais", {})
atuacoes = doc.get("atuacoes", [])
consultoria = self._extrair_consultoria(atuacoes)
coordenacoes_capes = self._extrair_coordenacoes_capes(atuacoes)
consultoria = self._extrair_consultoria(atuacoes)
inscricoes = self._extrair_inscricoes(atuacoes)
avaliacoes = self._extrair_avaliacoes_comissao(atuacoes)
premiacoes = self._extrair_premiacoes(atuacoes)
bolsas = self._extrair_bolsas_cnpq(atuacoes)
participacoes = self._extrair_participacoes(atuacoes)
orientacoes = self._extrair_orientacoes(atuacoes)
coorientacoes = self._extrair_coorientacoes(atuacoes)
membros_banca = self._extrair_membros_banca(atuacoes)
consultor = Consultor(
id_pessoa=id_pessoa,
nome=dados_pessoais.get("nome", "N/A"),
cpf=dados_pessoais.get("cpf"),
coordenacoes_capes=coordenacoes_capes,
coordenacoes_programas=[], # PPG vem do job/ETL de Componente B
consultoria=consultoria,
inscricoes=inscricoes,
avaliacoes_comissao=avaliacoes,
premiacoes=premiacoes,
bolsas_cnpq=bolsas,
participacoes=participacoes,
orientacoes=orientacoes + coorientacoes,
membros_banca=membros_banca,
)
consultor.pontuacao = self.calculador.calcular_pontuacao_completa(consultor)
@@ -457,8 +534,6 @@ class ConsultorRepositoryImpl(ConsultorRepository):
consultores = []
for doc in docs:
consultor = await self._construir_consultor(doc)
score_es = doc.get("_score_es", 0)
consultor.score_es = score_es
consultores.append(consultor)
consultores_ordenados = sorted(

70
backend/src/interface/api/routes.py
View File

@@ -31,7 +31,7 @@ async def obter_ranking(
limite: int = Query(default=100, ge=1, le=1000, description="Limite de consultores"),
offset: int = Query(default=0, ge=0, description="Offset para paginação"),
componente: Optional[str] = Query(
default=None, description="Filtrar por componente (a, b, c, d)"
default=None, description="Filtrar por bloco (a, c, d)"
),
repository: ConsultorRepositoryImpl = Depends(get_repository),
):
@@ -53,7 +53,7 @@ async def obter_ranking(
async def obter_ranking_detalhado(
limite: int = Query(default=100, ge=1, le=1000, description="Limite de consultores"),
componente: Optional[str] = Query(
default=None, description="Filtrar por componente (a, b, c, d)"
default=None, description="Filtrar por bloco (a, c, d)"
),
repository: ConsultorRepositoryImpl = Depends(get_repository),
):
@@ -95,9 +95,6 @@ async def ranking_paginado(
ativo: Optional[bool] = Query(default=None, description="Filtrar por status ativo"),
ranking_repo = Depends(get_ranking_repository),
):
"""
Retorna ranking paginado do Oracle (pré-calculado).
"""
total = ranking_repo.contar_total(filtro_ativo=ativo)
consultores = ranking_repo.buscar_paginado(page=page, size=size, filtro_ativo=ativo)
@@ -132,50 +129,51 @@ async def buscar_por_nome(
]
def _calcular_continuidade(anos_consecutivos: int) -> int:
if anos_consecutivos >= 8:
return 15
elif anos_consecutivos >= 5:
return 10
elif anos_consecutivos >= 3:
return 5
return 0
def _consultor_resumo_from_ranking(c):
consultoria = None
coordenacoes_capes = None
coordenacoes_programas = None
inscricoes = None
avaliacoes_comissao = None
premiacoes = None
bolsas_cnpq = None
participacoes = None
orientacoes = None
membros_banca = None
try:
jd = json.loads(c.json_detalhes) if c.json_detalhes else {}
if isinstance(jd, dict):
consultoria = jd.get("consultoria")
coordenacoes_capes = jd.get("coordenacoes_capes")
coordenacoes_programas = jd.get("coordenacoes_programas")
inscricoes = jd.get("inscricoes")
avaliacoes_comissao = jd.get("avaliacoes_comissao")
premiacoes = jd.get("premiacoes")
if consultoria and isinstance(consultoria, dict):
anos_consec = consultoria.get("anos_consecutivos") or consultoria.get("anos_completos") or 0
consultoria["continuidade"] = _calcular_continuidade(anos_consec)
consultoria["anos_consecutivos"] = anos_consec
bolsas_cnpq = jd.get("bolsas_cnpq")
participacoes = jd.get("participacoes")
orientacoes = jd.get("orientacoes")
membros_banca = jd.get("membros_banca")
except Exception:
consultoria = None
pass
return ConsultorRankingResumoSchema(
id_pessoa=c.id_pessoa,
nome=c.nome,
posicao=c.posicao,
pontuacao_total=c.pontuacao_total,
componente_a=c.componente_a,
componente_b=c.componente_b,
componente_c=c.componente_c,
componente_d=c.componente_d,
bloco_a=c.componente_a,
bloco_c=c.componente_c,
bloco_d=c.componente_d,
ativo=c.ativo,
anos_atuacao=c.anos_atuacao,
consultoria=consultoria,
coordenacoes_capes=coordenacoes_capes,
coordenacoes_programas=coordenacoes_programas,
inscricoes=inscricoes,
avaliacoes_comissao=avaliacoes_comissao,
premiacoes=premiacoes,
bolsas_cnpq=bolsas_cnpq,
participacoes=participacoes,
orientacoes=orientacoes,
membros_banca=membros_banca,
)
@@ -183,23 +181,24 @@ def _consultor_resumo_from_ranking(c):
async def ranking_estatisticas(
ranking_repo = Depends(get_ranking_repository),
):
"""
Retorna estatísticas do ranking.
"""
estatisticas = ranking_repo.obter_estatisticas()
distribuicao = ranking_repo.obter_distribuicao()
return EstatisticasRankingSchema(
**estatisticas,
total_consultores=estatisticas.get("total_consultores", 0),
total_ativos=estatisticas.get("total_ativos", 0),
total_inativos=estatisticas.get("total_inativos", 0),
ultima_atualizacao=estatisticas.get("ultima_atualizacao"),
pontuacao_media=estatisticas.get("pontuacao_media", 0),
pontuacao_maxima=estatisticas.get("pontuacao_maxima", 0),
pontuacao_minima=estatisticas.get("pontuacao_minima", 0),
media_blocos=estatisticas.get("media_componentes", {}),
distribuicao=distribuicao
)
@router.get("/ranking/status", response_model=JobStatusSchema)
async def status_processamento():
"""
Retorna o status do job de processamento do ranking.
"""
return JobStatusSchema(**job_status.to_dict())
@@ -209,9 +208,6 @@ async def processar_ranking(
request: ProcessarRankingRequestSchema = ProcessarRankingRequestSchema(),
job = Depends(get_processar_job),
):
"""
Dispara o processamento do ranking em background.
"""
if job_status.is_running:
raise HTTPException(status_code=409, detail="Job já está em execução")

98
backend/src/interface/schemas/consultor_schema.py
View File

@@ -1,4 +1,4 @@
from pydantic import BaseModel, Field
from pydantic import BaseModel
from typing import List, Optional
@@ -10,6 +10,7 @@ class PeriodoSchema(BaseModel):
class CoordenacaoCapesSchema(BaseModel):
codigo: str
tipo: str
area_avaliacao: str
periodo: PeriodoSchema
@@ -17,51 +18,84 @@ class CoordenacaoCapesSchema(BaseModel):
ja_coordenou_antes: bool
class CoordenacaoProgramaSchema(BaseModel):
id_programa: int
nome_programa: str
codigo_programa: str
nota_ppg: str
modalidade: str
area_avaliacao: str
periodo: PeriodoSchema
class ConsultoriaSchema(BaseModel):
total_eventos: int
eventos_recentes: int
primeiro_evento: str
ultimo_evento: str
continuidade: int
areas: List[str]
codigo: str
situacao: str
anos_completos: int
periodo: PeriodoSchema
areas: List[str]
anos_consecutivos: int
retornos: int
vezes_responsavel: int
class InscricaoSchema(BaseModel):
codigo: str
tipo: str
premio: str
ano: int
situacao: str
class AvaliacaoComissaoSchema(BaseModel):
codigo: str
tipo: str
premio: str
ano: int
comissao_tipo: str
class PremiacaoSchema(BaseModel):
codigo: str
tipo: str
nome_premio: str
ano: int
pontos: int
class ComponentePontuacaoSchema(BaseModel):
class BolsaCNPQSchema(BaseModel):
codigo: str
nivel: str
area: str
class ParticipacaoSchema(BaseModel):
codigo: str
tipo: str
descricao: str
ano: Optional[int] = None
class OrientacaoSchema(BaseModel):
codigo: str
tipo: str
nivel: str
ano: Optional[int] = None
class MembroBancaSchema(BaseModel):
codigo: str
tipo: str
nivel: str
ano: Optional[int] = None
class PontuacaoAtuacaoSchema(BaseModel):
codigo: str
base: int
tempo: int
extras: int
bonus: int
retorno: int
total: int
quantidade: int
class PontuacaoBlocoSchema(BaseModel):
bloco: str
total: int
atuacoes: List[PontuacaoAtuacaoSchema]
class PontuacaoCompletaSchema(BaseModel):
componente_a: ComponentePontuacaoSchema
componente_b: ComponentePontuacaoSchema
componente_c: ComponentePontuacaoSchema
componente_d: ComponentePontuacaoSchema
bloco_a: PontuacaoBlocoSchema
bloco_c: PontuacaoBlocoSchema
bloco_d: PontuacaoBlocoSchema
pontuacao_total: int
@@ -72,6 +106,9 @@ class ConsultorResumoSchema(BaseModel):
ativo: bool
veterano: bool
pontuacao_total: int
bloco_a: int
bloco_c: int
bloco_d: int
rank: Optional[int] = None
@@ -83,9 +120,14 @@ class ConsultorDetalhadoSchema(BaseModel):
ativo: bool
veterano: bool
coordenacoes_capes: List[CoordenacaoCapesSchema]
coordenacoes_programas: List[CoordenacaoProgramaSchema]
consultoria: Optional[ConsultoriaSchema] = None
inscricoes: List[InscricaoSchema]
avaliacoes_comissao: List[AvaliacaoComissaoSchema]
premiacoes: List[PremiacaoSchema]
bolsas_cnpq: List[BolsaCNPQSchema]
participacoes: List[ParticipacaoSchema]
orientacoes: List[OrientacaoSchema]
membros_banca: List[MembroBancaSchema]
pontuacao: PontuacaoCompletaSchema
rank: Optional[int] = None

16
backend/src/interface/schemas/ranking_schema.py
View File

@@ -8,16 +8,20 @@ class ConsultorRankingResumoSchema(BaseModel):
nome: str
posicao: Optional[int]
pontuacao_total: float
componente_a: float
componente_b: float
componente_c: float
componente_d: float
bloco_a: float
bloco_c: float
bloco_d: float
ativo: bool
anos_atuacao: float
consultoria: Optional[dict] = None
coordenacoes_capes: Optional[list] = None
coordenacoes_programas: Optional[list] = None
inscricoes: Optional[list] = None
avaliacoes_comissao: Optional[list] = None
premiacoes: Optional[list] = None
bolsas_cnpq: Optional[list] = None
participacoes: Optional[list] = None
orientacoes: Optional[list] = None
membros_banca: Optional[list] = None
class RankingPaginadoResponseSchema(BaseModel):
@@ -36,7 +40,7 @@ class EstatisticasRankingSchema(BaseModel):
pontuacao_media: float
pontuacao_maxima: float
pontuacao_minima: float
media_componentes: dict
media_blocos: dict
distribuicao: List[dict]