shudao-main/shudao-chat-py/services/tencent_speech_service.py

import base64
import json
import time
from typing import Any

from utils.config import settings
from utils.logger import logger


CONTENT_TYPE_BY_CODEC = {
    "mp3": "audio/mpeg",
    "wav": "audio/wav",
    "pcm": "audio/pcm",
}


class TencentSpeechService:
    def __init__(self):
        self.config = settings.speech

    def _ensure_available(self):
        if not self.config.enabled:
            raise ValueError("语音服务未启用")
        if self.config.provider != "tencent_cloud":
            raise ValueError(f"暂不支持的语音服务商: {self.config.provider}")

        credentials = self.config.tencent
        if not credentials.app_id or not credentials.secret_id or not credentials.secret_key:
            raise ValueError("腾讯云语音服务凭证未配置")

    def _load_tencent_sdk(self):
        try:
            from tencentcloud.common import credential
            from tencentcloud.common.exception.tencent_cloud_sdk_exception import (
                TencentCloudSDKException,
            )
            from tencentcloud.asr.v20190614 import asr_client, models as asr_models
            from tencentcloud.tts.v20190823 import tts_client, models as tts_models
        except ImportError as exc:
            raise RuntimeError(
                "未安装腾讯云 SDK，请先安装 tencentcloud-sdk-python"
            ) from exc

        return {
            "credential": credential,
            "TencentCloudSDKException": TencentCloudSDKException,
            "asr_client": asr_client,
            "asr_models": asr_models,
            "tts_client": tts_client,
            "tts_models": tts_models,
        }

    def _build_credentials(self, credential_module):
        cfg = self.config.tencent
        return credential_module.Credential(cfg.secret_id, cfg.secret_key)

    @staticmethod
    def _read_upload_bytes(upload_file) -> bytes:
        source = getattr(upload_file, "file", None)
        if source is None:
            raise ValueError("未读取到音频文件")

        if hasattr(source, "seek"):
            source.seek(0)

        data = source.read()
        if hasattr(source, "seek"):
            source.seek(0)

        if not data:
            raise ValueError("音频文件为空")

        return data

    def transcribe_file(self, upload_file, user_id: str = "", request_id: str | None = None) -> dict[str, Any]:
        self._ensure_available()

        if not getattr(self.config.transcribe, "enabled", True):
            raise ValueError("语音转文字未启用")

        audio_bytes = self._read_upload_bytes(upload_file)
        max_size_bytes = self.config.transcribe.max_audio_size_mb * 1024 * 1024
        if len(audio_bytes) > max_size_bytes:
            raise ValueError(
                f"音频文件超过 {self.config.transcribe.max_audio_size_mb}MB，当前版本请缩短录音后重试"
            )

        sdk = self._load_tencent_sdk()
        credentials = self._build_credentials(sdk["credential"])
        client = sdk["asr_client"].AsrClient(credentials, self.config.tencent.region)
        models = sdk["asr_models"]

        usr_audio_key = request_id or f"{user_id or 'user'}-{int(time.time() * 1000)}"
        params = {
            "ProjectId": 0,
            "SubServiceType": 2,
            "EngSerViceType": self.config.transcribe.engine_model_type,
            "SourceType": self.config.transcribe.source_type,
            "VoiceFormat": self.config.transcribe.voice_format,
            "UsrAudioKey": usr_audio_key,
            "Data": base64.b64encode(audio_bytes).decode("utf-8"),
            "DataLen": len(audio_bytes),
            "WordInfo": self.config.transcribe.word_info,
            "FilterDirty": self.config.transcribe.filter_dirty,
            "FilterModal": self.config.transcribe.filter_modal,
            "FilterPunc": self.config.transcribe.filter_punc,
            "ConvertNumMode": self.config.transcribe.convert_num_mode,
        }

        if self.config.transcribe.hotword_id:
            params["HotwordId"] = self.config.transcribe.hotword_id
        elif self.config.transcribe.hotword_list:
            params["HotwordList"] = self.config.transcribe.hotword_list

        if self.config.transcribe.customization_id:
            params["CustomizationId"] = self.config.transcribe.customization_id

        if self.config.transcribe.replace_text_id:
            params["ReplaceTextId"] = self.config.transcribe.replace_text_id

        req = models.SentenceRecognitionRequest()
        req.from_json_string(json.dumps(params))

        try:
            resp = client.SentenceRecognition(req)
            text = getattr(resp, "Result", "") or ""
            request_id = getattr(resp, "RequestId", "")
            if not text.strip():
                raise RuntimeError("腾讯云语音识别未返回文本内容")

            return {
                "text": text.strip(),
                "request_id": request_id,
                "raw": resp.to_json_string(),
            }
        except sdk["TencentCloudSDKException"] as exc:
            logger.error("[speech] Tencent ASR failed: %s", exc)
            raise RuntimeError(f"腾讯云语音识别失败: {exc}") from exc

    def synthesize_text(
        self,
        text: str,
        *,
        session_id: str | None = None,
        voice_type: int | None = None,
        speed: float | None = None,
        volume: float | None = None,
    ) -> dict[str, Any]:
        self._ensure_available()

        if not getattr(self.config.synthesize, "enabled", True):
            raise ValueError("语音播报未启用")

        normalized_text = (text or "").strip()
        if not normalized_text:
            raise ValueError("播报文本不能为空")

        if len(normalized_text) > self.config.synthesize.basic_text_limit_chars:
            raise ValueError(
                f"播报文本超过 {self.config.synthesize.basic_text_limit_chars} 字，请先分段后重试"
            )

        sdk = self._load_tencent_sdk()
        credentials = self._build_credentials(sdk["credential"])
        client = sdk["tts_client"].TtsClient(credentials, self.config.tencent.region)
        models = sdk["tts_models"]

        codec = self.config.synthesize.codec.lower()
        params = {
            "Text": normalized_text,
            "SessionId": session_id or f"speech-{int(time.time() * 1000)}",
            "ProjectId": 0,
            "ModelType": 1,
            "Volume": volume if volume is not None else self.config.synthesize.volume,
            "Speed": speed if speed is not None else self.config.synthesize.speed,
            "VoiceType": voice_type if voice_type is not None else self.config.synthesize.voice_type,
            "SampleRate": self.config.synthesize.sample_rate,
            "Codec": codec,
            "PrimaryLanguage": self.config.synthesize.primary_language,
            "EnableSubtitle": self.config.synthesize.enable_subtitle,
        }

        req = models.TextToVoiceRequest()
        req.from_json_string(json.dumps(params))

        try:
            resp = client.TextToVoice(req)
            audio_base64 = getattr(resp, "Audio", "") or ""
            request_id = getattr(resp, "RequestId", "")
            if not audio_base64:
                raise RuntimeError("腾讯云语音合成未返回音频内容")

            return {
                "audio_bytes": base64.b64decode(audio_base64),
                "content_type": CONTENT_TYPE_BY_CODEC.get(codec, "application/octet-stream"),
                "request_id": request_id,
                "raw": resp.to_json_string(),
            }
        except sdk["TencentCloudSDKException"] as exc:
            logger.error("[speech] Tencent TTS failed: %s", exc)
            raise RuntimeError(f"腾讯云语音合成失败: {exc}") from exc


tencent_speech_service = TencentSpeechService()