"""Video processing module with proper temp file cleanup.""" import os import tempfile from dataclasses import dataclass from typing import List, Tuple import ffprobe @dataclass class VideoResult: """Video processing result.""" output_path: str thumbnail_path: str duration: float width: int height: int fps: float file_size: int class VideoProcessor: """Video processor with proper temp file cleanup.""" def __init__(self, temp_dir: str = None): """ Initialize video processor. Args: temp_dir: Temporary directory for intermediate files. Uses system temp dir if not specified. """ self.temp_dir = temp_dir or tempfile.gettempdir() def concatenate_videos( self, input_paths: List[str], output_path: str, resolution: Tuple[int, int] = (1920, 1080), fps: int = 25, ) -> VideoResult: """ Concatenate multiple videos into one. Args: input_paths: List of input video paths output_path: Output video path resolution: Target resolution (width, height) fps: Target frame rate Returns: VideoResult with processing details """ if not input_paths: raise ValueError("input_paths cannot be empty") # Ensure output directory exists os.makedirs(os.path.dirname(output_path), exist_ok=True) # P2-6: Use NamedTemporaryFile with cleanup for concat list file concat_file = None try: # Create concat file in temp directory with auto cleanup concat_file = tempfile.NamedTemporaryFile( mode='w', suffix='.txt', dir=self.temp_dir, delete=True, ) # Write concat list for path in input_paths: concat_file.write(f"file '{os.path.abspath(path)}'\n") concat_file.flush() # Run ffmpeg concat width, height = resolution ( ffprobe.input(concat_file.name, format="concat", safe=0) .output( output_path, vcodec="libx264", acodec="aac", s=f"{width}x{height}", r=fps, preset="medium", crf=23, ) .overwrite_output() .run(capture_stdout=True, capture_stderr=True) ) # Clean up concat file explicitly (will be cleaned on context exit anyway) concat_file.close() # Generate thumbnail thumbnail_path = self.generate_thumbnail(output_path) # Get video info probe = ffprobe.probe(output_path) video_info = next(s for s in probe["streams"] if s["codec_type"] == "video") duration = float(probe["format"]["duration"]) width = int(video_info["width"]) height = int(video_info["height"]) # Get FPS fps_str = video_info.get("r_frame_rate", "25/1") fps_parts = fps_str.split("/") fps_value = float(fps_parts[0]) / float(fps_parts[1]) if len(fps_parts) == 2 else float(fps_parts[0]) file_size = os.path.getsize(output_path) return VideoResult( output_path=output_path, thumbnail_path=thumbnail_path, duration=duration, width=width, height=height, fps=fps_value, file_size=file_size, ) except ffprobe.Error as e: stderr = e.stderr.decode() if e.stderr else "" raise RuntimeError(f"FFmpeg error: {stderr}") finally: # Ensure concat file is cleaned up even on exception if concat_file is not None: try: concat_file.close() except Exception: pass def generate_thumbnail( self, video_path: str, timestamp: float = 1.0, output_path: str = None, ) -> str: """ Generate thumbnail from video. Args: video_path: Path to video file timestamp: Time in seconds to capture frame output_path: Output thumbnail path. Defaults to video_path + '.jpg' Returns: Path to generated thumbnail """ if output_path is None: output_path = f"{os.path.splitext(video_path)[0]}_thumb.jpg" try: ( ffprobe.input(video_path, ss=timestamp) .output(output_path, vframes=1, format="image2", vcodec="mjpeg") .overwrite_output() .run(capture_stdout=True, capture_stderr=True) ) return output_path except ffprobe.Error as e: stderr = e.stderr.decode() if e.stderr else "" raise RuntimeError(f"FFmpeg thumbnail error: {stderr}") def get_video_info(self, video_path: str) -> dict: """ Get video metadata. Args: video_path: Path to video file Returns: Dictionary with video metadata """ try: probe = ffprobe.probe(video_path) video_info = next(s for s in probe["streams"] if s["codec_type"] == "video") duration = float(probe["format"]["duration"]) width = int(video_info["width"]) height = int(video_info["height"]) fps_str = video_info.get("r_frame_rate", "25/1") fps_parts = fps_str.split("/") fps_value = float(fps_parts[0]) / float(fps_parts[1]) if len(fps_parts) == 2 else float(fps_parts[0]) return { "duration": duration, "width": width, "height": height, "fps": fps_value, "codec": video_info.get("codec_name"), "bitrate": int(probe["format"].get("bit_rate", 0)), } except ffprobe.Error as e: stderr = e.stderr.decode() if e.stderr else "" raise RuntimeError(f"FFmpeg probe error: {stderr}")