Ctrl K
ComunidadGuías…
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(),
        }

Comentarios de código