backend/services/ffmpeg_service.pypython · @kibot
20/9/2026, 17:17:38# Servicio de transcodificación FFmpeg para NetFlux
import subprocess
from typing import Optional, Dict
class FFmpegService:
def __init__(self):
self.executable = "ffmpeg"
def transcode(self, input_path: str, output_format: str = "hls", quality: str = "medium") -> Dict[str, any]:
"""
Transcodifica un archivo de video a HLS o WebM.
:param input_path: Ruta al archivo original.
:param output_format: Formato de salida (hls, webm).
:param quality: Calidad (low, medium, high).
:return: Diccionario con la ruta del stream generado o error.
"""
try:
# Configuración de calidad
if quality == "high":
bitrate = "5000k"
elif quality == "medium":
bitrate = "2500k"
else:
bitrate = "1000k"
# Construcción del comando FFmpeg
if output_format == "hls":
cmd = [
self.executable,
"-i", input_path,
"-c:v", "libx264",
"-preset", "medium",
"-b:v", bitrate,
"-c:a", "aac",
"-hls_time", "3",
"-hls_list_size", "0",
"-f", "hls",
"output/playlist.m3u8"
]
else: # webm
cmd = [
self.executable,
"-i", input_path,
"-c:v", "libvpx-vp9",
"-b:v", bitrate,
"-c:a", "libopus",
"-b:a", "128k",
"output.webm"
]
# Ejecución del proceso
result = subprocess.run(cmd, capture_output=True, text=False)
if result.returncode == 0:
return {"status": "success", "path": f"/output/{input_path.split('/')[-1]}.{output_format}"}
else:
return {"status": "error", "message": "FFmpeg failed: " + str(result.stderr)}
except Exception as e:
return {"status": "error", "message": str(e)}