#!/usr/bin/env python # -*- coding: utf-8 -*- import base64 import hashlib import hmac import json import time import uuid from datetime import datetime from urllib import parse import requests from loguru import logger from utils.config import OssConfig 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) string_to_sign = 'GET' + '&' + AccessToken._encode_text('/') + '&' + AccessToken._encode_text(query_string) 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).decode('utf-8') signature = AccessToken._encode_text(signature) full_url = 'http://nls-meta.cn-shanghai.aliyuncs.com/?Signature=%s&%s' % (signature, query_string) try: response = requests.get(full_url) response.raise_for_status() root_obj = response.json() if 'Token' in root_obj: token = root_obj['Token']['Id'] expire_time = root_obj['Token']['ExpireTime'] return token, expire_time except requests.exceptions.RequestException as e: logger.error(f"获取Token失败: {e}") logger.error(f"获取Token失败: {response.text}") return None, None class TtsHeader: def __init__(self, appkey, token): self.appkey = appkey self.token = token def to_dict(self): return {'appkey': self.appkey, 'token': self.token} class TtsContext: def __init__(self, device_id): self.device_id = device_id def to_dict(self): return {'device_id': self.device_id} class TtsRequest: def __init__(self, voice, sample_rate, format, enable_subtitle, text): self.voice = voice self.sample_rate = sample_rate self.format = format self.enable_subtitle = enable_subtitle self.text = text def to_dict(self): return { 'voice': self.voice, 'sample_rate': self.sample_rate, 'format': self.format, 'enable_subtitle': self.enable_subtitle, 'text': self.text } class TtsPayload: def __init__(self, enable_notify, notify_url, tts_request): self.enable_notify = enable_notify self.notify_url = notify_url self.tts_request = tts_request def to_dict(self): return { 'enable_notify': self.enable_notify, 'notify_url': self.notify_url, 'tts_request': self.tts_request.to_dict() } class TtsBody: def __init__(self, tts_header, tts_context, tts_payload): self.tts_header = tts_header self.tts_context = tts_context self.tts_payload = tts_payload def to_dict(self): return { 'header': self.tts_header.to_dict(), 'context': self.tts_context.to_dict(), 'payload': self.tts_payload.to_dict() } class AliyunTTS: def __init__(self): self.access_key_id = OssConfig["OSS_ACCESS_KEY_ID"] self.access_key_secret = OssConfig["OSS_ACCESS_KEY_SECRET"] self.app_key = OssConfig["APP_KEY"] self.token = None self.expire_time = None def get_token(self): """获取并缓存访问令牌""" if not self.token or time.time() + 60 > self.expire_time: self.token, self.expire_time = AccessToken.create_token( self.access_key_id, self.access_key_secret ) if self.token: logger.info(f"获取Token成功，有效期至: {time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(self.expire_time))}") else: logger.error("获取Token失败") return self.token def synthesize(self, text, voice="xiaoyun", format="mp3", sample_rate=16000, use_polling=True): """ 当 use_polling=False 时，阿里云语音合成会采用 回调模式 而非轮询模式。此时，需要提供一个 公网可访问的回调 URL， 阿里云会在语音合成完成后主动发送请求到该 URL，通知合成结果。 """ """长文本语音合成""" token = self.get_token() if not token: return None th = TtsHeader(self.app_key, token) tc = TtsContext("mydevice") tr = TtsRequest(voice, sample_rate, format, False, text) notify_url = "" if use_polling else "http://your-public-server.com/tts-callback" tp = TtsPayload(use_polling, notify_url, tr) tb = TtsBody(th, tc, tp) body = json.dumps(tb.to_dict()) polling_url = "https://nls-gateway.cn-shanghai.aliyuncs.com/rest/v1/tts/async" return request_long_tts(body, self.app_key, token, use_polling, polling_url) def request_long_tts(tts_body, appkey, token, use_polling=True, polling_url=None): """发送长文本语音合成请求""" url = 'https://nls-gateway.cn-shanghai.aliyuncs.com/rest/v1/tts/async' headers = {'Content-Type': 'application/json'} try: response = requests.post(url, data=tts_body, headers=headers) response.raise_for_status() json_data = response.json() if "error_code" in json_data and json_data["error_code"] == 20000000: task_id = json_data['data']['task_id'] request_id = json_data['request_id'] logger.info(f"语音合成任务已提交，task_id: {task_id}") if use_polling and polling_url: return wait_loop_for_complete(polling_url, appkey, token, task_id, request_id) return task_id, request_id else: logger.error(f"请求失败: {json_data}") return None, None except requests.exceptions.RequestException as e: logger.error(f'请求异常: {e}') return None, None def wait_loop_for_complete(url, appkey, token, task_id, request_id, max_retries=30): """轮询等待合成完成""" full_url = f"{url}?appkey={appkey}&task_id={task_id}&token={token}&request_id={request_id}" logger.info(f"开始轮询任务状态: {task_id}") for retries in range(max_retries): try: response = requests.get(full_url) response.raise_for_status() json_data = response.json() if "data" in json_data and "audio_address" in json_data["data"]: audio_address = json_data["data"]["audio_address"] if audio_address: logger.info(f"合成完成! audio_address = {audio_address}") return audio_address else: logger.info(f"第 {retries + 1}/{max_retries} 次轮询: 合成中...") elif "error_code" in json_data and json_data["error_code"] != 20000000: logger.error(f"合成失败: {json_data.get('error_message', '未知错误')}") return None except requests.exceptions.RequestException as e: logger.warning(f"轮询请求异常: {e}") time.sleep(10) logger.warning(f"已达到最大轮询次数({max_retries})，任务可能仍在处理中") return None if __name__ == "__main__": # 准备请求文本 tts_text = """生活中总有一些故事能让我们感受到温暖和智慧，赵元任的传奇经历就是这样一个值得分享的好故事...""" tts_client = AliyunTTS() # 语音合成 logger.info("开始语音合成...") mp3_url = tts_client.synthesize(tts_text) logger.info(f"合成的url: {mp3_url}") if not mp3_url: logger.error("语音合成失败，程序退出")