Add utility scripts and documentation
- Add TIPOS_ATUACAO_ELASTICSEARCH.md: mapping of ES activity types - Add TOP_10_RANKING_CAPES.md: sample ranking output documentation - Add backend/scripts/: utility scripts for analysis and debugging - analise_detalhada.py: detailed consultant analysis - auditar_ranking.py: ranking audit tool - bootstrap_ranking.sh: bootstrap script - buscar_consultores_especificos.py: search specific consultants - popular_componente_b.py: populate component B - top10_ranking.py: generate top 10 ranking - Add scripts/reload_atuacapes.sh: reload ES index script
This commit is contained in:
Frederico Castro
265
backend/scripts/top10_ranking.py
Normal file
265
backend/scripts/top10_ranking.py
Normal file
@@ -0,0 +1,265 @@
|
||||
#!/usr/bin/env python3
|
||||
import sys
|
||||
import os
|
||||
import asyncio
|
||||
from datetime import datetime
|
||||
from typing import List, Dict, Any
|
||||
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'src'))
|
||||
|
||||
from infrastructure.elasticsearch.client import ElasticsearchClient
|
||||
from domain.services.calculador_pontuacao import CalculadorPontuacao
|
||||
from domain.entities.consultor import Consultor, CoordenacaoCapes, Consultoria, Premiacao, Periodo
|
||||
|
||||
ES_URL = "http://elastic-atuacapes.hom.capes.gov.br:9200"
|
||||
ES_INDEX = "atuacapes"
|
||||
ES_USER = "admin-atuacapes"
|
||||
ES_PASSWORD = "O}!S0bj%FhJ:"
|
||||
|
||||
|
||||
def parse_date(date_str):
|
||||
if not date_str:
|
||||
return None
|
||||
try:
|
||||
if '/' in date_str:
|
||||
return datetime.strptime(date_str[:10], '%d/%m/%Y')
|
||||
else:
|
||||
return datetime.strptime(date_str[:10], '%Y-%m-%d')
|
||||
except:
|
||||
return None
|
||||
|
||||
|
||||
def extrair_coordenacoes_capes(atuacoes: List[Dict]) -> List[CoordenacaoCapes]:
|
||||
coordenacoes_data = [
|
||||
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",
|
||||
]
|
||||
]
|
||||
|
||||
result = []
|
||||
for coord in coordenacoes_data:
|
||||
dados = coord.get("dadosCoordenacaoArea") or coord.get("dadosHistoricoCoordenacaoArea") or {}
|
||||
|
||||
nome = coord.get("nome", "")
|
||||
if "câmara" in nome.lower() or "camara" in nome.lower():
|
||||
tipo = "CAM"
|
||||
elif "mestrado profissional" in nome.lower():
|
||||
tipo = "CAJ-MP"
|
||||
elif "adjunt" in nome.lower():
|
||||
tipo = "CAJ"
|
||||
else:
|
||||
tipo = "CA"
|
||||
|
||||
inicio = parse_date(coord.get("inicio") or dados.get("inicio"))
|
||||
fim = parse_date(coord.get("fim") or dados.get("fim"))
|
||||
|
||||
if not inicio:
|
||||
continue
|
||||
|
||||
area = dados.get("areaAvaliacao", {})
|
||||
if isinstance(area, dict):
|
||||
area_nome = area.get("nome", "N/A")
|
||||
else:
|
||||
area_nome = str(area) if area else "N/A"
|
||||
|
||||
periodo = Periodo(inicio=inicio, fim=fim)
|
||||
|
||||
result.append(CoordenacaoCapes(
|
||||
tipo=tipo,
|
||||
area_avaliacao=area_nome,
|
||||
periodo=periodo,
|
||||
areas_adicionais=[],
|
||||
ja_coordenou_antes=False
|
||||
))
|
||||
|
||||
return result
|
||||
|
||||
|
||||
def extrair_consultoria(atuacoes: List[Dict]) -> Consultoria:
|
||||
consultorias = [
|
||||
a for a in atuacoes
|
||||
if a.get("tipo") in ["Consultor", "Histórico de Consultoria"]
|
||||
]
|
||||
|
||||
if not consultorias:
|
||||
return None
|
||||
|
||||
datas = []
|
||||
areas = set()
|
||||
total_eventos = len(consultorias)
|
||||
vezes_responsavel = 0
|
||||
|
||||
for cons in consultorias:
|
||||
dados = cons.get("dadosConsultoria", {})
|
||||
|
||||
inicio = parse_date(cons.get("inicio"))
|
||||
fim = parse_date(cons.get("fim"))
|
||||
|
||||
if inicio:
|
||||
datas.append(inicio)
|
||||
if fim:
|
||||
datas.append(fim)
|
||||
|
||||
area = dados.get("areaAvaliacao")
|
||||
if area:
|
||||
areas.add(area)
|
||||
|
||||
if dados.get("responsavel"):
|
||||
vezes_responsavel += 1
|
||||
|
||||
if not datas:
|
||||
return None
|
||||
|
||||
primeiro_evento = min(datas)
|
||||
ultimo_evento = max(datas)
|
||||
|
||||
limite_recente = datetime.now().replace(year=datetime.now().year - 2)
|
||||
eventos_recentes = sum(1 for d in datas if d >= limite_recente)
|
||||
|
||||
anos = (datetime.now() - primeiro_evento).days / 365.25
|
||||
|
||||
return Consultoria(
|
||||
total_eventos=total_eventos,
|
||||
eventos_recentes=eventos_recentes,
|
||||
primeiro_evento=primeiro_evento,
|
||||
ultimo_evento=ultimo_evento,
|
||||
areas=list(areas),
|
||||
situacao="Ativo" if eventos_recentes > 0 else "Histórico",
|
||||
anos_completos=int(anos),
|
||||
anos_consecutivos=int(anos),
|
||||
retornos=0
|
||||
)
|
||||
|
||||
|
||||
def extrair_premiacoes(atuacoes: List[Dict]) -> 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",
|
||||
]
|
||||
]
|
||||
|
||||
result = []
|
||||
mapa_pontos = {
|
||||
"Premiação Prêmio": 60,
|
||||
"Avaliação Prêmio": 40,
|
||||
"Inscrição Prêmio": 20,
|
||||
}
|
||||
|
||||
for prem in premiacoes_data:
|
||||
tipo = prem.get("tipo", "")
|
||||
dados = prem.get("dadosPremiacaoPremio") or prem.get("dadosParticipacaoPremio") or prem.get("dadosParticipacaoInscricaoPremio") or {}
|
||||
|
||||
nome_premio = dados.get("premio", "") or prem.get("descricao", "")
|
||||
ano_str = dados.get("ano") or prem.get("inicio", "")
|
||||
|
||||
try:
|
||||
ano = int(ano_str) if isinstance(ano_str, (int, str)) and str(ano_str).isdigit() else 0
|
||||
except:
|
||||
ano = 0
|
||||
|
||||
pontos = mapa_pontos.get(tipo, 0)
|
||||
|
||||
result.append(Premiacao(
|
||||
tipo=tipo,
|
||||
nome_premio=nome_premio,
|
||||
ano=ano,
|
||||
pontos=pontos
|
||||
))
|
||||
|
||||
return result
|
||||
|
||||
|
||||
async def main():
|
||||
print("Conectando ao Elasticsearch...")
|
||||
es_client = ElasticsearchClient(ES_URL, ES_INDEX, ES_USER, ES_PASSWORD)
|
||||
await es_client.connect()
|
||||
|
||||
try:
|
||||
print("Buscando candidatos com boost...")
|
||||
docs = await es_client.buscar_candidatos_ranking(size=100)
|
||||
print(f"Encontrados {len(docs)} candidatos")
|
||||
|
||||
calculador = CalculadorPontuacao()
|
||||
consultores = []
|
||||
|
||||
print("\nProcessando consultores...")
|
||||
for idx, doc in enumerate(docs, 1):
|
||||
id_pessoa = doc.get("id")
|
||||
dados_pessoais = doc.get("dadosPessoais", {})
|
||||
nome = dados_pessoais.get("nome", "N/A")
|
||||
cpf = dados_pessoais.get("cpf", "")
|
||||
atuacoes = doc.get("atuacoes", [])
|
||||
|
||||
coordenacoes_capes = extrair_coordenacoes_capes(atuacoes)
|
||||
consultoria = extrair_consultoria(atuacoes)
|
||||
premiacoes = extrair_premiacoes(atuacoes)
|
||||
|
||||
consultor = Consultor(
|
||||
id_pessoa=id_pessoa,
|
||||
nome=nome,
|
||||
cpf=cpf,
|
||||
coordenacoes_capes=coordenacoes_capes,
|
||||
coordenacoes_programas=[],
|
||||
consultoria=consultoria,
|
||||
premiacoes=premiacoes
|
||||
)
|
||||
|
||||
pontuacao = calculador.calcular_pontuacao_completa(consultor)
|
||||
consultor.pontuacao = pontuacao
|
||||
|
||||
consultores.append(consultor)
|
||||
|
||||
if idx % 10 == 0:
|
||||
print(f" Processados {idx}/{len(docs)} consultores...")
|
||||
|
||||
print("\nOrdenando por pontuação...")
|
||||
consultores.sort(key=lambda c: c.pontuacao_total, reverse=True)
|
||||
|
||||
print("\n" + "="*100)
|
||||
print("TOP 10 CONSULTORES - RANKING CAPES")
|
||||
print("="*100)
|
||||
|
||||
for rank, c in enumerate(consultores[:10], 1):
|
||||
print(f"\n{rank}º LUGAR - {c.nome}")
|
||||
print(f" ID: {c.id_pessoa}")
|
||||
print(f" PONTUAÇÃO TOTAL: {c.pontuacao_total:.2f} pontos")
|
||||
print(f" Componente A (Coordenação CAPES): {c.pontuacao.componente_a.total:.2f}")
|
||||
print(f" Base: {c.pontuacao.componente_a.base} | Tempo: {c.pontuacao.componente_a.tempo}")
|
||||
print(f" Extras: {c.pontuacao.componente_a.extras} | Bônus: {c.pontuacao.componente_a.bonus} | Retorno: {c.pontuacao.componente_a.retorno}")
|
||||
print(f" Componente B (Coordenação PPG): {c.pontuacao.componente_b.total:.2f}")
|
||||
print(f" Componente C (Consultoria): {c.pontuacao.componente_c.total:.2f}")
|
||||
if c.consultoria:
|
||||
print(f" Base: {c.pontuacao.componente_c.base} | Tempo: {c.pontuacao.componente_c.tempo}")
|
||||
print(f" Eventos: {c.consultoria.total_eventos} | Recentes: {c.consultoria.eventos_recentes}")
|
||||
print(f" Áreas: {', '.join(c.consultoria.areas[:3])}")
|
||||
print(f" Componente D (Premiações): {c.pontuacao.componente_d.total:.2f}")
|
||||
if c.premiacoes:
|
||||
print(f" Total de premiações: {len(c.premiacoes)}")
|
||||
|
||||
print(f" Anos de atuação: {c.anos_atuacao}")
|
||||
print(f" Ativo: {'Sim' if c.ativo else 'Não'}")
|
||||
|
||||
if c.coordenacoes_capes:
|
||||
print(f" Coordenações CAPES:")
|
||||
for coord in c.coordenacoes_capes[:3]:
|
||||
status = "ATIVA" if coord.periodo.ativo else "ENCERRADA"
|
||||
print(f" - {coord.tipo}: {coord.area_avaliacao} ({status})")
|
||||
|
||||
print("\n" + "="*100)
|
||||
print(f"\nEstatísticas Gerais:")
|
||||
print(f" Total de candidatos processados: {len(consultores)}")
|
||||
print(f" Pontuação máxima: {consultores[0].pontuacao_total:.2f}")
|
||||
print(f" Pontuação mínima (top 10): {consultores[9].pontuacao_total:.2f}")
|
||||
print(f" Média de pontuação (top 10): {sum(c.pontuacao_total for c in consultores[:10])/10:.2f}")
|
||||
|
||||
finally:
|
||||
await es_client.close()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
Reference in New Issue
Block a user