311 lines
10 KiB
Python
311 lines
10 KiB
Python
"""Motor principal de la aplicación."""
|
|
|
|
from typing import Optional
|
|
from pathlib import Path
|
|
from PySide6.QtCore import QObject, Signal, QThreadPool
|
|
|
|
from .file_watcher import FileWatcher, FileStabilityChecker, calculate_file_hash
|
|
from .restore_worker import RestoreWorker
|
|
from ..db.job_repository import JobRepository
|
|
from ..db.event_repository import EventRepository
|
|
from ..db.config_repository import ConfigRepository
|
|
from ..constants import JobStatus, DEFAULT_CONFIG
|
|
from ..utils.logger import app_logger
|
|
|
|
|
|
class EngineSignals(QObject):
|
|
"""Señales del motor."""
|
|
job_created = Signal(str) # job_id
|
|
stats_updated = Signal(dict) # stats
|
|
config_loaded = Signal(dict) # config
|
|
|
|
|
|
class RestoreEngine(QObject):
|
|
"""Motor principal de restauración."""
|
|
|
|
def __init__(self):
|
|
"""Inicializa el motor."""
|
|
super().__init__()
|
|
|
|
self.signals = EngineSignals()
|
|
|
|
# Thread pool para workers
|
|
self._thread_pool = QThreadPool.globalInstance()
|
|
|
|
# File watcher
|
|
self._file_watcher: Optional[FileWatcher] = None
|
|
|
|
# Estado
|
|
self._running = False
|
|
self._paused = False
|
|
|
|
# Configuración
|
|
self._config = self._load_config()
|
|
|
|
app_logger.info("RestoreEngine inicializado")
|
|
|
|
def _load_config(self) -> dict:
|
|
"""Carga la configuración desde la base de datos."""
|
|
config = ConfigRepository.get("app_config", DEFAULT_CONFIG.copy())
|
|
|
|
# Asegurar que tiene todas las claves
|
|
for key, value in DEFAULT_CONFIG.items():
|
|
if key not in config:
|
|
config[key] = value
|
|
|
|
self.signals.config_loaded.emit(config)
|
|
return config
|
|
|
|
def save_config(self, config: dict):
|
|
"""
|
|
Guarda la configuración.
|
|
|
|
Args:
|
|
config: Configuración a guardar
|
|
"""
|
|
ConfigRepository.set("app_config", config)
|
|
self._config = config
|
|
app_logger.info("Configuración guardada")
|
|
|
|
# Reconfigurar file watcher si está corriendo
|
|
if self._running:
|
|
self._restart_file_watcher()
|
|
|
|
def get_config(self) -> dict:
|
|
"""Obtiene la configuración actual."""
|
|
return self._config.copy()
|
|
|
|
def start(self):
|
|
"""Inicia el motor."""
|
|
if self._running:
|
|
app_logger.warning("Motor ya está corriendo")
|
|
return
|
|
|
|
# Validar configuración
|
|
if not self._validate_config():
|
|
app_logger.error("Configuración inválida, no se puede iniciar el motor")
|
|
return
|
|
|
|
# Configurar thread pool
|
|
extract_workers = self._config["concurrency"]["extract_workers"]
|
|
restore_workers = self._config["concurrency"]["restore_workers"]
|
|
max_threads = extract_workers + restore_workers
|
|
self._thread_pool.setMaxThreadCount(max_threads)
|
|
|
|
app_logger.info(
|
|
f"Thread pool configurado: {max_threads} threads "
|
|
f"(extract: {extract_workers}, restore: {restore_workers})"
|
|
)
|
|
|
|
# Iniciar file watcher
|
|
if self._config["features"]["auto_scan_enabled"]:
|
|
self._start_file_watcher()
|
|
|
|
self._running = True
|
|
self._paused = False
|
|
|
|
EventRepository.create("INFO", "Motor iniciado")
|
|
app_logger.info("Motor iniciado")
|
|
|
|
def stop(self):
|
|
"""Detiene el motor."""
|
|
if not self._running:
|
|
return
|
|
|
|
# Detener file watcher
|
|
if self._file_watcher:
|
|
self._file_watcher.stop()
|
|
self._file_watcher = None
|
|
|
|
# Esperar a que terminen los workers
|
|
self._thread_pool.waitForDone(msecs=30000) # 30s timeout
|
|
|
|
self._running = False
|
|
|
|
EventRepository.create("INFO", "Motor detenido")
|
|
app_logger.info("Motor detenido")
|
|
|
|
def pause(self):
|
|
"""Pausa el procesamiento (no toma nuevos jobs)."""
|
|
self._paused = True
|
|
if self._file_watcher:
|
|
self._file_watcher.stop()
|
|
|
|
EventRepository.create("INFO", "Motor pausado")
|
|
app_logger.info("Motor pausado")
|
|
|
|
def resume(self):
|
|
"""Reanuda el procesamiento."""
|
|
if not self._running:
|
|
return
|
|
|
|
self._paused = False
|
|
if self._config["features"]["auto_scan_enabled"]:
|
|
self._start_file_watcher()
|
|
|
|
EventRepository.create("INFO", "Motor reanudado")
|
|
app_logger.info("Motor reanudado")
|
|
|
|
def is_running(self) -> bool:
|
|
"""Verifica si el motor está corriendo."""
|
|
return self._running
|
|
|
|
def is_paused(self) -> bool:
|
|
"""Verifica si el motor está pausado."""
|
|
return self._paused
|
|
|
|
def scan_now(self):
|
|
"""Fuerza un escaneo manual de la carpeta de entrada."""
|
|
if self._file_watcher:
|
|
self._file_watcher.scan_now()
|
|
else:
|
|
# Crear watcher temporal para un escaneo
|
|
self._start_file_watcher()
|
|
if self._file_watcher:
|
|
self._file_watcher.scan_now()
|
|
|
|
def process_file(self, file_path: str):
|
|
"""
|
|
Procesa un archivo manualmente.
|
|
|
|
Args:
|
|
file_path: Ruta al archivo ZIP
|
|
"""
|
|
try:
|
|
self._on_file_ready(file_path)
|
|
except Exception as e:
|
|
app_logger.error(f"Error procesando archivo {file_path}: {e}")
|
|
|
|
def get_stats(self) -> dict:
|
|
"""Obtiene estadísticas del motor."""
|
|
stats = JobRepository.get_stats()
|
|
stats["active_threads"] = self._thread_pool.activeThreadCount()
|
|
stats["max_threads"] = self._thread_pool.maxThreadCount()
|
|
stats["running"] = self._running
|
|
stats["paused"] = self._paused
|
|
|
|
# Tiempos promedio
|
|
avg_times = JobRepository.get_average_times()
|
|
stats.update(avg_times)
|
|
|
|
return stats
|
|
|
|
def _validate_config(self) -> bool:
|
|
"""Valida que la configuración sea correcta."""
|
|
paths = self._config["paths"]
|
|
|
|
# Validar carpetas requeridas
|
|
required_paths = ["input_folder", "extract_folder", "data_sql_folder"]
|
|
for key in required_paths:
|
|
if not paths.get(key):
|
|
app_logger.error(f"Falta configurar: {key}")
|
|
return False
|
|
|
|
path = Path(paths[key])
|
|
if not path.exists():
|
|
try:
|
|
path.mkdir(parents=True, exist_ok=True)
|
|
except Exception as e:
|
|
app_logger.error(f"No se pudo crear carpeta {key}: {e}")
|
|
return False
|
|
|
|
# Validar 7-Zip
|
|
if not paths.get("seven_zip_exe"):
|
|
app_logger.error("Falta configurar ruta a 7-Zip")
|
|
return False
|
|
|
|
if not Path(paths["seven_zip_exe"]).exists():
|
|
app_logger.error(f"7-Zip no encontrado en: {paths['seven_zip_exe']}")
|
|
return False
|
|
|
|
return True
|
|
|
|
def _start_file_watcher(self):
|
|
"""Inicia el file watcher."""
|
|
if self._file_watcher:
|
|
self._file_watcher.stop()
|
|
|
|
# Configurar stability checker
|
|
stability_config = self._config["stability"]
|
|
stability_checker = FileStabilityChecker(
|
|
check_interval=stability_config["check_interval_seconds"],
|
|
stable_duration=stability_config["stable_duration_seconds"],
|
|
use_ready_marker=stability_config["use_ready_marker"]
|
|
)
|
|
|
|
# Crear file watcher
|
|
self._file_watcher = FileWatcher(
|
|
watch_folder=self._config["paths"]["input_folder"],
|
|
on_file_ready=self._on_file_ready,
|
|
scan_interval=self._config["features"]["scan_interval_seconds"],
|
|
stability_checker=stability_checker
|
|
)
|
|
|
|
self._file_watcher.start()
|
|
|
|
def _restart_file_watcher(self):
|
|
"""Reinicia el file watcher con nueva configuración."""
|
|
if self._running and not self._paused:
|
|
self._start_file_watcher()
|
|
|
|
def _on_file_ready(self, file_path: str):
|
|
"""
|
|
Callback cuando un archivo está listo para procesar.
|
|
|
|
Args:
|
|
file_path: Ruta al archivo
|
|
"""
|
|
if self._paused:
|
|
app_logger.info(f"Motor pausado, ignorando archivo: {file_path}")
|
|
return
|
|
|
|
try:
|
|
# Calcular hash para evitar duplicados
|
|
file_hash = calculate_file_hash(file_path)
|
|
|
|
if JobRepository.exists_by_hash(file_hash):
|
|
app_logger.warning(f"Archivo ya procesado (hash duplicado): {file_path}")
|
|
return
|
|
|
|
# Crear job
|
|
file_name = Path(file_path).name
|
|
job_id = JobRepository.create(file_path, file_name, file_hash)
|
|
|
|
app_logger.info(f"Job creado: {job_id} para archivo {file_name}")
|
|
EventRepository.create("INFO", f"Nuevo job creado: {file_name}", job_id)
|
|
|
|
self.signals.job_created.emit(job_id)
|
|
|
|
# Crear worker y encolar
|
|
worker = RestoreWorker(
|
|
job_id=job_id,
|
|
config=self._config,
|
|
dry_run=self._config["features"]["dry_run_mode"]
|
|
)
|
|
|
|
# Conectar señales
|
|
worker.signals.job_completed.connect(self._on_job_completed)
|
|
|
|
# Ejecutar en thread pool
|
|
self._thread_pool.start(worker)
|
|
|
|
app_logger.info(f"Worker encolado para job {job_id}")
|
|
|
|
except Exception as e:
|
|
app_logger.error(f"Error creando job para {file_path}: {e}", exc_info=True)
|
|
EventRepository.create("ERROR", f"Error creando job: {e}")
|
|
|
|
def _on_job_completed(self, job_id: str, success: bool):
|
|
"""
|
|
Callback cuando un job se completa.
|
|
|
|
Args:
|
|
job_id: ID del job
|
|
success: Si fue exitoso
|
|
"""
|
|
app_logger.info(f"Job {job_id} completado - Éxito: {success}")
|
|
|
|
# Actualizar estadísticas
|
|
stats = self.get_stats()
|
|
self.signals.stats_updated.emit(stats)
|