Spaces:
Sleeping
Sleeping
| # application/utils/text_to_speech.py - NEW FILE | |
| import asyncio | |
| import edge_tts | |
| from aiohttp import web # Import aiohttp | |
| async def generate_tts(text: str) -> bytes: | |
| """ | |
| Generates speech from text using edge-tts and returns the audio as a byte stream. | |
| """ | |
| try: | |
| communicate = edge_tts.Communicate(text, "en-US-GuyNeural") # Or any other preferred voice | |
| return await communicate.stream() | |
| except Exception as e: | |
| print(f"Error in TTS generation: {e}") | |
| return b"" # Return empty bytes on error | |
| async def tts_handler(request: web.Request) -> web.Response: | |
| """ | |
| aiohttp handler for the /tts endpoint. This is the correct way to handle | |
| streaming with aiohttp (and thus edge-tts). | |
| """ | |
| text = request.query.get('text') | |
| if not text: | |
| return web.Response(status=400, text="No text provided") | |
| try: | |
| response = web.StreamResponse() | |
| response.content_type = 'audio/mpeg' | |
| await response.prepare(request) | |
| async for chunk in generate_tts(text): | |
| await response.write(chunk) | |
| await response.write_eof() | |
| return response | |
| except Exception as e: | |
| print(f"Error in TTS handler: {e}") | |
| return web.Response(status=500, text=str(e)) |