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], }, }