diff --git a/voice/ali/ali_api.py b/voice/ali/ali_api.py index 9d366a0..cac0c8c 100644 --- a/voice/ali/ali_api.py +++ b/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', diff --git a/voice/ali/ali_voice.py b/voice/ali/ali_voice.py index c9e90cc..3a57442 100644 --- a/voice/ali/ali_voice.py +++ b/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 \ No newline at end of file + return self.token diff --git a/voice/factory.py b/voice/factory.py index 01229eb..ed80758 100644 --- a/voice/factory.py +++ b/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