#!/usr/bin/env python3 # -*- coding: utf-8 -*- """Submit two LibTV Eleven Music V3 audio nodes for the coin-pouch product video.""" import json import subprocess import time from pathlib import Path LIBTV = "/Users/bot1/.local/bin/libtv" PROJECT_ID = "6739cc5ca9d1415ba4a82b9673f3931e" PROJECT_ROOT = Path("/Users/bot1/Volumes/root_for_ai/AI工作区/通用_产品宣传视频_古钱币杜邦纸钱袋包_20260530_1702") OUT_DIR = PROJECT_ROOT / "04_audio_bgm" / "libtv_music_20260605" LOG_DIR = PROJECT_ROOT / "prompts" / "audio_bgm" / "libtv_music_20260605" SUMMARY = LOG_DIR / "submission_summary.json" OUT_DIR.mkdir(parents=True, exist_ok=True) LOG_DIR.mkdir(parents=True, exist_ok=True) RUN_ID = time.strftime("%Y%m%d_%H%M%S") PROMPTS = [ { "code": "E_bright_premium_product", "label": "E｜明亮高级产品片", "node": f"钱袋子BGM_明亮高级产品片_{RUN_ID}", "prompt": "30-second instrumental background music for a bright premium cultural creative product film. Warm, elegant, emotional, product-commercial friendly. Ancient coins become lively and a yellow DuPont paper coin pouch becomes the hero product. Light marimba or pizzicato pulse, soft modern synth pad, warm bass, delicate metallic sparkle, gentle uplifting chord movement, clear positive emotion. Modern oriental craft feeling, bright and refined. No vocals, no lyrics, not dark, not suspenseful, not museum ambient, not sad, not epic trailer.", "x": "1000", "y": "0", }, { "code": "F_lively_coin_product_arc", "label": "F｜钱币小生命产品转场", "node": f"钱袋子BGM_钱币小生命产品转场_{RUN_ID}", "prompt": "30-second instrumental background music for a lively polished product storytelling video. Different ancient coin characters walk forward with tiny rounded arms and legs, then the yellow DuPont paper coin pouch gathers them and ends in a clean product showcase. Gentle plucked strings, tiny rhythmic metal clicks like coin footsteps, soft hand percussion, bright bell accents, warm cinematic chord lift, satisfying final resolve. Playful but not childish, premium but not cold. No vocals, no lyrics, not dark, not empty ambient, not cartoon comedy, no heavy drums.", "x": "1000", "y": "360", }, ] def run_cmd(args, timeout=600, check=False): started = time.time() res = subprocess.run(args, text=True, capture_output=True, timeout=timeout) return { "args": args, "returncode": res.returncode, "stdout": res.stdout, "stderr": res.stderr, "elapsed_sec": round(time.time() - started, 2), "ok": res.returncode == 0, } def ffprobe(path: Path): try: res = subprocess.run( ["ffprobe", "-v", "error", "-show_entries", "format=duration", "-of", "default=nk=1:nw=1", str(path)], text=True, capture_output=True, timeout=30, ) if res.returncode != 0: return {"ok": False, "stderr": res.stderr} return {"ok": True, "duration": float((res.stdout or "0").strip() or 0)} except Exception as e: return {"ok": False, "error": str(e)} def download_new(node_name: str, before_names): cmd = [LIBTV, "download", "-p", PROJECT_ID, "-n", node_name, "-o", str(OUT_DIR)] res = run_cmd(cmd, timeout=180) after = {p.name for p in OUT_DIR.glob("*") if p.is_file()} new_files = sorted(after - before_names) return res, [str(OUT_DIR / name) for name in new_files] def main(): summary = { "status": "starting", "run_id": RUN_ID, "project_id": PROJECT_ID, "project_url": f"https://www.liblib.tv/canvas?projectId={PROJECT_ID}", "model": "vocal-music / Eleven Music V3", "duration_ms": 30000, "outputs_dir": str(OUT_DIR), "items": [], } for item in PROMPTS: prompt_path = LOG_DIR / f"{item['code']}_prompt.txt" prompt_path.write_text(item["prompt"], encoding="utf-8") create_cmd = [ LIBTV, "node", "--x", item["x"], "--y", item["y"], "create", item["node"], "-p", PROJECT_ID, "-t", "audio", "--prompt", item["prompt"], "-s", "scene=Music", "-s", "model=vocal-music", "-s", "duration=30000", "--run", ] create_res = run_cmd(create_cmd, timeout=600) record = { "code": item["code"], "label": item["label"], "node": item["node"], "prompt_path": str(prompt_path), "create": create_res, "download_attempts": [], "downloaded_files": [], "ffprobe": [], } summary["items"].append(record) SUMMARY.write_text(json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8") if not create_res["ok"]: continue before = {p.name for p in OUT_DIR.glob("*") if p.is_file()} # Try for up to ~5 minutes. Some nodes become downloadable immediately after create --run; others need polling. for attempt in range(1, 21): dl_res, new_files = download_new(item["node"], before) record["download_attempts"].append({ "attempt": attempt, "returncode": dl_res["returncode"], "stdout_tail": dl_res["stdout"][-2000:], "stderr_tail": dl_res["stderr"][-2000:], "new_files": new_files, }) if new_files: record["downloaded_files"].extend(new_files) break # If download says not ready / no resource, wait and try again. time.sleep(15) SUMMARY.write_text(json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8") for f in record["downloaded_files"]: p = Path(f) record["ffprobe"].append({"file": f, "bytes": p.stat().st_size if p.exists() else None, "probe": ffprobe(p) if p.exists() else {"ok": False, "error": "missing"}}) SUMMARY.write_text(json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8") summary["status"] = "completed" SUMMARY.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()