Python 实现 X 新推文推送到飞书/企业微信群：轮询、SQLite 去重与 Webhook 机器人
## 一、需求场景 很多团队都有"盯 X"的需求： - 市场团队关注竞品官方账号，一有新功能发布就要知道； - 研发团队关注几个开源项目维护者的动态； - 公关团队需要第一时间发现品牌名称被大量讨论的情况。 这些信息如果靠人工刷，既容易遗漏又占用精力。更好的方式是让程序自动发现新推文，并直接推送到团队日常使用的飞书或企业微信群。 本文实现一个轻量服务，满足以下要求： 1. 支持按**账号**和**关键词**两种方式订阅； 1. 同一条推文**只推送一次**，服务重启后也不会重复推送； 1. 首次启动时**不推送历史内容**，避免刷屏； 1. 同时支持**飞书**和**企业微信**机器人，并遵守它们各自的频率和长度限制； 1. 发送失败的消息能够**自动重试**。 ## 二、整体架构 ``` 定时触发（每 N 分钟） └→ 轮询：对每个订阅调用搜索接口，取最新推文 └→ SQLite：INSERT OR IGNORE 去重，新推文状态记为 pending └→ 发送器：取出 pending 记录，合并成消息 ├→ 飞书自定义机器人（签名 + 消息卡片） └→ 企业微信群机器人（markdown） └→ 成功标记为 sent，失败保留 pending 等待下轮重试 ``` 这里把"发现新推文"和"发送消息"拆成两个阶段，中间用 SQLite 中的状态字段衔接。这样做的好处是：即使发送环节失败（比如机器人被限流），推文也已经入库，下一轮会自动重发，不会丢失。 ## 三、订阅配置与轮询 本文示例使用第三方接口 TwexAPI 的高级搜索接口获取推文。它支持 X 原生搜索语法，`sortBy` 设为 `Latest` 时按时间倒序返回结果。账号订阅可以用 `from:` 运算符实现，多个账号用 `OR` 连接，这样一次请求就能覆盖多个账号。 ``` # config.py 示例代码 import os BASE_URL = "https://api.twexapi.io" API_HEADERS = {"Authorization": f"Bearer {os.environ['DATA_API_KEY']}"} FEISHU_WEBHOOK = os.environ.get("FEISHU_WEBHOOK") # https://open.feishu.cn/open-apis/bot/v2/hook/xxx FEISHU_SECRET = os.environ.get("FEISHU_SECRET") # 开启签名校验后获得的密钥 WECOM_WEBHOOK = os.environ.get("WECOM_WEBHOOK") # https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxx SUBSCRIPTIONS = [ {"name": "竞品官号", "query": "(from:account_a OR from:account_b) -filter:replies"}, {"name": "品牌舆情", "query": "\"YourBrand\" -filter:retweets"}, ] POLL_INTERVAL = 300 # 秒 ``` ``` # poller.py 示例代码 import time import requests from config import BASE_URL, API_HEADERS def fetch_latest(query: str) -> list[dict]: """只取第一页：轮询场景下，两次轮询之间的新推文通常不会超过一页。""" for attempt in range(3): r = requests.post(f"{BASE_URL}/twitter/advanced_search/page", headers=API_HEADERS, json={"searchTerms": [query], "sortBy": "Latest"}, timeout=30) if r.status_code == 429: time.sleep(int(r.headers.get("Retry-After", 10))) continue r.raise_for_status() return [t for t in r.json().get("data") or [] if t] return [] ``` 轮询只取第一页（最多 20 条）。如果某个关键词在 5 分钟内的新推文经常超过 20 条，说明它的热度已经不适合逐条推送，应该改为"每小时汇总"的形式，或者在查询中加上 `min_faves:` 之类的条件，只推送有一定互动量的内容。 ## 四、SQLite 去重与首次运行基线 去重最可靠的方式是以 `tweet_id` 作为主键，借助数据库的唯一约束实现幂等写入： ``` # store.py 示例代码 import sqlite3 import time DDL = """ CREATE TABLE IF NOT EXISTS tweets ( tweet_id TEXT PRIMARY KEY, sub_name TEXT NOT NULL, author TEXT, text TEXT, url TEXT, created_at TEXT, status TEXT NOT NULL DEFAULT 'pending', -- pending / sent / skipped retries INTEGER NOT NULL DEFAULT 0, inserted REAL NOT NULL ); CREATE INDEX IF NOT EXISTS idx_status ON tweets(status); CREATE TABLE IF NOT EXISTS subs_seen (sub_name TEXT PRIMARY KEY); """ class Store: def __init__(self, path="tweets.db"): self.db = sqlite3.connect(path) self.db.executescript(DDL) def is_first_run(self, sub: str) -> bool: return self.db.execute("SELECT 1 FROM subs_seen WHERE sub_name=?", (sub,)).fetchone() is None def save(self, sub: str, tweets: list[dict]) -> int: """写入新推文；首次运行的订阅直接标记为 skipped，作为基线。""" status = "skipped" if self.is_first_run(sub) else "pending" n = 0 with self.db: for t in tweets: u = t.get("user") or {} cur = self.db.execute( "INSERT OR IGNORE INTO tweets VALUES (?,?,?,?,?,?,?,0,?)", (t["tweet_id"], sub, u.get("screen_name"), t.get("full_text") or t.get("text") or "", f"https://x.com/{u.get('screen_name')}/status/{t['tweet_id']}", t.get("created_at"), status, time.time())) n += cur.rowcount self.db.execute("INSERT OR IGNORE INTO subs_seen VALUES (?)", (sub,)) return n if status == "pending" else 0 def pending(self, limit=10) -> list[tuple]: return self.db.execute( "SELECT tweet_id, sub_name, author, text, url FROM tweets " "WHERE status='pending' AND retries str: string_to_sign = f"{timestamp}\n{secret}" code = hmac.new(string_to_sign.encode("utf-8"), digestmod=hashlib.sha256).digest() return base64.b64encode(code).decode("utf-8") def build_card(title: str, items: list[tuple]) -> dict: elements = [] for _id, sub, author, text, url in items: elements.append({"tag": "markdown", "content": f"**@{author}**：{text[:300]}\n[查看原文]({url})"}) elements.append({"tag": "hr"}) return { "schema": "2.0", "header": {"title": {"tag": "plain_text", "content": title}, "template": "blue"}, "body": {"elements": elements[:-1]}, } def send_feishu(title: str, items: list[tuple]) -> bool: payload = {"msg_type": "interactive", "card": build_card(title, items)} if FEISHU_SECRET: ts = int(time.time()) payload.update(timestamp=str(ts), sign=gen_sign(ts, FEISHU_SECRET)) r = requests.post(FEISHU_WEBHOOK, json=payload, timeout=10) data = r.json() if data.get("code") != 0: print("feishu error:", data) # 19021 签名失败；11232 触发限流 return False return True ``` 飞书自定义机器人的几个限制需要特别注意： - **频率限制**：单个机器人每分钟 100 次、每秒 5 次。文档还建议避开整点和半点发送，否则可能因为系统压力触发 `11232` 限流错误； - **请求体大小**：不能超过 20KB，所以单张卡片里放的推文数量要有上限，正文也要截断； - **判断成功的方式**：HTTP 状态码为 200 并不代表发送成功，要看响应体中的 `code` 是否为 0。 ## 六、企业微信群机器人 ### 1. 创建 在企业微信群中添加"消息推送"（原"群机器人"），即可获得 Webhook 地址：`https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxxx`。地址中的 `key` 就是凭证，务必妥善保管。 ### 2. 发送 markdown 消息 企业微信群机器人支持 text、markdown、markdown_v2、图片、图文、文件、语音和模板卡片等消息类型。对于推文推送，markdown 类型是比较合适的选择：支持加粗、链接、引用和三种内置字体颜色。 ``` # wecom.py 示例代码 import requests from config import WECOM_WEBHOOK MAX_BYTES = 4000 # 文档限制 4096 字节，预留余量 def build_markdown(title: str, items: list[tuple]) -> str: parts = [f"### {title}"] for _id, sub, author, text, url in items: block = f"\n**@{author}** {sub}\n> {text[:200]}\n[查看原文]({url})" if len("\n".join(parts + [block]).encode("utf-8")) > MAX_BYTES: parts.append("\n……更多内容见下一条消息") break parts.append(block) return "\n".join(parts) def send_wecom(title: str, items: list[tuple]) -> bool: payload = {"msgtype": "markdown", "markdown": {"content": build_markdown(title, items)}} r = requests.post(WECOM_WEBHOOK, json=payload, timeout=10) data = r.json() if data.get("errcode") != 0: print("wecom error:", data) return False return True ``` 企业微信的限制： - **长度**：markdown 内容不超过 4096 字节（注意是字节，UTF-8 编码下一个汉字占 3 个字节），所以代码中按字节而不是按字符数判断； - **频率**：每个机器人每分钟不能超过 20 条消息。这个限制比飞书严格得多，因此**合并发送**几乎是必需的； - **判断成功**：看响应中的 `errcode` 是否为 0。 另外，markdown 消息中引用块 `>` 只对当前行生效。如果推文正文中有换行，建议先把换行替换为空格，再放进引用块，否则排版会乱。 ## 七、把各部分串起来 ``` # main.py 示例代码 import time from config import SUBSCRIPTIONS, POLL_INTERVAL, FEISHU_WEBHOOK, WECOM_WEBHOOK from poller import fetch_latest from store import Store from feishu import send_feishu from wecom import send_wecom def clean(items): return [(i, s, a, (t or "").replace("\n", " "), u) for i, s, a, t, u in items] def run_once(store: Store): for sub in SUBSCRIPTIONS: n = store.save(sub["name"], fetch_latest(sub["query"])) if n: print(f"[{sub['name']}] 新推文 {n} 条") time.sleep(1) batch = clean(store.pending(limit=8)) # 每次最多合并 8 条 if not batch: return title = f"X 动态提醒（{len(batch)} 条）" ok = True if FEISHU_WEBHOOK: ok &= send_feishu(title, batch) if WECOM_WEBHOOK: ok &= send_wecom(title, batch) store.mark([b[0] for b in batch], ok) if __name__ == "__main__": store = Store() while True: try: run_once(store) except Exception as e: # 单轮失败不影响下一轮 print("round failed:", e) time.sleep(POLL_INTERVAL) ``` 这里有一个简化处理：同时配置两个渠道时，只要有一个发送失败，整批都会在下一轮重发，成功的那个渠道可能收到重复消息。如果对重复比较敏感，可以在表中为每个渠道分别记录发送状态。 ## 八、部署与运维 1. **运行方式**：可以直接用 `systemd` 守护上面的循环进程，配置 `Restart=always`；也可以去掉 `while` 循环，改为每 5 分钟由 cron 执行一次 `run_once`，配合 `flock` 防止重叠； 1. **密钥管理**：Webhook 地址、签名密钥和数据接口密钥都通过环境变量注入，不写入代码，也不提交到仓库。飞书和企业微信的文档都特别提醒，不要把 Webhook 地址公开到 GitHub 或博客等地方； 1. **轮询间隔**：间隔越短，消息越及时，但调用量也越大。对于官号动态，5～10 分钟通常就足够了； 1. **数据清理**：定期删除较早的记录，例如只保留 30 天，既能控制数据库大小，也符合"最小保留"原则； 1. **监控**：统计连续失败次数，超过阈值时通过另一个渠道告警，避免推送服务"静默失效"。 ## 九、可以继续扩展的方向 - **关键词高亮与分级**：命中核心词时使用红色卡片模板（飞书 `template` 改为 `red`），并 @ 相关负责人； - **互动阈值过滤**：只推送点赞或转发超过一定数量的推文，降低噪音； - **大模型摘要**：对长推文或串推先生成一句话摘要再推送； - **翻译**：把英文推文翻译成中文，方便团队阅读。 ## 十、合规提示 只订阅和推送公开内容；推送时附原文链接，而不是大段复制原文；不针对个人做持续监控；遵守数据来源平台的服务条款、企业内部的信息安全规定和相关法律法规。 ## 十一、小结 1. 把"发现新推文"和"发送消息"解耦，中间用 SQLite 记录状态，可以同时解决去重、重启恢复和失败重试三个问题； 1. 首次运行基线可以避免新增订阅时刷屏； 1. 飞书签名的关键是"拼接字符串作为 HMAC 密钥、对空消息签名"，判断成功要看响应体中的 `code`； 1. 企业微信每分钟 20 条、markdown 4096 字节的限制，决定了必须合并发送并按字节截断。 > 说明：文中代码为示例代码，飞书和企业微信的消息格式及限制以其官方文档的最新版本为准。 ## 参考资料 - 飞书自定义机器人使用指南：[https://open.feishu.cn/document/client-docs/bot-v3/add-custom-bot](https://open.feishu.cn/document/client-docs/bot-v3/add-custom-bot) - 企业微信消息推送配置说明：[https://developer.work.weixin.qq.com/document/path/91770](https://developer.work.weixin.qq.com/document/path/91770) - 示例数据接口文档：[https://docs.twexapi.io](https://docs.twexapi.io)
