#!/usr/bin/env python3 # -*- coding: utf-8 -*- """Submit two LibTV Eleven Music V3 BGM variants based on two reference-audio rhythm targets.""" 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") RUN_ID = time.strftime("%Y%m%d_%H%M%S") OUT_DIR = PROJECT_ROOT / "04_audio_bgm" / f"libtv_reference_midtempo_{RUN_ID}" LOG_DIR = PROJECT_ROOT / "prompts" / "audio_bgm" / f"libtv_reference_midtempo_{RUN_ID}" SUMMARY = LOG_DIR / "submission_summary.json" OUT_DIR.mkdir(parents=True, exist_ok=True) LOG_DIR.mkdir(parents=True, exist_ok=True) REFERENCE_ANALYSIS = { "audio_caf06134b73d.mp3": { "duration_sec": 19.27, "estimated_bpm_candidates": [147, 146, 145, 144, 143], "role": "lower / steadier reference side" }, "audio_6acbeae60412.mp3": { "duration_sec": 24.74, "estimated_bpm_candidates": [178, 177, 176, 175, 174], "role": "faster / brighter reference side" }, "target": "two instrumental variants between the references, usable for a 15-second coin-pouch product film" } PROMPTS = [ { "code": "G_midtempo_refined_product", "label": "G｜参考折中-高级产品片", "node": f"钱袋子BGM_参考折中高级产品片_{RUN_ID}", "x": "1400", "y": "0", "prompt": ( "30-second instrumental background music for a short vertical cultural creative product film. " "Rhythm sits between two references: lower reference feels around 145-147 BPM, faster reference feels around 174-178 BPM; target a polished 155-165 BPM midtempo bounce with a calm half-time product-film warmth. " "Mood bright, warm, elegant, premium, emotionally friendly, modern oriental craft. " "Story arc: ancient coins wake up, gently walk forward together, a yellow DuPont paper coin pouch arrives, warm gold light reveals product texture, ending in a clean hero product shot. " "Arrangement: light marimba or plucked strings pulse, tiny metallic coin-click percussion, soft hand percussion, warm bass, gentle synth pad, delicate bell sparkles, subtle uplifting chord movement, satisfying clean final resolve. " "No vocals, no lyrics, no rap, no heavy drums, not dark, not suspenseful, not empty museum ambient, not childish cartoon comedy." ), }, { "code": "H_midtempo_lively_coin_story", "label": "H｜参考折中-钱币小生命", "node": f"钱袋子BGM_参考折中钱币小生命_{RUN_ID}", "x": "1400", "y": "360", "prompt": ( "30-second instrumental background music for a lively but premium coin-pouch product storytelling video. " "Use a tempo and energy between two references: not as steady/low as 145 BPM, not as fast/busy as 178 BPM; target around 160 BPM feel, light bounce, edit-friendly for 15 seconds. " "Different ancient coin characters with tiny rounded arms and legs walk forward together, slightly playful but refined; the yellow DuPont paper coin pouch softly catches them and becomes the final product hero. " "Arrangement: nimble plucked strings, small coin-footstep metal clicks, soft brush percussion, warm rounded bass, bright bell accents, gentle cinematic chord lift. " "Mood: bright, warm, lively, polished, delightful, premium product-commercial, modern oriental craft. " "No vocals, no lyrics, no singing, no heavy drums, not dark, not horror, not childish cartoon music, not empty ambient." ), }, ] def run_cmd(args, timeout=600): started = time.time() try: 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, } except Exception as e: return { "args": args, "returncode": None, "stdout": "", "stderr": f"{type(e).__name__}: {e}", "elapsed_sec": round(time.time() - started, 2), "ok": False, } def afinfo(path: Path): res = run_cmd(["/usr/bin/afinfo", str(path)], timeout=30) duration = None if res["ok"]: for line in res["stdout"].splitlines(): if "estimated duration:" in line: try: duration = float(line.split("estimated duration:", 1)[1].split("sec", 1)[0].strip()) except Exception: duration = None break return {"ok": res["ok"], "duration_sec": duration, "stdout_tail": res["stdout"][-1200:], "stderr_tail": res["stderr"][-1200:]} 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(): (LOG_DIR / "reference_analysis.json").write_text(json.dumps(REFERENCE_ANALYSIS, ensure_ascii=False, indent=2), encoding="utf-8") 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, "reference_analysis": REFERENCE_ANALYSIS, "outputs_dir": str(OUT_DIR), "prompt_dir": str(LOG_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": [], "afinfo": [], } 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()} for attempt in range(1, 25): 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, }) SUMMARY.write_text(json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8") if new_files: record["downloaded_files"].extend(new_files) break time.sleep(15) for f in record["downloaded_files"]: p = Path(f) record["afinfo"].append({"file": f, "bytes": p.stat().st_size if p.exists() else None, "probe": afinfo(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()