From 8a5d37a4db739e9a5f8e44692183de301b1c9a47 Mon Sep 17 00:00:00 2001 From: "DESKTOP-RTLN3BA\\$punk" Date: Fri, 17 Jul 2026 16:30:26 -0700 Subject: [PATCH] refactor: implement cross-country proxy rotation for Reddit and TikTok scrapers, enhancing resilience against IP blocks and improving search query handling --- surfsense_backend/.env.example | 8 +++ .../app/proprietary/platforms/reddit/fetch.py | 43 +++++++++++---- .../platforms/tiktok/orchestrator.py | 16 ++++-- .../platforms/tiktok/session/client.py | 7 ++- .../platforms/tiktok/session/proxy.py | 23 +++++--- surfsense_backend/app/utils/proxy/rotation.py | 34 ++++++++++++ .../scripts/e2e_instagram_scraper.py | 16 +++++- .../platforms/reddit/test_fetch_resilience.py | 4 ++ .../platforms/tiktok/test_orchestrator.py | 31 +++++++++++ .../tests/unit/utils/proxy/test_rotation.py | 53 +++++++++++++++++++ 10 files changed, 213 insertions(+), 22 deletions(-) create mode 100644 surfsense_backend/app/utils/proxy/rotation.py create mode 100644 surfsense_backend/tests/unit/utils/proxy/test_rotation.py diff --git a/surfsense_backend/.env.example b/surfsense_backend/.env.example index 94508431c..404bd3b44 100644 --- a/surfsense_backend/.env.example +++ b/surfsense_backend/.env.example @@ -373,6 +373,14 @@ TURNSTILE_SECRET_KEY= # (cyclic); server-side-rotating gateways ignore it. # PROXY_URL=http://user:pass@host:port # PROXY_URLS=http://user:pass@host1:port,http://user:pass@host2:port +# +# NOTE (dataimpulse): its default *worldwide* pool (a PROXY_URL with no +# "__cr." suffix) hands out exit IPs some targets hard-block — Reddit +# 403s every one, TikTok withholds its ttwid cookie — so those scrapers +# pin/rotate exit countries on their own (no config needed). Other scrapers +# (e.g. Google SERP) deliberately prefer the worldwide pool, so leave PROXY_URL +# unpinned unless you have a reason; pin a country only for your whole install by +# adding a "__cr." suffix above. # --- Google Search scraper: scale / warm sticky-IP pool ----------------------- # Per-process render concurrency AND the throughput lever: ceiling = diff --git a/surfsense_backend/app/proprietary/platforms/reddit/fetch.py b/surfsense_backend/app/proprietary/platforms/reddit/fetch.py index 54c7ba125..4184e85bb 100644 --- a/surfsense_backend/app/proprietary/platforms/reddit/fetch.py +++ b/surfsense_backend/app/proprietary/platforms/reddit/fetch.py @@ -37,7 +37,15 @@ from urllib.parse import urlencode from scrapling.fetchers import AsyncFetcher, FetcherSession -from app.utils.proxy import get_proxy_url +from app.utils.proxy import get_geo_proxy_url, get_proxy_url + +# Shared cross-country rotation walk (also used by the TikTok sibling). Kept under +# the historical private names this module and its tests reference. +from app.utils.proxy.rotation import ( + FALLBACK_COUNTRIES as _FALLBACK_COUNTRIES, + country_for_rotation as _country_for_rotation, + rotation_countries as _rotation_countries, +) logger = logging.getLogger(__name__) @@ -68,7 +76,16 @@ _current_session: ContextVar[_RotatingSession | None] = ContextVar( # different handling per status (spec section 3). _ROTATE_STATUS = 403 _BACKOFF_STATUS = 429 -_MAX_ROTATIONS = 3 +# Rotating an IP is cheap (close + reopen one keep-alive connection through the +# gateway + a 2-request warm ≈ a few seconds) and each rotation also walks to the +# next country pool, so we spend rotations liberally: neither a dirty IP nor a +# wholly-blocked country pool should fail a job. 8 ≥ len(_FALLBACK_COUNTRIES), so +# a job tries every country at least once before giving up. Worst case (a genuine +# global block) costs _MAX_ROTATIONS bounded warm attempts before +# RedditAccessBlockedError. +# ponytail: 8 caps that worst case at ~30s; raise if every pool gets dirty at +# once, lower if a real global block is wasting time. +_MAX_ROTATIONS = 8 _MAX_BACKOFFS = 4 _BACKOFF_BASE_S = 5.0 @@ -155,11 +172,13 @@ class _RotatingSession: """Owns one live ``FetcherSession`` (sticky IP) and can swap it for a fresh one. ``rotate()`` closes the current keep-alive connection and opens a new one, so - the rotating gateway hands out a different residential exit IP. Because the - ``loid`` cookie binds to the exit IP, ``rotate()`` also drops the warmed - state — the next fetch re-warms on the new IP. Used sequentially within a - single flow (never shared across concurrent tasks), so no locking is needed. - ``session`` is ``None`` only when no proxy is configured. + the rotating gateway hands out a different residential exit IP — walking to + the next country pool (see :func:`_country_for_rotation`) so a wholly-blocked + pool can't fail the flow. Because the ``loid`` cookie binds to the exit IP, + ``rotate()`` also drops the warmed state — the next fetch re-warms on the new + IP. Used sequentially within a single flow (never shared across concurrent + tasks), so no locking is needed. ``session`` is ``None`` only when no proxy + is configured. """ def __init__(self) -> None: @@ -167,11 +186,13 @@ class _RotatingSession: self.session: Any | None = None self.rotations = 0 self.warmed = False + self.country = "" self._last_at = 0.0 async def _open(self) -> None: - proxy = get_proxy_url() self.warmed = False + self.country = _country_for_rotation(self.rotations) + proxy = get_geo_proxy_url(self.country) if proxy is None: self._cm = self.session = None return @@ -194,7 +215,11 @@ class _RotatingSession: await self.close() self.rotations += 1 await self._open() - logger.info("[reddit] rotated proxy session (rotation #%d)", self.rotations) + logger.info( + "[reddit] rotated proxy session (rotation #%d, country=%s)", + self.rotations, + self.country, + ) return self.session async def pace(self) -> None: diff --git a/surfsense_backend/app/proprietary/platforms/tiktok/orchestrator.py b/surfsense_backend/app/proprietary/platforms/tiktok/orchestrator.py index a738233a6..9a285dc1d 100644 --- a/surfsense_backend/app/proprietary/platforms/tiktok/orchestrator.py +++ b/surfsense_backend/app/proprietary/platforms/tiktok/orchestrator.py @@ -10,6 +10,7 @@ from __future__ import annotations from collections.abc import AsyncIterator from typing import Any +from urllib.parse import quote from .extraction.timestamps import now_iso from .flows import FetchCommentsFn, FetchFn, FetchListingFn, FetchUsersFn @@ -31,15 +32,18 @@ from .targets.types import TikTokTarget _PROFILE_URL = "https://www.tiktok.com/@{name}" _HASHTAG_URL = "https://www.tiktok.com/tag/{tag}" +_SEARCH_URL = "https://www.tiktok.com/search?q={query}" _EXPLORE_URL = "https://www.tiktok.com/explore" def _resolve_targets(input_model: TikTokScrapeInput) -> list[TikTokTarget]: - """Build the target list from the URL/profile/hashtag sources. + """Build the target list from every input source. - A raw ``tiktok.com/search?...`` URL passed explicitly in - ``startUrls``/``postURLs`` still resolves here and keeps its native listing - routing; there is no keyword-search shortcut. + ``searchQueries`` map to the same ``tiktok.com/search?q=`` targets that a raw + search URL in ``startUrls``/``postURLs`` resolves to, so both share the + listing flow's parse/dedupe/cap/empty-ErrorItem contract (the anonymous + search feed often withholds results, degrading to one honest ErrorItem rather + than a silent empty). """ targets: list[TikTokTarget] = [] for entry in input_model.startUrls: @@ -55,6 +59,10 @@ def _resolve_targets(input_model: TikTokScrapeInput) -> list[TikTokTarget]: targets.append(TikTokTarget("profile", name, _PROFILE_URL.format(name=name))) for tag in input_model.hashtags: targets.append(TikTokTarget("hashtag", tag, _HASHTAG_URL.format(tag=tag))) + for query in input_model.searchQueries: + resolved = resolve_target(_SEARCH_URL.format(query=quote(query))) + if resolved is not None: + targets.append(resolved) return targets diff --git a/surfsense_backend/app/proprietary/platforms/tiktok/session/client.py b/surfsense_backend/app/proprietary/platforms/tiktok/session/client.py index e5be1e419..4f374c536 100644 --- a/surfsense_backend/app/proprietary/platforms/tiktok/session/client.py +++ b/surfsense_backend/app/proprietary/platforms/tiktok/session/client.py @@ -26,7 +26,12 @@ logger = logging.getLogger(__name__) # 403 => IP blocked; rotate and re-warm. 429 => rate limited; back off same IP. _ROTATE_STATUS = 403 _BACKOFF_STATUS = 429 -_MAX_ROTATIONS = 3 +# Each rotation walks to the next country pool (see session/proxy.py). The bare +# worldwide pool never mints ``ttwid`` (proven live 2026-07-17), so warming +# relies on reaching a good country pool — budget enough rotations to try every +# one at least once (>= len(rotation FALLBACK_COUNTRIES)). Rotating is cheap +# (reopen one keep-alive connection + a 1-request warm), so spend liberally. +_MAX_ROTATIONS = 8 _MAX_BACKOFFS = 4 _BACKOFF_BASE_S = 5.0 diff --git a/surfsense_backend/app/proprietary/platforms/tiktok/session/proxy.py b/surfsense_backend/app/proprietary/platforms/tiktok/session/proxy.py index dca6c776b..ab53ea6f5 100644 --- a/surfsense_backend/app/proprietary/platforms/tiktok/session/proxy.py +++ b/surfsense_backend/app/proprietary/platforms/tiktok/session/proxy.py @@ -18,7 +18,8 @@ from typing import Any from scrapling.fetchers import FetcherSession -from app.utils.proxy import get_proxy_url +from app.utils.proxy import get_geo_proxy_url +from app.utils.proxy.rotation import country_for_rotation logger = logging.getLogger(__name__) @@ -37,9 +38,13 @@ _current_session: ContextVar[_RotatingSession | None] = ContextVar( class _RotatingSession: """Owns one live ``FetcherSession`` (sticky IP); ``rotate()`` swaps the IP. - Used sequentially within a single flow (never shared across concurrent - tasks), so no locking is needed. ``session`` is ``None`` only when no proxy - is configured. + Each open walks to the next country pool (see + :func:`app.utils.proxy.rotation.country_for_rotation`): TikTok withholds the + anonymous ``ttwid`` cookie from the provider's default worldwide pool but + mints it on country-pinned exits, so re-drawing within the same pool never + warms — spreading rotations across country pools does. Used sequentially + within a single flow (never shared across concurrent tasks), so no locking is + needed. ``session`` is ``None`` only when no proxy is configured. """ def __init__(self) -> None: @@ -47,11 +52,13 @@ class _RotatingSession: self.session: Any | None = None self.rotations = 0 self.warmed = False + self.country = "" self._last_at = 0.0 async def _open(self) -> None: - proxy = get_proxy_url() self.warmed = False + self.country = country_for_rotation(self.rotations) + proxy = get_geo_proxy_url(self.country) if proxy is None: self._cm = self.session = None return @@ -74,7 +81,11 @@ class _RotatingSession: await self.close() self.rotations += 1 await self._open() - logger.info("[tiktok] rotated proxy session (rotation #%d)", self.rotations) + logger.info( + "[tiktok] rotated proxy session (rotation #%d, country=%s)", + self.rotations, + self.country, + ) return self.session async def pace(self) -> None: diff --git a/surfsense_backend/app/utils/proxy/rotation.py b/surfsense_backend/app/utils/proxy/rotation.py new file mode 100644 index 000000000..d8be37923 --- /dev/null +++ b/surfsense_backend/app/utils/proxy/rotation.py @@ -0,0 +1,34 @@ +"""Cross-country exit rotation for warm-session scrapers. + +Some targets (Reddit's ``loid``, TikTok's ``ttwid``) silently withhold their +anonymous session cookie from the provider's *default worldwide* pool but hand +it out freely on **country-pinned** exits (proven live: a bare-pool homepage +hit returns 200 with an empty cookie jar, while a us/gb/de/nl-pinned hit mints +the cookie every time). A warm-on-block flow that only re-draws from the same +worldwide pool therefore burns every rotation on cookie-less IPs and fails. + +Walking a spread of country pools instead lets the flow escape a wholly-blocked +pool. The provider's configured country leads (so an operator's choice is +honoured first); the fallbacks are large, reliable residential pools. Non-geo +providers (e.g. the custom single-URL provider) ignore the country and re-draw +their one URL, so this is a harmless no-op there. +""" + +from __future__ import annotations + +from app.utils.proxy.registry import get_active_provider + +# Walk order after the configured country. Ordered by pool size / reliability. +FALLBACK_COUNTRIES = ("us", "gb", "de", "ca", "nl", "fr") + + +def rotation_countries() -> tuple[str, ...]: + """Ordered, de-duplicated exit countries with the configured one leading.""" + lead = get_active_provider().get_location() + return tuple(dict.fromkeys(c for c in (lead, *FALLBACK_COUNTRIES) if c)) + + +def country_for_rotation(n: int) -> str: + """Exit country for rotation index ``n`` (cycles the list, wrapping around).""" + countries = rotation_countries() + return countries[n % len(countries)] diff --git a/surfsense_backend/scripts/e2e_instagram_scraper.py b/surfsense_backend/scripts/e2e_instagram_scraper.py index a5e356f99..b74c20ca6 100644 --- a/surfsense_backend/scripts/e2e_instagram_scraper.py +++ b/surfsense_backend/scripts/e2e_instagram_scraper.py @@ -51,8 +51,14 @@ from app.proprietary.platforms.instagram.fetch import ( # noqa: E402 ) from app.proprietary.platforms.instagram.url_resolver import resolve_url # noqa: E402 -_PROFILE = "natgeo" -_SEARCH_TERM = "national geographic" +# Canonical public targets. Override from the CLI to test any real-world case: +# python scripts/e2e_instagram_scraper.py [search term] +# Note: web_profile_info intermittently 400s for *business/creator* accounts +# (IG server bug on the ig_business_category_subvertical schema); a regular +# public account is the reliable smoke target. +_DEFAULT_PROFILE = "natgeo" +_PROFILE = sys.argv[1] if len(sys.argv) > 1 else _DEFAULT_PROFILE +_SEARCH_TERM = sys.argv[2] if len(sys.argv) > 2 else "national geographic" _FIXTURE_DIR = _BACKEND_ROOT / "tests" / "unit" / "platforms" / "instagram" / "fixtures" @@ -179,6 +185,12 @@ async def step5_search() -> bool: async def step6_dump_fixtures(post_url: str | None) -> bool: _hr("STEP 6 — dump trimmed, anonymized fixtures for offline tests") + if _PROFILE != _DEFAULT_PROFILE: + return _check( + "dumped fixtures", + True, + f"skipped (custom profile {_PROFILE!r} would clobber committed fixtures)", + ) profile = await fetch_json("api/v1/users/web_profile_info/", {"username": _PROFILE}) _FIXTURE_DIR.mkdir(parents=True, exist_ok=True) wrote = [] diff --git a/surfsense_backend/tests/unit/platforms/reddit/test_fetch_resilience.py b/surfsense_backend/tests/unit/platforms/reddit/test_fetch_resilience.py index 610855f5c..6f5362fd5 100644 --- a/surfsense_backend/tests/unit/platforms/reddit/test_fetch_resilience.py +++ b/surfsense_backend/tests/unit/platforms/reddit/test_fetch_resilience.py @@ -266,3 +266,7 @@ async def test_fan_out_closes_all_sessions_on_early_stop(monkeypatch): async def test_fan_out_empty_jobs_is_noop(): out = [x async for x in scraper.fan_out([])] assert out == [] + + +# Cross-country rotation lives in app.utils.proxy.rotation and is shared with the +# TikTok sibling; its unit tests live in tests/unit/utils/proxy/test_rotation.py. diff --git a/surfsense_backend/tests/unit/platforms/tiktok/test_orchestrator.py b/surfsense_backend/tests/unit/platforms/tiktok/test_orchestrator.py index 3a4e42776..f42db7023 100644 --- a/surfsense_backend/tests/unit/platforms/tiktok/test_orchestrator.py +++ b/surfsense_backend/tests/unit/platforms/tiktok/test_orchestrator.py @@ -169,6 +169,37 @@ async def test_listing_dedupes_then_caps_per_target(): assert [i["id"] for i in items] == ["1", "2"] +async def test_search_query_resolves_to_listing_target(): + # searchQueries must produce a search target (not be silently dropped): a + # query with results flows through the listing parse/dedupe/cap path. + async def fake_listing(url: str, _count: int) -> list[dict]: + assert "search?q=meal%20prep" in url # query wired into the search URL + return [{"id": "1", "author": {"uniqueId": "a"}}] + + items = await scrape_tiktok( + TikTokScrapeInput(searchQueries=["meal prep"], resultsPerPage=5), + fetch=_no_html, + fetch_listing=fake_listing, + ) + assert [i["id"] for i in items] == ["1"] + + +async def test_empty_search_query_degrades_to_error_item(): + # A withheld anonymous search feed must surface one honest ErrorItem tagged + # with the query, never a silent empty. + async def fake_listing(_url: str, _count: int) -> list[dict]: + return [] + + items = await scrape_tiktok( + TikTokScrapeInput(searchQueries=["meal prep"], resultsPerPage=5), + fetch=_no_html, + fetch_listing=fake_listing, + ) + assert len(items) == 1 + assert items[0]["errorCode"] == "no_items" + assert items[0]["input"] == "meal prep" + + async def test_empty_listing_emits_error_item(): # A trust-gated/empty feed (0 videos) must surface one honest ErrorItem, # tagged with errorCode, rather than vanishing silently. diff --git a/surfsense_backend/tests/unit/utils/proxy/test_rotation.py b/surfsense_backend/tests/unit/utils/proxy/test_rotation.py new file mode 100644 index 000000000..09aa05441 --- /dev/null +++ b/surfsense_backend/tests/unit/utils/proxy/test_rotation.py @@ -0,0 +1,53 @@ +"""Unit tests for the shared cross-country exit rotation helper. + +Offline: the provider is faked via monkeypatch, so no network/proxy is touched. +""" + +from __future__ import annotations + +from app.utils.proxy import rotation + + +class _Prov: + def __init__(self, location: str) -> None: + self._location = location + + def get_location(self) -> str: + return self._location + + +def test_lead_country_leads_and_dedupes(monkeypatch): + # The configured/default exit country leads and isn't duplicated in the walk. + monkeypatch.setattr(rotation, "get_active_provider", lambda: _Prov("gb")) + countries = rotation.rotation_countries() + + assert countries[0] == "gb" + assert len(countries) == len(set(countries)) # de-duplicated + assert set(rotation.FALLBACK_COUNTRIES) <= set(countries) # fallbacks kept + + +def test_no_configured_country_uses_fallbacks_only(monkeypatch): + # A bare PROXY_URL (no country) leaves just the fallback pools, in order. + monkeypatch.setattr(rotation, "get_active_provider", lambda: _Prov("")) + assert rotation.rotation_countries() == rotation.FALLBACK_COUNTRIES + + +def test_walk_covers_every_country_and_wraps(monkeypatch): + # A whole-pool block can't stall the walk: every country is reached, and the + # index cycles (wraps) rather than running off the end. + monkeypatch.setattr(rotation, "get_active_provider", lambda: _Prov("us")) + countries = rotation.rotation_countries() + tried = {rotation.country_for_rotation(n) for n in range(len(countries))} + + assert tried == set(countries) + assert rotation.country_for_rotation(len(countries)) == countries[0] # wraps + + +def test_caller_budgets_cover_every_country(): + # Each warm-on-block caller must budget enough rotations to try every pool at + # least once, else a wholly-blocked lead pool could fail a job prematurely. + from app.proprietary.platforms.reddit import fetch as reddit_fetch + from app.proprietary.platforms.tiktok.session import client as tiktok_client + + assert reddit_fetch._MAX_ROTATIONS >= len(rotation.FALLBACK_COUNTRIES) + assert tiktok_client._MAX_ROTATIONS >= len(rotation.FALLBACK_COUNTRIES)