微信群红包提醒与统计机器人

前言

微信群红包是企业促活、社群运营的常见手段，但手动盯群、逐条统计费时费力——尤其是多群并发时，往往等你反应过来，红包已过期。本文介绍如何基于 WechatApi 微信群管理机器人 构建一套自动化红包提醒与统计系统：实时监听群消息、识别红包事件、推送提醒、汇总领取数据，全程无需人工值守。

红包提醒机器人的核心需求拆解

在动手写代码之前，先把业务诉求翻译成技术需求，否则很容易写出一个"能跑但没用"的东西。

常见的红包相关场景可以分为三类：

除此之外还有一个隐性需求：多群隔离。企业往往同时维护十几个乃至几十个微信群，统计数据必须按群区分，不能混在一起。

实现上述功能的前提是能够稳定接收微信群消息，并且能够主动发送消息和拉取群成员列表。这正是选用 WechatApi 个人微信 API 的核心原因：它基于 iPad 协议实现，稳定性远优于 Hook/注入方案，且提供标准 HTTP 接口，任何语言都能接入。

消息接收原理：Webhook 回调机制

WechatApi 采用 Webhook 推送 模式，即微信端收到消息后，平台主动向你配置的回调地址发送 HTTP POST 请求，消息体为 JSON 格式。你的服务只需监听这个地址，解析消息类型即可。

关键消息类型字段（MsgType）与红包相关的主要有：

• 49（XML 消息）：红包、转账、小程序卡片等富媒体消息均通过此类型下发，需解析内嵌 XML 的 type 子字段。红包对应 type=2000。
• 10000（系统通知）：领取红包的系统提示（"XXX 领取了你的红包"）通过此类型推送。

一个典型的红包消息回调数据结构示例：

json{
  "ret": 200,
  "msg": "ok",
  "data": {
    "MsgType": 49,
    "FromUserName": "@@abcdef1234567890",
    "ToUserName": "wxid_xxxxxxxxxxxxxxx",
    "Content": "<msg><appmsg><title>微信红包</title><type>2000</type>...</appmsg></msg>",
    "CreateTime": 1718256000,
    "MsgId": "9988776655443322110",
    "RoomTopic": "技术交流群",
    "SenderWxId": "wxid_sender001"
  }
}

字段说明：

• FromUserName 以 @@ 开头表示群消息。
• SenderWxId 是发红包的群成员 wxid。
• Content 中的 XML 需要自行解析，<type>2000</type> 是红包的识别标志。

消息处理流程：从接收到提醒

整个流程可以拆分为五个步骤：

第一步：搭建 Webhook 接收服务

用 Python Flask 快速搭起一个接收端点：

pythonfrom flask import Flask, request, jsonify
import xml.etree.ElementTree as ET
import requests, threading, time, json

app = Flask(__name__)

# 存储红包记录：{MsgId: {group, sender, send_time, receivers: []}}
redpacket_store = {}

WECHATAPI_HOST = "https://api.wechatapi.net"   # 示意，非真实地址
VIDEOS_API_TOKEN = "your_token_here"            # 替换为控制台申请的 token
APP_ID = "your_device_appid"                    # 替换为设备 appId

def send_group_message(room_id, content):
    """调用 WechatApi 向群发送文本消息"""
    url = f"{WECHATAPI_HOST}/message/sendText"
    headers = {"VideosApi-token": VIDEOS_API_TOKEN, "Content-Type": "application/json"}
    payload = {"appId": APP_ID, "toWxId": room_id, "content": content}
    resp = requests.post(url, headers=headers, json=payload, timeout=10)
    return resp.json()

def parse_redpacket(xml_content):
    """从 XML Content 中判断是否红包消息"""
    try:
        root = ET.fromstring(xml_content)
        msg_type = root.find(".//type")
        if msg_type is not None and msg_type.text == "2000":
            title = root.findtext(".//title", default="微信红包")
            return True, title
    except Exception:
        pass
    return False, None

@app.route("/wechat/callback", methods=["POST"])
def callback():
    data = request.json or {}
    event = data.get("data", {})
    msg_type = event.get("MsgType")
    from_user = event.get("FromUserName", "")

    # 仅处理群消息
    if not from_user.startswith("@@"):
        return jsonify({"code": 0})

    group_id = from_user
    msg_id = event.get("MsgId", "")

    if msg_type == 49:
        is_rp, title = parse_redpacket(event.get("Content", ""))
        if is_rp:
            # 记录红包
            redpacket_store[msg_id] = {
                "group": group_id,
                "sender": event.get("SenderWxId"),
                "send_time": event.get("CreateTime"),
                "receivers": [],
                "room_topic": event.get("RoomTopic", "未知群")
            }
            # 立即提醒
            tip = f"【红包提醒】群里有红包，手慢无！\n来自：{event.get('SenderWxId')}"
            send_group_message(group_id, tip)
            # 启动定时过期提醒
            threading.Thread(
                target=expire_reminder, args=(msg_id, group_id), daemon=True
            ).start()

    elif msg_type == 10000:
        # 领取通知，格式："wxid_xxx 领取了红包"，简单解析
        notice = event.get("Content", "")
        for mid, info in redpacket_store.items():
            if info["group"] == group_id:
                info["receivers"].append(notice)

    return jsonify({"code": 0})

def expire_reminder(msg_id, group_id, delay=1200):
    """20 分钟后检查红包是否仍有意义，再次提醒"""
    time.sleep(delay)
    info = redpacket_store.get(msg_id)
    if info and len(info.get("receivers", [])) == 0:
        send_group_message(group_id, "【红包提醒】红包还没人领，快来抢！")

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

代码中的 WECHATAPI_HOST、VIDEOS_API_TOKEN、APP_ID 均为占位符，实际值在 WechatApi 控制台 注册设备后获取。鉴权请求头统一为 VideosApi-token，这是 WechatApi 所有接口的标准鉴权方式。

第二步：调用消息发送接口推送提醒

发消息的接口调用范式非常简洁，以 cURL 为例：

bashcurl -X POST "https://api.wechatapi.net/message/sendText" \
  -H "VideosApi-token: your_token_here" \
  -H "Content-Type: application/json" \
  -d '{
    "appId": "your_device_appid",
    "toWxId": "@@群ID",
    "content": "【红包提醒】群里来红包啦，手慢无！"
  }'

正常返回：

json{
  "ret": 200,
  "msg": "发送成功",
  "data": {
    "msgId": "11223344556677889900"
  }
}

ret=200 表示成功，data.msgId 是消息的唯一 ID，可用于后续对账。若 ret 非 200，需检查 token 是否有效、设备是否在线。

第三步：统计领取情况

领取统计依赖两个数据来源：

1. 系统通知消息（MsgType=10000）：每当有群成员领取红包，微信会在群内推送一条系统消息，内容大致为"xxx 领取了 xxx 的红包"。通过解析这条消息，可以记录领取人。

1. 群成员列表接口：调用 WechatApi 的获取群成员接口，拿到群内所有成员 wxid，与已领取列表做差集，即可得到"未领取名单"。

获取群成员列表的调用示例：

pythondef get_group_members(group_id):
    url = f"{WECHATAPI_HOST}/group/memberList"
    headers = {"VideosApi-token": VIDEOS_API_TOKEN}
    payload = {"appId": APP_ID, "groupId": group_id}
    resp = requests.post(url, headers=headers, json=payload, timeout=10)
    result = resp.json()
    if result.get("ret") == 200:
        return result["data"].get("memberList", [])
    return []

def generate_report(msg_id):
    """生成红包领取报告"""
    info = redpacket_store.get(msg_id)
    if not info:
        return "未找到该红包记录"
    total_members = get_group_members(info["group"])
    received_count = len(info["receivers"])
    total_count = len(total_members)
    report = (
        f"【红包统计报告】\n"
        f"群名：{info['room_topic']}\n"
        f"发红包人：{info['sender']}\n"
        f"群成员总数：{total_count}\n"
        f"已领取：{received_count} 人\n"
        f"未领取：{total_count - received_count} 人\n"
    )
    return report

多群并发管理：不同群独立统计

实际运营场景中，一个微信号往往同时在几十个社群中。关键点在于 FromUserName（即 @@群ID）是全局唯一的，因此以它作为 key 分群存储数据，天然就做到了多群隔离。

如果需要更复杂的多群策略（比如不同群有不同提醒话术、不同统计周期），可以维护一个群配置表：

这类配置建议存入数据库（SQLite/MySQL 均可），启动时加载到内存，收到消息后按群 ID 匹配配置项，选对应的话术和时间窗口。

WechatApi 底层基于 iPad 协议 实现，与微信服务器交互的方式和真实 iPad 客户端完全一致，因此在消息接收的完整性上有保障——不会像某些 Web 协议方案那样漏掉消息或频繁掉线，这对多群并发场景尤为重要。

定时汇报与日报生成

除了实时提醒，运营团队通常还需要每日汇总报告。实现方式有两种：

方式一：定时脚本

用 APScheduler 或系统 crontab，每天固定时间调用 generate_report 函数，将报告发到指定微信群或企业通知群。

方式二：关键词触发

在 Webhook 回调中增加关键词检测逻辑：当群内有人发送"今日红包统计"时，机器人自动回复当日汇总报告。这种方式更灵活，运营人员随时可以查询。

pythonif msg_type == 1:  # 普通文本消息
    text = event.get("Content", "").strip()
    if text == "今日红包统计":
        report_lines = []
        for mid, info in redpacket_store.items():
            if info["group"] == group_id:
                report_lines.append(generate_report(mid))
        reply = "\n---\n".join(report_lines) if report_lines else "今日暂无红包记录"
        send_group_message(group_id, reply)

常见问题与注意事项

1. 消息去重

微信在网络抖动时可能重复推送同一条消息，务必用 MsgId 做去重，避免同一个红包被提醒多次。

2. XML 解析容错

红包的 Content 字段有时会包含 CDATA 或特殊字符，建议使用 lxml 库替代标准库 xml.etree.ElementTree，容错性更强。

3. 设备在线检测

若微信设备掉线，Webhook 推送会中断，提醒将失效。建议每隔 5 分钟调用一次设备心跳检测接口，发现掉线及时告警（邮件/钉钉/飞书均可）。

4. 红包金额隐私

微信红包的具体金额在系统通知消息中通常不携带，若需要金额统计，需结合红包发出时的 XML 内容（部分红包类型会携带金额字段）或借助其他业务系统数据。不要依赖抓包或私有协议字段，合规为先。

5. 频率限制

群内提醒消息不宜过于频繁，否则容易被群成员屏蔽。建议同一群内同一红包最多提醒 2 次（发出时 + 过期前），且两次间隔不低于 15 分钟。

更多接口细节可参考 WechatApi 开发文档 和 微信机器人开发指南，文档中有完整的参数列表和错误码说明。

小结

本文系统介绍了基于 WechatApi 构建微信群红包提醒与统计机器人的完整方案：从 Webhook 回调接收消息、解析红包事件，到实时提醒、过期二次提醒、多群并发隔离，再到定时日报和关键词触发统计。核心调用范式为 HTTP POST + JSON，鉴权使用 VideosApi-token 请求头，业务参数中 appId 标识设备，返回体格式统一为 {"ret":200,"msg":"...","data":{...}}，接入成本极低。

如果你的团队正在做社群运营，或者需要更复杂的群管理自动化（签到、关键词回复、成员欢迎等），WechatApi 提供的 微信二次开发 能力可以作为统一底座，在同一套接口体系下扩展所有功能，避免维护多套方案的成本。注册试用地址：https://newmanager.wechatapi.net/dashboard/。