""" make_video.py - 场景图 + ASR 时间戳 + 音频 → 视频 流程： 1. 读取 voice.mp3 获取总时长 2. 读取 scene_plan.json 获取场景列表 3. 如果有 result.json（ASR 对齐），按对齐结果分配时间；否则平均分配 4. 用 moviepy 将场景图在对应时段显示，配上音频生成视频 """ import os import json import math import re import numpy as np from PIL import Image, ImageDraw, ImageFont from moviepy import ( AudioFileClip, ImageClip, CompositeVideoClip, ColorClip, concatenate_videoclips ) def load_scene_plan(workspace: str) -> dict: """读取 scene_plan.json""" path = os.path.join(workspace, 'scene_plan.json') with open(path, 'r', encoding='utf-8') as f: return json.load(f) def load_asr_result(workspace: str) -> dict: """读取 result.json（ASR 对齐结果），如果不存在返回 None""" path = os.path.join(workspace, "result.json") if os.path.exists(path): with open(path, 'r', encoding='utf-8') as f: return json.load(f) return None def build_scene_timeline(scene_plan: dict, asr_result: dict, audio_duration: float, workspace: str = None) -> list: """ 构建场景时间线。优先使用 scene 中已有的 start_time/end_time（ASR 步骤写入）， 没有则根据 ASR 文本匹配分配，再没有则平均分配。 """ scenes = scene_plan["scenes"] num_scenes = len(scenes) # --- 策略 1：直接使用 scene 已有的时间戳（ASR 步骤写入的） --- if all("start_time" in s and "end_time" in s for s in scenes): timeline = [] for scene in scenes: timeline.append({ "scene_id": scene["scene_id"], "start": scene["start_time"], "end": scene["end_time"], "has_match": True, }) # 确保最后一个场景延伸到音频结尾 if timeline: timeline[-1]["end"] = audio_duration return timeline # --- 策略 2：部分有时间戳，部分没有 --- has_timestamp = [s for s in scenes if "start_time" in s and "end_time" in s] if has_timestamp and len(has_timestamp) < num_scenes: timeline = [] matched_end = 0.0 for scene in scenes: if "start_time" in scene and "end_time" in scene: timeline.append({ "scene_id": scene["scene_id"], "start": scene["start_time"], "end": scene["end_time"], }) matched_end = max(matched_end, scene["end_time"]) else: timeline.append({ "scene_id": scene["scene_id"], "start": None, "end": None, }) # 未匹配的场景按剩余时间平均分配 unmatched_indices = [i for i, item in enumerate(timeline) if item["start"] is None] if unmatched_indices: remaining = audio_duration - matched_end if remaining > 0: seg_dur = remaining / len(unmatched_indices) t = matched_end for idx in unmatched_indices: timeline[idx]["start"] = t timeline[idx]["end"] = t + seg_dur t += seg_dur else: for idx in unmatched_indices: timeline[idx]["start"] = matched_end timeline[idx]["end"] = matched_end + 2.0 if timeline: timeline[-1]["end"] = audio_duration return timeline # --- 策略 3：没有任何时间戳，用 ASR 文本匹配或平均分配 --- if asr_result is not None: # 准备匹配用的文本 scenes_for_match = [] for scene in scenes: s = dict(scene) text_val = s.get("text", "") lines_val = s.get("lines", "") raw = text_val or lines_val or "" if isinstance(raw, str) and raw: s["lines"] = [x.strip() for x in raw.split("。") if x.strip()] elif isinstance(raw, list): s["lines"] = raw else: s["lines"] = [] scenes_for_match.append(s) asr_segments = asr_result.get("segments", []) timeline = assign_scenes_to_segments(scenes_for_match, asr_segments, audio_duration) else: # 平均分配 duration_per_scene = audio_duration / num_scenes timeline = [] current_time = 0.0 for scene in scenes: start = current_time end = min(current_time + duration_per_scene, audio_duration) timeline.append({ "scene_id": scene["scene_id"], "start": start, "end": end, }) current_time = end return timeline def _normalize_text(text: str) -> str: """清理文本：去除标点、空格、转小写，用于相似度比较""" if not text: return "" # 去除所有非字母数字和非中文字符 text = re.sub(r'[^\w\u4e00-\u9fff]', '', text) return text.lower() def _longest_common_substring_ratio(s1: str, s2: str) -> float: """ 计算两个字符串的最长公共子串比例 使用动态规划，但为了性能优化： - 如果字符串太长，使用简化的滑动窗口方法 """ if not s1 or not s2: return 0.0 # 对于短字符串，使用精确的 DP 算法 if len(s1) * len(s2) < 100000: # 乘积小于 10万 m, n = len(s1), len(s2) # 优化空间复杂度：只用两行 prev = [0] * (n + 1) curr = [0] * (n + 1) max_len = 0 for i in range(1, m + 1): for j in range(1, n + 1): if s1[i-1] == s2[j-1]: curr[j] = prev[j-1] + 1 max_len = max(max_len, curr[j]) else: curr[j] = 0 prev, curr = curr, [0] * (n + 1) return max_len / max(len(s1), len(s2)) # 对于长字符串，使用简化的滑动窗口 return _sliding_window_similarity(s1, s2) def _sliding_window_similarity(s1: str, s2: str, window_size: int = 20) -> float: """ 滑动窗口相似度（用于长字符串的快速估算） 将 s1 分成多个窗口，在 s2 中查找最佳匹配 """ if len(s1) <= window_size: # 如果 s1 本身很短，直接检查是否在 s2 中 if s1 in s2: return 1.0 # 否则检查每个字符的出现 common = sum(1 for c in s1 if c in s2) return common / len(s1) if s1 else 0.0 # 分窗口检查 total_score = 0.0 num_windows = 0 for i in range(0, len(s1) - window_size + 1, window_size // 2): window = s1[i:i+window_size] if window in s2: total_score += 1.0 else: # 部分匹配 common = sum(1 for c in window if c in s2) total_score += common / window_size num_windows += 1 return total_score / num_windows if num_windows > 0 else 0.0 def assign_scenes_to_segments(scenes: list, asr_segments: list, audio_duration: float) -> list: """ 基于文本相似度的精确匹配 策略： 1. 每个 scene 有 text 字段（原文片段） 2. 在 ASR segments 中搜索这段文本的出现位置 3. 使用最长公共子串相似度容忍 ASR 识别错误 4. 阈值 60%，避免误匹配 """ SIMILARITY_THRESHOLD = 0.5 # 相似度阈值 MAX_WINDOW_SIZE = 20 # 最多合并 20 个 ASR segments（长句需要更多） # 预处理：清理 ASR segments 的文本 asr_cleaned = [] for seg in asr_segments: text = seg.get('text', '').strip() if text: asr_cleaned.append({ 'text': text, 'text_normalized': _normalize_text(text), 'start': seg['start'], 'end': seg['end'] }) def find_best_match(scene_text: str, start_seg_idx: int): """ 找到 scene_text 在 ASR segments 中的最佳匹配位置 使用滑动窗口尝试不同长度的 segment 组合 """ if not scene_text or not scene_text.strip(): return None scene_normalized = _normalize_text(scene_text) best_score = 0.0 best_match = None # 滑动窗口：尝试不同数量的 segment 组合 for window_size in range(1, min(MAX_WINDOW_SIZE + 1, len(asr_cleaned) - start_seg_idx + 1)): for i in range(start_seg_idx, len(asr_cleaned) - window_size + 1): # 合并多个 segments 的文本 combined_text = ''.join( seg['text_normalized'] for seg in asr_cleaned[i:i+window_size] ) # 计算相似度 similarity = _longest_common_substring_ratio(scene_normalized, combined_text) if similarity > best_score: best_score = similarity best_match = { 'start': asr_cleaned[i]['start'], 'end': asr_cleaned[i + window_size - 1]['end'], 'start_idx': i, 'end_idx': i + window_size - 1, 'similarity': similarity } # 如果已经达到很高相似度，提前返回 if similarity >= 0.90: return best_match # 只有达到阈值才算匹配成功 if best_score >= SIMILARITY_THRESHOLD: return best_match return None # 主匹配循环 timeline_raw = [] last_end_idx = 0 # 记录上一个匹配结束的位置 for scene in scenes: # 优先使用 text 字段，兼容旧 lines 字段 scene_text = scene.get('text', '') or scene.get('lines', '') if isinstance(scene_text, list): scene_text = ''.join(scene_text) match = find_best_match(scene_text, last_end_idx) if match: timeline_raw.append({ 'scene_id': scene['scene_id'], 'start': match['start'], 'end': match['end'], 'has_match': True, 'similarity': match['similarity'] }) # 下一个场景从当前匹配结束后开始 last_end_idx = match['end_idx'] + 1 else: timeline_raw.append({ 'scene_id': scene['scene_id'], 'start': None, 'end': None, 'has_match': False, 'similarity': 0.0 }) # 处理未匹配的场景：按剩余音频时间平均分配 unmatched_indices = [i for i, item in enumerate(timeline_raw) if not item['has_match']] if unmatched_indices: # 找到已匹配场景的最大结束时间 matched_end = 0.0 for item in timeline_raw: if item['has_match'] and item['end']: matched_end = max(matched_end, item['end']) remaining_duration = audio_duration - matched_end if remaining_duration > 0 and len(unmatched_indices) > 0: seg_dur = remaining_duration / len(unmatched_indices) current_t = matched_end for idx in unmatched_indices: timeline_raw[idx]['start'] = current_t timeline_raw[idx]['end'] = current_t + seg_dur timeline_raw[idx]['has_match'] = True current_t += seg_dur else: # 如果没有剩余时间，给一个默认时长 for idx in unmatched_indices: timeline_raw[idx]['start'] = matched_end timeline_raw[idx]['end'] = matched_end + 2.0 timeline_raw[idx]['has_match'] = True # 确保每个场景至少有 0.5 秒 for item in timeline_raw: if item['end'] is None or item['end'] <= item['start']: item['end'] = item['start'] + 0.5 # 消除间隙：确保时间连续 for i in range(len(timeline_raw) - 1): curr = timeline_raw[i] nxt = timeline_raw[i + 1] gap = nxt['start'] - curr['end'] if gap > 0.01: # 允许 10ms 的微小间隙 curr['end'] = nxt['start'] # 最后一个场景延伸到音频结尾 if timeline_raw: timeline_raw[-1]['end'] = audio_duration # 合并极短的场景（< 1秒）到下一个场景 merged = [] skip_next = False for i in range(len(timeline_raw)): if skip_next: skip_next = False continue item = dict(timeline_raw[i]) duration = item['end'] - item['start'] # 如果当前场景太短且不是最后一个，合并到下一个 if duration < 1.0 and i < len(timeline_raw) - 1: item['end'] = timeline_raw[i + 1]['end'] merged.append(item) skip_next = True else: merged.append(item) return merged def find_scene_image(workspace: str, scene_id: int) -> str: """查找场景图""" candidates = [ os.path.join(workspace, "scene", f"scene_{scene_id:03d}.png"), os.path.join(workspace, "scene", f"scene_{scene_id}.png"), ] for path in candidates: if os.path.exists(path): return path return None def create_video(workspace: str, timeline: list, audio_path: str, output_path: str, fps: int = 24, img_size: tuple = (1280, 720), subtitle: bool = True, log_fn=None): """ 用 moviepy 创建视频 Args: workspace: 工作区路径 timeline: 场景时间线 audio_path: 音频文件路径 output_path: 输出视频路径 fps: 帧率 img_size: 视频尺寸 (w, h) subtitle: 是否添加字幕 log_fn: 日志回调函数 """ def log(msg): if log_fn: log_fn(msg) else: print(msg) clips = [] for item in timeline: scene_id = item["scene_id"] start = item["start"] end = item["end"] duration = end - start if duration <= 0: continue img_path = find_scene_image(workspace, scene_id) if img_path: img_clip = ImageClip(img_path).with_duration(duration) img_clip = img_clip.resized(new_size=img_size) else: log(f" [警告] scene_{scene_id} 无图片，使用纯色背景") img_clip = ColorClip(size=img_size, color=(50, 50, 50)).with_duration(duration) img_clip = img_clip.with_start(start) clips.append(img_clip) # 字幕 if subtitle: asr_path = os.path.join(workspace, "result.json") if os.path.exists(asr_path): with open(asr_path, encoding="utf-8") as f: asr_result = json.load(f) segments = asr_result.get("segments", []) log(f"[*] 原始 ASR 片段: {len(segments)} 条") # 将 ASR segments 按时间分组（每段 3-5 秒，避免字幕堆满屏幕） def group_segments_into_sentences(segments: list) -> list: """ 将短的 ASR segments 合并成合理的字幕组。 支持中文（字级）和英文（词级）ASR 输出。 限制条件： - 每组最多 4 秒时长 - 每组最多 40 个字符（中文）/ 60 个字符（英文，含空格） - 遇到句号等结束标点时提前分组 """ if not segments: return [] sentence_endings = set('。！？.!?…\n') MAX_CHARS = 60 # 字符上限（英文含空格时会更长） MAX_DURATION = 4.0 def flush(text, start, end): if text and start is not None: grouped.append({"text": text.strip(), "start": start, "end": end}) grouped = [] current_words = [] # 词/字列表 current_start = None current_end = None for seg in segments: text = seg.get("text", "").strip() if not text: continue seg_start = seg.get("start", 0) seg_end = seg.get("end", 0) if current_start is None: current_start = seg_start # 判断加入当前词后是否超限 # 用空格拼英文，中文直接拼（中文 seg 通常每个词自带空格或无需加） preview_words = current_words + [text] # 根据是否含非 ASCII 字符判断是中文还是英文 is_ascii_word = all(ord(c) < 128 for c in text) if is_ascii_word: preview_text = " ".join(preview_words) else: preview_text = "".join(preview_words) preview_duration = seg_end - current_start # 先检查：加入后是否超限 → 如果是，先 flush 旧组，再开新组 overflow = ( len(preview_text) > MAX_CHARS or preview_duration > MAX_DURATION ) if overflow and current_words: # flush 当前组（不含新词） if is_ascii_word: flush(" ".join(current_words), current_start, current_end) else: flush("".join(current_words), current_start, current_end) current_words = [text] current_start = seg_start current_end = seg_end else: current_words.append(text) current_end = seg_end # 如果当前词是句尾标点，立刻 flush if text and text[-1] in sentence_endings: if is_ascii_word: flush(" ".join(current_words), current_start, current_end) else: flush("".join(current_words), current_start, current_end) current_words = [] current_start = None current_end = None # flush 剩余 if current_words: if all(ord(c) < 128 for c in current_words[0]): flush(" ".join(current_words), current_start or 0, current_end or 0) else: flush("".join(current_words), current_start or 0, current_end or 0) return grouped # 按句子分组 sentence_groups = group_segments_into_sentences(segments) log(f"[*] 添加字幕，共 {len(sentence_groups)} 个句子（从 {len(segments)} 个片段合并）") w, h = img_size font_size = max(int(h * 0.045), 20) margin_bottom = int(h * 0.10) max_chars_per_line = max(int(w / (font_size * 0.85)), 15) # 字体：优先黑体，fallback arial font_path = "C:/Windows/Fonts/simhei.ttf" if not os.path.exists(font_path): font_path = "C:/Windows/Fonts/arial.ttf" try: pil_font = ImageFont.truetype(font_path, font_size) except Exception: pil_font = ImageFont.load_default() # 字体度量 tmp_img = Image.new("RGBA", (1, 1)) tmp_draw = ImageDraw.Draw(tmp_img) metrics = tmp_draw.textbbox((0, 0), "gypj8q", font=pil_font) font_full_height = metrics[3] - metrics[1] def wrap_text(text: str, max_chars: int) -> list: """按字符数拆分多行，优先在标点处断行""" if len(text) <= max_chars: return [text] lines = [] while len(text) > max_chars: cut = max_chars for i in range(max_chars, max(max_chars - 5, 0), -1): if text[i] in "，。！？、；：\u201c\u201d\u2018\u2019\u2026\u2014,.!?;: ": cut = i + 1 break lines.append(text[:cut]) text = text[cut:] if text: lines.append(text) return lines def render_subtitle_pil(text_lines: list) -> np.ndarray: """用 PIL 渲染多行字幕，返回 RGBA numpy 数组""" line_spacing = int(font_size * 0.35) line_height = font_full_height + line_spacing line_widths = [] for line in text_lines: bbox = tmp_draw.textbbox((0, 0), line, font=pil_font) line_widths.append(bbox[2] - bbox[0]) img_w = max(line_widths) + 6 img_h = len(text_lines) * line_height + int(font_size * 0.3) img = Image.new("RGBA", (img_w, img_h), (0, 0, 0, 0)) draw = ImageDraw.Draw(img) for idx, line in enumerate(text_lines): x = (img_w - line_widths[idx]) // 2 y = idx * line_height for dx in (-2, -1, 0, 1, 2): for dy in (-2, -1, 0, 1, 2): if dx == 0 and dy == 0: continue draw.text((x + dx, y + dy), line, fill=(0, 0, 0, 255), font=pil_font) draw.text((x, y), line, fill=(255, 255, 255, 255), font=pil_font) return np.array(img) # 按句子显示字幕（而不是逐个片段） for sentence in sentence_groups: text = sentence["text"].strip() if not text: continue seg_start = sentence["start"] seg_end = sentence["end"] seg_dur = seg_end - seg_start if seg_dur <= 0: continue lines = wrap_text(text, max_chars_per_line) rgba = render_subtitle_pil(lines) txt_clip = ImageClip(rgba) txt_clip = txt_clip.with_position(("center", h - margin_bottom - rgba.shape[0])) txt_clip = txt_clip.with_duration(seg_dur).with_start(seg_start) clips.append(txt_clip) else: log("[提示] 未找到 result.json，跳过字幕") if not clips: raise ValueError("没有可用的场景片段！") # 加载音频 audio = AudioFileClip(audio_path) # 合成视频 video = CompositeVideoClip(clips, size=img_size) video = video.with_audio(audio) video = video.with_duration(audio.duration) # 导出 video.write_videofile( output_path, fps=fps, codec='libx264', audio_codec='aac', bitrate='8000k', ) video.close() log(f"[OK] 视频已保存: {output_path}") def main(workspace: str = None, fps: int = 24, size: str = "1280x720", subtitle: bool = True, log_fn=None): """ 主入口 Args: workspace: 工作区路径 fps: 帧率 size: 视频尺寸 "WxH" subtitle: 是否添加字幕 log_fn: 日志回调 """ from config import DEFAULT_VIDEO_SIZE if workspace is None: workspace = os.path.join(os.path.dirname(os.path.abspath(__file__)), "workspace", "1") w, h = map(int, (size or DEFAULT_VIDEO_SIZE).split("x")) img_size = (w, h) def log(msg): if log_fn: log_fn(msg) else: print(msg) log(f"[视频合成] 工作区: {workspace}") log(f"[视频合成] 尺寸: {img_size}, FPS: {fps}") # 1. 加载 scene_plan scene_plan = load_scene_plan(workspace) num_scenes = len(scene_plan["scenes"]) log(f"[视频合成] 共 {num_scenes} 个场景") # 2. 加载音频 audio_path = os.path.join(workspace, "voice.mp3") if not os.path.exists(audio_path): raise FileNotFoundError(f"音频文件不存在: {audio_path}") # 使用 try-finally 确保音频资源正确释放 audio = AudioFileClip(audio_path) try: audio_duration = audio.duration log(f"[视频合成] 音频时长: {audio_duration:.2f} 秒") finally: audio.close() # 3. 读取 ASR 结果 asr_result = load_asr_result(workspace) if asr_result: seg_count = len(asr_result.get("segments", [])) log(f"[视频合成] ASR: {seg_count} 个片段") else: log("[视频合成] 无 ASR 结果，按平均分配") # 4. 构建时间线 timeline = build_scene_timeline(scene_plan, asr_result, audio_duration, workspace=workspace) log("[视频合成] 场景时间分配：") matched_count = 0 unmatched_count = 0 for item in timeline: dur = item['end'] - item['start'] # 显示匹配质量 similarity = item.get('similarity', 0.0) if similarity > 0: quality = f"[{similarity:.0%}]" matched_count += 1 elif item.get('has_match', False): quality = "[OK]" matched_count += 1 else: quality = "[兜底]" unmatched_count += 1 log(f" Scene {item['scene_id']:2d}: {item['start']:6.2f}s - {item['end']:6.2f}s ({dur:.2f}s) {quality}") if unmatched_count > 0: log(f"[警告] {unmatched_count}/{len(timeline)} 个场景未精确匹配，使用兜底分配") else: log(f"[OK] 所有场景均成功匹配") # 5. 生成视频 has_image = sum(1 for item in timeline if find_scene_image(workspace, item["scene_id"]) is not None) log(f"[视频合成] 有图片场景: {has_image}/{len(timeline)}") if has_image == 0: log("[警告] 所有场景都没有图片，视频将使用纯色背景！") output_path = os.path.join(workspace, "output_video.mp4") create_video(workspace, timeline, audio_path, output_path, fps=fps, img_size=img_size, subtitle=subtitle, log_fn=log_fn) log(f"[完成] 视频: {output_path}") return output_path if __name__ == "__main__": main()