wordloop/backend/services/tts_service.py

from __future__ import annotations

import base64
import hashlib
import json
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path

from config import get_settings

_CACHE_DIR = Path(__file__).resolve().parent.parent / "data" / "tts_cache"
_NO_PROXY_OPENER = urllib.request.build_opener(urllib.request.ProxyHandler({}))
_TTS_USER_AGENT = "WordLoop-TTS/1.0"


def _open_tts_request(
    request: urllib.request.Request,
    *,
    timeout: int,
):
    # 统一绕过系统代理；公网网关会拒绝 Python-urllib 默认 UA（403）
    return _NO_PROXY_OPENER.open(request, timeout=timeout)


class TtsServiceError(Exception):
    pass


class TtsService:
    def is_configured(self) -> bool:
        settings = get_settings()
        return settings.tkmind_tts_enabled and bool(settings.tkmind_tts_base_url.strip())

    def synthesize(self, text: str) -> tuple[bytes, str]:
        cleaned = text.strip()
        if not cleaned:
            raise TtsServiceError("朗读文本不能为空")
        if len(cleaned) > 200:
            raise TtsServiceError("朗读文本过长")

        settings = get_settings()
        if not settings.tkmind_tts_enabled:
            raise TtsServiceError("单词朗读未启用")
        base_url = settings.tkmind_tts_base_url.strip().rstrip("/")
        if not base_url:
            raise TtsServiceError("TTS 服务地址未配置")

        cache_key = self._cache_key(cleaned, settings.tkmind_tts_spk_id)
        cached = self._read_cache(cache_key)
        if cached is not None:
            return cached, "audio/wav"

        audio_bytes, content_type = self._request_upstream(
            base_url=base_url,
            text=cleaned,
            spk_id=settings.tkmind_tts_spk_id.strip(),
            timeout=settings.tkmind_tts_timeout,
        )
        self._write_cache(cache_key, audio_bytes)
        return audio_bytes, content_type

    @staticmethod
    def _cache_key(text: str, spk_id: str) -> str:
        raw = f"{text.lower()}|{spk_id}".encode("utf-8")
        return hashlib.sha256(raw).hexdigest()

    @staticmethod
    def _read_cache(cache_key: str) -> bytes | None:
        path = _CACHE_DIR / f"{cache_key}.wav"
        if not path.is_file():
            return None
        return path.read_bytes()

    @staticmethod
    def _write_cache(cache_key: str, audio_bytes: bytes) -> None:
        _CACHE_DIR.mkdir(parents=True, exist_ok=True)
        path = _CACHE_DIR / f"{cache_key}.wav"
        path.write_bytes(audio_bytes)

    def _request_upstream(
        self,
        *,
        base_url: str,
        text: str,
        spk_id: str,
        timeout: int,
    ) -> tuple[bytes, str]:
        payload: dict[str, str] = {"tts_text": text}
        if spk_id:
            payload["spk_id"] = spk_id

        request = urllib.request.Request(
            f"{base_url}/inference_sft",
            data=urllib.parse.urlencode(payload).encode("utf-8"),
            headers={
                "Content-Type": "application/x-www-form-urlencoded",
                "User-Agent": _TTS_USER_AGENT,
            },
            method="POST",
        )
        try:
            with _open_tts_request(request, timeout=timeout) as response:
                content_type = response.headers.get_content_type() or "application/octet-stream"
                body = response.read()
        except urllib.error.HTTPError as exc:
            detail = exc.read().decode("utf-8", errors="replace")
            raise TtsServiceError(f"TTS 服务异常（HTTP {exc.code}）") from exc
        except urllib.error.URLError as exc:
            raise TtsServiceError(f"TTS 服务不可达：{exc.reason}") from exc

        if content_type.startswith("audio/"):
            if not body:
                raise TtsServiceError("TTS 返回空音频")
            return body, content_type

        return self._parse_json_response(body)

    @staticmethod
    def _parse_json_response(body: bytes) -> tuple[bytes, str]:
        try:
            payload = json.loads(body.decode("utf-8"))
        except (UnicodeDecodeError, json.JSONDecodeError) as exc:
            if body[:4] == b"RIFF":
                return body, "audio/wav"
            raise TtsServiceError("TTS 响应格式无效") from exc

        code = payload.get("code", 0)
        if code not in (0, "0", None):
            message = str(payload.get("message") or "TTS 合成失败")
            raise TtsServiceError(message)

        data = payload.get("data")
        audio_b64 = TtsService._extract_audio_base64(data)
        if not audio_b64:
            raise TtsServiceError("TTS 响应缺少音频数据")

        try:
            audio_bytes = base64.b64decode(audio_b64)
        except (ValueError, TypeError) as exc:
            raise TtsServiceError("TTS 音频解码失败") from exc
        if not audio_bytes:
            raise TtsServiceError("TTS 返回空音频")
        return audio_bytes, "audio/wav"

    @staticmethod
    def _extract_audio_base64(data: object) -> str | None:
        if isinstance(data, str) and data.strip():
            return data.strip()
        if isinstance(data, dict):
            for key in ("audio", "wav", "speech", "data"):
                value = data.get(key)
                if isinstance(value, str) and value.strip():
                    return value.strip()
        return None


tts_service = TtsService()