Telegram机器人定时清理群组消息的代码实现

在Telegram群组中，消息积累过多不仅会让重要信息被淹没，还会增加服务器存储负担。手动清理消息耗时费力，而通过机器人定时自动清理则能高效解决问题。本文将基于Python与Telegram Bot API，为您详细讲解如何编写一个能够定时清理群组消息的机器人，并附上可直接运行的完整代码。

一、前置准备与环境要求

在开始编码之前，您需要准备以下内容：

Python 3.7或更高版本：推荐使用3.9+，以支持最新特性。
python-telegram-bot库：建议使用20.x版本，安装命令：pip install python-telegram-bot==20.7
机器人Token：通过BotFather创建机器人并获取。
群组ID与管理员权限：机器人必须加入群组并被授予管理员身份，且具有“删除消息”的权限。

二、核心思路与实现方案

实现定时清理消息的核心逻辑分为两部分：

定时调度：使用python-telegram-bot自带的JobQueue轻松安排周期性任务，无需额外依赖。
消息清理：通过bot.delete_message()方法逐条删除。但由于API限制，只能删除24小时以内发出的消息，因此需要结合群组实际消息量设计清理策略。

三、代码实现步骤

下面按步骤编写完整代码。

步骤1：导入所需模块

import logging
from datetime import timedelta
from telegram import Update
from telegram.ext import Application, CommandHandler, ContextTypes

步骤2：初始化机器人与日志

logging.basicConfig(
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
    level=logging.INFO
)
logger = logging.getLogger(__name__)

BOT_TOKEN = "YOUR_BOT_TOKEN"  # 替换为实际的Token
GROUP_CHAT_ID = -1001234567890  # 替换为要清理的群组ID（支持超链接）

步骤3：定义清理函数

async def clean_group_messages(context: ContextTypes.DEFAULT_TYPE):
    """定时清理群组消息"""
    chat_id = GROUP_CHAT_ID
    try:
        # 获取最近消息（最多100条）
        messages = await context.bot.get_chat(chat_id)
        # 注意：get_chat不能获取消息列表，我们需要通过遍历消息ID来删除。
        # 实际上Telegram Bot API没有“获取历史消息”的接口，因此我们需要主动记录消息ID。
        # 此处我们用一个示例列表，实际应用中您需要将消息ID存储在外部（如数据库或内存）。
    except Exception as e:
        logger.error(f"清理失败: ")

由于Bot API不提供直接获取群组历史消息列表的方法，强烈建议在机器人运行时将收到的每条消息的ID保存下来。下面提供一个改进方案，使用一个内存队列记录最近N条消息ID，并在定时任务中删除。

步骤4（推荐）：使用内存队列记录消息并清理

from collections import deque

message_queue = deque(maxlen=200)  # 最多保存200条消息ID

async def store_message(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """存储群组中新消息的ID"""
    if update.message and update.message.chat.type in ("group", "supergroup"):
        if update.message.chat_id == GROUP_CHAT_ID:
            message_queue.append(update.message.message_id)

async def clean_group_messages(context: ContextTypes.DEFAULT_TYPE):
    """批量删除队列中的消息"""
    if not message_queue:
        return
    if context.bot.get_chat_member(GROUP_CHAT_ID, context.bot.id).status != "administrator":
        logger.warning("机器人不是管理员，无法删除消息")
        return
    while message_queue:
        msg_id = message_queue.popleft()
        try:
            await context.bot.delete_message(GROUP_CHAT_ID, msg_id)
        except Exception as e:
            logger.error(f"删除消息  失败: ")
    logger.info(f"已清理 {len(message_queue)} 条消息")

步骤5：设置定时任务并启动

def main():
    application = Application.builder()(BOT_TOKEN).build()

    # 处理新消息以记录ID
    application.add_handler(MessageHandler(filters.TEXT & filters.Chat(GROUP_CHAT_ID), store_message))

    # 添加定时任务：每10分钟执行一次清理
    job_queue = application.job_queue
    job_queue.run_repeating(clean_group_messages, interval=timedelta(minutes=10), first=timedelta(seconds=5))

    # 启动机器人
    application.run_polling()

if __name__ == "__main__":
    main()

四、完整示例代码

下面给出一个可直接运行的完整代码文件，请根据实际情况修改配置。

import logging
from collections import deque
from datetime import timedelta
from telegram import Update
from telegram.ext import Application, MessageHandler, filters, ContextTypes

logging.basicConfig(format="%(asctime)s - %(levelname)s - %(message)s", level=logging.INFO)
logger = logging.getLogger(__name__)

BOT_TOKEN = "YOUR_BOT_TOKEN"
GROUP_CHAT_ID = -1001234567890

message_queue = deque(maxlen=200)

async def store_message(update: Update, context: ContextTypes.DEFAULT_TYPE):
    if update.message and update.message.chat.type in ("group", "supergroup"):
        if update.message.chat_id == GROUP_CHAT_ID:
            message_queue.append(update.message.message_id)

async def clean_group_messages(context: ContextTypes.DEFAULT_TYPE):
    if not message_queue:
        return
    # 检查机器人权限
    try:
        member = await context.bot.get_chat_member(GROUP_CHAT_ID, context.bot.id)
        if member.status != "administrator":
            logger.error("机器人不是管理员，无法删除消息")
            return
    except Exception as e:
        logger.error(f"获取权限失败: ")
        return
    while message_queue:
        msg_id = message_queue.popleft()
        try:
            await context.bot.delete_message(GROUP_CHAT_ID, msg_id)
            logger.info(f"已删除消息 ")
        except Exception as e:
            logger.error(f"删除消息  失败: ")
    logger.info("本轮清理完成")

def main():
    application = Application.builder().token(BOT_TOKEN).build()
    application.add_handler(MessageHandler(filters.Chat(GROUP_CHAT_ID) & (~filters.COMMAND), store_message))
    application.job_queue.run_repeating(clean_group_messages, interval=timedelta(minutes=10), first=timedelta(seconds=5))
    application.run_polling()

if __name__ == "__main__":
    main()

五、注意事项与最佳实践

消息时间限制：Telegram Bot API只允许删除24小时内发送的消息。如果队列中有超过24小时的消息，删除会失败。建议在存储消息时记录时间戳，并在清理时过滤掉超时消息。
权限要求：机器人必须被设为管理员，并勾选“删除消息”权限，否则会抛出权限不足异常。
避免误删重要消息：可以设置白名单用户或关键词，在存储时跳过这些消息。
持久化存储：内存队列在机器人重启后会丢失，如果业务要求严谨，请使用SQLite或Redis存储消息ID。
合理设置清理频率：过于频繁会增加API调用次数，请根据群组消息量调整间隔。

六、扩展功能

支持按“每天固定时间”清理，只需将run_repeating改为run_daily。
清理后自动发送通知，告知管理员清理了哪些消息。
配合数据库，实现更精确的消息保留策略（如保留最近5条）。

总结

通过本文的引导，您已经可以使用Python为Telegram群组开发一个定时清理消息的机器人。核心在于利用JobQueue实现调度，并主动存储消息ID。请务必注意API的限制和权限要求，并根据实际场景优化代码。如果您希望部署到云服务器，可参考本站其他教程完成环境配置与运行。