Build a Twitter/X brand monitor

#!/usr/bin/env bash
# brand-monitor.sh
# Usage: ./brand-monitor.sh "acme corp" OR @acmecorp

set -euo pipefail

QUERY="${1:-acme corp}"
SEEN_FILE="${HOME}/.brand-monitor-seen.txt"
touch "$SEEN_FILE"

# URL-encode the query
ENCODED=$(python3 -c "import urllib.parse,sys; print(urllib.parse.quote(sys.argv[1]))" "$QUERY")

RESPONSE=$(curl -sf \
  "https://api.twitterapis.com/twitter/tweet/advanced_search?query=${ENCODED}&product=Latest" \
  -H "Authorization: Bearer ${TWITTERAPIS_KEY}")

# Extract tweet IDs and text
echo "$RESPONSE" | python3 - <<'PYEOF'
import json, sys

data = json.load(sys.stdin)
seen_file = os.path.expanduser("~/.brand-monitor-seen.txt")

import os
with open(seen_file) as f:
    seen = set(f.read().splitlines())

new_tweets = []
new_ids = []

for tweet in data.get("tweets", []):
    tid = tweet["id"]
    if tid not in seen:
        new_tweets.append(tweet)
        new_ids.append(tid)

if new_tweets:
    print(f"--- {len(new_tweets)} new mention(s) ---")
    for t in new_tweets:
        print(f"@{t['author']['username']}: {t['text'][:120]}")
        print(f"  https://x.com/{t['author']['username']}/status/{t['id']}")
        print()
    with open(seen_file, "a") as f:
        f.write("\n".join(new_ids) + "\n")
else:
    print("No new mentions.")
PYEOF

*/30 * * * * TWITTERAPIS_KEY=your_key /path/to/brand-monitor.sh '"acme corp" OR @acmecorp' >> /var/log/brand-monitor.log 2>&1

#!/usr/bin/env python3
"""
brand_monitor.py  --  TwitterAPIs brand monitor

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

Polls every POLL_INTERVAL_SECONDS. New mentions are printed to stdout
and optionally POSTed to WEBHOOK_URL.
"""

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"

KEYWORDS = [
    '"acme corp" OR @acmecorp -is:retweet lang:en',
    '#acme min_faves:5 -is:retweet',
]

POLL_INTERVAL_SECONDS = 30 * 60   # 30 minutes
SEEN_FILE = Path.home() / ".brand-monitor-seen.json"
WEBHOOK_URL = os.environ.get("WEBHOOK_URL")  # optional

# Cost note: len(KEYWORDS) calls per poll.
# At $0.0008/call and 48 polls/day:
#   2 keywords x 48 = 96 calls/day = $0.077/day

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

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


def load_seen() -> set:
    if SEEN_FILE.exists():
        return set(json.loads(SEEN_FILE.read_text()))
    return set()


def save_seen(seen: set) -> None:
    SEEN_FILE.write_text(json.dumps(list(seen)))


def search(query: str, retries: int = 3) -> list:
    """Fetch the first page of latest tweets matching query."""
    for attempt in range(retries):
        try:
            resp = session.get(
                f"{BASE_URL}/tweet/advanced_search",
                params={"query": query, "product": "Latest"},
                timeout=15,
            )
            if resp.status_code == 429:
                wait = 60 * (2 ** attempt)
                log.warning("Rate limited. Waiting %ss before retry.", wait)
                time.sleep(wait)
                continue
            resp.raise_for_status()
            return resp.json().get("tweets", [])
        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 []


def deliver(tweet: dict) -> None:
    """Print and optionally webhook new mentions."""
    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"New mention from @{handle}:\n{tweet['text']}\n{url}",
            "tweet_id": tweet["id"],
        }
        try:
            requests.post(WEBHOOK_URL, json=payload, timeout=5)
        except requests.RequestException as exc:
            log.warning("Webhook delivery failed: %s", exc)


def poll_once(seen: set) -> set:
    new_ids: list = []
    for query in KEYWORDS:
        tweets = search(query)
        for tweet in tweets:
            tid = tweet["id"]
            if tid not in seen:
                deliver(tweet)
                new_ids.append(tid)
    seen.update(new_ids)
    log.info("Poll complete. %d new mention(s). Seen set size: %d", len(new_ids), len(seen))
    return seen


def main() -> None:
    seen = load_seen()
    log.info("Starting brand monitor. %d keyword group(s). Poll interval: %ds.",
             len(KEYWORDS), POLL_INTERVAL_SECONDS)
    while True:
        seen = poll_once(seen)
        save_seen(seen)
        time.sleep(POLL_INTERVAL_SECONDS)


if __name__ == "__main__":
    main()

pip install requests

// brand-monitor.ts
// Usage:
//   TWITTERAPIS_KEY=your_key bun brand-monitor.ts
//   TWITTERAPIS_KEY=your_key npx tsx brand-monitor.ts

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

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 SEEN_FILE = join(homedir(), ".brand-monitor-seen.json");
const POLL_INTERVAL_MS = 30 * 60 * 1000; // 30 minutes
const WEBHOOK_URL = process.env.WEBHOOK_URL; // optional

// Cost: $0.0008 per call. Two keywords every 30 min = $0.077/day.

const KEYWORDS = [
  '"acme corp" OR @acmecorp -is:retweet lang:en',
  "#acme min_faves:5 -is:retweet",
];

// ---- seen-set helpers ----
function loadSeen(): Set<string> {
  if (existsSync(SEEN_FILE)) {
    return new Set(JSON.parse(readFileSync(SEEN_FILE, "utf8")));
  }
  return new Set();
}

function saveSeen(seen: Set<string>): void {
  writeFileSync(SEEN_FILE, JSON.stringify([...seen]));
}

// ---- API call with backoff ----
interface Tweet {
  id: string;
  text: string;
  author: { username: string; id: string; name: string };
  favorite_count?: number;
}

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

async function search(query: string, attempt = 0): Promise<Tweet[]> {
  const url = new URL(`${BASE_URL}/tweet/advanced_search`);
  url.searchParams.set("query", query);
  url.searchParams.set("product", "Latest");

  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. Waiting ${wait / 1000}s.`);
    await sleep(wait);
    return search(query, attempt + 1);
  }

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

  const data: SearchResponse = await res.json();
  return data.tweets ?? [];
}

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

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

// ---- poll ----
async function pollOnce(seen: Set<string>): Promise<void> {
  let newCount = 0;
  for (const query of KEYWORDS) {
    let tweets: Tweet[];
    try {
      tweets = await search(query);
    } catch (err) {
      console.error(`Search error for query "${query}":`, err);
      continue;
    }
    for (const tweet of tweets) {
      if (!seen.has(tweet.id)) {
        await deliver(tweet);
        seen.add(tweet.id);
        newCount++;
      }
    }
  }
  console.log(
    `[${new Date().toISOString()}] Poll done. ${newCount} new. Seen: ${seen.size}.`
  );
}

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

// ---- main ----
async function main(): Promise<void> {
  const seen = loadSeen();
  console.log(`Brand monitor started. ${KEYWORDS.length} keyword group(s). Interval: ${POLL_INTERVAL_MS / 60_000}min.`);

  while (true) {
    await pollOnce(seen);
    saveSeen(seen);
    await sleep(POLL_INTERVAL_MS);
  }
}

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

npm install node-fetch