微信群素材轮播与定时推送机器人

前言

很多运营团队每天面临同一个问题：几十个微信群需要定时发早报、活动海报、产品链接，靠人工一个个复制粘贴既费时又容易漏发。素材更新后还要手动替换，稍有疏忽就出现发错版本的事故。本文聚焦"微信群素材轮播 + 定时推送"这一场景，从调度设计、素材管理到实际代码落地，手把手讲清楚怎么用一套机器人方案彻底解放双手。

一、为什么个人微信群消息推送需要机器人

企业微信覆盖不到的场景

企业微信有官方群发接口，但绝大多数 C 端用户、私域流量沉淀的微信群仍在个人微信生态里。企业无法强制客户加企业微信，因此运营侧对个人微信群批量推送的需求始终存在。

人工运营的三大痛点

合规前提

微信官方明确禁止通过逆向客户端协议进行批量操作。使用第三方 API 前务必评估业务合规性，建议仅用于自有私域群、已授权的运营场景，不得用于骚扰用户或违规营销。

二、整体架构设计

一套完整的微信群素材轮播机器人由四个模块组成：

定时调度器（Scheduler）
      │
      ▼
素材管理器（Material Manager）── 素材库（图片/文字/链接）
      │
      ▼
发送执行器（Sender）── WechatApi HTTP 接口
      │
      ▼
回执记录器（Logger）── 发送日志 / 失败重试

定时调度器 负责按 cron 表达式触发任务，精确到分钟级。素材管理器 维护一个轮播队列，每次触发时自动取下一条素材，到头后回到队首循环。发送执行器 调用 个人微信 API 完成实际消息投递。回执记录器 将每次发送结果落库，便于监控告警。

这套架构与所用语言无关，Python、Node.js、Go 均可实现。以下以 Python 为例展开。

三、素材库与轮播队列设计

素材类型

微信群常见素材分三类：

• 纯文字：早报、每日金句、活动通知
• 图片（含图文混合）：海报、产品图、活动 banner
• 小程序卡片 / 链接卡片：落地页引流

轮播设计上建议用 JSON 文件或轻量数据库（SQLite）维护素材队列，每条素材包含以下字段：

json{
  "id": 1,
  "type": "text",
  "content": "【每日早报】2025-01-15\n📌 今日重点：...",
  "send_time": "08:00",
  "groups": ["group_wxid_001", "group_wxid_002"],
  "enabled": true,
  "cycle": "daily"
}

字段说明：

• type：text / image / link，决定调用哪个发送接口
• send_time：当天触发时刻，配合调度器使用
• groups：目标群的 wxid 列表，可精细控制每条素材的发送范围
• cycle：daily（每天）、weekly_mon（每周一）、custom（自定义间隔）

轮播指针管理

对于同一个群，有时需要按"第一天发 A，第二天发 B，第三天发 C，再回到 A"的方式循环。实现上用一张指针表记录每个群当前轮播到的素材索引：

python# carousel_state.json 示意
{
  "group_wxid_001": {"playlist": [1, 2, 3], "current": 0},
  "group_wxid_002": {"playlist": [4, 5],    "current": 1}
}

每次发送后将 current 加一取模，写回文件。这样即使服务重启，状态也不会丢失。

四、调用 WechatApi 实现定时发送

WechatApi 基于 微信 iPad 协议 实现，通过 HTTP API 操控个人微信账号发消息，无需额外安装客户端，云端直接调用，天然适合定时任务场景。

鉴权与基础参数

所有请求使用 HTTP POST + JSON Body，鉴权通过请求头 VideosApi-token 传入 API 密钥，appId 字段标识当前登录设备（同一账号在不同设备登录会产生不同 appId，控制台可查）。

pythonimport requests
import json

API_BASE = "https://api.wechatapi.net"   # 示意域名，请以控制台实际地址为准
TOKEN    = "your_videos_api_token_here"   # 替换为控制台生成的 Token
APP_ID   = "your_app_id_here"             # 替换为设备 appId

HEADERS = {
    "Content-Type": "application/json",
    "VideosApi-token": TOKEN
}

def send_text_to_group(group_wxid: str, content: str) -> dict:
    """向指定微信群发送文字消息"""
    payload = {
        "appId": APP_ID,
        "toWxid": group_wxid,
        "content": content
    }
    resp = requests.post(
        f"{API_BASE}/group/sendText",   # 示意路径
        headers=HEADERS,
        json=payload,
        timeout=15
    )
    return resp.json()


def send_image_to_group(group_wxid: str, image_url: str) -> dict:
    """向指定微信群发送图片（URL 方式）"""
    payload = {
        "appId": APP_ID,
        "toWxid": group_wxid,
        "imageUrl": image_url
    }
    resp = requests.post(
        f"{API_BASE}/group/sendImage",  # 示意路径
        headers=HEADERS,
        json=payload,
        timeout=30
    )
    return resp.json()

返回体结构

WechatApi 所有接口返回统一结构：

json{
  "ret": 200,
  "msg": "发送成功",
  "data": {
    "msgId": "wx_msg_20250115_001",
    "toWxid": "group_wxid_001",
    "createTime": 1736908800
  }
}

ret 为 200 表示成功，非 200 时 msg 字段会给出具体错误原因（如设备离线、群不存在、频率超限等）。建议在代码里统一判断 ret 而不要仅靠 HTTP 状态码。

五、定时调度器完整实现

推荐用 Python 的 APScheduler 库做调度，支持 cron 表达式，且可在进程内运行，无需额外部署 Celery / Redis。

pythonfrom apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers.cron import CronTrigger
import json
import logging
from datetime import datetime

logging.basicConfig(level=logging.INFO,
                    format="%(asctime)s [%(levelname)s] %(message)s")

scheduler = BlockingScheduler(timezone="Asia/Shanghai")


def load_materials() -> list:
    with open("materials.json", "r", encoding="utf-8") as f:
        return json.load(f)


def load_state() -> dict:
    try:
        with open("carousel_state.json", "r", encoding="utf-8") as f:
            return json.load(f)
    except FileNotFoundError:
        return {}


def save_state(state: dict):
    with open("carousel_state.json", "w", encoding="utf-8") as f:
        json.dump(state, f, ensure_ascii=False, indent=2)


def run_push_job(send_time_slot: str):
    """每分钟检查一次，匹配当前时间段需要发送的素材"""
    materials = load_materials()
    state = load_state()
    now_slot = datetime.now().strftime("%H:%M")

    for mat in materials:
        if not mat.get("enabled"):
            continue
        if mat.get("send_time") != now_slot:
            continue

        for group_wxid in mat.get("groups", []):
            # 轮播逻辑：取当前指针对应素材
            key = f"{group_wxid}_{mat['id']}"
            playlist = mat.get("playlist", [mat["id"]])
            idx = state.get(key, 0) % len(playlist)
            current_material_id = playlist[idx]

            # 实际发送（根据 type 调不同接口）
            if mat["type"] == "text":
                result = send_text_to_group(group_wxid, mat["content"])
            elif mat["type"] == "image":
                result = send_image_to_group(group_wxid, mat["imageUrl"])
            else:
                logging.warning(f"未知素材类型: {mat['type']}")
                continue

            if result.get("ret") == 200:
                logging.info(f"✓ 群 {group_wxid} 发送成功，素材 ID={current_material_id}")
                state[key] = (idx + 1) % len(playlist)
            else:
                logging.error(f"✗ 群 {group_wxid} 发送失败: {result.get('msg')}")

    save_state(state)


# 每分钟触发一次检查
scheduler.add_job(
    run_push_job,
    CronTrigger(minute="*", timezone="Asia/Shanghai"),
    args=[""],
    id="wechat_push_job"
)

if __name__ == "__main__":
    logging.info("微信群定时推送机器人启动...")
    scheduler.start()

启动只需一行命令：

bash# 建议用 screen 或 systemd 守护运行
screen -S wechat_bot
python wechat_push_bot.py

六、素材轮播的进阶玩法

按群分层投放

不同群的受众画像不同。VIP 付费群应该收到更高频、更深度的内容，而公域引流群只需每天一条引流卡片。建议在素材配置里给每个群单独维护一份播放列表，而不是所有群共用同一队列。

利用 微信群管理机器人 的群成员查询接口，还可以拉取每个群的成员数和标签，实现基于群属性的动态分组投放，比如："成员 > 200 人的群发海报，成员 ≤ 50 人的群发文字"。

A/B 测试文案

素材轮播天然支持 A/B 测试：把两条不同话术的文字素材加入同一播放列表，奇数次发 A 版，偶数次发 B 版，观察转化率差异。这需要在发送日志里记录发送版本，后期用数据比对效果。

失败重试与告警

网络抖动或设备临时离线都可能导致发送失败。建议维护一个失败队列，每隔 5 分钟重试一次，连续失败 3 次后通过邮件或钉钉告警。

pythonimport time

MAX_RETRY = 3

def send_with_retry(group_wxid, content, retry=0):
    result = send_text_to_group(group_wxid, content)
    if result.get("ret") != 200 and retry < MAX_RETRY:
        time.sleep(5)
        return send_with_retry(group_wxid, content, retry + 1)
    return result

富文本与图文混排

纯文字消息容易被忽略。WechatApi 支持发送图文消息（先发图片再发文字）来模拟图文混排效果。在素材配置里把图片和文字组合为一个"套餐"，发送时先投图、200ms 后投文，视觉上连贯自然：

pythonimport time

def send_image_text_combo(group_wxid, image_url, text):
    send_image_to_group(group_wxid, image_url)
    time.sleep(0.2)
    send_text_to_group(group_wxid, text)

七、部署与运维注意事项

设备稳定性

基于 微信二次开发 的 iPad 协议方案，账号稳定性依赖设备保持在线。建议：

1. 使用专号运营，不与日常沟通的主账号混用。
2. 控制单日单群发送频次，避免触发平台风控（通常建议同一群每天不超过 5 条）。
3. 不同群之间的发送间隔建议 3~10 秒，模拟人工操作节奏。

发送频率限制

日志与监控

建议将发送日志写入文件并配合 logrotate 定期归档。生产环境可接入 Prometheus + Grafana，监控每日发送成功率、失败率和延迟分布。

素材版本管理

使用 Git 管理 materials.json，每次更新素材时提交一个 commit，保留完整变更历史。配合 CI/CD 可以实现：运营同学在 Web 后台编辑素材 → 自动生成 JSON → 推送到服务器 → 机器人热加载，无需重启进程。

小结

微信群素材轮播与定时推送机器人的核心是三件事：结构化的素材管理、可靠的定时调度、稳定的消息发送通道。WechatApi 基于 iPad 协议提供了完善的个人微信 HTTP 接口，鉴权简单（VideosApi-token + appId），返回格式统一（ret/msg/data），天然契合自动化场景。配合 APScheduler 的 cron 调度和 JSON 轮播状态机，百行 Python 即可搭建一套生产可用的推送系统。

如果你的业务同时涉及客服对话接管，可以进一步整合 微信客服机器人 能力，让同一账号既能定时推送，又能智能回复用户消息，实现真正的"全自动私域运营"闭环。更多接口文档和 Demo 可访问 WechatApi 官网 查阅。