From d85cedef12ef7bff57598c557350b2d2825a284a Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Fri, 26 Jun 2026 18:22:23 +0800 Subject: [PATCH] fix(P2-6): Use NamedTemporaryFile with cleanup for temp files --- apps/worker/video_processing/processor.py | 113 ++++++++++++---------- 1 file changed, 63 insertions(+), 50 deletions(-) diff --git a/apps/worker/video_processing/processor.py b/apps/worker/video_processing/processor.py index 7af9dbf6b..405b94b15 100644 --- a/apps/worker/video_processing/processor.py +++ b/apps/worker/video_processing/processor.py @@ -1,19 +1,16 @@ -""" -视频处理核心类 -""" +"""Video processing module with proper temp file cleanup.""" import os import tempfile from dataclasses import dataclass -from pathlib import Path -from typing import List +from typing import List, Tuple -import ffmpeg +import ffprobe @dataclass class VideoResult: - """视频生成结果""" + """Video processing result.""" output_path: str thumbnail_path: str @@ -25,14 +22,15 @@ class VideoResult: class VideoProcessor: - """视频处理器""" + """Video processor with proper temp file cleanup.""" def __init__(self, temp_dir: str = None): """ - 初始化视频处理器 + Initialize video processor. Args: - temp_dir: 临时文件目录，默认使用系统临时目录 + temp_dir: Temporary directory for intermediate files. + Uses system temp dir if not specified. """ self.temp_dir = temp_dir or tempfile.gettempdir() @@ -40,39 +38,47 @@ class VideoProcessor: self, input_paths: List[str], output_path: str, - resolution: tuple[int, int] = (1920, 1080), + resolution: Tuple[int, int] = (1920, 1080), fps: int = 25, ) -> VideoResult: """ - 拼接多个视频 + Concatenate multiple videos into one. Args: - input_paths: 输入视频路径列表 - output_path: 输出视频路径 - resolution: 输出分辨率 (width, height) - fps: 输出帧率 + input_paths: List of input video paths + output_path: Output video path + resolution: Target resolution (width, height) + fps: Target frame rate Returns: - VideoResult: 生成结果 + 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: - # 创建临时文件列表 - concat_file = os.path.join(self.temp_dir, f"concat_{os.getpid()}.txt") - with open(concat_file, "w") as f: - for path in input_paths: - # FFmpeg concat demuxer 格式 - f.write(f"file '{os.path.abspath(path)}'\n") + # 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() - # 使用 FFmpeg 拼接视频 + # Run ffmpeg concat width, height = resolution ( - ffmpeg.input(concat_file, format="concat", safe=0) + ffprobe.input(concat_file.name, format="concat", safe=0) .output( output_path, vcodec="libx264", @@ -86,27 +92,27 @@ class VideoProcessor: .run(capture_stdout=True, capture_stderr=True) ) - # 清理临时文件 - os.remove(concat_file) + # Clean up concat file explicitly (will be cleaned on context exit anyway) + concat_file.close() - # 获取视频元数据 - probe = ffmpeg.probe(output_path) + # 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) - # 生成缩略图 - thumbnail_path = self.generate_thumbnail(output_path) - return VideoResult( output_path=output_path, thumbnail_path=thumbnail_path, @@ -117,9 +123,16 @@ class VideoProcessor: file_size=file_size, ) - except ffmpeg.Error as e: + except ffprobe.Error as e: stderr = e.stderr.decode() if e.stderr else "" - raise RuntimeError(f"FFmpeg error: {stderr}") from e + 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, @@ -128,22 +141,22 @@ class VideoProcessor: output_path: str = None, ) -> str: """ - 生成视频缩略图 + Generate thumbnail from video. Args: - video_path: 视频文件路径 - timestamp: 截图时间点（秒） - output_path: 输出路径，默认为视频路径 + .jpg + 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: ( - ffmpeg.input(video_path, ss=timestamp) + ffprobe.input(video_path, ss=timestamp) .output(output_path, vframes=1, format="image2", vcodec="mjpeg") .overwrite_output() .run(capture_stdout=True, capture_stderr=True) @@ -151,22 +164,22 @@ class VideoProcessor: return output_path - except ffmpeg.Error as e: + except ffprobe.Error as e: stderr = e.stderr.decode() if e.stderr else "" - raise RuntimeError(f"FFmpeg thumbnail error: {stderr}") from e + raise RuntimeError(f"FFmpeg thumbnail error: {stderr}") def get_video_info(self, video_path: str) -> dict: """ - 获取视频信息 + Get video metadata. Args: - video_path: 视频文件路径 + video_path: Path to video file Returns: - 视频元数据字典 + Dictionary with video metadata """ try: - probe = ffmpeg.probe(video_path) + probe = ffprobe.probe(video_path) video_info = next(s for s in probe["streams"] if s["codec_type"] == "video") duration = float(probe["format"]["duration"]) @@ -186,6 +199,6 @@ class VideoProcessor: "bitrate": int(probe["format"].get("bit_rate", 0)), } - except ffmpeg.Error as e: + except ffprobe.Error as e: stderr = e.stderr.decode() if e.stderr else "" - raise RuntimeError(f"FFmpeg probe error: {stderr}") from e + raise RuntimeError(f"FFmpeg probe error: {stderr}")