Twitter webhooks: tweets as they post.

import json, os, time, requests

API = "https://api.cheapxapi.com/twitter/tweet/advanced_search"
HEADERS = {"x-api-key": os.environ["CHEAPXAPI_KEY"]}
WEBHOOK_URL = os.environ["WEBHOOK_URL"]
QUERY = "from:NASA OR from:SpaceX"
INTERVAL = 60   # seconds between polls
MAX_PAGES = 5   # caps what one burst of new tweets can cost
STATE = "seen.json"

def search(cursor=""):
    params = {"query": QUERY, "queryType": "Latest"}
    if cursor:
        params["cursor"] = cursor
    res = requests.get(API, headers=HEADERS, params=params, timeout=75)
    res.raise_for_status()
    return res.json()

def new_tweets(seen):
    fresh, cursor = {}, ""
    for _ in range(MAX_PAGES):
        page = search(cursor)
        tweets = page.get("tweets", [])
        for t in tweets:
            if t["id"] not in seen:
                fresh[t["id"]] = t
        # A page that reaches a known tweet means everything older was seen.
        if any(t["id"] in seen for t in tweets) or not page.get("has_next_page") or not page.get("next_cursor"):
            break
        cursor = page["next_cursor"]
    return sorted(fresh.values(), key=lambda t: int(t["id"]))  # oldest first

def save(seen):
    with open(STATE, "w") as f:
        json.dump(seen[-2000:], f)

if os.path.exists(STATE):
    with open(STATE) as f:
        seen = json.load(f)
else:
    seen = [t["id"] for t in search().get("tweets", [])]
    save(seen)
    print(f"Watching {QUERY!r}, skipped {len(seen)} existing tweets", flush=True)

while True:
    try:
        for t in new_tweets(set(seen)):
            requests.post(WEBHOOK_URL, timeout=10, json={
                "id": t["id"], "url": t["url"], "author": t["author"]["userName"],
                "created_at": t["createdAt"], "text": t["text"],
            }).raise_for_status()
            seen.append(t["id"])
            save(seen)
    except requests.RequestException as err:
        print("Will retry next poll:", err, flush=True)
    time.sleep(INTERVAL)