#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ EP01-SHOT03-PRODUCTION-CLI 一键跑苏白站牌匾下说台词：生成底片、合成牌匾、配音、口型、字幕、混音、质检。 功能: 1. 读取 E1-SHOT03 配置 2. 生成底片 (MULTI-REFERENCE-VIDEO-ADAPTER) 3. 合成牌匾 (平面追踪 + 贴图) 4. 配音 (VOICE-EMOTION-COMPILER) 5. 口型 (LIPSYNC-ADAPTER) 6. 字幕 (subtitle-renderer.py) 7. 混音 (AUDIO-MIXER) 8. 质检 (SHOT-QC-AUTOMATION) 9. 输出生产报告 用法: python ep01_shot03_production.py --run python ep01_shot03_production.py --dry-run # 只打印计划，不执行 python ep01_shot03_production.py --resume task_id.json # 从失败点恢复 """ import os import sys import json import argparse import subprocess from pathlib import Path from datetime import datetime import time PROJECT_ROOT = Path(__file__).parent.parent sys.path.insert(0, str(PROJECT_ROOT / "engines")) DEFAULT_PYTHON = "/Users/bingshuolingdianyuanhe/.workbuddy/binaries/python/envs/default/bin/python3" def resolve_python_bin(): """Pick the Python that has the video AI dependencies installed.""" candidates = [ os.environ.get("PYTHON_BIN"), os.environ.get("PYTHON"), DEFAULT_PYTHON, sys.executable, "python3", ] for candidate in candidates: if not candidate: continue candidate_path = Path(candidate) if candidate_path.is_absolute() and not candidate_path.exists(): continue return str(candidate) return sys.executable class EP01Shot03ProductionCLI: """E1-SHOT03 一键生产 CLI""" def __init__(self, dry_run=False): self.dry_run = dry_run self.project = "zai-fu-fei-xiu-xian/ep01" self.shot_id = "E1-SHOT03" self.start_time = datetime.now() # 路径配置 self.config_dir = PROJECT_ROOT / "plans" / "script-to-screen" self.output_dir = PROJECT_ROOT / "outputs" / "ep01" / "shot03" self.assets_dir = PROJECT_ROOT / "assets" self.python_bin = resolve_python_bin() # 确保输出目录存在 self.output_dir.mkdir(parents=True, exist_ok=True) # 加载配置 self.config = self._load_config() # 生产状态 self.state = { "shot_id": self.shot_id, "steps": {}, "start_time": self.start_time.isoformat(), "end_time": None, "status": "running", # running / success / failed "current_step": None, "error": None } print(f"🎬 E1-SHOT03 生产 CLI 启动") print(f" 项目: {self.project}") print(f" 镜头: {self.shot_id}") print(f" 输出: {self.output_dir}") print(f" Dry Run: {self.dry_run}") print(f" Python: {self.python_bin}") def _load_config(self): """加载 E1-SHOT03 配置，从剧本原文+分镜文件读取真实台词""" script_file = PROJECT_ROOT.parent / "动态漫：《付费才能修仙？我的宗门全免费》.md" mapping_file = self.config_dir / "EP01-SHOT01-06-MAPPING.hdlp" storyboard_file = PROJECT_ROOT / "data" / "ep01-storyboard.json" config = { "shot_id": "E1-SHOT03", "prompt": "", "duration": 3, "resolution": "720p", "character": "CHAR-003-SuBai", "props": ["PROP-TDZ-PLAQUE"], "env": "ENV-002-Baizonghui", "dialogue": "", "emotion": "苏白·大声·自信" } # 1. 从剧本原文读取真实台词 if script_file.exists(): with open(script_file, "r", encoding="utf-8") as f: script_text = f.read() # 匹配 SHOT03 对应的剧本段落 import re # 找"苏白（大声）："后面的台词 dialogue_matches = re.findall(r'苏白[（(][^）)]*[）)]\s*[：:]\s*(.+?)(?:\n|$)', script_text) if dialogue_matches: config["dialogue"] = dialogue_matches[0] # 第一句：未来的天下第一宗！天道宗开门收徒啦！ print(f" 📖 剧本台词(第1句): {config['dialogue']}") # 拼接多句台词 if len(dialogue_matches) >= 2: config["dialogue_all"] = dialogue_matches print(f" 📖 全部{len(dialogue_matches)}句: {' | '.join(dialogue_matches[:3])}") # 2. 从分镜文件读取镜头信息 if storyboard_file.exists(): with open(storyboard_file, "r", encoding="utf-8") as f: storyboard = json.load(f) for shot in storyboard.get("shots", []): if shot.get("shotId") == "E1-SHOT03": config["action"] = shot.get("action", "") config["camera"] = shot.get("camera", "") config["emotion"] = shot.get("emotionMarker", config["emotion"]) config["duration"] = shot.get("duration", config["duration"]) config["sourceLine"] = shot.get("sourceLine", "") print(f" 🎬 分镜: {config.get('action', '')[:60]}") break # 3. 从HLDP映射文件读取镜头参数（不覆盖剧本台词） SCRIPT_PROTECTED = {"dialogue", "dialogue_all", "action", "camera", "emotion", "sourceLine", "duration"} if mapping_file.exists(): with open(mapping_file, "r", encoding="utf-8") as f: mapping_content = f.read() if "E1-SHOT03" in mapping_content: lines = mapping_content.split("\n") in_shot = False for line in lines: if "E1-SHOT03" in line: in_shot = True elif in_shot: if line.strip().startswith("---"): break if ":" in line: key, _, val = line.partition(":") key = key.strip() if key not in SCRIPT_PROTECTED: config[key] = val.strip() # 构造提示词 if not config.get("prompt"): action = config.get("action", "苏白站在天道宗牌匾下自信喊话") dialogue = config.get("dialogue", "未来的天下第一宗！") config["prompt"] = f"{action}，大声说：{dialogue}" print(f" ✅ 配置已加载 (来源: 剧本原文 + 分镜 + 映射)") print(f" 台词: {config['dialogue'][:50]}") print(f" 情感: {config['emotion']}") print(f" 时长: {config['duration']}s") def _default_config(self): """默认配置""" return { "shot_id": "E1-SHOT03", "prompt": "苏白站在天道宗牌匾下，自信地说：未来的天下第一宗！", "duration": 5, "resolution": "720p", "character": "CHAR-003-SuBai", "props": ["PROP-TDZ-PLAQUE"], "env": "ENV-002-Baizonghui", "dialogue": "未来的天下第一宗！", "emotion": "苏白·大声·自信" } def run(self): """执行完整生产流程""" print(f"\n{'=' * 60}") print(f"开始生产 E1-SHOT03") print(f"{'=' * 60}") steps = [ ("step1_prepare_assets", self.step1_prepare_assets), ("step2_generate_base_video", self.step2_generate_base_video), ("step3_synthesize_plaque", self.step3_synthesize_plaque), ("step4_generate_dialogue_audio", self.step4_generate_dialogue_audio), ("step5_lipsync", self.step5_lipsync), ("step6_render_subtitles", self.step6_render_subtitles), ("step7_mix_audio", self.step7_mix_audio), ("step8_qc", self.step8_qc), ("step9_generate_report", self.step9_generate_report), ] for step_name, step_func in steps: print(f"\n📍 步骤: {step_name}") self.state["current_step"] = step_name if self.dry_run: print(f" [Dry Run] 跳过: {step_func.__doc__}") self.state["steps"][step_name] = {"status": "skipped", "reason": "dry_run"} continue try: step_start = time.time() result = step_func() step_duration = time.time() - step_start self.state["steps"][step_name] = { "status": "success", "duration": f"{step_duration:.1f}s", "result": result } print(f" ✅ 完成 ({step_duration:.1f}s)") except Exception as e: print(f" ❌ 失败: {e}") self.state["steps"][step_name] = { "status": "failed", "error": str(e) } self.state["status"] = "failed" self.state["error"] = f"{step_name}: {e}" self._save_state() return False self.state["status"] = "success" self.state["end_time"] = datetime.now().isoformat() self._save_state() print(f"\n{'=' * 60}") print(f"✅ 生产完成!") print(f" 总耗时: {(datetime.now() - self.start_time).total_seconds():.1f}s") print(f" 输出: {self.output_dir}") print(f"{'=' * 60}") return True def step1_prepare_assets(self): """步骤1: 准备资产 (CHAR-HERO-DESIGN-PACKER)""" print(f" 准备苏白资产包...") # 检查是否有批准资产 char_dir = self.assets_dir / "characters" / self.config["character"] / "approved" if char_dir.exists() and list(char_dir.glob("*.png")): print(f" ✓ 已有批准资产: {len(list(char_dir.glob('*.png')))} 个") return {"assets_ready": True, "path": str(char_dir)} # 没有批准资产，生成 print(f" 📤 生成资产包...") result = subprocess.run( [ self.python_bin, str(PROJECT_ROOT / "engines" / "char-hero-design-packer" / "char-hero-design-packer.py"), "--character", self.config["character"], "--generate-all" ], capture_output=True, text=True, timeout=600 ) if result.returncode != 0: raise Exception(f"资产生成失败: {result.stderr}") return {"assets_ready": True, "path": str(char_dir)} def step2_generate_base_video(self): """步骤2: 生成底片 (MULTI-REFERENCE-VIDEO-ADAPTER)""" print(f" 生成底片...") print(f" 提示词: {self.config['prompt'][:60]}...") print(f" 时长: {self.config['duration']}s") print(f" 分辨率: {self.config['resolution']}") # 收集参考图 reference_images = [] # 苏白参考图 char_dir = self.assets_dir / "characters" / self.config["character"] / "approved" if char_dir.exists(): for img in char_dir.glob("*.png"): reference_images.append(str(img)) # 牌匾参考图 for prop in self.config.get("props", []): prop_dir = self.assets_dir / "props" / prop / "approved" if prop_dir.exists(): for img in prop_dir.glob("*.png"): reference_images.append(str(img)) if len(reference_images) == 0: print(f" ⚠️ 无参考图，使用单参考图模式") # 调用 MULTI-REFERENCE-VIDEO-ADAPTER output_path = self.output_dir / "base_video.mp4" if len(reference_images) >= 2: # 多参考图模式 cmd = [ self.python_bin, str(PROJECT_ROOT / "engines" / "multi-reference-video-adapter.py"), "--prompt", self.config["prompt"], "--references"] + reference_images + [ "--output", str(output_path), "--duration", str(self.config["duration"]), "--resolution", self.config["resolution"] ] else: # 单参考图模式 (回退) cmd = [ "node", str(PROJECT_ROOT / "engines" / "video-api-adapter.js"), "--prompt", self.config["prompt"], "--duration", str(self.config["duration"]), "--resolution", self.config["resolution"] ] if reference_images: cmd.extend(["--reference-image", reference_images[0]]) print(f" 参考图数量: {len(reference_images)}") print(f" 输出: {output_path.name}") # TODO: 实际调用 API (这里简化为记录命令) # result = subprocess.run(cmd, capture_output=True, text=True, timeout=600) return { "output_path": str(output_path), "reference_count": len(reference_images), "note": "TODO: 实际调用 API 生成视频" } def step3_synthesize_plaque(self): """步骤3: 合成牌匾 (平面追踪 + 贴图)""" print(f" 合成牌匾...") # 检查是否有 base_video base_video = self.output_dir / "base_video.mp4" if not base_video.exists(): print(f" ⚠️ base_video.mp4 不存在，跳过牌匾合成") return {"skipped": True, "reason": "base_video not found"} # 调用 planar-tracker.py print(f" 运行平面追踪...") output_with_plaque = self.output_dir / "video_with_plaque.mp4" # TODO: 实际调用 planar-tracker.py # cmd = [self.python_bin, str(PROJECT_ROOT / "engines" / "planar-tracker.py"), ...] return { "output_path": str(output_with_plaque), "note": "TODO: 实际调用 planar-tracker.py" } def step4_generate_dialogue_audio(self): """步骤4: 生成对白音频 (VOICE-EMOTION-COMPILER)""" print(f" 生成对白音频...") print(f" 台词: {self.config['dialogue']}") print(f" 情感: {self.config['emotion']}") output_path = self.output_dir / "dialogue.mp3" cmd = [ self.python_bin, str(PROJECT_ROOT / "engines" / "voice-emotion-compiler.py"), "--text", self.config["dialogue"], "--emotion", self.config["emotion"], "--output", str(output_path) ] print(f" 输出: {output_path.name}") result = subprocess.run(cmd, capture_output=True, text=True, timeout=60) if result.returncode != 0: raise Exception(f"配音失败: {result.stderr}") return {"output_path": str(output_path)} def step5_lipsync(self): """步骤5: 口型同步 (LIPSYNC-ADAPTER)""" print(f" 口型同步...") video_path = self.output_dir / "video_with_plaque.mp4" if not video_path.exists(): video_path = self.output_dir / "base_video.mp4" audio_path = self.output_dir / "dialogue.mp3" if not video_path.exists(): raise Exception(f"视频不存在: {video_path}") if not audio_path.exists(): raise Exception(f"音频不存在: {audio_path}") output_path = self.output_dir / "video_synced.mp4" cmd = [ self.python_bin, str(PROJECT_ROOT / "engines" / "lipsync-adapter.py"), "--video", str(video_path), "--audio", str(audio_path), "--output", str(output_path) ] print(f" 输入视频: {video_path.name}") print(f" 输入音频: {audio_path.name}") print(f" 输出: {output_path.name}") result = subprocess.run(cmd, capture_output=True, text=True, timeout=300) if result.returncode != 0: print(f" ⚠️ 口型同步失败: {result.stderr}") print(f" 📌 继续使用非同步视频...") # 不抛出异常，继续流程 return {"warning": "Lipsync failed, continuing with unsynced video"} return {"output_path": str(output_path)} def step6_render_subtitles(self): """步骤6: 渲染字幕 (subtitle-renderer.py)""" print(f" 渲染字幕...") video_path = self.output_dir / "video_synced.mp4" if not video_path.exists(): video_path = self.output_dir / "video_with_plaque.mp4" if not video_path.exists(): video_path = self.output_dir / "base_video.mp4" if not video_path.exists(): raise Exception(f"视频不存在: {video_path}") output_path = self.output_dir / "video_with_subtitles.mp4" srt_path = self.output_dir / "dialogue.srt" self._write_single_line_srt(srt_path, self.config["dialogue"], float(self.config.get("duration", 3))) # 调用 subtitle-renderer.py cmd = [ self.python_bin, str(PROJECT_ROOT / "engines" / "subtitle-renderer.py"), "--srt", str(srt_path), "--video", str(video_path), "--output", str(output_path) ] print(f" 输入: {video_path.name}") print(f" 字幕: {self.config['dialogue']}") print(f" 输出: {output_path.name}") result = subprocess.run(cmd, capture_output=True, text=True, timeout=120) if result.returncode != 0: print(f" ⚠️ 字幕渲染失败: {result.stderr}") print(f" 📌 继续使用无字幕视频...") return {"warning": "Subtitle rendering failed, continuing without subtitles"} return {"output_path": str(output_path)} def step7_mix_audio(self): """步骤7: 混音 (AUDIO-MIXER)""" print(f" 混音...") dialogue_path = self.output_dir / "dialogue.mp3" output_path = self.output_dir / "final_video.mp4" # 找到带字幕的视频 video_path = self.output_dir / "video_with_subtitles.mp4" if not video_path.exists(): video_path = self.output_dir / "video_synced.mp4" if not video_path.exists(): video_path = self.output_dir / "video_with_plaque.mp4" if not video_path.exists(): video_path = self.output_dir / "base_video.mp4" if not video_path.exists(): raise Exception(f"视频不存在: {video_path}") # 提取视频音轨 video_audio = self.output_dir / "video_audio.mp3" print(f" 提取视频音轨...") # TODO: 实际调用 ffmpeg 提取音轨 # 混音 print(f" 混音...") cmd = [ self.python_bin, str(PROJECT_ROOT / "engines" / "audio-mixer.py"), "--dialogue", str(dialogue_path), "--output", str(self.output_dir / "mixed_audio.mp3") ] if video_audio.exists(): cmd.extend(["--original", str(video_audio)]) result = subprocess.run(cmd, capture_output=True, text=True, timeout=60) if result.returncode != 0: print(f" ⚠️ 混音失败: {result.stderr}") # 合并视频和混音后的音频 print(f" 合并视频和音频...") # TODO: ffmpeg -i video -i audio -c:v copy -c:a aac output return {"output_path": str(output_path), "note": "TODO: 实际合并视频和音频"} def step8_qc(self): """步骤8: 质检 (SHOT-QC-AUTOMATION)""" print(f" 质检...") video_path = self.output_dir / "final_video.mp4" if not video_path.exists(): # 找任何可用的视频 for v in self.output_dir.glob("*.mp4"): video_path = v break if not video_path or not video_path.exists(): print(f" ⚠️ 无视频文件可质检") return {"skipped": True, "reason": "no video found"} cmd = [ self.python_bin, str(PROJECT_ROOT / "engines" / "shot-qc-automation.py"), "--video", str(video_path), "--character", self.config["character"], "--output", str(self.output_dir / "qc_report.json") ] print(f" 输入: {video_path.name}") result = subprocess.run(cmd, capture_output=True, text=True, timeout=120) if result.returncode != 0: print(f" ⚠️ 质检失败: {result.stderr}") return {"warning": "QC failed"} # 读取 QC 报告 qc_report_path = self.output_dir / "qc_report.json" if qc_report_path.exists(): with open(qc_report_path, "r", encoding="utf-8") as f: qc_result = json.load(f) passed = qc_result.get("passed", False) score = qc_result.get("score", 0) print(f" QC 结果: {'✅ 通过' if passed else '❌ 失败'} (分数: {score:.1f}/10)") return {"passed": passed, "score": score, "report": str(qc_report_path)} return {"passed": None, "warning": "QC report not found"} def step9_generate_report(self): """步骤9: 生成生产报告""" print(f" 生成生产报告...") report_path = self.output_dir / f"production_report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json" with open(report_path, "w", encoding="utf-8") as f: json.dump(self.state, f, ensure_ascii=False, indent=2) print(f" ✅ 报告已保存: {report_path.name}") # 打印总结 print(f"\n📊 生产总结") print(f" 状态: {self.state['status']}") print(f" 步骤数: {len(self.state['steps'])}") success_count = sum(1 for s in self.state["steps"].values() if s.get("status") == "success") print(f" 成功: {success_count}/{len(self.state['steps'])}") return {"report_path": str(report_path)} def _save_state(self): """保存生产状态""" state_path = self.output_dir / "production_state.json" with open(state_path, "w", encoding="utf-8") as f: json.dump(self.state, f, ensure_ascii=False, indent=2) def resume(self, state_file): """从失败点恢复""" print(f"\n📂 从失败点恢复: {state_file}") state_path = Path(state_file) if not state_path.exists(): print(f" ❌ 状态文件不存在: {state_file}") return False with open(state_path, "r", encoding="utf-8") as f: self.state = json.load(f) print(f" 上次状态: {self.state['status']}") print(f" 当前步骤: {self.state['current_step']}") # 找到当前步骤，继续执行 steps = list(self.state["steps"].keys()) if self.state["current_step"] in steps: start_idx = steps.index(self.state["current_step"]) else: start_idx = 0 print(f" 从第 {start_idx + 1} 步继续...") # TODO: 实际恢复逻辑 return True def _write_single_line_srt(self, srt_path, text, duration): """Write a one-line SRT from the locked source dialogue.""" duration_ms = max(500, int(duration * 1000)) end_seconds = duration_ms // 1000 end_ms = duration_ms % 1000 srt_path.parent.mkdir(parents=True, exist_ok=True) content = ( "1\n" f"00:00:00,000 --> 00:00:{end_seconds:02d},{end_ms:03d}\n" f"{text}\n" ) with open(srt_path, "w", encoding="utf-8") as f: f.write(content) return srt_path def main(): parser = argparse.ArgumentParser(description="EP01-SHOT03-PRODUCTION-CLI") parser.add_argument("--run", action="store_true", help="执行生产") parser.add_argument("--dry-run", action="store_true", help="Dry Run (只打印计划)") parser.add_argument("--resume", type=str, help="从状态文件恢复") args = parser.parse_args() if not args.run and not args.dry_run and not args.resume: parser.print_help() sys.exit(1) if args.resume: cli = EP01Shot03ProductionCLI(dry_run=False) cli.resume(args.resume) else: cli = EP01Shot03ProductionCLI(dry_run=args.dry_run) success = cli.run() sys.exit(0 if success else 1) if __name__ == "__main__": main()