跳至正文

给每日文章自动化加 watchdog：cron + 状态轮询 + 微信告警，再也不漏发

给每日文章自动化加 watchdog：cron + 状态轮询 + 微信告警，再也不漏发

今天下午 3 点老板问我”今天发了几篇”，我去 ai.hkras.com 后台一拉——cat28 发了，cat4 啥都没有。

文章自动化 watchdog

10:00 的自动化跑挂了。等到下午才发现，断更一整天。

这事我得治。根本原因不是”自动化写挂了”，是”挂了没人知道”。自动化总有概率失败（API 抽风、磁盘满、token 过期、配额耗尽…），但等到读者 / 老板发现就晚了。

今天我就把这个洞补上——给文章自动化加一个 watchdog。

一、需求拆解：watchdog 要干啥

我想要的很简单：

  • 每小时查一次：今天该发的 cat4（10:00）、cat28（11:00）到底发了没
  • 到了发布点 + 30 分钟还没发：触发告警
  • 告警走微信：直接发到我手机，不用我手动查后台

听起来 50 行 Python 的事，但真做起来踩了 5 个坑，下面一个一个说。

状态轮询

二、核心代码

先上完整脚本，再讲坑。这个脚本我直接放在服务器 /opt/watchdog/check_posts.py，每小时 cron 跑一次：

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""文章自动化 watchdog — 检查今日 cat4/cat28 是否已发布"""

import os
import sys
import json
import requests
from datetime import datetime, timedelta
import pytz

# —— 配置 ——
WP_URL = os.getenv("WP_URL", "https://ai.hkras.com")
WP_USER = os.getenv("WP_USER", "hewenqiang")
WP_PASS = os.getenv("WP_PASS")  # 走环境变量，不写死
WX_PROXY = os.getenv("WX_PROXY", "https://ai.hkras.com/wx-proxy.php")
WX_TOKEN = os.getenv("WX_TOKEN")  # 脱敏

# 预期发布时间表（BJT）
SCHEDULE = {
    4:  {"hour": 10, "minute": 0, "name": "技术笔记"},
    28: {"hour": 11, "minute": 0, "name": "AI战略转型"},
}

BJT = pytz.timezone("Asia/Shanghai")
GRACE_MIN = 30  # 发布点后宽限 30 分钟


def fetch_today_posts(cat_id: int):
    """拉今天已发布的这个分类的最新一篇"""
    now = datetime.now(BJT)
    start = now.replace(hour=0, minute=0, second=0, microsecond=0)
    try:
        r = requests.get(
            f"{WP_URL}/wp-json/wp/v2/posts",
            params={
                "categories": cat_id,
                "after": start.isoformat(),
                "per_page": 1,
                "orderby": "date",
                "order": "desc",
            },
            auth=(WP_USER, WP_PASS),
            timeout=10,
        )
        r.raise_for_status()
        return r.json()
    except Exception as e:
        return {"_error": str(e)}


def send_wechat(text: str):
    """走 wx-proxy.php 发送微信通知"""
    try:
        r = requests.post(
            WX_PROXY,
            json={
                "action": "send_text",
                "proxy_token": WX_TOKEN,
                "to_user": "hewenqiang",
                "content": text,
            },
            timeout=10,
        )
        r.raise_for_status()
        return r.json()
    except Exception as e:
        # 兜底：写本地日志
        with open("/opt/watchdog/wx_failed.log", "a") as f:
            f.write(f"{datetime.now(BJT).isoformat()}\t{text}\t{e}\n")
        return {"_error": str(e)}


def main():
    now = datetime.now(BJT)
    print(f"[{now.isoformat()}] watchdog 启动")

    for cat_id, sched in SCHEDULE.items():
        # 发布点时间戳
        publish_at = now.replace(
            hour=sched["hour"], minute=sched["minute"], second=0, microsecond=0
        )
        # 还没到发布点 → 跳过
        if now < publish_at:
            print(f"⏳ cat{cat_id} 还没到发布点 {sched['hour']:02d}:{sched['minute']:02d}")
            continue

        # 还没过宽限期 → 跳过
        if now < publish_at + timedelta(minutes=GRACE_MIN):
            print(
                f"⏳ cat{cat_id} 在宽限期 {GRACE_MIN}min 内，再观察"
            )
            continue

        # 真正检查：今天这个分类有没有已发布的文章
        posts = fetch_today_posts(cat_id)
        if isinstance(posts, dict) and posts.get("_error"):
            send_wechat(
                f"🚨 watchdog 异常\n"
                f"cat{cat_id} ({sched['name']}) 查询失败\n"
                f"err: {posts['_error']}"
            )
            continue

        if not posts:
            # 没发 → 告警
            send_wechat(
                f"🚨 文章漏发告警\n"
                f"分类: cat{cat_id} ({sched['name']})\n"
                f"预期: {sched['hour']:02d}:{sched['minute']:02d}\n"
                f"当前: {now.strftime('%H:%M')}\n"
                f"URL: {WP_URL}/wp-admin/edit.php?cat={cat_id}"
            )
            print(f"❌ cat{cat_id} 今日未发布，已告警")
        else:
            title = posts[0]["title"]["rendered"]
            print(f"✅ cat{cat_id} 已发布: {title[:30]}")


if __name__ == "__main__":
    main()

cron 这边每小小时跑一次：

# 注意：服务器是 UTC，下面 0-23 是 UTC 时间
# 北京时间 0-23 = UTC 16-15（次日）
# 我直接用绝对时间偏移：北京时间每小时 → UTC 减 8 小时
0 1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16 * * * cd /opt/watchdog && /usr/bin/python3 check_posts.py >> /opt/watchdog/run.log 2>&1

跑了一周，没再漏发过。

微信告警通道

三、5 个踩坑经验（每个都让我多花了半小时）

坑 1：服务器时区是 UTC，cron 不指定 TZ 就完蛋

我一开始写 0 10 * * * 想着北京 10:00 跑，结果发现服务器时区是 UTC——这个时间跑的是北京 18:00。watchdog 全天都在”等发布点”，真正的 10:00 反而没人查。

正解：要么在 cron 前面加 TZ=Asia/Shanghai（crontab 头部 CRON_TZ=Asia/Shanghai），要么把时间转成 UTC 写死。我选了后者，简单粗暴。

坑 2：WordPress REST API 过滤 categories 用 ID，不是 slug

我一开始写 ?categories=tech-notes，查回 0 篇，懵了半天。WP REST API 的 categories 参数只接受数字 ID——cat4 是 4，不是 "tech-notes"。

slugs 看 wp-admin 后台 → 文章 → 分类，鼠标悬停就能看到 tag_ID。也可以直接 GET /wp-json/wp/v2/categories?slug=tech-notes 查。

坑 3：after 参数必须带时区

?after=2026-08-19T00:00:00 这种 naive datetime WP 会按 UTC 解析——北京 0:00 发的文章会被漏掉，因为它在 WP 看来是”前一天的 16:00″。

正解：带 timezone 的 ISO 8601，例如 2026-08-19T00:00:00+08:00。

start = now.replace(hour=0, minute=0, second=0, microsecond=0)
# .isoformat() 会带上 +08:00（如果 BJT 是带 tzinfo 的）

坑 4：watchdog 自己也可能挂

第一次部署完我以为万事大吉。结果有一天我自己服务的 PHP-FPM 挂了，watchdog 也没跑成，没人发现。

正解是双层：
– 服务器内部 systemd 监控 watchdog 进程（Restart=on-failure）
– 云厂商的云监控再监控一次服务器本身（CPU / 内存 / 进程数）

而且 watchdog 的告警通道（wx-proxy.php）也要定期演练——我每周手动 curl 一次，确认 token 还能用。

坑 5：微信测试号每分钟限 20 条，别让 watchdog 太勤

我一开始图省事写”每 5 分钟跑一次”，结果某次 API 抽风，watchdog 连发 20 条微信轰炸我手机。

正解：
– 每小时跑一次足够（10:30 之前 cat4 没人查也没事，发文本来就在 10:00）
– 告警里加一个去重逻辑——同一类问题 1 小时内只发一次
– 真正紧急的事不要走微信测试号，走 Server 酱 / 阿里云告警 / 短信

双层监控

四、最后说两句

自动化这东西，写出来只完成了一半，能”感知失败”才算闭环。

我今天的教训：
– 10:00 跑挂了 → 没人知道 → 下午才发现
– 加了 watchdog → 下次再挂，10:30 微信就响了

代码就上面那些，核心是 cron + WP REST API + wx-proxy.php 三件套。如果你也在跑类似的定时发文 / 定时同步 pipeline，别等出问题再加 watchdog，今天就加。

不然你下一次发现漏发，可能就是读者比你先到。

（全文完）

发表回复

您的邮箱地址不会被公开。 必填项已用 * 标注