#!/usr/bin/env python3 """CLI mínimo que expone el orquestador principal. Este módulo proporciona la función `main()` que construye los adaptadores por defecto e invoca `PipelineOrchestrator.run(...)`. Está diseñado para reemplazar el antiguo `run_full_pipeline.py` como punto de entrada. """ from __future__ import annotations import argparse import glob import os import shutil import sys import tempfile from whisper_project.logging_config import configure_logging import logging from whisper_project.usecases.orchestrator import PipelineOrchestrator from whisper_project.infra.kokoro_adapter import KokoroHttpClient from whisper_project import config as project_config def main(): p = argparse.ArgumentParser() p.add_argument("--video", required=True) p.add_argument("--srt", help="SRT de entrada (opcional)") p.add_argument( "--kokoro-endpoint", required=False, default=project_config.KOKORO_ENDPOINT or "https://kokoro.example/api/synthesize", help=( "Endpoint HTTP de Kokoro. Si existe en ./config.toml se usará por defecto. " "(override con esta opción para cambiar)" ), ) p.add_argument( "--kokoro-key", required=False, default=project_config.KOKORO_API_KEY, help=( "API key para Kokoro. Si existe en ./config.toml se usará por defecto. " "(override con esta opción para cambiar)" ), ) p.add_argument("--voice", default="em_alex") p.add_argument("--kokoro-model", default="model") p.add_argument("--whisper-model", default="base") p.add_argument( "--translate-method", choices=[ "local", "gemini", "argos", "none", ], default="local", ) p.add_argument( "--gemini-key", default=None, help=( "API key para Gemini (si eliges " "--translate-method=gemini)" ), ) p.add_argument("--mix", action="store_true") p.add_argument("--mix-background-volume", type=float, default=0.1) p.add_argument("--keep-chunks", action="store_true") p.add_argument("--keep-temp", action="store_true") p.add_argument( "--dry-run", action="store_true", help="Simular pasos sin ejecutar", ) p.add_argument( "--srt-only", action="store_true", help="Solo extraer y traducir el SRT, devolver la ruta del SRT traducido y no ejecutar TTS ni quemado", ) p.add_argument( "--verbose", action="store_true", help="Activar logging DEBUG", ) args = p.parse_args() # configurar logging según flag (crea carpeta logs/ y handlers por tipo) configure_logging(args.verbose) video = os.path.abspath(args.video) if not os.path.exists(video): logging.getLogger(__name__).error("Vídeo no encontrado: %s", video) sys.exit(2) workdir = tempfile.mkdtemp(prefix="full_pipeline_") logging.info("Workdir creado: %s", workdir) try: # construir cliente Kokoro HTTP nativo e inyectarlo en el orquestador kokoro_client = KokoroHttpClient( args.kokoro_endpoint, api_key=args.kokoro_key, voice=args.voice, model=args.kokoro_model, ) orchestrator = PipelineOrchestrator( kokoro_endpoint=args.kokoro_endpoint, kokoro_key=args.kokoro_key, voice=args.voice, kokoro_model=args.kokoro_model, tts_client=kokoro_client, ) logging.info( "Lanzando orquestador (dry_run=%s, translate_method=%s, whisper_model=%s)", args.dry_run, args.translate_method, args.whisper_model, ) result = orchestrator.run( video=video, srt=args.srt, workdir=workdir, translate_method=args.translate_method, gemini_api_key=args.gemini_key, whisper_model=args.whisper_model, mix=args.mix, mix_background_volume=args.mix_background_volume, keep_chunks=args.keep_chunks, dry_run=args.dry_run, srt_only=args.srt_only, ) logging.info("Orquestador finalizado; resultado (burned_video): %s", getattr(result, "burned_video", None)) # Si no es dry-run, crear una subcarpeta por proyecto en output/ # (output/) y mover allí los artefactos generados. final_path = None if ( not args.dry_run and result and getattr(result, "burned_video", None) ): base = os.path.splitext(os.path.basename(video))[0] project_out = os.path.join(os.getcwd(), "output", base) try: os.makedirs(project_out, exist_ok=True) except Exception: pass # Mover el vídeo principal src = result.burned_video dest = os.path.join(project_out, os.path.basename(src)) try: if os.path.abspath(src) != os.path.abspath(dest): logging.info("Moviendo vídeo principal: %s -> %s", src, dest) shutil.move(src, dest) final_path = dest except Exception: final_path = src # También mover otros artefactos que empiecen por el basename try: pattern = os.path.join(os.getcwd(), f"{base}*") for p in glob.glob(pattern): # no mover el archivo fuente ya movido if os.path.abspath(p) == os.path.abspath(final_path): continue # mover sólo ficheros regulares try: if os.path.isfile(p): logging.info("Moviendo artefacto: %s -> %s", p, os.path.join(project_out, os.path.basename(p))) shutil.move(p, os.path.join(project_out, os.path.basename(p))) except Exception: pass except Exception: pass else: # En dry-run o sin resultado, no movemos nada final_path = getattr(result, "burned_video", None) # Si el usuario pidió sólo el SRT traducido, mover el srt al output// if args.srt_only and not args.dry_run and result and getattr(result, "srt_translated", None): base = os.path.splitext(os.path.basename(video))[0] project_out = os.path.join(os.getcwd(), "output", base) try: os.makedirs(project_out, exist_ok=True) except Exception: pass src = result.srt_translated dest = os.path.join(project_out, os.path.basename(src)) if src else None try: if src and dest and os.path.abspath(src) != os.path.abspath(dest): logging.info("Moviendo SRT traducido: %s -> %s", src, dest) shutil.move(src, dest) final_path = dest or src except Exception: final_path = src logging.info("Flujo completado. Vídeo final: %s", final_path) finally: if not args.keep_temp: try: logging.info("Eliminando workdir: %s", workdir) shutil.rmtree(workdir) except Exception: pass # Limpiar posibles directorios temporales residuales creados # por ejecuciones previas del pipeline: /tmp/full_pipeline_* try: tmp_pattern = "/tmp/full_pipeline_*" for tmpd in glob.glob(tmp_pattern): try: # no intentar borrar si no es un directorio if not os.path.isdir(tmpd): continue # evitar borrar el workdir actual por si acaso try: if os.path.abspath(tmpd) == os.path.abspath(workdir): continue except Exception: pass logging.info("Eliminando temporal residual: %s", tmpd) shutil.rmtree(tmpd) except Exception: # no fallar la limpieza por errores puntuales pass except Exception: pass if __name__ == "__main__": main()