diff --git a/main/xiaozhi-server/core/providers/tts/aliyun_cosy.py b/main/xiaozhi-server/core/providers/tts/aliyun_cosy.py new file mode 100644 index 00000000..f1e49d9f --- /dev/null +++ b/main/xiaozhi-server/core/providers/tts/aliyun_cosy.py @@ -0,0 +1,132 @@ +import uuid +import hmac +import hashlib +import base64 +import requests +from datetime import datetime +from core.providers.tts.base import TTSProviderBase +from config.logger import setup_logging +import time +import uuid +from urllib import parse + +from dashscope.audio.tts_v2 import SpeechSynthesizer, AudioFormat +import dashscope +import os + +TAG = __name__ +logger = setup_logging() + + +class AccessToken: + @staticmethod + def _encode_text(text): + encoded_text = parse.quote_plus(text) + return encoded_text.replace("+", "%20").replace("*", "%2A").replace("%7E", "~") + + @staticmethod + def _encode_dict(dic): + keys = dic.keys() + dic_sorted = [(key, dic[key]) for key in sorted(keys)] + encoded_text = parse.urlencode(dic_sorted) + return encoded_text.replace("+", "%20").replace("*", "%2A").replace("%7E", "~") + + @staticmethod + def create_token(access_key_id, access_key_secret): + parameters = { + "AccessKeyId": access_key_id, + "Action": "CreateToken", + "Format": "JSON", + "RegionId": "cn-shanghai", + "SignatureMethod": "HMAC-SHA1", + "SignatureNonce": str(uuid.uuid1()), + "SignatureVersion": "1.0", + "Timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), + "Version": "2019-02-28", + } + # 构造规范化的请求字符串 + query_string = AccessToken._encode_dict(parameters) + # print('规范化的请求字符串: %s' % query_string) + # 构造待签名字符串 + string_to_sign = ( + "GET" + + "&" + + AccessToken._encode_text("/") + + "&" + + AccessToken._encode_text(query_string) + ) + # print('待签名的字符串: %s' % string_to_sign) + # 计算签名 + secreted_string = hmac.new( + bytes(access_key_secret + "&", encoding="utf-8"), + bytes(string_to_sign, encoding="utf-8"), + hashlib.sha1, + ).digest() + signature = base64.b64encode(secreted_string) + # print('签名: %s' % signature) + # 进行URL编码 + signature = AccessToken._encode_text(signature) + # print('URL编码后的签名: %s' % signature) + # 调用服务 + full_url = "http://nls-meta.cn-shanghai.aliyuncs.com/?Signature=%s&%s" % ( + signature, + query_string, + ) + # print('url: %s' % full_url) + # 提交HTTP GET请求 + response = requests.get(full_url) + if response.ok: + root_obj = response.json() + key = "Token" + if key in root_obj: + token = root_obj[key]["Id"] + expire_time = root_obj[key]["ExpireTime"] + return token, expire_time + # print(response.text) + return None, None + + +class TTSProvider(TTSProviderBase): + + def __init__(self, config, delete_audio_file): + super().__init__(config, delete_audio_file) + + self.api_key = config.get("api_key") + self.voice = config.get("private_voice") + if self.voice.startswith("cosyvoice-v3-"): + self.model = "cosyvoice-v3" + elif self.voice.startswith("cosyvoice-v2-"): + self.model = "cosyvoice-v2" + else: + self.model = "cosyvoice-v1" + + def generate_filename(self, extension=".wav"): + return os.path.join(self.output_file, + f"tts-{__name__}{datetime.now().date()}@{uuid.uuid4().hex}{extension}") + + async def text_to_speak(self, text, output_file): + try: + # 设置 DashScope 的 API 密钥 + dashscope.api_key = self.api_key + + # 创建语音合成器实例,指定模型、声音和音频格式 + synthesizer = SpeechSynthesizer( + model=self.model, + voice=self.voice, + format=AudioFormat.WAV_16000HZ_MONO_16BIT, + # callback=callback + ) + + # 调用语音合成接口,获取音频数据 + audiodata = synthesizer.call(text) + + # 如果指定了输出文件,则将音频数据写入文件,并返回文件路径 + if output_file: + with open(output_file, 'wb') as f: + f.write(audiodata) + return output_file + else: + # 否则直接返回音频数据 + return audiodata + except Exception as e: + raise Exception(f"{__name__} error: {e}") diff --git a/main/xiaozhi-server/core/utils/wakeup_word.py b/main/xiaozhi-server/core/utils/wakeup_word.py index d2f4fb32..93d4e6b8 100644 --- a/main/xiaozhi-server/core/utils/wakeup_word.py +++ b/main/xiaozhi-server/core/utils/wakeup_word.py @@ -55,7 +55,7 @@ class WakeupWordsConfig: return self._config_cache try: - with open(self.config_file, "a+") as f: + with open(self.config_file, "a+", encoding="utf-8") as f: with FileLock(f, timeout=self._lock_timeout): f.seek(0) content = f.read() @@ -73,7 +73,7 @@ class WakeupWordsConfig: def _save_config(self, config: Dict): """保存配置到文件,使用文件锁保护""" try: - with open(self.config_file, "w") as f: + with open(self.config_file, "w", encoding="utf-8") as f: with FileLock(f, timeout=self._lock_timeout): yaml.dump(config, f, allow_unicode=True) self._config_cache = config