微信群防广告自动踢人机器人实现

前言

微信群一旦规模扩大，广告骚扰几乎是必然的麻烦。群主每天要花时间盯着群消息，发现广告后手动踢人，效率极低，稍不注意就会错过，导致广告在群里扩散，破坏群氛围。

对于运营多个群的团队来说，这个问题尤其突出。靠人工值守既消耗精力，又存在时间盲区。本文介绍一种基于关键词匹配与行为模式识别的自动踢人机器人方案，结合 HTTP 接口实现消息监听、广告检测和自动移除成员的完整链路，帮助群管理者从重复劳动中解放出来。

实际运营过程中，广告发送者的手法越来越隐蔽：有人把联系方式拆成多条发送，有人先混入群内潜伏数天再发广告，有人专门在凌晨或周末趁管理员不在线时集中轰炸。这些行为特征本身就是判断依据，单纯的关键词过滤远远不够，需要结合时间窗口、频率和行为序列来综合评估。本文的方案正是从这个角度出发，设计了一套多维度评分引擎，尽可能在降低误踢率的同时提高广告命中率。

一、整体架构设计

自动踢人机器人的核心逻辑不复杂，可以拆成三层：

消息接收层  →  广告检测层  →  执行层
（回调服务）     （规则引擎）   （踢人/禁言/警告）

消息接收层：通过设置回调地址，平台将群消息实时推送到你的服务端，程序解析消息内容、发送者信息和所在群 ID。

广告检测层：对消息文本进行多维度分析，包括关键词命中、正则匹配、URL 识别和频率异常检测。这是整个系统的核心，准确率决定了机器人的可用性。

执行层：检测到广告后，根据配置的策略执行相应动作——警告、静默移除或记录黑名单。

这三层松耦合，可以分别迭代优化，不影响整体稳定性。

二、广告识别规则体系

单纯的关键词黑名单误判率较高，"招聘"可能是广告，也可能是正常交流。一个实用的广告检测引擎需要多维度评分。

2.1 关键词分级

将敏感词按风险等级分级，累积权重超过阈值才触发。

等级	示例关键词	权重
高危	加我威信、私聊我、扫码领取、日入万元	80
中危	兼职、代理、返利、一件代发	40
低危	微信号、联系方式、优惠	20
URL	任何 http/https 链接	50

评分阈值建议设为 100，低于此值仅记录日志，不执行踢人。

2.2 行为模式识别

除了内容，发消息的行为本身也是重要信号：

入群即发：成员加入群 5 分钟内就发送消息，且命中任意低危词，权重翻倍
高频发送：同一成员 5 分钟内发超过 10 条消息
重复内容：同一条文本在 24 小时内重复出现超过 3 次
@全体成员：非管理员发送 @所有人，权重直接 +60

2.3 白名单机制

所有规则都要配合白名单，避免误踢：

管理员列表：群主和所有群管理员豁免所有规则
受信成员：手动标记的老成员，降低权重系数
关键词豁免：某些群本身就是行业讨论群，需要屏蔽特定词检测

三、回调服务搭建

回调服务需要一个公网可访问的 HTTP 服务端，接收平台推送过来的消息。

python# Flask 回调服务示例
# 具体字段以官方文档为准

from flask import Flask, request, jsonify
import json

app = Flask(__name__)

@app.route('/wechat/callback', methods=['POST'])
def wechat_callback():
    data = request.get_json()
    
    # 仅处理群消息（type 以平台文档为准）
    msg_type = data.get('type')
    from_wxid = data.get('fromWxid', '')
    
    # 群消息的 fromWxid 通常包含 @chatroom
    if '@chatroom' not in from_wxid:
        return jsonify({'code': 200})
    
    # 交给检测引擎处理
    handle_group_message(data)
    
    # 必须返回 200，否则平台会重试
    return jsonify({'code': 200})

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=8080)

回调地址必须返回 HTTP 200，否则平台会认为推送失败并重试，导致消息重复处理。建议先回包，再异步处理消息逻辑。

四、核心检测与踢人逻辑

4.1 广告评分引擎

pythonimport re
from dataclasses import dataclass, field
from typing import Dict, List

# 关键词权重表
HIGH_RISK_WORDS = ['加我威信', '私聊我', '扫码领取', '日入万元', '加我微信', '带你赚钱']
MID_RISK_WORDS  = ['兼职', '代理', '返利', '一件代发', '招代理', '佣金']
LOW_RISK_WORDS  = ['微信号', '联系方式', '优惠券', '限时']

URL_PATTERN = re.compile(r'https?://\S+|www\.\S+')

@dataclass
class MemberRecord:
    """记录成员行为"""
    join_time: float = 0
    msg_count_5min: int = 0
    last_window_start: float = 0
    recent_contents: List[str] = field(default_factory=list)

member_records: Dict[str, MemberRecord] = {}

def calc_ad_score(content: str, sender: str, room_id: str,
                  is_new_member: bool, timestamp: float) -> int:
    score = 0

    # 关键词权重
    for w in HIGH_RISK_WORDS:
        if w in content:
            score += 80
    for w in MID_RISK_WORDS:
        if w in content:
            score += 40
    for w in LOW_RISK_WORDS:
        if w in content:
            score += 20

    # URL 检测
    if URL_PATTERN.search(content):
        score += 50

    # 入群即发 bonus
    if is_new_member:
        score = int(score * 2)

    # @所有人
    if '@所有人' in content or '@All' in content:
        score += 60

    return score

4.2 发起踢人请求

pythonimport requests

BASE  = "https://你的接口域名"   # 注册后在官方文档获取
TOKEN = "你的Token"
APPID = "你的appId"
HEADERS = {"token": TOKEN}       # 鉴权字段名以官方文档为准

def kick_member(room_id: str, member_wxid: str):
    """从群中移除成员"""
    url = f"{BASE}/group/removeMember"
    payload = {
        "appId": APPID,
        "chatroomId": room_id,
        "memberWxid": member_wxid
    }
    resp = requests.post(url, json=payload, headers=HEADERS, timeout=10)
    result = resp.json()
    if result.get('ret') == 200:
        print(f"[踢人成功] {member_wxid} 已移出 {room_id}")
        return True
    else:
        print(f"[踢人失败] {result.get('msg')}")
        return False

def send_warning(room_id: str, member_wxid: str, reason: str):
    """踢人前先发警告"""
    url = f"{BASE}/message/postText"
    content = f"@{member_wxid} 检测到您发送了疑似广告内容（{reason}），请遵守群规。"
    payload = {
        "appId": APPID,
        "toWxid": room_id,
        "content": content,
        "ats": member_wxid
    }
    requests.post(url, json=payload, headers=HEADERS, timeout=10)

代码为示例，具体接口路径和字段以官方文档为准。

4.3 完整决策流程

pythonimport time

ADMIN_WXIDS = {'wxid_admin1', 'wxid_admin2'}  # 群管理员豁免列表
SCORE_THRESHOLD = 100  # 触发踢人的分值阈值
WARN_THRESHOLD  = 60   # 触发警告的分值阈值

def handle_group_message(data: dict):
    sender   = data.get('fromMemberWxid', '')  # 字段名以文档为准
    room_id  = data.get('fromWxid', '')
    content  = data.get('content', '')
    ts       = data.get('createTime', time.time())

    # 管理员豁免
    if sender in ADMIN_WXIDS:
        return

    # 判断是否新成员（5 分钟内入群）
    record = member_records.setdefault(sender, MemberRecord(join_time=ts))
    is_new = (ts - record.join_time) < 300

    score = calc_ad_score(content, sender, room_id, is_new, ts)

    if score >= SCORE_THRESHOLD:
        # 直接踢人，记录日志
        kick_member(room_id, sender)
        log_event('kick', sender, room_id, content, score)

    elif score >= WARN_THRESHOLD:
        # 发警告，记录异常
        send_warning(room_id, sender, f"得分{score}")
        log_event('warn', sender, room_id, content, score)

def log_event(action, sender, room_id, content, score):
    print(f"[{action.upper()}] sender={sender} room={room_id} score={score} content={content[:50]}")

五、接入托管 HTTP 接口

自建微信协议风险较高，维护成本大。目前有托管方案可以通过扫码接入个人微信，以 HTTP 接口形式提供消息收发和群管理能力。WechatApi 提供扫码登录、消息收发、好友与群管理等 REST 接口，HTTP 调用即可，不需要维护微信协议层。

接入流程如下：

调用登录接口获取二维码，扫码完成授权
调用 setCallback 接口，将回调地址设置为你的服务地址
服务端收到群消息推送后，运行检测逻辑
检测到广告调用 removeMember 接口完成踢人

pythondef setup_callback(callback_url: str):
    """设置消息回调地址"""
    url = f"{BASE}/callback/setCallback"
    payload = {
        "appId": APPID,
        "callbackUrl": callback_url
    }
    resp = requests.post(url, json=payload, headers=HEADERS, timeout=10)
    result = resp.json()
    print(f"回调设置结果: {result.get('msg')}")
    return result.get('ret') == 200

六、防误踢与稳定性优化

6.1 二次确认机制

对于评分在 100~150 之间的消息，可以设置"观察期"而不是立即踢人：

记录该成员为"疑似广告用户"
若 10 分钟内再次触发评分，则合并处理
两次命中后再执行踢人

这可以有效降低低质量关键词导致的误踢。

6.2 频率控制

踢人接口不宜调用过于密集，建议：

同一群每分钟最多踢 3 人
批量处理时加随机延迟（2~5 秒）
维护一个执行队列，顺序消费

pythonimport queue
import threading
import random
import time

kick_queue = queue.Queue()

def kick_worker():
    while True:
        task = kick_queue.get()
        room_id, member_wxid = task['room_id'], task['member_wxid']
        kick_member(room_id, member_wxid)
        time.sleep(random.uniform(2, 5))  # 随机间隔防触发限频
        kick_queue.task_done()

# 启动后台工作线程
t = threading.Thread(target=kick_worker, daemon=True)
t.start()

6.3 黑名单持久化

被踢成员的 wxid 应持久化存储，防止重新入群后继续发广告：

pythonimport json
import os

BLACKLIST_FILE = 'blacklist.json'

def load_blacklist() -> set:
    if os.path.exists(BLACKLIST_FILE):
        with open(BLACKLIST_FILE) as f:
            return set(json.load(f))
    return set()

def save_blacklist(bl: set):
    with open(BLACKLIST_FILE, 'w') as f:
        json.dump(list(bl), f, ensure_ascii=False)

blacklist = load_blacklist()

def add_to_blacklist(wxid: str):
    blacklist.add(wxid)
    save_blacklist(blacklist)

七、部署与上线注意事项

7.1 服务器要求

公网 IP + 开放回调端口（如 8080）
建议使用 Nginx 反向代理并配置 HTTPS，防止推送被拦截
内存占用不高，4 核 2G 的轻量云服务器足够跑多个群的监控

7.2 日志与告警

建议接入简单的告警通知，当系统触发踢人时发送通知给群主（可通过向群主发送私信实现）：

pythondef notify_admin(admin_wxid: str, room_id: str, kicked_wxid: str, score: int):
    url = f"{BASE}/message/postText"
    content = f"[防广告机器人] 已将 {kicked_wxid} 移出群 {room_id}，广告评分：{score}"
    payload = {
        "appId": APPID,
        "toWxid": admin_wxid,
        "content": content
    }
    requests.post(url, json=payload, headers=HEADERS, timeout=10)

7.3 分群配置

不同群可能有不同的容忍度，建议支持分群配置：

配置项	默认值	说明
score_threshold	100	踢人评分阈值
warn_threshold	60	警告评分阈值
new_member_minutes	5	入群多久算新成员
enable_url_check	true	是否检测链接
whitelist_wxids	[]	豁免成员列表

用 JSON 文件维护每个群的配置，启动时加载，支持热更新。

八、实操细节与注意事项

在实际落地过程中，有几个容易忽略的细节值得重点关注。

关键词库的持续维护：广告话术更新很快，上个月常见的"扫码领红包"这个月可能已经换成了"加群领福利"。建议每周抽查一次实际被拦截和漏过的消息，更新关键词库。关键词不必追求全覆盖，高危词命中一条就能触发足够权重，宁可少而精准，也不要堆砌大量低权重词拉高误判率。

时区与时间窗口的对齐：行为分析依赖时间戳，服务器时区必须与微信平台一致，否则"入群 5 分钟内发消息"这类判断会出现偏差。建议统一用 UTC 时间戳做内部计算，展示层再转换为本地时间。

内存中的成员记录清理：member_records 字典会随着成员增多而无限增长，生产环境中需要定期清理超过 24 小时未活跃的记录，或改用 TTL 缓存（如 cachetools.TTLCache）自动过期。

踢人权限的前提确认：机器人所使用的账号必须在目标群中具有管理员权限，否则踢人接口会一直返回失败。建议在服务启动时调用群信息接口，自动校验账号角色，不满足条件的群直接跳过处理，避免产生大量无效请求。

多账号场景下的去重：如果同一个群挂了多个监控账号（主号备号），同一条广告消息可能被两个账号分别检测到，触发两次踢人请求。可以引入 Redis 或本地锁，用"群 ID + 消息 ID"做去重键，保证每条消息只执行一次动作。

误踢的补救流程：自动化系统总会有误判。建议在群公告中说明机器人的存在，并提供申诉联系方式；同时保留完整的踢人日志（时间、群 ID、成员 wxid、触发内容、评分明细），方便事后核查和重新邀请。

灰度上线策略：新配置上线时不要直接全量启用踢人功能，可以先以"只记录不踢人"的模式运行 48 小时，观察日志中触发踢人的消息质量，确认误判率在可接受范围内再切换到执行模式。这个习惯能避免大规模误踢带来的群氛围损伤。

九、常见问题排查

问：回调收不到群消息？ 检查回调地址是否公网可达，curl 测试接口返回是否 200；确认微信账号保持在线；回调地址设置后需要账号重新在线才能生效。

问：踢人接口返回失败？ 确认账号在群内是管理员或群主权限；检查 chatroomId 是否正确（应含 @chatroom 后缀）；确认目标成员 wxid 准确。

问：误踢了正常成员怎么处理？ 建议保留踢人日志，可以手动重新邀请成员回群；同时降低评分阈值或将相关关键词移入豁免列表。

问：广告漏网怎么办？ 图片广告和语音广告无法通过关键词检测，需要结合 OCR 或 ASR 服务扩展检测维度；也可以对"入群即发图片"的行为单独加权。

总结

微信群防广告机器人的核心在于规则引擎的设计精度——过于宽松则广告漏网，过于严格则误踢合法用户。本文介绍的多维度评分体系、二次确认机制和分群配置方案，已经能覆盖大多数实际场景。规则库需要根据实际群情况持续调整，才能维持良好的识别率。实操层面要注意关键词库的定期维护、成员记录的内存管理、踢人权限的前置校验以及灰度上线策略，这些细节直接决定系统在生产环境中的稳定性和可靠性。