File size: 4,289 Bytes
56dc677
95cb26e
56dc677
 
674469e
56dc677
c7fc3b6
56dc677
 
 
 
 
 
 
 
 
4a13628
56dc677
 
 
 
 
 
 
95cb26e
56dc677
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
95cb26e
56dc677
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
674469e
 
56dc677
 
 
 
 
 
 
 
95cb26e
56dc677
 
95cb26e
56dc677
 
 
 
 
 
674469e
56dc677
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
674469e
56dc677
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
import base64
import io
import tempfile
import os
from gtts import gTTS
import pyttsx3

class TTSService:
    def __init__(self):
        self.models = {}
        self._initialize_models()
    
    def _initialize_models(self):
        """Initialize TTS models"""
        # gTTS is our primary method (always available)
        self.models["gtts"] = True
        
        # Try to initialize pyttsx3 as fallback
        try:
            self.models["pyttsx3"] = pyttsx3.init()
            print("✓ pyttsx3 TTS initialized")
        except:
            print("⚠️ pyttsx3 not available")
            self.models["pyttsx3"] = None
        
        # Coqui TTS is optional
        self.models["coqui"] = self._initialize_coqui_tts()
    
    def _initialize_coqui_tts(self):
        """Initialize Coqui TTS if available"""
        try:
            from TTS.api import TTS
            tts_model = TTS(model_name="tts_models/en/ljspeech/tacotron2-DDC", progress_bar=False)
            print("✓ Coqui TTS initialized")
            return tts_model
        except ImportError:
            print("⚠️ Coqui TTS not available. Install with: pip install TTS")
            return None
        except Exception as e:
            print(f"⚠️ Coqui TTS initialization failed: {e}")
            return None
    
    async def text_to_speech_base64(self, text: str, language: str = "en") -> str:
        """Convert text to base64 audio"""
        # Try gTTS first (most reliable and free)
        try:
            return await self._gtts_to_base64(text, language)
        except Exception as e:
            print(f"gTTS error: {e}")
        
        # Fallback to pyttsx3
        try:
            if self.models.get("pyttsx3"):
                return await self._pyttsx3_to_base64(text)
        except Exception as e:
            print(f"pyttsx3 error: {e}")
        
        # Final fallback to Coqui TTS
        try:
            if self.models.get("coqui"):
                return await self._coqui_to_base64(text)
        except Exception as e:
            print(f"Coqui TTS error: {e}")
        
        raise Exception("All TTS services failed")
    
    async def _gtts_to_base64(self, text: str, language: str) -> str:
        """Convert using gTTS"""
        tts = gTTS(text=text, lang=language, slow=False)
        audio_buffer = io.BytesIO()
        tts.write_to_fp(audio_buffer)
        audio_buffer.seek(0)
        return base64.b64encode(audio_buffer.getvalue()).decode('utf-8')
    
    async def _pyttsx3_to_base64(self, text: str) -> str:
        """Convert using pyttsx3"""
        engine = self.models["pyttsx3"]
        with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as temp_file:
            temp_path = temp_file.name
        
        engine.save_to_file(text, temp_path)
        engine.runAndWait()
        
        with open(temp_path, 'rb') as audio_file:
            audio_base64 = base64.b64encode(audio_file.read()).decode('utf-8')
        
        # Cleanup
        os.unlink(temp_path)
        return audio_base64
    
    async def _coqui_to_base64(self, text: str) -> str:
        """Convert using Coqui TTS"""
        with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as temp_file:
            temp_path = temp_file.name
        
        self.models["coqui"].tts_to_file(text=text, file_path=temp_path)
        
        with open(temp_path, 'rb') as audio_file:
            audio_base64 = base64.b64encode(audio_file.read()).decode('utf-8')
        
        # Cleanup
        os.unlink(temp_path)
        return audio_base64

# Simple TTS service that only uses gTTS (minimal dependencies)
class SimpleTTSService:
    def __init__(self):
        pass
    
    async def text_to_speech_base64(self, text: str, language: str = "en") -> str:
        """Convert text to base64 audio using only gTTS"""
        try:
            tts = gTTS(text=text, lang=language, slow=False)
            audio_buffer = io.BytesIO()
            tts.write_to_fp(audio_buffer)
            audio_buffer.seek(0)
            return base64.b64encode(audio_buffer.getvalue()).decode('utf-8')
        except Exception as e:
            print(f"gTTS error: {e}")
            # Return a placeholder audio or error message
            return "TTS_ERROR_PLACEHOLDER"