Python 实现 X 新推文推送到飞书/企业微信群：轮询、SQLite 去重与 Webhook 机器人

定时触发（每 N 分钟）
   └→ 轮询：对每个订阅调用搜索接口，取最新推文
        └→ SQLite：INSERT OR IGNORE 去重，新推文状态记为 pending
             └→ 发送器：取出 pending 记录，合并成消息
                  ├→ 飞书自定义机器人（签名 + 消息卡片）
                  └→ 企业微信群机器人（markdown）
                       └→ 成功标记为 sent，失败保留 pending 等待下轮重试

# 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 []

# 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 < 5 ORDER BY inserted LIMIT ?", (limit,)).fetchall()

    def mark(self, ids: list[str], ok: bool):
        with self.db:
            if ok:
                self.db.executemany("UPDATE tweets SET status='sent' WHERE tweet_id=?",
                                    [(i,) for i in ids])
            else:
                self.db.executemany("UPDATE tweets SET retries=retries+1 WHERE tweet_id=?",
                                    [(i,) for i in ids])

# feishu.py  示例代码
import base64
import hashlib
import hmac
import time
import requests
from config import FEISHU_WEBHOOK, FEISHU_SECRET


def gen_sign(timestamp: int, secret: str) -> 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

# 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}** <font color=\"comment\">{sub}</font>\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

# 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)