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

@@ -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.<country>" 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.<country>" suffix above.
 
 # --- Google Search scraper: scale / warm sticky-IP pool -----------------------
 # Per-process render concurrency AND the throughput lever: ceiling =

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:

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
 
 

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
 

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:

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)]

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 <profile> [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 = []

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.

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.

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)