fix(minimax): satisfy voice clone CI checks

This commit is contained in:
octo-patch 2026-10-30 16:49:51 +08:00
parent 41db04bbed
commit ece1047198
5 changed files with 59 additions and 21 deletions

View file

@ -6,7 +6,7 @@ from .transformation import (
VoiceCloneResponse,
)
__all__ = [ # mutable-ok: public export list is intentionally mutable for package compatibility
__all__ = [ # mutable-ok: public export list is intentionally mutable for package compatibility
"MinimaxVoiceCloneConfig",
"MinimaxVoiceCloneError",
"VoiceCloneResponse",

View file

@ -27,9 +27,9 @@ from litellm.types.llms.openai import FileTypes
class VoiceCloneResponse(TypedDict):
"""Normalized response returned after a successful voice clone."""
file_id: str # writable-ok: normalized response is intentionally a mutable compatibility payload
voice_id: str # writable-ok: normalized response is intentionally a mutable compatibility payload
model: str # writable-ok: normalized response is intentionally a mutable compatibility payload
file_id: str # writable-ok: normalized response is intentionally a mutable compatibility payload
voice_id: str # writable-ok: normalized response is intentionally a mutable compatibility payload
model: str # writable-ok: normalized response is intentionally a mutable compatibility payload
class MinimaxVoiceCloneError(BaseLLMException):
@ -62,19 +62,23 @@ class MinimaxVoiceCloneConfig:
def get_complete_url(cls, api_base: str | None = None, operation: str = "clone") -> str:
"""Return the endpoint URL for ``upload`` or ``clone``."""
try:
path = {"upload": cls.FILE_UPLOAD_PATH, "clone": cls.VOICE_CLONE_PATH}[operation] # mutable-ok: operation lookup is a small local dispatch table
path = {"upload": cls.FILE_UPLOAD_PATH, "clone": cls.VOICE_CLONE_PATH}[
operation
] # mutable-ok: operation lookup is a small local dispatch table
except KeyError as exc:
raise ValueError("operation must be 'upload' or 'clone'") from exc
return f"{cls._base_url(api_base)}{path}"
@staticmethod
def validate_environment(headers: dict[str, str] | None = None, api_key: str | None = None) -> dict[str, str]: # mutable-ok: provider payloads and public headers require mutable mappings
def validate_environment(
headers: dict[str, str] | None = None, api_key: str | None = None
) -> dict[str, str]: # mutable-ok: provider payloads and public headers require mutable mappings
"""Add MiniMax authentication while preserving caller headers."""
resolved_key = api_key or litellm.api_key or get_secret_str("MINIMAX_API_KEY")
if not resolved_key:
raise ValueError("MiniMax API key is required for voice cloning")
result = dict(headers or {}) # mutable-ok: caller headers must remain independently mutable
result = dict(headers or {}) # mutable-ok: caller headers must remain independently mutable
result["Authorization"] = f"Bearer {resolved_key}"
return result
@ -83,7 +87,9 @@ class MinimaxVoiceCloneConfig:
cls,
file: FileTypes,
purpose: str = "voice_clone",
) -> tuple[dict[str, tuple[str, bytes, str]], dict[str, str]]: # mutable-ok: provider payloads and public headers require mutable mappings
) -> tuple[
dict[str, tuple[str, bytes, str]], dict[str, str]
]: # mutable-ok: provider payloads and public headers require mutable mappings
"""Convert a LiteLLM file value to MiniMax multipart fields."""
if purpose not in cls.UPLOAD_PURPOSES:
allowed = ", ".join(sorted(cls.UPLOAD_PURPOSES))
@ -95,7 +101,9 @@ class MinimaxVoiceCloneConfig:
if not isinstance(content, bytes):
content = bytes(content)
content_type = extracted.get("content_type") or "application/octet-stream"
return {"file": (filename, content, content_type)}, {"purpose": purpose} # mutable-ok: multipart fields require mutable mapping for httpx
return {"file": (filename, content, content_type)}, {
"purpose": purpose
} # mutable-ok: multipart fields require mutable mapping for httpx
@classmethod
def transform_upload_response(cls, raw_response: httpx.Response) -> str:
@ -112,16 +120,24 @@ class MinimaxVoiceCloneConfig:
return file_id
@classmethod
def transform_clone_request(cls, file_id: str, voice_id: str, model: str) -> dict[str, str]: # mutable-ok: provider payloads and public headers require mutable mappings
def transform_clone_request(
cls, file_id: str, voice_id: str, model: str
) -> dict[str, str]: # mutable-ok: provider payloads and public headers require mutable mappings
"""Build the required MiniMax voice-clone JSON body."""
values = {"file_id": file_id, "voice_id": voice_id, "model": model} # mutable-ok: request body is serialized as mutable JSON
values = {
"file_id": file_id,
"voice_id": voice_id,
"model": model,
} # mutable-ok: request body is serialized as mutable JSON
for name, value in values.items():
if not isinstance(value, str) or not value.strip():
raise ValueError(f"{name} is required for MiniMax voice cloning")
if model not in cls.SUPPORTED_MODELS:
supported = ", ".join(sorted(cls.SUPPORTED_MODELS))
raise ValueError(f"Unsupported MiniMax voice-clone model {model!r}; supported models: {supported}")
return {name: value.strip() for name, value in values.items()} # mutable-ok: request body is serialized as mutable JSON
return {
name: value.strip() for name, value in values.items()
} # mutable-ok: request body is serialized as mutable JSON
@classmethod
def transform_clone_response(
@ -141,10 +157,16 @@ class MinimaxVoiceCloneConfig:
)
if not voice_id:
raise MinimaxVoiceCloneError("MiniMax clone response did not include voice_id", raw_response.status_code)
return {"file_id": file_id, "voice_id": voice_id, "model": model} # mutable-ok: parsed response is normalized from provider JSON
return {
"file_id": file_id,
"voice_id": voice_id,
"model": model,
} # mutable-ok: parsed response is normalized from provider JSON
@staticmethod
def _json_response(raw_response: httpx.Response) -> dict[str, object]: # mutable-ok: provider payloads and public headers require mutable mappings
def _json_response(
raw_response: httpx.Response,
) -> dict[str, object]: # mutable-ok: provider payloads and public headers require mutable mappings
try:
payload = raw_response.json()
except ValueError as exc:
@ -183,7 +205,9 @@ class MinimaxVoiceCloneConfig:
def _raise_for_api_status(payload: Mapping[str, object], http_status: int) -> None:
base_resp = payload.get("base_resp")
status_code = base_resp.get("status_code") if isinstance(base_resp, Mapping) else None
if http_status >= 400 or (status_code is not None and str(status_code) not in {"0", "200"}): # mutable-ok: status code allowlist is a local membership set
if http_status >= 400 or (
status_code is not None and str(status_code) not in {"0", "200"}
): # mutable-ok: status code allowlist is a local membership set
status_message = base_resp.get("status_msg") if isinstance(base_resp, Mapping) else None
detail = str(status_message or payload.get("message") or "MiniMax voice-clone request failed")
raise MinimaxVoiceCloneError(detail, http_status or int(status_code or 0))

View file

@ -2,4 +2,7 @@
from .main import avoice_clone, voice_clone
__all__ = ["avoice_clone", "voice_clone"] # mutable-ok: public export list is intentionally mutable for package compatibility
__all__ = [
"avoice_clone",
"voice_clone",
] # mutable-ok: public export list is intentionally mutable for package compatibility

View file

@ -31,7 +31,7 @@ def voice_clone(
api_base: str | None = None,
timeout: float | httpx.Timeout | None = None,
custom_llm_provider: str | None = "minimax",
extra_headers: dict[str, str] | None = None, # mutable-ok: caller may provide mutable extra headers
extra_headers: dict[str, str] | None = None, # mutable-ok: caller may provide mutable extra headers
client: httpx.Client | None = None,
) -> VoiceCloneResponse:
"""Upload an audio sample and create a MiniMax custom voice.
@ -70,7 +70,7 @@ def voice_clone(
clone_body = config.transform_clone_request(file_id=file_id, voice_id=voice_id, model=model)
clone_response = http_client.post(
config.get_complete_url(api_base=api_base, operation="clone"),
headers={**headers, "Content-Type": "application/json"}, # mutable-ok: JSON request body is passed to httpx
headers={**headers, "Content-Type": "application/json"}, # mutable-ok: JSON request body is passed to httpx
json=clone_body,
timeout=timeout_value,
)
@ -90,7 +90,7 @@ async def avoice_clone(
api_base: str | None = None,
timeout: float | httpx.Timeout | None = None,
custom_llm_provider: str | None = "minimax",
extra_headers: dict[str, str] | None = None, # mutable-ok: caller may provide mutable extra headers
extra_headers: dict[str, str] | None = None, # mutable-ok: caller may provide mutable extra headers
client: httpx.AsyncClient | None = None,
) -> VoiceCloneResponse:
"""Asynchronously upload and clone a MiniMax voice."""
@ -102,7 +102,7 @@ async def avoice_clone(
timeout_value = timeout if timeout is not None else 600.0
http_client = client or get_async_httpx_client(
llm_provider=LlmProviders.MINIMAX,
params={"timeout": timeout_value}, # mutable-ok: JSON request body is passed to httpx
params={"timeout": timeout_value}, # mutable-ok: JSON request body is passed to httpx
)
upload_response = await http_client.post(
config.get_complete_url(api_base=api_base, operation="upload"),
@ -115,7 +115,10 @@ async def avoice_clone(
clone_body = config.transform_clone_request(file_id=file_id, voice_id=voice_id, model=model)
clone_response = await http_client.post(
config.get_complete_url(api_base=api_base, operation="clone"),
headers={**headers, "Content-Type": "application/json"}, # mutable-ok: provider request and export collections are intentionally mutable
headers={
**headers,
"Content-Type": "application/json",
}, # mutable-ok: provider request and export collections are intentionally mutable
json=clone_body,
timeout=timeout_value,
)

View file

@ -37,6 +37,14 @@ def test_prompt_audio_purpose_is_supported() -> None:
assert data == {"purpose": "prompt_audio"}
def test_invalid_operation_and_upload_purpose_are_rejected() -> None:
with pytest.raises(ValueError, match="operation must be"):
MinimaxVoiceCloneConfig.get_complete_url(operation="invalid")
with pytest.raises(ValueError, match="file upload purpose"):
MinimaxVoiceCloneConfig.transform_upload_request(("sample.wav", b"audio"), "invalid")
def test_upload_response_extracts_nested_file_id() -> None:
file_id = MinimaxVoiceCloneConfig.transform_upload_response(
response({"file": {"file_id": "file-123"}, "base_resp": {"status_code": 0}})