Python Telegram Bot无法检测上传文件求助（python-telegram-bot 13.7）

问题背景 #

使用python-telegram-bot 13.7版本、Python 3.11开发Telegram文件管理Bot，期望实现类Google Drive的文件上传管理功能。Bot运行无报错，但执行/upload指令激活上传模式后，发送文件无法被检测到，无法完成保存。

用户提供的核心代码片段：

def upload(update: Update, context: CallbackContext) -> None:
    if update.message.text == '/upload':
        context.chat_data['uploading'] = True
        update.message.reply_text('Upload mode activated. Send files to upload them. Send /uploadstop to stop uploading.')
        return

    if 'uploading' in context.chat_data and context.chat_data['uploading']:
        try:
            file = update.message.document
            file_name = file.file_name
            file_id = file.file_id
            newFile = context.bot.get_file(file_id)
            newFile.download(file_name)

            if 'uploaded_files' not in context.chat_data:
                context.chat_data['uploaded_files'] = []

            context.chat_data['uploaded_files'].append(file_name)
            update.message.reply_text(f'File {file_name} uploaded successfully!')

            # Show the file status as uploading and uploaded to chat_data
            if 'file_status' not in context.chat_data:
                context.chat_data['file_status'] = {}

            context.chat_data['file_status'][file_name] = 'Uploading'
            update.message.reply_text(f'File {file_name} status: {context.chat_data["file_status"][file_name]}')
            context.chat_data['file_status'][file_name] = 'Uploaded'
            update.message.reply_text(f'File {file_name} status: {context.chat_data["file_status"][file_name]}')
        except Exception as e:
            logging.error(f'Error uploading file: {e}')
            update.message.reply_text(f'Error uploading file: {e}')

问题分析 #

1. 消息类型不兼容：当前upload函数仅通过update.message.text判断文本指令，但用户上传文件时，消息类型为document，此时update.message.text为None，无法触发后续文件处理逻辑。
2. 处理器注册缺失：未明确注册document类型消息的处理器，导致文件消息无法被upload函数捕获。

解决方案 #

1. 调整消息处理逻辑，兼容指令与文件消息 #

修改upload函数，先处理文本指令，再判断文件消息并验证上传模式：

import logging
import os

def upload(update: Update, context: CallbackContext) -> None:
    # 处理/upload激活指令
    if update.message and update.message.text == '/upload':
        context.chat_data['uploading'] = True
        update.message.reply_text('Upload mode activated. Send files to upload them. Send /uploadstop to stop uploading.')
        return
    
    # 处理/uploadstop停止指令
    if update.message and update.message.text == '/uploadstop':
        context.chat_data['uploading'] = False
        update.message.reply_text('Upload mode deactivated.')
        return

    # 仅在上传模式下处理文件消息
    if 'uploading' in context.chat_data and context.chat_data['uploading']:
        try:
            # 校验是否为文件消息
            if not update.message.document:
                update.message.reply_text('Please send a valid file to upload.')
                return
                
            file = update.message.document
            file_name = file.file_name
            file_id = file.file_id
            
            # 创建专属上传目录（可选优化）
            UPLOAD_DIR = 'bot_uploads'
            os.makedirs(UPLOAD_DIR, exist_ok=True)
            save_path = os.path.join(UPLOAD_DIR, file_name)
            
            newFile = context.bot.get_file(file_id)
            newFile.download(save_path)

            # 记录已上传文件
            if 'uploaded_files' not in context.chat_data:
                context.chat_data['uploaded_files'] = []
            context.chat_data['uploaded_files'].append(save_path)
            
            # 更新并反馈文件状态
            if 'file_status' not in context.chat_data:
                context.chat_data['file_status'] = {}
            
            context.chat_data['file_status'][file_name] = 'Uploading'
            update.message.reply_text(f'File {file_name} status: {context.chat_data["file_status"][file_name]}')
            
            context.chat_data['file_status'][file_name] = 'Uploaded'
            update.message.reply_text(f'File {file_name} uploaded successfully to {save_path}!')
            
        except Exception as e:
            logging.error(f'Error uploading file: {e}')
            update.message.reply_text(f'Error uploading file: {str(e)}')

2. 完整注册消息处理器 #

在Bot初始化代码中，同时注册文本指令和文件类型的处理器，确保upload函数能捕获对应消息：

from telegram.ext import Updater, CommandHandler, MessageHandler, Filters

def main():
    # 替换为你的Bot Token
    updater = Updater("YOUR_TELEGRAM_BOT_TOKEN")
    dp = updater.dispatcher

    # 注册指令处理器
    dp.add_handler(CommandHandler("upload", upload))
    dp.add_handler(CommandHandler("uploadstop", upload))
    # 注册文件消息处理器
    dp.add_handler(MessageHandler(Filters.document, upload))

    updater.start_polling()
    updater.idle()

if __name__ == '__main__':
    main()

3. 额外优化建议 #

• 重复文件名处理：可在文件名后添加时间戳或序号，避免覆盖已上传文件：

  from datetime import datetime
  timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
  save_path = os.path.join(UPLOAD_DIR, f"{timestamp}_{file_name}")
  

• 权限控制：添加用户白名单，限制仅指定用户可使用上传功能。

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