usky-human/digitalHuman/engine/tts/tencentTTS.py

# -*- coding: utf-8 -*-


from ..builder import TTSEngines
from ..engineBase import BaseTTSEngine
import hashlib
import hmac
import time
import json
from uuid import uuid4
from datetime import datetime, timezone
from typing import Tuple, Dict
from digitalHuman.protocol import *
from digitalHuman.utils import logger, httpxAsyncClient
from pydantic import BaseModel
from typing import List, Optional
from decimal import Decimal


__all__ = ["TencentApiTts"]


MAX_INPUT_LENGTH = 150

# neutral(中性)、sad(悲伤)、happy(高兴)、angry(生气)、fear(恐惧)、sajiao(撒娇)、amaze(震惊)、disgusted(厌恶)、peaceful(平静)
# 中性、悲伤、高兴、生气、恐惧、撒娇、震惊、厌恶、平静
class TencentVoiceEmotion(StrEnum):
    NEUTRAL = "neutral"
    SAD = "sad"
    HAPPY = "happy"
    ANGRY = "angry"
    FEAR = "fear"
    SAJIAO = "sajiao"
    AMAZE = "amaze"
    DISGUSTED = "disgusted"
    PEACEFUL = "peaceful"

class TencentVoiceDesc(BaseModel):
    id: int
    name: str
    gender: GENDER_TYPE
    language: str
    multi_emotional: bool

VOICE_LIST = [
    TencentVoiceDesc(id=501000, name="智斌", gender=GENDER_TYPE.MALE, language="中文", multi_emotional=False),
    TencentVoiceDesc(id=501001, name="智兰", gender=GENDER_TYPE.FEMALE, language="中文", multi_emotional=False),
    TencentVoiceDesc(id=501002, name="智菊", gender=GENDER_TYPE.FEMALE, language="中文", multi_emotional=False),
    TencentVoiceDesc(id=501003, name="智宇", gender=GENDER_TYPE.MALE, language="中文", multi_emotional=False),
    TencentVoiceDesc(id=501004, name="月华", gender=GENDER_TYPE.FEMALE, language="中文", multi_emotional=False),
    TencentVoiceDesc(id=501005, name="飞镜", gender=GENDER_TYPE.MALE, language="中文", multi_emotional=False),
    TencentVoiceDesc(id=501006, name="千嶂", gender=GENDER_TYPE.MALE, language="中文", multi_emotional=False),
    TencentVoiceDesc(id=501007, name="浅草", gender=GENDER_TYPE.MALE, language="中文", multi_emotional=False),
    TencentVoiceDesc(id=501008, name="WeJames", gender=GENDER_TYPE.MALE, language="英文", multi_emotional=False),
    TencentVoiceDesc(id=501009, name="WeWinny", gender=GENDER_TYPE.FEMALE, language="中文", multi_emotional=False),
    TencentVoiceDesc(id=601000, name="爱小溪", gender=GENDER_TYPE.FEMALE, language="中文", multi_emotional=True),
    TencentVoiceDesc(id=601001, name="爱小洛", gender=GENDER_TYPE.FEMALE, language="中文", multi_emotional=True),
    TencentVoiceDesc(id=601002, name="爱小辰", gender=GENDER_TYPE.MALE, language="中文", multi_emotional=True),
    TencentVoiceDesc(id=601003, name="爱小荷", gender=GENDER_TYPE.FEMALE, language="中文", multi_emotional=True),
    TencentVoiceDesc(id=601004, name="爱小树", gender=GENDER_TYPE.MALE, language="中文", multi_emotional=True),
    TencentVoiceDesc(id=601005, name="爱小静", gender=GENDER_TYPE.FEMALE, language="中文", multi_emotional=True),
    TencentVoiceDesc(id=601006, name="爱小耀", gender=GENDER_TYPE.MALE, language="中文", multi_emotional=True),
    TencentVoiceDesc(id=601007, name="爱小叶", gender=GENDER_TYPE.FEMALE, language="中文", multi_emotional=True),
    TencentVoiceDesc(id=601008, name="爱小豪", gender=GENDER_TYPE.MALE, language="中文", multi_emotional=True),
    TencentVoiceDesc(id=601009, name="爱小芊", gender=GENDER_TYPE.FEMALE, language="中文", multi_emotional=True),
    TencentVoiceDesc(id=601010, name="爱小娇", gender=GENDER_TYPE.FEMALE, language="中文", multi_emotional=True),
    TencentVoiceDesc(id=601011, name="爱小川", gender=GENDER_TYPE.MALE, language="中文", multi_emotional=True),
    TencentVoiceDesc(id=601012, name="爱小璟", gender=GENDER_TYPE.FEMALE, language="中文", multi_emotional=True),
    TencentVoiceDesc(id=601013, name="爱小伊", gender=GENDER_TYPE.FEMALE, language="中文", multi_emotional=True),
    TencentVoiceDesc(id=601014, name="爱小简", gender=GENDER_TYPE.MALE, language="中文", multi_emotional=True),
]

class TencentCloudApiKey(BaseModel):
    secret_id: str
    secret_key: str

def findVoice(name: str) -> Optional[TencentVoiceDesc]:
    for voice in VOICE_LIST:
        if voice.name == name:
            return voice
    return None

def sign(key, msg: str):
    return hmac.new(key, msg.encode("utf-8"), hashlib.sha256).digest()

@TTSEngines.register("Tencent-API")
class TencentApiTts(BaseTTSEngine): 
    def setup(self):
        self._url = "https://tts.tencentcloudapi.com"
    
    def _buildRequest(
        self, 
        input: TextMessage, 
        tencentApiKey: TencentCloudApiKey, 
        voice: str, 
        volume: float, 
        speed: float, 
        emotionCategory: str = TencentVoiceEmotion.NEUTRAL
    ) -> Tuple[Dict, str]:
        service = "tts"
        host = "tts.tencentcloudapi.com"
        version = "2019-08-23"
        action = "TextToVoice"
        algorithm = "TC3-HMAC-SHA256"
        timestamp = int(time.time())
        date = datetime.fromtimestamp(timestamp, timezone.utc).strftime("%Y-%m-%d")
        tencentVoice = findVoice(voice)
        if not tencentVoice:
            raise ValueError("voice not found")
        params = {
            "Text": input.data,
            "SessionId": str(uuid4()),
            "VoiceType": tencentVoice.id,
            # "Codec": "wav",
            "Codec": "mp3",
            "Volume": volume,
            "Speed": speed,
            "EmotionCategory": emotionCategory
        }
        payload = json.dumps(params)
        # ************* 步骤 1：拼接规范请求串 *************
        http_request_method = "POST"
        canonical_uri = "/"
        canonical_querystring = ""
        ct = "application/json; charset=utf-8"
        canonical_headers = "content-type:%s\nhost:%s\nx-tc-action:%s\n" % (ct, host, action.lower())
        signed_headers = "content-type;host;x-tc-action"
        hashed_request_payload = hashlib.sha256(payload.encode("utf-8")).hexdigest()
        canonical_request = (http_request_method + "\n" +
                            canonical_uri + "\n" +
                            canonical_querystring + "\n" +
                            canonical_headers + "\n" +
                            signed_headers + "\n" +
                            hashed_request_payload)

        # ************* 步骤 2：拼接待签名字符串 *************
        credential_scope = date + "/" + service + "/" + "tc3_request"
        hashed_canonical_request = hashlib.sha256(canonical_request.encode("utf-8")).hexdigest()
        string_to_sign = (algorithm + "\n" +
                        str(timestamp) + "\n" +
                        credential_scope + "\n" +
                        hashed_canonical_request)

        # ************* 步骤 3：计算签名 *************
        secret_date = sign(("TC3" + tencentApiKey.secret_key).encode("utf-8"), date)
        secret_service = sign(secret_date, service)
        secret_signing = sign(secret_service, "tc3_request")
        signature = hmac.new(secret_signing, string_to_sign.encode("utf-8"), hashlib.sha256).hexdigest()

        # ************* 步骤 4：拼接 Authorization *************
        authorization = (algorithm + " " +
                        "Credential=" + tencentApiKey.secret_id + "/" + credential_scope + ", " +
                        "SignedHeaders=" + signed_headers + ", " +
                        "Signature=" + signature)

        # ************* 步骤 5：构造并发起请求 *************
        headers = {
            "Authorization": authorization,
            "Content-Type": "application/json; charset=utf-8",
            "Host": host,
            "X-TC-Action": action,
            "X-TC-Timestamp": str(timestamp),
            "X-TC-Version": version
        }

        return (headers, payload)

    async def voices(self, **kwargs) -> List[VoiceDesc]:
        return [VoiceDesc(name=v.name, gender=v.gender) for v in VOICE_LIST]
    
    async def run(self, input: TextMessage, **kwargs) -> AudioMessage:
        # 参数校验
        paramters = self.checkParameter(**kwargs)
        voice = paramters["voice"]
        speed = paramters["speed"]
        volume = paramters["volume"]
        SECRECT_ID = paramters["secret_id"]
        SECRECT_KEY = paramters["secret_key"]
        tencentCloudApiKey = TencentCloudApiKey(secret_id=SECRECT_ID, secret_key=SECRECT_KEY)
        headers, payload = self._buildRequest(input, tencentCloudApiKey, voice, volume, speed) 
        logger.debug(f"[TTS] Engine input: {input.data}")
        response = await httpxAsyncClient.post(self._url, headers=headers, data=payload)
        if response.status_code != 200:
            raise RuntimeError(f"Builtin tts api error: {response.status_code}")
        audio = response.json()["Response"]["Audio"]
        message = AudioMessage(
            data=audio,
            sampleRate=16000,
            sampleWidth=2,
        )
        return message