How to Scrape Twitter/X Posts

"""
KnoxProxy python twitter scraper for X (Twitter) posts and public profile
timelines. Uses the public syndication endpoints designed for embedding
tweets and profile widgets, neither of which requires a logged-in session.
"""
import random
import time

import requests

PROXY_HOST = "proxy.knoxproxy.com"
PROXY_PORT = 10000  # residential rotating
PROXY_USER = "your_username"
PROXY_PASS = "your_password"

HEADERS = {
    "User-Agent": (
        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
        "(KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36"
    ),
    "Accept": "application/json",
}


def proxy_dict() -> dict:
    proxy_url = f"http://{PROXY_USER}:{PROXY_PASS}@{PROXY_HOST}:{PROXY_PORT}"
    return {"http": proxy_url, "https": proxy_url}


def fetch_tweet(tweet_id: str, max_retries: int = 3) -> dict:
    """Scrape a single tweet by ID from the public syndication endpoint."""
    url = "https://cdn.syndication.twimg.com/tweet-result"
    params = {"id": tweet_id, "lang": "en"}

    for attempt in range(1, max_retries + 1):
        try:
            resp = requests.get(url, headers=HEADERS, params=params, proxies=proxy_dict(), timeout=20)
            if resp.status_code == 200:
                return parse_tweet(resp.json(), tweet_id)
            if resp.status_code == 429:
                print(f"Attempt {attempt}: rate limited, rotating IP before retry")
            else:
                print(f"Attempt {attempt}: X returned status {resp.status_code}")
        except requests.exceptions.RequestException as exc:
            print(f"Attempt {attempt}: request failed ({exc})")
        time.sleep(random.uniform(4, 8))

    raise RuntimeError(f"Failed to fetch tweet {tweet_id} after {max_retries} attempts")


def parse_tweet(payload: dict, tweet_id: str) -> dict:
    photos = payload.get("photos", [])
    video = payload.get("video")
    return {
        "tweet_id": tweet_id,
        "text": payload.get("text"),
        "author": payload.get("user", {}).get("screen_name"),
        "like_count": payload.get("favorite_count"),
        "retweet_count": payload.get("conversation_count"),
        "media_urls": [p.get("url") for p in photos] + ([video.get("poster")] if video else []),
    }


def fetch_profile_timeline(screen_name: str, max_retries: int = 3) -> list[dict]:
    """Pull a rolling window of recent public posts and media for one profile.

    This is the same widget feed X serves to power embedded timelines on
    external sites, so it stays reachable without a login while X's own
    timeline and search pages sit behind the authentication wall. This is
    the core of a twitter media scraper: photos and a video poster travel
    in the same payload as the post text.
    """
    url = "https://cdn.syndication.twimg.com/timeline/profile"
    params = {"screen_name": screen_name, "lang": "en"}

    for attempt in range(1, max_retries + 1):
        try:
            resp = requests.get(url, headers=HEADERS, params=params, proxies=proxy_dict(), timeout=20)
            if resp.status_code == 200:
                body = resp.json()
                items = body.get("body", {}).get("items", [])
                # NOTE: item keys mirror the tweet-result payload above;
                # re-verify against a live profile if X changes the widget shape.
                return [parse_tweet(item, item.get("id_str", "")) for item in items]
            if resp.status_code == 429:
                print(f"Attempt {attempt}: rate limited on profile timeline, rotating IP")
            else:
                print(f"Attempt {attempt}: X returned status {resp.status_code}")
        except requests.exceptions.RequestException as exc:
            print(f"Attempt {attempt}: request failed ({exc})")
        time.sleep(random.uniform(4, 8))

    raise RuntimeError(f"Failed to fetch profile timeline for {screen_name} after {max_retries} attempts")


if __name__ == "__main__":
    print(fetch_tweet("1743456789012345678"))
    time.sleep(random.uniform(4, 8))

    for tweet in fetch_profile_timeline("examplehandle"):
        print(tweet)
        time.sleep(random.uniform(4, 8))

"""
Append parsed KnoxProxy twitter scraper output to a CSV file for analysis,
skipping tweet IDs that have already been stored. Built for the fetch
functions in twitter_x_scraper.py.
"""
import csv
import os

FIELDNAMES = ["tweet_id", "text", "author", "like_count", "retweet_count", "media_urls"]


def load_existing_ids(path: str) -> set[str]:
    if not os.path.exists(path):
        return set()
    with open(path, newline="", encoding="utf-8") as f:
        return {row["tweet_id"] for row in csv.DictReader(f)}


def append_tweets(path: str, tweets: list[dict]) -> int:
    existing = load_existing_ids(path)
    new_rows = [t for t in tweets if t["tweet_id"] not in existing]

    write_header = not os.path.exists(path)
    with open(path, "a", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=FIELDNAMES)
        if write_header:
            writer.writeheader()
        for row in new_rows:
            row = dict(row)
            row["media_urls"] = ";".join(row.get("media_urls") or [])
            writer.writerow(row)

    return len(new_rows)


if __name__ == "__main__":
    from twitter_x_scraper import fetch_profile_timeline

    tweets = fetch_profile_timeline("examplehandle")
    added = append_tweets("tweets.csv", tweets)
    print(f"Added {added} new rows to tweets.csv")

{
  "tweet_id": "1743456789012345678",
  "text": "shipping a small update to the dashboard today",
  "author": "examplehandle",
  "like_count": 1420,
  "retweet_count": 96,
  "media_urls": []
}
Added 4 new rows to tweets.csv