add ali voice output

增加阿里云语音输出接口

voice/ali/ali_api.py

@@ -7,9 +7,9 @@ wechat：cheung-z-x
 Description:
 
 """
+
 import json
 import time
-
 import requests
 import datetime
 import hashlib
@@ -23,12 +23,22 @@ from common.tmp_dir import TmpDir
 
 
 def text_to_speech_aliyun(url, text, appkey, token):
-    # 请求的headers
+    """
+    使用阿里云的文本转语音服务将文本转换为语音。
+
+    参数:
+    - url (str): 阿里云文本转语音服务的端点URL。
+    - text (str): 要转换为语音的文本。
+    - appkey (str): 您的阿里云appkey。
+    - token (str): 阿里云API的认证令牌。
+
+    返回值:
+    - str: 成功时输出音频文件的路径，否则为None。
+    """
     headers = {
         "Content-Type": "application/json",
     }
 
-    # 请求的payload
     data = {
         "text": text,
         "appkey": appkey,
@@ -36,20 +46,15 @@ def text_to_speech_aliyun(url, text, appkey, token):
         "format": "wav"
     }
 
-    # 发送POST请求
     response = requests.post(url, headers=headers, data=json.dumps(data))
 
-    # 检查响应状态码和内容类型
     if response.status_code == 200 and response.headers['Content-Type'] == 'audio/mpeg':
-        # 构造唯一的文件名
         output_file = TmpDir().path() + "reply-" + str(int(time.time())) + "-" + str(hash(text) & 0x7FFFFFFF) + ".wav"
 
-        # 将响应内容写入文件
         with open(output_file, 'wb') as file:
             file.write(response.content)
         logger.debug(f"音频文件保存成功，文件名：{output_file}")
     else:
-        # 打印错误信息
         logger.debug("响应状态码: {}".format(response.status_code))
         logger.debug("响应内容: {}".format(response.text))
         output_file = None
@@ -58,28 +63,55 @@ def text_to_speech_aliyun(url, text, appkey, token):
 
 
 class AliyunTokenGenerator:
+    """
+    用于生成阿里云服务认证令牌的类。
+
+    属性:
+    - access_key_id (str): 您的阿里云访问密钥ID。
+    - access_key_secret (str): 您的阿里云访问密钥秘密。
+    """
+
     def __init__(self, access_key_id, access_key_secret):
         self.access_key_id = access_key_id
         self.access_key_secret = access_key_secret
 
     def sign_request(self, parameters):
-        # 将参数排序
+        """
+        为阿里云服务签名请求。
+
+        参数:
+        - parameters (dict): 请求的参数字典。
+
+        返回值:
+        - str: 请求的签名签章。
+        """
+        # 将参数按照字典顺序排序
         sorted_params = sorted(parameters.items())
 
-        # 构造待签名的字符串
+        # 构造待签名的查询字符串
         canonicalized_query_string = ''
         for (k, v) in sorted_params:
             canonicalized_query_string += '&' + self.percent_encode(k) + '=' + self.percent_encode(v)
 
+        # 构造用于签名的字符串
         string_to_sign = 'GET&%2F&' + self.percent_encode(canonicalized_query_string[1:])  # 使用GET方法
 
-        # 计算签名
+        # 使用HMAC算法计算签名
         h = hmac.new((self.access_key_secret + "&").encode('utf-8'), string_to_sign.encode('utf-8'), hashlib.sha1)
         signature = base64.encodebytes(h.digest()).strip()
 
         return signature
 
     def percent_encode(self, encode_str):
+        """
+        对字符串进行百分比编码。
+
+        参数:
+        - encode_str (str): 要编码的字符串。
+
+        返回值:
+        - str: 编码后的字符串。
+        """
         encode_str = str(encode_str)
         res = urllib.parse.quote(encode_str, '')
         res = res.replace('+', '%20')
@@ -88,6 +120,12 @@ class AliyunTokenGenerator:
         return res
 
     def get_token(self):
+        """
+        获取阿里云服务的令牌。
+
+        返回值:
+        - str: 获取到的令牌。
+        """
         # 设置请求参数
         params = {
             'Format': 'JSON',

voice/ali/ali_voice.py

@@ -20,15 +20,11 @@ from voice.ali.ali_api import AliyunTokenGenerator
 from voice.ali.ali_api import text_to_speech_aliyun
 
 
-def textContainsEmoji(text):
-    # 此正则表达式匹配大多数表情符号和特殊字符
-    pattern = re.compile(
-        '[\U0001F600-\U0001F64F\U0001F300-\U0001F5FF\U0001F680-\U0001F6FF\U0001F700-\U0001F77F\U0001F780-\U0001F7FF\U0001F800-\U0001F8FF\U0001F900-\U0001F9FF\U0001FA00-\U0001FA6F\U0001FA70-\U0001FAFF\U00002702-\U000027B0\U00002600-\U000026FF]')
-    return bool(pattern.search(text))
-
-
 class AliVoice(Voice):
     def __init__(self):
+        """
+        初始化AliVoice类，从配置文件加载必要的配置。
+        """
         try:
             curdir = os.path.dirname(__file__)
             config_path = os.path.join(curdir, "config.json")
@@ -43,13 +39,17 @@ class AliVoice(Voice):
         except Exception as e:
             logger.warn("AliVoice init failed: %s, ignore " % e)
 
-    # def voiceToText(self, voice_file):
-    #     pass
-
     def textToVoice(self, text):
+        """
+        将文本转换为语音文件。
+
+        :param text: 要转换的文本。
+        :return: 返回一个Reply对象，其中包含转换得到的语音文件或错误信息。
+        """
+        # 清除文本中的非中文、非英文和非基本字符
         text = re.sub(r'[^\u4e00-\u9fa5\u3040-\u30FF\uAC00-\uD7AFa-zA-Z0-9'
                       r'äöüÄÖÜáéíóúÁÉÍÓÚàèìòùÀÈÌÒÙâêîôûÂÊÎÔÛçÇñÑ，。！？,.]', '', text)
-        # 提取 token_id 值
+        # 提取有效的token
         token_id = self.get_valid_token()
         fileName = text_to_speech_aliyun(self.api_url, text, self.appkey, token_id)
         if fileName:
@@ -60,6 +60,11 @@ class AliVoice(Voice):
         return reply
 
     def get_valid_token(self):
+        """
+        获取有效的阿里云token。
+
+        :return: 返回有效的token字符串。
+        """
         current_time = time.time()
         if self.token is None or current_time >= self.token_expire_time:
             get_token = AliyunTokenGenerator(self.access_key_id, self.access_key_secret)
@@ -71,4 +76,4 @@ class AliVoice(Voice):
             logger.debug(f"新获取的阿里云token：{self.token}")
         else:
             logger.debug("使用缓存的token")
-        return self.token
+        return self.token

voice/factory.py

@@ -29,6 +29,15 @@ def create_voice(voice_type):
         from voice.azure.azure_voice import AzureVoice
 
         return AzureVoice()
+    elif voice_type == "elevenlabs":
+        from voice.elevent.elevent_voice import ElevenLabsVoice
+
+        return ElevenLabsVoice()
+
+    elif voice_type == "linkai":
+        from voice.linkai.linkai_voice import LinkAIVoice
+
+        return LinkAIVoice()
     elif voice_type == "ali":
         from voice.ali.ali_voice import AliVoice