Telegram Bot发送音频异常：YouTube转MP3失败报错排查

问题描述 #

开发Telegram Bot时，调用YouTube转MP3的RapidAPI接口实现功能：用户发送YouTube视频URL，Bot返回可流式播放/下载的MP3文件。但实际运行中，Bot始终返回错误提示：

"Sorry, I couldn't convert the video. Please try again with a valid YouTube URL"

当前使用的代码如下：

async def convert_to_mp3(update, context):
    # Get the message text
    message_text = update.message.text
    
    # Check if the message contains a YouTube URL
    if 'youtube.com' in message_text or 'youtu.be' in message_text:
        # API endpoint and headers
        url = "https://youtube-mp315.p.rapidapi.com/"
        headers = {
            "X-RapidAPI-Key": "rapidAPI-Key",
            "X-RapidAPI-Host": "youtube-mp315.p.rapidapi.com"
        }
        
        # Query parameters
        querystring = {"url": message_text}

        # Making the API call
        response = requests.get(url, headers=headers, params=querystring)
        data = response.json()
        print("API Response:", data)  # Debugging


        # Check if the response contains the MP3 URL
        if 'url' in data:
            await update.message.reply_audio(data['url'])
        else:
            await update.message.reply_text("Sorry, I couldn't convert the video. Please try again with a valid YouTube URL.")
    else:
        await update.message.reply_text("Please send a valid YouTube video URL.")

问题排查与修复 #

1. 同步请求阻塞异步事件循环 #

代码中使用requests.get同步HTTP请求，而Telegram Bot的处理函数是async异步函数，同步调用会阻塞事件循环，导致请求超时或响应异常。需替换为异步HTTP库aiohttp。

2. URL验证不严谨 #

仅通过字符串包含判断是否为YouTube URL，无法过滤带多余字符（空格、换行）或无效格式的链接，需用正则严格匹配。

3. 缺少错误状态与异常处理 #

未检查API响应状态码，也未捕获请求过程中的异常，无法定位具体失败原因（如API密钥无效、调用次数超限、网络错误等）。

修复后的代码 #

import aiohttp
import re

async def convert_to_mp3(update, context):
    message_text = update.message.text.strip()
    
    # 严格匹配YouTube URL格式
    youtube_pattern = r'(https?://)?(www\.)?(youtube\.com/watch\?v=|youtu\.be/)[\w-]+'
    if not re.match(youtube_pattern, message_text):
        await update.message.reply_text("请发送有效的YouTube视频链接。")
        return
    
    url = "https://youtube-mp315.p.rapidapi.com/"
    headers = {
        "X-RapidAPI-Key": "你的RapidAPI密钥",
        "X-RapidAPI-Host": "youtube-mp315.p.rapidapi.com"
    }
    querystring = {"url": message_text}
    
    try:
        async with aiohttp.ClientSession() as session:
            async with session.get(url, headers=headers, params=querystring) as response:
                # 检查API响应状态码
                if response.status != 200:
                    await update.message.reply_text(f"API请求失败，状态码：{response.status}")
                    print(f"API Status Code: {response.status}")
                    return
                
                data = await response.json()
                print("API Response:", data)
                
                if 'url' in data:
                    await update.message.reply_audio(data['url'])
                else:
                    # 提取API返回的错误信息
                    error_msg = data.get('error', '未知错误')
                    print(f"API Error: {error_msg}")
                    await update.message.reply_text(f"转换失败：{error_msg}，请检查链接后重试。")
    except Exception as e:
        print(f"请求异常：{str(e)}")
        await update.message.reply_text("转换过程中出现异常，请稍后重试。")

关键修复点 #

• 用aiohttp替代requests，适配异步函数的事件循环
• 正则匹配严格验证YouTube URL格式，过滤无效输入
• 增加响应状态码检查，及时捕获API请求失败
• 提取API错误信息，便于调试和给用户精准提示
• 全局异常捕获，避免单个请求失败导致Bot崩溃

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