一、需求背景与整体思路

信息爆炸的时代,手动刷新新闻网站效率很低。我们可以用 Python 写一个轻量脚本,定时抓取 RSS 源中的最新文章,筛选出未推送过的条目,再通过企业微信群机器人 Webhook 自动推送到群里。整个方案无需服务器常驻进程,一台能跑 Python 的机器或定时任务平台即可完成。

整体流程可以概括为四步:

  1. 拉取 RSS:使用 feedparser 解析 RSS/Atom 源,得到标题、链接、发布时间等字段。
  2. 去重判断:用本地文件或 SQLite 记录已推送条目的唯一标识,避免重复推送。
  3. 格式化消息:把新闻整理成企业微信支持的 Markdown 消息体。
  4. 发送推送:调用企业微信群机器人 Webhook 接口完成推送。
flowchart TD
    A[定时触发] --> B[拉取 RSS 源]
    B --> C[解析条目]
    C --> D{是否已推送}
    D -- 是 --> E[跳过]
    D -- 否 --> F[格式化 Markdown]
    F --> G[调用企业微信 Webhook]
    G --> H[记录已推送 ID]
    H --> E

二、环境准备

先安装依赖库。核心只需要两个:feedparser 负责解析 RSS,requests 负责发送 HTTP 请求。

pip install feedparser requests

接着在企业微信群里创建机器人:进入群聊 → 右上角「…」→「群机器人」→「添加机器人」,创建后复制 Webhook 地址,形如:

https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx

注意:Webhook 中的 key 相当于密码,不要提交到公开仓库。建议通过环境变量注入。

三、核心代码实现

3.1 拉取并解析 RSS

feedparser.parse() 可以直接接受 URL,返回的 entries 是条目列表。不同源字段略有差异,取 link 作为唯一标识最稳妥。

import feedparser
def fetch_entries(feed_url):
feed = feedparser.parse(feed_url)
if feed.bozo:
print(f"[WARN] RSS 解析异常: {feed.bozo_exception}")
entries = []
for item in feed.entries:
entries.append({
"id": item.get("link") or item.get("id"),
"title": item.get("title", "无标题"),
"link": item.get("link", ""),
"published": item.get("published", ""),
})
return entries

3.2 本地去重记录

用一个 JSON 文件保存已推送的 ID 集合,简单可靠。数据量大时可以换成 SQLite。

import json
import os
SEEN_FILE = "seen_ids.json"
def load_seen():
if not os.path.exists(SEEN_FILE):
return set()
with open(SEEN_FILE, "r", encoding="utf-8") as f:
return set(json.load(f))
def save_seen(seen):
with open(SEEN_FILE, "w", encoding="utf-8") as f:
json.dump(sorted(seen), f, ensure_ascii=False, indent=2)

3.3 构造并发送企业微信消息

企业微信群机器人支持 text、markdown、news 等消息类型。这里用 markdown,可读性更好。注意企业微信 markdown 对链接语法支持有限,正文中直接给出 URL 更稳妥。

import requests
def send_wecom(webhook, title, entries):
lines = [f"## {title}", ""]
for e in entries:
lines.append(f"{e['title']}")
lines.append(f"> {e['published']}")
lines.append(f"查看原文")
lines.append("")
content = "\n".join(lines)
payload = {
    "msgtype": "markdown",
    "markdown": {"content": content[:4000]}
}
resp = requests.post(webhook, json=payload, timeout=10)
resp.raise_for_status()
result = resp.json()
if result.get("errcode") != 0:
    raise RuntimeError(f"推送失败: {result}")
return result</code></pre>
3.4 组装主流程
import os
FEEDS = [
{"name": "科技新闻", "url": "https://example.com/rss/tech.xml"},
{"name": "行业动态", "url": "https://example.com/rss/industry.xml"},
]
def main():
webhook = os.environ["WECOM_WEBHOOK"]
seen = load_seen()
new_seen = set(seen)
for feed in FEEDS:
    entries = fetch_entries(feed["url"])
    fresh = [e for e in entries if e["id"] and e["id"] not in seen]
    if not fresh:
        print(f"[INFO] {feed['name']} 无新内容")
        continue
    send_wecom(webhook, feed["name"], fresh)
    new_seen.update(e["id"] for e in fresh)
    print(f"[INFO] {feed['name']} 推送 {len(fresh)} 条")

save_seen(new_seen)
if name == "main":
main()
四、定时运行方案
脚本本身只执行一次,需要外部调度。常见做法有三种:
方式
适用场景
说明
crontab
Linux 服务器
配置简单,适合长期运行
Windows 任务计划
本地 Windows 机器
图形化配置,无需额外依赖
GitHub Actions
无服务器
用 schedule 触发,需处理状态持久化
以 crontab 为例,每 30 分钟执行一次:
*/30 * * * * cd /opt/rss-bot && /usr/bin/python3 main.py >> run.log 2>&1
五、常见问题与优化建议
消息长度限制:企业微信 markdown 内容上限约 4096 字节,条目多时要分批发送或截断。
频率限制:单个机器人每分钟最多 20 条消息,批量推送需加 time.sleep 控制节奏。
RSS 源不稳定:给 requests 加超时和重试,feedparser 解析失败时记录日志而不是直接崩溃。
去重文件并发:多实例运行时建议改用 SQLite 并加唯一索引,避免竞态。
敏感信息保护:Webhook key 一律走环境变量,日志中不要打印完整 URL。
按以上步骤,一个可用的 RSS 新闻自动推送脚本就完成了。后续可以按需扩展:增加关键词过滤、接入多个群、把推送记录写入数据库做统计等。
Logo

DAMO开发者矩阵,由阿里巴巴达摩院和中国互联网协会联合发起,致力于探讨最前沿的技术趋势与应用成果,搭建高质量的交流与分享平台,推动技术创新与产业应用链接,围绕“人工智能与新型计算”构建开放共享的开发者生态。

更多推荐