#!/usr/bin/env python3 # -*- coding: utf-8 -*- """Submit four instrumental Suno.cn BGM directions for the coin-pouch video. Uses only the REST API documented in the suno-cn-music skill. """ import json import os import pathlib import re import sys import traceback import urllib.error import urllib.parse import urllib.request BASE = "https://mcp.suno.cn" API_KEY = os.environ.get("SUNO_CN_API_KEY", "") PROJECT = pathlib.Path("/Users/bot1/Volumes/root_for_ai/AI工作区/通用_产品宣传视频_古钱币杜邦纸钱袋包_20260530_1702") OUT_DIR = PROJECT / "04_audio_bgm" LOG_DIR = PROJECT / "prompts" / "audio_bgm" OUT_DIR.mkdir(parents=True, exist_ok=True) LOG_DIR.mkdir(parents=True, exist_ok=True) DIRECTIONS = [ { "code": "A_artifact_awakening", "label": "A｜文物苏醒感", "title": "Coin Pouch Artifact Awakening Instrumental", "prompt": "Short instrumental background music for a 15-second product video: ancient Chinese coins awaken in a dim cinematic tabletop world. Mysterious but refined museum atmosphere, low warm drone, sparse guqin-like plucks, soft stone chime, tiny bronze shimmer, subtle air movement. Slow opening, gentle magical rise, no vocals, no singing, no lyrics, no pop beat, not epic, not horror. Clean intro suitable for artifact display and transformation.", "tags": "instrumental, cinematic ambient, ancient Chinese, guqin, stone chime, bronze shimmer, museum, mysterious, refined, no vocals", }, { "code": "B_coin_little_life", "label": "B｜钱币小生命感", "title": "Coin Pouch Little Life Instrumental", "prompt": "Short instrumental background music for a 15-second product video: different ancient coins grow tiny rounded arms and legs and walk forward loosely on a tabletop. Light, clever, charming but not childish. Gentle pizzicato strings, soft woodblock taps, tiny metal clicks, warm handpan accents, subtle playful rhythm like little footsteps. No vocals, no singing, no lyrics, no cartoon comedy, no big drums.", "tags": "instrumental, light cinematic, pizzicato, woodblock, tiny metal clicks, playful, warm, charming, no vocals", }, { "code": "C_premium_product_showcase", "label": "C｜产品广告高级感", "title": "Coin Pouch Premium Product Showcase Instrumental", "prompt": "Short instrumental background music for a 15-second premium product advertisement: a DuPont paper coin pouch lands, catches the walking coins, its patterns softly glow, then the product is shown cleanly. Minimal cinematic electronic bed, soft pulse, elegant metal clicks, warm sub bass, restrained luxury, modern clean commercial finish. No vocals, no singing, no lyrics, no strong dance beat, no epic trailer drums.", "tags": "instrumental, premium product ad, minimal cinematic electronic, elegant, warm sub bass, metal clicks, modern, clean, no vocals", }, { "code": "D_mixed_story_arc", "label": "D｜推荐混合版", "title": "Coin Pouch Story Arc Instrumental", "prompt": "Short instrumental background music for a 15-second product storytelling video with three clear emotional sections: 1) mysterious ancient coin display, 2) tiny coins become alive and walk forward, 3) DuPont paper coin pouch appears and becomes a premium product hero shot. Seamless story arc, ancient-to-modern transition, low ambient drone, guqin-like plucks, tiny footstep percussion, soft metal shimmer, elegant final resolve. Pure instrumental, no vocals, no singing, no lyrics, no pop song structure.", "tags": "instrumental, cinematic story arc, ancient to modern, guqin, ambient, tiny percussion, metal shimmer, premium resolve, no vocals", }, ] def redact(text: str) -> str: if API_KEY: text = text.replace(API_KEY, "********") text = re.sub(r"sk-[A-Za-z0-9_\-]+", "sk-********", text) return text def request_json(method: str, path: str, body=None, timeout=60): url = BASE + path data = None headers = {"Authorization": f"Bearer {API_KEY}"} if body is not None: data = json.dumps(body, ensure_ascii=False).encode("utf-8") headers["Content-Type"] = "application/json; charset=utf-8" req = urllib.request.Request(url, data=data, headers=headers, method=method) try: with urllib.request.urlopen(req, timeout=timeout) as resp: raw = resp.read().decode("utf-8", errors="replace") try: parsed = json.loads(raw) except Exception: parsed = {"_raw": raw} return {"ok": True, "status": resp.status, "headers": dict(resp.headers), "json": parsed, "url": url, "method": method} except urllib.error.HTTPError as e: raw = e.read().decode("utf-8", errors="replace") return {"ok": False, "status": e.code, "headers": dict(e.headers), "body": raw, "url": url, "method": method, "error_type": "HTTPError", "error": str(e)} except Exception as e: return {"ok": False, "status": None, "headers": {}, "body": "", "url": url, "method": method, "error_type": type(e).__name__, "error": str(e), "trace": traceback.format_exc()} def save_error_response(err): return { "method": err.get("method"), "url": err.get("url"), "status": err.get("status"), "error_type": err.get("error_type"), "error": err.get("error"), "headers": err.get("headers"), "body": err.get("body"), "trace": err.get("trace"), } def main(): if not API_KEY: print(json.dumps({"ok": False, "error": "NO_API_KEY", "message": "SUNO_CN_API_KEY is not set"}, ensure_ascii=False)) return 2 submitted = [] errors = [] for d in DIRECTIONS: body = { "prompt": d["prompt"], "mv": "chirp-fenix", "title": d["title"], "tags": d["tags"], "custom_mode": False, "instrumental": True, } (LOG_DIR / f"{d['code']}_request.json").write_text(json.dumps(body, ensure_ascii=False, indent=2), encoding="utf-8") res = request_json("POST", "/mcp/api/generate", body=body, timeout=60) if not res["ok"] or res.get("status") != 200: errors.append({"direction": d, "response": save_error_response(res)}) continue data = res.get("json") or {} serials = data.get("serial_nos") or data.get("serialNos") or data.get("serial_no") or [] if isinstance(serials, str): serials = [serials] submitted.append({"direction": d, "serials": [str(s) for s in serials], "message": data.get("message"), "raw": data}) all_serials = [s for item in submitted for s in item["serials"] if s] query = None saved = [] tasks_summary = [] if all_serials: path_ids = ",".join(urllib.parse.quote(s, safe="") for s in all_serials) query = request_json("GET", f"/mcp/api/task/{path_ids}?wait=45", timeout=75) if query.get("ok") and query.get("status") == 200: qj = query.get("json") or {} tasks = qj.get("tasks") or qj.get("data") or qj.get("list") or [] if isinstance(tasks, dict): tasks = [tasks] serial_to_direction = {} for item in submitted: for s in item["serials"]: serial_to_direction[s] = item["direction"] for task in tasks if isinstance(tasks, list) else []: serial = str(task.get("serial_no") or task.get("serialNo") or task.get("id") or "") direction = serial_to_direction.get(serial, {"code": "unknown", "label": "unknown"}) status = str(task.get("status") or "") play_url = task.get("play_url") or task.get("audio_url") or task.get("url") item = { "serial_no": serial, "direction_code": direction.get("code"), "direction_label": direction.get("label"), "status": status, "title": task.get("title"), "duration": task.get("duration"), "play_url": play_url, "fail_reason": task.get("fail_reason"), } tasks_summary.append(item) if status.lower() == "success" and play_url: local_path = OUT_DIR / f"coin_pouch_{direction.get('code')}_{serial}.mp3" try: with urllib.request.urlopen(play_url, timeout=90) as r: content = r.read() local_path.write_bytes(content) item["local_path"] = str(local_path) item["bytes"] = len(content) saved.append(item) except Exception as e: item["download_error"] = f"{type(e).__name__}: {e}" summary = { "ok": len(errors) == 0, "submitted": [ {"direction_code": item["direction"]["code"], "direction_label": item["direction"]["label"], "serials": item["serials"], "message": item["message"]} for item in submitted ], "errors": errors, "queried_serials": all_serials, "query_status": None if query is None else query.get("status"), "query_ok": None if query is None else query.get("ok"), "query_error": None if not query or query.get("ok") else save_error_response(query), "tasks": tasks_summary, "saved": saved, "out_dir": str(OUT_DIR), "prompt_log_dir": str(LOG_DIR), } summary_path = LOG_DIR / "suno_four_directions_submission_summary_20260605.json" summary_path.write_text(redact(json.dumps(summary, ensure_ascii=False, indent=2)), encoding="utf-8") print(redact(json.dumps(summary, ensure_ascii=False, indent=2))) return 0 if not errors else 1 if __name__ == "__main__": sys.exit(main())