feat(tiktok): return profile metadata even when video listing is gated

A profile's account data (name, followers, bio, verification) lives in the
page's rehydration blob and loads over plain HTTP without a signed request,
so emit it first and always. The video listing needs a signed item_list XHR
that TikTok withholds from anonymous sessions, so it stays best-effort and
degrades to an ErrorItem. A blocked profile now yields its metadata instead
of only an ErrorItem.

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

@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-from .author import parse_author
+from .author import parse_author, parse_profile
 from .hydration import extract_rehydration_data
 from .item_list import items_from_response
 from .scopes import user_info, video_item_struct
@@ -12,6 +12,7 @@ __all__ = [
     "extract_rehydration_data",
     "items_from_response",
     "parse_author",
+    "parse_profile",
     "parse_video",
     "user_info",
     "video_item_struct",

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

@@ -29,3 +29,10 @@ def build_author_meta(author: dict[str, Any], stats: dict[str, Any]) -> dict[str
 def parse_author(user_info: dict[str, Any]) -> dict[str, Any]:
     """Map a ``webapp.user-detail`` ``userInfo`` (``{user, stats}``) to authorMeta."""
     return build_author_meta(user_info.get("user") or {}, user_info.get("stats") or {})
+
+
+def parse_profile(user_info: dict[str, Any]) -> dict[str, Any]:
+    """Map a ``userInfo`` to a standalone :class:`TikTokProfileItem` output dict."""
+    from ..schemas.items import TikTokProfileItem
+
+    return TikTokProfileItem(**parse_author(user_info)).to_output()

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

@@ -0,0 +1,37 @@
+"""Profile flow: reliable blob metadata first, then the (gated) video listing.
+
+A profile's account data (name, followers, bio, verification) lives in the page's
+rehydration blob and loads over plain HTTP without a signed request, so we emit it
+first and always. The video listing needs a signed ``item_list`` XHR that TikTok
+withholds from anonymous sessions, so it is best-effort: it streams videos when it
+loads and degrades to an ErrorItem (via :func:`iter_listing`) when withheld. The
+metadata item therefore survives even when the videos are blocked.
+"""
+
+from __future__ import annotations
+
+from collections.abc import AsyncIterator
+from typing import Any
+
+from ..extraction import extract_rehydration_data, parse_profile, user_info
+from ..extraction.timestamps import now_iso
+from ..targets.types import TikTokTarget
+from . import FetchFn, FetchListingFn
+from .listing import iter_listing
+
+
+async def iter_profile(
+    target: TikTokTarget,
+    *,
+    cap: int,
+    fetch: FetchFn,
+    fetch_listing: FetchListingFn,
+) -> AsyncIterator[dict[str, Any]]:
+    html = await fetch(target.url)
+    info = user_info(extract_rehydration_data(html) or {}) if html else None
+    if info:
+        item = parse_profile(info)
+        item["scrapedAt"] = now_iso()
+        yield item
+    async for out in iter_listing(target, cap=cap, fetch_listing=fetch_listing):
+        yield out

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

@@ -14,6 +14,7 @@ from urllib.parse import quote
 
 from .flows import FetchFn, FetchListingFn
 from .flows.listing import iter_listing
+from .flows.profile import iter_profile
 from .flows.video import iter_video
 from .schemas import TikTokScrapeInput
 from .session import fetch_html, fetch_item_list
@@ -57,6 +58,8 @@ def _dispatch(
 ) -> AsyncIterator[dict[str, Any]]:
     if target.kind == "video":
         return iter_video(target, fetch=fetch)
+    if target.kind == "profile":
+        return iter_profile(target, cap=cap, fetch=fetch, fetch_listing=fetch_listing)
     return iter_listing(target, cap=cap, fetch_listing=fetch_listing)
 
 

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

@@ -8,6 +8,7 @@ from .items import (
     CommentItem,
     ErrorItem,
     MusicMeta,
+    TikTokProfileItem,
     TikTokVideoItem,
     VideoMeta,
 )
@@ -18,6 +19,7 @@ __all__ = [
     "ErrorItem",
     "MusicMeta",
     "StartUrl",
+    "TikTokProfileItem",
     "TikTokScrapeInput",
     "TikTokVideoItem",
     "VideoMeta",

surfsense_backend/app/proprietary/platforms/tiktok/schemas/items.py

@@ -30,6 +30,21 @@ class AuthorMeta(BaseModel):
     video: int | None = None
 
 
+class TikTokProfileItem(AuthorMeta):
+    """A profile's public metadata, read from the page's rehydration blob.
+
+    Emitted even when the video listing is withheld from anonymous access, so a
+    blocked profile still yields its account data (name, followers, bio,
+    verification) instead of only an ErrorItem. Distinguishable from a video item
+    by the absence of ``webVideoUrl``/``text``.
+    """
+
+    scrapedAt: str | None = None
+
+    def to_output(self) -> dict[str, Any]:
+        return self.model_dump(exclude_none=False)
+
+
 class MusicMeta(BaseModel):
     model_config = ConfigDict(extra="allow")
 

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

@@ -10,6 +10,33 @@ import json
 from app.proprietary.platforms.tiktok import TikTokScrapeInput, scrape_tiktok
 
 
+async def _no_html(_url: str) -> str:
+    """Fetch stub that yields no rehydration blob (skips profile metadata)."""
+    return ""
+
+
+def _profile_page(username: str, followers: int, videos: int) -> str:
+    blob = {
+        "__DEFAULT_SCOPE__": {
+            "webapp.user-detail": {
+                "userInfo": {
+                    "user": {
+                        "id": "u1",
+                        "uniqueId": username,
+                        "nickname": "Nick",
+                        "verified": True,
+                    },
+                    "stats": {"followerCount": followers, "videoCount": videos},
+                }
+            }
+        }
+    }
+    return (
+        '<script id="__UNIVERSAL_DATA_FOR_REHYDRATION__" '
+        f'type="application/json">{json.dumps(blob)}</script>'
+    )
+
+
 def _video_page(video_id: str, username: str) -> str:
     blob = {
         "__DEFAULT_SCOPE__": {
@@ -81,12 +108,57 @@ async def test_scrape_profile_returns_listing_items():
         ]
 
     items = await scrape_tiktok(
-        TikTokScrapeInput(profiles=["a"], resultsPerPage=5), fetch_listing=fake_listing
+        TikTokScrapeInput(profiles=["a"], resultsPerPage=5),
+        fetch=_no_html,
+        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_profile_emits_metadata_then_videos():
+    # The blob metadata item comes first and is billable; videos follow.
+    async def fake_fetch(_url: str) -> str:
+        return _profile_page("a", followers=100, videos=2)
+
+    async def fake_listing(_url: str, _count: int) -> list[dict]:
+        return [{"id": "1", "author": {"uniqueId": "a"}}]
+
+    items = await scrape_tiktok(
+        TikTokScrapeInput(profiles=["a"], resultsPerPage=5),
+        fetch=fake_fetch,
+        fetch_listing=fake_listing,
+    )
+    assert len(items) == 2
+    profile, video = items
+    assert "webVideoUrl" not in profile  # metadata item, not a video
+    assert profile["name"] == "a"
+    assert profile["fans"] == 100
+    assert profile["verified"] is True
+    assert profile["scrapedAt"] is not None
+    assert video["id"] == "1"
+
+
+async def test_profile_metadata_survives_blocked_listing():
+    # Videos withheld from anonymous access: we still return the profile metadata
+    # (not just an ErrorItem), so a blocked profile isn't a total loss.
+    async def fake_fetch(_url: str) -> str:
+        return _profile_page("a", followers=100, videos=9)
+
+    async def fake_listing(_url: str, _count: int) -> list[dict]:
+        return []
+
+    items = await scrape_tiktok(
+        TikTokScrapeInput(profiles=["a"], resultsPerPage=5),
+        fetch=fake_fetch,
+        fetch_listing=fake_listing,
+    )
+    assert len(items) == 2
+    assert items[0]["name"] == "a"
+    assert items[0]["fans"] == 100
+    assert items[1]["errorCode"] == "no_items"
+
+
 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"}]
@@ -105,6 +177,7 @@ async def test_empty_listing_emits_error_item():
 
     items = await scrape_tiktok(
         TikTokScrapeInput(profiles=["nasa"], resultsPerPage=5),
+        fetch=_no_html,
         fetch_listing=fake_listing,
     )
     assert len(items) == 1