maolai7/agent-skills/wechat-watch

<skill-dir>/
├── SKILL.md                              ← skill 定义（AI 读取此文件执行）
├── services/
│   └── wechat-download-api/              ← 微信 API 服务
│       ├── app.py                        ← 服务入口
│       └── data/
│           └── rss.db                    ← 文章缓存数据库
├── scripts/
│   └── check_articles.py                 ← 检查新文章脚本
└── data/
    └── pushed_articles.json              ← 已推送文章记录

curl -s http://localhost:5000/api/health

curl -s http://localhost:5000/api/admin/status

curl -s http://localhost:5000/api/rss/subscriptions

curl -X DELETE http://localhost:5000/api/rss/subscribe/{fakeid}

┌─────────────────────────────────────────────────────────────┐
│                    wechat-download-api                      │
│                                                             │
│  RSS Poller（每小时自动）                                    │
│      ↓                                                      │
│  轮询订阅的公众号 → 缓存到 SQLite（含完整内容）               │
│                                                             │
│  数据表：                                                   │
│  - subscriptions：订阅列表                                  │
│  - articles：文章缓存（title, content, publish_time 等）    │
│                                                             │
└─────────────────────────────────────────────────────────────┘
                          ↓
┌─────────────────────────────────────────────────────────────┐
│                    wechat-watch skill                       │
│                                                             │
│  Cron (10:00 / 18:00)                                       │
│      ↓                                                      │
│  从 SQLite 读取缓存文章                                     │
│      ↓                                                      │
│  按 publish_time 筛选时间段                                 │
│      ↓                                                      │
│  总结 + 推送到飞书                                          │
│                                                             │
└─────────────────────────────────────────────────────────────┘

import time
from datetime import datetime, timedelta
# 早上10点任务：前一天18点到当天10点
if hour == 10:
    start_time = int((datetime.now() - timedelta(hours=16)).timestamp())  # 昨天18:00
    end_time = int(datetime.now().replace(hour=10, minute=0, second=0).timestamp())
# 晚上18点任务：当天10点到18点
if hour == 18:
    start_time = int(datetime.now().replace(hour=10, minute=0, second=0).timestamp())
    end_time = int(datetime.now().replace(hour=18, minute=0, second=0).timestamp())
# 筛选文章
for article in articles:
    if start_time <= article['publish_time'] <= end_time:
        # 这是新文章，需要推送

import sqlite3
from datetime import datetime, timedelta
# 连接数据库（使用相对路径，基于 skill 目录）
import os
skill_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))  # skill 根目录
db_path = os.path.join(skill_dir, "services", "wechat-download-api", "data", "rss.db")
conn = sqlite3.connect(db_path)
conn.row_factory = sqlite3.Row
# 计算时间段
now = datetime.now()
if now.hour == 10:
    # 早上10点：前一天18点到当天10点
    start_time = int((now - timedelta(hours=16)).timestamp())  # 昨天18:00
    end_time = int(now.replace(hour=10, minute=0, second=0).timestamp())
else:
    # 晚上18点：当天10点到18点
    start_time = int(now.replace(hour=10, minute=0, second=0).timestamp())
    end_time = int(now.replace(hour=18, minute=0, second=0).timestamp())
# 从缓存读取文章
cursor = conn.execute("""
    SELECT a.*, s.nickname
    FROM articles a
    JOIN subscriptions s ON a.fakeid = s.fakeid
    WHERE a.publish_time BETWEEN ? AND ?
    ORDER BY a.publish_time DESC
""", (start_time, end_time))
articles = [dict(row) for row in cursor]
conn.close()

📰 公众号文章更新（X 篇）
查询时间范围：YYYY-MM-DD HH:00 ~ YYYY-MM-DD HH:00
 
---
 
**【公众号名称】**
 
- **文章标题**：XXX
- **内容总结**：XXX...
- **原文链接**：[链接](URL)
 
---
 
**【另一个公众号】**
 
- **文章标题**：XXX
- **内容总结**：XXX...
- **原文链接**：[链接](URL)

📰 公众号文章更新
查询时间范围：YYYY-MM-DD HH:00 ~ YYYY-MM-DD HH:00
 
暂无新文章更新。

# 检查 Docker
docker --version
 
# 检查 Python
python3 --version

cd <skill-dir>/services
git clone https://github.com/tmwgsicp/wechat-download-api.git
cd wechat-download-api
cp env.example .env
docker-compose up -d

cd <skill-dir>/services
git clone https://github.com/tmwgsicp/wechat-download-api.git
cd wechat-download-api
bash start.sh

{
  "pushed_ids": [
    "article_id_1",
    "article_id_2"
  ],
  "last_check": "2026-03-24T18:00:00+08:00",
  "last_time_range": {
    "start": "2026-03-24T10:00:00+08:00",
    "end": "2026-03-24T18:00:00+08:00"
  }
}

 
### subscriptions.json（可选）
 
如果用户想用自己的订阅管理而不是服务的：
 

 
## 注意事项
 
1. **登录有效期**：约 4 天，过期后需重新扫码
2. **检查频率**：每天两次（10:00 和 18:00），按时间段筛选
3. **推送目标**：老板私聊，不是群聊
4. **图片发送**：二维码使用 `MEDIA:` 语法发送到飞书
5. **服务端口**：默认 5000，确保端口未被占用
6. **数据持久化**：项目 data 目录需要持久化，避免重启丢失数据
7. **无更新也推送**：每次检查后都要推送结果，无更新则说明"暂无新文章"
 
## 风控防护脚本
 
已封装请求逻辑到脚本，自动处理间隔：
 

 
脚本内置间隔规则：
- 搜索公众号：3 秒
- 获取文章内容：**5 秒**
- 添加订阅：3 秒
 
**执行时请使用脚本，不要直接调用 API，避免忘记间隔。**
 
## ⚠️ 风控防护（重要）
 
**微信会检测频繁请求并触发风控**，返回"请求过于频繁，请4秒后重试"。
 
**必须遵守以下间隔规则：**
 
| 操作 | 最小间隔 |
|------|---------|
| 搜索公众号 | 3 秒 |
| 获取文章列表 | 3 秒 |
| 获取文章内容 | **5 秒** |
| 添加订阅 | 3 秒 |
 
**示例代码：**
 

 
**如果触发风控：**
- 等待 10-30 分钟后重试
- 检查请求间隔是否足够
- 考虑配置代理池（设置 `PROXY_URLS` 环境变量）
 
## 用户交互指令
 
用户可以说：
- "帮我订阅公众号 XXX" → 搜索并添加订阅
- "查看我的公众号订阅" → 列出已订阅的公众号
- "取消订阅 XXX" → 取消订阅
- "检查公众号更新" → 手动触发一次轮询
- "重新登录微信" → 获取新二维码扫码
 
---
 
## 定时任务配置（推荐）
 
建议配置两个定时任务，每天自动检查并推送公众号文章更新。
 
### 任务 1：早上 10:00 推送
 
检查时间段：**昨天 18:00 ~ 今天 10:00**（16 小时，覆盖晚上 + 早上）
 

 
### 任务 2：晚上 18:00 推送
 
检查时间段：**今天 10:00 ~ 今天 18:00**
 

 
### 配置说明
 
| 配置项 | 说明 | 如何修改 |
|--------|------|----------|
| `agentId` | 执行任务的 agent | 保持 `info_collector` 或改为你的 agent ID |
| `delivery.to` | 推送目标用户 | **必须修改**为你的飞书 user ID |
| `schedule.expr` | cron 表达式 | 可根据需要调整时间 |
 
### 如何获取飞书 user ID
 
1. 在飞书中打开与 bot 的私聊
2. 发送任意消息
3. 查看 OpenClaw 日志或会话信息，找到 `user:ou_xxx` 格式的 ID
 
### 推送格式
 
**有新文章时：**

 
**无新文章时：**

 
### 设计说明
 
- **脚本只负责输出内容**：`check_articles.py` 输出飞书格式的 Markdown 到 stdout
- **推送由框架处理**：定时任务的 `delivery` 配置自动将输出推送到目标用户
- **通用性强**：skill 本身不写死 user ID，每个用户只需修改定时任务配置中的 `delivery.to`