2007 lines
88 KiB
Python
2007 lines
88 KiB
Python
# LIB · 蛋蛋短剧生产管线 · 命令行入口
|
||
# D181 · 2026-07-10 · v2.0
|
||
# Usage: python -m lib.cli <command> [args]
|
||
# 59 commands · 24 engines · 8 rounds of iteration
|
||
import sys
|
||
import os
|
||
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
|
||
|
||
from lib.models import get_db, init
|
||
|
||
def cmd_init():
|
||
init()
|
||
|
||
def cmd_project(action="list", project_id="", name=""):
|
||
"""项目管理: list / new / rm"""
|
||
db = get_db()
|
||
if action == "list":
|
||
rows = db.execute("SELECT id, name, created_at FROM projects ORDER BY created_at").fetchall()
|
||
if not rows:
|
||
print("还没有项目。用 project new <id> <name> 创建")
|
||
return
|
||
for r in rows:
|
||
eps = db.execute("SELECT COUNT(*) FROM episodes WHERE project_id = ?", (r["id"],)).fetchone()
|
||
print(f" {r['id']:20s} {r['name']:15s} {eps[0]}集 {r['created_at']}")
|
||
elif action == "new":
|
||
if not project_id or not name:
|
||
print("❌ 用法: project new <id> <名称>")
|
||
return
|
||
db.execute("INSERT OR IGNORE INTO projects (id, name) VALUES (?, ?)", (project_id, name))
|
||
db.commit()
|
||
print(f"✅ 项目 {project_id} ({name}) 已创建")
|
||
else:
|
||
print(f"❌ 未知操作: {action} (list/new)")
|
||
|
||
def cmd_episode(action="list", project_id="", episode_id="", number="1"):
|
||
"""剧集管理: list / new"""
|
||
db = get_db()
|
||
if action == "list":
|
||
rows = db.execute("""
|
||
SELECT e.id, e.project_id, e.number, e.status, p.name as project_name
|
||
FROM episodes e JOIN projects p ON e.project_id = p.id ORDER BY p.id, e.number
|
||
""").fetchall()
|
||
if not rows:
|
||
print("还没有剧集。用 episode new <project_id> <episode_id> [编号] 创建")
|
||
return
|
||
for r in rows:
|
||
shots = db.execute("SELECT COUNT(*) FROM shots WHERE episode_id = ?", (r["id"],)).fetchone()
|
||
print(f" {r['project_name']:12s} E{r['number']:02d} {r['id']:20s} {r['status']:8s} {shots[0]}镜")
|
||
elif action == "new":
|
||
if not project_id or not episode_id:
|
||
print("❌ 用法: episode new <project_id> <episode_id> [编号]")
|
||
return
|
||
# 验证项目存在
|
||
proj = db.execute("SELECT id FROM projects WHERE id = ?", (project_id,)).fetchone()
|
||
if not proj:
|
||
print(f"❌ 项目 {project_id} 不存在，先创建: project new {project_id} <名称>")
|
||
return
|
||
db.execute("INSERT OR IGNORE INTO episodes (id, project_id, number, status) VALUES (?, ?, ?, 'pending')",
|
||
(episode_id, project_id, int(number)))
|
||
db.commit()
|
||
print(f"✅ 剧集 {episode_id} (第{number}集) 已创建")
|
||
|
||
def cmd_asset(action="list", project_id="", asset_type="CHAR", name="", file_path=""):
|
||
"""资产管理: list / add / approve"""
|
||
db = get_db()
|
||
if action == "list":
|
||
pid = project_id or "deep-sea-voyage"
|
||
rows = db.execute("SELECT * FROM assets WHERE project_id = ? ORDER BY type, name", (pid,)).fetchall()
|
||
if not rows:
|
||
print(f"项目 {pid} 还没有资产")
|
||
return
|
||
for r in rows:
|
||
s = "✅" if r["status"] == "approved" else "🔴"
|
||
print(f" [{r['type']:6s}] {s} {r['name']:25s} {r['file_path']}")
|
||
elif action == "add":
|
||
if not project_id or not name or not file_path:
|
||
print("❌ 用法: asset add <project_id> CHAR|ENV|PROP <名称> <文件路径>")
|
||
return
|
||
aid = f"{project_id}-{asset_type}-{name}"
|
||
db.execute("INSERT OR REPLACE INTO assets (id, project_id, type, name, file_path, status) VALUES (?, ?, ?, ?, ?, 'pending')",
|
||
(aid, project_id, asset_type, name, file_path))
|
||
db.commit()
|
||
print(f"✅ 资产 {aid} 已添加")
|
||
elif action == "approve":
|
||
if not project_id or not name:
|
||
print("❌ 用法: asset approve <project_id> <资产名>")
|
||
return
|
||
db.execute("UPDATE assets SET status = 'approved', approved_at = datetime('now') WHERE project_id = ? AND name = ?",
|
||
(project_id, name))
|
||
db.commit()
|
||
print(f"✅ 资产 {name} 已批准")
|
||
|
||
def cmd_shot_list(episode="DSV-EP01"):
|
||
db = get_db()
|
||
rows = db.execute("""
|
||
SELECT shot_number, description, status, model, seedance_task_id, output_path, error_log
|
||
FROM shots WHERE episode_id = ? ORDER BY shot_number
|
||
""", (episode,)).fetchall()
|
||
if not rows:
|
||
print(f"No shots found for {episode}. Run: python -m lib.cli sync-shots")
|
||
return
|
||
print(f"\n{'#':4s} {'状态':10s} {'描述':40s} {'模型':25s} {'任务ID':30s}")
|
||
print("-" * 115)
|
||
for r in rows:
|
||
status_icon = {"pending":"🔴","generating":"🟡","done":"✅","failed":"❌"}.get(r["status"], "❓")
|
||
desc = (r["description"] or "")[:38]
|
||
task = (r["seedance_task_id"] or "")[:28]
|
||
print(f"S{r['shot_number']:>2s} {status_icon} {r['status']:8s} {desc:40s} {r['model']:25s} {task:30s}")
|
||
|
||
def cmd_shot_status(shot_id):
|
||
db = get_db()
|
||
r = db.execute("""
|
||
SELECT * FROM shots WHERE id = ?
|
||
""", (shot_id,)).fetchone()
|
||
if not r:
|
||
print(f"Shot {shot_id} not found")
|
||
return
|
||
print(f"\nShot {r['shot_number']} · {r['status']}")
|
||
print(f" 描述: {r['description']}")
|
||
print(f" 剧本: {r['script_ref']}")
|
||
print(f" 镜头: {r['camera']} · {r['duration_sec']}s")
|
||
print(f" 模型: {r['model']} · {r['resolution']}")
|
||
print(f" 提示词: {r['prompt']}")
|
||
print(f" 任务ID: {r['seedance_task_id']}")
|
||
print(f" 输出: {r['output_path']}")
|
||
if r['error_log']:
|
||
print(f" ❌ 错误: {r['error_log']}")
|
||
# Show linked assets
|
||
assets = db.execute("""
|
||
SELECT a.type, a.name, sa.role FROM shot_assets sa
|
||
JOIN assets a ON sa.asset_id = a.id WHERE sa.shot_id = ?
|
||
""", (shot_id,)).fetchall()
|
||
if assets:
|
||
print(f" 关联资产:")
|
||
for a in assets:
|
||
print(f" [{a['role']}] {a['type']}: {a['name']}")
|
||
|
||
def cmd_asset_list(project="deep-sea-voyage"):
|
||
db = get_db()
|
||
rows = db.execute("""
|
||
SELECT id, type, name, status, approved_at FROM assets
|
||
WHERE project_id = ? ORDER BY type, name
|
||
""", (project,)).fetchall()
|
||
print(f"\n{'类型':6s} {'名称':25s} {'状态':10s} {'批准时间':20s} {'ID'}")
|
||
print("-" * 75)
|
||
for r in rows:
|
||
s = "✅" if r["status"] == "approved" else "🔴"
|
||
print(f"{r['type']:6s} {r['name']:25s} {s} {r['status']:8s} {(r['approved_at'] or ''):20s} {r['id']}")
|
||
|
||
def cmd_sync_shots():
|
||
"""Import shots from SHOT-LIST-EP01.hdlp into database"""
|
||
db = get_db()
|
||
shot_list_path = os.path.join(
|
||
os.path.dirname(os.path.dirname(__file__)),
|
||
"projects", "deep-sea-voyage", "shots", "SHOT-LIST-EP01.hdlp"
|
||
)
|
||
if not os.path.exists(shot_list_path):
|
||
print(f"Shot list not found: {shot_list_path}")
|
||
return
|
||
|
||
with open(shot_list_path, 'r', encoding='utf-8') as f:
|
||
content = f.read()
|
||
|
||
# Parse shots - simple regex approach
|
||
import re
|
||
shots = []
|
||
blocks = re.split(r'### Shot (\d+)', content)
|
||
for i in range(1, len(blocks), 2):
|
||
num = blocks[i]
|
||
text = blocks[i+1] if i+1 < len(blocks) else ""
|
||
shots.append((num, text.strip()))
|
||
|
||
for num, text in shots:
|
||
shot_id = f"DSV-EP01-S{num.zfill(2)}"
|
||
# Extract description
|
||
desc_match = re.search(r'\*\*剧本\*\*[：:]\s*(.+?)(?:\n|$)', text)
|
||
desc = desc_match.group(1)[:80] if desc_match else f"Shot {num}"
|
||
# Extract duration
|
||
dur_match = re.search(r'(\d+)s', text.split('\n')[0] if '\n' in text else text)
|
||
dur = float(dur_match.group(1)) if dur_match else 6.0
|
||
# Determine status
|
||
status = "done" if num == "01" else "pending"
|
||
|
||
db.execute("""
|
||
INSERT OR REPLACE INTO shots
|
||
(id, episode_id, shot_number, description, script_ref, camera, duration_sec, status, output_path)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||
""", (
|
||
shot_id, "DSV-EP01", f"S{num.zfill(2)}",
|
||
desc, f"EP01-SCRIPT-LOCK line {num}",
|
||
"9:16 vertical", dur, status,
|
||
f"outputs/videos/SHOT-{num.zfill(2)}.mp4" if status == "done" else None
|
||
))
|
||
db.commit()
|
||
print(f"Synced {len(shots)} shots from SHOT-LIST-EP01.hdlp")
|
||
|
||
def cmd_link_asset(shot_id, asset_id, role):
|
||
db = get_db()
|
||
db.execute("INSERT OR REPLACE INTO shot_assets (shot_id, asset_id, role) VALUES (?, ?, ?)",
|
||
(shot_id, asset_id, role))
|
||
db.commit()
|
||
print(f"Linked: {shot_id} ← [{role}] {asset_id}")
|
||
|
||
def cmd_task_log(shot_id=None):
|
||
db = get_db()
|
||
if shot_id:
|
||
rows = db.execute("""
|
||
SELECT * FROM generation_tasks WHERE shot_id = ? ORDER BY created_at DESC
|
||
""", (shot_id,)).fetchall()
|
||
else:
|
||
rows = db.execute("""
|
||
SELECT * FROM generation_tasks ORDER BY created_at DESC LIMIT 20
|
||
""").fetchall()
|
||
for r in rows:
|
||
icon = {"submitted":"📤","running":"🔄","succeeded":"✅","failed":"❌"}.get(r["status"], "❓")
|
||
print(f"{icon} {r['id'][:20]:20s} {r['shot_id']:15s} {r['task_type']:12s} {r['status']}")
|
||
|
||
def cmd_exp_add(category, title, content):
|
||
db = get_db()
|
||
db.execute("INSERT INTO experience_log (category, title, content) VALUES (?, ?, ?)",
|
||
(category, title, content))
|
||
db.commit()
|
||
print(f"Experience logged: [{category}] {title}")
|
||
|
||
def cmd_exp_list(category=None):
|
||
db = get_db()
|
||
if category:
|
||
rows = db.execute("SELECT * FROM experience_log WHERE category = ? ORDER BY created_at DESC",
|
||
(category,)).fetchall()
|
||
else:
|
||
rows = db.execute("SELECT * FROM experience_log ORDER BY created_at DESC LIMIT 20").fetchall()
|
||
for r in rows:
|
||
print(f"[{r['category']}] {r['title']} ({r['created_at']})")
|
||
print(f" {r['content'][:120]}")
|
||
|
||
def cmd_chat(prompt):
|
||
from lib.doubao_chat import chat
|
||
r = chat(prompt)
|
||
if "content" in r:
|
||
print(r["content"])
|
||
else:
|
||
print(f"❌ {r}")
|
||
|
||
def cmd_breakdown(file_path, episode_num=1):
|
||
from lib.doubao_chat import breakdown_script
|
||
with open(file_path, "r", encoding="utf-8") as f:
|
||
script = f.read()
|
||
r = breakdown_script(script, int(episode_num))
|
||
if "content" in r:
|
||
print(r["content"])
|
||
print(f"\n📊 {r['model']} · {r['tokens'].get('total_tokens',0)} tokens")
|
||
else:
|
||
print(f"❌ {r}")
|
||
|
||
def cmd_collect(shot_id):
|
||
from lib.seedance import check_and_collect
|
||
r = check_and_collect(shot_id)
|
||
if r.get("error"): print(f"❌ {r['error']}")
|
||
elif r.get("status") == "done": print(f"✅ {shot_id} 完成")
|
||
elif r.get("status") == "running": print(f"🔄 生成中")
|
||
else: print(f"❓ {r}")
|
||
|
||
def cmd_render(episode_id="DSV-EP01"):
|
||
from lib.composer import render_episode
|
||
r = render_episode(episode_id)
|
||
if "error" in r: print(f"❌ {r['error']}")
|
||
else: print(f"✅ {r['shot_count']}镜 {r['total_duration']}s → {r['output_path']}")
|
||
|
||
def cmd_compose(episode_id="DSV-EP01"):
|
||
from lib.composer import concat_shots
|
||
r = concat_shots(episode_id)
|
||
if "error" in r: print(f"❌ {r['error']}")
|
||
else: print(f"✅ {r['shot_count']}镜拼接 → {r['output_path']}")
|
||
|
||
def cmd_batch(episode_id="DSV-EP01", max_parallel="3", retry="false", model=""):
|
||
"""批量提交生成任务: --max-parallel 3 --retry true --model kling|wan"""
|
||
from lib.seedance import batch_submit
|
||
mp = int(max_parallel)
|
||
rf = retry.lower() in ("true", "1", "yes")
|
||
m = (model or "").lower()
|
||
|
||
if m and m != "seedance":
|
||
# 非Seedance模型：逐镜gen，不走batch_submit
|
||
db = get_db()
|
||
where = "episode_id = ? AND status IN ('pending', 'failed')" if rf else "episode_id = ? AND status = 'pending'"
|
||
shots = db.execute(f"SELECT id FROM shots WHERE {where} ORDER BY shot_number", (episode_id,)).fetchall()
|
||
if not shots:
|
||
print("⚠️ 没有待生成的镜头")
|
||
return
|
||
print(f"🎬 批量 {m} 生成 {len(shots)} 镜...")
|
||
for s in shots:
|
||
cmd_gen(s["id"], m)
|
||
return
|
||
|
||
batch_submit(episode_id, max_parallel=mp, retry_failed=rf)
|
||
|
||
def cmd_dub(shot_id, text, character="林昊"):
|
||
"""配音+自动SRT"""
|
||
from lib.tts import tts_for_character, words_to_srt
|
||
audio_path = f"projects/deep-sea-voyage/audio/{shot_id}-{character}.mp3"
|
||
srt_path = f"projects/deep-sea-voyage/audio/{shot_id}-{character}.srt"
|
||
r = tts_for_character(text, character, audio_path)
|
||
if r and r["words"]:
|
||
srt = words_to_srt(r["words"], srt_path)
|
||
print(f"✅ {len(r['words'])}句 → {srt}")
|
||
else:
|
||
print("❌ 配音失败")
|
||
|
||
def cmd_image(mode="t2i", prompt="", ref="", output="", size="1440x2560"):
|
||
"""图像生成: t2i(文生图) / i2i(图生图) / multi(多图融合)"""
|
||
from lib.seedream import seedream_t2i, seedream_i2i, seedream_multi, seedream_download
|
||
|
||
if not prompt:
|
||
print("❌ 缺少提示词 prompt")
|
||
print("用法: python -m lib.cli image t2i \"提示词\" [ref路径] [输出路径] [尺寸]")
|
||
return
|
||
|
||
out = output or f"outputs/images/{mode}_{hash(prompt) & 0xFFFF:04x}.jpg"
|
||
os.makedirs(os.path.dirname(out) or ".", exist_ok=True)
|
||
|
||
print(f"🎨 {mode} 生成中...")
|
||
if mode == "t2i":
|
||
r = seedream_t2i(prompt, size)
|
||
elif mode == "i2i":
|
||
if not ref:
|
||
print("❌ i2i需要参考图路径: image i2i \"提示词\" ref.png")
|
||
return
|
||
r = seedream_i2i(prompt, ref, size)
|
||
elif mode == "multi":
|
||
refs = ref.split(",") if ref else []
|
||
if len(refs) < 2:
|
||
print("❌ multi需要至少2张参考图（逗号分隔）")
|
||
return
|
||
r = seedream_multi(prompt, refs, size)
|
||
else:
|
||
print(f"❌ 未知模式: {mode} (t2i/i2i/multi)")
|
||
return
|
||
|
||
if "error" in r:
|
||
print(f"❌ {r['error']}")
|
||
return
|
||
|
||
p = seedream_download(r, out)
|
||
if p:
|
||
print(f"✅ {p}")
|
||
else:
|
||
print(f"❌ 下载失败")
|
||
|
||
def cmd_qc(shot_id="", video_path="", character=""):
|
||
"""镜头质检: 竖屏/字幕/换脸/牌匾/遮挡/现代物品"""
|
||
import subprocess as _sp, json as _json
|
||
project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
|
||
if not video_path and shot_id:
|
||
db = get_db()
|
||
shot = db.execute("SELECT output_path FROM shots WHERE id = ? AND status = 'done'", (shot_id,)).fetchone()
|
||
if shot and shot["output_path"]:
|
||
video_path = os.path.join(project_root, shot["output_path"])
|
||
|
||
if not video_path or not os.path.exists(video_path):
|
||
print(f"❌ 视频不存在: {video_path or shot_id}")
|
||
return
|
||
|
||
engine = os.path.join(project_root, "engines", "shot-qc-automation.py")
|
||
out = f"{video_path}.qc-report.json"
|
||
cmd = [sys.executable, engine, "--video", video_path, "--output", out]
|
||
if character: cmd += ["--character", character]
|
||
|
||
if not os.path.exists(engine):
|
||
print(f"⚠️ QC引擎未找到: {engine}")
|
||
return
|
||
|
||
print(f"🔍 质检 {os.path.basename(video_path)}...")
|
||
r = _sp.run(cmd, capture_output=True, text=True, timeout=120)
|
||
if r.returncode == 0 and os.path.exists(out):
|
||
report = _json.load(open(out, "r", encoding="utf-8"))
|
||
passed = report.get("passed", False)
|
||
score = report.get("score", 0)
|
||
print(f"{'✅ 通过' if passed else '❌ 未通过'} · 得分: {score:.1f}/10")
|
||
else:
|
||
print(f"⚠️ 质检完成（无JSON报告）: {r.stderr[:200] or r.stdout[:200]}")
|
||
|
||
def cmd_lipsync(shot_id="", video_path="", audio_path="", output=""):
|
||
"""口型同步: 视频+音频 → 口型匹配视频"""
|
||
import subprocess as _sp
|
||
project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
|
||
if not video_path and shot_id:
|
||
db = get_db()
|
||
shot = db.execute("SELECT output_path FROM shots WHERE id = ? AND status = 'done'", (shot_id,)).fetchone()
|
||
if shot and shot["output_path"]:
|
||
video_path = os.path.join(project_root, shot["output_path"])
|
||
|
||
if not video_path or not audio_path:
|
||
print("❌ 用法: lipsync DSV-EP01-S01 配音.mp3")
|
||
return
|
||
|
||
engine = os.path.join(project_root, "engines", "lipsync-adapter.py")
|
||
if not os.path.exists(engine):
|
||
print(f"⚠️ 口型引擎未找到，跳过")
|
||
return
|
||
|
||
out = output or video_path.replace(".mp4", "-synced.mp4")
|
||
print(f"👄 口型同步...")
|
||
r = _sp.run([sys.executable, engine, "--video", video_path, "--audio", audio_path, "--output", out],
|
||
capture_output=True, text=True, timeout=300)
|
||
if r.returncode == 0: print(f"✅ {out}")
|
||
else: print(f"⚠️ 口型同步失败: {r.stderr[:200]}")
|
||
|
||
def cmd_mix(video_path="", dialogue="", bgm="", output=""):
|
||
"""音频混音: 对白+BGM → 混音后合并视频"""
|
||
import subprocess as _sp, shutil
|
||
project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
|
||
if not video_path or not dialogue:
|
||
print("❌ 用法: mix video.mp4 dialogue.mp3 [bgm.mp3]")
|
||
return
|
||
|
||
engine = os.path.join(project_root, "engines", "audio-mixer.py")
|
||
out = output or video_path.replace(".mp4", "-mixed.mp4")
|
||
mixed_audio = out.replace(".mp4", ".mp3")
|
||
|
||
if os.path.exists(engine):
|
||
cmd = [sys.executable, engine, "--dialogue", dialogue, "--output", mixed_audio]
|
||
if bgm and os.path.exists(bgm): cmd += ["--bgm", bgm]
|
||
print(f"🔊 混音...")
|
||
r = _sp.run(cmd, capture_output=True, text=True, timeout=60)
|
||
if r.returncode != 0:
|
||
print(f"⚠️ 引擎混音失败，用兜底方案: {r.stderr[:100]}")
|
||
from lib.tts import mix_audio_video
|
||
r2 = mix_audio_video(video_path, dialogue, out)
|
||
print(f"{'✅' if isinstance(r2, str) else '❌'} {r2}")
|
||
return
|
||
else:
|
||
from lib.tts import mix_audio_video
|
||
r2 = mix_audio_video(video_path, dialogue, out)
|
||
print(f"{'✅' if isinstance(r2, str) else '❌'} {r2}")
|
||
return
|
||
|
||
# 合并视频+混音
|
||
ffmpeg = shutil.which("ffmpeg") or os.path.expanduser("~/.workbuddy/binaries/ffmpeg.exe")
|
||
if os.path.exists(ffmpeg) and os.path.exists(mixed_audio):
|
||
r3 = _sp.run([ffmpeg, "-y", "-i", video_path, "-i", mixed_audio,
|
||
"-c:v", "copy", "-c:a", "aac", "-b:a", "192k", "-shortest",
|
||
"-map", "0:v:0", "-map", "1:a:0", out],
|
||
capture_output=True, text=True, timeout=60)
|
||
print(f"{'✅' if r3.returncode == 0 else '❌'} {out}")
|
||
|
||
def cmd_emotion(text="", emotion="", output=""):
|
||
"""情感配音: 角色·情感·强度 → 带情感的TTS"""
|
||
import subprocess as _sp
|
||
project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
|
||
if not text or not emotion:
|
||
print("❌ 用法: emotion \"台词\" \"苏白·大声·自信\" [output.mp3]")
|
||
print("情感格式: 角色名·情绪·强度 (如: 苏白·大声·自信)")
|
||
return
|
||
|
||
engine = os.path.join(project_root, "engines", "voice-emotion-compiler.py")
|
||
out = output or f"outputs/audio/emotion_{hash(text) & 0xFFFF:04x}.mp3"
|
||
os.makedirs(os.path.dirname(out) or ".", exist_ok=True)
|
||
|
||
if os.path.exists(engine):
|
||
print(f"🎭 情感配音: {emotion}")
|
||
r = _sp.run([sys.executable, engine, "--text", text, "--emotion", emotion, "--output", out],
|
||
capture_output=True, text=True, timeout=60)
|
||
if r.returncode == 0: print(f"✅ {out}")
|
||
else: print(f"❌ 情感配音失败: {r.stderr[:200]}")
|
||
else:
|
||
# 兜底: 用普通TTS
|
||
from lib.tts import tts
|
||
r = tts(text, out)
|
||
print(f"⚠️ 情感引擎未找到，用普通TTS兜底: {out}")
|
||
|
||
def cmd_subtitles(srt_path="", video_path="", output="", style="short-drama-bold"):
|
||
"""字幕渲染: SRT → 视频烧录（5种样式）"""
|
||
import subprocess as _sp
|
||
project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
|
||
if not srt_path or not video_path:
|
||
print("❌ 用法: subtitles input.srt video.mp4 [output] [style]")
|
||
print("样式: reference-drama / short-drama-bold / clean-white / black-box / outlined-white")
|
||
return
|
||
|
||
engine = os.path.join(project_root, "engines", "subtitle-renderer.py")
|
||
out = output or video_path.replace(".mp4", "-subtitled.mp4")
|
||
|
||
if os.path.exists(engine):
|
||
print(f"📝 字幕渲染 ({style})...")
|
||
r = _sp.run([sys.executable, engine, "--srt", srt_path, "--video", video_path,
|
||
"--output", out, "--style", style],
|
||
capture_output=True, text=True, timeout=120)
|
||
if r.returncode == 0: print(f"✅ {out}")
|
||
else: print(f"⚠️ 字幕引擎失败: {r.stderr[:200]}")
|
||
else:
|
||
# 兜底: FFmpeg subtitles filter
|
||
import shutil
|
||
ffmpeg = shutil.which("ffmpeg") or os.path.expanduser("~/.workbuddy/binaries/ffmpeg.exe")
|
||
if os.path.exists(ffmpeg):
|
||
print(f"📝 FFmpeg字幕烧录...")
|
||
r = _sp.run([ffmpeg, "-y", "-i", video_path,
|
||
"-vf", f"subtitles='{srt_path}':force_style='Fontsize=24,Alignment=2'",
|
||
"-c:a", "copy", out],
|
||
capture_output=True, text=True, timeout=60)
|
||
print(f"{'✅' if r.returncode == 0 else '❌'} {out}")
|
||
else:
|
||
print("❌ ffmpeg未找到")
|
||
|
||
def cmd_track(mode="track", video_path="", output=""):
|
||
"""平面追踪: track(特征追踪) / stabilize(稳定化) / pin(文字贴合) / motion(运动数据导出)"""
|
||
import subprocess as _sp
|
||
project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
|
||
if not video_path:
|
||
print("❌ 用法: track track|stabilize|pin|motion video.mp4")
|
||
return
|
||
|
||
engine = os.path.join(project_root, "engines", "planar-tracker.py")
|
||
if not os.path.exists(engine):
|
||
print(f"⚠️ 追踪引擎未找到: {engine}")
|
||
return
|
||
|
||
out = output or f"{video_path}.track-{mode}.json"
|
||
print(f"🎯 平面追踪 ({mode})...")
|
||
r = _sp.run([sys.executable, engine, mode, video_path, "--output", out],
|
||
capture_output=True, text=True, timeout=120)
|
||
if r.returncode == 0:
|
||
print(f"✅ {out}")
|
||
else:
|
||
print(f"❌ {r.stderr[:200]}")
|
||
|
||
def cmd_director(script_path="", shots_dir="", output=""):
|
||
"""自动导演: 分镜→匹配镜头→追踪→分轨→时间线→成片"""
|
||
import subprocess as _sp, json as _json
|
||
project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
|
||
if not script_path or not shots_dir:
|
||
print("❌ 用法: director 分镜.json 镜头目录/ [output.mp4]")
|
||
return
|
||
|
||
engine = os.path.join(project_root, "engines", "auto-cut-director.js")
|
||
out = output or f"outputs/episodes/director-{os.urandom(3).hex()}.mp4"
|
||
os.makedirs(os.path.dirname(out) or ".", exist_ok=True)
|
||
|
||
if not os.path.exists(engine):
|
||
print(f"⚠️ 导演引擎未找到，用 render 兜底")
|
||
# 兜底：简单的concat
|
||
result = {"shots_dir": shots_dir, "script": script_path, "note": "director engine not found, use 'render' instead"}
|
||
print(f"💡 请改用: python -m lib.cli render")
|
||
return
|
||
|
||
# 构造 autoCut 配置
|
||
cfg = {
|
||
"script": script_path,
|
||
"shots": shots_dir,
|
||
"output": out,
|
||
"resolution": {"w": 1080, "h": 1920},
|
||
"fps": 24,
|
||
"stabilize": False,
|
||
"autoMix": True
|
||
}
|
||
cfg_json = out + ".director-cfg.json"
|
||
with open(cfg_json, "w", encoding="utf-8") as f:
|
||
_json.dump(cfg, f, ensure_ascii=False)
|
||
|
||
print(f"🎬 自动导演...")
|
||
r = _sp.run(["node", "-e",
|
||
f"const acd=require('{engine.replace(chr(92),'/')}');"
|
||
f"const cfg=require('{cfg_json.replace(chr(92),'/')}');"
|
||
f"acd.autoCut(cfg).then(r=>{{console.log(JSON.stringify(r));process.exit(0);}})"
|
||
f".catch(e=>{{console.error(e.message);process.exit(1);}})"],
|
||
capture_output=True, text=True, timeout=600)
|
||
|
||
if os.path.exists(cfg_json): os.unlink(cfg_json)
|
||
|
||
if r.returncode == 0:
|
||
try:
|
||
result = _json.loads(r.stdout)
|
||
print(f"✅ {result.get('outputPath', out)}")
|
||
print(f" 时长: {result.get('duration', '?')}s")
|
||
except:
|
||
print(f"✅ {out}")
|
||
else:
|
||
print(f"❌ 导演失败: {r.stderr[:300]}")
|
||
print(f"💡 请改用: python -m lib.cli render")
|
||
|
||
def cmd_pack(character="CHAR-003-SuBai", *extra_args):
|
||
"""角色资产生成: 多视角角色定妆照 → 自动批准入库
|
||
args: <角色ID> [--project <项目>] [--views front,side,hand] [--dry-run]
|
||
v2.0: 支持主视角→参考图链路 + seed锁定"""
|
||
import subprocess as _sp
|
||
project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
|
||
# 检测引擎类型 (.js vs .py)
|
||
engine_js = os.path.join(project_root, "engines", "char-hero-design-packer", "char-hero-design-packer.js")
|
||
engine_py = os.path.join(project_root, "engines", "char-hero-design-packer", "char-hero-design-packer.py")
|
||
|
||
engine = None
|
||
runtime = None
|
||
if os.path.exists(engine_js):
|
||
engine = engine_js
|
||
runtime = "node"
|
||
elif os.path.exists(engine_py):
|
||
engine = engine_py
|
||
runtime = "python"
|
||
|
||
if not engine:
|
||
print(f"⚠️ 资产打包引擎未找到 (char-hero-design-packer.js/.py)")
|
||
print(f"💡 请用: python -m lib.cli image t2i \"角色定妆照描述\"")
|
||
return
|
||
|
||
# 解析参数
|
||
args = list(extra_args)
|
||
project = None
|
||
views = None
|
||
breakdown = None
|
||
dry_run = False
|
||
|
||
i = 0
|
||
while i < len(args):
|
||
if args[i] == "--project" and i+1 < len(args):
|
||
project = args[i+1]; i += 2
|
||
elif args[i] == "--views" and i+1 < len(args):
|
||
views = args[i+1]; i += 2
|
||
elif args[i] == "--breakdown" and i+1 < len(args):
|
||
breakdown = args[i+1]; i += 2
|
||
elif args[i] in ("--dry-run", "--dry_run", "true"):
|
||
dry_run = True; i += 1
|
||
else:
|
||
i += 1
|
||
|
||
if not project:
|
||
# 尝试从角色ID推断项目
|
||
import glob
|
||
proj_dirs = glob.glob(os.path.join(project_root, "projects", "*"))
|
||
for pd in proj_dirs:
|
||
breakdowns = glob.glob(os.path.join(pd, "*BREAKDOWN*V*.hdlp"))
|
||
for bd in breakdowns:
|
||
with open(bd, "r", encoding="utf-8") as f:
|
||
if character in f.read():
|
||
project = os.path.basename(pd)
|
||
print(f"🔍 自动检测项目: {project}")
|
||
break
|
||
if project: break
|
||
|
||
engine_args = ["--character", character, "--generate-all"]
|
||
engine_args.append("--project"); engine_args.append(project or "default")
|
||
if views: engine_args.append("--views"); engine_args.append(views)
|
||
if breakdown: engine_args.append("--breakdown"); engine_args.append(breakdown)
|
||
if dry_run: engine_args.append("--dry-run")
|
||
|
||
print(f"🎒 打包角色资产: {character}")
|
||
print(f" 项目: {project or '(未指定)'}")
|
||
print(f" 引擎: {os.path.basename(engine)} ({runtime})")
|
||
if views: print(f" 视角: {views}")
|
||
if dry_run: print(f" 🔍 干运行模式")
|
||
print()
|
||
|
||
if runtime == "node":
|
||
r = _sp.run(["C:/Users/rtyyr/.workbuddy/binaries/node/versions/22.22.2/node.exe", engine] + engine_args,
|
||
capture_output=True, text=True, timeout=600)
|
||
else:
|
||
r = _sp.run([sys.executable, engine] + engine_args,
|
||
capture_output=True, text=True, timeout=600)
|
||
|
||
if r.stderr:
|
||
for line in r.stderr.strip().split("\n"):
|
||
print(f" {line}")
|
||
|
||
if r.returncode == 0:
|
||
if not dry_run:
|
||
print(f"\n✅ {character} 角色资产打包完成")
|
||
try:
|
||
db = get_db()
|
||
db.execute("UPDATE assets SET status = 'approved', approved_at = datetime('now') WHERE name LIKE ?",
|
||
(f"%{character}%",))
|
||
db.commit()
|
||
except: pass
|
||
else:
|
||
print(f"\n❌ 打包失败: {r.stderr[:300] or r.stdout[:300]}")
|
||
|
||
def cmd_prompt(scene_text="", output=""):
|
||
"""HLDP提示词: 场景描述 → Seedance提示词（HLDP引擎/AI兜底）"""
|
||
import subprocess as _sp, json as _json
|
||
project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
|
||
if not scene_text:
|
||
print("❌ 用法: prompt \"场景描述\"")
|
||
print("示例: prompt \"苏白站在天道宗牌匾下，自信地喊话招生\"")
|
||
return
|
||
|
||
engine = os.path.join(project_root, "engines", "hldp-prompt.js")
|
||
if os.path.exists(engine):
|
||
print(f"🎯 HLDP提示词引擎...")
|
||
r = _sp.run(["node", "-e",
|
||
f"const hp=require('{engine.replace(chr(92),'/')}');"
|
||
f"const r=hp.understandScene({{description:'{scene_text}'}});"
|
||
f"console.log(JSON.stringify(r,null,2));"],
|
||
capture_output=True, text=True, timeout=30)
|
||
if r.returncode == 0 and r.stdout.strip():
|
||
try:
|
||
data = _json.loads(r.stdout)
|
||
for k, v in data.items() if isinstance(data, dict) else []:
|
||
if isinstance(v, str) and len(v) < 200:
|
||
print(f" {k}: {v}")
|
||
if not isinstance(data, dict):
|
||
print(r.stdout[:500])
|
||
except:
|
||
print(r.stdout[:500])
|
||
return
|
||
|
||
# 兜底: 豆包AI
|
||
from lib.doubao_chat import generate_keyframes
|
||
print(f"🤖 AI兜底...")
|
||
r = generate_keyframes(scene_text, [], [])
|
||
if "content" in r: print(r["content"])
|
||
else: print(f"❌ {r}")
|
||
|
||
def cmd_produce(shot_id="", dry_run="false"):
|
||
"""单镜一站式生产: 资产→底片→配音→口型→字幕→混音→质检→报告"""
|
||
import subprocess as _sp
|
||
|
||
if not shot_id:
|
||
print("❌ 用法: produce <shot_id> [--dry-run]")
|
||
print("示例: produce E1-SHOT03")
|
||
return
|
||
|
||
project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
engine = os.path.join(project_root, "engines", "ep01_shot03_production.py")
|
||
|
||
if not os.path.exists(engine):
|
||
print(f"⚠️ 生产编排引擎未找到")
|
||
print(f"💡 手工分步: image → gen → dub → lipsync → subtitles → mix → qc → render")
|
||
return
|
||
|
||
is_dry = dry_run.lower() in ("true", "1", "yes", "--dry-run")
|
||
flag = "--dry-run" if is_dry else "--run"
|
||
|
||
print(f"🎬 单镜生产: {shot_id}")
|
||
print(f" 步骤: 准备资产 → 生成底片 → 合成特效 → 配音 → 口型 → 字幕 → 混音 → 质检 → 报告")
|
||
if is_dry: print(f" 🔍 DRY RUN模式（不实际执行）")
|
||
|
||
r = _sp.run([sys.executable, engine, flag], capture_output=True, text=True, timeout=900)
|
||
print(r.stdout[-2000:] if len(r.stdout) > 2000 else r.stdout)
|
||
if r.returncode != 0:
|
||
print(f"❌ 生产失败: {r.stderr[:500]}")
|
||
|
||
def cmd_srt_timing(script_path="", audio_dir="", output=""):
|
||
"""剧本→SRT时间轴: 从剧本JSON+配音文件自动生成带时间戳的字幕"""
|
||
import subprocess as _sp
|
||
project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
|
||
if not script_path:
|
||
print("❌ 用法: srt-timing 剧本.json 配音目录/ [输出.srt]")
|
||
print("剧本格式: JSON(含台词) / MD(含对话) / TXT(纯文本)")
|
||
return
|
||
|
||
engine = os.path.join(project_root, "engines", "subtitle-pipeline", "srt-tools", "script-to-srt-with-timing.py")
|
||
out = output or script_path.replace(".json", ".srt").replace(".md", ".srt").replace(".txt", ".srt")
|
||
|
||
if os.path.exists(engine):
|
||
print(f"⏱ 生成字幕时间轴...")
|
||
cmd = [sys.executable, engine, "--script", script_path, "--output", out]
|
||
if audio_dir: cmd += ["--audio-dir", audio_dir]
|
||
r = _sp.run(cmd, capture_output=True, text=True, timeout=60)
|
||
if r.returncode == 0: print(f"✅ {out}")
|
||
else: print(f"❌ {r.stderr[:300]}")
|
||
else:
|
||
# 兜底: 用 lib/tts.py 从文本生成简单SRT
|
||
print(f"⚠️ 字幕引擎未找到，用简单兜底方案")
|
||
with open(script_path, "r", encoding="utf-8") as f:
|
||
text = f.read()
|
||
lines = [l.strip() for l in text.split("\n") if l.strip() and not l.startswith("#")][:50]
|
||
from lib.tts import words_to_srt
|
||
# 每行假设3秒
|
||
words = [{"text": line, "start": i*3, "end": (i+1)*3} for i, line in enumerate(lines)]
|
||
words_to_srt(words, out)
|
||
print(f"✅ {out} (简单时间轴，每句3秒)")
|
||
|
||
def cmd_font(action="check"):
|
||
"""字幕字体管理: check(检测) / list(列表) / install(安装)"""
|
||
import subprocess as _sp
|
||
project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
engine = os.path.join(project_root, "engines", "subtitle-pipeline", "font-manager", "subtitle-font-manager.py")
|
||
|
||
if not os.path.exists(engine):
|
||
print("⚠️ 字体管理器未找到")
|
||
print("💡 推荐中文字体: PingFang SC / Noto Sans SC / Source Han Sans")
|
||
return
|
||
|
||
if action in ("check", "list"):
|
||
print(f"🔤 检查字体...")
|
||
r = _sp.run([sys.executable, engine, "--check"], capture_output=True, text=True, timeout=30)
|
||
print(r.stdout[:1000] if r.stdout else r.stderr[:300])
|
||
elif action == "install":
|
||
print(f"📥 安装中文字体...")
|
||
r = _sp.run([sys.executable, engine, "--install"], capture_output=True, text=True, timeout=60)
|
||
print(r.stdout[:500] or "安装完成" if r.returncode == 0 else f"❌ {r.stderr[:200]}")
|
||
|
||
def cmd_split_audio(video_path="", output_dir=""):
|
||
"""音频分轨: 视频 → 分离对白/BGM/音效三轨"""
|
||
import subprocess as _sp
|
||
project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
|
||
if not video_path:
|
||
print("❌ 用法: split-audio video.mp4 [输出目录]")
|
||
return
|
||
|
||
engine = os.path.join(project_root, "engines", "audio-stem-splitter.js")
|
||
out_dir = output_dir or f"{video_path}.stems"
|
||
|
||
if os.path.exists(engine):
|
||
print(f"🎵 音频分轨...")
|
||
r = _sp.run(["node", "-e",
|
||
f"const sas=require('{engine.replace(chr(92),'/')}');"
|
||
f"sas.splitAudioStems('{video_path.replace(chr(92),'/')}',"
|
||
f"{{outputDir:'{out_dir.replace(chr(92),'/')}'}})"
|
||
f".then(r=>{{console.log(JSON.stringify(r));process.exit(0);}})"
|
||
f".catch(e=>{{console.error(e.message);process.exit(1);}})"],
|
||
capture_output=True, text=True, timeout=120)
|
||
if r.returncode == 0:
|
||
stems = ["voice.wav", "bgm.wav", "sfx.wav"]
|
||
found = [s for s in stems if os.path.exists(os.path.join(out_dir, s))]
|
||
print(f"✅ 分轨完成: {', '.join(found) if found else out_dir}")
|
||
else:
|
||
print(f"❌ {r.stderr[:300]}")
|
||
else:
|
||
print(f"⚠️ 音频分轨引擎未找到")
|
||
print(f"💡 用FFmpeg手动分轨: ffmpeg -i video.mp4 -vn voice.wav")
|
||
|
||
def cmd_char_qc(image_path="", character="", video_path=""):
|
||
"""角色质检: 角色存在感评分(0-10) / 跨镜一致性对比"""
|
||
import subprocess as _sp
|
||
project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
engine = os.path.join(project_root, "engines", "character-distinctiveness-qc", "character-distinctiveness-qc.py")
|
||
|
||
if not os.path.exists(engine):
|
||
print(f"⚠️ 角色QC引擎未找到")
|
||
return
|
||
|
||
if image_path and character:
|
||
print(f"🔍 角色质检: {character}")
|
||
r = _sp.run([sys.executable, engine, "--image", image_path, "--character", character],
|
||
capture_output=True, text=True, timeout=60)
|
||
print(r.stdout[:500] if r.stdout else f"❌ {r.stderr[:200]}")
|
||
elif video_path and character:
|
||
print(f"🔍 角色跨镜一致性: {character}")
|
||
r = _sp.run([sys.executable, engine, "--batch", video_path, "--character", character],
|
||
capture_output=True, text=True, timeout=120)
|
||
print(r.stdout[:500] if r.stdout else f"❌ {r.stderr[:200]}")
|
||
else:
|
||
print("❌ 用法: char-qc --image 角色图.png --character CHAR-003-SuBai")
|
||
print(" char-qc --video 镜头目录/ --character CHAR-003-SuBai")
|
||
|
||
def cmd_eye(video_path="", director_encoding=""):
|
||
"""铸渊之眼: 视频逐帧分析 → 与导演编码对比 → 差异报告"""
|
||
import subprocess as _sp
|
||
project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
|
||
if not video_path:
|
||
print("❌ 用法: eye video.mp4 [导演编码.json]")
|
||
return
|
||
|
||
engine = os.path.join(project_root, "engines", "zhuyuan-eye.js")
|
||
cmd = ["node", engine, video_path]
|
||
if director_encoding: cmd.append(director_encoding)
|
||
|
||
if os.path.exists(engine):
|
||
print(f"👁 铸渊之眼分析...")
|
||
r = _sp.run(cmd, capture_output=True, text=True, timeout=120)
|
||
print(r.stdout[:1000] if r.stdout else f"❌ {r.stderr[:300]}")
|
||
else:
|
||
print(f"⚠️ 铸渊之眼未找到")
|
||
|
||
def cmd_multi_ref(prompt="", references="", output="", action="check"):
|
||
"""多参考视频适配: check(API探测) / gen(多参考生成)"""
|
||
import subprocess as _sp
|
||
project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
engine = os.path.join(project_root, "engines", "multi-reference-video-adapter.py")
|
||
|
||
if not os.path.exists(engine):
|
||
print(f"⚠️ 多参考适配器未找到")
|
||
return
|
||
|
||
if action == "check":
|
||
print(f"🔍 探测API能力...")
|
||
r = _sp.run([sys.executable, engine, "--check-api"], capture_output=True, text=True, timeout=30)
|
||
print(r.stdout[:1000] if r.stdout else r.stderr[:300])
|
||
elif action == "gen":
|
||
if not prompt or not references:
|
||
print("❌ 用法: multi-ref gen \"提示词\" \"ref1.png,ref2.png\" [output.mp4]")
|
||
return
|
||
refs = references.split(",")
|
||
out = output or f"outputs/videos/multi-ref-{hash(prompt) & 0xFFFF:04x}.mp4"
|
||
cmd = [sys.executable, engine, "--prompt", prompt, "--references"] + refs + ["--output", out]
|
||
print(f"🎬 多参考生成 ({len(refs)}张参考图)...")
|
||
r = _sp.run(cmd, capture_output=True, text=True, timeout=120)
|
||
if r.returncode == 0: print(f"✅ {out}")
|
||
else: print(f"❌ {r.stderr[:300]}")
|
||
else:
|
||
print("❌ 用法: multi-ref check|gen")
|
||
|
||
def cmd_parse(script_path="", output=""):
|
||
"""脚本解析: 原始剧本→结构化JSON (确定性解析，不调AI)"""
|
||
import subprocess as _sp, json as _json
|
||
project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
|
||
if not script_path:
|
||
print("❌ 用法: parse 剧本.txt [输出.json]")
|
||
print("支持格式: 腾讯文档(第N集/人物/△描述/名字(情绪):台词)")
|
||
return
|
||
|
||
engine = os.path.join(project_root, "engines", "script-parser.js")
|
||
out = output or script_path.replace(".txt", ".json").replace(".md", ".json")
|
||
|
||
if os.path.exists(engine) and os.path.exists(script_path):
|
||
with open(script_path, "r", encoding="utf-8") as f:
|
||
script_text = f.read()[:50000]
|
||
print(f"📖 解析剧本...")
|
||
r = _sp.run(["node", "-e",
|
||
f"const sp=require('{engine.replace(chr(92),'/')}');"
|
||
f"const r=sp.parseScript(`{script_text.replace('`','\\`').replace('$','\\$')}`);"
|
||
f"console.log(JSON.stringify(r,null,2));"],
|
||
capture_output=True, text=True, timeout=30)
|
||
if r.returncode == 0 and r.stdout.strip():
|
||
try:
|
||
data = _json.loads(r.stdout)
|
||
with open(out, "w", encoding="utf-8") as f:
|
||
_json.dump(data, f, ensure_ascii=False, indent=2)
|
||
stats = data.get("stats", {})
|
||
eps = data.get("episodes", [])
|
||
scenes = sum(len(ep.get("scenes", [])) for ep in eps)
|
||
chars = len(stats.get("characters", []))
|
||
print(f"✅ {out}")
|
||
print(f" {len(eps)}集 {scenes}场 {chars}个角色")
|
||
except:
|
||
print(f"⚠️ 解析结果不是JSON，直接输出:")
|
||
print(r.stdout[:500])
|
||
else:
|
||
# 兜底用AI
|
||
print(f"⚠️ 引擎解析失败，用AI拆解...")
|
||
from lib.doubao_chat import breakdown_script
|
||
with open(script_path, "r", encoding="utf-8") as f:
|
||
text = f.read()
|
||
r = breakdown_script(text)
|
||
if "content" in r:
|
||
with open(out, "w", encoding="utf-8") as f:
|
||
f.write(r["content"])
|
||
print(f"✅ {out} (AI拆解)")
|
||
else:
|
||
print(f"❌ {r}")
|
||
else:
|
||
# 兜底AI
|
||
from lib.doubao_chat import breakdown_script
|
||
with open(script_path, "r", encoding="utf-8") as f:
|
||
text = f.read()
|
||
print(f"🤖 AI拆解...")
|
||
r = breakdown_script(text)
|
||
if "content" in r:
|
||
with open(out, "w", encoding="utf-8") as f:
|
||
f.write(r["content"])
|
||
print(f"✅ {out}")
|
||
|
||
def cmd_adapt(director_json="", output=""):
|
||
"""导演编码适配: 导演JSON→video-editor.js编辑参数"""
|
||
import subprocess as _sp, json as _json
|
||
project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
|
||
if not director_json:
|
||
print("❌ 用法: adapt 导演编码.json [输出编辑配置.json]")
|
||
return
|
||
|
||
engine = os.path.join(project_root, "engines", "hldp-director-adapter.js")
|
||
out = output or director_json.replace(".json", "-editor-cfg.json")
|
||
|
||
if os.path.exists(engine):
|
||
print(f"🎬 导演编码→编辑参数...")
|
||
r = _sp.run(["node", "-e",
|
||
f"const adapt=require('{engine.replace(chr(92),'/')}').adapt;"
|
||
f"const cfg=require('{director_json.replace(chr(92),'/')}');"
|
||
f"console.log(JSON.stringify(adapt(cfg),null,2));"],
|
||
capture_output=True, text=True, timeout=30)
|
||
if r.returncode == 0:
|
||
with open(out, "w", encoding="utf-8") as f:
|
||
f.write(r.stdout)
|
||
print(f"✅ {out}")
|
||
else:
|
||
print(f"❌ {r.stderr[:300]}")
|
||
else:
|
||
print(f"⚠️ 导演适配器未找到")
|
||
|
||
def cmd_batch_dub(episode_id="DSV-EP01", character=""):
|
||
"""批量配音: 为剧集所有镜头批量配音"""
|
||
from lib.tts import tts_for_character
|
||
db = get_db()
|
||
|
||
# 从剧本找台词（简化版：从shots表的script_ref和description拼接）
|
||
shots = db.execute("""
|
||
SELECT id, shot_number, description, script_ref FROM shots
|
||
WHERE episode_id = ? ORDER BY shot_number
|
||
""", (episode_id,)).fetchall()
|
||
|
||
if not shots:
|
||
print(f"❌ 没有镜头，先 sync-shots")
|
||
return
|
||
|
||
print(f"🎭 批量配音 {len(shots)} 镜")
|
||
if not character:
|
||
print("💡 未指定角色，将使用默认林昊音色")
|
||
character = "林昊"
|
||
|
||
for s in shots:
|
||
sid = s["id"]
|
||
# 从description提取可能的台词
|
||
desc = s["description"] or s["script_ref"] or f"镜头{s['shot_number']}"
|
||
text = desc.split("：")[-1].split("说")[-1].strip() if ("：" in desc or "说" in desc) else desc[:60]
|
||
|
||
audio_path = f"outputs/audio/{sid}-{character}.mp3"
|
||
srt_path = f"outputs/audio/{sid}-{character}.srt"
|
||
|
||
r = tts_for_character(text, character, audio_path)
|
||
if r and r.get("words"):
|
||
from lib.tts import words_to_srt
|
||
words_to_srt(r["words"], srt_path)
|
||
print(f" ✅ {sid}: {text[:30]}...")
|
||
else:
|
||
print(f" ❌ {sid}")
|
||
|
||
def cmd_batch_qc(episode_id="DSV-EP01"):
|
||
"""批量质检: 对剧集所有已完成镜头逐一QC"""
|
||
import subprocess as _sp
|
||
project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
engine = os.path.join(project_root, "engines", "shot-qc-automation.py")
|
||
|
||
db = get_db()
|
||
shots = db.execute("""
|
||
SELECT id, shot_number, output_path FROM shots
|
||
WHERE episode_id = ? AND status = 'done' AND output_path IS NOT NULL
|
||
ORDER BY shot_number
|
||
""", (episode_id,)).fetchall()
|
||
|
||
if not shots:
|
||
print(f"❌ 没有已完成的镜头")
|
||
return
|
||
|
||
print(f"🔍 批量质检 {len(shots)} 镜\n")
|
||
results = []
|
||
|
||
for s in shots:
|
||
video_path = os.path.join(project_root, s["output_path"])
|
||
if not os.path.exists(video_path):
|
||
print(f" ⚠️ {s['id']}: 视频文件不存在")
|
||
continue
|
||
|
||
out = f"{video_path}.qc-report.json"
|
||
r = _sp.run([sys.executable, engine, "--video", video_path, "--output", out],
|
||
capture_output=True, text=True, timeout=120)
|
||
|
||
if r.returncode == 0 and os.path.exists(out):
|
||
import json as _json
|
||
with open(out, "r", encoding="utf-8") as f:
|
||
report = _json.load(f)
|
||
passed = report.get("passed", False)
|
||
score = report.get("score", 0)
|
||
icon = "✅" if passed else "❌"
|
||
results.append({"shot": s["id"], "passed": passed, "score": score})
|
||
print(f" {icon} {s['id']}: {score:.1f}/10")
|
||
else:
|
||
print(f" ❌ {s['id']}: QC失败")
|
||
|
||
passed = sum(1 for r in results if r["passed"])
|
||
avg = sum(r["score"] for r in results) / len(results) if results else 0
|
||
print(f"\n📊 {passed}/{len(results)} 通过 · 均分: {avg:.1f}/10")
|
||
|
||
def cmd_env():
|
||
"""环境检测: 我在哪 / 有什么 / 能做什么"""
|
||
import shutil, socket
|
||
|
||
print(f"\n{'='*50}")
|
||
print(f"🖥 环境检测")
|
||
print(f"{'='*50}")
|
||
|
||
# 主机
|
||
hostname = socket.gethostname()
|
||
print(f"\n📍 主机: {hostname}")
|
||
print(f"🐍 Python: {sys.version.split()[0]}")
|
||
|
||
# 目录
|
||
project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
print(f"📁 项目: {project_root}")
|
||
|
||
# FFmpeg
|
||
ffmpeg = shutil.which("ffmpeg") or os.path.expanduser("~/.workbuddy/binaries/ffmpeg.exe")
|
||
print(f"🎬 FFmpeg: {'✅ ' + ffmpeg if os.path.exists(ffmpeg) else '❌ 未找到'}")
|
||
|
||
# Node
|
||
node = shutil.which("node")
|
||
print(f"📦 Node: {'✅ ' + node if node else '❌ 未找到'}")
|
||
|
||
# 引擎数量
|
||
engines_dir = os.path.join(project_root, "engines")
|
||
if os.path.exists(engines_dir):
|
||
py = len([f for f in os.listdir(engines_dir) if f.endswith(".py")])
|
||
js = len([f for f in os.listdir(engines_dir) if f.endswith(".js")])
|
||
print(f"⚙ 引擎: {py}个Python + {js}个Node")
|
||
|
||
# API密钥
|
||
try:
|
||
from lib.secrets import get_ark_key
|
||
get_ark_key()
|
||
print(f"🔑 API密钥: ✅ 已配置")
|
||
except:
|
||
print(f"🔑 API密钥: ❌ 未设置 ARK_API_KEY")
|
||
|
||
# 数据库
|
||
db_path = os.path.join(os.path.dirname(__file__), "lib.db")
|
||
if os.path.exists(db_path):
|
||
db = get_db()
|
||
projs = db.execute("SELECT COUNT(*) FROM projects").fetchone()[0]
|
||
shots_total = db.execute("SELECT COUNT(*) FROM shots").fetchone()[0]
|
||
shots_done = db.execute("SELECT COUNT(*) FROM shots WHERE status = 'done'").fetchone()[0]
|
||
print(f"🗄 数据库: {projs}项目 {shots_total}镜({shots_done}完成)")
|
||
|
||
print(f"\n{'='*50}")
|
||
|
||
def cmd_config(section=""):
|
||
"""查看配置: agents(智能体) / voices(音色) / models(模型)"""
|
||
project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
|
||
if not section or section == "agents":
|
||
agents_dir = os.path.join(project_root, "config", "agents")
|
||
if os.path.exists(agents_dir):
|
||
agents = sorted([f for f in os.listdir(agents_dir) if f.endswith(".hdlp")])
|
||
print(f"\n🤖 智能体 ({len(agents)}个):")
|
||
for a in agents:
|
||
print(f" {a}")
|
||
|
||
if not section or section == "voices":
|
||
try:
|
||
from lib.tts import VOICES
|
||
print(f"\n🎭 音色表 ({len(VOICES)}个):")
|
||
for name, voice in sorted(VOICES.items()):
|
||
print(f" {name:8s} → {voice}")
|
||
except:
|
||
pass
|
||
|
||
if not section or section == "models":
|
||
from lib.doubao_chat import MODELS
|
||
print(f"\n🧠 模型:")
|
||
for k, v in MODELS.items():
|
||
print(f" {k:10s} → {v}")
|
||
|
||
if section and section not in ("agents", "voices", "models"):
|
||
print(f"❌ 未知配置区: {section} (agents/voices/models)")
|
||
|
||
def cmd_backup():
|
||
"""数据库备份: 复制lib.db到带时间戳的备份文件"""
|
||
import shutil
|
||
from datetime import datetime
|
||
|
||
db_path = os.path.join(os.path.dirname(__file__), "lib.db")
|
||
if not os.path.exists(db_path):
|
||
print("❌ 数据库不存在，先 init")
|
||
return
|
||
|
||
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||
backup_dir = os.path.join(os.path.dirname(os.path.dirname(__file__)), "outputs", "backups")
|
||
os.makedirs(backup_dir, exist_ok=True)
|
||
backup_path = os.path.join(backup_dir, f"lib-{ts}.db")
|
||
|
||
shutil.copy2(db_path, backup_path)
|
||
size = os.path.getsize(backup_path)
|
||
print(f"✅ 备份: {backup_path} ({size/1024:.0f}KB)")
|
||
|
||
# 清理旧备份（保留最近10个）
|
||
backups = sorted([f for f in os.listdir(backup_dir) if f.startswith("lib-")])
|
||
for old in backups[:-10]:
|
||
os.remove(os.path.join(backup_dir, old))
|
||
|
||
def cmd_workflow(name="", episode_id="DSV-EP01"):
|
||
"""工作流模板: make-ep(制作一集)"""
|
||
import subprocess as _sp
|
||
workflows = {"make-ep": "制作一集: parse-sync-batch-dub-qc-render"}
|
||
if not name:
|
||
print("📋 可用工作流:")
|
||
for k, v in workflows.items(): print(f" {k:12s} {v}")
|
||
return
|
||
if name == "make-ep":
|
||
steps = [("sync-shots", f"python -m lib.cli sync-shots"),
|
||
("batch", f"python -m lib.cli batch {episode_id}"),
|
||
("batch-dub", f"python -m lib.cli batch-dub {episode_id}"),
|
||
("batch-qc", f"python -m lib.cli batch-qc {episode_id}"),
|
||
("render", f"python -m lib.cli render {episode_id}")]
|
||
print(f"\n🎬 make-ep {episode_id}\n")
|
||
for i, (n, c) in enumerate(steps):
|
||
print(f"[{i+1}/{len(steps)}] {n}...")
|
||
_sp.run(c, shell=True, cwd=os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
print(f"\n✅ make-ep 完成")
|
||
|
||
def cmd_info(video_path=""):
|
||
"""视频信息: 时长/分辨率/编码/大小"""
|
||
import subprocess as _sp, shutil
|
||
if not video_path:
|
||
print("❌ 用法: info video.mp4"); return
|
||
ffprobe = shutil.which("ffprobe") or shutil.which("ffmpeg")
|
||
if not ffprobe: print("❌ ffprobe/ffmpeg 未找到"); return
|
||
r = _sp.run([ffprobe, "-v", "quiet", "-show_entries", "format=duration,size", "-of", "csv=p=0", video_path],
|
||
capture_output=True, text=True, timeout=10)
|
||
parts = r.stdout.strip().split(",") if r.stdout.strip() else []
|
||
dur = float(parts[0]) if len(parts) > 0 and parts[0] else 0
|
||
sz = int(parts[1]) if len(parts) > 1 and parts[1] else 0
|
||
r2 = _sp.run([ffprobe, "-v", "quiet", "-select_streams", "v:0", "-show_entries", "stream=width,height,codec_name",
|
||
"-of", "csv=p=0", video_path], capture_output=True, text=True, timeout=10)
|
||
vp = r2.stdout.strip().split(",") if r2.stdout.strip() else []
|
||
print(f"\n📹 {os.path.basename(video_path)}")
|
||
print(f" 时长: {dur:.1f}s 大小: {sz/1048576:.1f}MB" if sz else f" 时长: {dur:.1f}s")
|
||
if len(vp) >= 2: print(f" 分辨率: {vp[0]}x{vp[1]} 编码: {vp[2] if len(vp)>2 else '?'}")
|
||
|
||
def cmd_characters(script_path=""):
|
||
"""角色提取: 从剧本解析角色列表"""
|
||
import subprocess as _sp, json as _json
|
||
if not script_path: print("❌ 用法: characters 剧本.txt"); return
|
||
project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
engine = os.path.join(project_root, "engines", "script-parser.js")
|
||
if os.path.exists(engine) and os.path.exists(script_path):
|
||
with open(script_path, "r", encoding="utf-8") as f: script = f.read()[:50000]
|
||
r = _sp.run(["node", "-e", f"const sp=require('{engine.replace(chr(92),'/')}');const r=sp.extractAllCharacters(`{script.replace('`','\\`').replace('$','\\$')}`);console.log(JSON.stringify(r));"],
|
||
capture_output=True, text=True, timeout=30)
|
||
if r.returncode == 0 and r.stdout.strip():
|
||
try:
|
||
chars = _json.loads(r.stdout)
|
||
if isinstance(chars, list):
|
||
print(f"👥 {len(chars)}个角色:"); [print(f" · {c}") for c in chars]
|
||
else: print(r.stdout[:500])
|
||
except: print(r.stdout[:500])
|
||
else: print("⚠️ 无法解析，请用 parse 命令")
|
||
|
||
def cmd_sub_ab(video_path="", srt_path="", styles=""):
|
||
"""字幕A/B测试: 多种样式对比"""
|
||
import subprocess as _sp
|
||
if not video_path or not srt_path: print("❌ 用法: sub-ab video.mp4 subs.srt"); return
|
||
project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
engine = os.path.join(project_root, "engines", "subtitle-pipeline", "ab-test-tools", "subtitle-style-ab-test.py")
|
||
out_dir = f"outputs/ab-test/{os.urandom(2).hex()}"
|
||
os.makedirs(out_dir, exist_ok=True)
|
||
if os.path.exists(engine):
|
||
cmd = [sys.executable, engine, "--video", video_path, "--srt", srt_path, "--output-dir", out_dir]
|
||
if styles:
|
||
for s in styles.split(","): cmd += ["--styles", s.strip()]
|
||
r = _sp.run(cmd, capture_output=True, text=True, timeout=120)
|
||
print(f"{'✅' if r.returncode==0 else '❌'} {out_dir}")
|
||
else: print("⚠️ A/B测试引擎未找到")
|
||
|
||
def cmd_ref_sub(image_path="", output=""):
|
||
"""参考字幕分析: 从截图提取字幕样式参数"""
|
||
import subprocess as _sp
|
||
if not image_path: print("❌ 用法: ref-sub 参考截图.png"); return
|
||
project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
engine = os.path.join(project_root, "engines", "subtitle-pipeline", "reference-analysis", "reference-subtitle-analyzer.py")
|
||
out = output or f"{image_path}.sub-analysis.json"
|
||
if os.path.exists(engine):
|
||
r = _sp.run([sys.executable, engine, "--image", image_path, "--output", out], capture_output=True, text=True, timeout=30)
|
||
print(f"{'✅' if r.returncode==0 else '❌'} {out}")
|
||
else: print("⚠️ 参考分析引擎未找到")
|
||
|
||
def cmd_shot_set(shot_id="", status=""):
|
||
"""手动设置镜头状态: done / pending / failed"""
|
||
db = get_db()
|
||
if not shot_id or status not in ("done", "pending", "failed"):
|
||
print("❌ 用法: shot-set <shot_id> done|pending|failed")
|
||
return
|
||
db.execute("UPDATE shots SET status = ?, updated_at = datetime('now') WHERE id = ?", (status, shot_id))
|
||
db.commit()
|
||
print(f"✅ {shot_id} → {status}")
|
||
|
||
def cmd_clean(what="temp"):
|
||
"""清理: temp(临时文件) / outputs(输出目录) / db(重置数据库)"""
|
||
import shutil
|
||
project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
|
||
if what == "temp":
|
||
temps = ["outputs/temp_concat.txt"]
|
||
count = 0
|
||
for pattern in temps + ["*.director-cfg.json", "*.qc-report.json"]:
|
||
for f in os.listdir(os.path.join(project_root, "outputs")):
|
||
fp = os.path.join(project_root, "outputs", f)
|
||
if os.path.isfile(fp) and ("temp" in f or ".qc-report" in f or ".director-cfg" in f):
|
||
os.remove(fp); count += 1
|
||
db_path = os.path.join(os.path.dirname(__file__), "lib.db")
|
||
print(f"✅ 清理 {count} 个临时文件")
|
||
|
||
elif what == "outputs":
|
||
out_dir = os.path.join(project_root, "outputs")
|
||
if os.path.exists(out_dir):
|
||
size = sum(os.path.getsize(os.path.join(dp, f)) for dp, _, files in os.walk(out_dir) for f in files)
|
||
print(f"⚠️ 即将删除 outputs/ 目录 ({size/1048576:.1f}MB)")
|
||
print(" 确认? (此命令仅提示，不实际删除)")
|
||
print(" 手动执行: rm -rf outputs/")
|
||
|
||
elif what == "db":
|
||
db_path = os.path.join(os.path.dirname(__file__), "lib.db")
|
||
if os.path.exists(db_path):
|
||
print(f"⚠️ 即将重置数据库: {db_path}")
|
||
print(" 确认? (此命令仅提示，不实际删除)")
|
||
print(" 手动执行: rm lib.db && python -m lib.cli init")
|
||
else:
|
||
print(f"❌ 用法: clean temp|outputs|db")
|
||
"""工作流模板: make-ep(制作一集)"""
|
||
import subprocess as _sp
|
||
|
||
workflows = {
|
||
"make-ep": "制作一集: parse→sync→batch→batch-dub→batch-qc→render",
|
||
}
|
||
|
||
if not name:
|
||
print("📋 可用工作流:")
|
||
for k, v in workflows.items():
|
||
print(f" {k:12s} {v}")
|
||
print("\n用法: workflow make-ep [episode_id]")
|
||
return
|
||
|
||
if name == "make-ep":
|
||
steps = [
|
||
("sync-shots", f"python -m lib.cli sync-shots"),
|
||
("batch", f"python -m lib.cli batch {episode_id}"),
|
||
("batch-dub", f"python -m lib.cli batch-dub {episode_id}"),
|
||
("batch-qc", f"python -m lib.cli batch-qc {episode_id}"),
|
||
("render", f"python -m lib.cli render {episode_id}"),
|
||
]
|
||
print(f"\n🎬 工作流: make-ep {episode_id}")
|
||
print(f" 步骤: sync → batch → dub → qc → render\n")
|
||
for i, (name, cmd) in enumerate(steps):
|
||
print(f"[{i+1}/{len(steps)}] {name}...")
|
||
r = _sp.run(cmd, shell=True, cwd=os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
if r.returncode != 0 and name != "batch-qc":
|
||
print(f" ⚠️ {name} 返回非零: {r.returncode}")
|
||
print(f"\n✅ make-ep 完成")
|
||
|
||
def cmd_storyboard(script_path="", episode_num="1", output=""):
|
||
"""全剧分镜生成: 剧本→逐镜画面描述+景别+时长+角色"""
|
||
import subprocess as _sp, json as _json
|
||
project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
|
||
if not script_path:
|
||
print("❌ 用法: storyboard 剧本.txt [集号] [输出.json]")
|
||
return
|
||
|
||
engine = os.path.join(project_root, "engines", "hldp-prompt.js")
|
||
out = output or script_path.replace(".txt", "-storyboard.md").replace(".md", "-storyboard.md")
|
||
|
||
if os.path.exists(engine) and os.path.exists(script_path):
|
||
with open(script_path, "r", encoding="utf-8") as f:
|
||
script = f.read()[:30000]
|
||
print(f"🎬 生成分镜脚本 (第{episode_num}集)...")
|
||
|
||
# Try episodeToStoryboard
|
||
scenes = [{"description": script[:2000]}] # simplified scene
|
||
r = _sp.run(["node", "-e",
|
||
f"const hp=require('{engine.replace(chr(92),'/')}');"
|
||
f"const r=hp.episodeToStoryboard({_json.dumps(scenes)},{episode_num});"
|
||
f"console.log(r||JSON.stringify(r));"],
|
||
capture_output=True, text=True, timeout=60)
|
||
|
||
if r.returncode == 0 and r.stdout.strip():
|
||
with open(out, "w", encoding="utf-8") as f:
|
||
f.write(r.stdout)
|
||
# Count shots roughly
|
||
shots_count = r.stdout.count("### Shot") or r.stdout.count("shot_number")
|
||
print(f"✅ {out} (~{shots_count}镜)" if shots_count else f"✅ {out}")
|
||
else:
|
||
# Fallback to AI
|
||
print(f"⚠️ HLDP引擎失败，用AI兜底...")
|
||
from lib.doubao_chat import breakdown_script
|
||
r2 = breakdown_script(script, int(episode_num))
|
||
if "content" in r2:
|
||
with open(out, "w", encoding="utf-8") as f:
|
||
f.write(r2["content"])
|
||
print(f"✅ {out} (AI)")
|
||
else:
|
||
print(f"❌ {r2}")
|
||
else:
|
||
from lib.doubao_chat import breakdown_script
|
||
with open(script_path, "r", encoding="utf-8") as f:
|
||
script = f.read()
|
||
print(f"🤖 AI分镜...")
|
||
r = breakdown_script(script, int(episode_num))
|
||
if "content" in r:
|
||
with open(out, "w", encoding="utf-8") as f:
|
||
f.write(r["content"])
|
||
print(f"✅ {out}")
|
||
|
||
def cmd_export(what="shots", episode_id="DSV-EP01", output=""):
|
||
"""数据导出: shots→JSON / assets→JSON / costs→CSV"""
|
||
import json as _json
|
||
db = get_db()
|
||
|
||
if what == "shots":
|
||
rows = db.execute("SELECT * FROM shots WHERE episode_id = ? ORDER BY shot_number", (episode_id,)).fetchall()
|
||
data = [dict(r) for r in rows]
|
||
out = output or f"exports/{episode_id}-shots.json"
|
||
elif what == "assets":
|
||
rows = db.execute("SELECT * FROM assets ORDER BY type, name").fetchall()
|
||
data = [dict(r) for r in rows]
|
||
out = output or "exports/assets.json"
|
||
elif what == "costs":
|
||
rows = db.execute("""
|
||
SELECT gt.*, s.shot_number FROM generation_tasks gt
|
||
JOIN shots s ON gt.shot_id = s.id ORDER BY gt.created_at
|
||
""").fetchall()
|
||
data = [dict(r) for r in rows]
|
||
out = output or "exports/costs.csv"
|
||
os.makedirs(os.path.dirname(out) or ".", exist_ok=True)
|
||
with open(out, "w", encoding="utf-8") as f:
|
||
f.write("shot,type,cost_estimated,cost_actual,status,created_at\n")
|
||
for r in data:
|
||
f.write(f"{r.get('shot_number','')},{r.get('task_type','')},{r.get('cost_estimated',0)},{r.get('cost_actual',0)},{r.get('status','')},{r.get('created_at','')}\n")
|
||
print(f"✅ {out} ({len(data)}条)")
|
||
return
|
||
else:
|
||
print(f"❌ 未知导出类型: {what} (shots/assets/costs)")
|
||
return
|
||
|
||
os.makedirs(os.path.dirname(out) or ".", exist_ok=True)
|
||
with open(out, "w", encoding="utf-8") as f:
|
||
_json.dump(data, f, ensure_ascii=False, indent=2, default=str)
|
||
print(f"✅ {out} ({len(data)}条)")
|
||
|
||
def cmd_safe_qc(video_path="", subtitle_ass="", output=""):
|
||
"""字幕安全区QC: 检测面部遮挡/贴边/对比度低"""
|
||
import subprocess as _sp
|
||
project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
|
||
if not video_path:
|
||
print("❌ 用法: safe-qc video.mp4 [subtitles.ass]")
|
||
return
|
||
|
||
engine = os.path.join(project_root, "engines", "subtitle-pipeline", "qc-tools", "subtitle-safe-area-qc.py")
|
||
out = output or f"{video_path}.safe-qc.json"
|
||
|
||
if os.path.exists(engine):
|
||
cmd = [sys.executable, engine, "--video", video_path, "--output", out]
|
||
if subtitle_ass: cmd += ["--subtitle-ass", subtitle_ass]
|
||
print(f"🔍 字幕安全区QC...")
|
||
r = _sp.run(cmd, capture_output=True, text=True, timeout=60)
|
||
if r.returncode == 0:
|
||
import json as _json
|
||
if os.path.exists(out):
|
||
report = _json.load(open(out, "r", encoding="utf-8"))
|
||
issues = report.get("issues", []) if isinstance(report, dict) else []
|
||
print(f"{'✅ 安全' if not issues else '⚠️ ' + str(len(issues)) + '个问题'}")
|
||
else:
|
||
print(f"✅ QC完成")
|
||
else:
|
||
print(f"❌ {r.stderr[:200]}")
|
||
else:
|
||
print(f"⚠️ 安全区QC引擎未找到")
|
||
|
||
def cmd_gen(shot_id, model=""):
|
||
"""提交视频生成任务: --model seedance|kling|wan, --dry-run, --seed <N>, --lock-seed"""
|
||
import subprocess as _sp
|
||
db = get_db()
|
||
shot = db.execute("SELECT * FROM shots WHERE id = ?", (shot_id,)).fetchone()
|
||
if not shot:
|
||
print(f"镜头 {shot_id} 不存在")
|
||
return
|
||
if shot["status"] == "done":
|
||
print(f"⚠️ {shot_id} 已完成。如需重跑请先 shot-set {shot_id} pending")
|
||
return
|
||
|
||
refs = db.execute("SELECT sa.role, a.file_path FROM shot_assets sa JOIN assets a ON sa.asset_id = a.id WHERE sa.shot_id = ?", (shot_id,)).fetchall()
|
||
ref_paths = [(r["role"], r["file_path"]) for r in refs]
|
||
prompt = shot["prompt"] or shot["description"]
|
||
if not prompt:
|
||
print(f"❌ {shot_id} 没有提示词")
|
||
return
|
||
|
||
# --dry-run 模式
|
||
if model == "--dry-run" or model == "true":
|
||
print(f"\n🔍 DRY RUN: {shot_id}")
|
||
print(f" 提示词: {prompt[:100]}...")
|
||
print(f" 时长: {shot['duration_sec']}s")
|
||
print(f" 参考图: {len(ref_paths)}张")
|
||
print(f" 锁定Seed: {shot['seed_value'] if shot['seed_value'] and shot['lock_seed'] else '无'}")
|
||
print(f" 预估成本: ~¥{int(shot['duration_sec'] or 6) * 0.5:.1f}")
|
||
print(f"\n💡 确认后执行: python -m lib.cli gen {shot_id}")
|
||
return
|
||
|
||
# --lock-seed 模式
|
||
if model == "--lock-seed":
|
||
if shot["seed_value"]:
|
||
db.execute("UPDATE shots SET lock_seed = 1 WHERE id = ?", (shot_id,))
|
||
db.commit()
|
||
print(f"🔒 {shot_id} Seed={shot['seed_value']} 已锁定，后续生成将复用此Seed保持角色一致")
|
||
else:
|
||
print(f"❌ {shot_id} 还没有Seed值，先生成一次")
|
||
return
|
||
|
||
# --seed 手动指定
|
||
seed_val = None
|
||
if model and model.startswith("--seed="):
|
||
seed_val = int(model.split("=")[1])
|
||
model = "seedance"
|
||
elif shot["lock_seed"] and shot["seed_value"]:
|
||
seed_val = shot["seed_value"]
|
||
print(f"🔒 使用锁定Seed={seed_val}")
|
||
|
||
m = (model or "seedance").lower()
|
||
project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
|
||
if m in ("kling", "wan"):
|
||
engine_name = f"{m}-official-adapter.js" if m == "kling" else f"{m}-api-adapter.js"
|
||
engine = os.path.join(project_root, "engines", engine_name)
|
||
if os.path.exists(engine):
|
||
print(f"🎬 {m.upper()}生成 {shot_id}...")
|
||
dur = int(shot["duration_sec"] or 6)
|
||
r = _sp.run(["node", "-e",
|
||
f"const gen=require('{engine.replace(chr(92),'/')}');"
|
||
f"gen.generateVideo({{prompt:`{prompt}`,duration:{dur},"
|
||
f"outputPath:'outputs/videos/{shot_id}.mp4'}})"
|
||
f".then(r=>{{console.log(JSON.stringify(r));process.exit(0);}})"
|
||
f".catch(e=>{{console.error(e.message);process.exit(1);}})"],
|
||
capture_output=True, text=True, timeout=120)
|
||
if r.returncode == 0:
|
||
db.execute("UPDATE shots SET status = 'generating', model = ? WHERE id = ?", (m, shot_id)); db.commit()
|
||
print(f"📤 {shot_id} → {m.upper()}")
|
||
else: print(f"❌ {r.stderr[:200]}")
|
||
else: print(f"⚠️ {m}引擎未找到，回退Seedance"); m = "seedance"
|
||
|
||
if m == "seedance":
|
||
from lib.seedance import generate_shot_async
|
||
r = generate_shot_async(shot_id, prompt, ref_paths, int(shot["duration_sec"] or 6))
|
||
if "error" in r: print(f"❌ {r['error']}")
|
||
else: print(f"📤 {shot_id} 已提交")
|
||
|
||
def cmd_flow(episode_id="DSV-EP01"):
|
||
"""管线可视化: 展示剧集生产流程当前进度"""
|
||
db = get_db()
|
||
shots = db.execute("""
|
||
SELECT shot_number, status, description, output_path, duration_sec
|
||
FROM shots WHERE episode_id = ? ORDER BY shot_number
|
||
""", (episode_id,)).fetchall()
|
||
|
||
if not shots:
|
||
print(f"❌ {episode_id} 没有镜头")
|
||
return
|
||
|
||
icons = {"done": "✅", "generating": "🔄", "pending": "🔴", "failed": "❌"}
|
||
|
||
print(f"\n{'='*70}")
|
||
print(f"🔗 {episode_id} · 生产管线")
|
||
print(f"{'='*70}\n")
|
||
|
||
for i, s in enumerate(shots):
|
||
icon = icons.get(s["status"], "❓")
|
||
dur = f"{s['duration_sec']:.0f}s" if s["duration_sec"] else "?"
|
||
desc = (s["description"] or "")[:50]
|
||
|
||
# 绘制管线连接
|
||
if i == 0:
|
||
prefix = " "
|
||
else:
|
||
prev_status = shots[i-1]["status"]
|
||
connector = "━" if prev_status == "done" else "╌"
|
||
prefix = f" {connector}→ "
|
||
|
||
print(f"{prefix}{icon} {s['shot_number']:6s} [{dur:4s}] {desc}")
|
||
|
||
# 统计
|
||
done = sum(1 for s in shots if s["status"] == "done")
|
||
total = len(shots)
|
||
total_dur = sum(s["duration_sec"] or 0 for s in shots if s["status"] == "done")
|
||
|
||
print(f"\n{'─'*70}")
|
||
print(f"📊 {done}/{total} 镜完成")
|
||
if done > 0:
|
||
print(f"⏱ 已完成: {total_dur:.0f}s")
|
||
print(f"🔜 下一步:")
|
||
if done == total:
|
||
print(f" python -m lib.cli render {episode_id}")
|
||
elif done > 0:
|
||
print(f" python -m lib.cli qc {shots[done-1]['shot_number']} # 质检最后一镜")
|
||
print(f" python -m lib.cli batch {episode_id} # 继续生成")
|
||
else:
|
||
print(f" python -m lib.cli batch {episode_id} # 开始批量生成")
|
||
else:
|
||
print(f"🔜 下一步: python -m lib.cli batch {episode_id}")
|
||
print(f"{'='*70}\n")
|
||
|
||
def cmd_shot_order(episode_id="DSV-EP01", shot_number="", new_number=""):
|
||
"""镜头重排: 调整镜头顺序"""
|
||
db = get_db()
|
||
|
||
if not shot_number or not new_number:
|
||
# 显示当前顺序
|
||
rows = db.execute("""
|
||
SELECT shot_number, description FROM shots
|
||
WHERE episode_id = ? ORDER BY shot_number
|
||
""", (episode_id,)).fetchall()
|
||
print(f"\n{episode_id} 镜头顺序:")
|
||
for r in rows:
|
||
print(f" {r['shot_number']:6s} {r['description'][:60] if r['description'] else ''}")
|
||
print("\n用法: shot-order DSV-EP01 S05 S03 (把S05移到S03位置)")
|
||
return
|
||
|
||
# 交换两个镜头的序号
|
||
db.execute("UPDATE shots SET shot_number = '__TEMP__' WHERE episode_id = ? AND shot_number = ?",
|
||
(episode_id, shot_number))
|
||
db.execute("UPDATE shots SET shot_number = ? WHERE episode_id = ? AND shot_number = ?",
|
||
(shot_number, episode_id, new_number))
|
||
db.execute("UPDATE shots SET shot_number = ? WHERE episode_id = ? AND shot_number = '__TEMP__'",
|
||
(new_number, episode_id))
|
||
db.commit()
|
||
print(f"✅ {shot_number} ↔ {new_number}")
|
||
|
||
def cmd_batch_resume(episode_id="DSV-EP01"):
|
||
"""从中断恢复批量任务"""
|
||
from lib.seedance import batch_resume
|
||
batch_resume(episode_id)
|
||
|
||
def cmd_status(episode_id="DSV-EP01"):
|
||
"""查看剧集全局状态：镜头进度 + 资产批准率 + 预估时长"""
|
||
db = get_db()
|
||
|
||
# 镜头统计
|
||
shots = db.execute("""
|
||
SELECT status, COUNT(*) as count FROM shots
|
||
WHERE episode_id = ? GROUP BY status
|
||
""", (episode_id,)).fetchall()
|
||
|
||
# 资产统计
|
||
assets = db.execute("""
|
||
SELECT type, status, COUNT(*) as count FROM assets
|
||
GROUP BY type, status
|
||
""").fetchall()
|
||
|
||
# 成本统计
|
||
cost = db.execute("""
|
||
SELECT COALESCE(SUM(cost_estimated), 0) as estimated,
|
||
COALESCE(SUM(cost_actual), 0) as actual
|
||
FROM generation_tasks
|
||
""").fetchone()
|
||
|
||
print(f"\n{'='*50}")
|
||
print(f"📊 {episode_id} · 全局状态")
|
||
print(f"{'='*50}")
|
||
|
||
# 镜头
|
||
shot_map = {r["status"]: r["count"] for r in shots}
|
||
total_shots = sum(shot_map.values())
|
||
done = shot_map.get("done", 0)
|
||
generating = shot_map.get("generating", 0)
|
||
pending = shot_map.get("pending", 0)
|
||
failed_n = shot_map.get("failed", 0)
|
||
|
||
bar_len = 30
|
||
done_bar = int(done / max(total_shots, 1) * bar_len)
|
||
gen_bar = int(generating / max(total_shots, 1) * bar_len)
|
||
pend_bar = bar_len - done_bar - gen_bar
|
||
|
||
print(f"\n🎬 镜头: {total_shots}镜")
|
||
print(f" ✅ done: {done} 🔄 generating: {generating} 🔴 pending: {pending} ❌ failed: {failed_n}")
|
||
print(f" [{'█'*done_bar}{'░'*gen_bar}{'·'*pend_bar}] {done}/{total_shots}")
|
||
|
||
# 时长预估
|
||
if done > 0:
|
||
done_shots = db.execute("""
|
||
SELECT SUM(duration_sec) as total_dur FROM shots
|
||
WHERE episode_id = ? AND status = 'done'
|
||
""", (episode_id,)).fetchone()
|
||
print(f" ⏱ 已完成时长: {done_shots['total_dur']:.0f}s")
|
||
|
||
# 资产
|
||
print(f"\n📦 资产:")
|
||
asset_types = {}
|
||
for r in assets:
|
||
t = r["type"]
|
||
if t not in asset_types: asset_types[t] = {"approved": 0, "pending": 0}
|
||
asset_types[t][r["status"]] = r["count"]
|
||
for t, counts in asset_types.items():
|
||
total = counts["approved"] + counts["pending"]
|
||
print(f" {t}: {counts['approved']}/{total} 已批准")
|
||
|
||
# 成本
|
||
print(f"\n💰 成本:")
|
||
print(f" 预估: ¥{cost['estimated']:.2f} 实际(按token): {cost['actual']:.0f}")
|
||
|
||
print(f"\n{'='*50}")
|
||
|
||
def cmd_note(shot_id="", text=""):
|
||
"""镜头备注: 添加/查看镜头备注"""
|
||
db = get_db()
|
||
if not shot_id:
|
||
rows = db.execute("SELECT id, prompt FROM shots WHERE prompt LIKE '%@note:%'").fetchall()
|
||
if rows:
|
||
print("📝 带备注的镜头:")
|
||
for r in rows:
|
||
n = r["prompt"].split("@note:")[-1].strip()[:60]
|
||
print(f" {r['id']:20s} {n}")
|
||
else: print("还没有备注。用法: note <shot_id> <内容>")
|
||
return
|
||
if not text:
|
||
shot = db.execute("SELECT prompt FROM shots WHERE id = ?", (shot_id,)).fetchone()
|
||
if shot and shot["prompt"] and "@note:" in (shot["prompt"] or ""):
|
||
print(f"📝 {shot_id}: {shot['prompt'].split('@note:')[-1].strip()}")
|
||
else: print(f"💡 {shot_id} 暂无备注"); return
|
||
return
|
||
shot = db.execute("SELECT prompt FROM shots WHERE id = ?", (shot_id,)).fetchone()
|
||
if not shot: print(f"❌ {shot_id} 不存在"); return
|
||
base = (shot["prompt"] or "").split("@note:")[0].strip()
|
||
db.execute("UPDATE shots SET prompt = ? WHERE id = ?", (f"{base}\n@note: {text}", shot_id))
|
||
db.commit()
|
||
print(f"✅ {shot_id} 备注已保存")
|
||
|
||
def cmd_deps():
|
||
"""依赖检测: 检查Python/Node包是否安装"""
|
||
import importlib, shutil
|
||
|
||
print(f"\n{'='*50}")
|
||
print(f"📦 依赖检测")
|
||
print(f"{'='*50}\n")
|
||
|
||
# Python
|
||
py_deps = {"edge_tts": "Edge-TTS配音", "PIL": "Pillow图像处理",
|
||
"cv2": "OpenCV视觉", "numpy": "NumPy数值计算"}
|
||
print("🐍 Python:")
|
||
for mod, desc in py_deps.items():
|
||
try:
|
||
importlib.import_module(mod)
|
||
print(f" ✅ {mod:12s} {desc}")
|
||
except:
|
||
print(f" ❌ {mod:12s} {desc} — pip install {mod.split('.')[0]}")
|
||
|
||
# Node
|
||
print("\n📦 Node:")
|
||
node = shutil.which("node")
|
||
if node:
|
||
import subprocess as _sp
|
||
for pkg, desc in [("edge-tts","(Python包)"), ("sharp","图像处理"), ("ffmpeg-static","FFmpeg静态")]:
|
||
r = _sp.run([node, "-e", f"try{{require('{pkg}');process.exit(0)}}catch(e){{process.exit(1)}}"],
|
||
capture_output=True)
|
||
print(f" {'✅' if r.returncode==0 else '⚠️'} {pkg:15s} {desc}")
|
||
|
||
# Tools
|
||
print("\n🔧 工具:")
|
||
for tool in ["ffmpeg", "ffprobe", "git", "curl"]:
|
||
p = shutil.which(tool)
|
||
print(f" {'✅' if p else '❌'} {tool:10s} {'→ '+p if p else '未安装'}")
|
||
|
||
print(f"\n{'='*50}")
|
||
|
||
def cmd_history(limit="20"):
|
||
"""生成历史: 最近的任务记录+成本"""
|
||
db = get_db()
|
||
rows = db.execute("""
|
||
SELECT gt.*, s.shot_number, s.episode_id FROM generation_tasks gt
|
||
JOIN shots s ON gt.shot_id = s.id ORDER BY gt.created_at DESC LIMIT ?
|
||
""", (int(limit),)).fetchall()
|
||
if not rows: print("还没有生成历史"); return
|
||
icons = {"submitted":"📤","running":"🔄","succeeded":"✅","failed":"❌"}
|
||
print(f"\n📜 最近 {len(rows)} 条\n")
|
||
for r in rows:
|
||
i = icons.get(r["status"], "❓")
|
||
c = f"¥{r['cost_estimated']:.1f}" if r["cost_estimated"] else "?"
|
||
print(f" {i} {r['shot_number']:8s} {r['task_type']:6s} {r['status']:10s} {c:8s} {r['created_at']}")
|
||
|
||
def cmd_report(episode_id="DSV-EP01", output=""):
|
||
"""剧集总结报告: HTML格式(镜头+成本+时间线)"""
|
||
db = get_db()
|
||
shots = db.execute("""
|
||
SELECT s.*, COALESCE(SUM(gt.cost_estimated),0) as total_cost
|
||
FROM shots s LEFT JOIN generation_tasks gt ON s.id = gt.shot_id
|
||
WHERE s.episode_id = ? GROUP BY s.id ORDER BY s.shot_number
|
||
""", (episode_id,)).fetchall()
|
||
if not shots: print(f"❌ {episode_id} 没有镜头"); return
|
||
|
||
costs = db.execute("""
|
||
SELECT COALESCE(SUM(cost_estimated),0) as est FROM generation_tasks gt
|
||
JOIN shots s ON gt.shot_id = s.id WHERE s.episode_id = ?
|
||
""", (episode_id,)).fetchone()
|
||
|
||
done = sum(1 for s in shots if s["status"] == "done")
|
||
total = len(shots)
|
||
dur = sum(s["duration_sec"] or 0 for s in shots if s["status"] == "done")
|
||
ico = {"done":"✅","generating":"🔄","pending":"🔴","failed":"❌"}
|
||
|
||
out = output or f"outputs/reports/{episode_id}-report.html"
|
||
os.makedirs(os.path.dirname(out), exist_ok=True)
|
||
|
||
rows = "".join(
|
||
f"<tr><td>{ico.get(s['status'],'?')}</td><td>{s['shot_number']}</td>"
|
||
f"<td>{(s['duration_sec'] or 0):.0f}s</td><td>{(s['description'] or '')[:50]}</td>"
|
||
f"<td>¥{s['total_cost']:.1f}</td></tr>"
|
||
for s in shots
|
||
)
|
||
|
||
pct = done/max(total,1)*100
|
||
html = f"""<!DOCTYPE html><html lang="zh"><head><meta charset="UTF-8">
|
||
<title>{episode_id} 报告</title>
|
||
<style>body{{font-family:system-ui,sans-serif;max-width:900px;margin:40px auto;padding:20px;color:#333}}
|
||
h1{{border-bottom:2px solid #4A90D9;padding-bottom:10px}}
|
||
.g{{display:flex;gap:40px;margin:20px 0;padding:15px;background:#f5f7fa;border-radius:8px}}
|
||
.g div{{text-align:center}} .v{{font-size:24px;font-weight:bold;color:#4A90D9}}
|
||
table{{width:100%;border-collapse:collapse}} th,td{{padding:8px 12px;text-align:left;border-bottom:1px solid #eee}}
|
||
th{{background:#4A90D9;color:white}} .f{{margin-top:30px;color:#999;font-size:12px}}
|
||
.bar{{height:8px;background:#e0e0e0;border-radius:4px}} .bar div{{height:100%;background:#4A90D9;border-radius:4px;width:{pct:.0f}%}}</style></head><body>
|
||
<h1>🎬 {episode_id} · 生产报告</h1>
|
||
<div class="g"><div>📊 进度<div class="v">{done}/{total}</div></div>
|
||
<div>⏱ 已完成<div class="v">{dur:.0f}s</div></div>
|
||
<div>💰 成本<div class="v">¥{costs['est']:.1f}</div></div></div>
|
||
<div class="bar"><div></div></div>
|
||
<table><tr><th>状态</th><th>镜头</th><th>时长</th><th>描述</th><th>成本</th></tr>{rows}</table>
|
||
<div class="f">蛋蛋短剧生产管线 · cang-ying · D181</div></body></html>"""
|
||
|
||
with open(out, "w", encoding="utf-8") as f: f.write(html)
|
||
print(f"✅ {out} ({os.path.getsize(out)/1024:.0f}KB)")
|
||
|
||
def cmd_inherit(source_ep="", target_ep=""):
|
||
"""跨集资产继承: 从源剧集复制资产到目标剧集"""
|
||
db = get_db()
|
||
if not source_ep or not target_ep:
|
||
eps = db.execute("SELECT id, project_id, number FROM episodes ORDER BY project_id, number").fetchall()
|
||
print("📋 现有剧集:"); [print(f" {e['project_id']:15s} {e['id']:20s} E{e['number']:02d}") for e in eps]
|
||
print("\n用法: inherit <源剧集> <目标剧集>")
|
||
return
|
||
src = db.execute("SELECT * FROM episodes WHERE id = ?", (source_ep,)).fetchone()
|
||
tgt = db.execute("SELECT * FROM episodes WHERE id = ?", (target_ep,)).fetchone()
|
||
if not src or not tgt: print("❌ 剧集不存在"); return
|
||
assets = db.execute("SELECT * FROM assets WHERE project_id = ?", (src["project_id"],)).fetchall()
|
||
count = 0
|
||
for a in assets:
|
||
new_id = a["id"].replace(src["project_id"], tgt["project_id"])
|
||
db.execute("""INSERT OR IGNORE INTO assets (id, project_id, type, name, file_path, status, approved_at)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?)""",
|
||
(new_id, tgt["project_id"], a["type"], a["name"], a["file_path"], a["status"], a["approved_at"]))
|
||
count += 1
|
||
db.commit()
|
||
print(f"✅ {source_ep} → {target_ep}: 继承 {count} 个资产")
|
||
seeds = db.execute("SELECT seed_value, shot_number FROM shots WHERE episode_id = ? AND seed_value > 0 AND lock_seed = 1",
|
||
(source_ep,)).fetchall()
|
||
if seeds:
|
||
print(f"🔒 {len(seeds)} 个锁定Seed可复用:")
|
||
for s in seeds: print(f" {s['shot_number']}: --seed={s['seed_value']}")
|
||
|
||
def cmd_compare(shot_id="", models="seedance,kling,wan"):
|
||
"""A/B对比: 同镜头多模型并行生成"""
|
||
db = get_db()
|
||
if not shot_id: print("❌ 用法: compare <shot_id> [模型列表]"); return
|
||
ml = [m.strip() for m in models.split(",")]
|
||
cost = int(db.execute("SELECT duration_sec FROM shots WHERE id = ?", (shot_id,)).fetchone()["duration_sec"] or 6) * 0.5 * len(ml)
|
||
print(f"\n🔬 {shot_id} × {len(ml)}模型: {', '.join(m.upper() for m in ml)}")
|
||
print(f" 预估总成本: ~¥{cost:.1f}\n")
|
||
for m in ml:
|
||
if m in ("seedance", "kling", "wan"):
|
||
print(f" 📤 {m.upper()}..."); cmd_gen(shot_id, m)
|
||
print(f"\n💡 查看结果: python -m lib.cli status")
|
||
|
||
COMMANDS = {
|
||
"init": cmd_init, "status": cmd_status,
|
||
"project": cmd_project, "episode": cmd_episode,
|
||
"shots": cmd_shot_list, "shot": cmd_shot_status,
|
||
"assets": cmd_asset_list, "asset": cmd_asset, "sync-shots": cmd_sync_shots, "link": cmd_link_asset,
|
||
"tasks": cmd_task_log, "exp-add": cmd_exp_add, "exp-list": cmd_exp_list,
|
||
"chat": cmd_chat, "breakdown": cmd_breakdown,
|
||
"gen": cmd_gen, "collect": cmd_collect, "batch": cmd_batch,
|
||
"batch-resume": cmd_batch_resume,
|
||
"image": cmd_image,
|
||
"qc": cmd_qc, "lipsync": cmd_lipsync, "mix": cmd_mix,
|
||
"emotion": cmd_emotion, "subtitles": cmd_subtitles,
|
||
"srt-timing": cmd_srt_timing, "font": cmd_font, "split-audio": cmd_split_audio,
|
||
"char-qc": cmd_char_qc, "eye": cmd_eye, "multi-ref": cmd_multi_ref,
|
||
"track": cmd_track, "director": cmd_director,
|
||
"pack": cmd_pack, "prompt": cmd_prompt,
|
||
"parse": cmd_parse, "adapt": cmd_adapt,
|
||
"produce": cmd_produce,
|
||
"batch-dub": cmd_batch_dub, "batch-qc": cmd_batch_qc,
|
||
"env": cmd_env, "config": cmd_config, "backup": cmd_backup,
|
||
"workflow": cmd_workflow,
|
||
"storyboard": cmd_storyboard, "export": cmd_export, "safe-qc": cmd_safe_qc,
|
||
"shot-order": cmd_shot_order, "shot-set": cmd_shot_set,
|
||
"info": cmd_info, "characters": cmd_characters,
|
||
"sub-ab": cmd_sub_ab, "ref-sub": cmd_ref_sub,
|
||
"clean": cmd_clean,
|
||
"flow": cmd_flow, "history": cmd_history,
|
||
"report": cmd_report,
|
||
"note": cmd_note,
|
||
"deps": cmd_deps,
|
||
"inherit": cmd_inherit, "compare": cmd_compare,
|
||
"render": cmd_render, "compose": cmd_compose,
|
||
"dub": cmd_dub,
|
||
}
|
||
|
||
# ====== 命令帮助文本 ======
|
||
HELP_TEXT = {
|
||
"init": "init — 初始化数据库，创建默认项目",
|
||
"status": "status [episode_id] — 全局状态: 镜头进度+资产+成本",
|
||
"project": "project list|new <id> <name> — 项目管理",
|
||
"episode": "episode list|new <project_id> <episode_id> [编号] — 剧集管理",
|
||
"asset": "asset list|add|approve <project_id> <类型> <名> [路径] — 资产管理",
|
||
"shots": "shots [episode_id] — 镜头列表(状态图标)",
|
||
"shot": "shot <shot_id> — 镜头详情(含关联资产)",
|
||
"assets": "assets [project_id] — 资产列表",
|
||
"sync-shots":"sync-shots — 从分镜文件导入镜头到数据库",
|
||
"link": "link <shot_id> <asset_id> <角色> — 关联资产到镜头",
|
||
"tasks": "tasks [shot_id] — 生成任务日志",
|
||
"exp-add": "exp-add <分类> <标题> <内容> — 添加经验",
|
||
"exp-list": "exp-list [分类] — 经验列表",
|
||
"chat": "chat <提示词> — 豆包对话",
|
||
"breakdown": "breakdown <剧本文件> [集号] — AI剧本拆解",
|
||
"gen": "gen <shot_id> [--model seedance|kling|wan] [--dry-run] — 提交生成\n"
|
||
" --dry-run: 预览提示词和预估成本，不实际提交",
|
||
"collect": "collect <shot_id> — 收集生成结果",
|
||
"batch": "batch [episode_id] [max_parallel] [retry] — 批量并行生成",
|
||
"batch-resume":"batch-resume [episode_id] — 中断恢复",
|
||
"image": "image t2i|i2i|multi <提示词> [ref] [输出] [尺寸] — Seedream图像生成",
|
||
"qc": "qc <shot_id> — 镜头质检(竖屏/字幕/换脸/牌匾/遮挡/现代物品)",
|
||
"lipsync": "lipsync <shot_id> <配音.mp3> — 口型同步",
|
||
"mix": "mix <视频> <配音> [bgm] — 音频混音",
|
||
"emotion": "emotion <台词> <角色·情绪·强度> [输出] — 情感配音",
|
||
"subtitles": "subtitles <srt> <视频> [输出] [样式] — 字幕烧录(5种样式)",
|
||
"srt-timing":"srt-timing <剧本> [配音目录] [输出] — 剧本→自动时间轴SRT",
|
||
"font": "font check|list|install — 字幕字体管理",
|
||
"split-audio":"split-audio <视频> [输出目录] — 音频分轨(对白/BGM/音效)",
|
||
"char-qc": "char-qc --image <图> --character <角色ID> — 角色辨识度QC",
|
||
"eye": "eye <视频> [导演编码.json] — 铸渊之眼逐帧分析",
|
||
"multi-ref": "multi-ref check|gen <提示词> <ref1,ref2> — 多参考适配",
|
||
"track": "track track|stabilize|pin|motion <视频> — 平面追踪",
|
||
"director": "director <分镜.json> <镜头目录> — 自动导演编排",
|
||
"pack": "pack <角色ID> [--project <项目>] [--views front,side,hand] [--dry-run] — 角色资产打包(主视角→参考+seed锁定)",
|
||
"prompt": "prompt <场景描述> — HLDP提示词引擎",
|
||
"parse": "parse <剧本> [输出] — 确定性剧本解析+AI兜底",
|
||
"adapt": "adapt <导演编码.json> — 导演编码→编辑参数",
|
||
"produce": "produce <shot_id> [--dry-run] — 单镜9步一站式生产",
|
||
"batch-dub": "batch-dub [episode_id] [角色名] — 批量配音",
|
||
"batch-qc": "batch-qc [episode_id] — 批量质检",
|
||
"env": "env — 环境检测(主机/Python/FFmpeg/Node/引擎/API密钥/DB)",
|
||
"config": "config [agents|voices|models] — 查看配置",
|
||
"backup": "backup — 数据库备份(保留最近10份)",
|
||
"workflow": "workflow [make-ep] [episode_id] — 工作流模板",
|
||
"storyboard":"storyboard <剧本> [集号] [输出] — 全剧分镜生成",
|
||
"export": "export shots|assets|costs [episode_id] — 数据导出",
|
||
"safe-qc": "safe-qc <视频> [字幕.ass] — 字幕安全区QC",
|
||
"shot-order":"shot-order [episode_id] [from] [to] — 镜头重排",
|
||
"shot-set": "shot-set <shot_id> done|pending|failed — 手动设置状态",
|
||
"info": "info <视频> — 视频元数据(时长/分辨率/编码/大小)",
|
||
"characters":"characters <剧本> — 提取角色列表",
|
||
"sub-ab": "sub-ab <视频> <srt> [样式列表] — 字幕A/B测试",
|
||
"ref-sub": "ref-sub <截图> [输出] — 参考字幕样式分析",
|
||
"clean": "clean temp|outputs|db — 清理(仅提示，不实际删除)",
|
||
"flow": "flow [episode_id] — 管线可视化，显示每个镜头的进度和下一步建议",
|
||
"history": "history [数量] — 生成历史+成本一览",
|
||
"report": "report [episode_id] — 生成HTML生产报告",
|
||
"note": "note [shot_id] [内容] — 镜头备注增/查",
|
||
"ref-sub": "ref-sub <截图> [输出] — 参考字幕样式分析",
|
||
"render": "render [episode_id] — 一键成片",
|
||
"compose": "compose [episode_id] — 多镜头拼接",
|
||
"dub": "dub <shot_id> <台词> [角色名] — 配音+自动SRT",
|
||
}
|
||
|
||
if __name__ == "__main__":
|
||
if len(sys.argv) >= 2 and sys.argv[1] in ("--version", "-V", "version"):
|
||
print("LIB · 蛋蛋短剧生产管线 · v2.0 · D181")
|
||
print("cang-ying · 光湖语言世界 · 胖头鱼语言子系统")
|
||
print(f"Python {sys.version.split()[0]} · {len(COMMANDS)} commands")
|
||
sys.exit(0)
|
||
|
||
if len(sys.argv) < 2 or sys.argv[1] not in COMMANDS:
|
||
print("LIB · 蛋蛋短剧生产管线 · v2.0 · D181")
|
||
print(f"Usage: python -m lib.cli <command> [args]")
|
||
print(f"Commands: {', '.join(sorted(COMMANDS.keys()))}")
|
||
if len(sys.argv) > 1:
|
||
print(f"\n❌ 未知命令: {sys.argv[1]}")
|
||
# 模糊匹配建议
|
||
import difflib
|
||
suggestions = difflib.get_close_matches(sys.argv[1], COMMANDS.keys(), n=3)
|
||
if suggestions:
|
||
print(f"💡 你是想说: {', '.join(suggestions)}?")
|
||
print(f"\n💡 python -m lib.cli <cmd> --help 查看命令帮助")
|
||
sys.exit(1)
|
||
|
||
cmd_name = sys.argv[1]
|
||
args = sys.argv[2:]
|
||
|
||
# --help 支持
|
||
if "--help" in args or "-h" in args:
|
||
help_text = HELP_TEXT.get(cmd_name, f"{cmd_name} — 无详细帮助")
|
||
print(f"📖 {help_text}")
|
||
sys.exit(0)
|
||
|
||
# --dry-run 透传 (gen/produce 专用)
|
||
if "--dry-run" in args:
|
||
args = [a for a in args if a != "--dry-run"] + ["true"] if cmd_name == "produce" else args
|
||
|
||
cmd = COMMANDS[cmd_name]
|
||
try:
|
||
cmd(*args)
|
||
except TypeError as e:
|
||
print(f"❌ 参数错误")
|
||
help_text = HELP_TEXT.get(cmd_name, "")
|
||
if help_text:
|
||
print(f"📖 {help_text}")
|
||
sys.exit(1)
|
||
except Exception as e:
|
||
print(f"💥 {type(e).__name__}: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
sys.exit(1)
|