Azure认知服务文本转语音REST API出现ConnectTimeout错误求助

问题描述 #

使用Azure Cognitive Services Text-to-speech REST API批量转换文件夹内数千个文本文件为语音，前期转换正常，但处理20-30个文件后出现超时错误，包括TimeoutError、urllib3.exceptions.MaxRetryError及requests.exceptions.ConnectTimeout。已设置30秒转换间隔，问题仍存在，需要无需手动重启的稳定运行方案。

错误信息 #

TimeoutError: [WinError 10060] A connection attempt failed because the connected party did not properly respond after a period of time, or established connection failed because connected host has failed to respond

urllib3.exceptions.MaxRetryError: HTTPSConnectionPool(host='eastus.api.cognitive.microsoft.com', port=443): Max retries exceeded with url: /sts/v1.0/issueToken (Caused by ConnectTimeoutError(<urllib3.connection.HTTPSConnection object at 0x000001F63AF32650>, 'Connection to eastus.api.cognitive.microsoft.com timed out. (connect timeout=None)'))

raise ConnectTimeout(e, request=request)
requests.exceptions.ConnectTimeout: HTTPSConnectionPool(host='eastus.api.cognitive.microsoft.com', port=443): Max retries exceeded with url: /sts/v1.0/issueToken (Caused by ConnectTimeoutError(<urllib3.connection.HTTPSConnection object at 0x000001F63AF32650>, 'Connection to eastus.api.cognitive.microsoft.com timed out. (connect timeout=None)'))

当前脚本代码 #

import os
import requests
import time
import chardet

subscription_key = 'here my subscription key'
region = 'eastus'
voice_name = 'es-MX-DaliaNeural'
output_format = 'audio-24khz-96kbitrate-mono-mp3'

tts_url = f'https://{region}.tts.speech.microsoft.com/cognitiveservices/v1'
headers = {
    'Authorization': '',
    'Content-Type': 'application/ssml+xml',
    'X-Microsoft-OutputFormat': output_format,
    'User-Agent': 'YOUR_RESOURCE_NAME'
}

# looping through all text files in the input folder
input_folder = 'C:/path/to/text/files'
output_folder = 'C:/path/to/folder'
for filename in os.listdir(input_folder):
    # Check if the file is a text file
    if filename.endswith('.txt'):
        # Read the contents of the file and detect the encoding
        with open(os.path.join(input_folder, filename), 'rb') as f:
            rawdata = f.read()
            encoding = chardet.detect(rawdata)['encoding']
            text = rawdata.decode(encoding)

        # creating the SSML body for the TTS request
        ssml = f'<speak version="1.0" xmlns="http://www.w3.org/2001/10/synthesis" xmlns:mstts="https://www.w3.org/2001/mstts" xml:lang="es-MX"><voice name="{voice_name}">{text}</voice></speak>'

        # getting the access token for the TTS service
        token_url = f'https://{region}.api.cognitive.microsoft.com/sts/v1.0/issueToken'
        token_headers = {'Ocp-Apim-Subscription-Key': subscription_key}
        response = requests.post(token_url, headers=token_headers)
        access_token = response.text

        headers['Authorization'] = f'Bearer {access_token}'

        response = requests.Post(tts_url, headers=headers, data=ssml.encode('utf-8'))

        if response.status_code == 200:
            # save the audio content to a file
            audio_filename = os.path.splitext(filename)[0] + '.mp3'
            with open(os.path.join(output_folder, audio_filename), 'wb') as f:
                f.write(response.content)
            print(f'Successfully converted "{filename}" to speech')
        else:
            print(f'Error converting "{filename}" to speech: {response.content}')

        time.Sleep(30)

解决方案 #

核心优化方向 #

1. 复用Access Token（有效期10分钟），减少token接口请求次数
2. 添加请求超时与自动重试机制，处理临时网络波动
3. 捕获全链路异常，确保脚本不中断
4. 修复原脚本中的语法错误

优化后的脚本 #

import os
import requests
import time
import chardet
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

subscription_key = 'here my subscription key'
region = 'eastus'
voice_name = 'es-MX-DaliaNeural'
output_format = 'audio-24khz-96kbitrate-mono-mp3'

tts_url = f'https://{region}.tts.speech.microsoft.com/cognitiveservices/v1'
token_url = f'https://{region}.api.cognitive.microsoft.com/sts/v1.0/issueToken'

# 配置请求重试与超时策略
session = requests.Session()
retry_strategy = Retry(
    total=5,
    backoff_factor=1,  # 指数退避：1s→2s→4s→8s→16s
    status_forcelist=[429, 500, 502, 503, 504],
    allowed_methods=["POST"]
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("https://", adapter)
session.mount("http://", adapter)

headers = {
    'Content-Type': 'application/ssml+xml',
    'X-Microsoft-OutputFormat': output_format,
    'User-Agent': 'YOUR_RESOURCE_NAME'
}

# 初始化Token与过期时间
access_token = ""
token_expiry_time = 0

input_folder = 'C:/path/to/text/files'
output_folder = 'C:/path/to/folder'
os.makedirs(output_folder, exist_ok=True)

for filename in os.listdir(input_folder):
    if not filename.endswith('.txt'):
        continue
    
    # 读取文件内容
    file_path = os.path.join(input_folder, filename)
    try:
        with open(file_path, 'rb') as f:
            rawdata = f.read()
            encoding = chardet.detect(rawdata)['encoding'] or 'utf-8'
            text = rawdata.decode(encoding)
    except Exception as e:
        print(f'读取文件失败 {filename}: {str(e)}')
        continue

    # 生成SSML
    ssml = f'<speak version="1.0" xmlns="http://www.w3.org/2001/10/synthesis" xmlns:mstts="https://www.w3.org/2001/mstts" xml:lang="es-MX"><voice name="{voice_name}">{text}</voice></speak>'

    # 检查Token是否过期，过期则重新获取
    current_time = time.time()
    if current_time >= token_expiry_time or not access_token:
        try:
            token_response = session.post(token_url, headers={'Ocp-Apim-Subscription-Key': subscription_key}, timeout=10)
            token_response.raise_for_status()
            access_token = token_response.text
            # 设置Token过期时间为9分钟（留1分钟缓冲）
            token_expiry_time = current_time + 540
            print("已获取新的Access Token")
        except Exception as e:
            print(f'获取Token失败: {str(e)}')
            time.sleep(5)
            continue

    # 发送TTS请求，带重试逻辑
    headers['Authorization'] = f'Bearer {access_token}'
    success = False
    retry_count = 0
    max_retries = 3
    while not success and retry_count < max_retries:
        try:
            tts_response = session.post(tts_url, headers=headers, data=ssml.encode('utf-8'), timeout=20)
            tts_response.raise_for_status()
            
            # 保存音频文件
            audio_filename = os.path.splitext(filename)[0] + '.mp3'
            audio_path = os.path.join(output_folder, audio_filename)
            with open(audio_path, 'wb') as f:
                f.write(tts_response.content)
            
            print(f'成功转换: {filename}')
            success = True
        except requests.exceptions.RequestException as e:
            retry_count += 1
            wait_time = 2 ** retry_count
            print(f'转换失败 {filename} (重试 {retry_count}/{max_retries}): {str(e)}, 等待{wait_time}秒后重试')
            time.sleep(wait_time)
            # 若为权限错误，标记Token过期
            if "401" in str(e) or "Unauthorized" in str(e):
                token_expiry_time = 0

    if not success:
        print(f'最终转换失败: {filename}')

    # 控制请求频率
    time.sleep(10)

优化点说明 #

• Token复用：每9分钟重新获取一次Token，避免频繁调用Token接口
• 智能重试：针对网络错误、服务器错误自动重试，采用指数退避策略避免触发限流
• 超时控制：所有请求添加超时参数，防止脚本无限阻塞
• 异常兜底：覆盖文件读取、Token获取、TTS请求全环节异常，确保脚本持续运行
• 语法修复：修正原脚本中requests.Post、time.Sleep的大小写错误

内容的提问来源于stack exchange，提问作者eera5607