diff --git a/voice/ali/ali_api.py b/voice/ali/ali_api.py new file mode 100644 index 0000000..cac0c8c --- /dev/null +++ b/voice/ali/ali_api.py @@ -0,0 +1,152 @@ +# coding=utf-8 +""" +Author: chazzjimel +Email: chazzjimel@gmail.com +wechat：cheung-z-x + +Description: + +""" + +import json +import time +import requests +import datetime +import hashlib +import hmac +import base64 +import urllib.parse +import uuid + +from common.log import logger +from common.tmp_dir import TmpDir + + +def text_to_speech_aliyun(url, text, appkey, token): + """ + 使用阿里云的文本转语音服务将文本转换为语音。 + + 参数: + - url (str): 阿里云文本转语音服务的端点URL。 + - text (str): 要转换为语音的文本。 + - appkey (str): 您的阿里云appkey。 + - token (str): 阿里云API的认证令牌。 + + 返回值: + - str: 成功时输出音频文件的路径，否则为None。 + """ + headers = { + "Content-Type": "application/json", + } + + data = { + "text": text, + "appkey": appkey, + "token": token, + "format": "wav" + } + + 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 + + return output_file + + +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') + res = res.replace('*', '%2A') + res = res.replace('%7E', '~') + return res + + def get_token(self): + """ + 获取阿里云服务的令牌。 + + 返回值: + - str: 获取到的令牌。 + """ + # 设置请求参数 + params = { + 'Format': 'JSON', + 'Version': '2019-02-28', + 'AccessKeyId': self.access_key_id, + 'SignatureMethod': 'HMAC-SHA1', + 'Timestamp': datetime.datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ"), + 'SignatureVersion': '1.0', + 'SignatureNonce': str(uuid.uuid4()), # 使用uuid生成唯一的随机数 + 'Action': 'CreateToken', + 'RegionId': 'cn-shanghai' + } + + # 计算签名 + signature = self.sign_request(params) + params['Signature'] = signature + + # 构造请求URL + url = 'http://nls-meta.cn-shanghai.aliyuncs.com/?' + urllib.parse.urlencode(params) + + # 发送请求 + response = requests.get(url) + + return response.text diff --git a/voice/ali/ali_voice.py b/voice/ali/ali_voice.py new file mode 100644 index 0000000..3a57442 --- /dev/null +++ b/voice/ali/ali_voice.py @@ -0,0 +1,79 @@ +# -*- coding: utf-8 -*- +""" +Author: chazzjimel +Email: chazzjimel@gmail.com +wechat：cheung-z-x + +Description: +ali voice service + +""" +import json +import os +import re +import time + +from bridge.reply import Reply, ReplyType +from common.log import logger +from voice.voice import Voice +from voice.ali.ali_api import AliyunTokenGenerator +from voice.ali.ali_api import text_to_speech_aliyun + + +class AliVoice(Voice): + def __init__(self): + """ + 初始化AliVoice类，从配置文件加载必要的配置。 + """ + try: + curdir = os.path.dirname(__file__) + config_path = os.path.join(curdir, "config.json") + with open(config_path, "r") as fr: + config = json.load(fr) + self.token = None + self.token_expire_time = 0 + self.api_url = config.get("api_url") + self.appkey = config.get("appkey") + self.access_key_id = config.get("access_key_id") + self.access_key_secret = config.get("access_key_secret") + except Exception as e: + logger.warn("AliVoice init failed: %s, ignore " % e) + + def textToVoice(self, text): + """ + 将文本转换为语音文件。 + + :param text: 要转换的文本。 + :return: 返回一个Reply对象，其中包含转换得到的语音文件或错误信息。 + """ + # 清除文本中的非中文、非英文和非基本字符 + text = re.sub(r'[^\u4e00-\u9fa5\u3040-\u30FF\uAC00-\uD7AFa-zA-Z0-9' + r'äöüÄÖÜáéíóúÁÉÍÓÚàèìòùÀÈÌÒÙâêîôûÂÊÎÔÛçÇñÑ，。！？,.]', '', text) + # 提取有效的token + token_id = self.get_valid_token() + fileName = text_to_speech_aliyun(self.api_url, text, self.appkey, token_id) + if fileName: + logger.info("[Ali] textToVoice text={} voice file name={}".format(text, fileName)) + reply = Reply(ReplyType.VOICE, fileName) + else: + reply = Reply(ReplyType.ERROR, "抱歉，语音合成失败") + 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) + token_str = get_token.get_token() + token_data = json.loads(token_str) + self.token = token_data["Token"]["Id"] + # 将过期时间减少一小段时间（例如5分钟），以避免在边界条件下的过期 + self.token_expire_time = token_data["Token"]["ExpireTime"] - 300 + logger.debug(f"新获取的阿里云token：{self.token}") + else: + logger.debug("使用缓存的token") + return self.token diff --git a/voice/ali/config.json.template b/voice/ali/config.json.template new file mode 100644 index 0000000..7f8d0e6 --- /dev/null +++ b/voice/ali/config.json.template @@ -0,0 +1,6 @@ +{ + "api_url": "https://nls-gateway-cn-shanghai.aliyuncs.com/stream/v1/tts", + "appkey": "", + "access_key_id": "", + "access_key_secret": "" +} \ No newline at end of file diff --git a/voice/factory.py b/voice/factory.py index 8725e29..ed80758 100644 --- a/voice/factory.py +++ b/voice/factory.py @@ -36,5 +36,10 @@ def create_voice(voice_type): elif voice_type == "linkai": from voice.linkai.linkai_voice import LinkAIVoice + return LinkAIVoice() + elif voice_type == "ali": + from voice.ali.ali_voice import AliVoice + + return AliVoice() raise RuntimeError