fetch_x/fetch_x_posts.py

#!/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())