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

feat: Aprimora cálculo de pontuação e extração de dados

Browse Source 
- Adiciona campos situacao, anos_completos, anos_consecutivos e retornos
  na entidade Consultoria para suportar regras documentadas
- Implementa mesclagem de períodos sobrepostos para evitar contagem dupla
- Melhora componente A com cálculo por área e detecção de retornos
- Ajusta componente B com bônus por nota PPG
- Refatora componente C com bônus de continuidade e retorno
- Implementa componente D com classificação de nível de prêmio
  (Grande Prêmio, PCT, Interfarma, outros) e pontuação diferenciada
- Trata datas inconsistentes (fim < início) como períodos em aberto
- Extrai situacaoConsultoria do campo dadosConsultoria.situacaoConsultoria
This commit is contained in:
Frederico Castro
2025-12-09 22:57:46 -03:00
parent 9a8332b740
commit ff4d838f34
7 changed files with 348 additions and 83 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -48,6 +48,30 @@ class ConsultorRepositoryImpl(ConsultorRepository):
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
@@ -63,22 +87,51 @@ class ConsultorRepositoryImpl(ConsultorRepository):
if not consultorias:
return None
datas_inicio_consultoria = [
self._parse_date(c.get("inicio"))
for c in consultorias
]
datas_inicio_consultoria = [d for d in datas_inicio_consultoria if d]
periodos: List[Periodo] = []
situacoes: List[str] = []
areas: List[str] = []
if not datas_inicio_consultoria:
for c in consultorias:
dc = c.get("dadosConsultoria", {}) or {}
situacao = dc.get("situacaoConsultoria") or c.get("situacaoConsultoria")
if situacao:
situacoes.append(situacao)
inicio = (
self._parse_date(dc.get("inicioVinculacao"))
or self._parse_date(dc.get("inicioSituacao"))
or self._parse_date(c.get("inicio"))
)
fim = (
self._parse_date(dc.get("fimVinculacao"))
or self._parse_date(dc.get("inativacaoSituacao"))
or self._parse_date(c.get("fim"))
)
if inicio and fim and fim < inicio:
fim = None # dados inconsistentes: trata como em aberto
if inicio:
try:
periodos.append(Periodo(inicio=inicio, fim=fim))
except ValueError:
continue
area = dc.get("areaAvaliacao") or c.get("areaAvaliacao")
if area:
areas.append(area)
periodos = [p for p in periodos if p.inicio]
if not periodos:
return None
eventos_sae = [
a for a in atuacoes if a.get("tipo") == "Evento"
]
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)
# considerar últimos 24 meses como janela de atividade
limite_recente = datetime.now() - timedelta(days=730)
eventos_recentes = 0
for ev in eventos_sae:
@@ -86,30 +139,25 @@ class ConsultorRepositoryImpl(ConsultorRepository):
if data_fim and data_fim >= limite_recente:
eventos_recentes += 1
dados_consultoria = consultorias[0].get("dadosConsultoria", {}) or {}
areas = []
for c in consultorias:
dc = c.get("dadosConsultoria", {}) or {}
area = dc.get("areaAvaliacao") or c.get("areaAvaliacao")
if area:
areas.append(area)
areas = list(set(areas)) if areas else ["N/A"]
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"]
vezes_responsavel = sum(1 for c in consultorias if c.get("responsavel", False))
datas_fim_consultoria = [
self._parse_date(c.get("fim"))
for c in consultorias
]
datas_fim_consultoria = [d for d in datas_fim_consultoria if d]
situacao_final = situacoes[0] if situacoes else "N/A"
return Consultoria(
total_eventos=total_eventos,
eventos_recentes=eventos_recentes,
primeiro_evento=min(datas_inicio_consultoria),
ultimo_evento=max(datas_fim_consultoria) if datas_fim_consultoria else datetime.now(),
primeiro_evento=primeiro_evento,
ultimo_evento=ultimo_evento,
vezes_responsavel=vezes_responsavel,
areas=areas,
situacao=situacao_final,
anos_completos=anos_total,
anos_consecutivos=anos_consecutivos,
retornos=retornos,
)
def _extrair_coordenacoes_capes(
@@ -127,14 +175,24 @@ class ConsultorRepositoryImpl(ConsultorRepository):
resultado = []
for coord in coordenacoes:
inicio = self._parse_date(coord.get("inicio"))
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
tipo = self._inferir_tipo_coordenacao(coord)
fim = self._parse_date(coord.get("fim"))
fim = (
self._parse_date(dados_coord.get("fimVinculacao"))
or self._parse_date(coord.get("fim"))
)
if inicio and fim and fim < inicio:
fim = None # ignora fins inconsistentes para não quebrar cálculo
dados_coord = coord.get("dadosCoordenacaoArea", {}) or {}
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:
@@ -178,45 +236,151 @@ class ConsultorRepositoryImpl(ConsultorRepository):
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
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)
def _pontuar_participacao_premio(self, nivel: str, tipo_participacao: str) -> int:
tipo = (tipo_participacao or "").lower()
if "avaliador" in tipo:
return 2 # teto final tratado em componente D
if "coordenador" 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
def _extrair_premiacoes(self, atuacoes: List[Dict[str, Any]]) -> List[Premiacao]:
premiacoes_data = [
a
for a in atuacoes
if a.get("tipo")
in [
"Premiação Prêmio",
"Avaliação Prêmio",
"Inscrição Prêmio",
"Premiação",
]
]
premiacoes = []
for prem in premiacoes_data:
pontos = self._calcular_pontos_premiacao(prem.get("tipo", ""))
inicio = self._parse_date(prem.get("inicio"))
ano = inicio.year if inicio else datetime.now().year
premiacoes.append(
Premiacao(
tipo=prem.get("tipo", "N/A"),
nome_premio=prem.get("descricao", "N/A"),
ano=ano,
pontos=pontos,
)
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 {}
)
return premiacoes
# 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
def _calcular_pontos_premiacao(self, tipo: str) -> int:
# Aproximação das regras (D) seguindo .claude/rules/ranking-consultores-capes.md
mapa = {
"Premiação Prêmio": 150,
"Premiação": 150,
"Avaliação Prêmio": 40,
"Inscrição Prêmio": 10,
}
return mapa.get(tipo, 0)
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),
)
)
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)
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),
)
)
return premiacoes
async def _construir_consultor(self, doc: Dict[str, Any]) -> Consultor:
id_pessoa = doc["id"]