648 lines
24 KiB
Python
648 lines
24 KiB
Python
|
#!/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()
|