Feat/Fix: Paralelismo y Coreccion de lista.
This commit is contained in:
@@ -89,12 +89,11 @@ public class ElectoralApiService : IElectoralApiService
|
||||
return response.IsSuccessStatusCode ? await response.Content.ReadFromJsonAsync<RepartoBancasDto>() : null;
|
||||
}
|
||||
|
||||
public async Task<List<string[]>?> GetTelegramasTotalizadosAsync(string authToken, string distritoId, string seccionId, int? categoriaId = null)
|
||||
public async Task<List<string>?> GetTelegramasTotalizadosAsync(string authToken, string distritoId, string seccionId, int? categoriaId = null)
|
||||
{
|
||||
var client = _httpClientFactory.CreateClient("ElectoralApiClient");
|
||||
var requestUri = $"/api/resultados/getTelegramasTotalizados?distritoId={distritoId}&seccionId={seccionId}";
|
||||
|
||||
// Añadimos el parámetro categoriaId a la URL SÓLO si se proporciona un valor.
|
||||
if (categoriaId.HasValue)
|
||||
{
|
||||
requestUri += $"&categoriaId={categoriaId.Value}";
|
||||
@@ -103,8 +102,9 @@ public class ElectoralApiService : IElectoralApiService
|
||||
var request = new HttpRequestMessage(HttpMethod.Get, requestUri);
|
||||
request.Headers.Add("Authorization", $"Bearer {authToken}");
|
||||
var response = await client.SendAsync(request);
|
||||
// Si la respuesta es 400, devolvemos null para que el worker sepa que falló.
|
||||
return response.IsSuccessStatusCode ? await response.Content.ReadFromJsonAsync<List<string[]>>() : null;
|
||||
|
||||
// Ahora deserializamos al tipo correcto: List<string>
|
||||
return response.IsSuccessStatusCode ? await response.Content.ReadFromJsonAsync<List<string>>() : null;
|
||||
}
|
||||
|
||||
public async Task<TelegramaFileDto?> GetTelegramaFileAsync(string authToken, string mesaId)
|
||||
|
||||
@@ -14,7 +14,7 @@ public interface IElectoralApiService
|
||||
Task<List<AgrupacionDto>?> GetAgrupacionesAsync(string authToken, string distritoId, int categoriaId);
|
||||
Task<ResultadosDto?> GetResultadosAsync(string authToken, string distritoId, string seccionId, string? municipioId, int categoriaId);
|
||||
Task<RepartoBancasDto?> GetBancasAsync(string authToken, string distritoId, string? seccionProvincialId, int categoriaId);
|
||||
Task<List<string[]>?> GetTelegramasTotalizadosAsync(string authToken, string distritoId, string seccionId, int? categoriaId = null);
|
||||
Task<List<string>?> GetTelegramasTotalizadosAsync(string authToken, string distritoId, string seccionId, int? categoriaId = null);
|
||||
Task<TelegramaFileDto?> GetTelegramaFileAsync(string authToken, string mesaId);
|
||||
Task<ResumenDto?> GetResumenAsync(string authToken, string distritoId);
|
||||
Task<EstadoRecuentoGeneralDto?> GetEstadoRecuentoGeneralAsync(string authToken, string distritoId, int categoriaId);
|
||||
|
||||
@@ -2,6 +2,7 @@ using Elecciones.Database;
|
||||
using Elecciones.Database.Entities;
|
||||
using Elecciones.Infrastructure.Services;
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
using System.Collections.Concurrent;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using System;
|
||||
using System.Linq;
|
||||
@@ -56,18 +57,18 @@ public class Worker : BackgroundService
|
||||
_logger.LogInformation("--- Iniciando sondeo de Resultados Municipales ---");
|
||||
await SondearResultadosMunicipalesAsync(authToken, stoppingToken);
|
||||
|
||||
_logger.LogInformation("--- Iniciando sondeo de Proyección de Bancas ---");
|
||||
await SondearProyeccionBancasAsync(authToken, stoppingToken);
|
||||
|
||||
_logger.LogInformation("--- Iniciando sondeo de Nuevos Telegramas ---");
|
||||
await SondearNuevosTelegramasAsync(authToken, stoppingToken);
|
||||
|
||||
_logger.LogInformation("--- Iniciando sondeo de Resumen Provincial ---");
|
||||
await SondearResumenProvincialAsync(authToken, stoppingToken);
|
||||
|
||||
_logger.LogInformation("--- Iniciando sondeo de Estado de Recuento General ---");
|
||||
await SondearEstadoRecuentoGeneralAsync(authToken, stoppingToken);
|
||||
|
||||
_logger.LogInformation("--- Iniciando sondeo de Proyección de Bancas ---");
|
||||
await SondearProyeccionBancasAsync(authToken, stoppingToken);
|
||||
|
||||
_logger.LogInformation("--- Iniciando sondeo de Nuevos Telegramas ---");
|
||||
await SondearNuevosTelegramasAsync(authToken, stoppingToken);
|
||||
|
||||
try
|
||||
{
|
||||
_logger.LogInformation("Ciclo de sondeo completado. Esperando 5 minutos para el siguiente...");
|
||||
@@ -83,174 +84,179 @@ public class Worker : BackgroundService
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Descarga y sincroniza los catálogos base (Categorías, Ámbitos, Agrupaciones)
|
||||
/// desde la API a la base de datos local. Se ejecuta una sola vez al iniciar el worker.
|
||||
/// Utiliza una estrategia de guardado en lotes para manejar grandes volúmenes de datos
|
||||
/// sin sobrecargar la base de datos.
|
||||
/// </summary>
|
||||
/// <param name="stoppingToken">El token de cancelación para detener la operación.</param>
|
||||
private async Task SincronizarCatalogosMaestrosAsync(CancellationToken stoppingToken)
|
||||
{
|
||||
try
|
||||
/// Descarga y sincroniza los catálogos base (Categorías, Ámbitos, Agrupaciones)
|
||||
/// desde la API a la base de datos local. Se ejecuta una sola vez al iniciar el worker.
|
||||
/// Utiliza una estrategia de guardado en lotes para manejar grandes volúmenes de datos
|
||||
/// sin sobrecargar la base de datos.
|
||||
/// </summary>
|
||||
/// <param name="stoppingToken">El token de cancelación para detener la operación.</param>
|
||||
private async Task SincronizarCatalogosMaestrosAsync(CancellationToken stoppingToken)
|
||||
{
|
||||
_logger.LogInformation("Iniciando sincronización de catálogos maestros...");
|
||||
|
||||
// PASO 1: Obtener el token de autenticación. Sin él, no podemos hacer nada.
|
||||
var authToken = await _apiService.GetAuthTokenAsync();
|
||||
if (string.IsNullOrEmpty(authToken) || stoppingToken.IsCancellationRequested)
|
||||
try
|
||||
{
|
||||
_logger.LogError("No se pudo obtener token para la sincronización de catálogos. La operación se cancela.");
|
||||
return;
|
||||
}
|
||||
_logger.LogInformation("Iniciando sincronización de catálogos maestros...");
|
||||
|
||||
// Creamos un scope de servicios para obtener una instancia fresca de DbContext.
|
||||
using var scope = _serviceProvider.CreateScope();
|
||||
var dbContext = scope.ServiceProvider.GetRequiredService<EleccionesDbContext>();
|
||||
|
||||
// PASO 2: Sincronizar las categorías electorales.
|
||||
// Es un catálogo pequeño y es la base para las siguientes consultas.
|
||||
var categoriasApi = await _apiService.GetCategoriasAsync(authToken);
|
||||
if (categoriasApi is null || !categoriasApi.Any())
|
||||
{
|
||||
_logger.LogWarning("La API no devolvió datos para el catálogo de Categorías. La sincronización no puede continuar.");
|
||||
return;
|
||||
}
|
||||
|
||||
var distinctCategorias = categoriasApi.GroupBy(c => c.CategoriaId).Select(g => g.First()).OrderBy(c => c.Orden).ToList();
|
||||
_logger.LogInformation("Se procesarán {count} categorías electorales.", distinctCategorias.Count);
|
||||
|
||||
var categoriasEnDb = await dbContext.CategoriasElectorales.ToDictionaryAsync(c => c.Id, c => c, stoppingToken);
|
||||
foreach (var categoriaDto in distinctCategorias)
|
||||
{
|
||||
if (!categoriasEnDb.ContainsKey(categoriaDto.CategoriaId))
|
||||
// PASO 1: Obtener el token de autenticación. Sin él, no podemos hacer nada.
|
||||
var authToken = await _apiService.GetAuthTokenAsync();
|
||||
if (string.IsNullOrEmpty(authToken) || stoppingToken.IsCancellationRequested)
|
||||
{
|
||||
dbContext.CategoriasElectorales.Add(new CategoriaElectoral { Id = categoriaDto.CategoriaId, Nombre = categoriaDto.Nombre, Orden = categoriaDto.Orden });
|
||||
_logger.LogError("No se pudo obtener token para la sincronización de catálogos. La operación se cancela.");
|
||||
return;
|
||||
}
|
||||
}
|
||||
// Guardamos las categorías primero para asegurar su existencia.
|
||||
await dbContext.SaveChangesAsync(stoppingToken);
|
||||
|
||||
// PASO 3: Cargar los catálogos existentes en memoria para una comparación eficiente.
|
||||
// Esto evita hacer miles de consultas a la BD dentro de un bucle.
|
||||
|
||||
// Para los ámbitos, creamos una clave única robusta que funciona incluso con campos nulos.
|
||||
var ambitosEnDb = new Dictionary<string, AmbitoGeografico>();
|
||||
var todosLosAmbitos = await dbContext.AmbitosGeograficos.ToListAsync(stoppingToken);
|
||||
foreach (var ambito in todosLosAmbitos)
|
||||
{
|
||||
string clave = $"{ambito.NivelId}|{ambito.DistritoId}|{ambito.SeccionProvincialId}|{ambito.SeccionId}|{ambito.MunicipioId}|{ambito.CircuitoId}|{ambito.EstablecimientoId}|{ambito.MesaId}";
|
||||
if (!ambitosEnDb.ContainsKey(clave))
|
||||
// Creamos un scope de servicios para obtener una instancia fresca de DbContext.
|
||||
using var scope = _serviceProvider.CreateScope();
|
||||
var dbContext = scope.ServiceProvider.GetRequiredService<EleccionesDbContext>();
|
||||
|
||||
// PASO 2: Sincronizar las categorías electorales.
|
||||
// Es un catálogo pequeño y es la base para las siguientes consultas.
|
||||
var categoriasApi = await _apiService.GetCategoriasAsync(authToken);
|
||||
if (categoriasApi is null || !categoriasApi.Any())
|
||||
{
|
||||
ambitosEnDb.Add(clave, ambito);
|
||||
_logger.LogWarning("La API no devolvió datos para el catálogo de Categorías. La sincronización no puede continuar.");
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
var agrupacionesEnDb = await dbContext.AgrupacionesPoliticas.ToDictionaryAsync(a => a.Id, a => a, stoppingToken);
|
||||
|
||||
// Variable para llevar la cuenta del total de registros insertados.
|
||||
int totalCambiosGuardados = 0;
|
||||
var distinctCategorias = categoriasApi.GroupBy(c => c.CategoriaId).Select(g => g.First()).OrderBy(c => c.Orden).ToList();
|
||||
_logger.LogInformation("Se procesarán {count} categorías electorales.", distinctCategorias.Count);
|
||||
|
||||
// PASO 4: Iterar sobre cada categoría para sincronizar sus ámbitos y agrupaciones.
|
||||
foreach (var categoria in distinctCategorias)
|
||||
{
|
||||
if (stoppingToken.IsCancellationRequested) break;
|
||||
_logger.LogInformation("--- Sincronizando datos para la categoría: {Nombre} (ID: {Id}) ---", categoria.Nombre, categoria.CategoriaId);
|
||||
|
||||
var catalogoDto = await _apiService.GetCatalogoAmbitosAsync(authToken, categoria.CategoriaId);
|
||||
if (catalogoDto != null)
|
||||
var categoriasEnDb = await dbContext.CategoriasElectorales.ToDictionaryAsync(c => c.Id, c => c, stoppingToken);
|
||||
foreach (var categoriaDto in distinctCategorias)
|
||||
{
|
||||
// 4.1 - Procesar y añadir ÁMBITOS nuevos al DbContext
|
||||
foreach (var ambitoDto in catalogoDto.Ambitos)
|
||||
if (!categoriasEnDb.ContainsKey(categoriaDto.CategoriaId))
|
||||
{
|
||||
string claveUnica = $"{ambitoDto.NivelId}|{ambitoDto.CodigoAmbitos.DistritoId}|{ambitoDto.CodigoAmbitos.SeccionProvincialId}|{ambitoDto.CodigoAmbitos.SeccionId}|{ambitoDto.CodigoAmbitos.MunicipioId}|{ambitoDto.CodigoAmbitos.CircuitoId}|{ambitoDto.CodigoAmbitos.EstablecimientoId}|{ambitoDto.CodigoAmbitos.MesaId}";
|
||||
|
||||
if (!ambitosEnDb.ContainsKey(claveUnica))
|
||||
{
|
||||
var nuevoAmbito = new AmbitoGeografico
|
||||
{
|
||||
Nombre = ambitoDto.Nombre,
|
||||
NivelId = ambitoDto.NivelId,
|
||||
DistritoId = ambitoDto.CodigoAmbitos.DistritoId,
|
||||
SeccionProvincialId = ambitoDto.CodigoAmbitos.SeccionProvincialId,
|
||||
SeccionId = ambitoDto.CodigoAmbitos.SeccionId,
|
||||
MunicipioId = ambitoDto.CodigoAmbitos.MunicipioId,
|
||||
CircuitoId = ambitoDto.CodigoAmbitos.CircuitoId,
|
||||
EstablecimientoId = ambitoDto.CodigoAmbitos.EstablecimientoId,
|
||||
MesaId = ambitoDto.CodigoAmbitos.MesaId,
|
||||
};
|
||||
dbContext.AmbitosGeograficos.Add(nuevoAmbito);
|
||||
ambitosEnDb.Add(claveUnica, nuevoAmbito); // Añadir también al diccionario en memoria
|
||||
}
|
||||
dbContext.CategoriasElectorales.Add(new CategoriaElectoral { Id = categoriaDto.CategoriaId, Nombre = categoriaDto.Nombre, Orden = categoriaDto.Orden });
|
||||
}
|
||||
}
|
||||
// Guardamos las categorías primero para asegurar su existencia.
|
||||
await dbContext.SaveChangesAsync(stoppingToken);
|
||||
|
||||
// 4.2 - Procesar y añadir AGRUPACIONES nuevas al DbContext
|
||||
var provincia = catalogoDto.Ambitos.FirstOrDefault(a => a.NivelId == 10);
|
||||
if (provincia != null && !string.IsNullOrEmpty(provincia.CodigoAmbitos.DistritoId))
|
||||
// PASO 3: Cargar los catálogos existentes en memoria para una comparación eficiente.
|
||||
// Esto evita hacer miles de consultas a la BD dentro de un bucle.
|
||||
|
||||
// Para los ámbitos, creamos una clave única robusta que funciona incluso con campos nulos.
|
||||
var ambitosEnDb = new Dictionary<string, AmbitoGeografico>();
|
||||
var todosLosAmbitos = await dbContext.AmbitosGeograficos.ToListAsync(stoppingToken);
|
||||
foreach (var ambito in todosLosAmbitos)
|
||||
{
|
||||
string clave = $"{ambito.NivelId}|{ambito.DistritoId}|{ambito.SeccionProvincialId}|{ambito.SeccionId}|{ambito.MunicipioId}|{ambito.CircuitoId}|{ambito.EstablecimientoId}|{ambito.MesaId}";
|
||||
if (!ambitosEnDb.ContainsKey(clave))
|
||||
{
|
||||
// Usamos un try-catch porque no todas las categorías tienen agrupaciones a nivel provincial.
|
||||
try
|
||||
ambitosEnDb.Add(clave, ambito);
|
||||
}
|
||||
}
|
||||
|
||||
var agrupacionesEnDb = await dbContext.AgrupacionesPoliticas.ToDictionaryAsync(a => a.Id, a => a, stoppingToken);
|
||||
|
||||
// Variable para llevar la cuenta del total de registros insertados.
|
||||
int totalCambiosGuardados = 0;
|
||||
|
||||
// PASO 4: Iterar sobre cada categoría para sincronizar sus ámbitos y agrupaciones.
|
||||
foreach (var categoria in distinctCategorias)
|
||||
{
|
||||
if (stoppingToken.IsCancellationRequested) break;
|
||||
_logger.LogInformation("--- Sincronizando datos para la categoría: {Nombre} (ID: {Id}) ---", categoria.Nombre, categoria.CategoriaId);
|
||||
|
||||
var catalogoDto = await _apiService.GetCatalogoAmbitosAsync(authToken, categoria.CategoriaId);
|
||||
if (catalogoDto != null)
|
||||
{
|
||||
// 4.1 - Procesar y añadir ÁMBITOS nuevos al DbContext
|
||||
foreach (var ambitoDto in catalogoDto.Ambitos)
|
||||
{
|
||||
var agrupacionesApi = await _apiService.GetAgrupacionesAsync(authToken, provincia.CodigoAmbitos.DistritoId, categoria.CategoriaId);
|
||||
if (agrupacionesApi != null && agrupacionesApi.Any())
|
||||
string claveUnica = $"{ambitoDto.NivelId}|{ambitoDto.CodigoAmbitos.DistritoId}|{ambitoDto.CodigoAmbitos.SeccionProvincialId}|{ambitoDto.CodigoAmbitos.SeccionId}|{ambitoDto.CodigoAmbitos.MunicipioId}|{ambitoDto.CodigoAmbitos.CircuitoId}|{ambitoDto.CodigoAmbitos.EstablecimientoId}|{ambitoDto.CodigoAmbitos.MesaId}";
|
||||
|
||||
if (!ambitosEnDb.ContainsKey(claveUnica))
|
||||
{
|
||||
foreach (var agrupacionDto in agrupacionesApi)
|
||||
var nuevoAmbito = new AmbitoGeografico
|
||||
{
|
||||
if (!agrupacionesEnDb.ContainsKey(agrupacionDto.IdAgrupacion))
|
||||
Nombre = ambitoDto.Nombre,
|
||||
NivelId = ambitoDto.NivelId,
|
||||
DistritoId = ambitoDto.CodigoAmbitos.DistritoId,
|
||||
SeccionProvincialId = ambitoDto.CodigoAmbitos.SeccionProvincialId,
|
||||
SeccionId = ambitoDto.CodigoAmbitos.SeccionId,
|
||||
MunicipioId = ambitoDto.CodigoAmbitos.MunicipioId,
|
||||
CircuitoId = ambitoDto.CodigoAmbitos.CircuitoId,
|
||||
EstablecimientoId = ambitoDto.CodigoAmbitos.EstablecimientoId,
|
||||
MesaId = ambitoDto.CodigoAmbitos.MesaId,
|
||||
};
|
||||
dbContext.AmbitosGeograficos.Add(nuevoAmbito);
|
||||
ambitosEnDb.Add(claveUnica, nuevoAmbito); // Añadir también al diccionario en memoria
|
||||
}
|
||||
}
|
||||
|
||||
// 4.2 - Procesar y añadir AGRUPACIONES nuevas al DbContext
|
||||
var provincia = catalogoDto.Ambitos.FirstOrDefault(a => a.NivelId == 10);
|
||||
if (provincia != null && !string.IsNullOrEmpty(provincia.CodigoAmbitos.DistritoId))
|
||||
{
|
||||
// Usamos un try-catch porque no todas las categorías tienen agrupaciones a nivel provincial.
|
||||
try
|
||||
{
|
||||
var agrupacionesApi = await _apiService.GetAgrupacionesAsync(authToken, provincia.CodigoAmbitos.DistritoId, categoria.CategoriaId);
|
||||
if (agrupacionesApi != null && agrupacionesApi.Any())
|
||||
{
|
||||
foreach (var agrupacionDto in agrupacionesApi)
|
||||
{
|
||||
var nuevaAgrupacion = new AgrupacionPolitica
|
||||
if (!agrupacionesEnDb.ContainsKey(agrupacionDto.IdAgrupacion))
|
||||
{
|
||||
Id = agrupacionDto.IdAgrupacion,
|
||||
IdTelegrama = agrupacionDto.IdAgrupacionTelegrama,
|
||||
Nombre = agrupacionDto.NombreAgrupacion
|
||||
};
|
||||
dbContext.AgrupacionesPoliticas.Add(nuevaAgrupacion);
|
||||
agrupacionesEnDb.Add(nuevaAgrupacion.Id, nuevaAgrupacion);
|
||||
var nuevaAgrupacion = new AgrupacionPolitica
|
||||
{
|
||||
Id = agrupacionDto.IdAgrupacion,
|
||||
IdTelegrama = agrupacionDto.IdAgrupacionTelegrama,
|
||||
Nombre = agrupacionDto.NombreAgrupacion
|
||||
};
|
||||
dbContext.AgrupacionesPoliticas.Add(nuevaAgrupacion);
|
||||
agrupacionesEnDb.Add(nuevaAgrupacion.Id, nuevaAgrupacion);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "No se pudieron obtener agrupaciones para la categoría '{catNombre}' ({catId}).", categoria.Nombre, categoria.CategoriaId);
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "No se pudieron obtener agrupaciones para la categoría '{catNombre}' ({catId}).", categoria.Nombre, categoria.CategoriaId);
|
||||
}
|
||||
}
|
||||
|
||||
// Después de procesar todos los ámbitos y agrupaciones de UNA categoría, guardamos los cambios.
|
||||
// Esto divide la inserción masiva de ~50,000 registros en 3 transacciones más pequeñas,
|
||||
// evitando timeouts y fallos en la base de datos.
|
||||
if (dbContext.ChangeTracker.HasChanges())
|
||||
{
|
||||
int cambiosEnLote = await dbContext.SaveChangesAsync(stoppingToken);
|
||||
totalCambiosGuardados += cambiosEnLote;
|
||||
_logger.LogInformation("Guardados {count} registros de catálogo para la categoría '{catNombre}'.", cambiosEnLote, categoria.Nombre);
|
||||
}
|
||||
}
|
||||
|
||||
// Después de procesar todos los ámbitos y agrupaciones de UNA categoría, guardamos los cambios.
|
||||
// Esto divide la inserción masiva de ~50,000 registros en 3 transacciones más pequeñas,
|
||||
// evitando timeouts y fallos en la base de datos.
|
||||
if (dbContext.ChangeTracker.HasChanges())
|
||||
{
|
||||
int cambiosEnLote = await dbContext.SaveChangesAsync(stoppingToken);
|
||||
totalCambiosGuardados += cambiosEnLote;
|
||||
_logger.LogInformation("Guardados {count} registros de catálogo para la categoría '{catNombre}'.", cambiosEnLote, categoria.Nombre);
|
||||
}
|
||||
// Ya no hay un SaveChangesAsync() gigante aquí.
|
||||
_logger.LogInformation("{count} nuevos registros de catálogo han sido guardados en total.", totalCambiosGuardados);
|
||||
_logger.LogInformation("Sincronización de catálogos maestros finalizada.");
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, "Ocurrió un error CRÍTICO durante la sincronización de catálogos.");
|
||||
}
|
||||
|
||||
// Ya no hay un SaveChangesAsync() gigante aquí.
|
||||
_logger.LogInformation("{count} nuevos registros de catálogo han sido guardados en total.", totalCambiosGuardados);
|
||||
_logger.LogInformation("Sincronización de catálogos maestros finalizada.");
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, "Ocurrió un error CRÍTICO durante la sincronización de catálogos.");
|
||||
}
|
||||
}
|
||||
|
||||
// El resto de los métodos (SondearResultadosMunicipalesAsync, GuardarResultadosDeAmbitoAsync, etc.)
|
||||
// se mantienen como en la versión anterior que te proporcioné. Los incluyo aquí para
|
||||
// que tengas el archivo completo y sin errores.
|
||||
/// <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)
|
||||
// El MunicipioId es opcional en la BD, lo quitamos del Where para asegurar que traiga todos los partidos
|
||||
.Select(a => new { a.Id, a.Nombre, a.MunicipioId, a.SeccionId, a.DistritoId })
|
||||
.ToListAsync(stoppingToken);
|
||||
|
||||
@@ -259,8 +265,8 @@ private async Task SincronizarCatalogosMaestrosAsync(CancellationToken stoppingT
|
||||
_logger.LogWarning("No se encontraron Partidos (NivelId 30) en la BD para sondear resultados.");
|
||||
return;
|
||||
}
|
||||
_logger.LogInformation("Iniciando sondeo de resultados para {count} municipios (Partidos)...", municipiosASondear.Count);
|
||||
|
||||
// Obtenemos la categoría "CONCEJALES", ya que los resultados municipales aplican a esta.
|
||||
var categoriaConcejales = await dbContext.CategoriasElectorales
|
||||
.AsNoTracking()
|
||||
.FirstOrDefaultAsync(c => c.Nombre.Contains("CONCEJALES"), stoppingToken);
|
||||
@@ -271,29 +277,64 @@ private async Task SincronizarCatalogosMaestrosAsync(CancellationToken stoppingT
|
||||
return;
|
||||
}
|
||||
|
||||
var todosLosResultados = new Dictionary<int, Elecciones.Core.DTOs.ResultadosDto>();
|
||||
foreach (var municipio in municipiosASondear)
|
||||
// PASO 2: Ejecutar las consultas a la API con paralelismo controlado.
|
||||
|
||||
// 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 = 10;
|
||||
// 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 =>
|
||||
{
|
||||
if (stoppingToken.IsCancellationRequested) break;
|
||||
|
||||
var resultados = await _apiService.GetResultadosAsync(
|
||||
authToken, municipio.DistritoId!, municipio.SeccionId!, null, categoriaConcejales.Id
|
||||
);
|
||||
|
||||
if (resultados != null)
|
||||
// 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
|
||||
{
|
||||
todosLosResultados[municipio.Id] = resultados;
|
||||
}
|
||||
}
|
||||
// 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 (todosLosResultados.Any())
|
||||
// Si la API devuelve datos válidos...
|
||||
if (resultados != null)
|
||||
{
|
||||
// ...los guardamos en el diccionario concurrente.
|
||||
resultadosPorId[municipio.Id] = resultados;
|
||||
}
|
||||
}
|
||||
finally
|
||||
{
|
||||
// ¡CRUCIAL! Liberamos el pase del semáforo, permitiendo que la siguiente
|
||||
// tarea en espera pueda comenzar su ejecución.
|
||||
semaforo.Release();
|
||||
}
|
||||
});
|
||||
|
||||
// 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())
|
||||
{
|
||||
// La llamada ahora es correcta porque el método receptor espera 3 argumentos
|
||||
await GuardarResultadosDeMunicipiosAsync(dbContext, todosLosResultados, stoppingToken);
|
||||
// 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);
|
||||
}
|
||||
}
|
||||
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.");
|
||||
}
|
||||
}
|
||||
@@ -498,10 +539,10 @@ private async Task SincronizarCatalogosMaestrosAsync(CancellationToken stoppingT
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Busca en la API si hay nuevos telegramas totalizados que no se encuentren en la base de datos local.
|
||||
/// Este método itera sobre cada Partido/Municipio (que la API identifica como "Sección" con NivelId = 30),
|
||||
/// obtiene la lista de IDs de telegramas para cada uno, los compara con los IDs locales,
|
||||
/// y finalmente descarga y guarda solo los que son nuevos.
|
||||
/// Busca y descarga nuevos telegramas de forma masiva y concurrente.
|
||||
/// Este método crea una lista de todas las combinaciones de Partido/Categoría,
|
||||
/// las consulta a la API con un grado de paralelismo controlado, y cada tarea concurrente
|
||||
/// maneja su propia lógica de descarga y guardado en la base de datos.
|
||||
/// </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>
|
||||
@@ -509,16 +550,17 @@ private async Task SincronizarCatalogosMaestrosAsync(CancellationToken stoppingT
|
||||
{
|
||||
try
|
||||
{
|
||||
using var scope = _serviceProvider.CreateScope();
|
||||
var dbContext = scope.ServiceProvider.GetRequiredService<EleccionesDbContext>();
|
||||
// PASO 1: Obtener los datos base para las consultas.
|
||||
// Usamos un DbContext inicial solo para leer los catálogos.
|
||||
using var initialScope = _serviceProvider.CreateScope();
|
||||
var initialDbContext = initialScope.ServiceProvider.GetRequiredService<EleccionesDbContext>();
|
||||
|
||||
// Obtenemos todos los partidos (NivelId=30) y todas las categorías para iterar
|
||||
var partidos = await dbContext.AmbitosGeograficos
|
||||
var partidos = await initialDbContext.AmbitosGeograficos
|
||||
.AsNoTracking()
|
||||
.Where(a => a.NivelId == 30 && a.DistritoId != null && a.SeccionId != null)
|
||||
.ToListAsync(stoppingToken);
|
||||
|
||||
var categorias = await dbContext.CategoriasElectorales
|
||||
var categorias = await initialDbContext.CategoriasElectorales
|
||||
.AsNoTracking()
|
||||
.ToListAsync(stoppingToken);
|
||||
|
||||
@@ -528,26 +570,29 @@ private async Task SincronizarCatalogosMaestrosAsync(CancellationToken stoppingT
|
||||
return;
|
||||
}
|
||||
|
||||
_logger.LogInformation("Iniciando sondeo de Telegramas nuevos para {partidosCount} partidos y {categoriasCount} categorías...", partidos.Count, categorias.Count);
|
||||
// Creamos una lista de todas las consultas que necesitamos hacer (135 partidos * 3 categorías = 405 consultas).
|
||||
var combinaciones = partidos.SelectMany(partido => categorias, (partido, categoria) => new { partido, categoria });
|
||||
|
||||
foreach (var partido in partidos)
|
||||
const int GRADO_DE_PARALELISMO = 10;
|
||||
var semaforo = new SemaphoreSlim(GRADO_DE_PARALELISMO);
|
||||
|
||||
_logger.LogInformation("Iniciando sondeo de Telegramas para {count} combinaciones... con paralelismo de {degree}", combinaciones.Count(), GRADO_DE_PARALELISMO);
|
||||
|
||||
var tareas = combinaciones.Select(async item =>
|
||||
{
|
||||
if (stoppingToken.IsCancellationRequested) break;
|
||||
|
||||
foreach (var categoria in categorias)
|
||||
await semaforo.WaitAsync(stoppingToken);
|
||||
try
|
||||
{
|
||||
if (stoppingToken.IsCancellationRequested) break;
|
||||
var idsDeApi = await _apiService.GetTelegramasTotalizadosAsync(authToken, item.partido.DistritoId!, item.partido.SeccionId!, item.categoria.Id);
|
||||
|
||||
// Llamamos al servicio pasando la categoriaId.
|
||||
// Si mañana la API ya no lo necesita, simplemente lo ignorará.
|
||||
// Si lo vuelve a necesitar en el futuro, nuestro código ya está preparado.
|
||||
var listaTelegramasApi = await _apiService.GetTelegramasTotalizadosAsync(authToken, partido.DistritoId!, partido.SeccionId!, categoria.Id);
|
||||
|
||||
// El resto de la lógica es la misma: si la respuesta es válida y contiene datos...
|
||||
if (listaTelegramasApi is { Count: > 0 })
|
||||
if (idsDeApi is { Count: > 0 })
|
||||
{
|
||||
var idsDeApi = listaTelegramasApi.Select(t => t[0]).Distinct().ToList();
|
||||
var idsYaEnDb = await dbContext.Telegramas
|
||||
using var innerScope = _serviceProvider.CreateScope();
|
||||
var innerDbContext = innerScope.ServiceProvider.GetRequiredService<EleccionesDbContext>();
|
||||
|
||||
// --- CORRECCIÓN CLAVE ---
|
||||
// 'idsDeApi' ya es una List<string>, no necesitamos hacer .Select(t => t[0])
|
||||
var idsYaEnDb = await innerDbContext.Telegramas
|
||||
.Where(t => idsDeApi.Contains(t.Id))
|
||||
.Select(t => t.Id)
|
||||
.ToListAsync(stoppingToken);
|
||||
@@ -556,11 +601,12 @@ private async Task SincronizarCatalogosMaestrosAsync(CancellationToken stoppingT
|
||||
|
||||
if (!nuevosTelegramasIds.Any())
|
||||
{
|
||||
continue;
|
||||
return;
|
||||
}
|
||||
|
||||
_logger.LogInformation("Se encontraron {count} telegramas nuevos en el partido '{nombre}' para la categoría '{cat}'. Descargando...", nuevosTelegramasIds.Count, partido.Nombre, categoria.Nombre);
|
||||
_logger.LogInformation("Se encontraron {count} telegramas nuevos en '{partido}' para '{cat}'. Descargando...", nuevosTelegramasIds.Count, item.partido.Nombre, item.categoria.Nombre);
|
||||
|
||||
// Iteramos y descargamos cada nuevo telegrama.
|
||||
foreach (var mesaId in nuevosTelegramasIds)
|
||||
{
|
||||
if (stoppingToken.IsCancellationRequested) break;
|
||||
@@ -570,18 +616,28 @@ private async Task SincronizarCatalogosMaestrosAsync(CancellationToken stoppingT
|
||||
var nuevoTelegrama = new Telegrama
|
||||
{
|
||||
Id = telegramaFile.NombreArchivo,
|
||||
AmbitoGeograficoId = partido.Id,
|
||||
AmbitoGeograficoId = item.partido.Id,
|
||||
ContenidoBase64 = telegramaFile.Imagen,
|
||||
FechaEscaneo = DateTime.Parse(telegramaFile.FechaEscaneo).ToUniversalTime(),
|
||||
FechaTotalizacion = DateTime.Parse(telegramaFile.FechaTotalizacion).ToUniversalTime()
|
||||
};
|
||||
await dbContext.Telegramas.AddAsync(nuevoTelegrama, stoppingToken);
|
||||
await innerDbContext.Telegramas.AddAsync(nuevoTelegrama, stoppingToken);
|
||||
}
|
||||
}
|
||||
await dbContext.SaveChangesAsync(stoppingToken);
|
||||
// Guardamos los cambios de ESTA tarea específica en la BD.
|
||||
await innerDbContext.SaveChangesAsync(stoppingToken);
|
||||
}
|
||||
}
|
||||
}
|
||||
finally
|
||||
{
|
||||
// Liberamos el semáforo para que otra tarea pueda comenzar.
|
||||
semaforo.Release();
|
||||
}
|
||||
});
|
||||
|
||||
// Ejecutamos todas las tareas en paralelo y esperamos a que finalicen.
|
||||
await Task.WhenAll(tareas);
|
||||
|
||||
_logger.LogInformation("Sondeo de Telegramas completado.");
|
||||
}
|
||||
catch (Exception ex)
|
||||
|
||||
Reference in New Issue
Block a user