diff --git a/litellm/llms/minimax/voice_clone/__init__.py b/litellm/llms/minimax/voice_clone/__init__.py index b8af979100c..38c8bd856b8 100644 --- a/litellm/llms/minimax/voice_clone/__init__.py +++ b/litellm/llms/minimax/voice_clone/__init__.py @@ -6,7 +6,7 @@ from .transformation import ( VoiceCloneResponse, ) -__all__ = [ +__all__ = [ # mutable-ok: public export list is intentionally mutable for package compatibility "MinimaxVoiceCloneConfig", "MinimaxVoiceCloneError", "VoiceCloneResponse", diff --git a/litellm/llms/minimax/voice_clone/transformation.py b/litellm/llms/minimax/voice_clone/transformation.py index d803afc3fc2..2db82f8b85e 100644 --- a/litellm/llms/minimax/voice_clone/transformation.py +++ b/litellm/llms/minimax/voice_clone/transformation.py @@ -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 - voice_id: str - model: str + 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,19 @@ 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] + 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]: + 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 {}) + result = dict(headers or {}) # mutable-ok: caller headers must remain independently mutable result["Authorization"] = f"Bearer {resolved_key}" return result @@ -83,7 +83,7 @@ class MinimaxVoiceCloneConfig: cls, file: FileTypes, purpose: str = "voice_clone", - ) -> tuple[dict[str, tuple[str, bytes, str]], dict[str, str]]: + ) -> 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 +95,7 @@ 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} + 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 +112,16 @@ class MinimaxVoiceCloneConfig: return file_id @classmethod - def transform_clone_request(cls, file_id: str, voice_id: str, model: str) -> dict[str, str]: + 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} + 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()} + 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 +141,10 @@ 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} + 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]: + 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 +183,7 @@ 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"}): + 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)) diff --git a/litellm/voice_clone/__init__.py b/litellm/voice_clone/__init__.py index 889acc78f38..616ddbac412 100644 --- a/litellm/voice_clone/__init__.py +++ b/litellm/voice_clone/__init__.py @@ -2,4 +2,4 @@ from .main import avoice_clone, voice_clone -__all__ = ["avoice_clone", "voice_clone"] +__all__ = ["avoice_clone", "voice_clone"] # mutable-ok: public export list is intentionally mutable for package compatibility diff --git a/litellm/voice_clone/main.py b/litellm/voice_clone/main.py index a019f6dc28b..457f12286fb 100644 --- a/litellm/voice_clone/main.py +++ b/litellm/voice_clone/main.py @@ -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, + 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"}, + 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, + 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}, + 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,7 @@ 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"}, + headers={**headers, "Content-Type": "application/json"}, # mutable-ok: provider request and export collections are intentionally mutable json=clone_body, timeout=timeout_value, )