Telegram机器人实现黑名单功能封禁用户

在Telegram机器人开发中，黑名单功能是保障群组秩序、防止恶意用户骚扰和滥发信息的利器。通过将违规用户加入黑名单，机器人可以快速封禁其访问权限，让群组管理者无需手动频繁操作。本文将带你从零开始实现一个完整的黑名单系统，包括数据存储、封禁命令、自动拦截和解封机制，并提供最佳实践建议。

为什么需要黑名单功能？

Telegram群组或频道经常受到刷屏、广告、辱骂等行为的侵扰。虽然Telegram提供了管理员踢人功能，但机器人层面的黑名单具有更灵活的自动化和持久化能力：

自动防护：当用户触发规则时，机器人可以立即封禁，无需人工干预。
跨群组管理：同一个机器人服务多个群组时，黑名单可以全局生效，封禁恶意用户的所有访问。
持久记忆：即使被封禁用户离开群组再次加入，机器人也能识别并拒绝服务。
可扩展性：黑名单可以结合积分系统、举报机制，构建复杂的用户管理体系。

数据模型设计：使用SQLite存储黑名单

黑名单数据必须持久化，首选轻量级的SQLite。设计一张简单的表，记录被封禁用户的ID、封禁时间、原因和解封状态：

CREATE TABLE IF NOT EXISTS blacklist (
    user_id INTEGER PRIMARY KEY,
    banned_at INTEGER NOT NULL,
    reason TEXT,
    is_active INTEGER DEFAULT 1
);

其中user_id是Telegram用户ID，banned_at为封禁时间戳（Unix毫秒），reason可选，is_active用于软删除，便于解封时保留历史记录。你也可以增加admin_id字段记录执行封禁的管理员。

实现封禁命令：/ban

我们使用Python和python-telegram-bot v20+异步框架来实现。首先编写添加黑名单的辅助函数：

import sqlite3
from datetime import datetime

def add_to_blacklist(user_id, reason=''):
    conn = sqlite3.connect('bot.db')
    cursor = conn.cursor()
    cursor.execute(
        'INSERT OR REPLACE INTO blacklist (user_id, banned_at, reason, is_active) VALUES (?, ?, ?, 1)',
        (user_id, datetime.now().timestamp() * 1000, reason)
    )
    conn.commit()
    conn.close()

然后定义/ban命令处理器，支持以下用法：

/ban @username：按用户名封禁
/ban userId：按ID封禁
/ban @username 原因：附带封禁原因

from telegram import Update, BotCommand
from telegram.ext import Application, CommandHandler, ContextTypes

async def ban_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    if not update.effective_message:
        return
    args = context.args
    if not args:
        await update.message.reply_text('用法：/ban <用户ID或@用户名> [原因]')
        return

    target = None
    reason = ' '.join(args[1:]) if len(args) > 1 else ''

    # 解析@用户名或ID
    if args[0].startswith('@'):
        username = args[0][1:]
        # 需要先查询用户ID，此处简化为直接使用发起用户ID作为演示
        # 实际项目中可通过搜索或缓存获取
        target = username
    else:
        target = int(args[0])

    # 添加黑名单
    if isinstance(target, int):
        add_to_blacklist(target, reason)
        # 如果机器人在群组中，可同时执行踢出操作
        try:
            await context.bot.ban_chat_member(update.effective_chat.id, target)
        except Exception as e:
            await update.message.reply_text(f'封禁失败：')
        await update.message.reply_text(f'用户  已被加入黑名单。')
    else:
        await update.message.reply_text('尚未实现通过用户名封禁，请使用用户ID。')

注意：实际中如果只有用户名，需要先通过get_updates或搜索接口获取用户ID。对于自动化封禁，建议将违规用户ID直接传入。

在消息处理中拦截黑名单用户

仅封禁了群组成员还不够，机器人还需要拒绝来自黑名单用户的任何交互。实现一个检查函数，并在所有处理器最前端调用：

def is_banned(user_id: int) -> bool:
    conn = sqlite3.connect('bot.db')
    cursor = conn.cursor()
    cursor.execute('SELECT is_active FROM blacklist WHERE user_id = ?', (user_id,))
    row = cursor.fetchone()
    conn.close()
    return row is not None and row[0] == 1

然后使用装饰器或中间件模式，在每次更新处理前检查。例如使用python-telegram-bot的TypeHandler或自定义装饰器：

from functools import wraps
from telegram import Update

def blacklist_check(func):
    @wraps(func)
    async def wrapper(update: Update, context: ContextTypes.DEFAULT_TYPE):
        user = update.effective_user
        if user and is_banned(user.id):
            # 可以选择静默忽略或发送提示
            await update.message.reply_text('您已被禁止使用此机器人。')
            return
        return await func(update, context)
    return wrapper

# 应用装饰器
@blacklist_check
async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
    await update.message.reply_text('你好！欢迎使用。')

对于群组自动封禁，还可以在ChatMemberHandler中检测用户重新加入时立即踢出：

from telegram.ext import ChatMemberHandler

async def track_chat_member(update: Update, context: ContextTypes.DEFAULT_TYPE):
    new_status = update.chat_member.new_chat_member
    if new_status.status == 'member' and is_banned(new_status.user.id):
        await context.bot.ban_chat_member(update.effective_chat.id, new_status.user.id)
        await context.bot.send_message(
            update.effective_chat.id,
            f"用户 {new_status.user.full_name} 被黑名单拦截，已自动封禁。"
        )

实现解封命令：/unban

有封禁自然也有解封。管理可以执行/unban命令，将用户移出黑名单并恢复访问：

def remove_from_blacklist(user_id: int):
    conn = sqlite3.connect('bot.db')
    cursor = conn.cursor()
    cursor.execute('UPDATE blacklist SET is_active = 0 WHERE user_id = ?', (user_id,))
    conn.commit()
    conn.close()

async def unban_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    args = context.args
    if not args:
        await update.message.reply_text('用法：/unban 用户ID')
        return
    user_id = int(args[0])
    remove_from_blacklist(user_id)
    # 尝试解除群组封禁
    try:
        await context.bot.unban_chat_member(update.effective_chat.id, user_id)
    except Exception:
        pass
    await update.message.reply_text(f'用户  已解封。')

最佳实践与注意事项

使用数据库索引：为user_id和is_active建立联合索引，提高查询速度。
异步数据库操作：SQLite同步操作可能阻塞事件循环，建议使用aiosqlite或异步ORM。
防止误封：封禁前判断用户是否管理员、是否机器人自身，避免封禁管理。
封禁原因记录：详细记录封禁原因、时间、操作者，便于审计和投诉处理。
定期清理：对于永久封禁的用户，可以定期归档旧记录，避免数据膨胀。
并行扩展：若机器人数较多，可考虑将黑名单放入Redis，提高共享性和响应速度。

总结

黑名单功能是Telegram机器人维护社群秩序的重要工具。通过数据库持久化、封禁/解封命令和消息拦截机制，你可以灵活控制用户访问权，实现自动化管理。本文的方法可直接应用于生产环境，也可根据需求扩展出白名单、警告次数组等更复杂的体系。希望你能够利用黑名单功能，打造更加安全、友好的Telegram群组。

Telegram机器人实现黑名单功能封禁用户

为什么需要黑名单功能？

数据模型设计：使用SQLite存储黑名单

实现封禁命令：/ban

在消息处理中拦截黑名单用户

实现解封命令：/unban

最佳实践与注意事项

总结

了解Telegram

下载与安装

常见问题

继续阅读