Download Twitter (X) Media — Bulk API Workflow (Photos + Videos)

import os, requests

HEADERS = {"X-API-Key": os.environ["TWITTERAPI_IO_KEY"]}
BASE = "https://api.twitterapi.io"

def extract_media_urls(tweet: dict) -> list[dict]:
    """Return list of {kind, url, tweet_id, author, index} for each media in tweet."""
    out = []
    author = tweet.get("user", {}).get("userName", "unknown")
    for i, m in enumerate(tweet.get("media", [])):
        if m["type"] == "photo":
            out.append({
                "kind": "photo",
                "url": m["media_url_https"],
                "tweet_id": tweet["id"],
                "author": author,
                "index": i,
                "ext": m["media_url_https"].rsplit(".", 1)[-1].split("?")[0],
            })
        elif m["type"] in ("video", "animated_gif"):
            variants = m.get("video_info", {}).get("variants", [])
            mp4s = [v for v in variants if v["content_type"] == "video/mp4"]
            if mp4s:
                best = max(mp4s, key=lambda v: v.get("bitrate", 0))
                out.append({
                    "kind": m["type"],
                    "url": best["url"],
                    "tweet_id": tweet["id"],
                    "author": author,
                    "index": i,
                    "ext": "mp4",
                })
            else:
                hls = next((v for v in variants
                            if v["content_type"] == "application/x-mpegURL"), None)
                if hls:
                    out.append({"kind": "video_hls", "url": hls["url"],
                                "tweet_id": tweet["id"], "author": author,
                                "index": i, "ext": "mp4"})
    return out

def enum_user_media(handle: str) -> list[dict]:
    urls, cursor = [], None
    while True:
        params = {"userName": handle}
        if cursor: params["cursor"] = cursor
        r = requests.get(f"{BASE}/twitter/user/last_tweets",
                         headers=HEADERS, params=params, timeout=15)
        r.raise_for_status()
        resp = r.json()
        for t in resp.get("tweets", []):
            urls.extend(extract_media_urls(t))
        cursor = resp.get("next_cursor")
        if not cursor: break
    return urls

urls = enum_user_media("your_handle")
print(f"found {len(urls)} media objects")
print(f"photos: {sum(1 for u in urls if u['kind'] == 'photo')}")
print(f"videos: {sum(1 for u in urls if u['kind'] in ('video', 'animated_gif'))}")
print(f"HLS-only: {sum(1 for u in urls if u['kind'] == 'video_hls')}")

import json, time
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path

OUT_DIR = Path("media_download")
OUT_DIR.mkdir(exist_ok=True)
LOG = open("download_log.jsonl", "a")

def fetch_one(item: dict, retries: int = 3) -> dict:
    fname = f"{item['author']}_{item['tweet_id']}_{item['index']}.{item['ext']}"
    dest = OUT_DIR / fname
    if dest.exists():
        return {"item": item, "result": "skip_exists"}
    for attempt in range(retries):
        try:
            r = requests.get(item["url"], stream=True, timeout=30)
            if r.status_code == 404:
                return {"item": item, "result": "404_deleted"}
            if r.status_code == 429:
                time.sleep(2 ** attempt * 5)
                continue
            r.raise_for_status()
            with open(dest, "wb") as f:
                for chunk in r.iter_content(chunk_size=64 * 1024):
                    f.write(chunk)
            return {"item": item, "result": "ok", "bytes": dest.stat().st_size}
        except Exception as e:
            if attempt == retries - 1:
                return {"item": item, "result": f"fail:{e}"}
            time.sleep(2 ** attempt)
    return {"item": item, "result": "exhausted_retries"}

def bulk_download(urls: list[dict], workers: int = 6):
    with ThreadPoolExecutor(max_workers=workers) as pool:
        futures = {pool.submit(fetch_one, u): u for u in urls}
        stats = {"ok": 0, "skip": 0, "404": 0, "fail": 0, "bytes": 0}
        for f in as_completed(futures):
            res = f.result()
            LOG.write(json.dumps(res, default=str) + "\n")
            r = res.get("result", "")
            if r == "ok":
                stats["ok"] += 1
                stats["bytes"] += res.get("bytes", 0)
            elif r.startswith("skip"):
                stats["skip"] += 1
            elif r.startswith("404"):
                stats["404"] += 1
            else:
                stats["fail"] += 1
    LOG.flush()
    return stats

# stats = bulk_download(urls, workers=6)
# print(f"downloaded {stats['ok']} files, {stats['bytes'] / 1024**2:.1f} MB")

import subprocess

def fetch_hls(item: dict) -> dict:
    fname = f"{item['author']}_{item['tweet_id']}_{item['index']}.mp4"
    dest = OUT_DIR / fname
    if dest.exists():
        return {"item": item, "result": "skip_exists"}
    # Option A — yt-dlp
    try:
        subprocess.run(
            ["yt-dlp", "-o", str(dest), "--quiet", "--no-warnings", item["url"]],
            check=True, timeout=300,
        )
        return {"item": item, "result": "ok", "bytes": dest.stat().st_size}
    except subprocess.CalledProcessError as e:
        return {"item": item, "result": f"yt_dlp_fail:{e.returncode}"}
    except subprocess.TimeoutExpired:
        return {"item": item, "result": "timeout"}

# For a mixed batch, route HLS separately (yt-dlp is slower per-file):
hls_items = [u for u in urls if u["kind"] == "video_hls"]
direct_items = [u for u in urls if u["kind"] != "video_hls"]
# direct_items → bulk_download() from Step 2
# hls_items → sequential fetch_hls() (yt-dlp handles internal concurrency)