Upload utils.py

utils.py

@@ -1,214 +1,214 @@
-# utils.py
-import os
-import yaml
-import subprocess
-from pathlib import Path
-from dataclasses import dataclass
-import shlex # Para manejar argumentos de línea de comandos de forma segura
-from __future__ import annotations
-from dataclasses import dataclass
-from typing import List, Optional, Callable
-import re
-import xml.etree.ElementTree as ET
-
-
-def incrustar_subtitulos_ffmpeg(
-    input_video_path: str,
-    srt_path: str,
-    output_video_path: str,
-    codificar_audio: bool = False,
-    sobreescribir: bool = False
-):
-    """
-    Incrusta (quema/hardsub) un archivo SRT sobre un vídeo MP4 usando FFmpeg.
-
-    IMPORTANTE: Requiere que FFmpeg esté instalado en el sistema.
-
-    :param input_video_path: Ruta al archivo de vídeo MP4 de entrada.
-    :param srt_path: Ruta al archivo de subtítulos SRT.
-    :param output_video_path: Ruta donde se guardará el nuevo vídeo con subtítulos.
-    :param codificar_audio: Si es True, recodifica el audio (más lento, pero a veces soluciona problemas).
-                            Si es False (por defecto), copia el stream de audio (-c:a copy).
-    :param sobreescribir: Si es True, permite sobreescribir el archivo de salida si ya existe.
-    :raises FileNotFoundError: Si no se encuentra FFmpeg o alguno de los archivos de entrada.
-    :raises subprocess.CalledProcessError: Si FFmpeg devuelve un error durante el proceso.
-    """
-
-    if not os.path.exists(input_video_path):
-        raise FileNotFoundError(f"Vídeo de entrada no encontrado: {input_video_path}")
-    if not os.path.exists(srt_path):
-        raise FileNotFoundError(f"Archivo SRT no encontrado: {srt_path}")
-
-    # Comando base de FFmpeg para incrustar subtítulos
-    # -i: archivo de entrada
-    # -vf: filtro de vídeo, usamos 'subtitles=' para quemar el srt
-    # -c:v: copia el stream de vídeo original (rápido, sin recodificar el vídeo)
-    
-    # IMPORTANTE: Para que FFmpeg pueda quemar subtítulos, debe usar el filtro 'subtitles'
-    # y *no* copiar el stream de vídeo (-c:v copy). Debemos recodificar el vídeo (-c:v libx264).
-    # Solo copiaremos el audio para ahorrar tiempo a menos que se indique lo contrario.
-
-    audio_codec_arg = ["-c:a", "copy"] if not codificar_audio else []
-    
-    # Se recomienda usar -c:v libx264 y -preset fast para una buena velocidad/calidad
-    video_codec_arg = ["-c:v", "libx264", "-preset", "fast"]
-
-    ffmpeg_command = [
-        "ffmpeg",
-        "-i", input_video_path,
-        "-vf", f"subtitles={srt_path}", # Filtro para quemar el SRT
-        *video_codec_arg,
-        *audio_codec_arg,
-        output_video_path
-    ]
-
-    if sobreescribir:
-        ffmpeg_command.insert(1, "-y") # Añade la opción de sobreescribir
-
-    print(f"Ejecutando comando FFmpeg: {' '.join(shlex.quote(arg) for arg in ffmpeg_command)}")
-
-    try:
-        # Ejecutar el comando FFmpeg
-        subprocess.run(
-            ffmpeg_command,
-            check=True,                 # Lanza CalledProcessError si el código de retorno no es 0
-            stdout=subprocess.PIPE,     # Captura la salida estándar (para no saturar)
-            stderr=subprocess.PIPE,     # Captura la salida de error (donde FFmpeg imprime el progreso)
-            text=True
-        )
-        print(f"\nÉxito: Vídeo con subtítulos guardado en: {output_video_path}")
-
-    except FileNotFoundError:
-        print("ERROR: El comando 'ffmpeg' no fue encontrado.")
-        print("Asegúrate de que FFmpeg esté instalado y accesible en tu PATH.")
-        raise
-    except subprocess.CalledProcessError as e:
-        print(f"ERROR: FFmpeg falló con el código {e.returncode}")
-        print("Salida de error de FFmpeg:")
-        print(e.stderr)
-        raise
-
-
-@dataclass
-class AppConfig:
-    app: dict
-    api: dict
-    storage: dict
-    sqlite: dict
-    security: dict
-    ui: dict
-
-def load_config(path: str = "config.yaml") -> AppConfig:
-    path = Path(path)
-    if not path.exists():
-        raise FileNotFoundError(f"No encuentro {path.resolve()}. Sube un config.yaml válido.")
-    with path.open("r", encoding="utf-8") as f:
-        cfg = yaml.safe_load(f) or {}
-    for k in ["app", "api", "storage", "sqlite", "security", "ui"]:
-        if k not in cfg:
-            raise ValueError(f"Falta la clave '{k}' en config.yaml")
-    return AppConfig(**cfg)
-
-def ensure_dirs(root: str | Path):
-    root = Path(root)
-    (root / "uploads").mkdir(parents=True, exist_ok=True)
-    (root / "results").mkdir(parents=True, exist_ok=True)
-
-def save_bytes(path: str | Path, data: bytes):
-    path = Path(path)
-    path.parent.mkdir(parents=True, exist_ok=True)
-    with open(path, "wb") as f:
-        f.write(data)
-
-def save_text(path: str | Path, text: str):
-    path = Path(path)
-    path.parent.mkdir(parents=True, exist_ok=True)
-    with open(path, "w", encoding="utf-8") as f:
-        f.write(text)
-
-def human_size(num_bytes: int) -> str:
-    units = ["B", "KB", "MB", "GB", "TB", "PB"]
-    n = float(num_bytes)
-    for u in units:
-        if n < 1024.0:
-            return f"{n:.1f} {u}"
-        n /= 1024.0
-    return f"{n:.1f} EB"
-
-def recortar_video(input_path: str, output_path: str, duracion_segundos: int = 240):
-    """
-    Corta los primeros `duracion_segundos` de un MP4 sin recodificar (rápido y sin pérdida).
-    Requiere ffmpeg en PATH (en conda-forge ya viene).
-    """
-    input_path = str(Path(input_path))
-    output_path = str(Path(output_path))
-
-    cmd = [
-        "ffmpeg",
-        "-y",                 # sobrescribe salida
-        "-hide_banner",
-        "-loglevel", "error", # mensajes limpios
-        "-ss", "0",           # inicio
-        "-i", input_path,     # entrada
-        "-t", str(duracion_segundos),  # duración
-        "-c", "copy",         # copia streams sin recodificar
-        output_path
-    ]
-    subprocess.run(cmd, check=True)
-
-
-# ---- Núcleo: SRT -> ESF (XML string) ----
-
-TIME_RE = re.compile(
-    r"(?P<start>\d{2}:\d{2}:\d{2}[,\.]\d{3})\s*-->\s*(?P<end>\d{2}:\d{2}:\d{2}[,\.]\d{3})"
-)
-
-@dataclass
-class Cue:
-    index: int
-    start: str  # "HH:MM:SS.mmm"
-    end: str    # "HH:MM:SS.mmm"
-    text: str
-
-def _norm_ts(ts: str) -> str:
-    """Convierte '01:02:03,456' -> '01:02:03.456'."""
-    return ts.replace(",", ".")
-
-def _parse_srt(srt_text: str) -> List[Cue]:
-    """Parsea SRT a una lista de cues normalizados."""
-    srt_text = srt_text.replace("\r\n", "\n").replace("\r", "\n")
-    blocks = [b.strip() for b in re.split(r"\n\s*\n", srt_text) if b.strip()]
-    cues: List[Cue] = []
-
-    for block in blocks:
-        lines = block.split("\n")
-        # Detectar si la primera línea es índice
-        idx = None
-        if lines and lines[0].strip().isdigit():
-            idx = int(lines[0].strip())
-            time_candidates = lines[1:]
-        else:
-            idx = len(cues) + 1
-            time_candidates = lines
-
-        m = None
-        time_line_idx = None
-        for i, ln in enumerate(time_candidates[:3]):  # robustez
-            mm = TIME_RE.search(ln)
-            if mm:
-                m = mm
-                time_line_idx = i
-                break
-        if not m:
-            raise ValueError(f"Bloque SRT sin tiempos válidos (índice {idx}):\n{block}")
-
-        start = _norm_ts(m.group("start"))
-        end = _norm_ts(m.group("end"))
-        text_lines = time_candidates[time_line_idx + 1 :]
-        text = "\n".join(text_lines).strip()
-
-        cues.append(Cue(index=idx, start=start, end=end, text=text))
-
+from __future__ import annotations
+
+# utils.py
+import os
+import yaml
+import subprocess
+from pathlib import Path
+from dataclasses import dataclass
+import shlex # Para manejar argumentos de línea de comandos de forma segura
+from typing import List, Optional, Callable
+import re
+import xml.etree.ElementTree as ET
+
+
+def incrustar_subtitulos_ffmpeg(
+    input_video_path: str,
+    srt_path: str,
+    output_video_path: str,
+    codificar_audio: bool = False,
+    sobreescribir: bool = False
+):
+    """
+    Incrusta (quema/hardsub) un archivo SRT sobre un vídeo MP4 usando FFmpeg.
+
+    IMPORTANTE: Requiere que FFmpeg esté instalado en el sistema.
+
+    :param input_video_path: Ruta al archivo de vídeo MP4 de entrada.
+    :param srt_path: Ruta al archivo de subtítulos SRT.
+    :param output_video_path: Ruta donde se guardará el nuevo vídeo con subtítulos.
+    :param codificar_audio: Si es True, recodifica el audio (más lento, pero a veces soluciona problemas).
+                            Si es False (por defecto), copia el stream de audio (-c:a copy).
+    :param sobreescribir: Si es True, permite sobreescribir el archivo de salida si ya existe.
+    :raises FileNotFoundError: Si no se encuentra FFmpeg o alguno de los archivos de entrada.
+    :raises subprocess.CalledProcessError: Si FFmpeg devuelve un error durante el proceso.
+    """
+
+    if not os.path.exists(input_video_path):
+        raise FileNotFoundError(f"Vídeo de entrada no encontrado: {input_video_path}")
+    if not os.path.exists(srt_path):
+        raise FileNotFoundError(f"Archivo SRT no encontrado: {srt_path}")
+
+    # Comando base de FFmpeg para incrustar subtítulos
+    # -i: archivo de entrada
+    # -vf: filtro de vídeo, usamos 'subtitles=' para quemar el srt
+    # -c:v: copia el stream de vídeo original (rápido, sin recodificar el vídeo)
+    
+    # IMPORTANTE: Para que FFmpeg pueda quemar subtítulos, debe usar el filtro 'subtitles'
+    # y *no* copiar el stream de vídeo (-c:v copy). Debemos recodificar el vídeo (-c:v libx264).
+    # Solo copiaremos el audio para ahorrar tiempo a menos que se indique lo contrario.
+
+    audio_codec_arg = ["-c:a", "copy"] if not codificar_audio else []
+    
+    # Se recomienda usar -c:v libx264 y -preset fast para una buena velocidad/calidad
+    video_codec_arg = ["-c:v", "libx264", "-preset", "fast"]
+
+    ffmpeg_command = [
+        "ffmpeg",
+        "-i", input_video_path,
+        "-vf", f"subtitles={srt_path}", # Filtro para quemar el SRT
+        *video_codec_arg,
+        *audio_codec_arg,
+        output_video_path
+    ]
+
+    if sobreescribir:
+        ffmpeg_command.insert(1, "-y") # Añade la opción de sobreescribir
+
+    print(f"Ejecutando comando FFmpeg: {' '.join(shlex.quote(arg) for arg in ffmpeg_command)}")
+
+    try:
+        # Ejecutar el comando FFmpeg
+        subprocess.run(
+            ffmpeg_command,
+            check=True,                 # Lanza CalledProcessError si el código de retorno no es 0
+            stdout=subprocess.PIPE,     # Captura la salida estándar (para no saturar)
+            stderr=subprocess.PIPE,     # Captura la salida de error (donde FFmpeg imprime el progreso)
+            text=True
+        )
+        print(f"\nÉxito: Vídeo con subtítulos guardado en: {output_video_path}")
+
+    except FileNotFoundError:
+        print("ERROR: El comando 'ffmpeg' no fue encontrado.")
+        print("Asegúrate de que FFmpeg esté instalado y accesible en tu PATH.")
+        raise
+    except subprocess.CalledProcessError as e:
+        print(f"ERROR: FFmpeg falló con el código {e.returncode}")
+        print("Salida de error de FFmpeg:")
+        print(e.stderr)
+        raise
+
+
+@dataclass
+class AppConfig:
+    app: dict
+    api: dict
+    storage: dict
+    sqlite: dict
+    security: dict
+    ui: dict
+
+def load_config(path: str = "config.yaml") -> AppConfig:
+    path = Path(path)
+    if not path.exists():
+        raise FileNotFoundError(f"No encuentro {path.resolve()}. Sube un config.yaml válido.")
+    with path.open("r", encoding="utf-8") as f:
+        cfg = yaml.safe_load(f) or {}
+    for k in ["app", "api", "storage", "sqlite", "security", "ui"]:
+        if k not in cfg:
+            raise ValueError(f"Falta la clave '{k}' en config.yaml")
+    return AppConfig(**cfg)
+
+def ensure_dirs(root: str | Path):
+    root = Path(root)
+    (root / "uploads").mkdir(parents=True, exist_ok=True)
+    (root / "results").mkdir(parents=True, exist_ok=True)
+
+def save_bytes(path: str | Path, data: bytes):
+    path = Path(path)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    with open(path, "wb") as f:
+        f.write(data)
+
+def save_text(path: str | Path, text: str):
+    path = Path(path)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    with open(path, "w", encoding="utf-8") as f:
+        f.write(text)
+
+def human_size(num_bytes: int) -> str:
+    units = ["B", "KB", "MB", "GB", "TB", "PB"]
+    n = float(num_bytes)
+    for u in units:
+        if n < 1024.0:
+            return f"{n:.1f} {u}"
+        n /= 1024.0
+    return f"{n:.1f} EB"
+
+def recortar_video(input_path: str, output_path: str, duracion_segundos: int = 240):
+    """
+    Corta los primeros `duracion_segundos` de un MP4 sin recodificar (rápido y sin pérdida).
+    Requiere ffmpeg en PATH (en conda-forge ya viene).
+    """
+    input_path = str(Path(input_path))
+    output_path = str(Path(output_path))
+
+    cmd = [
+        "ffmpeg",
+        "-y",                 # sobrescribe salida
+        "-hide_banner",
+        "-loglevel", "error", # mensajes limpios
+        "-ss", "0",           # inicio
+        "-i", input_path,     # entrada
+        "-t", str(duracion_segundos),  # duración
+        "-c", "copy",         # copia streams sin recodificar
+        output_path
+    ]
+    subprocess.run(cmd, check=True)
+
+
+# ---- Núcleo: SRT -> ESF (XML string) ----
+
+TIME_RE = re.compile(
+    r"(?P<start>\d{2}:\d{2}:\d{2}[,\.]\d{3})\s*-->\s*(?P<end>\d{2}:\d{2}:\d{2}[,\.]\d{3})"
+)
+
+@dataclass
+class Cue:
+    index: int
+    start: str  # "HH:MM:SS.mmm"
+    end: str    # "HH:MM:SS.mmm"
+    text: str
+
+def _norm_ts(ts: str) -> str:
+    """Convierte '01:02:03,456' -> '01:02:03.456'."""
+    return ts.replace(",", ".")
+
+def _parse_srt(srt_text: str) -> List[Cue]:
+    """Parsea SRT a una lista de cues normalizados."""
+    srt_text = srt_text.replace("\r\n", "\n").replace("\r", "\n")
+    blocks = [b.strip() for b in re.split(r"\n\s*\n", srt_text) if b.strip()]
+    cues: List[Cue] = []
+
+    for block in blocks:
+        lines = block.split("\n")
+        # Detectar si la primera línea es índice
+        idx = None
+        if lines and lines[0].strip().isdigit():
+            idx = int(lines[0].strip())
+            time_candidates = lines[1:]
+        else:
+            idx = len(cues) + 1
+            time_candidates = lines
+
+        m = None
+        time_line_idx = None
+        for i, ln in enumerate(time_candidates[:3]):  # robustez
+            mm = TIME_RE.search(ln)
+            if mm:
+                m = mm
+                time_line_idx = i
+                break
+        if not m:
+            raise ValueError(f"Bloque SRT sin tiempos válidos (índice {idx}):\n{block}")
+
+        start = _norm_ts(m.group("start"))
+        end = _norm_ts(m.group("end"))
+        text_lines = time_candidates[time_line_idx + 1 :]
+        text = "\n".join(text_lines).strip()
+
+        cues.append(Cue(index=idx, start=start, end=end, text=text))
+
     # Re-indexar por si venía desordenado
     for i, c in enumerate(cues, 1):
         c.index = i