feat(tiktok): browser-driven signed listings

surfsense_backend/app/proprietary/platforms/tiktok/extraction/__init__.py

@@ -4,11 +4,13 @@ from __future__ import annotations
 
 from .author import parse_author
 from .hydration import extract_rehydration_data
+from .item_list import items_from_response
 from .scopes import user_info, video_item_struct
 from .video import parse_video
 
 __all__ = [
     "extract_rehydration_data",
+    "items_from_response",
     "parse_author",
     "parse_video",
     "user_info",

surfsense_backend/app/proprietary/platforms/tiktok/extraction/item_list.py

@@ -0,0 +1,30 @@
+"""Item structs from a captured ``item_list`` / search API response body.
+
+Profile and hashtag listings return ``{"itemList": [...]}``; search returns
+``{"data": [{"item": {...}}]}``. Both element shapes are the same itemStruct
+:func:`parse_video` already consumes.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+
+def items_from_response(body: Any) -> list[dict[str, Any]]:
+    """Return the itemStructs carried by one API response, or ``[]``."""
+    if not isinstance(body, dict):
+        return []
+
+    item_list = body.get("itemList")
+    if isinstance(item_list, list):
+        return [i for i in item_list if isinstance(i, dict)]
+
+    data = body.get("data")
+    if isinstance(data, list):
+        return [
+            entry["item"]
+            for entry in data
+            if isinstance(entry, dict) and isinstance(entry.get("item"), dict)
+        ]
+
+    return []

surfsense_backend/app/proprietary/platforms/tiktok/flows/__init__.py

@@ -5,4 +5,9 @@ from __future__ import annotations
 from collections.abc import AsyncIterator, Awaitable, Callable
 
 FetchFn = Callable[[str], Awaitable[str | None]]
+"""Fetch a page's HTML by URL (blob-first video flow)."""
+
+FetchListingFn = Callable[[str, int], Awaitable[list[dict]]]
+"""Load a listing page and return up to ``count`` captured itemStructs."""
+
 FlowResult = AsyncIterator[dict]

surfsense_backend/app/proprietary/platforms/tiktok/flows/listing.py

@@ -0,0 +1,37 @@
+"""Listing flow shared by profile, hashtag, and search targets.
+
+The browser seam returns raw itemStructs captured from the signed ``item_list``
+XHRs; this maps each to the output contract, drops duplicate video ids, and
+stops at the per-target ``cap``.
+"""
+
+from __future__ import annotations
+
+from collections.abc import AsyncIterator
+from typing import Any
+
+from ..extraction import parse_video
+from ..extraction.timestamps import now_iso
+from ..targets.types import TikTokTarget
+from . import FetchListingFn
+
+
+async def iter_listing(
+    target: TikTokTarget, *, cap: int, fetch_listing: FetchListingFn
+) -> AsyncIterator[dict[str, Any]]:
+    if cap <= 0:
+        return
+    seen: set[str] = set()
+    emitted = 0
+    for item in await fetch_listing(target.url, cap):
+        out = parse_video(item)
+        video_id = out.get("id")
+        if video_id is not None:
+            if video_id in seen:
+                continue
+            seen.add(video_id)
+        out["scrapedAt"] = now_iso()
+        yield out
+        emitted += 1
+        if emitted >= cap:
+            return

surfsense_backend/app/proprietary/platforms/tiktok/orchestrator.py

@@ -2,35 +2,29 @@
 
 Targets run sequentially on one warm sticky IP; ``limit`` is collector policy
 applied by :func:`scrape_tiktok`, never baked into a flow. Each kind routes to
-its flow via :func:`_dispatch`.
+its flow via :func:`_dispatch`: video URLs read the rehydration blob over HTTP,
+listings capture signed item_list XHRs through the stealth browser.
 """
 
 from __future__ import annotations
 
-import logging
 from collections.abc import AsyncIterator
 from typing import Any
 from urllib.parse import quote
 
-from .flows import FetchFn
+from .flows import FetchFn, FetchListingFn
+from .flows.listing import iter_listing
 from .flows.video import iter_video
 from .schemas import TikTokScrapeInput
-from .session import fetch_html, proxy_session
+from .session import fetch_html, fetch_item_list, proxy_session
 from .targets import resolve_target
 from .targets.types import TikTokTarget
 
-logger = logging.getLogger(__name__)
-
 _PROFILE_URL = "https://www.tiktok.com/@{name}"
 _HASHTAG_URL = "https://www.tiktok.com/tag/{tag}"
 _SEARCH_URL = "https://www.tiktok.com/search?q={query}"
 
 
-async def _empty() -> AsyncIterator[dict[str, Any]]:
-    for _ in ():
-        yield {}
-
-
 def _resolve_targets(input_model: TikTokScrapeInput) -> list[TikTokTarget]:
     """Build the target list from every input source, dropping unresolved URLs."""
     targets: list[TikTokTarget] = []
@@ -54,21 +48,31 @@ def _resolve_targets(input_model: TikTokScrapeInput) -> list[TikTokTarget]:
     return targets
 
 
-def _dispatch(target: TikTokTarget, *, fetch: FetchFn) -> AsyncIterator[dict[str, Any]]:
+def _dispatch(
+    target: TikTokTarget,
+    *,
+    cap: int,
+    fetch: FetchFn,
+    fetch_listing: FetchListingFn,
+) -> AsyncIterator[dict[str, Any]]:
     if target.kind == "video":
         return iter_video(target, fetch=fetch)
-    # Listings come from the signed item_list API, not the blob.
-    logger.debug("[tiktok] no blob flow for %s target", target.kind)
-    return _empty()
+    return iter_listing(target, cap=cap, fetch_listing=fetch_listing)
 
 
 async def iter_tiktok(
-    input_model: TikTokScrapeInput, *, fetch: FetchFn = fetch_html
+    input_model: TikTokScrapeInput,
+    *,
+    fetch: FetchFn = fetch_html,
+    fetch_listing: FetchListingFn = fetch_item_list,
 ) -> AsyncIterator[dict[str, Any]]:
     """Yield normalized items for every resolved target, in order."""
+    cap = input_model.resultsPerPage
     async with proxy_session():
         for target in _resolve_targets(input_model):
-            async for item in _dispatch(target, fetch=fetch):
+            async for item in _dispatch(
+                target, cap=cap, fetch=fetch, fetch_listing=fetch_listing
+            ):
                 yield item
 
 
@@ -77,12 +81,13 @@ async def scrape_tiktok(
     *,
     limit: int | None = None,
     fetch: FetchFn = fetch_html,
+    fetch_listing: FetchListingFn = fetch_item_list,
 ) -> list[dict[str, Any]]:
     """Collect :func:`iter_tiktok` into a list, honoring an optional ``limit``."""
     from app.capabilities.core.progress import emit_progress
 
     results: list[dict[str, Any]] = []
-    async for item in iter_tiktok(input_model, fetch=fetch):
+    async for item in iter_tiktok(input_model, fetch=fetch, fetch_listing=fetch_listing):
         results.append(item)
         emit_progress("scraping", current=len(results), total=limit, unit="item")
         if limit is not None and len(results) >= limit:

surfsense_backend/app/proprietary/platforms/tiktok/session/__init__.py

@@ -4,12 +4,14 @@ from __future__ import annotations
 
 from .client import fetch_html
 from .errors import TikTokAccessBlockedError
+from .listing import fetch_item_list
 from .proxy import bind_proxy_holder, open_proxy_holder, proxy_session
 
 __all__ = [
     "TikTokAccessBlockedError",
     "bind_proxy_holder",
     "fetch_html",
+    "fetch_item_list",
     "open_proxy_holder",
     "proxy_session",
 ]

surfsense_backend/app/proprietary/platforms/tiktok/session/listing.py

@@ -0,0 +1,92 @@
+"""Browser-driven listing fetch: let TikTok sign its own ``item_list`` XHRs.
+
+Profile/hashtag/search listings need signed requests (``X-Gnarly``) whose
+algorithm rev's monthly and reads a browser canvas fingerprint. Rather than port
+and chase that signer, we load the page in the stealth browser we already run
+(patchright-Chromium, via the web-crawler tier) and capture the itemStruct JSON
+the page's own scripts fetch while scrolling. The browser is the client, so it
+signs correctly for whatever version TikTok ships.
+
+The pure response-shape parsing lives in :func:`items_from_response`; this module
+is the untested browser-I/O glue (covered by the e2e smoke, not unit tests).
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from typing import Any
+
+from scrapling.fetchers import StealthyFetcher
+
+from app.proprietary.web_crawler.stealth import (
+    build_stealthy_kwargs,
+    get_stealth_config,
+)
+from app.utils.proxy import get_proxy_url
+
+from ..extraction import items_from_response
+
+logger = logging.getLogger(__name__)
+
+# XHR paths that carry itemStructs for the three listing kinds.
+_ITEM_LIST_MARKERS = (
+    "/api/post/item_list",
+    "/api/challenge/item_list",
+    "/api/search/",
+)
+# Bounded scroll: a dead page can't loop forever, and a live one stops early
+# once enough items are captured.
+_SCROLL_MAX_ROUNDS = 20
+_SCROLL_SETTLE_MS = 1200
+
+
+def _build_page_action(collected: list[dict[str, Any]], target_count: int):
+    """A sync ``page_action`` that captures item_list XHRs while scrolling."""
+
+    def _on_response(response: Any) -> None:
+        try:
+            if not any(marker in response.url for marker in _ITEM_LIST_MARKERS):
+                return
+            body = response.json()
+        except Exception:
+            return
+        collected.extend(items_from_response(body))
+
+    def page_action(page: Any) -> Any:
+        page.on("response", _on_response)
+        try:
+            last_height = 0
+            for _ in range(_SCROLL_MAX_ROUNDS):
+                if len(collected) >= target_count:
+                    break
+                page.evaluate("window.scrollTo(0, document.body.scrollHeight)")
+                page.wait_for_timeout(_SCROLL_SETTLE_MS)
+                height = page.evaluate("document.body.scrollHeight")
+                if not height or height <= last_height:
+                    break
+                last_height = height
+        except Exception as exc:
+            logger.debug("[tiktok] listing scroll aborted: %s", exc)
+        return page
+
+    return page_action
+
+
+def _fetch_sync(url: str, target_count: int) -> list[dict[str, Any]]:
+    collected: list[dict[str, Any]] = []
+    kwargs = build_stealthy_kwargs(get_stealth_config())
+    StealthyFetcher.fetch(
+        url,
+        headless=True,
+        network_idle=False,
+        proxy=get_proxy_url(),
+        page_action=_build_page_action(collected, target_count),
+        **kwargs,
+    )
+    return collected[:target_count]
+
+
+async def fetch_item_list(page_url: str, target_count: int) -> list[dict[str, Any]]:
+    """Return up to ``target_count`` itemStructs from a listing page's XHRs."""
+    return await asyncio.to_thread(_fetch_sync, page_url, target_count)

surfsense_backend/tests/unit/platforms/tiktok/test_item_list.py

@@ -0,0 +1,25 @@
+"""Pulling item structs out of captured item_list / search API responses."""
+
+from __future__ import annotations
+
+from app.proprietary.platforms.tiktok.extraction import items_from_response
+
+
+def test_reads_item_list_shape():
+    body = {"itemList": [{"id": "1"}, {"id": "2"}], "hasMore": True}
+    assert items_from_response(body) == [{"id": "1"}, {"id": "2"}]
+
+
+def test_reads_search_data_shape():
+    body = {"data": [{"type": 1, "item": {"id": "9"}}, {"type": 4, "item": {}}]}
+    assert items_from_response(body) == [{"id": "9"}, {}]
+
+
+def test_skips_malformed_entries():
+    body = {"data": [{"type": 1}, "junk", {"item": {"id": "7"}}]}
+    assert items_from_response(body) == [{"id": "7"}]
+
+
+def test_returns_empty_for_unrelated_json():
+    assert items_from_response({"statusCode": 0}) == []
+    assert items_from_response("nope") == []

surfsense_backend/tests/unit/platforms/tiktok/test_orchestrator.py

@@ -71,3 +71,27 @@ async def test_scrape_skips_unrecognized_urls():
         TikTokScrapeInput(postURLs=["https://example.com/x"]), fetch=fake_fetch
     )
     assert items == []
+
+
+async def test_scrape_profile_returns_listing_items():
+    async def fake_listing(_url: str, _count: int) -> list[dict]:
+        return [
+            {"id": "1", "author": {"uniqueId": "a"}},
+            {"id": "2", "author": {"uniqueId": "a"}},
+        ]
+
+    items = await scrape_tiktok(
+        TikTokScrapeInput(profiles=["a"], resultsPerPage=5), fetch_listing=fake_listing
+    )
+    assert [i["id"] for i in items] == ["1", "2"]
+    assert items[0]["webVideoUrl"] == "https://www.tiktok.com/@a/video/1"
+
+
+async def test_listing_dedupes_then_caps_per_target():
+    async def fake_listing(_url: str, _count: int) -> list[dict]:
+        return [{"id": "1"}, {"id": "1"}, {"id": "2"}, {"id": "3"}]
+
+    items = await scrape_tiktok(
+        TikTokScrapeInput(hashtags=["x"], resultsPerPage=2), fetch_listing=fake_listing
+    )
+    assert [i["id"] for i in items] == ["1", "2"]