Flask集成Telegram Bot遇Event loop is closed错误求助

问题重现 #

用Flask处理HTTP请求，调用Telegram Bot发送消息时，首次请求正常返回200，但第二次请求触发以下错误并返回500：

telegram.error.NetworkError: Unknown error in HTTP implementation: RuntimeError('Event loop is closed')

移除Telegram消息发送逻辑后，功能恢复正常。

问题原因 #

全局创建的telegram.Bot实例底层依赖HTTP客户端（httpx），第一次请求时客户端会绑定到当前请求的事件循环。请求结束后该循环被销毁，第二次请求时Bot实例仍尝试使用已关闭的旧循环，导致报错。

解决方案 #

方案1：改用同步Telegram Bot方法（最简单）

直接使用同步版的send_message，避免事件循环管理问题：

from flask import Flask, request, jsonify
import logging
import os
from dotenv import load_dotenv
import telegram
from telegram import InlineKeyboardButton, InlineKeyboardMarkup
from database_central import find_user_by_port

logging.basicConfig(level=logging.INFO)
load_dotenv()

app = Flask(__name__)
bot = telegram.Bot(token=os.getenv("BOT_TOKEN"))

# 改为普通同步函数，移除await
def send_blocked_message(user_id, port):
    keyboard = [[InlineKeyboardButton("Разблокировать", callback_data='unlock')]]
    reply_markup = InlineKeyboardMarkup(keyboard)
    bot.send_message(chat_id=user_id, text='Ваш ключ заблокирован из-за подозрительной активности. Если Вы считаете что это ошибка, то нажмите кнопку ниже.', reply_markup=reply_markup)
    logging.info(f"Port {port} has been blocked for user {user_id}.")

# 改回同步视图
@app.route('/block_port', methods=['POST'])
def block_port():
    port = request.form.get('port')
    if port:
        user_id = find_user_by_port(port)
        if user_id:
            send_blocked_message(user_id, port)
            logging.info(f"Port {port} has been blocked")
            return jsonify({'message': f'Port {port} has been blocked for user {user_id}.'}), 200
        else:
            logging.info(f"Port {port} has been blocked, but no user found.")
            return jsonify({'message': f'Port {port} has been blocked, but no user found.'}), 200
    else:
        return jsonify({'error': 'Port not provided.'}), 400

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5001)

方案2：每次请求创建新的Bot实例（保留异步逻辑）

避免复用绑定了旧事件循环的Bot实例，每次发送消息时创建新实例：

from flask import Flask, request, jsonify
import logging
import os
from dotenv import load_dotenv
import telegram
from telegram import InlineKeyboardButton, InlineKeyboardMarkup
from database_central import find_user_by_port

logging.basicConfig(level=logging.INFO)
load_dotenv()

app = Flask(__name__)
BOT_TOKEN = os.getenv("BOT_TOKEN")

async def send_blocked_message(user_id, port):
    # 每次调用时创建新的Bot实例
    bot = telegram.Bot(token=BOT_TOKEN)
    keyboard = [[InlineKeyboardButton("Разблокировать", callback_data='unlock')]]
    reply_markup = InlineKeyboardMarkup(keyboard)
    await bot.send_message(chat_id=user_id, text='Ваш ключ заблокирован из-за подозрительной активности. Если Вы считаете что это ошибка, то нажмите кнопку ниже.', reply_markup=reply_markup)
    logging.info(f"Port {port} has been blocked for user {user_id}.")

@app.route('/block_port', methods=['POST'])
async def block_port():
    port = request.form.get('port')
    if port:
        user_id = find_user_by_port(port)
        if user_id:
            await send_blocked_message(user_id, port)
            logging.info(f"Port {port} has been blocked")
            return jsonify({'message': f'Port {port} has been blocked for user {user_id}.'}), 200
        else:
            logging.info(f"Port {port} has been blocked, but no user found.")
            return jsonify({'message': f'Port {port} has been blocked, but no user found.'}), 200
    else:
        return jsonify({'error': 'Port not provided.'}), 400

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5001)

方案3：使用AsyncBot手动管理事件循环（进阶）

如果需要长期复用Bot实例，用AsyncBot并确保每次请求的事件循环是活跃的：

from flask import Flask, request, jsonify
import logging
import os
import asyncio
from dotenv import load_dotenv
from telegram.ext import AsyncBot
from telegram import InlineKeyboardButton, InlineKeyboardMarkup
from database_central import find_user_by_port

logging.basicConfig(level=logging.INFO)
load_dotenv()

app = Flask(__name__)
bot = AsyncBot(token=os.getenv("BOT_TOKEN"))

async def send_blocked_message(user_id, port):
    keyboard = [[InlineKeyboardButton("Разблокировать", callback_data='unlock')]]
    reply_markup = InlineKeyboardMarkup(keyboard)
    await bot.send_message(chat_id=user_id, text='Ваш ключ заблокирован из-за подозрительной активности. Если Вы считаете что это ошибка, то нажмите кнопку ниже.', reply_markup=reply_markup)
    logging.info(f"Port {port} has been blocked for user {user_id}.")

@app.route('/block_port', methods=['POST'])
async def block_port():
    port = request.form.get('port')
    if port:
        user_id = find_user_by_port(port)
        if user_id:
            # 确保事件循环存在
            try:
                loop = asyncio.get_running_loop()
            except RuntimeError:
                loop = asyncio.new_event_loop()
                asyncio.set_event_loop(loop)
            await send_blocked_message(user_id, port)
            logging.info(f"Port {port} has been blocked")
            return jsonify({'message': f'Port {port} has been blocked for user {user_id}.'}), 200
        else:
            logging.info(f"Port {port} has been blocked, but no user found.")
            return jsonify({'message': f'Port {port} has been blocked, but no user found.'}), 200
    else:
        return jsonify({'error': 'Port not provided.'}), 400

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5001)

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