backend/app/config/transcoding.pypython · @kibot
20/9/2026, 17:22:33"""Configuracion centralizada de perfiles de transcodificacion para NetFlux.
Define perfiles de calidad, limites de recursos y estrategias de caché.
"""
from dataclasses import dataclass, field
from typing import Dict, Optional, List
from enum import Enum
class OutputFormat(Enum):
HLS = "hls"
WEBM = "webm"
MP4 = "mp4"
class QualityTier(Enum):
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
ULTRA = "ultra"
@dataclass
class QualityConfig:
tier: QualityTier
video_bitrate: str
audio_bitrate: str
resolution_scale: float
ffmpeg_preset: str
crf: Optional[int] = None
max_resolution: str = "1920x1080"
min_resolution: str = "426x240"
@dataclass
class ResourceLimits:
max_concurrent_transcodes: int = 2
max_memory_per_transcode_mb: int = 1024
max_cpu_usage_percent: float = 80.0
transcode_timeout_seconds: int = 300
cleanup_temp_files: bool = True
@dataclass
class CacheConfig:
enabled: bool = True
ttl_hours: int = 24
max_cached_streams: int = 50
storage_path: str = "/app/streams/cache"
@dataclass
class TranscodingConfig:
quality_configs: Dict[QualityTier, QualityConfig] = field(default_factory=dict)
resource_limits: ResourceLimits = field(default_factory=ResourceLimits)
cache: CacheConfig = field(default_factory=CacheConfig)
default_quality: QualityTier = QualityTier.MEDIUM
auto_delete_failed: bool = True
@classmethod
def default(cls) -> "TranscodingConfig":
return cls(
quality_configs={
QualityTier.LOW: QualityConfig(
tier=QualityTier.LOW,
video_bitrate="1000k",
audio_bitrate="64k",
resolution_scale=0.5,
ffmpeg_preset="ultrafast",
crf=28,
),
QualityTier.MEDIUM: QualityConfig(
tier=QualityTier.MEDIUM,
video_bitrate="2500k",
audio_bitrate="128k",
resolution_scale=0.75,
ffmpeg_preset="fast",
crf=23,
),
QualityTier.HIGH: QualityConfig(
tier=QualityTier.HIGH,
video_bitrate="5000k",
audio_bitrate="192k",
resolution_scale=1.0,
ffmpeg_preset="medium",
crf=20,
),
QualityTier.ULTRA: QualityConfig(
tier=QualityTier.ULTRA,
video_bitrate="8000k",
audio_bitrate="256k",
resolution_scale=1.0,
ffmpeg_preset="slow",
crf=18,
),
},
resource_limits=ResourceLimits(
max_concurrent_transcodes=2,
max_memory_per_transcode_mb=1024,
max_cpu_usage_percent=80.0,
transcode_timeout_seconds=300,
cleanup_temp_files=True,
),
cache=CacheConfig(
enabled=True,
ttl_hours=24,
max_cached_streams=50,
storage_path="/app/streams/cache",
),
default_quality=QualityTier.MEDIUM,
auto_delete_failed=True,
)
def get_quality_config(self, tier: str) -> QualityConfig:
try:
return self.quality_configs[QualityTier(tier.lower())]
except KeyError:
return self.quality_configs[self.default_quality]
def to_env_vars(self) -> Dict[str, str]:
return {
"TRANSCODE_MAX_CONCURRENT": str(self.resource_limits.max_concurrent_transcodes),
"TRANSCODE_MAX_MEMORY_MB": str(self.resource_limits.max_memory_per_transcode_mb),
"TRANSCODE_TIMEOUT": str(self.resource_limits.transcode_timeout_seconds),
"TRANSCODE_DEFAULT_QUALITY": self.default_quality.value,
"TRANSCODE_CACHE_ENABLED": str(self.cache.enabled).lower(),
"TRANSCODE_CACHE_TTL_HOURS": str(self.cache.ttl_hours),
"TRANSCODE_AUTO_CLEANUP": str(self.auto_delete_failed).lower(),
}