#!/usr/bin/env python3 import json import re import sys import time from pathlib import Path PROJECT_ROOT = Path('/Users/bot1/Volumes/root_for_ai/AI工作区/通用_产品宣传视频_古钱币杜邦纸钱袋包_20260530_1702') SKILL_SCRIPTS = Path('/Users/bot1/.hermes/backups/seedance_default_remove_libtv_20260522_184712/locked-skills/creative/libtv-skill/scripts') sys.path.insert(0, str(SKILL_SCRIPTS)) from upload_file import upload_file from _common import create_session, build_project_url RUN_ID = '20260530_174949' SHEETS_DIR = PROJECT_ROOT / 'deliverables/storyboard_sheets_20260530' PROMPT_MD = SHEETS_DIR / '动态视频提示词_按4组分镜_20260530.md' OUT_DIR = PROJECT_ROOT / 'outputs/libtv-seedance-4x15-20260530_174949' SUMMARY_PATH = OUT_DIR / 'submission_summary.json' REGISTRY_PATH = Path('/Users/bot1/.hermes/profiles/video/state/libtv/project_registry.json') USER_CANVAS_PROJECT_ID = '6739cc5ca9d1415ba4a82b9673f3931e' SCENES = [ (1, '古代文物造型转现代、轻颤苏醒、站起卡通化', SHEETS_DIR / '合并分镜1_2x2.png'), (2, '刚成人的钱币小角色行走、一枚受惊轻跃', SHEETS_DIR / '合并分镜2_2x2.png'), (3, '三枚硬币逃跑、被钱袋包吸成金色光点并成为纹样', SHEETS_DIR / '合并分镜3_2x2.png'), (4, '钱袋包纹样亮起、手提与背包产品展示', SHEETS_DIR / '合并分镜4_2x2.png'), ] def extract_scene_section(text: str, n: int) -> str: start_pat = rf'^# 分镜{n}｜.*$' m = re.search(start_pat, text, flags=re.M) if not m: raise RuntimeError(f'cannot find section for scene {n}') start = m.start() m_next = re.search(r'^# 分镜\d+｜|^# 可选：整条连续视频提交稿', text[m.end():], flags=re.M) end = m.end() + m_next.start() if m_next else len(text) return text[start:end].strip() def make_message(scene_no: int, title: str, section: str, ref_url: str) -> str: marker = f'COIN_POUCH_SEGMENT_{scene_no:02d}_{RUN_ID}' return f'''【{marker}】 请使用 Seedance 2.0 生成第 {scene_no} 段视频，时长 15 秒，比例 3:4，清晰度 720p。 本段主题：{title} 参考分镜图：{ref_url} 重要执行要求： - 参考分镜图只作为镜头顺序、主体特征、动作状态和产品结构参考；成片必须是连续单画面视频，不要生成九宫格、分镜板、拼贴图或画中画。 - 严格按下面提示词生成，不要新增分镜里没有的角色、logo、文字、产品结构或夸张机械装置。 - 钱币角色和黄色钱袋包的外形、材质、颜色、纹样以参考分镜图为准。 - 声音只生成环境声和动作音效，不生成背景音乐，不生成旋律型 BGM。 {section} ''' def update_registry(project_uuid: str, session_id: str): try: if REGISTRY_PATH.exists(): data = json.loads(REGISTRY_PATH.read_text(encoding='utf-8')) else: data = [] key = 'coin-pouch-video' found = False for item in data: if item.get('project_key') == key: item.update({ 'display_name': '古钱币杜邦纸钱袋包产品宣传视频', 'libtv_project_uuid': project_uuid, 'project_url': build_project_url(project_uuid), 'session_id': session_id, 'last_used_at': time.strftime('%Y-%m-%d'), 'status': 'active', }) aliases = set(item.get('aliases') or []) aliases.update(['钱袋子视频', '古钱币', '杜邦纸包', '钱袋包', 'coin pouch']) item['aliases'] = sorted(aliases) found = True break if not found: data.append({ 'project_key': key, 'display_name': '古钱币杜邦纸钱袋包产品宣传视频', 'libtv_project_uuid': project_uuid, 'project_url': build_project_url(project_uuid), 'session_id': session_id, 'aliases': ['钱袋子视频', '古钱币', '杜邦纸包', '钱袋包', 'coin pouch'], 'status': 'active', 'last_used_at': time.strftime('%Y-%m-%d'), }) REGISTRY_PATH.parent.mkdir(parents=True, exist_ok=True) REGISTRY_PATH.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding='utf-8') except Exception as e: print(f'WARN registry_update_failed: {e}', file=sys.stderr) def main(): OUT_DIR.mkdir(parents=True, exist_ok=True) if not PROMPT_MD.exists(): raise SystemExit(f'prompt_missing: {PROMPT_MD}') prompt_text = PROMPT_MD.read_text(encoding='utf-8') uploaded = [] for scene_no, title, path in SCENES: if not path.exists(): raise SystemExit(f'storyboard_sheet_missing: {path}') up = upload_file(str(path)) url = up.get('url') if not url: raise SystemExit(f'upload_failed_no_url: scene {scene_no}') uploaded.append({'scene_no': scene_no, 'title': title, 'path': str(path), 'url': url}) session_id = '' project_uuid = '' submitted = [] for item in uploaded: scene_no = item['scene_no'] title = item['title'] section = extract_scene_section(prompt_text, scene_no) msg = make_message(scene_no, title, section, item['url']) msg_path = OUT_DIR / f'segment_{scene_no:02d}_libtv_message.txt' msg_path.write_text(msg, encoding='utf-8') data = create_session(session_id=session_id, message=msg) session_id = data.get('sessionId') or session_id project_uuid = data.get('projectUuid') or project_uuid if not session_id: raise SystemExit(f'no_session_returned_after_scene_{scene_no}') submitted.append({ 'scene_no': scene_no, 'title': title, 'marker': f'COIN_POUCH_SEGMENT_{scene_no:02d}_{RUN_ID}', 'reference_url': item['url'], 'message_path': str(msg_path), 'session_id': session_id, 'project_uuid': project_uuid, 'project_url': build_project_url(project_uuid), 'submitted_at': time.strftime('%Y-%m-%dT%H:%M:%S%z'), }) # Small spacing keeps the session messages in order and avoids hammering the API. time.sleep(5) if project_uuid and session_id: update_registry(project_uuid, session_id) summary = { 'status': 'submitted', 'run_id': RUN_ID, 'user_canvas_project_id': USER_CANVAS_PROJECT_ID, 'user_canvas_url': build_project_url(USER_CANVAS_PROJECT_ID), 'session_id': session_id, 'project_uuid': project_uuid, 'project_url': build_project_url(project_uuid), 'output_dir': str(OUT_DIR), 'prompt_md': str(PROMPT_MD), 'submitted_segments': submitted, 'created_at': time.strftime('%Y-%m-%dT%H:%M:%S%z'), } SUMMARY_PATH.write_text(json.dumps(summary, ensure_ascii=False, indent=2), encoding='utf-8') print(json.dumps(summary, ensure_ascii=False, indent=2)) if __name__ == '__main__': main()