"""Repositorio para gestionar jobs.""" from typing import Optional, List from datetime import datetime from dataclasses import dataclass import uuid from .database import db from ..constants import JobStatus @dataclass class Job: """Modelo de Job.""" job_id: str created_at: str updated_at: str source_path: str source_name: str source_hash: Optional[str] node_name: Optional[str] db_name: Optional[str] status: str attempts: int last_error: Optional[str] started_at: Optional[str] finished_at: Optional[str] total_ms: Optional[int] extract_ms: Optional[int] restore_ms: Optional[int] filelist_ms: Optional[int] class JobRepository: """Repositorio para operaciones con jobs.""" @staticmethod def create(source_path: str, source_name: str, source_hash: str) -> str: """ Crea un nuevo job. Args: source_path: Ruta completa del archivo fuente source_name: Nombre del archivo source_hash: Hash del archivo Returns: job_id del job creado """ job_id = str(uuid.uuid4()) now = datetime.utcnow().isoformat() db.execute(""" INSERT INTO jobs ( job_id, created_at, updated_at, source_path, source_name, source_hash, status, attempts ) VALUES (?, ?, ?, ?, ?, ?, ?, ?) """, (job_id, now, now, source_path, source_name, source_hash, JobStatus.QUEUED, 0)) return job_id @staticmethod def get(job_id: str) -> Optional[Job]: """Obtiene un job por ID.""" row = db.fetchone("SELECT * FROM jobs WHERE job_id = ?", (job_id,)) if not row: return None return Job(**dict(row)) @staticmethod def get_all( status: Optional[str] = None, limit: Optional[int] = None, offset: int = 0 ) -> List[Job]: """ Obtiene todos los jobs con filtros opcionales. Args: status: Filtrar por estado limit: Límite de resultados offset: Desplazamiento Returns: Lista de jobs """ query = "SELECT * FROM jobs" params = [] if status: query += " WHERE status = ?" params.append(status) query += " ORDER BY created_at DESC" if limit: query += f" LIMIT {limit} OFFSET {offset}" rows = db.fetchall(query, tuple(params)) return [Job(**dict(row)) for row in rows] @staticmethod def update_status( job_id: str, status: str, error: Optional[str] = None, increment_attempts: bool = False ): """Actualiza el estado de un job.""" now = datetime.utcnow().isoformat() updates = ["status = ?", "updated_at = ?"] params = [status, now] if error is not None: updates.append("last_error = ?") params.append(error) if increment_attempts: updates.append("attempts = attempts + 1") if status == JobStatus.EXTRACTING and not db.fetchone( "SELECT started_at FROM jobs WHERE job_id = ?", (job_id,) )["started_at"]: updates.append("started_at = ?") params.append(now) if status in [JobStatus.COMPLETED, JobStatus.FAILED, JobStatus.CANCELLED]: updates.append("finished_at = ?") params.append(now) params.append(job_id) db.execute( f"UPDATE jobs SET {', '.join(updates)} WHERE job_id = ?", tuple(params) ) @staticmethod def update_node_and_db(job_id: str, node_name: str, db_name: str): """Actualiza el nodo y base de datos de un job.""" now = datetime.utcnow().isoformat() db.execute( "UPDATE jobs SET node_name = ?, db_name = ?, updated_at = ? WHERE job_id = ?", (node_name, db_name, now, job_id) ) @staticmethod def update_timing( job_id: str, total_ms: Optional[int] = None, extract_ms: Optional[int] = None, restore_ms: Optional[int] = None, filelist_ms: Optional[int] = None ): """Actualiza los tiempos de ejecución.""" updates = [] params = [] if total_ms is not None: updates.append("total_ms = ?") params.append(total_ms) if extract_ms is not None: updates.append("extract_ms = ?") params.append(extract_ms) if restore_ms is not None: updates.append("restore_ms = ?") params.append(restore_ms) if filelist_ms is not None: updates.append("filelist_ms = ?") params.append(filelist_ms) if updates: params.append(job_id) db.execute( f"UPDATE jobs SET {', '.join(updates)} WHERE job_id = ?", tuple(params) ) @staticmethod def exists_by_hash(source_hash: str) -> bool: """Verifica si existe un job con el hash dado.""" row = db.fetchone( "SELECT COUNT(*) as count FROM jobs WHERE source_hash = ?", (source_hash,) ) return row["count"] > 0 @staticmethod def get_stats() -> dict: """Obtiene estadísticas de jobs.""" stats = { "total": 0, "queued": 0, "running": 0, "completed": 0, "failed": 0 } rows = db.fetchall("SELECT status, COUNT(*) as count FROM jobs GROUP BY status") for row in rows: status = row["status"] count = row["count"] stats["total"] += count if status == JobStatus.QUEUED: stats["queued"] = count elif status in [JobStatus.EXTRACTING, JobStatus.RESTORING, JobStatus.CLEANING]: stats["running"] += count elif status == JobStatus.COMPLETED: stats["completed"] = count elif status in [JobStatus.FAILED, JobStatus.FAILED_RESTART, JobStatus.CANCELLED]: stats["failed"] += count return stats @staticmethod def get_average_times() -> dict: """Obtiene tiempos promedio de ejecución.""" row = db.fetchone(""" SELECT AVG(total_ms) as avg_total, AVG(extract_ms) as avg_extract, AVG(restore_ms) as avg_restore FROM jobs WHERE status = ? AND total_ms IS NOT NULL """, (JobStatus.COMPLETED,)) return { "total_ms": int(row["avg_total"] or 0), "extract_ms": int(row["avg_extract"] or 0), "restore_ms": int(row["avg_restore"] or 0) }