You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

Telegram Bot第三次函数调用后无响应问题求助

问题定位

Bot执行到第三个函数后无响应,由以下代码缺陷共同导致:

  • 消息路由缺失:自定义的test函数未添加@bot.message_handler装饰器,pyTelegramBotAPI依靠装饰器注册的规则做消息分发,未注册的函数永远不会被轮询逻辑触发。
  • 裸异常捕获吞掉所有错误:代码中多处使用无指定异常类型的except:块,索引越界、API请求失败、JSON解析错误、键值不存在等所有运行时错误都会被直接拦截,既不打印日志也不返回提示,表象就是Bot卡住无响应。典型场景是用户仅发送youtube未带搜索词时,访问query[1]直接触发索引错误,被外层异常块捕获后直接pass,无任何反馈。
  • 网络请求未设置超时:调用RapidAPI的requests请求未配置超时时间,一旦接口网络不通、无响应,整个同步轮询线程会被挂起,无法处理任何后续消息。
  • 接口错误处理逻辑断裂:判断RapidAPI返回503、429状态码并发送提示后,没有终止执行流程,后续代码依然尝试解析错误响应体中的视频链接字段,必然触发KeyError被吞。
  • 遍历列表同时修改元素:遍历搜索词列表时直接执行remove操作,容易出现搜索词拼接遗漏、索引错位的问题。
  • 导入冲突:开头from email import message和telebot回调传入的message对象命名空间冲突,会偶发参数解析异常。
  • 轮询参数配置错误:bot.infinity_polling(1440)传入的1440为长轮询超时时间(单位秒),过长的超时设置会导致消息更新延迟,同时未开启异常打印,出错后无任何日志输出。
  • 边界校验缺失:未判断用户发送的消息是否为文本、是否带齐指令参数,用户发送贴纸、图片等非文本内容,或指令缺参数时会直接触发异常被吞。
修复步骤
  • 删除无用的冲突导入from email import message
  • 为test函数添加消息处理器装饰器,配置对应的消息匹配规则
  • 替换所有裸except:块,指定捕获的具体异常类型,添加错误日志打印
  • 所有网络请求增加超时配置,避免线程挂死
  • 增加入参边界校验:用户发送非文本内容、指令缺少参数时直接返回提示,终止后续流程
  • 接口返回非200状态码时,发送提示后直接return,不再执行后续响应解析逻辑
  • 优化搜索词拼接逻辑,取消遍历删元素的写法,直接对消息文本做切片处理提取搜索词
  • 调整轮询参数,设置合理超时,开启错误追踪
修正后的可运行代码
import os
import json
import telebot
import requests
from pytube import *
from dotenv import load_dotenv

load_dotenv()

API_KEY = os.getenv("API_KEY")
RAPID_KEY = os.getenv("RAPID_API")
bot = telebot.TeleBot(API_KEY)

@bot.message_handler(commands=["start"])
def help(message):
    try:
        bot.send_message(message.chat.id, "发送“youtube + 视频名称”即可搜索对应视频。\n")
    except Exception as e:
        print(f"发送帮助信息出错: {str(e)}")
        bot.send_message(message.chat.id, "获取帮助信息失败,机器人可能已离线。\n")
 
def data_validation(message):
    # 过滤非文本消息
    if not message.text:
        return False
    query = message.text.split() 
    if len(query) < 1:
        return False
    return query[0].lower() == "youtube"

@bot.message_handler(func=data_validation)
def search(message):
    query = message.text.split()
    if len(query) < 2:
        bot.send_message(message.chat.id, "请在youtube指令后输入要搜索的视频关键词")
        return
    if query[1].lower() == "-d":
        # 可自行补充-d参数对应的下载逻辑
        return
    
    # 直接切片提取搜索词,无需遍历删除元素
    search_query = ' '.join(query[1:])
    url = "https://youtube-search-results.p.rapidapi.com/youtube-search/"
    querystring = {"q":search_query}
    headers = {
        "X-RapidAPI-Key": RAPID_KEY,
        "X-RapidAPI-Host": "youtube-search-results.p.rapidapi.com"
    }
    try:
        # 增加10秒超时避免请求挂死
        response = requests.request("GET", url, headers=headers, params=querystring, timeout=10)
        if response.status_code == 503:
            bot.send_message(message.chat.id, "当前Rapid API服务不可用,请稍后再试。\n")
            return
        if response.status_code == 429:
            bot.send_message(message.chat.id, "今日接口调用额度已用尽,请24小时后再试。\n")
            return
        if response.status_code != 200:
            bot.send_message(message.chat.id, "搜索请求出错,请稍后重试")
            return
        result = json.loads(response.text)
        first_link = str(result["items"][0]["url"])
        bot.send_message(message.chat.id, f"{first_link}\n")
    except Exception as e:
        print(f"搜索视频出错: {str(e)}")
        bot.send_message(message.chat.id, "视频加载失败。\n")

# 注册test指令的消息处理器
@bot.message_handler(func=lambda msg: msg.text and msg.text.split()[0].lower() == "test")
def test(message):
    print(message.text.split())
    bot.send_message(message.chat.id, "Test message")

# 配置合理轮询参数,开启日志打印
bot.infinity_polling(timeout=30, long_polling_timeout=30, logger_level=20)

修复后运行时控制台会输出所有错误日志,不会再出现无响应无提示的问题,test函数可正常触发。

内容的提问来源于stack exchange,提问作者Novous

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.08.27 08:54:19