リアルタイム監視

特定アカウントの新しいツイートやキーワードへの言及を検出し、Xの最新データをアプリケーションに取り込みます。このガイドでは、Sorsa APIを定期的に呼び出すプル型のポーリングで、ほぼリアルタイムの監視パイプラインを構築します。検出までの時間は、ポーリング間隔、APIの応答時間、対象フィードや検索インデックスに投稿が現れるタイミングで決まります。新しい投稿が必ず次のレスポンスに含まれると考えず、チェックポイント、ページネーション、重複排除を中心に設計してください。

試作は無料： 全エンドポイントを最初の無料100リクエストで使えます。付与は1回限り、カード不要、有効期限なしです。以下の監視処理を立ち上げ、投稿の検出とSlack・Discordへの振り分けを確認してからプランを選べます。ポーリングはリクエストを多く消費するため、後述の表から間隔に合う有料プランを見積もってください。

注：追加の構成パターンと一連の実装例は、ブログのREST APIによるTwitterのリアルタイム監視を参照してください。

ポーリングによる監視の仕組み

ソーシャルプラットフォームのデータ取得には、プッシュ型（ストリーミング、Webhook）とプル型（ポーリング）があります。Sorsaはポーリングを使います。手順は4つです。

定期的に（1〜30秒おきに）エンドポイントを呼び出す。
既知のツイートIDと結果を比較して、新しい投稿を見つける。
新しい投稿を処理する：通知、保存、SlackやDiscordなどへの配信。
繰り返す。

ツイートIDには作成日時が含まれ、Pythonの整数やJavaScriptのBigIntで比較できます。確認済みの最大IDはチェックポイントに使えますが、投稿が遅れて現れたり順不同になったりする場合があります。本番では、取得する時間範囲を重ね、IDで重複を除去してください。障害後の再開には、チェックポイントの保存と読み込みを明示的に実装する必要があります。

適切なエンドポイントを選ぶ

監視対象	エンドポイント	メソッド	選ぶ理由
単一のアカウント	`/user-tweets`	POST	ユーザーのタイムラインから最新投稿を返す
最大5,000アカウントをまとめて	`/list-tweets`	GET	1リクエストでXリストの全メンバーをカバー
キーワードやハッシュタグ	`/search-tweets`	POST	全検索演算子に対応。`order: latest`で時系列順
アカウントへの@メンション	`/mentions`	POST	エンゲージメントフィルター付きのメンション追跡専用機能

レベル1：単一アカウントの監視

最も簡単な例です。/user-tweetsの最初のページを繰り返し取得し、最後に確認したIDより新しいツイートを出力します。最初の取得成功時は基準を設定し、既存の投稿は出力しません。これは開発用の例です。確認の間や停止中に1ページを超える投稿があると、取り逃す可能性があります。

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リストを使います。

レベル2：1リクエストで複数アカウントを監視する

Xリストには最大5,000アカウントをまとめられます。/list-tweetsは全メンバーの最新投稿を1回で返すため、本番の複数アカウント監視の基本パターンです。詳細はリストとコミュニティを参照してください。

ステップ1：公開Xリストを作成する

Xのリストでリストを作成します。
監視対象を追加します（最大5,000件）。
公開に設定します。非公開リストには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)

効率の向上。 50アカウントを個別に10秒間隔で確認すると、1日50 × 8,640 = 432,000リクエストです。同じ50アカウントを1リストにまとめると、1日8,640リクエストで済み、50分の1になります。他のパターンはAPI利用の最適化を参照してください。

/list-tweetsは1ページ最大20ツイートです。1回の間隔中にそれ以上投稿される場合は、間隔を2〜3秒にするか、既知のIDに到達するまでnext_cursorで続きを取得してください。

レベル3：キーワードやハッシュタグの監視

アカウントの代わりに、/search-tweetsをorder: "latest"で呼び出し、条件に一致する結果を時系列順に取得します。

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利用量の計算

以下は、1回に1ページ、固定間隔、再試行なしの想定です。監視処理の数を掛け、追加ページや再試行も加算してください。サンプルはレスポンスの後に待機するため、実際の周期には通信と処理の時間も含まれます。

間隔	1時間のリクエスト数	1日のリクエスト数	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リクエストで、監視の試作から一連の動作確認まで行えます。継続運用では上の月間利用量に合わせてください。1つのループなら30〜60秒間隔はPro（月間100,000件）、10秒間隔はEnterprise（月間500,000件）に収まります。複数の監視を並列実行すると合計も増えるため、全体の量で選びます。詳細は料金を参照してください。

標準プランを超える速度や利用量が必要な場合は、営業にカスタム割り当てをご相談いただくか、Discordでお問い合わせください。

本番運用に向けた対策

上記は開発用の例です。本番では次の5点に対応してください。

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()に置き換えます。1周期の全ページを処理または永続キューに保存した後に、新しいチェックポイントを保存します。ページ取得や配信に失敗したら進めないでください。再起動時は、未取得の範囲をページネーションで埋めてから新しいチェックポイントを採用します。

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を含む全エンドポイントの仕様。

はじめに

基本的な仕組み

活用ガイド

サポートとリソース

ポーリングによる監視の仕組み

適切なエンドポイントを選ぶ

レベル1：単一アカウントの監視

Python

JavaScript

レベル2：1リクエストで複数アカウントを監視する

ステップ1：公開Xリストを作成する

ステップ2：リストを定期取得する

レベル3：キーワードやハッシュタグの監視

新しい投稿をSlack、Discord、任意のHTTP送信先に渡す

Incoming WebhookでSlackへ送信

Discord

Telegram

独自のHTTPエンドポイント

API利用量の計算

本番運用に向けた対策

1. 再起動後も`last_seen_id`を保持する

2. エラー時に指数バックオフを使う

3. 取得と処理を分離する

4. 監視処理自体を監視する

5. 例外的なケースに対応する

次のステップ

​ポーリングによる監視の仕組み

​適切なエンドポイントを選ぶ

​レベル1：単一アカウントの監視

​Python

​JavaScript

​レベル2：1リクエストで複数アカウントを監視する

​ステップ1：公開Xリストを作成する

​ステップ2：リストを定期取得する

​レベル3：キーワードやハッシュタグの監視

​新しい投稿をSlack、Discord、任意のHTTP送信先に渡す

​Incoming WebhookでSlackへ送信

​Discord

​Telegram

​独自のHTTPエンドポイント

​API利用量の計算

​本番運用に向けた対策

​1. 再起動後もlast_seen_idを保持する

​2. エラー時に指数バックオフを使う

​3. 取得と処理を分離する

​4. 監視処理自体を監視する

​5. 例外的なケースに対応する

​次のステップ

ポーリングによる監視の仕組み

適切なエンドポイントを選ぶ

レベル1：単一アカウントの監視

Python

JavaScript

レベル2：1リクエストで複数アカウントを監視する

ステップ1：公開Xリストを作成する

ステップ2：リストを定期取得する

レベル3：キーワードやハッシュタグの監視

新しい投稿をSlack、Discord、任意のHTTP送信先に渡す

Incoming WebhookでSlackへ送信

Discord

Telegram

独自のHTTPエンドポイント

API利用量の計算

本番運用に向けた対策

1. 再起動後も`last_seen_id`を保持する

2. エラー時に指数バックオフを使う

3. 取得と処理を分離する

4. 監視処理自体を監視する

5. 例外的なケースに対応する

次のステップ