Stream the latest tweets for a keyword

#!/usr/bin/env bash
# keyword-stream.sh
# Usage: ./keyword-stream.sh "#openai lang:en -is:retweet"

set -euo pipefail

QUERY="${1:-#openai lang:en -is:retweet}"
HWM_FILE="${HOME}/.kwstream-hwm.txt"

# Load high-water mark (empty on first run)
HWM=""
if [[ -f "$HWM_FILE" ]]; then
  HWM=$(cat "$HWM_FILE")
fi

ENCODED=$(python3 -c "import urllib.parse,sys; print(urllib.parse.quote(sys.argv[1]))" "$QUERY")
URL="https://api.twitterapis.com/twitter/tweet/advanced_search?query=${ENCODED}&product=Latest"

RESPONSE=$(curl -sf "$URL" -H "Authorization: Bearer ${TWITTERAPIS_KEY}")

python3 - <<PYEOF
import json, sys

data = json.loads("""${RESPONSE}""")
hwm = "${HWM}"
new_hwm = hwm
new_tweets = []

for tweet in data.get("tweets", []):
    tid = tweet["id"]
    # IDs are numeric strings; compare as integers
    if not hwm or int(tid) > int(hwm):
        new_tweets.append(tweet)
        if not new_hwm or int(tid) > int(new_hwm):
            new_hwm = tid

if new_tweets:
    print(f"--- {len(new_tweets)} new tweet(s) ---")
    for t in sorted(new_tweets, key=lambda x: int(x["id"])):
        print(f"@{t['author']['username']}: {t['text'][:120]}")
        print(f"  https://x.com/{t['author']['username']}/status/{t['id']}")
        print()
    with open("${HWM_FILE}", "w") as f:
        f.write(new_hwm)
else:
    print("No new tweets.")
PYEOF

while true; do
  ./keyword-stream.sh '"#openai" lang:en -is:retweet'
  sleep 30
done

*/5 * * * * TWITTERAPIS_KEY=your_key /path/to/keyword-stream.sh '"#openai" lang:en -is:retweet' >> /var/log/kwstream.log 2>&1

#!/usr/bin/env python3
"""
keyword_stream.py -- TwitterAPIs keyword stream via polling

Usage:
    export TWITTERAPIS_KEY="your_key_here"
    python3 keyword_stream.py

Cost: $0.0008 per call (one call per poll interval, more if paginating catch-up pages).
"""

import os
import json
import time
import logging
import requests
from pathlib import Path

# ---------- config ----------
API_KEY = os.environ["TWITTERAPIS_KEY"]
BASE_URL = "https://api.twitterapis.com/twitter"

# The keyword you want to stream. Supports full Twitter operator syntax.
QUERY = "#openai lang:en -is:retweet"

POLL_INTERVAL_SECONDS = 30        # how often to check for new tweets
MAX_CATCHUP_PAGES = 10            # stop paginating after this many pages
HWM_FILE = Path.home() / ".kwstream-hwm.json"  # stores high-water mark per query

# ---------- setup ----------
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s",
)
log = logging.getLogger("kwstream")

session = requests.Session()
session.headers.update({"Authorization": f"Bearer {API_KEY}"})


# ---------- high-water mark helpers ----------
def load_hwm() -> dict:
    if HWM_FILE.exists():
        return json.loads(HWM_FILE.read_text())
    return {}


def save_hwm(hwm: dict) -> None:
    HWM_FILE.write_text(json.dumps(hwm))


# ---------- API call with backoff ----------
def search_page(query: str, cursor: str = "", retries: int = 4) -> dict:
    """Fetch one page of latest tweets. Returns the full response dict."""
    params: dict = {"query": query, "product": "Latest"}
    if cursor:
        params["cursor"] = cursor

    for attempt in range(retries):
        try:
            resp = session.get(
                f"{BASE_URL}/tweet/advanced_search",
                params=params,
                timeout=15,
            )
            if resp.status_code == 429:
                wait = 60 * (2 ** attempt)
                log.warning("Rate limited. Retrying in %ss.", wait)
                time.sleep(wait)
                continue
            resp.raise_for_status()
            return resp.json()
        except requests.RequestException as exc:
            log.error("Request error (attempt %d/%d): %s", attempt + 1, retries, exc)
            if attempt < retries - 1:
                time.sleep(5 * (2 ** attempt))
    return {"tweets": [], "next_cursor": ""}


# ---------- collect all new tweets ----------
def fetch_new_tweets(query: str, hwm_id: str) -> tuple[list, str]:
    """
    Walk pages newest-first, collecting tweets with ID > hwm_id.
    Returns (new_tweets_oldest_first, new_hwm_id).
    """
    new_tweets: list = []
    new_hwm = hwm_id
    cursor = ""

    for page_num in range(MAX_CATCHUP_PAGES):
        data = search_page(query, cursor=cursor)
        tweets = data.get("tweets", [])

        caught_up = False
        for tweet in tweets:
            tid = tweet["id"]
            if hwm_id and int(tid) <= int(hwm_id):
                # Reached a tweet we already processed
                caught_up = True
                break
            new_tweets.append(tweet)
            if not new_hwm or int(tid) > int(new_hwm):
                new_hwm = tid

        if caught_up or not data.get("next_cursor") or not data.get("tweets"):
            break

        cursor = data["next_cursor"]
        log.debug("Page %d done. Fetching next page.", page_num + 1)

    # Reverse so we process oldest-first
    new_tweets.reverse()
    return new_tweets, new_hwm


# ---------- delivery ----------
WEBHOOK_URL = os.environ.get("WEBHOOK_URL")  # optional


def deliver(tweet: dict) -> None:
    handle = tweet["author"]["username"]
    url = f"https://x.com/{handle}/status/{tweet['id']}"
    log.info("NEW @%s: %s", handle, tweet["text"][:100])
    log.info("    %s", url)

    if WEBHOOK_URL:
        payload = {
            "text": f"@{handle}: {tweet['text']}\n{url}",
            "tweet_id": tweet["id"],
            "author": handle,
        }
        try:
            requests.post(WEBHOOK_URL, json=payload, timeout=5)
        except requests.RequestException as exc:
            log.warning("Webhook failed: %s", exc)


# ---------- poll loop ----------
def poll_once(hwm: dict) -> dict:
    hwm_id = hwm.get(QUERY, "")
    new_tweets, new_hwm_id = fetch_new_tweets(QUERY, hwm_id)

    for tweet in new_tweets:
        deliver(tweet)

    log.info(
        "Poll done. %d new tweet(s). HWM: %s",
        len(new_tweets),
        new_hwm_id or "(none yet)",
    )

    if new_hwm_id:
        hwm[QUERY] = new_hwm_id
    return hwm


def main() -> None:
    hwm = load_hwm()
    log.info(
        "Keyword stream started. Query: %r. Interval: %ds.",
        QUERY,
        POLL_INTERVAL_SECONDS,
    )
    while True:
        hwm = poll_once(hwm)
        save_hwm(hwm)
        time.sleep(POLL_INTERVAL_SECONDS)


if __name__ == "__main__":
    main()

pip install requests

// keyword-stream.ts
// Usage:
//   TWITTERAPIS_KEY=your_key bun keyword-stream.ts
//   TWITTERAPIS_KEY=your_key npx tsx keyword-stream.ts
//
// Cost: $0.0008 per call.

import { writeFileSync, readFileSync, existsSync } from "node:fs";
import { homedir } from "node:os";
import { join } from "node:path";

// ---------- config ----------
const API_KEY = process.env.TWITTERAPIS_KEY;
if (!API_KEY) throw new Error("Set TWITTERAPIS_KEY environment variable.");

const BASE_URL = "https://api.twitterapis.com/twitter";
const QUERY = "#openai lang:en -is:retweet";
const POLL_INTERVAL_MS = 30_000;     // 30 seconds
const MAX_CATCHUP_PAGES = 10;
const HWM_FILE = join(homedir(), ".kwstream-hwm.json");
const WEBHOOK_URL = process.env.WEBHOOK_URL; // optional

// ---------- types ----------
interface Tweet {
  id: string;
  text: string;
  author: { id: string; username: string; name: string };
  favorite_count?: number;
  retweet_count?: number;
}

interface SearchResponse {
  tweets: Tweet[];
  next_cursor: string;
}

// ---------- high-water mark helpers ----------
function loadHwm(): Record<string, string> {
  if (existsSync(HWM_FILE)) {
    return JSON.parse(readFileSync(HWM_FILE, "utf8"));
  }
  return {};
}

function saveHwm(hwm: Record<string, string>): void {
  writeFileSync(HWM_FILE, JSON.stringify(hwm));
}

// ---------- API call with backoff ----------
async function searchPage(
  query: string,
  cursor = "",
  attempt = 0
): Promise<SearchResponse> {
  const url = new URL(`${BASE_URL}/tweet/advanced_search`);
  url.searchParams.set("query", query);
  url.searchParams.set("product", "Latest");
  if (cursor) url.searchParams.set("cursor", cursor);

  const res = await fetch(url, {
    headers: { Authorization: `Bearer ${API_KEY}` },
  });

  if (res.status === 429) {
    const wait = 60_000 * Math.pow(2, attempt);
    console.warn(`Rate limited. Retrying in ${wait / 1000}s.`);
    await sleep(wait);
    return searchPage(query, cursor, attempt + 1);
  }

  if (!res.ok) {
    const body = await res.text();
    throw new Error(`Search failed ${res.status}: ${body}`);
  }

  return res.json() as Promise<SearchResponse>;
}

// ---------- collect all new tweets ----------
async function fetchNewTweets(
  query: string,
  hwmId: string
): Promise<{ tweets: Tweet[]; newHwmId: string }> {
  const collected: Tweet[] = [];
  let newHwmId = hwmId;
  let cursor = "";

  for (let page = 0; page < MAX_CATCHUP_PAGES; page++) {
    const data = await searchPage(query, cursor);
    let caughtUp = false;

    for (const tweet of data.tweets) {
      if (hwmId && BigInt(tweet.id) <= BigInt(hwmId)) {
        caughtUp = true;
        break;
      }
      collected.push(tweet);
      if (!newHwmId || BigInt(tweet.id) > BigInt(newHwmId)) {
        newHwmId = tweet.id;
      }
    }

    if (caughtUp || !data.next_cursor || data.tweets.length === 0) break;
    cursor = data.next_cursor;
    console.log(`Page ${page + 1} done, fetching next.`);
  }

  // Reverse so oldest-first
  collected.reverse();
  return { tweets: collected, newHwmId };
}

// ---------- delivery ----------
async function deliver(tweet: Tweet): Promise<void> {
  const tweetUrl = `https://x.com/${tweet.author.username}/status/${tweet.id}`;
  console.log(`NEW @${tweet.author.username}: ${tweet.text.slice(0, 100)}`);
  console.log(`    ${tweetUrl}`);

  if (WEBHOOK_URL) {
    await fetch(WEBHOOK_URL, {
      method: "POST",
      headers: { "Content-Type": "application/json" },
      body: JSON.stringify({
        text: `@${tweet.author.username}: ${tweet.text}\n${tweetUrl}`,
        tweet_id: tweet.id,
        author: tweet.author.username,
      }),
    }).catch((err) => console.warn("Webhook failed:", err));
  }
}

// ---------- poll loop ----------
async function pollOnce(hwm: Record<string, string>): Promise<void> {
  const hwmId = hwm[QUERY] ?? "";

  let tweets: Tweet[];
  let newHwmId: string;
  try {
    ({ tweets, newHwmId } = await fetchNewTweets(QUERY, hwmId));
  } catch (err) {
    console.error("Fetch error:", err);
    return;
  }

  for (const tweet of tweets) {
    await deliver(tweet);
  }

  console.log(
    `[${new Date().toISOString()}] Poll done. ${tweets.length} new. HWM: ${newHwmId || "(none yet)"}.`
  );

  if (newHwmId) hwm[QUERY] = newHwmId;
}

function sleep(ms: number): Promise<void> {
  return new Promise((r) => setTimeout(r, ms));
}

// ---------- main ----------
async function main(): Promise<void> {
  const hwm = loadHwm();
  console.log(`Keyword stream started. Query: ${JSON.stringify(QUERY)}. Interval: ${POLL_INTERVAL_MS / 1000}s.`);

  while (true) {
    await pollOnce(hwm);
    saveHwm(hwm);
    await sleep(POLL_INTERVAL_MS);
  }
}

main().catch((err) => {
  console.error("Fatal:", err);
  process.exit(1);
});

npm install --save-dev tsx typescript
npx tsx keyword-stream.ts