249 lines
8.9 KiB
Python
249 lines
8.9 KiB
Python
# LIB · 蛋蛋短剧生产管线 · 命令行入口
|
||
# D180 · 2026-07-10
|
||
# Usage: python -m lib.cli <command> [args]
|
||
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_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}")
|
||
|
||
COMMANDS = {
|
||
"init": cmd_init,
|
||
"shots": cmd_shot_list,
|
||
"shot": cmd_shot_status,
|
||
"assets": cmd_asset_list,
|
||
"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_generate,
|
||
"collect": cmd_collect,
|
||
}
|
||
|
||
def cmd_generate(shot_id):
|
||
"""提交 Seedance 生成任务（异步）"""
|
||
from lib.seedance import generate_shot_async
|
||
from lib.models import get_db
|
||
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} 已完成，需爸爸确认是否重跑")
|
||
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
|
||
|
||
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} 已提交 · task_id: {r['task_id'][:20]}...")
|
||
print(f" 轮询: python -m lib.cli collect {shot_id}")
|
||
|
||
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} 完成 → {r['output_path']}")
|
||
elif r.get("status") == "running":
|
||
print(f"🔄 {shot_id} 生成中（ID: {r['task_id'][:20]}...）")
|
||
else:
|
||
print(f"❓ {r}")
|
||
|
||
if __name__ == "__main__":
|
||
if len(sys.argv) < 2 or sys.argv[1] not in COMMANDS:
|
||
print("LIB · 蛋蛋短剧生产管线 · D180")
|
||
print("Usage: python -m lib.cli <command> [args]")
|
||
print(f"Commands: {', '.join(COMMANDS.keys())}")
|
||
sys.exit(1)
|
||
cmd = COMMANDS[sys.argv[1]]
|
||
args = sys.argv[2:]
|
||
try:
|
||
cmd(*args)
|
||
except TypeError as e:
|
||
print(f"参数错误: {e}")
|
||
print(f"Usage: python -m lib.cli {sys.argv[1]} <args>")
|