# -*- coding: utf-8 -*- """ 视频处理模块 包含视频剪辑、合并、淡入淡出等操作 """ import os import json import logging from .constants import FFMPEG_CMD, FFPROBE_CMD, DEFAULT_VIDEO_PARAMS, SUBTITLE_STYLE from .utils import run_cmd, ensure_dir, get_clip_num logger = logging.getLogger(__name__) def get_video_info(video_path): """ 获取视频信息 Args: video_path: 视频路径 Returns: 视频信息字典 """ cmd = f'"{FFPROBE_CMD}" -v error -show_entries format=duration,size -of json "{video_path}"' import subprocess result = subprocess.run(cmd, shell=True, capture_output=True, text=True, encoding='utf-8', errors='ignore') try: import json as json_mod return json_mod.loads(result.stdout) except: return {} def extract_clip(video_src, start, end, output_path, fade_duration=1): """ 从视频提取片段 Args: video_src: 源视频路径 start: 开始时间(秒) end: 结束时间(秒) output_path: 输出路径 fade_duration: 淡入淡出时长 Returns: True if success """ duration = end - start # 使用掐头去尾精确控制 cmd = f'"{FFMPEG_CMD}" -y -ss {start} -i "{video_src}" -t {duration} ' if fade_duration > 0: # 添加淡入淡出 cmd += f'-vf "fade=t=in:st=0:d={fade_duration},fade=t=out:st={duration-fade_duration}:d={fade_duration}" ' cmd += f'-c:v libx264 -crf 20 -c:a aac -y "{output_path}"' success = run_cmd(cmd) if success: logger.info(f"Extracted clip: {output_path}") else: logger.error(f"Failed to extract clip: {output_path}") return success def merge_clips(clip_paths, output_path, inter_dir=None): """ 合并多个视频片段 Args: clip_paths: 片段路径列表 output_path: 输出路径 inter_dir: 中间目录（用于保存concat list） Returns: True if success """ if not clip_paths: logger.error("No clips to merge") return False # 按编号排序 sorted_clips = sorted(clip_paths, key=lambda p: get_clip_num(p)) # 创建concat list if inter_dir: list_path = os.path.join(inter_dir, "concat_list.txt") else: list_path = os.path.join(os.path.dirname(output_path), "concat_list.txt") with open(list_path, 'w', encoding='utf-8') as f: for p in sorted_clips: clip_num = get_clip_num(p) # 验证clip对应的json存在 if inter_dir: json_path = os.path.join(inter_dir, f"clip{clip_num}.json") if not os.path.exists(json_path): logger.warning(f"Skipping clip{clip_num} - no JSON found") continue f.write(f"file '{p}'\n") # 合并 cmd = f'"{FFMPEG_CMD}" -y -f concat -safe 0 -i "{list_path}" -c copy -y "{output_path}"' success = run_cmd(cmd) if success: logger.info(f"Merged {len(sorted_clips)} clips -> {output_path}") else: logger.error(f"Failed to merge clips") return success def burn_subtitles(video_path, srt_path, output_path): """ 烧录字幕到视频 Args: video_path: 输入视频路径 srt_path: 字幕文件路径 output_path: 输出路径 Returns: True if success """ # Windows路径转义 srt_escaped = srt_path.replace('\\', '/').replace('D:/', 'D\\:/') filter_str = f"subtitles='{srt_escaped}':force_style='{SUBTITLE_STYLE}'" cmd = f'"{FFMPEG_CMD}" -y -i "{video_path}" -vf "{filter_str}" -c:a copy -y "{output_path}"' success = run_cmd(cmd) if success: logger.info(f"Burned subtitles -> {output_path}") else: logger.error(f"Failed to burn subtitles") return success def burn_dual_subtitles(video_path, title_srt_path, content_srt_path, output_path, title_fontsize=90, title_color="FFFF00", subtitle_fontsize=24, subtitle_color="FFFFFF"): """ 烧录两层字幕到视频（标题在屏幕正中，正文在下方） Args: video_path: 输入视频路径 title_srt_path: 标题字幕文件路径 content_srt_path: 正文字幕文件路径 output_path: 输出路径 title_fontsize: 标题字号 title_color: 标题颜色(HTML格式如FFFF00) subtitle_fontsize: 正文字号 subtitle_color: 正文颜色 Returns: True if success """ # Windows路径转义 title_escaped = title_srt_path.replace('\\', '/').replace('D:/', 'D\\:/') content_escaped = content_srt_path.replace('\\', '/').replace('D:/', 'D\\:/') # 转换颜色格式：HTML (FFFF00) -> FFmpeg BGR (&H00FFFF) def html_to_bgr(color): if color.startswith('&H') or color.startswith('0x'): return color # HTML format like FFFF00 -> FFmpeg BGR format r = int(color[0:2], 16) g = int(color[2:4], 16) b = int(color[4:6], 16) return f"&H00{b:02X}{g:02X}{r:02X}" title_bgr = html_to_bgr(title_color) subtitle_bgr = html_to_bgr(subtitle_color) # 标题样式：使用SRT+force_style，Alignment=5水平居中，垂直位置由MarginV控制 # 正文字样式：底部居中，24字号，白色，带描边 content_style = f"FontName=微软雅黑,FontSize={subtitle_fontsize},PrimaryColour={subtitle_bgr},Alignment=2,MarginV=20,Outline=1,Shadow=1" # 使用两个独立字幕滤镜分别渲染，然后叠加 # 标题使用Alignment=5,MarginV=0（正中） title_style = f"FontName=微软雅黑,FontSize={title_fontsize},PrimaryColour={title_bgr},Alignment=5,MarginV=0,Outline=3,Shadow=2" # 使用两个字幕滤镜叠加，然后映射视频+原始音频 # 标题使用Alignment=5,MarginV=0（正中） title_style = f"FontName=微软雅黑,FontSize={title_fontsize},PrimaryColour={title_bgr},Alignment=5,MarginV=0,Outline=3,Shadow=2" # 使用两个字幕滤镜叠加 filter_str = f"[0:v]subtitles='{title_escaped}':force_style='{title_style}',subtitles='{content_escaped}':force_style='{content_style}'[out]" # 保留原始音频 - 映射视频输出和原始音频 cmd = f'"{FFMPEG_CMD}" -y -i "{video_path}" -filter_complex "{filter_str}" -map "[out]" -map 0:a? -c:a copy -c:v libx264 -crf 18 -y "{output_path}"' success = run_cmd(cmd) if success: logger.info(f"Burned dual subtitles (title+content) -> {output_path}") else: logger.error(f"Failed to burn dual subtitles") return success def extract_key_frames(video_path, timestamps, output_dir, prefix="frame"): """ 从视频提取关键帧 Args: video_path: 视频路径 timestamps: 时间戳列表 output_dir: 输出目录 prefix: 文件名前缀 Returns: 帧文件路径列表 """ ensure_dir(output_dir) frames = [] for i, ts in enumerate(timestamps): output_path = os.path.join(output_dir, f"{prefix}_{i:03d}.jpg") cmd = f'"{FFMPEG_CMD}" -y -ss {ts} -i "{video_path}" -vframes 1 -q:v 2 "{output_path}"' if run_cmd(cmd) and os.path.exists(output_path): frames.append(output_path) else: logger.warning(f"Failed to extract frame at {ts}s") return frames class VideoPipeline: """视频流水线封装""" def __init__(self, config): self.config = config self.video_params = config.get('video_params', DEFAULT_VIDEO_PARAMS) self.video_src = config.get('video_src') self.output_dir = config.get('output_dir') self.inter_dir = os.path.join(self.output_dir, 'intermediates') ensure_dir(self.output_dir) ensure_dir(self.inter_dir) def extract_clips(self, clips): """ 提取所有视频片段 Args: clips: 片段配置列表 Returns: 片段路径列表 """ clip_paths = [] fade_dur = self.video_params.get('fade_duration', 1) for i, clip in enumerate(clips, 1): clip_path = os.path.join(self.inter_dir, f"clip{i}.mp4") fade_path = os.path.join(self.inter_dir, f"clip{i}_fade.mp4") # 提取片段 success = extract_clip( self.video_src, clip['start'], clip['end'], clip_path, fade_duration=0 # 先不添加淡出 ) if success and fade_dur > 0: # 添加淡入淡出 from .utils import to_srt_time duration = clip['end'] - clip['start'] fade_in_end = fade_dur fade_out_start = max(0, duration - fade_dur) cmd = f'"{FFMPEG_CMD}" -y -i "{clip_path}" ' cmd += f'-vf "fade=t=in:st=0:d={fade_dur},fade=t=out:st={fade_out_start}:d={fade_dur}" ' cmd += f'-c:v libx264 -crf 20 -c:a aac -y "{fade_path}"' if run_cmd(cmd): clip_paths.append(fade_path) else: # 淡出失败，使用原始片段 clip_paths.append(clip_path) else: clip_paths.append(clip_path) return clip_paths def merge(self, clip_paths, output_name="concat_merged.mp4"): """合并片段""" output_path = os.path.join(self.output_dir, output_name) return merge_clips(clip_paths, output_path, self.inter_dir) def burn(self, video_path, srt_path, output_name="final.mp4"): """烧录字幕""" output_path = os.path.join(self.output_dir, output_name) return burn_subtitles(video_path, srt_path, output_path)