实时监测

发现特定账号的新推文，及时追踪关键词提及，并将实时 X 数据输入应用。本指南介绍如何基于 Sorsa API，使用主动拉取的轮询模式构建近实时监测管道。检测延迟取决于轮询间隔、API 响应时间，以及帖子何时进入所选信息流或搜索索引。设计时应围绕检查点、分页和去重，不要假设每条新帖子都会出现在下一次响应中。

免费构建原型： 初始赠送的 100 次请求可访问所有 Sorsa 端点，一次性赠送，无需信用卡，永不过期。足以搭建下方任意监测器、确认能够发现实时推文，并在选购套餐前验证 Slack 或 Discord 分流。轮询消耗较多请求，请根据下方用量表和轮询间隔选择付费套餐。

注意： 更多架构模式和端到端示例请参阅博客上的使用 REST API 实时监测 Twitter。

轮询监测的工作方式

从社交平台获取数据有推送式（流式传输、webhook）和拉取式（轮询）两种方式。Sorsa API 使用轮询，分为四步：

定期轮询端点，间隔为 1–30 秒。
比较结果与之前见过的推文 ID，识别新内容。
处理新推文： 发送提醒、保存，或分发至 Slack、Discord 等。
重复。

推文 ID 编码了创建时间，可通过 Python 整数或 JavaScript BigInt 比较。最大已见 ID 可作为检查点，但帖子可能延迟出现或乱序返回。生产环境应重复检查一段重叠时间窗口并按 ID 去重。崩溃后恢复需要显式保存和加载检查点。

选择合适的端点

监测目标	端点	方法	选择原因
单个账号	`/user-tweets`	POST	返回该用户时间线中的最新推文
同时最多 5,000 个账号	`/list-tweets`	GET	一次请求覆盖 X 列表中的全部成员
关键词或话题标签	`/search-tweets`	POST	完整支持搜索运算符，使用 `order: latest` 按时间排序
账号的 @提及	`/mentions`	POST	专为提及追踪设计，支持互动筛选

第一级：监测单个账号

这是最简单的情况。循环轮询 /user-tweets 的第一页，输出 ID 大于上次已见 ID 的推文。首次成功轮询时只建立基线，不输出现有推文。这是开发示例：如果两次轮询之间或进程停止期间新增内容超过一页，可能漏掉帖子。

Python

import requests
import time

API_KEY = "YOUR_API_KEY"
USERNAME = "elonmusk"
POLL_INTERVAL = 5  # seconds

URL = "https://api.sorsa.io/v3/user-tweets"
HEADERS = {"ApiKey": API_KEY, "Content-Type": "application/json"}

last_seen_id = None

print(f"Monitoring @{USERNAME}...")

while True:
    try:
        resp = requests.post(URL, headers=HEADERS, json={"username": USERNAME}, timeout=30)
        resp.raise_for_status()
        tweets = resp.json().get("tweets", [])

        if tweets:
            # Snowflake IDs arrive as strings. Use the highest (newest) ID in the
            # batch; this stays correct even if a pinned tweet appears first.
            top_id = max(int(t["id"]) for t in tweets)

            if last_seen_id is None:
                last_seen_id = top_id
                print(f"Baseline set: {last_seen_id}")
            else:
                new_tweets = [t for t in tweets if int(t["id"]) > last_seen_id]
                for tweet in reversed(new_tweets):  # oldest first
                    print(f"[NEW] @{USERNAME}: {tweet['full_text'][:140]}")
                if new_tweets:
                    last_seen_id = top_id

    except requests.exceptions.RequestException as e:
        print(f"Error: {e}")
        time.sleep(POLL_INTERVAL * 2)
        continue

    time.sleep(POLL_INTERVAL)

JavaScript

const API_KEY = "YOUR_API_KEY";
const USERNAME = "elonmusk";
const POLL_INTERVAL = 5000;

let lastSeenId = null;
console.log(`Monitoring @${USERNAME}...`);

while (true) {
  try {
    const resp = await fetch("https://api.sorsa.io/v3/user-tweets", {
      method: "POST",
      headers: { "ApiKey": API_KEY, "Content-Type": "application/json" },
      body: JSON.stringify({ username: USERNAME }),
    });
    if (!resp.ok) throw new Error(`HTTP ${resp.status}`);

    const tweets = (await resp.json()).tweets || [];

    if (tweets.length > 0) {
      // BigInt avoids precision loss on 64-bit Snowflake IDs.
      // Use the highest ID in the batch (robust if a pinned tweet appears first).
      let topId = 0n;
      for (const t of tweets) {
        const id = BigInt(t.id);
        if (id > topId) topId = id;
      }

      if (lastSeenId === null) {
        lastSeenId = topId;
        console.log(`Baseline set: ${lastSeenId}`);
      } else {
        const newTweets = tweets.filter((t) => BigInt(t.id) > lastSeenId);
        for (const t of [...newTweets].reverse()) {
          console.log(`[NEW] @${USERNAME}: ${t.full_text.slice(0, 140)}`);
        }
        if (newTweets.length) lastSeenId = topId;
      }
    }
  } catch (err) {
    console.error(`Error: ${err.message}`);
    await new Promise((r) => setTimeout(r, POLL_INTERVAL * 2));
    continue;
  }
  await new Promise((r) => setTimeout(r, POLL_INTERVAL));
}

这种方式不适合大量账号。监测 50 个账号需要 50 个独立循环和 50 倍请求。此时应使用 X 列表。

第二级：通过一次请求监测多个账号

X 列表最多可包含 5,000 个账号。/list-tweets 一次 API 调用即可返回全部成员合并后的最新推文。这是生产环境多账号监测的默认方案，详情见列表与社群。

第 1 步：创建公开 X 列表

前往 X Lists 创建列表。
添加要监测的账号，最多 5,000 个。
将列表设为 Public。API 无法访问私有列表。
从 URL 复制列表 ID。例如 https://x.com/i/lists/1234567890 的 ID 为 1234567890。

第 2 步：轮询列表

import requests
import time

API_KEY = "YOUR_API_KEY"
LIST_ID = "YOUR_LIST_ID"
POLL_INTERVAL = 5

URL = f"https://api.sorsa.io/v3/list-tweets?list_id={LIST_ID}"
HEADERS = {"ApiKey": API_KEY, "Accept": "application/json"}


def monitor_list(callback, interval=POLL_INTERVAL):
    """Poll an X List and call `callback` for each new tweet detected."""
    last_seen_id = None
    print(f"Monitoring List {LIST_ID} (interval: {interval}s)")

    while True:
        try:
            resp = requests.get(URL, headers=HEADERS, timeout=10)
            resp.raise_for_status()
            tweets = resp.json().get("tweets", [])

            if not tweets:
                time.sleep(interval)
                continue

            top_id = max(int(t["id"]) for t in tweets)

            if last_seen_id is None:
                last_seen_id = top_id
                print(f"Baseline set: {last_seen_id}")
            else:
                new_tweets = [t for t in tweets if int(t["id"]) > last_seen_id]
                if new_tweets:
                    for tweet in reversed(new_tweets):
                        callback(tweet)
                    last_seen_id = top_id

        except requests.exceptions.RequestException as e:
            print(f"Request error: {e}. Retrying in {interval * 2}s")
            time.sleep(interval * 2)
            continue

        time.sleep(interval)


def on_new_tweet(tweet):
    user = tweet["user"]
    print(f"[NEW] @{user['username']}: {tweet['full_text'][:120]}")
    print(
        f"       Likes: {tweet.get('likes_count', 0)} | "
        f"RTs: {tweet.get('retweet_count', 0)} | "
        f"Views: {tweet.get('view_count', 'N/A')}\n"
    )


if __name__ == "__main__":
    monitor_list(on_new_tweet)

效率提升。 以 10 秒间隔分别轮询 50 个账号，每天消耗 50 × 8,640 = 432,000 次请求。将同样 50 个账号放入一个列表，以 10 秒间隔轮询，每天仅需 8,640 次，减少 50 倍。更多方式见优化 API 使用。

/list-tweets 每页最多返回 20 条。如果成员在一次轮询间隔内发布更多推文，可将间隔缩短至 2–3 秒，或使用 next_cursor 分页，直到遇到已见过的 ID。

第三级：监测关键词或话题标签

不必追踪具体账号，也可以使用 order: "latest" 轮询 /search-tweets，按时间获取匹配查询的结果。

import requests
import time

API_KEY = "YOUR_API_KEY"
QUERY = '("your brand" OR @yourbrand) lang:en'
POLL_INTERVAL = 10

URL = "https://api.sorsa.io/v3/search-tweets"
HEADERS = {"ApiKey": API_KEY, "Content-Type": "application/json"}


def monitor_keyword(query, callback, interval=10):
    last_seen_id = None
    print(f"Monitoring: {query} (interval: {interval}s)")

    while True:
        try:
            resp = requests.post(
                URL,
                headers=HEADERS,
                json={"query": query, "order": "latest"},
                timeout=10,
            )
            resp.raise_for_status()
            tweets = resp.json().get("tweets", [])

            if tweets:
                top_id = max(int(t["id"]) for t in tweets)
                if last_seen_id is None:
                    last_seen_id = top_id
                    print(f"Baseline set: {last_seen_id}")
                else:
                    new_tweets = [t for t in tweets if int(t["id"]) > last_seen_id]
                    for tweet in reversed(new_tweets):
                        callback(tweet)
                    if new_tweets:
                        last_seen_id = top_id

        except requests.exceptions.RequestException as e:
            print(f"Error: {e}")
            time.sleep(interval * 2)
            continue

        time.sleep(interval)


monitor_keyword(QUERY, on_new_tweet, interval=10)

查询字符串可使用任意搜索运算符。例如监测品牌的高互动英文提及，并排除转推：

monitor_keyword('"your brand" min_faves:10 lang:en -filter:retweets', on_new_tweet)

将新推文发送到 Slack、Discord 或任意 HTTP 端点

轮询循环负责生成数据，回调函数决定如何处理每条新推文。回调只是一个函数，因此同一个监测器可连接任何支持 HTTP 的目标。

通过 Incoming Webhook 发送到 Slack

import requests

SLACK_WEBHOOK_URL = "https://hooks.slack.com/services/YOUR/SLACK/WEBHOOK"


def send_to_slack(tweet):
    user = tweet["user"]
    text = (
        f"*New tweet from @{user['username']}*\n"
        f"{tweet['full_text']}\n"
        f"Likes: {tweet.get('likes_count', 0)} | "
        f"RTs: {tweet.get('retweet_count', 0)} | "
        f"Views: {tweet.get('view_count', 'N/A')}\n"
        f"https://x.com/{user['username']}/status/{tweet['id']}"
    )
    requests.post(SLACK_WEBHOOK_URL, json={"text": text})


# Plug into any monitor:
monitor_list(send_to_slack)
# or: monitor_keyword("bitcoin lang:en min_faves:50", send_to_slack)

Discord

DISCORD_WEBHOOK_URL = "https://discord.com/api/webhooks/YOUR/WEBHOOK"


def send_to_discord(tweet):
    user = tweet["user"]
    content = (
        f"**@{user['username']}** just tweeted:\n"
        f"{tweet['full_text']}\n"
        f"https://x.com/{user['username']}/status/{tweet['id']}"
    )
    requests.post(DISCORD_WEBHOOK_URL, json={"content": content})

TELEGRAM_BOT_TOKEN = "YOUR_BOT_TOKEN"
TELEGRAM_CHAT_ID = "YOUR_CHAT_ID"


def send_to_telegram(tweet):
    user = tweet["user"]
    text = (
        f"New tweet from @{user['username']}\n\n"
        f"{tweet['full_text']}\n\n"
        f"https://x.com/{user['username']}/status/{tweet['id']}"
    )
    requests.post(
        f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage",
        json={"chat_id": TELEGRAM_CHAT_ID, "text": text},
    )

自定义 HTTP 端点

def send_to_internal_api(tweet):
    requests.post(
        "https://internal.example.com/events/twitter",
        json={
            "tweet_id": tweet["id"],
            "username": tweet["user"]["username"],
            "text": tweet["full_text"],
            "metrics": {
                "likes": tweet.get("likes_count", 0),
                "retweets": tweet.get("retweet_count", 0),
                "views": tweet.get("view_count", 0),
            },
            "url": f"https://x.com/{tweet['user']['username']}/status/{tweet['id']}",
        },
        headers={"Authorization": "Bearer YOUR_INTERNAL_TOKEN"},
        timeout=5,
    )

API 用量估算

下表假设每轮只有一页、固定调度且无需重试。请乘以监测器数量，并加上额外页面和重试次数。示例循环在收到响应后休眠，因此实际周期还包含网络和处理时间。

间隔	每小时请求	每天请求	每月请求（30 天）
1 秒	3,600	86,400	2,592,000
5 秒	720	17,280	518,400
10 秒	360	8,640	259,200
30 秒	120	2,880	86,400
1 分钟	60	1,440	43,200

根据应用可接受的延迟和信息流活跃程度选择间隔。缩短间隔会增加请求数，但不能保证帖子立即出现在搜索结果中。免费 100 次请求足够构建并端到端验证原型。持续监测时，按表中月度用量选套餐：单个循环以 30–60 秒间隔运行适合 Pro（每月 100,000 次），10 秒间隔适合 Enterprise（每月 500,000 次）。这些数字按单个监测器计算，多个并行运行会相应增加总量，请按合计用量选择。完整套餐见价格。

超过标准套餐的速率或用量需求，请联系销售定制配额，或在 Discord 咨询。

生产环境加固

上方示例适用于开发。生产环境需要处理以下五点。

1. 在重启之间持久化 `last_seen_id`

脚本崩溃后如果不知道上次检查点，可能重复处理旧推文、发送重复提醒，或静默跳过缺口。将最后已见 ID 保存在文件、数据库或 Redis 中。

import json
import os

STATE_FILE = "monitor_state.json"


def load_state():
    if os.path.exists(STATE_FILE):
        with open(STATE_FILE) as f:
            return json.load(f).get("last_seen_id")
    return None


def save_state(last_seen_id):
    with open(STATE_FILE, "w") as f:
        json.dump({"last_seen_id": last_seen_id}, f)

将监测器初始化的 last_seen_id = None 替换为 last_seen_id = load_state()。在一轮所有页面处理完成或进入持久化队列后，保存新检查点。页面获取或投递失败时不要推进检查点。重启后应先分页补齐缺口，再接受更新的检查点。

2. 对错误使用指数退避

网络问题、速率限制（HTTP 429）和临时 API 错误不可避免。应逐步增加等待时间并设置上限，而不是立即不断重试。完整说明见错误码。

retry_delay = POLL_INTERVAL
MAX_DELAY = 60

while True:
    try:
        resp = requests.get(URL, headers=HEADERS, timeout=10)
        if resp.status_code == 429:
            print(f"Rate limited. Backing off {retry_delay}s")
            time.sleep(retry_delay)
            retry_delay = min(retry_delay * 2, MAX_DELAY)
            continue
        resp.raise_for_status()
        retry_delay = POLL_INTERVAL  # reset on success
        # process tweets
    except requests.exceptions.RequestException as e:
        print(f"Error: {e}")
        time.sleep(retry_delay)
        retry_delay = min(retry_delay * 2, MAX_DELAY)
        continue

    time.sleep(POLL_INTERVAL)

3. 将轮询与处理分离

不要在轮询循环内同步执行 NLP、数据库写入、外部 API 调用等耗时操作。下游变慢会导致轮询落后。将新推文放入队列，由独立工作进程处理。

from collections import deque
import threading

tweet_queue = deque()


def polling_loop():
    """Fast loop: poll and enqueue. No heavy work here."""
    # Standard polling code, but instead of calling callback(tweet):
    # tweet_queue.append(tweet)
    pass


def processing_worker():
    """Separate thread: dequeue and dispatch."""
    while True:
        if tweet_queue:
            tweet = tweet_queue.popleft()
            send_to_slack(tweet)
            save_to_database(tweet)
        else:
            time.sleep(0.1)


threading.Thread(target=processing_worker, daemon=True).start()
polling_loop()

更大工作负载可将内存 deque 替换为 Redis、RabbitMQ、SQS 或现有技术栈使用的消息系统。

4. 监测监测器本身

记录每轮的时间戳、新推文数、响应时间和错误。如果过去 N 分钟没有成功完成轮询，应发出提醒。静默故障会造成难以察觉的数据缺口。API 运行情况可查看 Sorsa 状态页。

5. 处理边界情况

页面溢出和延迟到达： 持续使用 next_cursor，直到覆盖上次检查点以来的时间段。保留少量重叠并按存储的 ID 去重，避免仅因延迟结果的 ID 较旧而丢弃。前面的首页示例没有实现这种回填。 回调投递： 检查 webhook 响应状态，使用有限重试或持久化队列。Sorsa 请求成功不代表 Slack、Discord 或数据库已接收事件。

已删除推文： 如果推文在获取后、回调前被删除，URL 会返回 404，应视为正常情况。
受保护账号： 追踪用户设为私有后，/user-tweets 返回空列表。记录并继续。
置顶推文： /user-tweets 的第一条往往是置顶内容，而非最新内容。不要直接用 tweets[0] 作为最新 ID；应使用 max(int(t["id"]) for t in tweets)（如上方示例），或按 created_at 排序。
转推： 转推的 tweet["retweeted_status"] 有值，可据此决定是否纳入。
回复限制： is_replies_limited 表示作者限制了回复，对某些监测场景有价值。

后续步骤

搜索运算符：通过高级筛选减少关键词监测噪声
追踪提及：支持互动筛选的专用 @提及端点
速率限制：处理 429 错误和请求节奏
分页：结合实时监测回填历史数据
API 参考：/list-tweets、/user-tweets、/search-tweets 及全部端点的完整规范

入门

核心概念

实用指南

支持与资源

轮询监测的工作方式

选择合适的端点

第一级：监测单个账号

Python

JavaScript

第二级：通过一次请求监测多个账号

第 1 步：创建公开 X 列表

第 2 步：轮询列表

第三级：监测关键词或话题标签

将新推文发送到 Slack、Discord 或任意 HTTP 端点

通过 Incoming Webhook 发送到 Slack

Discord

Telegram

自定义 HTTP 端点

API 用量估算

生产环境加固

1. 在重启之间持久化 `last_seen_id`

2. 对错误使用指数退避

3. 将轮询与处理分离

4. 监测监测器本身

5. 处理边界情况

后续步骤

​轮询监测的工作方式

​选择合适的端点

​第一级：监测单个账号

​Python

​JavaScript

​第二级：通过一次请求监测多个账号

​第 1 步：创建公开 X 列表

​第 2 步：轮询列表

​第三级：监测关键词或话题标签

​将新推文发送到 Slack、Discord 或任意 HTTP 端点

​通过 Incoming Webhook 发送到 Slack

​Discord

​Telegram

​自定义 HTTP 端点

​API 用量估算

​生产环境加固

​1. 在重启之间持久化 last_seen_id

​2. 对错误使用指数退避

​3. 将轮询与处理分离

​4. 监测监测器本身

​5. 处理边界情况

​后续步骤

轮询监测的工作方式

选择合适的端点

第一级：监测单个账号

Python

JavaScript

第二级：通过一次请求监测多个账号

第 1 步：创建公开 X 列表

第 2 步：轮询列表

第三级：监测关键词或话题标签

将新推文发送到 Slack、Discord 或任意 HTTP 端点

通过 Incoming Webhook 发送到 Slack

Discord

Telegram

自定义 HTTP 端点

API 用量估算

生产环境加固

1. 在重启之间持久化 `last_seen_id`

2. 对错误使用指数退避

3. 将轮询与处理分离

4. 监测监测器本身

5. 处理边界情况

后续步骤