extract last 5 tweets from a twitter account as guest

extract_tweets-ssr.py

#!/usr/bin/env python3
"""
Extract tweet id, timestamp, and text from an X/Twitter profile timeline
using recent SSR as guest. Output as plaintext, json or rss.

Usage:
    python3 extract_tweets.py elonmusk
    python3 extract_tweets.py cnn
    python3 extract_tweets.py nasa --json
    python3 extract_tweets.py nasa --rss > nasa.xml
    python3 extract_tweets.py nasa --resolve-links   # swap t.co links for real URLs

    # Parse an HTML file you already saved instead of fetching live:
    python3 extract_tweets.py elonmusk --file saved_page.html
"""
import argparse
import datetime
import json
import re
import socket
import sys
import time
import urllib.error
import urllib.request
import email.utils as email_utils
from xml.sax.saxutils import escape

TCO_URL_RE = re.compile(r'https?://t\.co/\w+')
_JS_ESCAPE_RE = re.compile(
    r'\\u[dD][89abAB][0-9a-fA-F]{2}\\u[dD][c-fC-F][0-9a-fA-F]{2}'  # surrogate pair
    r'|\\u[0-9a-fA-F]{4}'                                          # \uXXXX
    r'|\\n|\\t|\\r|\\"|\\\\|\\/'                                   # simple escapes
)

_SIMPLE_ESCAPES = {'\\n': '\n', '\\t': '\t', '\\r': '\r', '\\"': '"', '\\\\': '\\', '\\/': '/'}


def _js_unescape_match(m: re.Match) -> str:
    s = m.group(0)
    if s in _SIMPLE_ESCAPES:
        return _SIMPLE_ESCAPES[s]
    if len(s) == 12:  # \uXXXX\uXXXX surrogate pair -> one astral code point
        high = int(s[2:6], 16)
        low = int(s[8:12], 16)
        return chr(0x10000 + (high - 0xD800) * 0x400 + (low - 0xDC00))
    return chr(int(s[2:6], 16))  # \uXXXX


def unescape_js_string(s: str) -> str:
    return _JS_ESCAPE_RE.sub(_js_unescape_match, s)


def fetch_profile_html(account: str) -> str:
    url = f"https://x.com/{account}"
    req = urllib.request.Request(
        url,
        headers={
            "User-Agent": (
                "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
                "AppleWebKit/537.36 (KHTML, like Gecko) "
                "Chrome/128.0.0.0 Safari/537.36"
            ),
            "Accept-Language": "en-US,en;q=0.9",
        },
    )
    with urllib.request.urlopen(req, timeout=20) as resp:
        return resp.read().decode("utf-8", errors="replace")


def extract_tweets(html: str):
    scripts = re.findall(r'<script[^>]*>(.*?)</script>', html, re.DOTALL)

    candidates = [s for s in scripts if 'tweet_results' in s and 'full_text' in s]

    results = []
    for s in candidates:
        tweet_iter = list(re.finditer(
            r'(?<![a-zA-Z_])tweet_results:\$R\[\d+\]=\{id:"[^"]+",rest_id:"(\d+)"',
            s
        ))
        for i, m in enumerate(tweet_iter):
            rest_id = m.group(1)
            start = m.end()
            end = tweet_iter[i + 1].start() if i + 1 < len(tweet_iter) else len(s)
            chunk = s[start:end]

            cm = re.search(r'created_at_ms:(\d+)', chunk)
            fm = re.search(r'full_text:"((?:[^"\\]|\\.)*)"', chunk)

            created_ms = cm.group(1) if cm else None
            text = fm.group(1) if fm else None
            if text:
                text = unescape_js_string(text)

            if created_ms and text:
                ts = datetime.datetime.fromtimestamp(
                    int(created_ms) / 1000, tz=datetime.timezone.utc
                ).strftime('%Y-%m-%d %H:%M:%S UTC')
                results.append({
                    "tweet_id": rest_id,
                    "timestamp": ts,
                    "created_ms": int(created_ms),
                    "text": text,
                })

    seen = set()
    deduped = []
    for r in results:
        if r["tweet_id"] not in seen:
            seen.add(r["tweet_id"])
            deduped.append(r)
    return deduped


def resolve_tco_url(url: str, timeout: int = 8, debug: bool = False, user_agent: str = None) -> str:
    headers = {
        "User-Agent": user_agent or "TwitterAndroid/10.53.2-release.0 (310532000-r-0)"
    }
    try:
        req = urllib.request.Request(url, method="HEAD", headers=headers)
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            final = resp.geturl()
            if debug:
                print(
                    f"  [ok] {url} -> status {resp.status}, final {final}",
                    file=sys.stderr,
                )
            return final
    except urllib.error.HTTPError as e:
        if e.code == 405:
            req = urllib.request.Request(url, method="GET", headers=headers)
            with urllib.request.urlopen(req, timeout=timeout) as resp:
                final = resp.geturl()
                if debug:
                    print(
                        f"  [ok, via GET after 405] {url} -> status {resp.status}, final {final}",
                        file=sys.stderr,
                    )
                return final
        raise


def resolve_links_in_tweets(tweets: list, debug: bool = False, delay: float = 0.0, user_agent: str = None) -> None:
    cache = {}
    errors = {}
    for t in tweets:
        matches = set(TCO_URL_RE.findall(t["text"]))
        for short_url in matches:
            if short_url not in cache and short_url not in errors:
                try:
                    cache[short_url] = resolve_tco_url(short_url, debug=debug, user_agent=user_agent)
                except Exception as e:
                    errors[short_url] = e
                    if debug:
                        print(f"  [error] {short_url}: {type(e).__name__}: {e}", file=sys.stderr)
                    continue
                if delay:
                    time.sleep(delay)
            resolved = cache.get(short_url)
            if resolved and resolved != short_url:
                t["text"] = t["text"].replace(short_url, resolved)
            elif resolved and debug:
                print(f"  [unchanged] {short_url} resolved to itself (no redirect followed)", file=sys.stderr)

    if errors:
        print(
            f"Warning: could not resolve {len(errors)} link(s); left as t.co "
            "URLs in the output.",
            file=sys.stderr,
        )
        if not debug:
            print("  Re-run with --debug to see per-link status/errors.", file=sys.stderr)


def build_rss(account: str, tweets: list) -> str:
    channel_link = f"https://x.com/{account}"
    now = email_utils.format_datetime(datetime.datetime.now(datetime.timezone.utc))

    items = []
    for t in tweets:
        item_link = f"{channel_link}/status/{t['tweet_id']}"
        pub_date = email_utils.format_datetime(
            datetime.datetime.fromtimestamp(t["created_ms"] / 1000, tz=datetime.timezone.utc)
        )
        # Use the tweet text as both title (truncated) and full description.
        title = t["text"] if len(t["text"]) <= 80 else t["text"][:77] + "..."
        items.append(
            "    <item>\n"
            f"      <title>{escape(title)}</title>\n"
            f"      <link>{escape(item_link)}</link>\n"
            f"      <guid isPermaLink=\"true\">{escape(item_link)}</guid>\n"
            f"      <pubDate>{pub_date}</pubDate>\n"
            f"      <description>{escape(t['text'])}</description>\n"
            "    </item>"
        )

    return (
        '<?xml version="1.0" encoding="UTF-8"?>\n'
        '<rss version="2.0">\n'
        "  <channel>\n"
        f"    <title>@{escape(account)} on X</title>\n"
        f"    <link>{escape(channel_link)}</link>\n"
        f"    <description>Recent posts from @{escape(account)}</description>\n"
        f"    <lastBuildDate>{now}</lastBuildDate>\n"
        + "\n".join(items) +
        "\n  </channel>\n"
        "</rss>\n"
    )


def main():
    parser = argparse.ArgumentParser(
        description="Extract tweet id, timestamp, and text from an X/Twitter profile."
    )
    parser.add_argument(
        "account",
        help="X/Twitter handle to fetch, without the @ (e.g. elonmusk, cnn, nasa)",
    )
    parser.add_argument(
        "--file",
        help="Parse this local HTML file instead of fetching https://x.com/<account> live",
    )
    parser.add_argument(
        "--json",
        action="store_true",
        help="Print results as JSON instead of pipe-delimited text",
    )
    parser.add_argument(
        "--rss",
        action="store_true",
        help="Print results as an RSS 2.0 feed instead of pipe-delimited text",
    )
    parser.add_argument(
        "--resolve-links",
        action="store_true",
        help="Follow t.co redirects (via HEAD, falling back to GET) and swap "
             "in the real destination URL in each tweet's text",
    )
    parser.add_argument(
        "--debug",
        action="store_true",
        help="With --resolve-links, print each link's resolution attempt "
             "(status code, final URL, or error) to stderr",
    )
    parser.add_argument(
        "--delay",
        type=float,
        default=0.3,
        help="Seconds to sleep between t.co resolution requests, to avoid "
             "triggering rate-limiting/throttling on bursts of requests "
             "(default: 0.3)",
    )
    parser.add_argument(
        "--user-agent",
        default=None,
        help="Override the User-Agent sent when resolving t.co links",
    )
    args = parser.parse_args()

    if args.json and args.rss:
        parser.error("--json and --rss are mutually exclusive")

    account = args.account.lstrip("@")

    if args.file:
        with open(args.file, "rb") as f:
            html = f.read().decode("utf-8", errors="replace")
    else:
        try:
            html = fetch_profile_html(account)
        except Exception as e:
            print(f"Failed to fetch https://x.com/{account}: {e}", file=sys.stderr)
            sys.exit(1)

    tweets = extract_tweets(html)

    if not tweets:
        print(
            f"No tweets found for @{account}",
            file=sys.stderr,
        )
        sys.exit(1)

    if args.resolve_links:
        resolve_links_in_tweets(tweets, debug=args.debug, delay=args.delay, user_agent=args.user_agent)

    if args.rss:
        print(build_rss(account, tweets))
    elif args.json:
        print(json.dumps(tweets, indent=2))
    else:
        for t in tweets:
            print(f'{t["tweet_id"]} | {t["timestamp"]} | {t["text"]}')


if __name__ == "__main__":
    main()