如何为Telegram Bot实现无触发事件的定时清理不活跃用户任务

你的核心问题是定时任务无法获取Update对象——因为Update只有在触发事件（比如消息、命令）时才会生成，而定时任务是自动执行的，根本没有触发事件，所以不能依赖Update来操作群组。

下面提供两种适配你需求的解决方案，优先推荐第一种（贴合python-telegram-bot生态）：

方案一：使用python-telegram-bot自带的JobQueue #

python-telegram-bot v20的JobQueue已经和应用的异步事件循环深度集成，不需要依赖事件触发，直接可以拿到Bot实例来操作群组。

修改后的代码 #

from telegram import Bot
from telegram.ext import ApplicationBuilder, Update
import datetime
from datetime import timedelta

# 替换为你的目标群组ID，或从数据库中读取多群组ID
TARGET_CHAT_ID = -1001234567890

async def remove_inactive_users(bot: Bot) -> None:
    # 查询一周未活跃的用户
    condition = "TIMESTAMPDIFF(WEEK, LAST_TIMESTAMP, NOW()) > 1"
    success, data = db.select(tablename, "*", condition)

    if success and data:
        # 删除数据库中记录的不活跃用户
        db.delete(tablename, condition)
        # 批量移除用户
        for user_data in data:
            # 假设数据库返回的字段顺序是(uid, chat_id, ...)，如果是单群可直接用TARGET_CHAT_ID
            uid = user_data[0]
            chat_id = user_data[1] if len(user_data)>=2 else TARGET_CHAT_ID
            
            try:
                # 临时禁言30秒（Telegram中临时禁言会自动移除用户，之后可重新加入）
                await bot.ban_chat_member(
                    chat_id=chat_id,
                    user_id=uid,
                    until_date=datetime.datetime.now() + timedelta(seconds=30)
                )
                print(f"已移除不活跃用户 {uid}")
            except Exception as e:
                print(f"移除用户 {uid} 失败: {str(e)}")

if __name__ == "__main__":
    application = ApplicationBuilder().token(API_KEY).build()

    # 添加每日执行的定时任务
    application.job_queue.run_repeating(
        remove_inactive_users,
        interval=timedelta(days=1),
        first=datetime.datetime.now()  # 立即执行一次，之后每日执行
    )

    # 其他处理器（如用户加入、消息更新时间戳的逻辑）...

    application.run_polling(allowed_updates=Update.ALL_TYPES, drop_pending_updates=True)

方案二：继续使用BackgroundScheduler（需处理异步兼容） #

如果你坚持用BackgroundScheduler，需要解决异步函数的执行问题——因为python-telegram-bot的API都是异步的，而BackgroundScheduler默认在同步线程运行，需要把异步任务丢到应用的事件循环中执行。

修改后的代码 #

from telegram import Bot
from telegram.ext import ApplicationBuilder, Update
from apscheduler.schedulers.background import BackgroundScheduler
import asyncio
import datetime
from datetime import timedelta

TARGET_CHAT_ID = -1001234567890

async def remove_inactive_users(bot: Bot) -> None:
    condition = "TIMESTAMPDIFF(WEEK, LAST_TIMESTAMP, NOW()) > 1"
    success, data = db.select(tablename, "*", condition)

    if success and data:
        db.delete(tablename, condition)
        for user_data in data:
            uid = user_data[0]
            chat_id = user_data[1] if len(user_data)>=2 else TARGET_CHAT_ID
            
            try:
                await bot.ban_chat_member(
                    chat_id=chat_id,
                    user_id=uid,
                    until_date=datetime.datetime.now() + timedelta(seconds=30)
                )
                print(f"已移除不活跃用户 {uid}")
            except Exception as e:
                print(f"移除用户 {uid} 失败: {str(e)}")

# 包装异步函数，让BackgroundScheduler可以调用
def run_async_task(bot: Bot):
    loop = asyncio.get_event_loop()
    loop.create_task(remove_inactive_users(bot))

if __name__ == "__main__":
    application = ApplicationBuilder().token(API_KEY).build()
    bot = application.bot

    scheduler = BackgroundScheduler()
    # 添加每日执行任务，传入Bot实例
    scheduler.add_job(run_async_task, "interval", days=1, args=(bot,))
    scheduler.start()

    # 其他处理器...

    application.run_polling(allowed_updates=Update.ALL_TYPES, drop_pending_updates=True)

关键注意事项 #

• 数据库表优化：建议在用户记录表中添加chat_id字段，这样Bot管理多个群组时，能准确知道每个用户属于哪个群。
• 权限检查：确保Bot在目标群组中拥有「删除成员」的权限，否则ban_chat_member会执行失败。
• 异常处理：必须添加try-except捕获错误，避免单个用户处理失败导致整个定时任务崩溃。

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