Fix Worker
This commit is contained in:
@@ -11,38 +11,35 @@ public class CriticalDataWorker : BackgroundService
|
||||
private readonly ILogger<CriticalDataWorker> _logger;
|
||||
private readonly SharedTokenService _tokenService;
|
||||
private readonly IServiceProvider _serviceProvider;
|
||||
private readonly IElectoralApiService _apiService; // <-- DEPENDENCIA AÑADIDA
|
||||
private readonly IElectoralApiService _apiService;
|
||||
|
||||
// Inyectamos IElectoralApiService en el constructor
|
||||
public CriticalDataWorker(
|
||||
ILogger<CriticalDataWorker> logger,
|
||||
SharedTokenService tokenService,
|
||||
IServiceProvider serviceProvider,
|
||||
IElectoralApiService apiService) // <-- PARÁMETRO AÑADIDO
|
||||
IElectoralApiService apiService)
|
||||
{
|
||||
_logger = logger;
|
||||
_tokenService = tokenService;
|
||||
_serviceProvider = serviceProvider;
|
||||
_apiService = apiService; // <-- ASIGNACIÓN AÑADIDA
|
||||
_apiService = apiService;
|
||||
}
|
||||
|
||||
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
|
||||
{
|
||||
_logger.LogInformation("Worker de Datos Críticos iniciado.");
|
||||
|
||||
// Damos tiempo a la sincronización inicial del otro worker para que se complete.
|
||||
try
|
||||
{
|
||||
await Task.Delay(TimeSpan.FromMinutes(2), stoppingToken);
|
||||
}
|
||||
catch (TaskCanceledException) { return; } // Salir si la app se apaga durante la espera inicial
|
||||
catch (TaskCanceledException) { return; }
|
||||
|
||||
int cicloContador = 0;
|
||||
while (!stoppingToken.IsCancellationRequested)
|
||||
{
|
||||
var cicloInicio = DateTime.UtcNow;
|
||||
cicloContador++;
|
||||
|
||||
_logger.LogInformation("--- Iniciando Ciclo de Datos Críticos #{ciclo} ---", cicloContador);
|
||||
|
||||
var authToken = await _tokenService.GetValidAuthTokenAsync(stoppingToken);
|
||||
@@ -68,200 +65,122 @@ public class CriticalDataWorker : BackgroundService
|
||||
{
|
||||
await Task.Delay(tiempoDeEspera, stoppingToken);
|
||||
}
|
||||
catch (TaskCanceledException)
|
||||
{
|
||||
break;
|
||||
}
|
||||
catch (TaskCanceledException) { break; }
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Sondea los resultados electorales para todos los municipios/partidos de forma optimizada.
|
||||
/// Utiliza paralelismo controlado para ejecutar múltiples peticiones a la API simultáneamente
|
||||
/// sin sobrecargar la red, y luego guarda todos los resultados en la base de datos de forma masiva.
|
||||
/// </summary>
|
||||
/// <param name="authToken">El token de autenticación válido para la sesión.</param>
|
||||
/// <param name="stoppingToken">El token de cancelación para detener la operación.</param>
|
||||
private async Task SondearResultadosMunicipalesAsync(string authToken, CancellationToken stoppingToken)
|
||||
{
|
||||
try
|
||||
{
|
||||
// PASO 1: Preparar el DbContext y los datos necesarios.
|
||||
using var scope = _serviceProvider.CreateScope();
|
||||
var dbContext = scope.ServiceProvider.GetRequiredService<EleccionesDbContext>();
|
||||
|
||||
// Obtenemos de nuestra BD local la lista de todos los partidos (NivelId=30) que necesitamos consultar.
|
||||
var municipiosASondear = await dbContext.AmbitosGeograficos
|
||||
.AsNoTracking()
|
||||
.Where(a => a.NivelId == 30 && a.DistritoId != null && a.SeccionId != null)
|
||||
.Select(a => new { a.Id, a.Nombre, a.MunicipioId, a.SeccionId, a.DistritoId })
|
||||
.ToListAsync(stoppingToken);
|
||||
|
||||
if (!municipiosASondear.Any())
|
||||
{
|
||||
_logger.LogWarning("No se encontraron Partidos (NivelId 30) en la BD para sondear resultados.");
|
||||
return;
|
||||
}
|
||||
|
||||
// Obtenemos la categoría "CONCEJALES", ya que los resultados municipales aplican a esta.
|
||||
var categoriaConcejales = await dbContext.CategoriasElectorales
|
||||
var todasLasCategorias = await dbContext.CategoriasElectorales
|
||||
.AsNoTracking()
|
||||
.FirstOrDefaultAsync(c => c.Nombre.Contains("CONCEJALES"), stoppingToken);
|
||||
.ToListAsync(stoppingToken);
|
||||
|
||||
if (categoriaConcejales == null)
|
||||
if (!municipiosASondear.Any() || !todasLasCategorias.Any())
|
||||
{
|
||||
_logger.LogWarning("No se encontró la categoría 'CONCEJALES'. Omitiendo sondeo de resultados municipales.");
|
||||
_logger.LogWarning("No se encontraron Partidos (NivelId 30) o Categorías para sondear resultados.");
|
||||
return;
|
||||
}
|
||||
|
||||
// PASO 2: Ejecutar las consultas a la API con paralelismo controlado.
|
||||
_logger.LogInformation("Iniciando sondeo de resultados para {m} municipios y {c} categorías...", municipiosASondear.Count, todasLasCategorias.Count);
|
||||
|
||||
// Definimos cuántas peticiones queremos que se ejecuten simultáneamente.
|
||||
// Un valor entre 8 y 16 es generalmente seguro y ofrece una gran mejora de velocidad.
|
||||
const int GRADO_DE_PARALELISMO = 3;
|
||||
// Creamos un semáforo que actuará como un "control de acceso" con 10 pases libres.
|
||||
var semaforo = new SemaphoreSlim(GRADO_DE_PARALELISMO);
|
||||
|
||||
// Usamos un ConcurrentDictionary para almacenar los resultados. A diferencia de un Dictionary normal,
|
||||
// este permite que múltiples tareas escriban en él al mismo tiempo sin conflictos.
|
||||
var resultadosPorId = new ConcurrentDictionary<int, Elecciones.Core.DTOs.ResultadosDto>();
|
||||
|
||||
_logger.LogInformation("Iniciando sondeo de resultados para {count} municipios con un paralelismo de {degree}...", municipiosASondear.Count, GRADO_DE_PARALELISMO);
|
||||
|
||||
// Creamos una lista de tareas (Tasks), una por cada municipio a consultar.
|
||||
// El método .Select() no ejecuta las tareas todavía, solo las prepara.
|
||||
var tareas = municipiosASondear.Select(async municipio =>
|
||||
foreach (var municipio in municipiosASondear)
|
||||
{
|
||||
// Cada tarea debe "pedir permiso" al semáforo antes de ejecutarse.
|
||||
// Si ya hay 10 tareas en ejecución, esta línea esperará hasta que una termine.
|
||||
await semaforo.WaitAsync(stoppingToken);
|
||||
try
|
||||
{
|
||||
// Una vez que obtiene el permiso, ejecuta la petición a la API.
|
||||
var resultados = await _apiService.GetResultadosAsync(
|
||||
authToken, municipio.DistritoId!, municipio.SeccionId!, null, categoriaConcejales.Id
|
||||
);
|
||||
if (stoppingToken.IsCancellationRequested) break;
|
||||
|
||||
var tareasCategoria = todasLasCategorias.Select(async categoria =>
|
||||
{
|
||||
var resultados = await _apiService.GetResultadosAsync(authToken, municipio.DistritoId!, municipio.SeccionId!, null, categoria.Id);
|
||||
|
||||
// Si la API devuelve datos válidos...
|
||||
if (resultados != null)
|
||||
{
|
||||
// ...los guardamos en el diccionario concurrente.
|
||||
resultadosPorId[municipio.Id] = resultados;
|
||||
using var innerScope = _serviceProvider.CreateScope();
|
||||
var innerDbContext = innerScope.ServiceProvider.GetRequiredService<EleccionesDbContext>();
|
||||
|
||||
// --- LLAMADA CORRECTA ---
|
||||
await GuardarResultadosDeAmbitoAsync(innerDbContext, municipio.Id, categoria.Id, resultados, stoppingToken);
|
||||
}
|
||||
}
|
||||
finally
|
||||
{
|
||||
// ¡CRUCIAL! Liberamos el pase del semáforo, permitiendo que la siguiente
|
||||
// tarea en espera pueda comenzar su ejecución.
|
||||
semaforo.Release();
|
||||
// Añadir un pequeño retraso aleatorio para no parecer un robot
|
||||
await Task.Delay(TimeSpan.FromMilliseconds(new Random().Next(50, 251)), stoppingToken);
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
// Ahora sí, ejecutamos todas las tareas preparadas en paralelo y esperamos a que todas terminen.
|
||||
await Task.WhenAll(tareas);
|
||||
|
||||
// PASO 3: Guardar los resultados en la base de datos.
|
||||
// Solo procedemos si recolectamos al menos un resultado válido.
|
||||
if (resultadosPorId.Any())
|
||||
{
|
||||
// Llamamos a nuestro método de guardado masivo y optimizado, pasándole todos los resultados
|
||||
// recolectados para que los inserte en una única y eficiente transacción.
|
||||
await GuardarResultadosDeMunicipiosAsync(dbContext, resultadosPorId.ToDictionary(kv => kv.Key, kv => kv.Value), stoppingToken);
|
||||
await Task.WhenAll(tareasCategoria);
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
// Capturamos cualquier error inesperado en el proceso para que el worker no se detenga.
|
||||
_logger.LogError(ex, "Ocurrió un error inesperado durante el sondeo de resultados municipales.");
|
||||
}
|
||||
}
|
||||
|
||||
/// Realiza una operación "Upsert" (Update o Insert) de forma masiva y optimizada.
|
||||
/// Este método es llamado por SondearResultadosMunicipalesAsync.
|
||||
/// </summary>
|
||||
private async Task GuardarResultadosDeMunicipiosAsync(
|
||||
EleccionesDbContext dbContext,
|
||||
Dictionary<int, Elecciones.Core.DTOs.ResultadosDto> todosLosResultados,
|
||||
CancellationToken stoppingToken) // <-- PARÁMETRO AÑADIDO
|
||||
private async Task GuardarResultadosDeAmbitoAsync(
|
||||
EleccionesDbContext dbContext, int ambitoId, int categoriaId,
|
||||
Elecciones.Core.DTOs.ResultadosDto resultadosDto, CancellationToken stoppingToken)
|
||||
{
|
||||
// Obtenemos los IDs de todos los ámbitos que vamos a actualizar.
|
||||
var ambitoIds = todosLosResultados.Keys;
|
||||
var estadoRecuento = await dbContext.EstadosRecuentos.FindAsync(new object[] { ambitoId, categoriaId }, stoppingToken);
|
||||
|
||||
// --- OPTIMIZACIÓN 1: Cargar todos los datos existentes en memoria UNA SOLA VEZ ---
|
||||
var estadosRecuentoExistentes = await dbContext.EstadosRecuentos
|
||||
.Where(e => ambitoIds.Contains(e.AmbitoGeograficoId))
|
||||
.ToDictionaryAsync(e => e.AmbitoGeograficoId, stoppingToken);
|
||||
|
||||
var resultadosVotosExistentes = await dbContext.ResultadosVotos
|
||||
.Where(rv => ambitoIds.Contains(rv.AmbitoGeograficoId))
|
||||
.GroupBy(rv => rv.AmbitoGeograficoId)
|
||||
.ToDictionaryAsync(g => g.Key, g => g.ToDictionary(item => item.AgrupacionPoliticaId), stoppingToken);
|
||||
|
||||
_logger.LogInformation("Procesando en memoria los resultados de {count} municipios.", todosLosResultados.Count);
|
||||
|
||||
// --- OPTIMIZACIÓN 2: Procesar todo en memoria ---
|
||||
foreach (var kvp in todosLosResultados)
|
||||
if (estadoRecuento == null)
|
||||
{
|
||||
var ambitoId = kvp.Key;
|
||||
var resultadosDto = kvp.Value;
|
||||
|
||||
// Lógica Upsert para EstadoRecuento
|
||||
if (!estadosRecuentoExistentes.TryGetValue(ambitoId, out var estadoRecuento))
|
||||
{
|
||||
estadoRecuento = new EstadoRecuento { AmbitoGeograficoId = ambitoId };
|
||||
dbContext.EstadosRecuentos.Add(estadoRecuento);
|
||||
}
|
||||
|
||||
// Mapeo completo de propiedades para EstadoRecuento
|
||||
estadoRecuento.FechaTotalizacion = DateTime.Parse(resultadosDto.FechaTotalizacion).ToUniversalTime();
|
||||
estadoRecuento.MesasEsperadas = resultadosDto.EstadoRecuento.MesasEsperadas;
|
||||
estadoRecuento.MesasTotalizadas = resultadosDto.EstadoRecuento.MesasTotalizadas;
|
||||
estadoRecuento.MesasTotalizadasPorcentaje = resultadosDto.EstadoRecuento.MesasTotalizadasPorcentaje;
|
||||
estadoRecuento.CantidadElectores = resultadosDto.EstadoRecuento.CantidadElectores;
|
||||
estadoRecuento.CantidadVotantes = resultadosDto.EstadoRecuento.CantidadVotantes;
|
||||
estadoRecuento.ParticipacionPorcentaje = resultadosDto.EstadoRecuento.ParticipacionPorcentaje;
|
||||
|
||||
if (resultadosDto.ValoresTotalizadosOtros != null)
|
||||
{
|
||||
estadoRecuento.VotosEnBlanco = resultadosDto.ValoresTotalizadosOtros.VotosEnBlanco;
|
||||
estadoRecuento.VotosEnBlancoPorcentaje = resultadosDto.ValoresTotalizadosOtros.VotosEnBlancoPorcentaje;
|
||||
estadoRecuento.VotosNulos = resultadosDto.ValoresTotalizadosOtros.VotosNulos;
|
||||
estadoRecuento.VotosNulosPorcentaje = resultadosDto.ValoresTotalizadosOtros.VotosNulosPorcentaje;
|
||||
estadoRecuento.VotosRecurridos = resultadosDto.ValoresTotalizadosOtros.VotosRecurridos;
|
||||
estadoRecuento.VotosRecurridosPorcentaje = resultadosDto.ValoresTotalizadosOtros.VotosRecurridosPorcentaje;
|
||||
}
|
||||
|
||||
// Lógica Upsert para ResultadosVotos
|
||||
var votosDeAmbitoExistentes = resultadosVotosExistentes.GetValueOrDefault(ambitoId);
|
||||
foreach (var votoPositivoDto in resultadosDto.ValoresTotalizadosPositivos)
|
||||
{
|
||||
ResultadoVoto? resultadoVoto = null;
|
||||
if (votosDeAmbitoExistentes != null)
|
||||
{
|
||||
votosDeAmbitoExistentes.TryGetValue(votoPositivoDto.IdAgrupacion, out resultadoVoto);
|
||||
}
|
||||
|
||||
if (resultadoVoto == null)
|
||||
{
|
||||
resultadoVoto = new ResultadoVoto
|
||||
{
|
||||
AmbitoGeograficoId = ambitoId,
|
||||
AgrupacionPoliticaId = votoPositivoDto.IdAgrupacion
|
||||
};
|
||||
dbContext.ResultadosVotos.Add(resultadoVoto);
|
||||
}
|
||||
resultadoVoto.CantidadVotos = votoPositivoDto.Votos;
|
||||
resultadoVoto.PorcentajeVotos = votoPositivoDto.VotosPorcentaje;
|
||||
}
|
||||
estadoRecuento = new EstadoRecuento { AmbitoGeograficoId = ambitoId, CategoriaId = categoriaId };
|
||||
dbContext.EstadosRecuentos.Add(estadoRecuento);
|
||||
}
|
||||
|
||||
// --- OPTIMIZACIÓN 3: Guardar todos los cambios en UNA SOLA TRANSACCIÓN ---
|
||||
_logger.LogInformation("Guardando todos los cambios de resultados municipales en la base de datos...");
|
||||
// Ahora 'stoppingToken' es reconocido aquí
|
||||
await dbContext.SaveChangesAsync(stoppingToken);
|
||||
_logger.LogInformation("Guardado completado.");
|
||||
estadoRecuento.FechaTotalizacion = DateTime.Parse(resultadosDto.FechaTotalizacion).ToUniversalTime();
|
||||
estadoRecuento.MesasEsperadas = resultadosDto.EstadoRecuento.MesasEsperadas;
|
||||
estadoRecuento.MesasTotalizadas = resultadosDto.EstadoRecuento.MesasTotalizadas;
|
||||
estadoRecuento.MesasTotalizadasPorcentaje = resultadosDto.EstadoRecuento.MesasTotalizadasPorcentaje;
|
||||
estadoRecuento.CantidadElectores = resultadosDto.EstadoRecuento.CantidadElectores;
|
||||
estadoRecuento.CantidadVotantes = resultadosDto.EstadoRecuento.CantidadVotantes;
|
||||
estadoRecuento.ParticipacionPorcentaje = resultadosDto.EstadoRecuento.ParticipacionPorcentaje;
|
||||
|
||||
if (resultadosDto.ValoresTotalizadosOtros != null)
|
||||
{
|
||||
estadoRecuento.VotosEnBlanco = resultadosDto.ValoresTotalizadosOtros.VotosEnBlanco;
|
||||
estadoRecuento.VotosEnBlancoPorcentaje = resultadosDto.ValoresTotalizadosOtros.VotosEnBlancoPorcentaje;
|
||||
estadoRecuento.VotosNulos = resultadosDto.ValoresTotalizadosOtros.VotosNulos;
|
||||
estadoRecuento.VotosNulosPorcentaje = resultadosDto.ValoresTotalizadosOtros.VotosNulosPorcentaje;
|
||||
estadoRecuento.VotosRecurridos = resultadosDto.ValoresTotalizadosOtros.VotosRecurridos;
|
||||
estadoRecuento.VotosRecurridosPorcentaje = resultadosDto.ValoresTotalizadosOtros.VotosRecurridosPorcentaje;
|
||||
}
|
||||
|
||||
foreach (var votoPositivoDto in resultadosDto.ValoresTotalizadosPositivos)
|
||||
{
|
||||
var resultadoVoto = await dbContext.ResultadosVotos.FirstOrDefaultAsync(
|
||||
rv => rv.AmbitoGeograficoId == ambitoId &&
|
||||
rv.CategoriaId == categoriaId &&
|
||||
rv.AgrupacionPoliticaId == votoPositivoDto.IdAgrupacion,
|
||||
stoppingToken
|
||||
);
|
||||
|
||||
if (resultadoVoto == null)
|
||||
{
|
||||
resultadoVoto = new ResultadoVoto
|
||||
{
|
||||
AmbitoGeograficoId = ambitoId,
|
||||
CategoriaId = categoriaId,
|
||||
AgrupacionPoliticaId = votoPositivoDto.IdAgrupacion
|
||||
};
|
||||
dbContext.ResultadosVotos.Add(resultadoVoto);
|
||||
}
|
||||
resultadoVoto.CantidadVotos = votoPositivoDto.Votos;
|
||||
resultadoVoto.PorcentajeVotos = votoPositivoDto.VotosPorcentaje;
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
await dbContext.SaveChangesAsync(stoppingToken);
|
||||
}
|
||||
catch (DbUpdateException ex)
|
||||
{
|
||||
_logger.LogError(ex, "DbUpdateException al guardar resultados para AmbitoId {ambitoId} y CategoriaId {categoriaId}", ambitoId, categoriaId);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
@@ -420,11 +339,4 @@ public class CriticalDataWorker : BackgroundService
|
||||
_logger.LogError(ex, "Ocurrió un error CRÍTICO en el sondeo de Estado Recuento General.");
|
||||
}
|
||||
}
|
||||
|
||||
// Pega aquí los métodos:
|
||||
// - SondearResultadosMunicipalesAsync
|
||||
// - GuardarResultadosDeMunicipiosAsync
|
||||
// - SondearResumenProvincialAsync
|
||||
// - SondearEstadoRecuentoGeneralAsync
|
||||
// (Estos métodos necesitan IServiceProvider y SharedTokenService, que ya están inyectados)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user