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

feat(export): adicionar exportação Excel do ranking com barra de progresso

Browse Source 
- Novo endpoint GET /api/v1/ranking/exportar/excel
- Exporta apenas consultores com pontuação > 0
- Usa xlsxwriter para geração rápida (~40s para 300k registros)
- Layout profissional com formatação, filtros e cores condicionais
- Barra de progresso real no frontend com dois estados:
  - Animação indeterminada durante geração no servidor
  - Progresso real durante download do arquivo
- Botão de exportação integrado ao layout do sistema
- Suporte a cancelamento da exportação
This commit is contained in:
Frederico Castro
2025-12-28 01:45:52 -03:00
parent 015c8f5741
commit 840934a187
9 changed files with 822 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

216
backend/src/application/services/excel_service.py Normal file
View File

@@ -0,0 +1,216 @@
from io import BytesIO
from datetime import datetime
from typing import List, Dict, Any, Optional
import json
import xlsxwriter
class ExcelService:
def gerar_ranking_excel(
self,
consultores: List[Dict[str, Any]],
filtros_aplicados: Optional[Dict[str, Any]] = None
) -> bytes:
output = BytesIO()
wb = xlsxwriter.Workbook(output, {'in_memory': True, 'constant_memory': True})
ws = wb.add_worksheet('Ranking Consultores')
title_fmt = wb.add_format({
'font_name': 'Calibri', 'font_size': 16, 'bold': True,
'font_color': '#1F4E79', 'align': 'center', 'valign': 'vcenter'
})
subtitle_fmt = wb.add_format({
'font_name': 'Calibri', 'font_size': 11,
'font_color': '#666666', 'align': 'center', 'valign': 'vcenter'
})
header_fmt = wb.add_format({
'font_name': 'Calibri', 'font_size': 11, 'bold': True,
'font_color': '#FFFFFF', 'bg_color': '#1F4E79',
'align': 'center', 'valign': 'vcenter', 'text_wrap': True,
'border': 1, 'border_color': '#D9D9D9'
})
data_fmt = wb.add_format({
'font_name': 'Calibri', 'font_size': 10,
'border': 1, 'border_color': '#D9D9D9'
})
data_center_fmt = wb.add_format({
'font_name': 'Calibri', 'font_size': 10,
'align': 'center', 'valign': 'vcenter',
'border': 1, 'border_color': '#D9D9D9'
})
data_number_fmt = wb.add_format({
'font_name': 'Calibri', 'font_size': 10,
'align': 'center', 'valign': 'vcenter',
'border': 1, 'border_color': '#D9D9D9',
'num_format': '#,##0'
})
data_decimal_fmt = wb.add_format({
'font_name': 'Calibri', 'font_size': 10,
'align': 'center', 'valign': 'vcenter',
'border': 1, 'border_color': '#D9D9D9',
'num_format': '#,##0.0'
})
ativo_fmt = wb.add_format({
'font_name': 'Calibri', 'font_size': 10, 'bold': True,
'align': 'center', 'valign': 'vcenter',
'bg_color': '#C6EFCE', 'font_color': '#006100',
'border': 1, 'border_color': '#D9D9D9'
})
inativo_fmt = wb.add_format({
'font_name': 'Calibri', 'font_size': 10, 'bold': True,
'align': 'center', 'valign': 'vcenter',
'bg_color': '#FFC7CE', 'font_color': '#9C0006',
'border': 1, 'border_color': '#D9D9D9'
})
ws.merge_range('A1:O1', 'RANKING DE CONSULTORES CAPES', title_fmt)
ws.set_row(0, 30)
data_geracao = datetime.now().strftime('%d/%m/%Y às %H:%M')
total = len(consultores)
filtros_texto = self._formatar_filtros(filtros_aplicados) if filtros_aplicados else ""
subtitulo = f"Gerado em {data_geracao} | Total: {total:,} consultores com pontuação"
if filtros_texto:
subtitulo += f" | Filtros: {filtros_texto}"
ws.merge_range('A2:O2', subtitulo, subtitle_fmt)
ws.set_row(1, 20)
headers = [
("Posição", 10),
("ID", 12),
("Nome", 40),
("Pontuação Total", 15),
("Bloco A\n(Coord. CAPES)", 14),
("Bloco C\n(Consultoria)", 14),
("Bloco D\n(Prêmios/Aval.)", 14),
("Bloco E\n(Coord. PPG)", 14),
("Status", 10),
("Anos Atuação", 12),
("Selos", 30),
("Coord. CAPES", 25),
("Situação Consultoria", 18),
("Prêmios", 25),
("Titulação", 35),
]
header_row = 3
for col_idx, (header_text, width) in enumerate(headers):
ws.write(header_row, col_idx, header_text, header_fmt)
ws.set_column(col_idx, col_idx, width)
ws.set_row(header_row, 35)
for row_idx, consultor in enumerate(consultores):
excel_row = header_row + 1 + row_idx
detalhes = self._parse_json_detalhes(consultor.get("JSON_DETALHES"))
selos = consultor.get("SELOS") or ""
is_ativo = consultor.get("ATIVO") == "S"
coord_capes = self._extrair_coordenacoes_resumo(detalhes)
situacao_cons = self._extrair_situacao_consultoria(detalhes)
premios = self._extrair_premios_resumo(detalhes)
titulacao = self._extrair_titulacao(detalhes)
ws.write_number(excel_row, 0, consultor.get("POSICAO") or 0, data_center_fmt)
ws.write_number(excel_row, 1, consultor.get("ID_PESSOA") or 0, data_center_fmt)
ws.write_string(excel_row, 2, consultor.get("NOME") or "", data_fmt)
ws.write_number(excel_row, 3, float(consultor.get("PONTUACAO_TOTAL") or 0), data_number_fmt)
ws.write_number(excel_row, 4, float(consultor.get("COMPONENTE_A") or 0), data_number_fmt)
ws.write_number(excel_row, 5, float(consultor.get("COMPONENTE_C") or 0), data_number_fmt)
ws.write_number(excel_row, 6, float(consultor.get("COMPONENTE_D") or 0), data_number_fmt)
ws.write_number(excel_row, 7, float(consultor.get("COMPONENTE_E") or 0), data_number_fmt)
ws.write_string(excel_row, 8, "Ativo" if is_ativo else "Inativo", ativo_fmt if is_ativo else inativo_fmt)
ws.write_number(excel_row, 9, float(consultor.get("ANOS_ATUACAO") or 0), data_decimal_fmt)
ws.write_string(excel_row, 10, selos.replace(",", ", "), data_fmt)
ws.write_string(excel_row, 11, coord_capes, data_fmt)
ws.write_string(excel_row, 12, situacao_cons, data_fmt)
ws.write_string(excel_row, 13, premios, data_fmt)
ws.write_string(excel_row, 14, titulacao, data_fmt)
last_row = header_row + len(consultores)
ws.autofilter(header_row, 0, last_row, 14)
ws.freeze_panes(header_row + 1, 0)
wb.close()
output.seek(0)
return output.getvalue()
def _parse_json_detalhes(self, json_str) -> Dict[str, Any]:
if not json_str:
return {}
if hasattr(json_str, "read"):
json_str = json_str.read()
try:
return json.loads(json_str) if isinstance(json_str, str) else json_str
except (json.JSONDecodeError, TypeError):
return {}
def _formatar_filtros(self, filtros: Dict[str, Any]) -> str:
partes = []
if filtros.get("ativo") is not None:
partes.append("Ativos" if filtros["ativo"] else "Inativos")
if filtros.get("selos"):
partes.append(f"Selos: {', '.join(filtros['selos'])}")
return " | ".join(partes)
def _extrair_coordenacoes_resumo(self, detalhes: Dict[str, Any]) -> str:
coords = detalhes.get("coordenacoes_capes", [])
if not coords:
return ""
resumos = []
for c in coords[:3]:
tipo = c.get("tipo", "")
area = c.get("area", "")
if tipo and area:
resumos.append(f"{tipo}: {area}")
elif tipo:
resumos.append(tipo)
return "; ".join(resumos)
def _extrair_situacao_consultoria(self, detalhes: Dict[str, Any]) -> str:
cons = detalhes.get("consultoria")
if not cons:
return ""
return cons.get("situacao", "")
def _extrair_premios_resumo(self, detalhes: Dict[str, Any]) -> str:
premios = detalhes.get("premiacoes", [])
if not premios:
return ""
resumos = []
for p in premios[:3]:
premio = p.get("premio", "")
tipo = p.get("tipo", "")
ano = p.get("ano", "")
if premio:
texto = f"{premio}"
if tipo:
texto += f" ({tipo})"
if ano:
texto += f" - {ano}"
resumos.append(texto)
if len(premios) > 3:
resumos.append(f"+{len(premios) - 3} outros")
return "; ".join(resumos)
def _extrair_titulacao(self, detalhes: Dict[str, Any]) -> str:
titulacao = detalhes.get("titulacao")
if titulacao and isinstance(titulacao, str):
return titulacao
lattes = detalhes.get("lattes", {})
titulacoes = lattes.get("titulacoes", []) if lattes else []
if titulacoes:
primeira = titulacoes[0]
grau = primeira.get("grau", "")
ies = primeira.get("ies_sigla", "")
ano = primeira.get("ano", "")
if grau:
texto = grau
if ies:
texto += f" - {ies}"
if ano:
texto += f" ({ano})"
return texto
return ""

96
backend/src/infrastructure/oracle/ranking_repository.py
View File

@@ -394,3 +394,99 @@ class RankingOracleRepository:
raise RuntimeError(f"Erro ao limpar tabela: {e2}")
finally:
cursor.close()
def buscar_para_exportacao(
self,
filtro_ativo: Optional[bool] = None,
filtro_selos: Optional[List[str]] = None,
batch_size: int = 5000
) -> List[Dict[str, Any]]:
"""
Busca todos os consultores com pontuação > 0 para exportação.
Otimizado para alto volume usando fetch em batches.
Retorna dicts crus do Oracle para processamento direto.
"""
where_clauses = ["PONTUACAO_TOTAL > 0"]
params = {}
if filtro_ativo is not None:
where_clauses.append("ATIVO = :ativo")
params["ativo"] = "S" if filtro_ativo else "N"
if filtro_selos:
for i, selo in enumerate(filtro_selos):
param_name = f"selo_{i}"
where_clauses.append(f"((',' || UPPER(SELOS) || ',') LIKE '%,' || :{param_name} || ',%')")
params[param_name] = str(selo).upper()
where_clause = " AND ".join(where_clauses)
query = f"""
SELECT
ID_PESSOA,
NOME,
POSICAO,
PONTUACAO_TOTAL,
COMPONENTE_A,
COMPONENTE_B,
COMPONENTE_C,
COMPONENTE_D,
COMPONENTE_E,
ATIVO,
ANOS_ATUACAO,
SELOS,
JSON_DETALHES
FROM TB_RANKING_CONSULTOR
WHERE {where_clause}
ORDER BY POSICAO NULLS LAST, PONTUACAO_TOTAL DESC
"""
resultados = []
with self.client.get_connection() as conn:
cursor = conn.cursor()
try:
cursor.arraysize = batch_size
cursor.execute(query, params)
colunas = [col[0] for col in cursor.description]
while True:
rows = cursor.fetchmany(batch_size)
if not rows:
break
for row in rows:
registro = dict(zip(colunas, row))
json_det = registro.get("JSON_DETALHES")
if hasattr(json_det, "read"):
registro["JSON_DETALHES"] = json_det.read()
resultados.append(registro)
finally:
cursor.close()
return resultados
def contar_para_exportacao(
self,
filtro_ativo: Optional[bool] = None,
filtro_selos: Optional[List[str]] = None
) -> int:
"""
Conta consultores com pontuação > 0 que seriam exportados.
"""
where_clauses = ["PONTUACAO_TOTAL > 0"]
params = {}
if filtro_ativo is not None:
where_clauses.append("ATIVO = :ativo")
params["ativo"] = "S" if filtro_ativo else "N"
if filtro_selos:
for i, selo in enumerate(filtro_selos):
param_name = f"selo_{i}"
where_clauses.append(f"((',' || UPPER(SELOS) || ',') LIKE '%,' || :{param_name} || ',%')")
params[param_name] = str(selo).upper()
where_clause = " AND ".join(where_clauses)
query = f"SELECT COUNT(*) AS TOTAL FROM TB_RANKING_CONSULTOR WHERE {where_clause}"
results = self.client.executar_query(query, params)
return results[0]["TOTAL"] if results else 0

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

@@ -992,3 +992,78 @@ async def listar_areas_avaliacao(
return [AreaAvaliacaoSchema(**a) for a in areas]
except Exception as e:
raise HTTPException(status_code=500, detail=f"Erro ao listar areas de avaliacao: {str(e)}")
@router.get("/ranking/exportar/excel")
async def exportar_ranking_excel(
ativo: Optional[bool] = Query(default=None, description="Filtrar por status ativo"),
selos: Optional[str] = Query(default=None, description="Filtrar por selos (separados por vírgula)"),
oracle_repo = Depends(get_ranking_oracle_repo),
):
from ...application.services.excel_service import ExcelService
if not oracle_repo:
raise HTTPException(status_code=503, detail="Oracle não configurado")
selos_lista = (
[s.strip().upper() for s in selos.split(",") if s.strip()]
if selos
else None
)
total = oracle_repo.contar_para_exportacao(filtro_ativo=ativo, filtro_selos=selos_lista)
if total == 0:
raise HTTPException(
status_code=404,
detail="Nenhum consultor com pontuação encontrado para os filtros aplicados."
)
consultores = oracle_repo.buscar_para_exportacao(
filtro_ativo=ativo,
filtro_selos=selos_lista
)
filtros = {"ativo": ativo, "selos": selos_lista}
excel_service = ExcelService()
excel_bytes = excel_service.gerar_ranking_excel(consultores, filtros)
data_atual = datetime.now().strftime('%Y%m%d_%H%M')
nome_arquivo = f"ranking_consultores_capes_{data_atual}.xlsx"
from fastapi.responses import Response
return Response(
content=excel_bytes,
media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
headers={
"Content-Disposition": f'attachment; filename="{nome_arquivo}"',
"Content-Length": str(len(excel_bytes)),
"Cache-Control": "no-cache"
}
)
@router.get("/ranking/exportar/info")
async def info_exportacao(
ativo: Optional[bool] = Query(default=None, description="Filtrar por status ativo"),
selos: Optional[str] = Query(default=None, description="Filtrar por selos (separados por vírgula)"),
oracle_repo = Depends(get_ranking_oracle_repo),
):
if not oracle_repo:
raise HTTPException(status_code=503, detail="Oracle não configurado")
selos_lista = (
[s.strip().upper() for s in selos.split(",") if s.strip()]
if selos
else None
)
total = oracle_repo.contar_para_exportacao(filtro_ativo=ativo, filtro_selos=selos_lista)
return {
"total_consultores": total,
"filtros": {
"ativo": ativo,
"selos": selos_lista
},
"estimativa_tamanho_mb": round(total * 0.003, 2)
}