app/services/media_validation.py

from __future__ import annotations

import json
import mimetypes
import os
import struct
import subprocess
from pathlib import Path
from typing import Any


class MediaValidationError(ValueError):
    pass


ALLOWED_IMAGE_SUFFIXES = {".jpg", ".jpeg", ".png", ".webp"}
ALLOWED_REEL_SUFFIXES = {".mp4", ".mov"}
MAX_IMAGE_BYTES = 25 * 1024 * 1024
MAX_REEL_BYTES = 1024 * 1024 * 1024
MIN_REEL_DURATION_SECONDS = 3.0
MAX_REEL_DURATION_SECONDS = 90.0
MIN_REEL_WIDTH = 540
MIN_REEL_HEIGHT = 960
REEL_ASPECT_RATIO_MIN = 0.45
REEL_ASPECT_RATIO_MAX = 0.75


def _assert_local_file(path: str) -> Path:
    if not path or path.startswith(("http://", "https://")):
        raise MediaValidationError("media_path must be an absolute local file path")
    p = Path(path)
    if not p.is_absolute():
        raise MediaValidationError("media_path must be absolute")
    try:
        resolved = p.resolve(strict=True)
    except FileNotFoundError as exc:
        raise MediaValidationError(f"media_path does not exist: {path}") from exc
    if not resolved.is_file():
        raise MediaValidationError(f"media_path is not a file: {path}")
    return resolved


def _png_dimensions(path: Path) -> tuple[int, int] | None:
    with path.open("rb") as fh:
        header = fh.read(24)
    if header[:8] != b"\x89PNG\r\n\x1a\n":
        return None
    return struct.unpack(">II", header[16:24])


def _jpeg_dimensions(path: Path) -> tuple[int, int] | None:
    with path.open("rb") as fh:
        if fh.read(2) != b"\xff\xd8":
            return None
        while True:
            marker_start = fh.read(1)
            if not marker_start:
                return None
            if marker_start != b"\xff":
                continue
            marker = fh.read(1)
            while marker == b"\xff":
                marker = fh.read(1)
            if marker in {b"\xc0", b"\xc1", b"\xc2", b"\xc3"}:
                fh.read(3)
                height, width = struct.unpack(">HH", fh.read(4))
                return width, height
            length_bytes = fh.read(2)
            if len(length_bytes) != 2:
                return None
            length = struct.unpack(">H", length_bytes)[0]
            fh.seek(length - 2, os.SEEK_CUR)


def _image_dimensions(path: Path) -> tuple[int, int] | None:
    return _png_dimensions(path) or _jpeg_dimensions(path)


def validate_image_attachment(attachment: dict[str, Any]) -> dict[str, Any]:
    path = _assert_local_file(str(attachment.get("media_path") or ""))
    suffix = path.suffix.lower()
    if suffix not in ALLOWED_IMAGE_SUFFIXES:
        raise MediaValidationError(f"unsupported image type: {suffix}")
    size = path.stat().st_size
    if size <= 0:
        raise MediaValidationError("image file is empty")
    if size > MAX_IMAGE_BYTES:
        raise MediaValidationError(f"image file is too large: {size} bytes")
    mime_type = attachment.get("mime_type") or mimetypes.guess_type(path.name)[0] or "application/octet-stream"
    if not mime_type.startswith("image/"):
        raise MediaValidationError(f"unsupported image MIME type: {mime_type}")
    dims = _image_dimensions(path)
    return {
        **attachment,
        "media_path": str(path),
        "media_type": "image",
        "mime_type": mime_type,
        "size_bytes": size,
        "width": dims[0] if dims else attachment.get("width"),
        "height": dims[1] if dims else attachment.get("height"),
    }


def validate_page_post_media(attachments: list[dict[str, Any]], media_required: bool) -> list[dict[str, Any]]:
    if media_required and not attachments:
        raise MediaValidationError("media_required=true but no media_attachments were provided")
    validated: list[dict[str, Any]] = []
    for attachment in attachments:
        media_type = (attachment.get("media_type") or "image").lower()
        if media_type != "image":
            raise MediaValidationError("Page post media lane currently supports image attachments only")
        validated.append(validate_image_attachment(attachment))
    return validated


def validate_reel_asset(video_path: str) -> dict[str, Any]:
    path = _assert_local_file(video_path)
    suffix = path.suffix.lower()
    if suffix not in ALLOWED_REEL_SUFFIXES:
        raise MediaValidationError(f"unsupported Reels video type: {suffix}")
    size = path.stat().st_size
    if size <= 0:
        raise MediaValidationError("Reels video file is empty")
    if size > MAX_REEL_BYTES:
        raise MediaValidationError(f"Reels video is too large: {size} bytes")
    try:
        proc = subprocess.run(
            [
                "ffprobe",
                "-v",
                "error",
                "-select_streams",
                "v:0",
                "-show_entries",
                "stream=codec_name,width,height,duration:format=duration,format_name",
                "-of",
                "json",
                str(path),
            ],
            text=True,
            capture_output=True,
            timeout=30,
            check=False,
        )
    except FileNotFoundError as exc:
        raise MediaValidationError("ffprobe is required for Reels validation but is not installed") from exc
    except subprocess.TimeoutExpired as exc:
        raise MediaValidationError("ffprobe timed out during Reels validation") from exc
    if proc.returncode != 0:
        raise MediaValidationError(f"ffprobe could not read Reels video: {proc.stderr[:240]}")
    body = json.loads(proc.stdout or "{}")
    streams = body.get("streams") or []
    if not streams:
        raise MediaValidationError("Reels video has no video stream")
    stream = streams[0]
    width = int(stream.get("width") or 0)
    height = int(stream.get("height") or 0)
    duration = float(stream.get("duration") or body.get("format", {}).get("duration") or 0)
    codec = stream.get("codec_name")
    if codec not in {"h264", "hevc", "h265"}:
        raise MediaValidationError(f"unsupported Reels video codec: {codec}")
    if duration < MIN_REEL_DURATION_SECONDS or duration > MAX_REEL_DURATION_SECONDS:
        raise MediaValidationError(f"Reels duration must be {MIN_REEL_DURATION_SECONDS:g}-{MAX_REEL_DURATION_SECONDS:g} seconds")
    if width < MIN_REEL_WIDTH or height < MIN_REEL_HEIGHT:
        raise MediaValidationError(f"Reels resolution must be at least {MIN_REEL_WIDTH}x{MIN_REEL_HEIGHT}")
    aspect = width / height if height else 0
    if aspect < REEL_ASPECT_RATIO_MIN or aspect > REEL_ASPECT_RATIO_MAX:
        raise MediaValidationError("Reels video must be vertical 9:16-ish aspect ratio")
    return {
        "media_path": str(path),
        "media_type": "reel",
        "mime_type": mimetypes.guess_type(path.name)[0] or "video/mp4",
        "size_bytes": size,
        "width": width,
        "height": height,
        "duration_seconds": duration,
        "codec": codec,
        "requirements": {
            "suffixes": sorted(ALLOWED_REEL_SUFFIXES),
            "duration_seconds": [MIN_REEL_DURATION_SECONDS, MAX_REEL_DURATION_SECONDS],
            "min_resolution": [MIN_REEL_WIDTH, MIN_REEL_HEIGHT],
            "aspect_ratio": [REEL_ASPECT_RATIO_MIN, REEL_ASPECT_RATIO_MAX],
        },
    }