import logging import os import shutil import subprocess import tempfile from typing import List, Tuple import cv2 import numpy as np def extract_frames(video_path: str) -> Tuple[List[np.ndarray], float, int, int]: cap = cv2.VideoCapture(video_path) if not cap.isOpened(): raise ValueError("Unable to open video.") fps = cap.get(cv2.CAP_PROP_FPS) or 0.0 width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) frames: List[np.ndarray] = [] success, frame = cap.read() while success: frames.append(frame) success, frame = cap.read() cap.release() if not frames: raise ValueError("Video decode produced zero frames.") return frames, fps, width, height def _transcode_with_ffmpeg(src_path: str, dst_path: str) -> None: cmd = [ "ffmpeg", "-y", "-i", src_path, "-c:v", "libx264", "-preset", "veryfast", "-pix_fmt", "yuv420p", "-movflags", "+faststart", dst_path, ] process = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False) if process.returncode != 0: raise RuntimeError(process.stderr.decode("utf-8", errors="ignore")) def write_video(frames: List[np.ndarray], output_path: str, fps: float, width: int, height: int) -> None: if not frames: raise ValueError("No frames available for writing.") temp_fd, temp_path = tempfile.mkstemp(prefix="raw_", suffix=".mp4") os.close(temp_fd) writer = cv2.VideoWriter(temp_path, cv2.VideoWriter_fourcc(*"mp4v"), fps or 1.0, (width, height)) if not writer.isOpened(): os.remove(temp_path) raise ValueError("Failed to open VideoWriter.") for frame in frames: writer.write(frame) writer.release() try: _transcode_with_ffmpeg(temp_path, output_path) logging.debug("Transcoded video to H.264 for browser compatibility.") os.remove(temp_path) except FileNotFoundError: logging.warning("ffmpeg not found; serving fallback MP4V output.") shutil.move(temp_path, output_path) except RuntimeError as exc: logging.warning("ffmpeg transcode failed (%s); serving fallback MP4V output.", exc) shutil.move(temp_path, output_path)