准备工作与环境搭建
在开始编写代码之前，必须先完成基础环境的配置。本指南基于Python 3.8+版本进行开发，利用python-telegram-bot库实现高效的异步处理。请确保你的操作系统已经安装了Python环境。
1. 获取Bot Token
Telegram的所有机器人都必须通过Token进行身份验证。获取步骤如下：
- 在Telegram中搜索官方机器人 @BotFather。
- 点击“开始”按钮，发送命令 /newbot。
- 按照提示输入机器人的名称（例如：MyStableBot）和用户名（例如：MyStableBot_bot，用户名必须以_bot结尾）。
- 创建成功后，BotFather会返回一串以 Bot 结尾的字符串，例如：
123456789:ABCdefGHIjkLMNoPQRsTUVwxyz。
- 务必复制并保存这串Token，它是后续连接API的唯一凭证。
2. 获取群组ID
为了确保机器人只在指定的群组中工作，我们需要获取目标群组的Chat ID。这是最容易被忽略的一步。
- 将刚才创建的机器人拉入你的社群，并将其设置为 管理员（必须设置为管理员，否则机器人无法删除消息或踢人）。
- 在群组中发送任意消息，例如 /test。
- 在浏览器中访问以下API链接（将YOUR_BOT_TOKEN替换为上一步获取的Token）：
https://api.telegram.org/botYOUR_BOT_TOKEN/getUpdates
- 你会看到JSON格式的返回数据，找到
"chat":{"id":-1001234567890,...} 这一节点。
- 复制那个负数ID（例如 -1001234567890），这就是你的群组ID。
3. 安装依赖库
打开终端或命令行工具，执行以下命令安装必要的库。我们使用v20版本的库，它支持全异步模式，性能更强。
```bash
pip install python-telegram-bot==20.7
```
核心功能模块开发
我们将项目拆分为配置文件和主逻辑文件，便于维护。请在本地新建一个文件夹，并在其中创建 config.py 和 bot.py 两个文件。
1. 配置文件编写
配置文件用于集中管理敏感信息和规则。打开 config.py，填入以下内容。请确保将占位符替换为真实数据。
```python
config.py
你的机器人Token，从BotFather获取
BOT_TOKEN = "123456789:ABCdefGHIjkLMNoPQRsTUVwxyz"
你的社群群组ID，必须是负数
ALLOWED_CHAT_ID = -1001234567890
违规关键词列表，机器人会删除包含这些词的消息
BANNED_KEYWORDS = [
"兼职", "刷单", "代练", "菠菜", "彩票",
"http", "telegram.me", "t.me/", "引流"
]
频率限制：用户在几秒内只能发送几条消息
RATE_LIMIT_SECONDS = 3
RATE_LIMIT_COUNT = 2
欢迎语，新人进群时自动发送
WELCOME_MESSAGE = (
"欢迎新朋友加入社群！\n"
"请仔细阅读群规：\n"
"1. 禁止发布任何广告及引流链接\n"
"2. 禁止讨论敏感政治话题\n"
"3. 违规者将被自动踢出，谢谢配合！"
)
```
2. 敏感词过滤与自动踢人
这是维稳的核心功能。我们需要监听群组的所有文本消息，检查是否包含违禁词。如果包含，机器人立即撤回消息并尝试禁言或踢出该用户。
在 bot.py 中引入必要的模块并编写过滤逻辑：
```python
import logging
from datetime import datetime, timedelta
from telegram import Update
from telegram.ext import ApplicationBuilder, ContextTypes, MessageHandler, filters, ChatMemberHandler
from config import BOT_TOKEN, ALLOWED_CHAT_ID, BANNED_KEYWORDS, WELCOME_MESSAGE
开启日志记录，方便排查问题
logging.basicConfig(
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
level=logging.INFO
)
存储用户消息时间戳的字典，用于频率限制
user_history = {}
async def check_keywords(update: Update, context: ContextTypes.DEFAULT_TYPE):
"""检查消息是否包含违禁词"""
message = update.effective_message
user = update.effective_user
只处理指定群组的消息
if update.effective_chat.id != ALLOWED_CHAT_ID:
return
只处理文本消息，忽略图片、贴纸等
if not message.text:
return
1. 关键词过滤逻辑
text = message.text.lower()
for keyword in BANNED_KEYWORDS:
if keyword.lower() in text:
try:
删除违规消息
await message.delete()
尝试踢出用户（需要机器人是管理员）
await context.bot.ban_chat_member(chat_id=ALLOWED_CHAT_ID, user_id=user.id)
logging.warning(f"用户 {user.id} 因发送违禁词 '{keyword}' 已被踢出")
except Exception as e:
logging.error(f"处理违规失败: {e}")
return 只要命中一个词就停止检查
2. 频率限制逻辑（防刷屏）
now = datetime.now()
uid = user.id
如果用户不在历史记录中，初始化
if uid not in user_history:
user_history[uid] = []
清理该用户10秒之前的记录（避免内存泄漏）
user_history[uid] = [t for t in user_history[uid] if t > now - timedelta(seconds=10)]
添加当前消息时间
user_history[uid].append(now)
检查频率：如果在设定的秒数内发送了超过设定数量的消息
if len(user_history[uid]) > RATE_LIMIT_COUNT:
检查最早的消息和当前消息的时间差
time_diff = (user_history[uid][-1] - user_history[uid][0]).total_seconds()
if time_diff < RATE_LIMIT_SECONDS:
try:
await message.delete()
可以选择禁言一段时间，这里直接踢出
await context.bot.ban_chat_member(chat_id=ALLOWED_CHAT_ID, user_id=uid)
logging.warning(f"用户 {uid} 因刷屏已被踢出")
清理该用户记录
del user_history[uid]
except Exception as e:
logging.error(f"处理刷屏失败: {e}")
```
3. 新人入群自动审核与欢迎
为了维持群内氛围，我们需要在新人加入时自动发送群规，起到警示作用。同时，也可以在此处加入逻辑，检查新成员的账号注册时间，如果是刚注册的小号，可以直接踢出（本代码仅展示欢迎功能）。
```python
async def greet_new_members(update: Update, context: ContextTypes.DEFAULT_TYPE):
"""处理新成员加入事件"""
确保是群组聊天
if update.chat_member and update.chat_member.chat.id == ALLOWED_CHAT_ID:
new_status = update.chat_member.new_chat_member
if new_status.status == "member":
user = new_status.user
try:
await context.bot.send_message(
chat_id=ALLOWED_CHAT_ID,
text=WELCOME_MESSAGE
)
logging.info(f"用户 {user.full_name} ({user.id}) 加入群组")
except Exception as e:
logging.error(f"发送欢迎消息失败: {e}")
```
完整代码整合
将上述逻辑整合到 bot.py 的主入口中。确保文件结构完整，可以直接运行。
```python
bot.py 完整内容
import logging
from datetime import datetime, timedelta
from telegram import Update
from telegram.ext import ApplicationBuilder, ContextTypes, MessageHandler, filters, ChatMemberHandler
from config import BOT_TOKEN, ALLOWED_CHAT_ID, BANNED_KEYWORDS, WELCOME_MESSAGE, RATE_LIMIT_SECONDS, RATE_LIMIT_COUNT
logging.basicConfig(
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
level=logging.INFO
)
user_history = {}
async def check_keywords(update: Update, context: ContextTypes.DEFAULT_TYPE):
message = update.effective_message
user = update.effective_user
if update.effective_chat.id != ALLOWED_CHAT_ID:
return
if not message.text:
return
关键词过滤
text = message.text.lower()
for keyword in BANNED_KEYWORDS:
if keyword.lower() in text:
try:
await message.delete()
await context.bot.ban_chat_member(chat_id=ALLOWED_CHAT_ID, user_id=user.id)
logging.warning(f"用户 {user.id} 因发送违禁词 '{keyword}' 已被踢出")
except Exception as e:
logging.error(f"处理违规失败: {e}")
return
频率限制
now = datetime.now()
uid = user.id
if uid not in user_history:
user_history[uid] = []
user_history[uid] = [t for t in user_history[uid] if t > now - timedelta(seconds=10)]
user_history[uid].append(now)
if len(user_history[uid]) > RATE_LIMIT_COUNT:
time_diff = (user_history[uid][-1] - user_history[uid][0]).total_seconds()
if time_diff < RATE_LIMIT_SECONDS:
try:
await message.delete()
await context.bot.ban_chat_member(chat_id=ALLOWED_CHAT_ID, user_id=uid)
)
logging.warning(f"用户 {uid} 因刷屏已被踢出")
del user_history[uid]
except Exception as e:
logging.error(f"处理刷屏失败: {e}")
async def greet_new_members(update: Update, context: ContextTypes.DEFAULT_TYPE):
if update.chat_member and update.chat_member.chat.id == ALLOWED_CHAT_ID:
new_status = update.chat_member.new_chat_member
if new_status.status == "member":
user = new_status.user
try:
await context.bot.send_message(
chat_id=ALLOWED_CHAT_ID,
text=WELCOME_MESSAGE
)
logging.info(f"用户 {user.full_name} ({user.id}) 加入群组")
except Exception as e:
logging.error(f"发送欢迎消息失败: {e}")
if __name__ == '__main__':
创建Application实例
application = ApplicationBuilder().token(BOT_TOKEN).build()
注册消息处理器（处理文本消息）
message_handler = MessageHandler(filters.TEXT & ~filters.COMMAND, check_keywords)
application.add_handler(message_handler)
注册聊天成员处理器（处理进群事件）
chat_member_handler = ChatMemberHandler(greet_new_members, ChatMemberHandler.CHAT_MEMBER)
application.add_handler(chat_member_handler)
启动轮询
logging.info("机器人已启动，开始监听消息...")
application.run_polling()
```
部署与运行
代码编写完成后，需要进行最后的部署操作。为了保证机器人24小时稳定运行，建议在Linux服务器上使用 nohup 或 systemd 进行后台托管。
1. 本地测试运行
在终端中进入项目目录，执行以下命令启动机器人：
```bash
python3 bot.py
```
如果看到终端输出 机器人已启动，开始监听消息...，说明启动成功。此时可以在群组中发送包含“兼职”的消息进行测试，机器人应立即删除该消息。
2. 服务器后台运行
测试无误后，将 config.py 和 bot.py 上传至服务器。使用 nohup 命令让其后台运行，即使关闭SSH连接也不会停止：
```bash
nohup python3 bot.py > bot.log 2>&1 &
```
查看运行日志以确认状态：
```bash
tail -f bot.log
```
3. 维护与扩展
如果需要增加新的违禁词，只需编辑 config.py 中的 BANNED_KEYWORDS 列表，保存后重启机器人即可生效。如果需要调整防刷屏的灵敏度，修改 RATE_LIMIT_SECONDS 参数，数值越小越严格。
