为什么需要定时清理群成员?
在Telegram社群运营中,群组规模扩大后难免混入垃圾账号、广告机器人或长期沉默的成员。手动清理费时费力,且容易遗漏。定时清理群成员功能可以大幅提升群组管理效率,维护社群质量。
Telegram机器人API提供了踢出成员(kickChatMember)和封禁成员(banChatMember)等接口,结合后台定时任务,即可实现完全自动化的成员清理。本文将基于python-telegram-bot框架,演示从零构建一个定时清理机器人的完整流程。
一、明确需求与边界条件
在写代码之前,必须明确以下问题:
- 清理对象:是清理所有非活跃成员,还是只针对特定条件(如加入超过30天且未发言)?
- 清理方式:踢出(可重新加入)还是封禁(永久禁止)?
- 执行周期:每天、每周还是每月?
- 排除名单:管理员、VIP成员是否豁免?
建议先输出一份成员列表清单进行人工复核,确认规则无误后再启用自动执行。
二、机器人权限与API准备
机器人必须拥有以下权限才能正常清理成员:
- 通过BotFather创建机器人,获取
API Token。 - 将机器人设为群组管理员,并开启“封禁用户”权限。
- 若需要读取成员信息,还需开启“成员列表”相关权限。
关键API方法:
getChatAdministrators— 获取管理员列表,用于排除。getChatMember— 获取单个成员信息(如加入时间、最后发言时间)。unbanChatMember— 解除封禁(若需要先踢出再允许重新加入)。banChatMember— 封禁成员。
三、核心代码实现:成员信息收集与清理逻辑
以下示例使用python-telegram-bot v20+,展示如何遍历群组成员并清理超过30天未活跃的用户。
1. 获取群组成员列表
from telegram import Bot
import asyncio
async def get_group_members(bot, chat_id):
members = []
offset = 0
while True:
# 注意:Telegram API没有直接的“获取所有成员”方法,需要通过搜索或迭代方式
# 这里使用智能方式:通过 recent_chat_members 或利用 get_chat_member_count + 分批获取
# 实际更推荐使用 telethon 库,但本文保持简单
break # 简化代码,真实实现需参考相关库
return members
注意:官方Bot API并未提供“列出所有群成员”的端点。如果需要全量遍历,建议使用MTProto库(如Telethon)或让机器人每日记录成员加入事件。下文将展示更实用的替代方案。
2. 利用加入时间与活跃度筛选
在Bot API限制下,最可行的方法是在机器人本地数据库中记录每位成员的加入时间和最后发言时间。通过监听new_chat_members和message事件来更新状态。
from telegram import Update
from telegram.ext import Application, MessageHandler, filters, CommandHandler
# 假设使用SQLite存储用户状态
import sqlite3
conn = sqlite3.connect('members.db', check_same_thread=False)
async def on_new_member(update: Update, context):
for member in update.message.new_chat_members:
if not member.is_bot:
conn.execute('INSERT OR IGNORE INTO members (user_id, join_time, last_active) VALUES (?,?,?)',
(member.id, time.time(), time.time()))
conn.commit()
async def on_message(update: Update, context):
user_id = update.effective_user.id
conn.execute('UPDATE members SET last_active = ? WHERE user_id = ?', (time.time(), user_id))
conn.commit()
3. 定时清理任务
使用APScheduler定时执行清理函数:
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from datetime import datetime, timedelta
async def clean_inactive_members(context):
bot = context.bot
chat_id = -1001234567890 # 替换为你的群ID
admins = [admin.user.id for admin in await bot.get_chat_administrators(chat_id)]
cutoff = time.time() - 30 * 86400 # 30天前
rows = conn.execute('SELECT user_id FROM members WHERE last_active < ?', (cutoff,)).fetchall()
for row in rows:
user_id = row[0]
if user_id in admins:
continue
# 踢出群组(可再加入)
await bot.ban_chat_member(chat_id, user_id)
await bot.unban_chat_member(chat_id, user_id)
conn.execute('DELETE FROM members WHERE user_id = ?', (user_id,))
conn.commit()
scheduler = AsyncIOScheduler()
scheduler.add_job(clean_inactive_members, 'cron', hour=3, minute=0, args=[application], id='cleanup')
scheduler.start()
四、高级功能与技巧
1. 使用Telethon获取完整成员列表
如果群成员较少(<10000),可以使用Telethon的GetParticipantsRequest实现全量扫描,并直接获取加入日期。
from telethon.sync import TelegramClient
from telethon.tl.functions.channels import GetParticipantsRequest
from telethon.tl.types import ChannelParticipantsSearch
client = TelegramClient('session', api_id, api_hash)
client.start()
async def fetch_all(chat):
offset = 0
limit = 200
while True:
participants = await client(GetParticipantsRequest(
channel=chat, filter=ChannelParticipantsSearch(''), offset=offset,
limit=limit, hash=0))
if not participants.users:
break
for user in participants.users:
# 处理每个用户
offset += len(participants.users)
注意:Telethon需要API ID和API Hash,且使用用户账号身份运行,存在一定风险,建议使用小号。
2. 白名单与保护机制
在清理前,务必检查用户是否为管理员、是否在VIP白名单中,以及是否被标记为“受保护”。建议在数据库中增加is_exempt字段。
3. 日志与通知
每次清理操作应记录日志,并通过机器人发送执行摘要给群主或管理群,便于审计。
五、常见问题与避坑指南
- 权限不足:机器人未成为管理员或未开启封禁权限,会导致
Forbidden异常。 - 频率限制:Telegram API有速率限制,大规模清理时应适当
sleep(如每操作3-5次暂停1秒)。 - 踢出后不能重新加入:使用
ban后必须unban才能允许重新加入。 - 数据一致性:如果成员退出群组,应监听
left_chat_member事件及时删除本地记录。
六、完整运行与部署建议
建议使用Docker或supervisor部署机器人,确保常驻后台。同时,可将清理任务拆分为“预览模式”和“执行模式”,先发送待清理名单到管理群,得到确认后再实际执行,降低误操作风险。
总结
定时清理群成员的核心在于数据收集与调度策略。虽然Bot API存在限制,但结合本地数据库和监听事件,完全可以实现高效精准的自动化清理。务必做好权限校验、异常处理和日志审计,才能打造真正可靠的社群管理机器人。
如果你正在构建Telegram社群自动化工具,希望本文能为你提供实用的参考。动手实践时,先从测试群开始,逐步优化规则吧。