#!/usr/bin/env python3 """ 处理所有字幕，生成完整视频 流程： 1. 读取subtitles.srt获取所有字幕时间轴 2. 根据字幕时间轴分割input.mp4为所有片段 3. 为所有字幕生成带字幕的静态帧图片 4. 为所有字幕生成带500ms空白音的静态视频 5. 交替拼接原始片段和静态视频生成完整视频 """ import os import subprocess import re import textwrap import tempfile # 导入编码处理模块 from console_helper import ( safe_print, print_header, print_step, print_info, print_success, print_fail, print_done ) def parse_srt_file(subtitle_path): """解析SRT字幕文件，返回每句的字幕信息""" subtitles = [] try: with open(subtitle_path, 'r', encoding='utf-8') as f: content = f.read() # 按空行分割字幕块 blocks = content.strip().split('\n\n') for block in blocks: lines = block.strip().split('\n') if len(lines) >= 3: # 第一行是序号 line_number = int(lines[0].strip()) # 第二行是时间轴 time_line = lines[1].strip() # 获取字幕文本 text = ' '.join(lines[2:]) # 解析时间轴 match = re.match(r'(\d{2}:\d{2}:\d{2}[,.]\d{3})\s*-->\s*(\d{2}:\d{2}:\d{2}[,.]\d{3})', time_line) if match: start_time = parse_srt_time(match.group(1)) end_time = parse_srt_time(match.group(2)) subtitles.append({ 'number': line_number, 'start': start_time, 'end': end_time, 'text': text }) return subtitles except Exception as e: safe_print(f" 解析字幕文件失败: {e}") return [] def parse_srt_time(time_str): """解析SRT时间格式 (00:00:01,129) 转换为秒""" match = re.match(r'(\d{2}):(\d{2}):(\d{2})[,.](\d{3})', time_str) if match: hours, minutes, seconds, milliseconds = match.groups() return int(hours) * 3600 + int(minutes) * 60 + int(seconds) + int(milliseconds) / 1000 return 0 def cut_video_segment(input_video, start_time, end_time, output_path, ffmpeg_path="ffmpeg"): """剪切视频片段""" duration = end_time - start_time args = [ ffmpeg_path, '-y', '-ss', str(start_time), '-i', input_video, '-t', str(duration), '-c:v', 'libx264', '-c:a', 'aac', '-ar', '44100', '-ac', '1', '-avoid_negative_ts', 'make_zero', output_path ] try: subprocess.run(args, check=True, capture_output=True) return True except subprocess.CalledProcessError as e: safe_print(f" 剪切视频失败") return False def extract_frame_at_time(video_path, time_seconds, output_image, ffmpeg_path="ffmpeg"): """从视频中提取指定时间的帧""" args = [ ffmpeg_path, '-y', '-ss', str(time_seconds), '-i', video_path, '-vframes', '1', '-q:v', '2', output_image ] try: subprocess.run(args, check=True, capture_output=True) return True except subprocess.CalledProcessError as e: safe_print(f" 提取帧失败") return False def wrap_text(text, max_chars=35): """将长文本按指定字符数换行""" wrapped = textwrap.fill(text, width=max_chars) return wrapped def escape_text_for_ffmpeg(text): """为FFmpeg的drawtext滤镜转义文本""" escaped = text.replace("'", "'\\''") return escaped def add_subtitle_to_image(input_image, output_image, subtitle_text, ffmpeg_path="ffmpeg"): """为图片添加字幕""" wrapped_text = wrap_text(subtitle_text, max_chars=35) lines = wrapped_text.split('\n') filters = [] y_position = 50 line_height = 60 for i, line in enumerate(lines): safe_line = escape_text_for_ffmpeg(line) current_y = y_position + i * line_height filter_str = ( f"drawtext=text='{safe_line}':" f"fontsize=50:" f"fontcolor=yellow:" f"borderw=3:" f"bordercolor=red:" f"x=(w-text_w)/2:" f"y={current_y}" ) filters.append(filter_str) vf_filter = ','.join(filters) args = [ ffmpeg_path, '-y', '-i', input_image, '-vf', vf_filter, '-frames:v', '1', output_image ] try: subprocess.run(args, check=True, capture_output=True) return True except subprocess.CalledProcessError as e: safe_print(f" 添加字幕失败") return False def get_audio_duration(audio_path): """获取音频时长""" args = [ 'ffprobe', '-v', 'quiet', '-print_format', 'json', '-show_format', audio_path ] try: result = subprocess.run(args, check=True, capture_output=True, text=True) import json data = json.loads(result.stdout) return float(data['format']['duration']) except Exception as e: safe_print(f" 获取音频时长失败: {e}") return 0 def create_silent_audio(duration, output_path, ffmpeg_path="ffmpeg"): """创建静音音频""" args = [ ffmpeg_path, '-y', '-f', 'lavfi', '-i', 'anullsrc=r=44100:cl=mono', '-c:a', 'aac', '-ar', '44100', '-ac', '1', '-t', str(duration), output_path ] try: subprocess.run(args, check=True, capture_output=True) return True except subprocess.CalledProcessError as e: safe_print(f" 创建静音音频失败: {e}") return False def concatenate_audio_files(audio_files, output_path, ffmpeg_path="ffmpeg"): """使用filter_complex拼接多个音频文件""" inputs = [] filters = [] for i, audio_file in enumerate(audio_files): inputs.extend(['-i', audio_file]) filters.append(f"[{i}:a]") filters.append(f"concat=n={len(audio_files)}:v=0:a=1[out]") args = [ ffmpeg_path, '-y', ] + inputs + [ '-filter_complex', ''.join(filters), '-map', '[out]', '-c:a', 'aac', '-ar', '44100', '-ac', '1', output_path ] try: subprocess.run(args, check=True, capture_output=True) return True except subprocess.CalledProcessError as e: safe_print(f" 拼接音频失败: {e}") return False def create_static_video(image_path, audio_path, output_path, silence_duration=0.5, ffmpeg_path="ffmpeg"): """使用图片和音频生成静态视频，音频前加空白音""" temp_dir = tempfile.gettempdir() original_duration = get_audio_duration(audio_path) if original_duration == 0: safe_print(f" 无法获取音频时长") return False # 创建静音音频 silent_audio = os.path.join(temp_dir, f"silent_{os.path.basename(audio_path)}.aac") if not create_silent_audio(silence_duration, silent_audio): return False # 拼接静音音频和原始音频 combined_audio = os.path.join(temp_dir, f"combined_{os.path.basename(audio_path)}.aac") if not concatenate_audio_files([silent_audio, audio_path], combined_audio): try: os.remove(silent_audio) except: pass return False total_duration = silence_duration + original_duration # 生成视频 args = [ ffmpeg_path, '-y', '-loop', '1', '-i', image_path, '-i', combined_audio, '-c:v', 'libx264', '-c:a', 'aac', '-ar', '44100', '-ac', '1', '-t', str(total_duration), '-pix_fmt', 'yuv420p', '-shortest', output_path ] try: subprocess.run(args, check=True, capture_output=True) try: os.remove(silent_audio) os.remove(combined_audio) except: pass return True except subprocess.CalledProcessError as e: safe_print(f" 生成视频失败: {e}") try: os.remove(silent_audio) os.remove(combined_audio) except: pass return False def concatenate_videos(video_list, output_path, ffmpeg_path="ffmpeg"): """拼接多个视频""" list_file = 'concat_list.txt' with open(list_file, 'w', encoding='utf-8') as f: for video in video_list: escaped_path = video.replace('\\', '/') f.write(f"file '{escaped_path}'\n") args = [ ffmpeg_path, '-y', '-f', 'concat', '-safe', '0', '-i', list_file, '-c:v', 'libx264', '-c:a', 'aac', '-ar', '44100', '-ac', '1', '-pix_fmt', 'yuv420p', output_path ] try: subprocess.run(args, check=True, capture_output=True) os.remove(list_file) return True except subprocess.CalledProcessError as e: safe_print(f" 拼接视频失败: {e}") try: os.remove(list_file) except: pass return False def main(): input_video = "input.mp4" subtitle_file = "subtitles.srt" audio_dir = "tts_audio_all" segments_dir = "all_segments" frames_dir = "all_frames" frames_with_subtitle_dir = "all_frames_with_subtitle" static_videos_dir = "all_static_videos" output_video = "final_video.mp4" silence_duration = 0.5 # 500毫秒空白音 print_header("处理所有字幕生成完整视频") safe_print("") # 检查输入文件 if not os.path.exists(input_video): safe_print(f"错误: 输入视频 {input_video} 不存在") return if not os.path.exists(subtitle_file): safe_print(f"错误: 字幕文件 {subtitle_file} 不存在") return if not os.path.exists(audio_dir): safe_print(f"错误: 音频目录 {audio_dir} 不存在") return # 创建输出目录 for dir_path in [segments_dir, frames_dir, frames_with_subtitle_dir, static_videos_dir]: if not os.path.exists(dir_path): os.makedirs(dir_path) # 1. 解析字幕文件 print_step(1, 5, "解析字幕文件...") subtitles = parse_srt_file(subtitle_file) if not subtitles: safe_print("解析字幕失败，退出") return print_info("字幕总数", str(len(subtitles))) safe_print("") # 2. 分割视频片段 print_step(2, 5, "分割视频片段...") segment_files = [] for i, sub in enumerate(subtitles): output_segment = os.path.join(segments_dir, f"segment_{sub['number']:03d}.mp4") safe_print(f" 分割第{sub['number']}段 ({sub['start']:.2f}s - {sub['end']:.2f}s)...") if cut_video_segment(input_video, sub['start'], sub['end'], output_segment): segment_files.append(output_segment) else: safe_print(f" 分割第{sub['number']}段失败") print_info("成功分割", f"{len(segment_files)}/{len(subtitles)} 段") safe_print("") # 3. 提取帧并添加字幕 print_step(3, 5, "提取帧并添加字幕...") frame_files = [] for sub in subtitles: # 提取帧 frame_path = os.path.join(frames_dir, f"frame_{sub['number']:03d}.jpg") safe_print(f" 处理第{sub['number']}帧...") if extract_frame_at_time(input_video, sub['end'], frame_path): # 添加字幕 frame_with_subtitle = os.path.join(frames_with_subtitle_dir, f"frame_{sub['number']:03d}_with_subtitle.jpg") if add_subtitle_to_image(frame_path, frame_with_subtitle, sub['text']): frame_files.append(frame_with_subtitle) else: safe_print(f" 添加第{sub['number']}帧字幕失败") else: safe_print(f" 提取第{sub['number']}帧失败") print_info("成功处理", f"{len(frame_files)}/{len(subtitles)} 帧") safe_print("") # 4. 生成静态视频 print_step(4, 5, "生成静态视频...") static_video_files = [] for sub in subtitles: frame_path = os.path.join(frames_with_subtitle_dir, f"frame_{sub['number']:03d}_with_subtitle.jpg") audio_path = os.path.join(audio_dir, f"tts_line_{sub['number']}.wav") output_path = os.path.join(static_videos_dir, f"static_video_{sub['number']:03d}.mp4") if not os.path.exists(frame_path): safe_print(f" 警告: 帧文件 {frame_path} 不存在，跳过第{sub['number']}句") continue if not os.path.exists(audio_path): safe_print(f" 警告: 音频文件 {audio_path} 不存在，跳过第{sub['number']}句") continue safe_print(f" 生成第{sub['number']}句静态视频...") if create_static_video(frame_path, audio_path, output_path, silence_duration): static_video_files.append(output_path) else: safe_print(f" 生成第{sub['number']}句静态视频失败") print_info("成功生成", f"{len(static_video_files)}/{len(subtitles)} 个静态视频") safe_print("") # 5. 拼接所有视频 print_step(5, 5, "拼接所有视频...") video_list = [] for sub in subtitles: segment = os.path.join(segments_dir, f"segment_{sub['number']:03d}.mp4") static_video = os.path.join(static_videos_dir, f"static_video_{sub['number']:03d}.mp4") if os.path.exists(segment): video_list.append(segment) if os.path.exists(static_video): video_list.append(static_video) if not video_list: safe_print("错误: 没有找到任何视频文件") return print_info("总视频数", str(len(video_list))) safe_print("开始拼接...") if concatenate_videos(video_list, output_video): print_info("成功生成", output_video) else: safe_print("拼接视频失败") return print_done() if __name__ == '__main__': main()