fix(P2-6): use NamedTemporaryFile for automatic temp file cleanup
Deploy / Deploy Staging (push) Failing after 2s
Deploy / Build Production Runtime Images (push) Has been skipped
Deploy / Deploy Production (push) Has been skipped
Deploy / Production Browser E2E (push) Has been skipped
CI/CD Pipeline / Validate Code Quality And Tests (push) Has been cancelled
CI/CD Pipeline / Validate Code Quality And Tests (pull_request) Failing after 1m0s
Tests / test (pull_request) Failing after 1m0s
Tests / lint (pull_request) Failing after 1m0s

This commit is contained in:
2026-06-26 17:54:42 +08:00
parent 4c3dcfbf0f
commit 4fefed332a
+34 -31
View File
@@ -1,5 +1,5 @@
"""
视频处理核心类
视频处理器模块
"""
import os
@@ -13,7 +13,7 @@ import ffmpeg
@dataclass
class VideoResult:
"""视频生成结果"""
"""视频结果"""
output_path: str
thumbnail_path: str
@@ -29,10 +29,10 @@ class VideoProcessor:
def __init__(self, temp_dir: str = None):
"""
初始化视频处理器
初始化处理器
Args:
temp_dir: 临时文件目录,默认使用系统临时目录
temp_dir: 临时目录路径,默认系统临时目录
"""
self.temp_dir = temp_dir or tempfile.gettempdir()
@@ -44,35 +44,41 @@ class VideoProcessor:
fps: int = 25,
) -> VideoResult:
"""
拼接多个视频
合并多个视频
Args:
input_paths: 输入视频路径列表
output_path: 输出视频路径
resolution: 输出分辨率 (width, height)
fps: 输出帧率
resolution: 分辨率 (width, height)
fps: 帧率
Returns:
VideoResult: 生成结果
VideoResult: 视频结果
"""
if not input_paths:
raise ValueError("input_paths cannot be empty")
# 确保输出目录存在
# 创建输出目录
os.makedirs(os.path.dirname(output_path), exist_ok=True)
try:
# 创建临时文件列表
concat_file = os.path.join(self.temp_dir, f"concat_{os.getpid()}.txt")
with open(concat_file, "w") as f:
# P2-6: Use NamedTemporaryFile for automatic cleanup
concat_file_fd = tempfile.NamedTemporaryFile(
mode='w',
suffix='.txt',
delete=True,
dir=self.temp_dir
)
with concat_file_fd as concat_file:
for path in input_paths:
# FFmpeg concat demuxer 格式
f.write(f"file '{os.path.abspath(path)}'\n")
# FFmpeg concat demuxer 要求绝对路径
concat_file.write(f"file '{os.path.abspath(path)}'\n")
# Flush is automatic with context manager
# 使用 FFmpeg 拼接视频
# 使用 FFmpeg 合并
width, height = resolution
(
ffmpeg.input(concat_file, format="concat", safe=0)
ffmpeg.input(concat_file.name, format="concat", safe=0)
.output(
output_path,
vcodec="libx264",
@@ -86,19 +92,16 @@ class VideoProcessor:
.run(capture_stdout=True, capture_stderr=True)
)
# 清理临时文件
os.remove(concat_file)
# 获取视频元数据
# 获取视频信息
probe = ffmpeg.probe(output_path)
video_info = next(s for s in probe["streams"] if s["codec_type"] == "video")
video_stream = 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"])
width = int(video_stream["width"])
height = int(video_stream["height"])
# 计算帧率
fps_str = video_info.get("r_frame_rate", "25/1")
# 生成帧率
fps_str = video_stream.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])
@@ -131,15 +134,15 @@ class VideoProcessor:
生成视频缩略图
Args:
video_path: 视频文件路径
timestamp: 截图时间点(秒)
output_path: 输出路径,默认视频路径 + .jpg
video_path: 视频路径
timestamp: 截图时间点
output_path: 输出路径,默认视频同目录下生成 thumbnail.jpg
Returns:
缩略图路径
"""
if output_path is None:
output_path = f"{os.path.splitext(video_path)[0]}_thumb.jpg"
output_path = f"{os.path.splitext(video_path)[0]}_thumbnail.jpg"
try:
(
@@ -160,10 +163,10 @@ class VideoProcessor:
获取视频信息
Args:
video_path: 视频文件路径
video_path: 视频路径
Returns:
视频元数据字典
视频信息字典
"""
try:
probe = ffmpeg.probe(video_path)