diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 92bf479..6cdc1c6 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -36,7 +36,7 @@ jobs: run: ruff format --check . - name: Type check (mypy) - run: mypy tweet_scraper.py + run: mypy tweet_scraper - name: Tests (pytest + coverage) run: pytest --cov=tweet_scraper --cov-report=term-missing diff --git a/CLAUDE.md b/CLAUDE.md index 3b061b9..4a752da 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -7,19 +7,22 @@ hard rules come first because they matter most. A two-part tweet toolkit (monorepo — two independent subprojects): -- **Scraper** — `scraper/` (`tweet_scraper.py`), a Python 3.11+ CLI that pulls - tweets from twitterapi.io (pay-per-use) into a flat CSV. Config schema mirrors - Apify Tweet Scraper V2. +- **Scraper** — `scraper/` (`tweet_scraper/` package), a Python 3.11+ CLI that + pulls tweets from twitterapi.io (pay-per-use) into a flat CSV. Config schema + mirrors Apify Tweet Scraper V2. - **Studio** — `studio/` (Vite + React 18 + Tailwind 4), a browser app that ingests the CSV for tone/sentiment tagging and draft assistance. -Data flow: `scraper/vars.json → scraper/tweet_scraper.py → tweets.csv → Tweet Studio`. +Data flow: `scraper/vars.json → python -m tweet_scraper → tweets.csv → Tweet Studio`. ## Architecture -- `scraper/tweet_scraper.py` — single-module CLI. Key seams: `build_query` - (search syntax), `paged_get` (HTTP + retry + cursor paging), `flatten` - (tweet → CSV row), `main` (CLI orchestration). +- `scraper/tweet_scraper/` — the CLI package (flat public API preserved via + `__init__` re-exports, so `from tweet_scraper import X` still works). Modules: + `config` (vars.json load/validate + date parsing), `query` (`build_query`, + handle extraction), `api` (HTTP client: `paged_get` retry/cursor paging, + per-endpoint fetchers — patch HTTP here, e.g. `tweet_scraper.api.requests`), + `cli` (`flatten` tweet → CSV row, streaming writes, `main`). - `scraper/tests/` — pytest, fully mocked (no network, no API key). `scraper/conftest.py` puts the scraper dir on `sys.path`. - `studio/src/App.jsx` — the entire Studio UI (lexicons, tone patterns, CSV upload). @@ -45,7 +48,7 @@ Run the full loop from `scraper/` and make sure every step passes: 1. `ruff check .` — lint, zero errors 2. `ruff format --check .` — formatting -3. `mypy tweet_scraper.py` — type check +3. `mypy tweet_scraper` — type check 4. `pytest --cov=tweet_scraper` — all tests pass (CI runs the same four steps from `scraper/` — see `.github/workflows/ci.yml`.) @@ -54,7 +57,7 @@ Run the full loop from `scraper/` and make sure every step passes: - When I say **"validate"** or **"ship it"** → run all four Validation steps and report results before doing anything else. -- When I say **"dry run"** → `python tweet_scraper.py --dry-run` (never spends credits). +- When I say **"dry run"** → `python -m tweet_scraper --dry-run` (never spends credits). - When I say **"scrape"** → `--dry-run` first, show me the queries, wait for OK, then run for real. diff --git a/README.md b/README.md index 2ea4429..46c044a 100644 --- a/README.md +++ b/README.md @@ -9,7 +9,7 @@ A two-part toolkit for collecting tweets and turning them into content insights: browser and does tone/sentiment tagging, engagement stats, and draft assistance. ``` -scraper/vars.json ──▶ scraper/tweet_scraper.py ──▶ tweets.csv ──▶ Tweet Studio ──▶ insights +scraper/vars.json ──▶ python -m tweet_scraper ──▶ tweets.csv ──▶ Tweet Studio ──▶ insights ``` The two halves are independent (separate toolchains); work on them from their @@ -42,15 +42,18 @@ pip install -r requirements.txt # runtime dep: requests export TWITTERAPI_IO_KEY="your_key_here" # key comes from the env, never from a file cp vars.example.json vars.json # then edit vars.json for your run -python tweet_scraper.py # reads ./vars.json, writes ./tweets.csv +python -m tweet_scraper # reads ./vars.json, writes ./tweets.csv ``` +(`pip install -e .` also installs a `tweet-scraper` console command equivalent to +`python -m tweet_scraper`.) + Options: ```bash -python tweet_scraper.py --config path/to.json # use a different config -python tweet_scraper.py --out results.csv # change output path -python tweet_scraper.py --dry-run # print the queries, don't call the API (free) +python -m tweet_scraper --config path/to.json # use a different config +python -m tweet_scraper --out results.csv # change output path +python -m tweet_scraper --dry-run # print the queries, don't call the API (free) ``` **Always `--dry-run` first** — it shows the exact queries that would be billed, @@ -112,7 +115,7 @@ pip install -r requirements-dev.txt ruff check . # lint ruff format --check . # formatting -mypy tweet_scraper.py # type check +mypy tweet_scraper # type check pytest --cov=tweet_scraper # tests + coverage ``` @@ -129,12 +132,18 @@ pip install pre-commit && pre-commit install ``` scraper/ # Python CLI scraper - tweet_scraper.py # the CLI + tweet_scraper/ # the CLI package (flat public API via __init__) + __init__.py # re-exports the public API + __main__.py # enables `python -m tweet_scraper` + config.py # vars.json load/validate + date parsing + query.py # build_query + handle extraction + api.py # twitterapi.io HTTP client (paging/retry, fetchers) + cli.py # flatten → CSV, streaming writes, main tests/ # pytest suite (no network / no API key needed) conftest.py # makes tweet_scraper importable from tests requirements.txt # runtime deps requirements-dev.txt # dev/validation deps - pyproject.toml # ruff / mypy / pytest config (tooling only) + pyproject.toml # [project] packaging + ruff / mypy / pytest config vars.example.json # template config — copy to vars.json (git-ignored) tweets.sample.csv # tiny anonymized example output studio/ # Tweet Studio (Vite + React) @@ -146,10 +155,13 @@ CLAUDE.md # AI-assistant rules & validation loop ## Roadmap -- Packaging: add a `[project]` table + console entry point (`tweet-scraper`). -- Split `tweet_scraper.py` into a package as it grows (config / api / query / cli). -- Config schema validation (reject unknown keys / bad types early). -- Move to the `logging` module instead of `print(..., file=sys.stderr)`. +Done: packaging (`[project]` + `tweet-scraper` console entry point), the +`config / query / api / cli` package split, config-schema validation, and the +move to the `logging` module. Possible next steps: + +- Tweet Studio: list virtualization for very large CSVs (the Tweets tab caps the + rendered list and shows a "top N of M" notice today). +- Retry/backoff tuning and configurable rate limits for `paged_get`. ## License diff --git a/scraper/pyproject.toml b/scraper/pyproject.toml index e9a649c..e8f30d7 100644 --- a/scraper/pyproject.toml +++ b/scraper/pyproject.toml @@ -14,7 +14,7 @@ dependencies = ["requests>=2.31.0"] tweet-scraper = "tweet_scraper:main" [tool.setuptools] -py-modules = ["tweet_scraper"] +packages = ["tweet_scraper"] [tool.ruff] line-length = 100 diff --git a/scraper/tests/test_main_streaming.py b/scraper/tests/test_main_streaming.py index 9e91d7b..c13d858 100644 --- a/scraper/tests/test_main_streaming.py +++ b/scraper/tests/test_main_streaming.py @@ -59,7 +59,7 @@ def test_main_writes_scraped_tweets_to_csv(tmp_path): page = FakeResp(200, {"tweets": [_tweet("1"), _tweet("2")], "has_next_page": False}) with ( patch.dict(os.environ, {"TWITTERAPI_IO_KEY": "k"}), - patch("tweet_scraper.requests.get", return_value=page), + patch("tweet_scraper.api.requests.get", return_value=page), ): rc = _run_main(config, out) assert rc == 0 @@ -72,7 +72,7 @@ def test_main_writes_user_timeline_tweets(tmp_path): page = FakeResp(200, {"tweets": [_tweet("1"), _tweet("2")], "has_next_page": False}) with ( patch.dict(os.environ, {"TWITTERAPI_IO_KEY": "k"}), - patch("tweet_scraper.requests.get", return_value=page), + patch("tweet_scraper.api.requests.get", return_value=page), ): rc = _run_main(config, out) assert rc == 0 @@ -87,8 +87,8 @@ def test_main_persists_rows_fetched_before_a_midrun_crash(tmp_path): side_effect = [page1] + [requests.ConnectionError()] * 4 with ( patch.dict(os.environ, {"TWITTERAPI_IO_KEY": "k"}), - patch("tweet_scraper.time.sleep"), - patch("tweet_scraper.requests.get", side_effect=side_effect), + patch("tweet_scraper.api.time.sleep"), + patch("tweet_scraper.api.requests.get", side_effect=side_effect), pytest.raises(requests.ConnectionError), ): _run_main(config, out) @@ -104,7 +104,7 @@ def test_main_streams_replies_after_main_tweets(tmp_path): reply_page = FakeResp(200, {"tweets": [_tweet("99")], "has_next_page": False}) with ( patch.dict(os.environ, {"TWITTERAPI_IO_KEY": "k"}), - patch("tweet_scraper.requests.get", side_effect=[search_page, reply_page]), + patch("tweet_scraper.api.requests.get", side_effect=[search_page, reply_page]), ): rc = _run_main(config, out) assert rc == 0 @@ -119,7 +119,7 @@ def test_main_skips_reply_fetch_when_reply_count_zero(tmp_path): search_page = FakeResp(200, {"tweets": [_tweet("1", replyCount=0)], "has_next_page": False}) with ( patch.dict(os.environ, {"TWITTERAPI_IO_KEY": "k"}), - patch("tweet_scraper.requests.get", side_effect=[search_page]) as mock_get, + patch("tweet_scraper.api.requests.get", side_effect=[search_page]) as mock_get, ): rc = _run_main(config, out) assert rc == 0 diff --git a/scraper/tests/test_paging.py b/scraper/tests/test_paging.py index c5e86ed..e75ea5b 100644 --- a/scraper/tests/test_paging.py +++ b/scraper/tests/test_paging.py @@ -63,7 +63,7 @@ def test_extract_pagination_missing_is_false_empty(): def test_paged_get_single_page(): resp = FakeResp(200, {"tweets": [{"id": "1"}, {"id": "2"}], "has_next_page": False}) - with patch("tweet_scraper.requests.get", return_value=resp): + with patch("tweet_scraper.api.requests.get", return_value=resp): out = list(paged_get("http://x", {}, "key", limit=10)) assert [t["id"] for t in out] == ["1", "2"] @@ -72,7 +72,7 @@ def test_paged_get_respects_limit(): resp = FakeResp( 200, {"tweets": [{"id": "1"}, {"id": "2"}, {"id": "3"}], "has_next_page": False} ) - with patch("tweet_scraper.requests.get", return_value=resp): + with patch("tweet_scraper.api.requests.get", return_value=resp): out = list(paged_get("http://x", {}, "key", limit=2)) assert [t["id"] for t in out] == ["1", "2"] @@ -80,7 +80,7 @@ def test_paged_get_respects_limit(): def test_paged_get_follows_cursor_across_pages(): page1 = FakeResp(200, {"tweets": [{"id": "1"}], "has_next_page": True, "next_cursor": "c1"}) page2 = FakeResp(200, {"tweets": [{"id": "2"}], "has_next_page": False}) - with patch("tweet_scraper.requests.get", side_effect=[page1, page2]) as mock_get: + with patch("tweet_scraper.api.requests.get", side_effect=[page1, page2]) as mock_get: out = list(paged_get("http://x", {"query": "q"}, "key", limit=10)) assert [t["id"] for t in out] == ["1", "2"] # second request must carry the cursor returned by page 1 @@ -91,8 +91,8 @@ def test_paged_get_retries_on_429_then_succeeds(): bad = FakeResp(429, {}) good = FakeResp(200, {"tweets": [{"id": "1"}], "has_next_page": False}) with ( - patch("tweet_scraper.time.sleep"), - patch("tweet_scraper.requests.get", side_effect=[bad, good]), + patch("tweet_scraper.api.time.sleep"), + patch("tweet_scraper.api.requests.get", side_effect=[bad, good]), ): out = list(paged_get("http://x", {}, "key", limit=10)) assert [t["id"] for t in out] == ["1"] @@ -102,8 +102,8 @@ def test_paged_get_retries_on_500_then_succeeds(): bad = FakeResp(500, {}) good = FakeResp(200, {"tweets": [{"id": "1"}], "has_next_page": False}) with ( - patch("tweet_scraper.time.sleep"), - patch("tweet_scraper.requests.get", side_effect=[bad, good]), + patch("tweet_scraper.api.time.sleep"), + patch("tweet_scraper.api.requests.get", side_effect=[bad, good]), ): out = list(paged_get("http://x", {}, "key", limit=10)) assert [t["id"] for t in out] == ["1"] @@ -112,8 +112,8 @@ def test_paged_get_retries_on_500_then_succeeds(): def test_paged_get_retries_on_connection_error(): good = FakeResp(200, {"tweets": [{"id": "1"}], "has_next_page": False}) with ( - patch("tweet_scraper.time.sleep"), - patch("tweet_scraper.requests.get", side_effect=[requests.ConnectionError(), good]), + patch("tweet_scraper.api.time.sleep"), + patch("tweet_scraper.api.requests.get", side_effect=[requests.ConnectionError(), good]), ): out = list(paged_get("http://x", {}, "key", limit=10)) assert [t["id"] for t in out] == ["1"] @@ -121,7 +121,7 @@ def test_paged_get_retries_on_connection_error(): def test_paged_get_stops_when_no_tweets(): resp = FakeResp(200, {"tweets": [], "has_next_page": True, "next_cursor": "c1"}) - with patch("tweet_scraper.requests.get", return_value=resp): + with patch("tweet_scraper.api.requests.get", return_value=resp): out = list(paged_get("http://x", {}, "key", limit=10)) assert out == [] diff --git a/scraper/tests/test_tweet_scraper.py b/scraper/tests/test_tweet_scraper.py index 0df8506..c474875 100644 --- a/scraper/tests/test_tweet_scraper.py +++ b/scraper/tests/test_tweet_scraper.py @@ -66,7 +66,7 @@ def test_tweet_replies_calls_paged_get_with_correct_args(): "author": {"userName": "replier", "name": "Replier"}, "text": "great tweet", } - with patch("tweet_scraper.paged_get", return_value=iter([mock_reply])) as mock_pg: + with patch("tweet_scraper.api.paged_get", return_value=iter([mock_reply])) as mock_pg: results = list(tweet_replies("123", "test_key", 20)) mock_pg.assert_called_once_with( TWEET_REPLIES_URL, @@ -79,7 +79,7 @@ def test_tweet_replies_calls_paged_get_with_correct_args(): def test_tweet_replies_returns_empty_when_no_replies(): - with patch("tweet_scraper.paged_get", return_value=iter([])): + with patch("tweet_scraper.api.paged_get", return_value=iter([])): results = list(tweet_replies("123", "test_key", 20)) assert results == [] @@ -150,7 +150,7 @@ def test_stream_replies_writes_replies_for_tweet(): "author": {"userName": "replier", "name": "Replier"}, "text": "reply text", } - with patch("tweet_scraper.tweet_replies", return_value=iter([mock_reply])): + with patch("tweet_scraper.api.tweet_replies", return_value=iter([mock_reply])): written = _stream_replies(writer, seen, "100", "api_key", 100) assert written == 1 assert writer.rows[0]["id"] == "999" @@ -165,7 +165,7 @@ def test_stream_replies_deduplicates_already_seen_replies(): "author": {"userName": "r", "name": "R"}, "text": "dupe", } - with patch("tweet_scraper.tweet_replies", return_value=iter([mock_reply])): + with patch("tweet_scraper.api.tweet_replies", return_value=iter([mock_reply])): written = _stream_replies(writer, seen, "100", "api_key", 100) assert written == 0 assert writer.rows == [] diff --git a/scraper/tweet_scraper.py b/scraper/tweet_scraper.py deleted file mode 100644 index ab18dc5..0000000 --- a/scraper/tweet_scraper.py +++ /dev/null @@ -1,628 +0,0 @@ -#!/usr/bin/env python3 -""" -X (Twitter) tweet scraper using pay-per-use pricing via twitterapi.io. - -Why twitterapi.io? - The official X API (api.x.com) is subscription-only (Basic / Pro / Enterprise). - twitterapi.io is a third-party X data API that is true pay-per-use - (~$0.15 per 1,000 tweets, no monthly fee). Sign up at https://twitterapi.io - to get an API key. - -Configuration: reads vars.json in the current directory by default. The schema -mirrors the Apify Tweet Scraper V2 input so the same vars.json works here. - -Supported vars.json fields: - searchTerms list of keywords/phrases to search - twitterHandles list of @handles whose timelines to fetch - startUrls list of x.com/twitter.com profile URLs (handles are extracted) - author restrict searches to tweets from this handle (adds `from:`) - start ISO date "YYYY-MM-DD" — earliest tweet date (inclusive) - end ISO date "YYYY-MM-DD" — latest tweet date (inclusive) - maxItems max tweets per query (default 100) - sort "Latest" or "Top" - tweetLanguage e.g. "en" (adds `lang:en`) - onlyVerifiedUsers adds filter:verified - onlyTwitterBlue adds filter:blue_verified - onlyVideo adds filter:videos - onlyImage adds filter:images - onlyQuote adds filter:quote - includeSearchTerms if true, the source term/handle is written to a column - customMapFunction (ignored — JS function, not applicable here) - -Credit-saving notes: - * `start` / `end` become `since:` / `until:` in search queries, so the API - never returns (or charges for) tweets outside the window. - * User-timeline endpoints don't support server-side date filters, but this - script STOPS paging as soon as it sees tweets older than `start` — no - wasted credits on ancient history. - -Setup: - pip install requests - export TWITTERAPI_IO_KEY="your_key_here" - -Usage: - python tweet_scraper.py # reads ./vars.json, writes ./tweets.csv - python tweet_scraper.py --config path/to.json - python tweet_scraper.py --out results.csv - python tweet_scraper.py --dry-run # print built queries, don't call API -""" - -from __future__ import annotations - -import argparse -import csv -import json -import logging -import os -import sys -import time -from collections.abc import Iterator -from datetime import UTC, datetime, timedelta -from pathlib import Path -from typing import Any - -import requests - -log = logging.getLogger("tweet_scraper") - -API_BASE = "https://api.twitterapi.io" -ADVANCED_SEARCH_URL = f"{API_BASE}/twitter/tweet/advanced_search" -USER_LAST_TWEETS_URL = f"{API_BASE}/twitter/user/last_tweets" -TWEET_REPLIES_URL = f"{API_BASE}/twitter/tweet/replies/v2" - -DEFAULTS: dict[str, Any] = { - "author": None, - "customMapFunction": None, - "end": None, - "includeSearchTerms": False, - "maxItems": 100, - "onlyImage": False, - "onlyQuote": False, - "onlyTwitterBlue": False, - "onlyVerifiedUsers": False, - "onlyVideo": False, - "searchTerms": [], - "sort": "Latest", - "start": None, - "startUrls": [], - "tweetLanguage": None, - "twitterHandles": [], - "fetchReplies": False, -} - - -# ---------- date helpers ---------------------------------------------------- - - -def parse_config_date(s: str | None, *, end_of_day: bool = False) -> datetime | None: - """Parse an ISO 'YYYY-MM-DD' from vars.json into a UTC datetime.""" - if not s: - return None - dt = datetime.strptime(s, "%Y-%m-%d").replace(tzinfo=UTC) - if end_of_day: - dt = dt.replace(hour=23, minute=59, second=59) - return dt - - -_TWEET_TIME_FORMATS = ( - "%a %b %d %H:%M:%S %z %Y", # Twitter classic: "Thu Apr 17 14:30:00 +0000 2026" - "%Y-%m-%dT%H:%M:%S.%fZ", - "%Y-%m-%dT%H:%M:%SZ", -) - - -def tweet_created_at(t: dict[str, Any]) -> datetime | None: - s = t.get("createdAt") or t.get("created_at") - if not s: - return None - for fmt in _TWEET_TIME_FORMATS: - try: - return datetime.strptime(s, fmt).astimezone(UTC) - except ValueError: - continue - try: - return datetime.fromisoformat(s.replace("Z", "+00:00")).astimezone(UTC) - except ValueError: - return None - - -# ---------- config ---------------------------------------------------------- - - -# Accepted types per known config key. `type(None)` marks a nullable field. -# bool is intentionally allowed for maxItems too (it is an int subclass) — not -# worth special-casing for a config validator. -_CONFIG_TYPES: dict[str, tuple[type, ...]] = { - "author": (str, type(None)), - "customMapFunction": (str, type(None)), - "end": (str, type(None)), - "fetchReplies": (bool,), - "includeSearchTerms": (bool,), - "maxItems": (int, float), - "onlyImage": (bool,), - "onlyQuote": (bool,), - "onlyTwitterBlue": (bool,), - "onlyVerifiedUsers": (bool,), - "onlyVideo": (bool,), - "searchTerms": (list,), - "sort": (str,), - "start": (str, type(None)), - "startUrls": (list,), - "tweetLanguage": (str, type(None)), - "twitterHandles": (list,), -} - - -def validate_config(cfg: dict[str, Any]) -> None: - """Warn on unknown keys and reject known keys with the wrong type. - - Unknown keys only warn: the config schema mirrors the Apify Tweet Scraper V2 - input (a superset of what this script implements), so an existing Apify - vars.json must still load. A type mismatch on a known key is a hard error - because it would otherwise crash or misbehave deep in the run. - """ - for key in cfg: - if key not in _CONFIG_TYPES: - log.warning("unknown config key '%s' — ignored.", key) - for key, types in _CONFIG_TYPES.items(): - if key in cfg and not isinstance(cfg[key], types): - expected = " or ".join(t.__name__ for t in types) - got = type(cfg[key]).__name__ - sys.exit(f"Config error: '{key}' must be {expected}, got {got}.") - - -def load_config(path: Path) -> dict[str, Any]: - with path.open("r", encoding="utf-8") as f: - cfg = json.load(f) - if not isinstance(cfg, dict): - sys.exit("Config error: the top-level JSON in the config file must be an object.") - validate_config(cfg) - return {**DEFAULTS, **cfg} - - -def handle_from_url(url: str) -> str | None: - """Extract a handle from an x.com / twitter.com profile URL.""" - u = url.rstrip("/") - for domain in ("x.com/", "twitter.com/"): - if domain in u: - tail = u.split(domain, 1)[1] - handle = tail.split("/")[0].split("?")[0].lstrip("@") - return handle or None - return None - - -# ---------- query construction --------------------------------------------- - - -def build_query(term: str, cfg: dict[str, Any]) -> str: - parts: list[str] = [] - if term: - # Multi-word terms become exact-phrase; single tokens stay bare - parts.append(f'"{term}"' if " " in term else term) - if cfg.get("author"): - parts.append(f"from:{str(cfg['author']).lstrip('@')}") - if cfg.get("tweetLanguage"): - parts.append(f"lang:{cfg['tweetLanguage']}") - if cfg.get("onlyVerifiedUsers"): - parts.append("filter:verified") - if cfg.get("onlyTwitterBlue"): - parts.append("filter:blue_verified") - if cfg.get("onlyVideo"): - parts.append("filter:videos") - if cfg.get("onlyImage"): - parts.append("filter:images") - if cfg.get("onlyQuote"): - parts.append("filter:quote") - # Date range — server-side filtering saves credits. - # Twitter's `until:` is exclusive of that day, so add one day to make - # the user's `end` inclusive. - if cfg.get("start"): - parts.append(f"since:{cfg['start']}") - if cfg.get("end"): - try: - end_plus = datetime.strptime(cfg["end"], "%Y-%m-%d") + timedelta(days=1) - parts.append(f"until:{end_plus.strftime('%Y-%m-%d')}") - except ValueError: - parts.append(f"until:{cfg['end']}") - return " ".join(parts) - - -# ---------- HTTP ------------------------------------------------------------ - - -def get_api_key() -> str: - key = os.environ.get("TWITTERAPI_IO_KEY") or os.environ.get("TWITTERAPI_KEY") - if not key: - sys.exit( - "Missing API key. Set TWITTERAPI_IO_KEY in your environment.\n" - "Get one at https://twitterapi.io (pay-per-use, no subscription)." - ) - return key - - -def _extract_tweets(data: dict[str, Any]) -> list[dict[str, Any]]: - """Pull the tweet list out of a twitterapi.io response regardless of shape. - - Shapes observed: - advanced_search: {"tweets": [...], "has_next_page": ..., "next_cursor": ...} - user/last_tweets: {"data": {"tweets": [...], "pin_tweet": {...}}, ...} - """ - t = data.get("tweets") - if isinstance(t, list): - return [x for x in t if isinstance(x, dict)] - inner = data.get("data") - if isinstance(inner, dict): - t = inner.get("tweets") - if isinstance(t, list): - return [x for x in t if isinstance(x, dict)] - if isinstance(inner, list): - return [x for x in inner if isinstance(x, dict)] - return [] - - -def _extract_pagination(data: dict[str, Any]) -> tuple[bool, str]: - """Return (has_next_page, next_cursor), checking both top-level and nested data.""" - has_next = data.get("has_next_page") - cursor = data.get("next_cursor") - inner = data.get("data") - if isinstance(inner, dict): - if has_next is None: - has_next = inner.get("has_next_page") - if not cursor: - cursor = inner.get("next_cursor") - return bool(has_next), cursor or "" - - -def paged_get( - url: str, - params: dict[str, Any], - api_key: str, - limit: int, - max_retries: int = 3, -) -> Iterator[dict[str, Any]]: - headers = {"X-API-Key": api_key} - cursor = "" - yielded = 0 - while yielded < limit: - p = {**params} - if cursor: - p["cursor"] = cursor - - attempt = 0 - while True: - try: - resp = requests.get(url, headers=headers, params=p, timeout=60) - except requests.RequestException: - if attempt >= max_retries: - raise - time.sleep(1.5**attempt) - attempt += 1 - continue - if resp.status_code == 429: - time.sleep(2 + attempt) - attempt += 1 - if attempt > max_retries: - resp.raise_for_status() - continue - if resp.status_code >= 500 and attempt < max_retries: - time.sleep(1.5**attempt) - attempt += 1 - continue - resp.raise_for_status() - break - - data = resp.json() - tweets = _extract_tweets(data) - if not tweets: - return - for t in tweets: - if yielded >= limit: - return - yield t - yielded += 1 - has_next, cursor = _extract_pagination(data) - if not has_next or not cursor: - return - - -def search_tweets( - query: str, cfg: dict[str, Any], key: str, limit: int -) -> Iterator[dict[str, Any]]: - sort_val = str(cfg.get("sort", "Latest")).lower() - query_type = "Latest" if sort_val.startswith("latest") else "Top" - yield from paged_get( - ADVANCED_SEARCH_URL, - {"query": query, "queryType": query_type}, - key, - limit, - ) - - -def user_tweets( - handle: str, - key: str, - limit: int, - since: datetime | None = None, - until: datetime | None = None, -) -> Iterator[dict[str, Any]]: - """Yield a user's tweets, filtering by date range if supplied. - - The user-timeline endpoint has no server-side date filter, so we: - * skip tweets newer than `until` (rare — only if user backdates), - * stop paging completely once we see a tweet older than `since`, - which saves credits on paginating through ancient history. - Timelines are delivered newest-first, so the first tweet older than - `since` means everything after it is also older. - """ - for t in paged_get( - USER_LAST_TWEETS_URL, - {"userName": handle.lstrip("@")}, - key, - limit, - ): - tw_dt = tweet_created_at(t) - if tw_dt is None: - # Date-less tweet: yield it rather than drop silently. - yield t - continue - if until is not None and tw_dt > until: - # Too new — skip but keep scanning (pinned tweets, etc.). - continue - if since is not None and tw_dt < since: - # Older than our window; remaining tweets will be even older. - break - yield t - - -def tweet_replies(tweet_id: str, key: str, limit: int) -> Iterator[dict[str, Any]]: - yield from paged_get( - TWEET_REPLIES_URL, - {"tweetId": tweet_id}, - key, - limit, - ) - - -# ---------- flatten & write ------------------------------------------------ - -CSV_COLUMNS = [ - "id", - "url", - "createdAt", - "authorUsername", - "authorName", - "authorVerified", - "authorIsBlueVerified", - "text", - "lang", - "likeCount", - "retweetCount", - "replyCount", - "quoteCount", - "viewCount", - "isRetweet", - "isQuote", - "isReply", - "hasMedia", - "mediaUrls", - "sourceTerm", - "parentTweetId", -] - - -def flatten(t: dict[str, Any], source_term: str = "", parent_id: str = "") -> dict[str, Any]: - author = t.get("author") - if not isinstance(author, dict): - author = {} - ext_raw = t.get("extendedEntities") - ext: dict[str, Any] = ext_raw if isinstance(ext_raw, dict) else {} - ent_raw = t.get("entities") - ent: dict[str, Any] = ent_raw if isinstance(ent_raw, dict) else {} - media_items = ext.get("media") or ent.get("media") or [] - if not isinstance(media_items, list): - media_items = [] - media_urls: list[str] = [] - for m in media_items: - if isinstance(m, dict): - url = m.get("media_url_https") or m.get("media_url") - if url: - media_urls.append(str(url)) - handle = author.get("userName") or author.get("screen_name") or "" - tid = str(t.get("id") or t.get("id_str") or "") - return { - "id": tid, - "url": f"https://x.com/{handle}/status/{tid}" if handle and tid else t.get("url", ""), - "createdAt": t.get("createdAt") or t.get("created_at", ""), - "authorUsername": handle, - "authorName": author.get("name", ""), - "authorVerified": bool(author.get("verified", False)), - "authorIsBlueVerified": bool(author.get("isBlueVerified", False)), - "text": t.get("text") or t.get("full_text", ""), - "lang": t.get("lang", ""), - "likeCount": t.get("likeCount") or t.get("favorite_count", 0), - "retweetCount": t.get("retweetCount") or t.get("retweet_count", 0), - "replyCount": t.get("replyCount") or t.get("reply_count", 0), - "quoteCount": t.get("quoteCount") or t.get("quote_count", 0), - "viewCount": t.get("viewCount") or t.get("view_count", 0), - "isRetweet": bool(t.get("isRetweet") or t.get("retweeted_status")), - "isQuote": bool(t.get("isQuote") or t.get("quoted_status_id")), - "isReply": bool(t.get("isReply") or t.get("in_reply_to_status_id")), - "hasMedia": bool(media_urls), - "mediaUrls": ";".join(media_urls), - "sourceTerm": source_term, - "parentTweetId": parent_id, - } - - -def _write_new(writer: csv.DictWriter[str], seen: set[str], row: dict[str, Any]) -> bool: - """Write a flattened row to the CSV if its id is new and non-empty. - - Returns True if the row was written. Streaming each row as it is fetched - (instead of buffering everything until the end) means a mid-run failure - still leaves the already-paid-for tweets on disk. - """ - rid = row["id"] - if rid and rid not in seen: - seen.add(rid) - writer.writerow(row) - return True - return False - - -def _stream_replies( - writer: csv.DictWriter[str], - seen: set[str], - tweet_id: str, - key: str, - limit: int, -) -> int: - """Fetch a tweet's replies and stream the new (unseen) ones to the CSV. - - Returns the number of reply rows written. - """ - written = 0 - for reply in tweet_replies(tweet_id, key, limit): - if _write_new(writer, seen, flatten(reply, parent_id=tweet_id)): - written += 1 - return written - - -def _stream_query( - writer: csv.DictWriter[str], - seen: set[str], - tweets: Iterator[dict[str, Any]], - *, - source_term: str, - fetch_replies: bool, - reply_targets: list[str], -) -> int: - """Flatten, dedup, and stream a batch of tweets to the CSV. - - Each new tweet whose ``replyCount`` is positive has its id queued in - ``reply_targets`` (when ``fetch_replies`` is set) for a later reply pass. - Returns the number of new rows written. - """ - count = 0 - for tw in tweets: - row = flatten(tw, source_term=source_term) - if _write_new(writer, seen, row): - count += 1 - if fetch_replies and int(row.get("replyCount") or 0) > 0: - reply_targets.append(row["id"]) - return count - - -# ---------- main ------------------------------------------------------------ - - -def main() -> int: - parser = argparse.ArgumentParser(description="X tweet scraper (twitterapi.io, pay-per-use)") - parser.add_argument( - "--config", default="vars.json", type=Path, help="Path to vars.json (default: ./vars.json)" - ) - parser.add_argument( - "--out", default="tweets.csv", type=Path, help="Output CSV path (default: ./tweets.csv)" - ) - parser.add_argument( - "--dry-run", - action="store_true", - help="Print the queries that would be issued; don't call the API", - ) - args = parser.parse_args() - - # Progress/warnings go to stderr via logging; the dry-run report stays on - # stdout (it is the program's actual output, not progress chatter). - logging.basicConfig(level=logging.INFO, format="%(message)s", stream=sys.stderr) - - if not args.config.exists(): - sys.exit(f"Config not found: {args.config}") - - cfg = load_config(args.config) - max_items = int(cfg.get("maxItems") or 100) - since_dt = parse_config_date(cfg.get("start")) - until_dt = parse_config_date(cfg.get("end"), end_of_day=True) - - if cfg.get("customMapFunction"): - log.warning("customMapFunction is a JS Apify feature — ignored in this script.") - if since_dt and until_dt and since_dt > until_dt: - sys.exit(f"Invalid range: start ({cfg['start']}) is after end ({cfg['end']}).") - - # Collect queries - search_queries: list[tuple[str, str]] = [] # (source_term, query) - for term in cfg.get("searchTerms") or []: - search_queries.append((term, build_query(term, cfg))) - - handles: set[str] = {h.lstrip("@") for h in (cfg.get("twitterHandles") or []) if h} - for u in cfg.get("startUrls") or []: - h = handle_from_url(u) - if h: - handles.add(h) - - if args.dry_run: - print(f"Config: {args.config}") - print(f"Max items per query: {max_items}") - print(f"Sort: {cfg.get('sort')}") - print(f"Date range: {cfg.get('start') or '(any)'} -> {cfg.get('end') or '(any)'}") - print("Search queries:") - for term, q in search_queries: - print(f" [{term}] -> {q}") - print("User timelines (date-filtered client-side, paging stops at `start`):") - for h in sorted(handles): - print(f" @{h}") - return 0 - - key = get_api_key() - seen: set[str] = set() - reply_targets: list[str] = [] # ids of written tweets to fetch replies for - total = 0 - fetch_replies = bool(cfg.get("fetchReplies")) - include_terms = bool(cfg.get("includeSearchTerms")) - - args.out.parent.mkdir(parents=True, exist_ok=True) - # Stream rows to the CSV as they arrive so a mid-run failure (network error, - # rate-limit exhaustion, Ctrl-C) still leaves the already-paid-for tweets on - # disk — the `with` block flushes and closes the file as the exception unwinds. - with args.out.open("w", newline="", encoding="utf-8") as f: - writer = csv.DictWriter(f, fieldnames=CSV_COLUMNS) - writer.writeheader() - - for term, q in search_queries: - log.info("[search] %s", q) - count = _stream_query( - writer, - seen, - search_tweets(q, cfg, key, max_items), - source_term=term if include_terms else "", - fetch_replies=fetch_replies, - reply_targets=reply_targets, - ) - total += count - log.info(" -> %d new tweets (total: %d)", count, total) - - for h in sorted(handles): - log.info("[user ] @%s", h) - count = _stream_query( - writer, - seen, - user_tweets(h, key, max_items, since=since_dt, until=until_dt), - source_term=f"@{h}" if include_terms else "", - fetch_replies=fetch_replies, - reply_targets=reply_targets, - ) - total += count - log.info(" -> %d new tweets (total: %d)", count, total) - - if reply_targets: - log.info("[replies] fetching replies for tweets with replyCount > 0...") - for tid in reply_targets: - written = _stream_replies(writer, seen, tid, key, max_items) - total += written - if written: - log.info(" [replies] tweet %s -> %d replies (total: %d)", tid, written, total) - - log.info("Wrote %d unique tweets -> %s", total, args.out) - return 0 - - -if __name__ == "__main__": - sys.exit(main()) diff --git a/scraper/tweet_scraper/__init__.py b/scraper/tweet_scraper/__init__.py new file mode 100644 index 0000000..033aabc --- /dev/null +++ b/scraper/tweet_scraper/__init__.py @@ -0,0 +1,83 @@ +"""X (Twitter) tweet scraper using pay-per-use pricing via twitterapi.io. + +Why twitterapi.io? + The official X API (api.x.com) is subscription-only (Basic / Pro / Enterprise). + twitterapi.io is a third-party X data API that is true pay-per-use + (~$0.15 per 1,000 tweets, no monthly fee). Sign up at https://twitterapi.io + to get an API key. + +This package keeps a flat public API — ``from tweet_scraper import build_query`` +still works — while the implementation is split across submodules: + + config vars.json loading, validation, and date parsing + query search-query construction and handle extraction + api twitterapi.io HTTP client (auth, paging/retry, fetchers) + cli flatten → CSV, streaming writes, and the ``main`` entry point + +Configuration mirrors the Apify Tweet Scraper V2 input so the same vars.json +works here. See README.md for setup, fields, and usage. +""" + +from __future__ import annotations + +from . import api, cli, config, query +from .api import ( + ADVANCED_SEARCH_URL, + API_BASE, + TWEET_REPLIES_URL, + USER_LAST_TWEETS_URL, + _extract_pagination, + _extract_tweets, + get_api_key, + paged_get, + search_tweets, + tweet_replies, + user_tweets, +) +from .cli import ( + CSV_COLUMNS, + _stream_query, + _stream_replies, + _write_new, + flatten, + main, +) +from .config import ( + DEFAULTS, + load_config, + parse_config_date, + tweet_created_at, + validate_config, +) +from .query import build_query, handle_from_url + +__all__ = [ + "ADVANCED_SEARCH_URL", + "API_BASE", + "CSV_COLUMNS", + "DEFAULTS", + "TWEET_REPLIES_URL", + "USER_LAST_TWEETS_URL", + "_extract_pagination", + "_extract_tweets", + "_stream_query", + "_stream_replies", + "_write_new", + "api", + "build_query", + "cli", + "config", + "flatten", + "get_api_key", + "handle_from_url", + "load_config", + "main", + "paged_get", + "parse_config_date", + "query", + "search_tweets", + "tweet_created_at", + "tweet_replies", + "user_tweets", + "validate_config", +] diff --git a/scraper/tweet_scraper/__main__.py b/scraper/tweet_scraper/__main__.py new file mode 100644 index 0000000..c49b335 --- /dev/null +++ b/scraper/tweet_scraper/__main__.py @@ -0,0 +1,8 @@ +"""Enable `python -m tweet_scraper` (equivalent to the `tweet-scraper` console script).""" + +import sys + +from .cli import main + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scraper/tweet_scraper/api.py b/scraper/tweet_scraper/api.py new file mode 100644 index 0000000..2339bc3 --- /dev/null +++ b/scraper/tweet_scraper/api.py @@ -0,0 +1,172 @@ +"""twitterapi.io HTTP client: auth, paging/retry, and the per-endpoint fetchers.""" + +from __future__ import annotations + +import os +import sys +import time +from collections.abc import Iterator +from datetime import datetime +from typing import Any + +import requests + +from .config import tweet_created_at + +API_BASE = "https://api.twitterapi.io" +ADVANCED_SEARCH_URL = f"{API_BASE}/twitter/tweet/advanced_search" +USER_LAST_TWEETS_URL = f"{API_BASE}/twitter/user/last_tweets" +TWEET_REPLIES_URL = f"{API_BASE}/twitter/tweet/replies/v2" + + +def get_api_key() -> str: + key = os.environ.get("TWITTERAPI_IO_KEY") or os.environ.get("TWITTERAPI_KEY") + if not key: + sys.exit( + "Missing API key. Set TWITTERAPI_IO_KEY in your environment.\n" + "Get one at https://twitterapi.io (pay-per-use, no subscription)." + ) + return key + + +def _extract_tweets(data: dict[str, Any]) -> list[dict[str, Any]]: + """Pull the tweet list out of a twitterapi.io response regardless of shape. + + Shapes observed: + advanced_search: {"tweets": [...], "has_next_page": ..., "next_cursor": ...} + user/last_tweets: {"data": {"tweets": [...], "pin_tweet": {...}}, ...} + """ + t = data.get("tweets") + if isinstance(t, list): + return [x for x in t if isinstance(x, dict)] + inner = data.get("data") + if isinstance(inner, dict): + t = inner.get("tweets") + if isinstance(t, list): + return [x for x in t if isinstance(x, dict)] + if isinstance(inner, list): + return [x for x in inner if isinstance(x, dict)] + return [] + + +def _extract_pagination(data: dict[str, Any]) -> tuple[bool, str]: + """Return (has_next_page, next_cursor), checking both top-level and nested data.""" + has_next = data.get("has_next_page") + cursor = data.get("next_cursor") + inner = data.get("data") + if isinstance(inner, dict): + if has_next is None: + has_next = inner.get("has_next_page") + if not cursor: + cursor = inner.get("next_cursor") + return bool(has_next), cursor or "" + + +def paged_get( + url: str, + params: dict[str, Any], + api_key: str, + limit: int, + max_retries: int = 3, +) -> Iterator[dict[str, Any]]: + headers = {"X-API-Key": api_key} + cursor = "" + yielded = 0 + while yielded < limit: + p = {**params} + if cursor: + p["cursor"] = cursor + + attempt = 0 + while True: + try: + resp = requests.get(url, headers=headers, params=p, timeout=60) + except requests.RequestException: + if attempt >= max_retries: + raise + time.sleep(1.5**attempt) + attempt += 1 + continue + if resp.status_code == 429: + time.sleep(2 + attempt) + attempt += 1 + if attempt > max_retries: + resp.raise_for_status() + continue + if resp.status_code >= 500 and attempt < max_retries: + time.sleep(1.5**attempt) + attempt += 1 + continue + resp.raise_for_status() + break + + data = resp.json() + tweets = _extract_tweets(data) + if not tweets: + return + for t in tweets: + if yielded >= limit: + return + yield t + yielded += 1 + has_next, cursor = _extract_pagination(data) + if not has_next or not cursor: + return + + +def search_tweets( + query: str, cfg: dict[str, Any], key: str, limit: int +) -> Iterator[dict[str, Any]]: + sort_val = str(cfg.get("sort", "Latest")).lower() + query_type = "Latest" if sort_val.startswith("latest") else "Top" + yield from paged_get( + ADVANCED_SEARCH_URL, + {"query": query, "queryType": query_type}, + key, + limit, + ) + + +def user_tweets( + handle: str, + key: str, + limit: int, + since: datetime | None = None, + until: datetime | None = None, +) -> Iterator[dict[str, Any]]: + """Yield a user's tweets, filtering by date range if supplied. + + The user-timeline endpoint has no server-side date filter, so we: + * skip tweets newer than `until` (rare — only if user backdates), + * stop paging completely once we see a tweet older than `since`, + which saves credits on paginating through ancient history. + Timelines are delivered newest-first, so the first tweet older than + `since` means everything after it is also older. + """ + for t in paged_get( + USER_LAST_TWEETS_URL, + {"userName": handle.lstrip("@")}, + key, + limit, + ): + tw_dt = tweet_created_at(t) + if tw_dt is None: + # Date-less tweet: yield it rather than drop silently. + yield t + continue + if until is not None and tw_dt > until: + # Too new — skip but keep scanning (pinned tweets, etc.). + continue + if since is not None and tw_dt < since: + # Older than our window; remaining tweets will be even older. + break + yield t + + +def tweet_replies(tweet_id: str, key: str, limit: int) -> Iterator[dict[str, Any]]: + yield from paged_get( + TWEET_REPLIES_URL, + {"tweetId": tweet_id}, + key, + limit, + ) diff --git a/scraper/tweet_scraper/cli.py b/scraper/tweet_scraper/cli.py new file mode 100644 index 0000000..1bd41fd --- /dev/null +++ b/scraper/tweet_scraper/cli.py @@ -0,0 +1,258 @@ +"""Flatten tweets to CSV rows, stream them to disk, and orchestrate the CLI.""" + +from __future__ import annotations + +import argparse +import csv +import logging +import sys +from collections.abc import Iterator +from pathlib import Path +from typing import Any + +from . import api +from .config import load_config, parse_config_date +from .query import build_query, handle_from_url + +log = logging.getLogger("tweet_scraper") + + +# ---------- flatten & write ------------------------------------------------ + +CSV_COLUMNS = [ + "id", + "url", + "createdAt", + "authorUsername", + "authorName", + "authorVerified", + "authorIsBlueVerified", + "text", + "lang", + "likeCount", + "retweetCount", + "replyCount", + "quoteCount", + "viewCount", + "isRetweet", + "isQuote", + "isReply", + "hasMedia", + "mediaUrls", + "sourceTerm", + "parentTweetId", +] + + +def flatten(t: dict[str, Any], source_term: str = "", parent_id: str = "") -> dict[str, Any]: + author = t.get("author") + if not isinstance(author, dict): + author = {} + ext_raw = t.get("extendedEntities") + ext: dict[str, Any] = ext_raw if isinstance(ext_raw, dict) else {} + ent_raw = t.get("entities") + ent: dict[str, Any] = ent_raw if isinstance(ent_raw, dict) else {} + media_items = ext.get("media") or ent.get("media") or [] + if not isinstance(media_items, list): + media_items = [] + media_urls: list[str] = [] + for m in media_items: + if isinstance(m, dict): + url = m.get("media_url_https") or m.get("media_url") + if url: + media_urls.append(str(url)) + handle = author.get("userName") or author.get("screen_name") or "" + tid = str(t.get("id") or t.get("id_str") or "") + return { + "id": tid, + "url": f"https://x.com/{handle}/status/{tid}" if handle and tid else t.get("url", ""), + "createdAt": t.get("createdAt") or t.get("created_at", ""), + "authorUsername": handle, + "authorName": author.get("name", ""), + "authorVerified": bool(author.get("verified", False)), + "authorIsBlueVerified": bool(author.get("isBlueVerified", False)), + "text": t.get("text") or t.get("full_text", ""), + "lang": t.get("lang", ""), + "likeCount": t.get("likeCount") or t.get("favorite_count", 0), + "retweetCount": t.get("retweetCount") or t.get("retweet_count", 0), + "replyCount": t.get("replyCount") or t.get("reply_count", 0), + "quoteCount": t.get("quoteCount") or t.get("quote_count", 0), + "viewCount": t.get("viewCount") or t.get("view_count", 0), + "isRetweet": bool(t.get("isRetweet") or t.get("retweeted_status")), + "isQuote": bool(t.get("isQuote") or t.get("quoted_status_id")), + "isReply": bool(t.get("isReply") or t.get("in_reply_to_status_id")), + "hasMedia": bool(media_urls), + "mediaUrls": ";".join(media_urls), + "sourceTerm": source_term, + "parentTweetId": parent_id, + } + + +def _write_new(writer: csv.DictWriter[str], seen: set[str], row: dict[str, Any]) -> bool: + """Write a flattened row to the CSV if its id is new and non-empty. + + Returns True if the row was written. Streaming each row as it is fetched + (instead of buffering everything until the end) means a mid-run failure + still leaves the already-paid-for tweets on disk. + """ + rid = row["id"] + if rid and rid not in seen: + seen.add(rid) + writer.writerow(row) + return True + return False + + +def _stream_replies( + writer: csv.DictWriter[str], + seen: set[str], + tweet_id: str, + key: str, + limit: int, +) -> int: + """Fetch a tweet's replies and stream the new (unseen) ones to the CSV. + + Returns the number of reply rows written. + """ + written = 0 + for reply in api.tweet_replies(tweet_id, key, limit): + if _write_new(writer, seen, flatten(reply, parent_id=tweet_id)): + written += 1 + return written + + +def _stream_query( + writer: csv.DictWriter[str], + seen: set[str], + tweets: Iterator[dict[str, Any]], + *, + source_term: str, + fetch_replies: bool, + reply_targets: list[str], +) -> int: + """Flatten, dedup, and stream a batch of tweets to the CSV. + + Each new tweet whose ``replyCount`` is positive has its id queued in + ``reply_targets`` (when ``fetch_replies`` is set) for a later reply pass. + Returns the number of new rows written. + """ + count = 0 + for tw in tweets: + row = flatten(tw, source_term=source_term) + if _write_new(writer, seen, row): + count += 1 + if fetch_replies and int(row.get("replyCount") or 0) > 0: + reply_targets.append(row["id"]) + return count + + +# ---------- main ------------------------------------------------------------ + + +def main() -> int: + parser = argparse.ArgumentParser(description="X tweet scraper (twitterapi.io, pay-per-use)") + parser.add_argument( + "--config", default="vars.json", type=Path, help="Path to vars.json (default: ./vars.json)" + ) + parser.add_argument( + "--out", default="tweets.csv", type=Path, help="Output CSV path (default: ./tweets.csv)" + ) + parser.add_argument( + "--dry-run", + action="store_true", + help="Print the queries that would be issued; don't call the API", + ) + args = parser.parse_args() + + # Progress/warnings go to stderr via logging; the dry-run report stays on + # stdout (it is the program's actual output, not progress chatter). + logging.basicConfig(level=logging.INFO, format="%(message)s", stream=sys.stderr) + + if not args.config.exists(): + sys.exit(f"Config not found: {args.config}") + + cfg = load_config(args.config) + max_items = int(cfg.get("maxItems") or 100) + since_dt = parse_config_date(cfg.get("start")) + until_dt = parse_config_date(cfg.get("end"), end_of_day=True) + + if cfg.get("customMapFunction"): + log.warning("customMapFunction is a JS Apify feature — ignored in this script.") + if since_dt and until_dt and since_dt > until_dt: + sys.exit(f"Invalid range: start ({cfg['start']}) is after end ({cfg['end']}).") + + # Collect queries + search_queries: list[tuple[str, str]] = [] # (source_term, query) + for term in cfg.get("searchTerms") or []: + search_queries.append((term, build_query(term, cfg))) + + handles: set[str] = {h.lstrip("@") for h in (cfg.get("twitterHandles") or []) if h} + for u in cfg.get("startUrls") or []: + h = handle_from_url(u) + if h: + handles.add(h) + + if args.dry_run: + print(f"Config: {args.config}") + print(f"Max items per query: {max_items}") + print(f"Sort: {cfg.get('sort')}") + print(f"Date range: {cfg.get('start') or '(any)'} -> {cfg.get('end') or '(any)'}") + print("Search queries:") + for term, q in search_queries: + print(f" [{term}] -> {q}") + print("User timelines (date-filtered client-side, paging stops at `start`):") + for h in sorted(handles): + print(f" @{h}") + return 0 + + key = api.get_api_key() + seen: set[str] = set() + reply_targets: list[str] = [] # ids of written tweets to fetch replies for + total = 0 + fetch_replies = bool(cfg.get("fetchReplies")) + include_terms = bool(cfg.get("includeSearchTerms")) + + args.out.parent.mkdir(parents=True, exist_ok=True) + # Stream rows to the CSV as they arrive so a mid-run failure (network error, + # rate-limit exhaustion, Ctrl-C) still leaves the already-paid-for tweets on + # disk — the `with` block flushes and closes the file as the exception unwinds. + with args.out.open("w", newline="", encoding="utf-8") as f: + writer = csv.DictWriter(f, fieldnames=CSV_COLUMNS) + writer.writeheader() + + for term, q in search_queries: + log.info("[search] %s", q) + count = _stream_query( + writer, + seen, + api.search_tweets(q, cfg, key, max_items), + source_term=term if include_terms else "", + fetch_replies=fetch_replies, + reply_targets=reply_targets, + ) + total += count + log.info(" -> %d new tweets (total: %d)", count, total) + + for h in sorted(handles): + log.info("[user ] @%s", h) + count = _stream_query( + writer, + seen, + api.user_tweets(h, key, max_items, since=since_dt, until=until_dt), + source_term=f"@{h}" if include_terms else "", + fetch_replies=fetch_replies, + reply_targets=reply_targets, + ) + total += count + log.info(" -> %d new tweets (total: %d)", count, total) + + if reply_targets: + log.info("[replies] fetching replies for tweets with replyCount > 0...") + for tid in reply_targets: + written = _stream_replies(writer, seen, tid, key, max_items) + total += written + if written: + log.info(" [replies] tweet %s -> %d replies (total: %d)", tid, written, total) + + log.info("Wrote %d unique tweets -> %s", total, args.out) + return 0 diff --git a/scraper/tweet_scraper/config.py b/scraper/tweet_scraper/config.py new file mode 100644 index 0000000..f63b590 --- /dev/null +++ b/scraper/tweet_scraper/config.py @@ -0,0 +1,121 @@ +"""Config loading, validation, and date parsing for vars.json.""" + +from __future__ import annotations + +import json +import logging +import sys +from datetime import UTC, datetime +from pathlib import Path +from typing import Any + +log = logging.getLogger("tweet_scraper") + +DEFAULTS: dict[str, Any] = { + "author": None, + "customMapFunction": None, + "end": None, + "includeSearchTerms": False, + "maxItems": 100, + "onlyImage": False, + "onlyQuote": False, + "onlyTwitterBlue": False, + "onlyVerifiedUsers": False, + "onlyVideo": False, + "searchTerms": [], + "sort": "Latest", + "start": None, + "startUrls": [], + "tweetLanguage": None, + "twitterHandles": [], + "fetchReplies": False, +} + + +# ---------- date helpers ---------------------------------------------------- + + +def parse_config_date(s: str | None, *, end_of_day: bool = False) -> datetime | None: + """Parse an ISO 'YYYY-MM-DD' from vars.json into a UTC datetime.""" + if not s: + return None + dt = datetime.strptime(s, "%Y-%m-%d").replace(tzinfo=UTC) + if end_of_day: + dt = dt.replace(hour=23, minute=59, second=59) + return dt + + +_TWEET_TIME_FORMATS = ( + "%a %b %d %H:%M:%S %z %Y", # Twitter classic: "Thu Apr 17 14:30:00 +0000 2026" + "%Y-%m-%dT%H:%M:%S.%fZ", + "%Y-%m-%dT%H:%M:%SZ", +) + + +def tweet_created_at(t: dict[str, Any]) -> datetime | None: + s = t.get("createdAt") or t.get("created_at") + if not s: + return None + for fmt in _TWEET_TIME_FORMATS: + try: + return datetime.strptime(s, fmt).astimezone(UTC) + except ValueError: + continue + try: + return datetime.fromisoformat(s.replace("Z", "+00:00")).astimezone(UTC) + except ValueError: + return None + + +# ---------- config ---------------------------------------------------------- + + +# Accepted types per known config key. `type(None)` marks a nullable field. +# bool is intentionally allowed for maxItems too (it is an int subclass) — not +# worth special-casing for a config validator. +_CONFIG_TYPES: dict[str, tuple[type, ...]] = { + "author": (str, type(None)), + "customMapFunction": (str, type(None)), + "end": (str, type(None)), + "fetchReplies": (bool,), + "includeSearchTerms": (bool,), + "maxItems": (int, float), + "onlyImage": (bool,), + "onlyQuote": (bool,), + "onlyTwitterBlue": (bool,), + "onlyVerifiedUsers": (bool,), + "onlyVideo": (bool,), + "searchTerms": (list,), + "sort": (str,), + "start": (str, type(None)), + "startUrls": (list,), + "tweetLanguage": (str, type(None)), + "twitterHandles": (list,), +} + + +def validate_config(cfg: dict[str, Any]) -> None: + """Warn on unknown keys and reject known keys with the wrong type. + + Unknown keys only warn: the config schema mirrors the Apify Tweet Scraper V2 + input (a superset of what this script implements), so an existing Apify + vars.json must still load. A type mismatch on a known key is a hard error + because it would otherwise crash or misbehave deep in the run. + """ + for key in cfg: + if key not in _CONFIG_TYPES: + log.warning("unknown config key '%s' — ignored.", key) + for key, types in _CONFIG_TYPES.items(): + if key in cfg and not isinstance(cfg[key], types): + expected = " or ".join(t.__name__ for t in types) + got = type(cfg[key]).__name__ + sys.exit(f"Config error: '{key}' must be {expected}, got {got}.") + + +def load_config(path: Path) -> dict[str, Any]: + with path.open("r", encoding="utf-8") as f: + cfg = json.load(f) + if not isinstance(cfg, dict): + sys.exit("Config error: the top-level JSON in the config file must be an object.") + validate_config(cfg) + return {**DEFAULTS, **cfg} diff --git a/scraper/tweet_scraper/query.py b/scraper/tweet_scraper/query.py new file mode 100644 index 0000000..1b10253 --- /dev/null +++ b/scraper/tweet_scraper/query.py @@ -0,0 +1,50 @@ +"""Search-query construction and handle extraction.""" + +from __future__ import annotations + +from datetime import datetime, timedelta +from typing import Any + + +def handle_from_url(url: str) -> str | None: + """Extract a handle from an x.com / twitter.com profile URL.""" + u = url.rstrip("/") + for domain in ("x.com/", "twitter.com/"): + if domain in u: + tail = u.split(domain, 1)[1] + handle = tail.split("/")[0].split("?")[0].lstrip("@") + return handle or None + return None + + +def build_query(term: str, cfg: dict[str, Any]) -> str: + parts: list[str] = [] + if term: + # Multi-word terms become exact-phrase; single tokens stay bare + parts.append(f'"{term}"' if " " in term else term) + if cfg.get("author"): + parts.append(f"from:{str(cfg['author']).lstrip('@')}") + if cfg.get("tweetLanguage"): + parts.append(f"lang:{cfg['tweetLanguage']}") + if cfg.get("onlyVerifiedUsers"): + parts.append("filter:verified") + if cfg.get("onlyTwitterBlue"): + parts.append("filter:blue_verified") + if cfg.get("onlyVideo"): + parts.append("filter:videos") + if cfg.get("onlyImage"): + parts.append("filter:images") + if cfg.get("onlyQuote"): + parts.append("filter:quote") + # Date range — server-side filtering saves credits. + # Twitter's `until:` is exclusive of that day, so add one day to make + # the user's `end` inclusive. + if cfg.get("start"): + parts.append(f"since:{cfg['start']}") + if cfg.get("end"): + try: + end_plus = datetime.strptime(cfg["end"], "%Y-%m-%d") + timedelta(days=1) + parts.append(f"until:{end_plus.strftime('%Y-%m-%d')}") + except ValueError: + parts.append(f"until:{cfg['end']}") + return " ".join(parts)