Add scheduled X post fetcher deployment

fetch_x_posts.py

@@ -0,0 +1,178 @@
+#!/usr/bin/env python3
+"""Fetch recent posts from selected X accounts into a local SQLite database.
+
+Requires an X API bearer token.  See README.md for setup and scheduling.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import sqlite3
+import sys
+import time
+from datetime import UTC, datetime, timedelta
+from pathlib import Path
+from urllib.error import HTTPError, URLError
+from urllib.parse import urlencode
+from urllib.request import Request, urlopen
+
+
+API_URL = "https://api.x.com/2/tweets/search/recent"
+DEFAULT_DB = Path(__file__).with_name("x_posts.db")
+OVERLAP = timedelta(minutes=5)  # avoids gaps around a scheduled run boundary
+
+
+def utc_now() -> datetime:
+    return datetime.now(UTC).replace(microsecond=0)
+
+
+def iso_time(value: datetime) -> str:
+    return value.astimezone(UTC).strftime("%Y-%m-%dT%H:%M:%SZ")
+
+
+def init_db(connection: sqlite3.Connection) -> None:
+    connection.executescript(
+        """
+        CREATE TABLE IF NOT EXISTS posts (
+            id TEXT PRIMARY KEY,
+            username TEXT NOT NULL,
+            created_at TEXT NOT NULL,
+            text TEXT NOT NULL,
+            url TEXT NOT NULL,
+            public_metrics TEXT,
+            raw_json TEXT NOT NULL,
+            fetched_at TEXT NOT NULL
+        );
+        CREATE INDEX IF NOT EXISTS idx_posts_username_created
+            ON posts(username, created_at DESC);
+        CREATE TABLE IF NOT EXISTS state (
+            key TEXT PRIMARY KEY,
+            value TEXT NOT NULL
+        );
+        """
+    )
+
+
+def get_state(connection: sqlite3.Connection, key: str) -> str | None:
+    row = connection.execute("SELECT value FROM state WHERE key = ?", (key,)).fetchone()
+    return row[0] if row else None
+
+
+def set_state(connection: sqlite3.Connection, key: str, value: str) -> None:
+    connection.execute(
+        "INSERT INTO state(key, value) VALUES(?, ?) "
+        "ON CONFLICT(key) DO UPDATE SET value = excluded.value",
+        (key, value),
+    )
+
+
+def request_json(token: str, params: dict[str, str]) -> dict:
+    request = Request(
+        f"{API_URL}?{urlencode(params)}",
+        headers={"Authorization": f"Bearer {token}", "User-Agent": "x-post-fetcher/1.0"},
+    )
+    try:
+        with urlopen(request, timeout=30) as response:
+            return json.load(response)
+    except HTTPError as error:
+        body = error.read().decode("utf-8", errors="replace")
+        raise RuntimeError(f"X API returned HTTP {error.code}: {body}") from error
+    except URLError as error:
+        raise RuntimeError(f"Unable to reach X API: {error.reason}") from error
+
+
+def fetch_account(token: str, username: str, start_time: str) -> list[dict]:
+    """Fetch every matching post in the requested period, following pagination."""
+    posts: list[dict] = []
+    next_token: str | None = None
+    while True:
+        params = {
+            "query": f"from:{username} -is:retweet",
+            "start_time": start_time,
+            "max_results": "100",
+            "sort_order": "recency",
+            "tweet.fields": "created_at,public_metrics,referenced_tweets,lang,entities",
+        }
+        if next_token:
+            params["next_token"] = next_token
+        payload = request_json(token, params)
+        posts.extend(payload.get("data", []))
+        next_token = payload.get("meta", {}).get("next_token")
+        if not next_token:
+            return posts
+        time.sleep(1)  # be conservative with API rate limits
+
+
+def save_posts(connection: sqlite3.Connection, username: str, posts: list[dict], fetched_at: str) -> int:
+    saved = 0
+    for post in posts:
+        cursor = connection.execute(
+            """
+            INSERT OR IGNORE INTO posts
+                (id, username, created_at, text, url, public_metrics, raw_json, fetched_at)
+            VALUES (?, ?, ?, ?, ?, ?, ?, ?)
+            """,
+            (
+                post["id"], username,
+                post.get("created_at", ""), post.get("text", ""),
+                f"https://x.com/{username}/status/{post['id']}",
+                json.dumps(post.get("public_metrics"), ensure_ascii=False),
+                json.dumps(post, ensure_ascii=False), fetched_at,
+            ),
+        )
+        saved += cursor.rowcount
+    return saved
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="Fetch selected X accounts into SQLite.")
+    parser.add_argument("usernames", nargs="+", help="X usernames, with or without @")
+    parser.add_argument("--db", type=Path, default=DEFAULT_DB, help=f"SQLite database (default: {DEFAULT_DB})")
+    parser.add_argument("--initial-lookback-hours", type=int, default=24,
+                        help="How much history to fetch on the first run (max 168; default: 24)")
+    return parser.parse_args()
+
+
+def main() -> int:
+    args = parse_args()
+    token = os.environ.get("X_BEARER_TOKEN")
+    if not token:
+        print("Missing X_BEARER_TOKEN environment variable.", file=sys.stderr)
+        return 2
+    if not 1 <= args.initial_lookback_hours <= 168:
+        print("--initial-lookback-hours must be between 1 and 168.", file=sys.stderr)
+        return 2
+
+    usernames = list(dict.fromkeys(name.lstrip("@").strip() for name in args.usernames))
+    if any(not name for name in usernames):
+        print("Usernames cannot be empty.", file=sys.stderr)
+        return 2
+    args.db.parent.mkdir(parents=True, exist_ok=True)
+    now = utc_now()
+    fetched_at = iso_time(now)
+    with sqlite3.connect(args.db) as connection:
+        init_db(connection)
+        last_success = get_state(connection, "last_success_at")
+        if last_success:
+            start = datetime.fromisoformat(last_success.replace("Z", "+00:00")) - OVERLAP
+        else:
+            start = now - timedelta(hours=args.initial_lookback_hours)
+        # The recent-search endpoint cannot look back further than seven days.
+        start = max(start, now - timedelta(days=7))
+        start_time = iso_time(start)
+        total = 0
+        for username in usernames:
+            posts = fetch_account(token, username, start_time)
+            saved = save_posts(connection, username, posts, fetched_at)
+            total += saved
+            print(f"@{username}: fetched {len(posts)}, saved {saved} new post(s)")
+        # Only advance after all accounts completed successfully.
+        set_state(connection, "last_success_at", fetched_at)
+    print(f"Done. {total} new post(s) saved to {args.db}.")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())