fix(P2-6): Use NamedTemporaryFile with cleanup for temp files

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}")