99 lines
3.2 KiB
Python
99 lines
3.2 KiB
Python
"""
|
|
TTS Router — озвучка текста через Edge TTS.
|
|
Бесплатный, качественный синтез речи для польского языка.
|
|
Кэш хранится в файлах на диске.
|
|
"""
|
|
import logging
|
|
import io
|
|
import os
|
|
import hashlib
|
|
import edge_tts
|
|
from fastapi import APIRouter, Request, HTTPException
|
|
from fastapi.responses import Response
|
|
from auth import require_auth
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
router = APIRouter(prefix="/tts", tags=["tts"])
|
|
|
|
# Голоса Edge TTS для польского
|
|
VOICE_PL = "pl-PL-MarekNeural" # Мужской голос (основной)
|
|
VOICE_PL_FEMALE = "pl-PL-AgnieszkaNeural" # Женский голос
|
|
|
|
# Директория для кэша
|
|
CACHE_DIR = "/root/filwordspl/backend/tts_cache"
|
|
os.makedirs(CACHE_DIR, exist_ok=True)
|
|
|
|
|
|
def get_cache_path(voice_name: str, text: str) -> str:
|
|
"""Создаёт путь к файлу кэша на основе хеша текста и голоса"""
|
|
cache_key = hashlib.md5(f"{voice_name}:{text}".encode()).hexdigest()
|
|
return os.path.join(CACHE_DIR, f"{cache_key}.mp3")
|
|
|
|
|
|
@router.get("")
|
|
async def text_to_speech(request: Request, text: str = "", voice: str = "male"):
|
|
"""
|
|
Озвучивает текст через Edge TTS.
|
|
|
|
Параметры:
|
|
- text: текст для озвучки
|
|
- voice: male (Marek) или female (Agnieszka)
|
|
|
|
Возвращает: audio/mpeg
|
|
"""
|
|
await require_auth(request)
|
|
|
|
if not text or len(text) > 500:
|
|
raise HTTPException(400, "Text is required (max 500 characters)")
|
|
|
|
# Выбираем голос
|
|
if voice == "female":
|
|
voice_name = VOICE_PL_FEMALE
|
|
else:
|
|
voice_name = VOICE_PL
|
|
|
|
# Путь к файлу кэша
|
|
cache_path = get_cache_path(voice_name, text)
|
|
|
|
# Проверяем файловый кэш
|
|
if os.path.exists(cache_path):
|
|
logger.info(f"TTS: cache hit for text: {text[:50]}...")
|
|
with open(cache_path, 'rb') as f:
|
|
return Response(
|
|
content=f.read(),
|
|
media_type="audio/mpeg",
|
|
)
|
|
|
|
try:
|
|
logger.info(f"TTS: synthesizing text: {text[:50]}...")
|
|
|
|
# Синтезируем речь
|
|
communicate = edge_tts.Communicate(text, voice_name)
|
|
audio_data = io.BytesIO()
|
|
|
|
async for chunk in communicate.stream():
|
|
if chunk["type"] == "audio":
|
|
audio_data.write(chunk["data"])
|
|
|
|
audio_bytes = audio_data.getvalue()
|
|
|
|
if not audio_bytes:
|
|
raise HTTPException(500, "TTS generation failed")
|
|
|
|
# Сохраняем в файловый кэш
|
|
with open(cache_path, 'wb') as f:
|
|
f.write(audio_bytes)
|
|
|
|
logger.info(f"TTS: done, size={len(audio_bytes)} bytes, cached to {os.path.basename(cache_path)}")
|
|
|
|
return Response(
|
|
content=audio_bytes,
|
|
media_type="audio/mpeg",
|
|
)
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"TTS error: {str(e)}")
|
|
raise HTTPException(500, f"TTS error: {str(e)}") |