在Telegram社群高速发展的今天，广告机器人和垃圾信息几乎无处不在。它们混入群组、刷屏骚扰、诱导点击，严重拉低社群质量。作为群组管理员，手动删除消息既低效又不可持续。幸运的是，借助Telegram Bot API，我们可以开发出能够自动识别并过滤广告和垃圾消息的机器人。本文将系统梳理六种可落地的过滤策略，从简单到进阶，帮助开发者快速搭建一套适合自己场景的反垃圾体系。
一、垃圾消息的常见特征与识别思路
在设计过滤策略之前，我们需要先理解垃圾消息的共性。通过大量观察，广告和垃圾消息通常具备以下特征之一：
- 包含高频率关键词：如“加群”、“返利”、“代购”、“私聊”等，或品牌/币种名。
- 附带可疑链接：短链、非HTTPS链接、未知域名。
- 发送频率极高：同一用户短时间内发送多条消息。
- 内容重复度高：同一消息文本在群内反复出现。
- 账号“年轻”：注册时间短、无头像、用户名随机。
没有一种方法是万能的，最佳实践是组合多种策略，并把它们做成可配置的规则链。下面我们就从最基础的关键词过滤开始。
二、基于关键词的黑名单过滤
关键词过滤是最简单直接的方法。通过监听消息文本，与预设的黑名单列表做匹配，命中即删除。为了避免误杀，建议使用正则表达式并支持自定义白名单。
import re
from telegram import Update
from telegram.ext import Application, MessageHandler, filters
BLACKLIST_PATTERNS = [
r'加\s*群',
r'返利',
r'代购',
r'https?://(?!t\.me/)[^\s]+', # 允许t.me链接
]
def is_spam(text):
if not text:
return False
for pattern in BLACKLIST_PATTERNS:
if re.search(pattern, text, re.IGNORECASE):
return True
return False
async def handle_message(update: Update, context):
msg = update.message
if msg and is_spam(msg.text):
await msg.delete()
app = Application.builder().token("YOUR_BOT_TOKEN").build()
app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_message))
app.run_polling()注意：不要直接使用过于宽泛的规则（如“包含http就删”），否则会误删正常聊天中的链接。建议结合短链展开、域名黑名单等方式提高准确率。
三、利用频率限制与行为分析检测垃圾
垃圾桶通常会在短时间内高频发送消息。我们可以基于用户ID统计单位时间内的消息数，超过阈值则自动禁言或删除消息。
from collections import defaultdict, deque
import time
rate_map = defaultdict(deque) # user_id -> [(timestamp, msg_id)]
MAX_PER_MINUTE = 5
async def handle_rate(update, context):
user_id = update.effective_user.id
now = time.time()
# 清理超过60秒的旧记录
while rate_map[user_id] and now - rate_map[user_id][0][0] > 60:
rate_map[user_id].popleft()
rate_map[user_id].append((now, update.message.message_id))
if len(rate_map[user_id]) > MAX_PER_MINUTE:
await update.message.delete()
# 可选：禁言1分钟
await context.bot.restrict_chat_member(update.effective_chat.id, user_id, permissions=None, until_date=now+60)除了频率，行为分析还能包括：重复内容的检测（用哈希或相似度）、发送时间特征（比如凌晨广告较多）、@多人的动作（机器人很喜欢@all）等。这些信号可以进一步提高识别率。
四、基于用户信誉与入群时长的过滤策略
新入群的用户往往是垃圾信息的重灾区。我们可以要求用户先完成验证（如输入验证码、点击按钮）才能发言，或限制新用户在一定时间内只能发送有限的内容。
from telegram.ext import ChatMemberHandler
verification_pending = set() # 存储待验证的user_id
async def on_new_member(update, context):
for new_member in update.chat_member.new_chat_member:
user_id = new_member.user.id
verification_pending.add(user_id)
# 发送验证码或要求点击按钮，这里简化处理
await context.bot.send_message(user_id, "请发送“验证”完成验证，否则无法发言。")
async def check_verified(update, context):
user_id = update.effective_user.id
if user_id in verification_pending:
if update.message.text == "验证":
verification_pending.remove(user_id)
await update.message.reply_text("验证通过，欢迎加入！")
else:
await update.message.delete()
await update.message.reply_text("请先完成验证。")此外，还可以建立用户信誉数据库：记录用户被删除消息的次数、被举报次数、被封禁次数等。信誉低的用户自动进入审查队列。当然，这需要一定的存储，可以使用SQLite或Redis。
五、集成第三方反垃圾服务
如果不想从零实现，可以使用现成的反垃圾API。例如 SpamProtection 机器人提供的接口，或者 Cloudflare 等安全服务。通过调用这些服务，我们可以获得消息是否为垃圾的评分，再根据阈值决定是否删除。
import requests
def check_spam(text):
resp = requests.post("https://your-spam-service.com/check", json={"text": text}, timeout=2)
if resp.status_code == 200:
return resp.json().get("is_spam", False)
return False使用第三方服务可以减少开发量，但需要注意隐私：消息内容会发送给第三方，对于敏感群组要慎重。同时也可能产生额外费用，而且服务本身需要维护。
六、机器学习与自适应过滤（高级）
对于广告形式不断变化的场景，规则系统容易漏判。我们可以使用机器学习模型来分类消息。常用方法包括：
- 训练一个朴素贝叶斯或逻辑回归模型，基于词频向量。
- 使用TF-IDF将文本转换为特征。
- 定期从管理员标记的垃圾/正常消息中重新训练模型。
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.naive_bayes import MultinomialNB
# 假设已经有训练数据
X = ["免费领取比特币", "今晚聊聊天吧"]
y = [1, 0] # 1垃圾, 0正常
vectorizer = TfidfVectorizer()
X_vec = vectorizer.fit_transform(X)
model = MultinomialNB().fit(X_vec, y)
def is_spam_ml(text):
vec = vectorizer.transform([text])
prob = model.predict_proba(vec)[0][1]
return prob > 0.8为了提升效果，可以融合手工规则和模型结果，比如规则判定为垃圾但模型判定为正常的消息，交给管理员处理。机器学习的优点是可以不断更新，适应新垃圾；缺点是初期需要标注数据，而且CPU开销较大。对于大型群组，可以考虑将该功能独立成一个微服务。
七、实践建议与避坑指南
把多种策略组合起来时，务必注意以下几点：
- 避免误杀：删除操作是不可逆的，建议先“禁言+警告”而非直接删除，或者将垃圾消息转发到“争议区”供管理员复核。
- 记录日志：所有过滤动作都应记录，包括触发规则、消息内容（脱敏）、时间，以便审计和调整规则。
- 设置管理员白名单：管理员和信任用户不受限制，避免机器人屏蔽自己人。
- 使用群组权限：设置新用户无法发送多媒体、链接等，能有效降低垃圾信息量。
- 用户反馈机制：允许用户举报消息，举报次数达到阈值自动触发处理。
- 性能优化：高频群组中，建议使用异步处理和消息队列，避免阻塞更新处理。
总结
Telegram机器人过滤广告和垃圾消息是一项系统性工程，没有一劳永逸的方案。开发者应根据群组的实际规模、活跃度和垃圾形态，灵活组合关键词、频率、信誉、第三方服务以及机器学习等策略。关键原则是“宁可漏杀，不可误杀”，始终保障正常用户的聊天体验。希望本文的思路和代码能帮助你构建一个高效、稳健的反垃圾机器人，为你的社群营造一个清净的交流环境。