Mureka V9.5 批量生成配乐的 Python 脚本骨架

做 Mureka V9.5 的批量配乐生成，核心不是先写生成循环，而是先确认你能用什么方式提交生成任务。Mureka V9.5 目前没有公开的官方 API，脚本只能做成可替换接入点的骨架：根据你的账号权限，去对网页端接口、平台 API 或第三方封装做适配。下面给出请求构造、任务轮询和结果下载的 Python 骨架，并标注每段代码里必须人工确认的字段。

判断：Mureka V9.5 没有官方公开 API，批量生成脚本只能作为 HTTP 客户端骨架使用。操作动作：先确认鉴权方式和任务状态字段，再写轮询与下载；验证方式：用单条任务确认响应和文件格式。适用场景：已拥有可用接口地址和密钥的开发者。风险边界：请求体、状态取值、音频返回格式需要按实际接口替换，否则脚本无法运行。

确定可用的接入方式与鉴权字段

先分清接入方式。网页端接口通常要带 Cookie、Referer，并且登录态会过期；平台 API 一般要有 api_key 或 access_token；第三方封装可能把请求简化了，但字段和限额不透明。脚本里只放占位符，密钥从环境变量读。

import os

api_key = os.environ.get('MUREKA_API_KEY')
api_secret = os.environ.get('MUREKA_API_SECRET')
access_token = os.environ.get('MUREKA_ACCESS_TOKEN')
web_cookie = os.environ.get('MUREKA_WEB_COOKIE')

接入 URL 用 MUREKA_API_URL 单独存放，不要写死 key 或 secret 到代码里。

构造生成请求的 JSON 数据

生成请求的 JSON 字段经常因为版本不同而改名，比如情绪、曲风、BPM、时长、人声类型。以下是请求体骨架，字段代表含义，不是真实接口字段名。

payload = {
    'style': 'pop',
    'lyrics': '...',
    'bpm': 120,
    'duration': 30,
    'instrument': 'piano',
    'mood': 'upbeat',
    'voice': 'female',
}

style、mood 这类枚举值需要从实际产品候选列表里取，不能自己随便填；lyrics 从你的歌词文件读取。提交前先打印一次 payload，和抓包或文档里的请求体对比，缺哪个补哪个。

发送任务并轮询生成状态

不要在一次 HTTP 请求里等生成完成，Mureka V9.5 这类生成任务通常要几十秒甚至更久，同步等待会超时。正确做法是提交后拿到任务 ID，再按间隔轮询。

import time
import requests

url = os.environ.get('MUREKA_API_URL')
headers = {'Authorization': 'Bearer ' + access_token}  # 按实际鉴权格式替换

resp = requests.post(url, json=payload, headers=headers, timeout=30)
resp.raise_for_status()
task = resp.json()
task_id = task.get('task_id')
if not task_id:
    raise ValueError('响应里没有 task_id: ' + str(task))

while True:
    status_resp = requests.get(url + '/' + task_id, headers=headers, timeout=30)
    status_data = status_resp.json()
    state = status_data.get('status')
    if state == 'success':
        result_url = status_data.get('result_url')
        break
    if state == 'failed':
        raise RuntimeError(status_data.get('error_msg'))
    time.sleep(5)

这里的 Authorization 只是常见格式，不是 Mureka V9.5 的固定格式；状态值 pending、processing、success、failed 也需要按实际响应替换。如果查询接口支持回调地址，优先接回调，减少轮询压力。

下载并保存返回的音频文件

拿到的结果可能是可下载 URL，也可能是二进制流。先确认 response headers 里的 content-type，再选择保存方式。

import pathlib
import base64

if result_url:
    audio = requests.get(result_url, timeout=60).content
else:
    audio = base64.b64decode(status_data.get('audio_base64'))

out_dir = pathlib.Path('output')
out_dir.mkdir(exist_ok=True)
out_path = out_dir / ('track_' + task_id + '.mp3')
out_path.write_bytes(audio)

如果返回的是流式响应，就用 requests 的 iter_content 分块写文件，不要一次性读进内存，避免大文件占满内存。

批量处理中的失败重试与日志记录

批量处理的关键是单条失败不能拖死整批。每首配乐单独调用一个带重试的函数，失败记日志，继续下一条。

import logging

logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s')

def submit_with_retry(url, payload, headers, max_retries=3):
    for attempt in range(max_retries):
        try:
            resp = requests.post(url, json=payload, headers=headers, timeout=30)
            resp.raise_for_status()
            return resp.json()
        except requests.RequestException as exc:
            logging.warning('提交失败，第 %s 次重试: %s', attempt + 1, exc)
            time.sleep(2 ** attempt)
    raise RuntimeError('提交重试耗尽')

批量循环里单独捕获每条任务异常，失败写日志后 continue，不要中断整个脚本。重试次数和退避间隔建议从配置读，按实际接口限流情况调整。