import io import os import subprocess import tempfile from typing import Tuple _HEIC_EXTS = {".heic", ".heif"} _VIDEO_EXTS = {".mov", ".avi", ".m4v"} def to_jpeg_if_heic(data: bytes, filename: str) -> Tuple[bytes, str]: """Convert HEIC/HEIF to JPEG; return (data, ext) unchanged for all other types.""" ext = os.path.splitext(filename or "")[1].lower() if ext not in _HEIC_EXTS: return data, ext or ".jpg" try: import pillow_heif pillow_heif.register_heif_opener() from PIL import Image, ImageOps img = Image.open(io.BytesIO(data)) img = ImageOps.exif_transpose(img) img = img.convert("RGB") buf = io.BytesIO() img.save(buf, format="JPEG", quality=90) return buf.getvalue(), ".jpg" except Exception: return data, ext def to_mp4_if_needed(data: bytes, filename: str) -> Tuple[bytes, str]: """Convert MOV/AVI/M4V to H.264 MP4; return unchanged for MP4/WebM.""" ext = os.path.splitext(filename or "")[1].lower() if ext not in _VIDEO_EXTS: return data, ext or ".mp4" src_path = dst_path = None try: with tempfile.NamedTemporaryFile(suffix=ext, delete=False) as src: src.write(data) src_path = src.name dst_path = src_path[: -len(ext)] + ".mp4" result = subprocess.run( ["ffmpeg", "-i", src_path, "-c:v", "libx264", "-preset", "fast", "-crf", "23", "-c:a", "aac", "-movflags", "+faststart", "-y", dst_path], capture_output=True, timeout=300, ) if result.returncode == 0: with open(dst_path, "rb") as f: return f.read(), ".mp4" return data, ext except Exception: return data, ext finally: for p in [src_path, dst_path]: if p: try: os.unlink(p) except OSError: pass def convert_media(data: bytes, filename: str) -> Tuple[bytes, str]: """Convert HEIC→JPEG and MOV/AVI/M4V→MP4; pass everything else through.""" ext = os.path.splitext(filename or "")[1].lower() if ext in _HEIC_EXTS: return to_jpeg_if_heic(data, filename) if ext in _VIDEO_EXTS: return to_mp4_if_needed(data, filename) return data, ext or ".bin" def extract_video_thumb(video_path: str) -> str | None: """Extract a frame at ~1s from video_path and save as _thumb.jpg. Returns the thumb path on success, None on failure.""" base, _ = os.path.splitext(video_path) thumb_path = base + "_thumb.jpg" try: result = subprocess.run( ["ffmpeg", "-i", video_path, "-ss", "1", "-vframes", "1", "-q:v", "3", "-vf", "scale=480:-1", "-y", thumb_path], capture_output=True, timeout=30, ) if result.returncode == 0 and os.path.exists(thumb_path): return thumb_path # Fallback: first frame (for videos shorter than 1s) result2 = subprocess.run( ["ffmpeg", "-i", video_path, "-vframes", "1", "-q:v", "3", "-vf", "scale=480:-1", "-y", thumb_path], capture_output=True, timeout=30, ) return thumb_path if result2.returncode == 0 and os.path.exists(thumb_path) else None except Exception: return None