feat(selos): implementar sistema completo de selos para consultores
- Adicionar coluna SELOS na tabela TB_RANKING_CONSULTOR (migration v1.2) - Criar script popular_selos.py para popular/atualizar selos em batch - Simplificar para 22 selos com dados reais (remover selos sem dados) - Adicionar selo IDIOMA_MULTILINGUE para consultores com 3+ idiomas - Corrigir filtro de selos com query LIKE exata (evitar matches parciais) - Alinhar ícones entre FiltroSelos e ConsultorCard - Reorganizar filtro em 7 categorias: Coordenação, Consultoria, Avaliações, Premiações, Orientações, Participações, Características Selos disponíveis: CA, CAJ, CAJ_MP, CAM, PRESID_CAMARA, CONS_ATIVO, AVAL_COMIS, COORD_COMIS, AUTOR_GP, AUTOR_PREMIO, AUTOR_MENCAO, ORIENT_GP, ORIENT_PREMIO, ORIENT_MENCAO, COORIENT_GP, COORIENT_PREMIO, COORIENT_MENCAO, ORIENT_TESE, ORIENT_DISS, EVENTO, PROJ, IDIOMA_MULTILINGUE
This commit is contained in:
98
backend/scripts/popular_selos.py
Normal file
98
backend/scripts/popular_selos.py
Normal file
@@ -0,0 +1,98 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Script para popular a coluna SELOS na TB_RANKING_CONSULTOR.
|
||||
Lê JSON_DETALHES, extrai selos e atualiza a coluna.
|
||||
"""
|
||||
import os
|
||||
import sys
|
||||
import json
|
||||
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "src"))
|
||||
|
||||
from dotenv import load_dotenv
|
||||
load_dotenv(os.path.join(os.path.dirname(__file__), "..", ".env"))
|
||||
|
||||
from infrastructure.oracle.client import OracleClient
|
||||
from infrastructure.ranking_store import extrair_selos_entry
|
||||
|
||||
|
||||
def main():
|
||||
user = os.getenv("ORACLE_LOCAL_USER")
|
||||
password = os.getenv("ORACLE_LOCAL_PASSWORD")
|
||||
dsn = os.getenv("ORACLE_LOCAL_DSN", "localhost:1521/XEPDB1")
|
||||
|
||||
print(f"Conectando ao Oracle: {dsn}")
|
||||
client = OracleClient(user=user, password=password, dsn=dsn)
|
||||
client.connect()
|
||||
|
||||
recalcular_todos = "--all" in sys.argv
|
||||
|
||||
if recalcular_todos:
|
||||
query = """
|
||||
SELECT ID_PESSOA, JSON_DETALHES
|
||||
FROM TB_RANKING_CONSULTOR
|
||||
"""
|
||||
print("Re-calculando TODOS os selos...")
|
||||
else:
|
||||
query = """
|
||||
SELECT ID_PESSOA, JSON_DETALHES
|
||||
FROM TB_RANKING_CONSULTOR
|
||||
WHERE SELOS IS NULL
|
||||
"""
|
||||
print("Buscando consultores sem selos...")
|
||||
results = client.executar_query(query)
|
||||
print(f"Encontrados {len(results)} registros para atualizar")
|
||||
|
||||
batch_size = 500
|
||||
batch = []
|
||||
updated = 0
|
||||
|
||||
for r in results:
|
||||
id_pessoa = r["ID_PESSOA"]
|
||||
json_det = r["JSON_DETALHES"]
|
||||
|
||||
if hasattr(json_det, "read"):
|
||||
json_det = json_det.read()
|
||||
|
||||
try:
|
||||
detalhes = json.loads(json_det) if json_det else {}
|
||||
except (json.JSONDecodeError, TypeError):
|
||||
detalhes = {}
|
||||
|
||||
selos_set = extrair_selos_entry(detalhes)
|
||||
selos_str = ",".join(sorted(selos_set)) if selos_set else None
|
||||
|
||||
batch.append({"id_pessoa": id_pessoa, "selos": selos_str})
|
||||
|
||||
if len(batch) >= batch_size:
|
||||
_update_batch(client, batch)
|
||||
updated += len(batch)
|
||||
print(f" Atualizados: {updated}")
|
||||
batch = []
|
||||
|
||||
if batch:
|
||||
_update_batch(client, batch)
|
||||
updated += len(batch)
|
||||
print(f" Atualizados: {updated}")
|
||||
|
||||
print(f"Total de registros atualizados: {updated}")
|
||||
|
||||
|
||||
def _update_batch(client, batch):
|
||||
update_sql = """
|
||||
UPDATE TB_RANKING_CONSULTOR
|
||||
SET SELOS = :selos
|
||||
WHERE ID_PESSOA = :id_pessoa
|
||||
"""
|
||||
|
||||
with client.get_connection() as conn:
|
||||
cursor = conn.cursor()
|
||||
try:
|
||||
cursor.executemany(update_sql, batch)
|
||||
conn.commit()
|
||||
finally:
|
||||
cursor.close()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user