Underground-Digital/Workflow-Engine

1import io2import logging3from typing import Optional4 5from werkzeug.datastructures import FileStorage6 7from core.model_manager import ModelManager8from core.model_runtime.entities.model_entities import ModelType9from models.model import App, AppMode, AppModelConfig, Message10from services.errors.audio import (11    AudioTooLargeServiceError,12    NoAudioUploadedServiceError,13    ProviderNotSupportSpeechToTextServiceError,14    ProviderNotSupportTextToSpeechServiceError,15    UnsupportedAudioTypeServiceError,16)17 18FILE_SIZE = 3019FILE_SIZE_LIMIT = FILE_SIZE * 1024 * 102420ALLOWED_EXTENSIONS = ["mp3", "mp4", "mpeg", "mpga", "m4a", "wav", "webm", "amr"]21 22logger = logging.getLogger(__name__)23 24 25class AudioService:26    @classmethod27    def transcript_asr(cls, app_model: App, file: FileStorage, end_user: Optional[str] = None):28        if app_model.mode in {AppMode.ADVANCED_CHAT.value, AppMode.WORKFLOW.value}:29            workflow = app_model.workflow30            if workflow is None:31                raise ValueError("Speech to text is not enabled")32 33            features_dict = workflow.features_dict34            if "speech_to_text" not in features_dict or not features_dict["speech_to_text"].get("enabled"):35                raise ValueError("Speech to text is not enabled")36        else:37            app_model_config: AppModelConfig = app_model.app_model_config38 39            if not app_model_config.speech_to_text_dict["enabled"]:40                raise ValueError("Speech to text is not enabled")41 42        if file is None:43            raise NoAudioUploadedServiceError()44 45        extension = file.mimetype46        if extension not in [f"audio/{ext}" for ext in ALLOWED_EXTENSIONS]:47            raise UnsupportedAudioTypeServiceError()48 49        file_content = file.read()50        file_size = len(file_content)51 52        if file_size > FILE_SIZE_LIMIT:53            message = f"Audio size larger than {FILE_SIZE} mb"54            raise AudioTooLargeServiceError(message)55 56        model_manager = ModelManager()57        model_instance = model_manager.get_default_model_instance(58            tenant_id=app_model.tenant_id, model_type=ModelType.SPEECH2TEXT59        )60        if model_instance is None:61            raise ProviderNotSupportSpeechToTextServiceError()62 63        buffer = io.BytesIO(file_content)64        buffer.name = "temp.mp3"65 66        return {"text": model_instance.invoke_speech2text(file=buffer, user=end_user)}67 68    @classmethod69    def transcript_tts(70        cls,71        app_model: App,72        text: Optional[str] = None,73        voice: Optional[str] = None,74        end_user: Optional[str] = None,75        message_id: Optional[str] = None,76    ):77        from collections.abc import Generator78 79        from flask import Response, stream_with_context80 81        from app import app82        from extensions.ext_database import db83 84        def invoke_tts(text_content: str, app_model, voice: Optional[str] = None):85            with app.app_context():86                if app_model.mode in {AppMode.ADVANCED_CHAT.value, AppMode.WORKFLOW.value}:87                    workflow = app_model.workflow88                    if workflow is None:89                        raise ValueError("TTS is not enabled")90 91                    features_dict = workflow.features_dict92                    if "text_to_speech" not in features_dict or not features_dict["text_to_speech"].get("enabled"):93                        raise ValueError("TTS is not enabled")94 95                    voice = features_dict["text_to_speech"].get("voice") if voice is None else voice96                else:97                    text_to_speech_dict = app_model.app_model_config.text_to_speech_dict98 99                    if not text_to_speech_dict.get("enabled"):100                        raise ValueError("TTS is not enabled")101 102                    voice = text_to_speech_dict.get("voice") if voice is None else voice103 104                model_manager = ModelManager()105                model_instance = model_manager.get_default_model_instance(106                    tenant_id=app_model.tenant_id, model_type=ModelType.TTS107                )108                try:109                    if not voice:110                        voices = model_instance.get_tts_voices()111                        if voices:112                            voice = voices[0].get("value")113                        else:114                            raise ValueError("Sorry, no voice available.")115 116                    return model_instance.invoke_tts(117                        content_text=text_content.strip(), user=end_user, tenant_id=app_model.tenant_id, voice=voice118                    )119                except Exception as e:120                    raise e121 122        if message_id:123            message = db.session.query(Message).filter(Message.id == message_id).first()124            if message.answer == "" and message.status == "normal":125                return None126 127            else:128                response = invoke_tts(message.answer, app_model=app_model, voice=voice)129                if isinstance(response, Generator):130                    return Response(stream_with_context(response), content_type="audio/mpeg")131                return response132        else:133            response = invoke_tts(text, app_model, voice)134            if isinstance(response, Generator):135                return Response(stream_with_context(response), content_type="audio/mpeg")136            return response137 138    @classmethod139    def transcript_tts_voices(cls, tenant_id: str, language: str):140        model_manager = ModelManager()141        model_instance = model_manager.get_default_model_instance(tenant_id=tenant_id, model_type=ModelType.TTS)142        if model_instance is None:143            raise ProviderNotSupportTextToSpeechServiceError()144 145        try:146            return model_instance.get_tts_voices(language)147        except Exception as e:148            raise e149