app/services/validator.py

from __future__ import annotations

import mimetypes
import re
from pathlib import Path
from typing import Any

from app.core.config import Settings
from app.core.exceptions import InputError

MEDIA_EXTENSIONS = {
    ".3gp",
    ".aac",
    ".aiff",
    ".apng",
    ".avi",
    ".bin",
    ".bmp",
    ".flac",
    ".gif",
    ".heic",
    ".jpeg",
    ".jpg",
    ".m4a",
    ".m4v",
    ".mkv",
    ".m3u",
    ".m3u8",
    ".mov",
    ".mp3",
    ".mp4",
    ".mpeg",
    ".mpg",
    ".oga",
    ".ogg",
    ".opus",
    ".png",
    ".svg",
    ".tif",
    ".tiff",
    ".ts",
    ".vtt",
    ".wav",
    ".webm",
    ".webp",
    ".wmv",
    ".srt",
    ".ass",
    ".ssa",
}


class MediaValidator:
    def __init__(self, settings: Settings) -> None:
        self.settings = settings

    def safe_filename(self, filename: str | None, fallback: str = "input.bin") -> str:
        name = Path(filename or fallback).name.replace("\x00", "")
        if not name or name in {".", ".."}:
            raise InputError("Invalid filename")
        return name[:240]

    def validate_declared(self, filename: str, mime_type: str, size: int) -> None:
        if size <= 0:
            raise InputError("The uploaded media is empty")
        if size > self.settings.max_upload_size:
            raise InputError(
                "The media exceeds MAX_UPLOAD_SIZE",
                details={"size": size, "maximum": self.settings.max_upload_size},
            )
        suffix = Path(filename).suffix.lower()
        if suffix and suffix not in MEDIA_EXTENSIONS:
            raise InputError("Unsupported media file extension", details={"extension": suffix})
        normalized_mime = mime_type.split(";", 1)[0].lower()
        allowed = normalized_mime.startswith(
            ("audio/", "video/", "image/", "text/")
        ) or normalized_mime in {
            "application/octet-stream",
            "application/x-subrip",
            "application/vnd.apple.mpegurl",
            "application/x-mpegurl",
        }
        if normalized_mime and not allowed:
            raise InputError("Unsupported media MIME type", details={"mime_type": mime_type})

    def validate_probe(self, metadata: dict[str, Any]) -> None:
        streams = [
            *(metadata.get("video_streams") or []),
            *(metadata.get("audio_streams") or []),
        ]
        if not streams:
            raise InputError("The input has no decodable video or audio stream")
        for stream in streams:
            codec = stream.get("codec")
            if not isinstance(codec, str) or not re.fullmatch(r"[a-zA-Z0-9_.-]{1,64}", codec):
                raise InputError("The input contains an invalid or unsupported codec")
        duration = metadata.get("duration")
        if duration and float(duration) > self.settings.max_duration_seconds:
            raise InputError(
                "Media duration exceeds the configured limit",
                details={"duration": duration, "maximum": self.settings.max_duration_seconds},
            )
        resolution = metadata.get("resolution") or {}
        width, height = resolution.get("width", 0), resolution.get("height", 0)
        if width and height and width * height > self.settings.max_resolution_pixels:
            raise InputError(
                "Media resolution exceeds the configured limit",
                details={"width": width, "height": height},
            )

    @staticmethod
    def infer_mime(path: Path, fallback: str = "application/octet-stream") -> str:
        return mimetypes.guess_type(path.name)[0] or fallback


def as_bool(value: Any, default: bool = False) -> bool:
    if value is None:
        return default
    if isinstance(value, bool):
        return value
    return str(value).strip().lower() in {"1", "true", "yes", "on"}


def bounded_number(
    params: dict[str, Any], name: str, default: float, minimum: float, maximum: float
) -> float:
    try:
        value = float(params.get(name, default))
    except (TypeError, ValueError) as exc:
        raise InputError(f"'{name}' must be a number") from exc
    if not minimum <= value <= maximum:
        raise InputError(f"'{name}' must be between {minimum} and {maximum}")
    return value


def positive_int(params: dict[str, Any], name: str, default: int) -> int:
    try:
        value = int(params.get(name, default))
    except (TypeError, ValueError) as exc:
        raise InputError(f"'{name}' must be an integer") from exc
    if value <= 0:
        raise InputError(f"'{name}' must be greater than zero")
    return value