Add MiniMax music generation API client

README.md

@@ -11,6 +11,28 @@ A simple demo offering music generation by prompt and duration
 
 A demo of instant prompt-based music video generation based on Deforum Stable Diffusion colab
 
+### MiniMax music generation API
+
+[`minimax_music.py`](minimax_music.py) provides a small standard-library client for
+the MiniMax `music-3.0` model. It supports both the global and China endpoints and
+returns either a temporary audio URL or hex-encoded audio:
+
+```python
+import os
+from minimax_music import generate_music
+
+result = generate_music(
+    os.environ["MINIMAX_API_KEY"],
+    prompt="An uplifting cinematic theme with strings",
+    output_format="url",
+)
+print(result.audio)
+```
+
+Use `region="cn_zh"` for the China endpoint. That region also accepts the optional
+`aigc_watermark` argument. URL results expire after 24 hours, so download them
+promptly if they need to be retained.
+
 _______________________________________________________________________________________________________________________________________________________
 
 All music is generated by Mubert API – www.mubert.com/. If you want to use it for free – please, tag us on Twitter (https://twitter.com/mubertapp) and include hashtag #mubert.

minimax_music.py

@@ -0,0 +1,95 @@
+"""Small client for MiniMax's synchronous music-generation API."""
+
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass
+from typing import Any
+from urllib import error, request
+
+
+ENDPOINTS = {
+    "global_en": "https://api.minimax.io/v1/music_generation",
+    "cn_zh": "https://api.minimaxi.com/v1/music_generation",
+}
+DEFAULT_MODEL = "music-3.0"
+
+
+class MusicGenerationError(RuntimeError):
+    """Raised when the music-generation service rejects a request."""
+
+
+@dataclass(frozen=True)
+class MusicResult:
+    audio: str
+    status: int
+
+
+def generate_music(
+    api_key: str,
+    *,
+    prompt: str | None = None,
+    lyrics: str | None = None,
+    region: str = "global_en",
+    model: str = DEFAULT_MODEL,
+    output_format: str = "url",
+    audio_format: str = "mp3",
+    is_instrumental: bool | None = None,
+    aigc_watermark: bool | None = None,
+    timeout: float = 120,
+) -> MusicResult:
+    """Generate music and return its URL or hex-encoded audio payload.
+
+    URL results are temporary and should be downloaded within 24 hours.
+    ``aigc_watermark`` is supported only by the ``cn_zh`` endpoint.
+    """
+    if region not in ENDPOINTS:
+        raise ValueError(f"Unsupported region: {region}")
+    if output_format not in {"url", "hex"}:
+        raise ValueError("output_format must be 'url' or 'hex'")
+    if audio_format not in {"mp3", "wav", "pcm"}:
+        raise ValueError("audio_format must be 'mp3', 'wav', or 'pcm'")
+    if aigc_watermark is not None and region != "cn_zh":
+        raise ValueError("aigc_watermark is available only in region='cn_zh'")
+
+    payload: dict[str, Any] = {
+        "model": model,
+        "stream": False,
+        "output_format": output_format,
+        "audio_setting": {"format": audio_format},
+    }
+    for key, value in {
+        "prompt": prompt,
+        "lyrics": lyrics,
+        "is_instrumental": is_instrumental,
+        "aigc_watermark": aigc_watermark,
+    }.items():
+        if value is not None:
+            payload[key] = value
+
+    http_request = request.Request(
+        ENDPOINTS[region],
+        data=json.dumps(payload).encode("utf-8"),
+        headers={
+            "Authorization": f"Bearer {api_key}",
+            "Content-Type": "application/json",
+        },
+        method="POST",
+    )
+    try:
+        with request.urlopen(http_request, timeout=timeout) as response:
+            response_data = json.load(response)
+    except (error.HTTPError, error.URLError, json.JSONDecodeError) as exc:
+        raise MusicGenerationError(f"Music generation request failed: {exc}") from exc
+
+    base_response = response_data.get("base_resp", {})
+    if base_response.get("status_code") != 0:
+        message = base_response.get("status_msg", "unknown service error")
+        raise MusicGenerationError(message)
+
+    data = response_data.get("data", {})
+    status = data.get("status")
+    audio = data.get("audio")
+    if status != 2 or not audio:
+        raise MusicGenerationError(f"Music generation did not complete (status={status!r})")
+    return MusicResult(audio=audio, status=status)

test_minimax_music.py

@@ -0,0 +1,59 @@
+import io
+import json
+import unittest
+from unittest.mock import patch
+
+import minimax_music
+
+
+class FakeResponse(io.BytesIO):
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *args):
+        self.close()
+
+
+class GenerateMusicTests(unittest.TestCase):
+    @patch("minimax_music.request.urlopen")
+    def test_global_request_and_response(self, urlopen):
+        urlopen.return_value = FakeResponse(json.dumps({
+            "data": {"status": 2, "audio": "https://example.test/song.mp3"},
+            "base_resp": {"status_code": 0},
+        }).encode())
+
+        result = minimax_music.generate_music("test-key", prompt="calm piano")
+
+        sent = urlopen.call_args.args[0]
+        body = json.loads(sent.data)
+        self.assertEqual(sent.full_url, minimax_music.ENDPOINTS["global_en"])
+        self.assertEqual(sent.headers["Authorization"], "Bearer test-key")
+        self.assertEqual(body["model"], "music-3.0")
+        self.assertEqual(body["audio_setting"], {"format": "mp3"})
+        self.assertEqual(result.audio, "https://example.test/song.mp3")
+
+    @patch("minimax_music.request.urlopen")
+    def test_china_region_supports_watermark(self, urlopen):
+        urlopen.return_value = FakeResponse(json.dumps({
+            "data": {"status": 2, "audio": "deadbeef"},
+            "base_resp": {"status_code": 0},
+        }).encode())
+
+        minimax_music.generate_music(
+            "test-key", region="cn_zh", lyrics="la la", output_format="hex",
+            audio_format="wav", aigc_watermark=True,
+        )
+
+        sent = urlopen.call_args.args[0]
+        body = json.loads(sent.data)
+        self.assertEqual(sent.full_url, minimax_music.ENDPOINTS["cn_zh"])
+        self.assertTrue(body["aigc_watermark"])
+        self.assertEqual(body["output_format"], "hex")
+
+    def test_rejects_region_specific_field_on_global_endpoint(self):
+        with self.assertRaises(ValueError):
+            minimax_music.generate_music("test-key", aigc_watermark=True)
+
+
+if __name__ == "__main__":
+    unittest.main()