From ad8b73477a281a175cd579c12105f5ed02245207 Mon Sep 17 00:00:00 2001 From: CREDO23 Date: Wed, 8 Jul 2026 16:37:36 +0200 Subject: [PATCH] feat(tiktok): browser-driven signed listings --- .../platforms/tiktok/extraction/__init__.py | 2 + .../platforms/tiktok/extraction/item_list.py | 30 ++++++ .../platforms/tiktok/flows/__init__.py | 5 + .../platforms/tiktok/flows/listing.py | 37 ++++++++ .../platforms/tiktok/orchestrator.py | 41 +++++---- .../platforms/tiktok/session/__init__.py | 2 + .../platforms/tiktok/session/listing.py | 92 +++++++++++++++++++ .../unit/platforms/tiktok/test_item_list.py | 25 +++++ .../platforms/tiktok/test_orchestrator.py | 24 +++++ 9 files changed, 240 insertions(+), 18 deletions(-) create mode 100644 surfsense_backend/app/proprietary/platforms/tiktok/extraction/item_list.py create mode 100644 surfsense_backend/app/proprietary/platforms/tiktok/flows/listing.py create mode 100644 surfsense_backend/app/proprietary/platforms/tiktok/session/listing.py create mode 100644 surfsense_backend/tests/unit/platforms/tiktok/test_item_list.py diff --git a/surfsense_backend/app/proprietary/platforms/tiktok/extraction/__init__.py b/surfsense_backend/app/proprietary/platforms/tiktok/extraction/__init__.py index 0e6b1c3b7..d70a494a5 100644 --- a/surfsense_backend/app/proprietary/platforms/tiktok/extraction/__init__.py +++ b/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", diff --git a/surfsense_backend/app/proprietary/platforms/tiktok/extraction/item_list.py b/surfsense_backend/app/proprietary/platforms/tiktok/extraction/item_list.py new file mode 100644 index 000000000..7dddf891f --- /dev/null +++ b/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 [] diff --git a/surfsense_backend/app/proprietary/platforms/tiktok/flows/__init__.py b/surfsense_backend/app/proprietary/platforms/tiktok/flows/__init__.py index f2f4b67cc..a7809567d 100644 --- a/surfsense_backend/app/proprietary/platforms/tiktok/flows/__init__.py +++ b/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] diff --git a/surfsense_backend/app/proprietary/platforms/tiktok/flows/listing.py b/surfsense_backend/app/proprietary/platforms/tiktok/flows/listing.py new file mode 100644 index 000000000..44fcc33e2 --- /dev/null +++ b/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 diff --git a/surfsense_backend/app/proprietary/platforms/tiktok/orchestrator.py b/surfsense_backend/app/proprietary/platforms/tiktok/orchestrator.py index d924e3a9e..c7f2e9008 100644 --- a/surfsense_backend/app/proprietary/platforms/tiktok/orchestrator.py +++ b/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: diff --git a/surfsense_backend/app/proprietary/platforms/tiktok/session/__init__.py b/surfsense_backend/app/proprietary/platforms/tiktok/session/__init__.py index 659305055..409aa5987 100644 --- a/surfsense_backend/app/proprietary/platforms/tiktok/session/__init__.py +++ b/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", ] diff --git a/surfsense_backend/app/proprietary/platforms/tiktok/session/listing.py b/surfsense_backend/app/proprietary/platforms/tiktok/session/listing.py new file mode 100644 index 000000000..0d19281a2 --- /dev/null +++ b/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) diff --git a/surfsense_backend/tests/unit/platforms/tiktok/test_item_list.py b/surfsense_backend/tests/unit/platforms/tiktok/test_item_list.py new file mode 100644 index 000000000..00ab89b02 --- /dev/null +++ b/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") == [] diff --git a/surfsense_backend/tests/unit/platforms/tiktok/test_orchestrator.py b/surfsense_backend/tests/unit/platforms/tiktok/test_orchestrator.py index 85c601047..8520a1d85 100644 --- a/surfsense_backend/tests/unit/platforms/tiktok/test_orchestrator.py +++ b/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"]