aiogram发送带按钮视频遇Errno13错误及按钮回调失效求助
问题1:send_video 报错
ClientOSError: [Errno 13] 原因
你给FSInputFile传的path参数是目录路径,而非完整的视频文件路径,导致读取文件时出现权限或路径错误。
解决方案
- 拼接视频文件的完整路径,替换原
FSInputFile的初始化代码:# 原错误代码 # video_file = FSInputFile(path=video_path, filename=filename) # 修正后 full_video_path = os.path.join(video_path, filename) video_file = FSInputFile(path=full_video_path) - 检查视频文件的读写权限,确保当前运行脚本的用户能读取该文件。
问题2:按钮点击无响应 & 等待按钮点击推进流程
原因
- 未将
handle_button_click注册为回调事件处理器,bot无法监听按钮点击; - main函数未启动Dispatcher的事件循环,bot无法接收任何事件;
- 没有用异步机制暂停流程,发送视频后脚本直接结束,无法等待按钮输入。
解决方案
1. 注册回调处理器并启动事件循环
需要初始化Dispatcher,将按钮点击函数绑定到事件监听中,同时启动polling让bot持续接收消息。
2. 用asyncio.Future等待按钮输入
通过Future对象在发送视频后暂停流程,按钮点击时唤醒流程继续执行(多用户场景建议用aiogram的FSM状态机,更稳定)。
完整修正代码
from aiogram.types import InlineKeyboardMarkup, InlineKeyboardButton, FSInputFile from aiogram import Bot, types, Dispatcher from aiogram.utils.executor import start_polling import os import asyncio import logging logging.basicConfig(level=logging.WARNING) logger = logging.getLogger(__name__) # 存储各用户的等待Future,用于唤醒流程 chat_futures = {} async def send_videos(bot: Bot, video_path: str, chat_id: int, buttons: bool): """发送目录下所有视频,带按钮时等待用户选择语言""" try: for filename in os.listdir(video_path): full_video_path = os.path.join(video_path, filename) video_file = FSInputFile(path=full_video_path) if buttons: keyboard = InlineKeyboardMarkup(inline_keyboard=[ [ InlineKeyboardButton(text="de", callback_data="de"), InlineKeyboardButton(text="en", callback_data="en"), InlineKeyboardButton(text="fr", callback_data="fr"), InlineKeyboardButton(text="es", callback_data="es"), InlineKeyboardButton(text="jp", callback_data="jp"), InlineKeyboardButton(text="cn", callback_data="cn"), InlineKeyboardButton(text="kr", callback_data="kr") ] ]) await bot.send_video(chat_id=chat_id, video=video_file, caption=filename, reply_markup=keyboard) # 创建Future,等待按钮点击 future = asyncio.Future() chat_futures[chat_id] = future # 暂停流程,直到按钮点击后获取结果 selected_language = await future del chat_futures[chat_id] # 后续流程示例:根据选择的语言执行操作 await bot.send_message(chat_id, f"已选择{selected_language},继续执行后续逻辑...") else: await bot.send_video(chat_id=chat_id, video=video_file, caption=filename) except Exception as e: logger.warning(f"发送视频时出错: {e}") # 出错时清理对应Future if chat_id in chat_futures: chat_futures[chat_id].set_exception(e) del chat_futures[chat_id] async def handle_button_click(callback_query: types.CallbackQuery): language = callback_query.data chat_id = callback_query.from_user.id # 必须回应用户的按钮点击,避免Telegram重复发送请求 await callback_query.answer(f"已选择{language}") # 唤醒等待的流程 if chat_id in chat_futures: chat_futures[chat_id].set_result(language) def get_token(): # 替换为你的bot token获取逻辑 return "YOUR_BOT_TOKEN" async def main(): path = os.path.join(os.getcwd(), "Videos") bot_token = get_token() bot = Bot(token=bot_token) dp = Dispatcher(bot) # 注册回调处理器,监听所有按钮点击 dp.register_callback_query_handler(handle_button_click) # 替换为目标用户的chat_id chat_id = 123456789 # 启动发送视频的异步任务 asyncio.create_task(send_videos(bot, path, chat_id, buttons=True)) # 启动polling,持续监听事件 await start_polling(dp, skip_updates=True) if __name__ == "__main__": asyncio.run(main())
内容的提问来源于stack exchange,提问作者Moldy
相关产品推荐
相关产品推荐

