Telegram使用Pyrogram自动化群组清理:从脚本到定时任务的实战指南

本文详细介绍如何使用Pyrogram库编写脚本,实现Telegram群组的自动清理,包括删除指定消息、移除违规成员、退出僵尸群等,并提供定时任务配置方案。

阅读提示建议先浏览小标题,再根据需要深入阅读具体段落。

群组管理是Telegram使用中常见的痛点,尤其是活跃群组,每天产生大量垃圾信息、广告和违规成员。手动清理既耗时又效率低下,而Pyrogram作为Telegram的MTProto客户端库,允许我们编写脚本来自动化这些操作。本文将带你从零开始,使用Python和Pyrogram构建一个群组清理工具,实现消息过滤、成员管理、僵尸群退出的全自动化。

为什么选择Pyrogram?

Pyrogram是一个基于MTProto的现代Telegram客户端库,与Bot API不同,它可以模拟用户账号行为,拥有更丰富的功能:访问所有对话、删除他人消息、管理成员、主动退出群组等。这使得它成为自动化群组管理的理想选择。此外,Pyrogram支持异步操作,性能优异,且文档完善。

环境准备与API凭证获取

安装Pyrogram

pip install pyrogram tgcrypto

tgcrypto是加速库,强烈推荐安装。

获取API ID和API Hash

访问 my.telegram.org,登录后进入"API development tools",创建一个应用,即可获得 api_idapi_hash。注意保密,不要公开。

编写第一个自动化清理脚本

1. 客户端初始化与登录

from pyrogram import Client

api_id = 12345
api_hash = "abcdef1234567890"

app = Client("my_account", api_id=api_id, api_hash=api_hash)

async def main():
    async with app:
        print("登录成功")

app.run(main())

首次运行会在终端要求输入手机号和验证码,并生成session文件,之后自动登录。

2. 清理指定群组中的广告消息

下面的脚本会扫描指定群组最近的消息,删除包含"广告"或"加群"等关键词的消息。

from pyrogram import Client, filters
from pyrogram.enums import MessagesFilter

app = Client("my_account")

TARGET_GROUP = "my_group_username"  # 或 -100xxxxxxxxx
FORBIDDEN_WORDS = ["广告", "加群", "推广"]

async def delete_spam_messages():
    async with app:
        found = 0
        async for message in app.get_chat_history(TARGET_GROUP, limit=200):
            if message.text:
                if any(word in message.text.lower() for word in FORBIDDEN_WORDS):
                    await message.delete()
                    found += 1
                    print(f"已删除消息 {message.id}: {message.text[:30]}")
        print(f"共删除  条消息")

app.run(delete_spam_messages())

注意:删除他人消息需要你是管理员,且具有删除消息权限。

3. 移除不活跃的成员

如果群组规定成员需定期发言,你可以自动移除超过N天未说话的用户。

from datetime import datetime, timedelta
from pyrogram import Client

app = Client("my_account")

TARGET_GROUP = "my_group"
INACTIVE_DAYS = 30  # 30天未发言

async def kick_inactive_members():
    async with app:
        group = await app.get_chat(TARGET_GROUP)
        members = app.get_chat_members(group.id)
        now = datetime.now()
        kicked = 0
        async for member in members:
            # 跳过管理员和群主
            if member.status in ("administrator", "creator"):
                continue
            # 计算最后发言时间
            last_active = member.user.last_online_date or member.user.last_seen
            if last_active and (now - last_active).days > INACTIVE_DAYS:
                await app.ban_chat_member(group.id, member.user.id)
                await app.unban_chat_member(group.id, member.user.id)  # 解除封禁,但此操作会移除成员
                kicked += 1
                print(f"已移除 {member.user.first_name} (ID: {member.user.id})")
        print(f"共移除  名不活跃成员")

app.run(kick_inactive_members())

注意:last_online_date 不一定准确,且需要管理员权限。

4. 自动退出长时间无消息的群组

from pyrogram import Client

app = Client("my_account")

DAYS_WITHOUT_MESSAGE = 90

async def leave_dead_groups():
    async with app:
        async for dialog in app.get_dialogs():
            chat = dialog.chat
            if chat.type == "group":  # 仅处理普通群组,不处理频道
                last_message = dialog.top_message
                if last_message:
                    age = (datetime.now() - last_message.date).days
                    if age > DAYS_WITHOUT_MESSAGE:
                        await app.leave_chat(chat.id)
                        print(f"已退出群组 {chat.title} (ID: {chat.id})")

app.run(leave_dead_groups())

请谨慎使用,退出后若未保存邀请链接可能需要重新被拉入。

定时执行与部署

为了实现真正的自动化,我们可以将脚本部署在服务器上,并用定时任务触发。以下是常见方法:

使用Cron(Linux/macOS)

编辑crontab:

crontab -e

添加一行,每天凌晨2点运行清理脚本:

0 2 * * * /usr/bin/python3 /path/to/cleanup_script.py

注意:脚本需要支持命令行触发,即包含 if __name__ == "__main__": app.run(...)

使用Windows任务计划程序

创建基本任务,设置触发器为每日固定时间,操作指向python.exe和脚本路径。推荐使用Python的schtasks工具或GUI设置。

使用APScheduler(Python库)

为了更灵活的调度,可以用APScheduler在脚本内部循环执行。

from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.triggers.interval import IntervalTrigger
import asyncio
from pyrogram import Client

app = Client("my_account")
scheduler = AsyncIOScheduler()

async def cleanup_job():
    async with app:
        await delete_spam_messages()
        await kick_inactive_members()

if __name__ == "__main__":
    app.start()
    scheduler.add_job(cleanup_job, IntervalTrigger(hours=24), id="cleanup")
    scheduler.start()
    asyncio.get_event_loop().run_forever()

避免触发频率限制的最佳实践

调用Pyrogram的删除、封禁等操作会消耗API限额,过快的请求可能导致账号被临时限制。建议:

  • 在每次操作之间添加 await asyncio.sleep(1)
  • 使用 app.set_sleep_threshold(60) 调整睡眠阈值。
  • 不要并发执行大量删除操作,分批处理。
  • 使用代理(如SOCKS5)降低风险,但务必在配置中指定。

高级技巧:动态配置与白名单

为了避免误删重要消息或误踢成员,可以引入白名单机制。例如,只清理来自特定用户组的消息,或保护指定用户ID。

WHITELISTED_USERS = [123456789]  # 不清理这些用户的消息

async def delete_spam_messages():
    async for message in app.get_chat_history(TARGET_GROUP, limit=200):
        if message.from_user and message.from_user.id not in WHITELISTED_USERS:
            # 检查关键词...
            pass

总结

借助Pyrogram,我们只需几十行代码就能实现复杂的群组清理逻辑,配合定时任务即可实现全自动化管理。当然,自动清理需要谨慎,建议先在测试群中试运行,并逐步完善过滤规则。希望本文能帮助你从繁琐的群组维护中解放出来,享受更高效的管理体验。

如果你有更多关于Telegram自动化的想法,欢迎在评论区交流。

FAQ

安装与配置指南

常见问题

如何获取Pyrogram所需的api_id和api_hash?

访问 my.telegram.org,登录后点击 'API development tools',创建一个应用即可获得。注意不要泄露这两个值。

脚本运行时报错 'A wait of N seconds is required' 怎么办?

这是Telegram的频率限制提示。可适当增加sleep间隔,或使用set_sleep_threshold提高默认睡眠时间,避免连续请求。

可以用Pyrogram清理频道中的消息吗?

可以,但需要你有管理员权限。将对话类型改为channel,并使用get_chat_history遍历消息即可,操作与群组类似。

自动移除不活跃成员时会误踢吗?

有风险。建议先导出成员最后在线时间进行分析,并设置合理的不活跃天数。同时建议跳过管理员和已经移除过的用户。