diff --git a/apps/models_provider/impl/minimax_model_provider/model/tts.py b/apps/models_provider/impl/minimax_model_provider/model/tts.py index 50935e4c915..b8503448c13 100644 --- a/apps/models_provider/impl/minimax_model_provider/model/tts.py +++ b/apps/models_provider/impl/minimax_model_provider/model/tts.py @@ -43,6 +43,81 @@ def new_instance(model_type, model_name, model_credential: Dict[str, object], ** def check_auth(self): self.text_to_speech(_('Hello')) + def _post_voice_operation(self, path, **request_kwargs): + api_base = self.api_base.rstrip('/') + headers = {'Authorization': f'Bearer {self.api_key}'} + if 'json' in request_kwargs: + headers['Content-Type'] = 'application/json' + + response = requests.post( + f'{api_base}/{path.lstrip("/")}', + headers=headers, + timeout=60, + **request_kwargs, + ) + response.raise_for_status() + + result = response.json() + if result.get('base_resp', {}).get('status_code', 0) != 0: + error_msg = result.get('base_resp', {}).get('status_msg', 'Unknown error') + raise Exception(f'MiniMax voice API error: {error_msg}') + return result + + def _upload_voice_audio(self, audio_file, purpose): + if purpose not in {'voice_clone', 'prompt_audio'}: + raise ValueError(f'Unsupported MiniMax voice upload purpose: {purpose}') + + if hasattr(audio_file, 'read'): + result = self._post_voice_operation( + 'files/upload', + data={'purpose': purpose}, + files={'file': audio_file}, + ) + else: + with open(audio_file, 'rb') as file_handle: + result = self._post_voice_operation( + 'files/upload', + data={'purpose': purpose}, + files={'file': file_handle}, + ) + + file_id = result.get('file', {}).get('file_id') or result.get('file_id') + if not file_id: + raise Exception('MiniMax voice upload returned no file ID') + return file_id + + def upload_clone_audio(self, audio_file): + return self._upload_voice_audio(audio_file, 'voice_clone') + + def upload_prompt_audio(self, audio_file): + return self._upload_voice_audio(audio_file, 'prompt_audio') + + def voice_clone(self, file_id, voice_id, model=None, **params): + result = self._post_voice_operation( + 'voice_clone', + json={ + 'file_id': file_id, + 'voice_id': voice_id, + 'model': model or self.model, + **params, + }, + ) + return result.get('voice_id') or voice_id + + def voice_design(self, prompt, voice_id, **params): + result = self._post_voice_operation( + 'voice_design', + json={ + 'prompt': prompt, + 'voice_id': voice_id, + **params, + }, + ) + designed_voice_id = result.get('voice_id') + if not designed_voice_id: + raise Exception('MiniMax voice design returned no voice ID') + return designed_voice_id + def text_to_speech(self, text): text = _remove_empty_lines(text) api_base = self.api_base.rstrip('/')