微信消息定时提醒待办机器人

前言

你是否遇到过这样的场景：团队每天早上需要同步任务清单，却要靠人工去群里逐条发消息；客户的跟进节点忘了提醒，导致商机白白错过；或者重要事项明明设了日历，却因为没打开电脑而错过。把提醒嵌入微信——这是最自然的解法。本文介绍如何基于 WechatApi 个人微信 HTTP API 构建一套可生产使用的微信消息定时提醒待办机器人，覆盖架构设计、核心调用代码、参数说明和落地注意事项。

为什么选微信做提醒载体

企业内部已经有钉钉机器人、飞书 Webhook，为什么还要专门折腾微信消息提醒？原因很简单：受众在哪里，提醒就该在哪里。

对于大多数国内用户来说，微信是使用频次最高的 App，平均每天打开次数远超其他通讯软件。如果把待办提醒放在微信里，信息被看到的概率要比邮件、短信高出不止一个量级。更关键的是，很多业务场景面向的是外部客户或合作伙伴，他们未必装了钉钉或飞书，但几乎 100% 在用微信。

个人微信相比企业微信的优势在于门槛低——不需要对方认证企业资质，也不需要对方主动关注公众号。一个普通微信账号，授权给 API 服务即可直接发消息给好友或群。WechatApi 正是基于 iPad 协议实现了这一能力，稳定运行于私有服务器或云端，支持文本、图片、文件、小程序卡片等多种消息类型，非常适合做定时提醒的消息投递底层。

整体架构设计

一个完整的定时提醒待办机器人由三个模块组成：

模块	职责	技术选型示例
任务存储	记录提醒内容、触发时间、接收人	MySQL / SQLite / Redis Sorted Set
调度引擎	定时扫描、触发到期任务	APScheduler / Celery Beat / crontab
消息投递	调用微信 API 发送消息	WechatApi HTTP 接口

三者解耦的好处是每个部分都可以单独替换。比如调度引擎从简单的 crontab 升级为 Celery，不影响存储层和消息层的代码；消息类型从文本扩展到图文卡片，只需改投递模块。

数据流如下：

用户通过管理界面或另一个微信指令机器人（接收用户"提醒我明天 9 点开会"这类指令）创建待办记录，写入任务表。
调度引擎每分钟扫描任务表，找出 remind_at <= NOW() 且状态为"待发送"的记录。
对每条记录，构造消息体，调用 WechatApi 的发送接口，将结果写回任务表（成功/失败）。
失败的任务按指数退避策略重试，超过最大重试次数后标记为"失败"并告警。

任务表结构与调度逻辑

以 SQLite 为例，任务表建议包含以下字段：

sqlCREATE TABLE remind_tasks (
    id          INTEGER PRIMARY KEY AUTOINCREMENT,
    wxid        TEXT NOT NULL,        -- 接收人微信 ID（或群 ID）
    content     TEXT NOT NULL,        -- 提醒文本内容
    remind_at   DATETIME NOT NULL,    -- 预定发送时间（UTC）
    repeat_rule TEXT DEFAULT NULL,    -- cron 表达式，NULL 表示一次性
    status      TEXT DEFAULT 'pending',  -- pending / sent / failed
    retry_count INTEGER DEFAULT 0,
    created_at  DATETIME DEFAULT CURRENT_TIMESTAMP
);
CREATE INDEX idx_remind_at ON remind_tasks(remind_at, status);

调度引擎使用 Python APScheduler，每 60 秒触发一次扫描函数：

pythonfrom apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime, timezone
import sqlite3, requests

API_BASE   = "https://your-wechatapi-endpoint"   # 替换为实际接入点
API_TOKEN  = "YOUR_VIDEOS_API_TOKEN"              # 替换为控制台 token
APP_ID     = "YOUR_DEVICE_APP_ID"                 # 替换为设备 appId

def send_wechat_text(wxid: str, content: str) -> bool:
    """调用 WechatApi 发送文本消息，返回是否成功。"""
    payload = {
        "appId": APP_ID,
        "toWxId": wxid,
        "content": content,
    }
    headers = {
        "VideosApi-token": API_TOKEN,
        "Content-Type": "application/json",
    }
    try:
        resp = requests.post(
            f"{API_BASE}/api/sendTextMsg",
            json=payload,
            headers=headers,
            timeout=10,
        )
        data = resp.json()
        return data.get("ret") == 200
    except Exception as e:
        print(f"[ERROR] send_wechat_text failed: {e}")
        return False

def process_due_tasks():
    now = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
    conn = sqlite3.connect("reminders.db")
    cur  = conn.cursor()
    cur.execute(
        "SELECT id, wxid, content FROM remind_tasks "
        "WHERE remind_at <= ? AND status = 'pending' AND retry_count < 3",
        (now,)
    )
    rows = cur.fetchall()
    for task_id, wxid, content in rows:
        success = send_wechat_text(wxid, content)
        if success:
            cur.execute(
                "UPDATE remind_tasks SET status='sent' WHERE id=?", (task_id,)
            )
        else:
            cur.execute(
                "UPDATE remind_tasks SET retry_count = retry_count + 1 WHERE id=?",
                (task_id,)
            )
    conn.commit()
    conn.close()
    print(f"[{now}] processed {len(rows)} tasks.")

scheduler = BlockingScheduler()
scheduler.add_job(process_due_tasks, "interval", seconds=60)
scheduler.start()

这段代码已经是生产可用的骨架：扫描、发送、记录结果、有限重试，逻辑清晰且容易扩展。

WechatApi 核心接口参数说明

WechatApi 基于微信 iPad 协议实现，鉴权方式统一使用请求头 VideosApi-token，业务参数通过 JSON body 传入。以文本消息为例：

请求示例（curl）：

bashcurl -X POST "https://your-wechatapi-endpoint/api/sendTextMsg" \
  -H "VideosApi-token: YOUR_VIDEOS_API_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "appId":   "YOUR_DEVICE_APP_ID",
    "toWxId":  "wxid_xxxxxxxxxxxxxxx",
    "content": "【待办提醒】您有一个会议将在 10 分钟后开始，请做好准备。"
  }'

成功响应示例：

json{
  "ret": 200,
  "msg": "操作成功",
  "data": {
    "msgId": "8176543210987654321",
    "createTime": 1718329200
  }
}

主要参数一览：

参数名	位置	类型	说明
`VideosApi-token`	Header	string	控制台颁发的鉴权 token，每设备唯一
`appId`	Body	string	设备 ID，登录控制台后获取
`toWxId`	Body	string	接收方的 wxid 或群 chatroom ID
`content`	Body	string	消息文本，支持换行符 `\n`
`ret`	响应	int	`200` 表示成功，其他值见错误码文档
`data.msgId`	响应	string	消息唯一 ID，可用于对账

群消息与私聊接口参数一致，区别在于 toWxId 传群 ID（以 @chatroom 结尾）。如果需要在群里 @ 某人，追加 atWxIds 数组字段即可，详见官方开发文档 post.wechatapi.net。

进阶：支持用户自主配置提醒的指令机器人

只有运营人员手动录入任务的系统，使用场景太窄。更实用的做法是让用户自己通过微信发指令来设置提醒，机器人解析后写入任务表。这需要在消息接收端（Webhook 或轮询）处理入站消息。

接收到用户发来的消息后，通过正则或 NLP（调用大模型提取意图）解析出时间和提醒内容。以简单正则为例：

pythonimport re
from datetime import datetime, timedelta

def parse_remind_command(text: str):
    """
    支持格式示例：
      "提醒我明天上午9点 开周会"
      "3小时后提醒我 提交报告"
      "提醒我2026-06-14 15:30 客户回访"
    返回 (remind_at, content) 或 None
    """
    # 匹配绝对时间：YYYY-MM-DD HH:MM
    m = re.search(r"(\d{4}-\d{2}-\d{2})\s+(\d{2}:\d{2})\s+(.+)", text)
    if m:
        dt = datetime.strptime(f"{m.group(1)} {m.group(2)}", "%Y-%m-%d %H:%M")
        return dt, m.group(3).strip()

    # 匹配相对时间：N小时后
    m = re.search(r"(\d+)小时后\S*\s+(.+)", text)
    if m:
        dt = datetime.utcnow() + timedelta(hours=int(m.group(1)))
        return dt, m.group(2).strip()

    return None

解析成功后直接 INSERT 到 remind_tasks 表，下一个调度周期自然会处理。整个闭环无需人工干预。

这种"用微信管理微信提醒"的设计非常受销售、运营团队欢迎——他们不需要打开额外的 App 或网页，在聊天框里说一句话就完成了待办创建。借助 WechatApi 的微信机器人开发能力，这套逻辑可以很快搭建起来并接入已有业务系统。

定时重复提醒与 Cron 表达式

一次性提醒只是基础能力，很多场景需要周期性提醒：

每天 9:00 发送当日任务清单
每周一 8:30 发送本周 OKR 回顾
每月最后一个工作日 17:00 发送月报填写提醒

在任务表的 repeat_rule 字段存储标准 cron 表达式（如 0 9 * * 1-5 表示工作日每天 9 点），调度引擎在任务执行后检查该字段：若为 NULL 则标记为 sent；若有值，则根据 cron 规则计算下一次触发时间并更新 remind_at，状态重置为 pending。

APScheduler 内置了 CronTrigger，可以直接解析标准 cron 表达式，不需要自己写计算逻辑：

pythonfrom apscheduler.triggers.cron import CronTrigger

# 示例：动态注册一个每天 9 点触发的 job
scheduler.add_job(
    func=lambda: send_wechat_text("wxid_xxxxxxx", "早安，今日待办已就绪，请查收！"),
    trigger=CronTrigger.from_crontab("0 9 * * *"),
    id="daily_morning_brief",
    replace_existing=True,
)

如果任务量大，建议用数据库驱动的 jobstore（APScheduler 支持 SQLAlchemy jobstore），这样重启服务后已注册的定时任务不会丢失。

生产部署注意事项

1. 设备在线保障

WechatApi 依赖微信设备保持登录状态。建议监听设备离线回调（或定期调用设备状态查询接口），一旦掉线立即告警，触发重新扫码登录流程，避免提醒消息静默丢失。

2. 消息发送频率控制

微信对短时间内大量发送消息有频率限制。若有批量提醒场景（如同时给 500 人发早报），需要在调度层做限速：每条消息之间加随机延迟（建议 1-3 秒），分批次投递而非并发轰炸。WechatApi 文档中有详细的频控建议，上线前务必阅读。

3. 时区处理

服务器、数据库和前端三处时区必须统一，否则"明天 9 点"可能变成"后天 2 点"。推荐全链路使用 UTC 存储，仅在展示层做时区转换。数据库字段用 DATETIME 配合应用层显式传 UTC 时间戳，避免依赖数据库时区配置。

4. 幂等发送

调度引擎在极少数情况下可能重复触发同一任务（服务重启、时钟漂移等）。建议用 UPDATE ... WHERE status='pending' 的原子操作配合数据库行锁，或引入分布式锁（Redis SETNX），保证同一任务只被发送一次。

5. 内容安全

用户自定义的提醒内容需要过滤敏感词，避免机器人账号因发送违规内容而被限制。可以在写入任务表前接入文本审核 API（腾讯云/阿里云均有现成服务），高风险内容拦截并返回提示。

实际业务场景举例

以下几类场景已有团队在用这套方案落地，可直接参考：

销售 CRM 跟进提醒：销售在 CRM 里更新客户状态时触发写入提醒任务，到期后机器人在个人微信推送"今天需要回访客户 XX，上次沟通进展：……"，比邮件触达率高出数倍。

运营每日播报：电商团队设置每天 8:00 推送昨日销售数据摘要到运营群，数据从内部 BI 系统实时拉取后拼成消息文本，比人工整理快且不容易出错。

项目里程碑提醒：项目管理工具（如 Jira、飞书多维表格）通过 Webhook 将截止日期写入提醒表，机器人提前 1 天、当天分别在项目群提醒负责人，避免交付物临时失联。

个人待办助手：个人用户通过微信给机器人发指令（"提醒我下周三上午 10 点去医院体检"），机器人确认后写入任务，到期推送提醒。这也是微信二次开发中非常典型的个人效率工具方向。

小结

本文从架构设计到核心代码，完整拆解了如何用 WechatApi + APScheduler 搭建一套微信消息定时提醒待办机器人。核心要点回顾：

三层解耦：存储、调度、投递各自独立，便于维护和扩展；
调用范式：HTTP POST + JSON body，鉴权 VideosApi-token 放请求头，appId 标识设备，响应体 ret:200 即成功；
重复提醒：用 cron 表达式存储规则，任务执行后自动计算下次触发时间；
生产保障：设备在线监控、限速发送、时区统一、幂等保护缺一不可。

如果你正在评估微信消息自动化方案，WechatApi 提供了完整的个人微信 API 能力，支持文本、图片、文件、群管理等丰富接口，可以直接在控制台注册试用，接入成本极低。