feat(tiktok): surface comments, user_search, trending across all product surfaces

Brings the three newer verbs to parity with tiktok.scrape everywhere it was
wired: MCP tools (+ selfcheck manifest), the chat subagent prompt/description,
the playground catalog, the native docs page, and the SEO marketing page. Also
adds live e2e stages for comments, user search, and trending. Docs/FAQ now
state the real contract: profile metadata is reliable while its video list can
be withheld, and keyword video search is walled (use user search for accounts).
This commit is contained in:
CREDO23 2026-07-09 19:09:00 +02:00
parent 67b5472b9f
commit 6b15d610d9
8 changed files with 294 additions and 33 deletions

View file

@ -15,6 +15,12 @@ What it exercises (everything REAL — live network, live proxy, live browser):
Stage 5 full scrape_tiktok() pipeline on a hashtag.
Stage 6 search via the full pipeline: same graceful-degrade contract as
profile (results feed doesn't load for anonymous sessions).
Stage 7 comments on a real video URL (served anonymously once the panel
opens): real comments OR a single honest ErrorItem.
Stage 8 user search: the account-discovery XHR that DOES serve anonymous
headless sessions asserts real account records come back.
Stage 9 trending: the Explore feed of trending videos asserts real,
normalized video items come back.
On success it writes raw itemStructs under tests/fixtures/tiktok/ so the parser
suites can pin against real-shaped data without network.
@ -199,6 +205,61 @@ async def stage_search_listing() -> tuple[bool, list[dict[str, Any]]]:
return ok, items
async def stage_comments(video_url: str) -> tuple[bool, list[dict[str, Any]]]:
_hr("STAGE 7 — comments graceful-degrade")
print(f" target: {video_url}")
from app.proprietary.platforms.tiktok import scrape_tiktok_comments
# Comments load over a signed /api/comment/list XHR that TikTok serves to
# anonymous sessions once the panel opens. Pass if real comments come back
# OR a graceful ErrorItem (video has none / disabled / withheld).
items = await scrape_tiktok_comments(
[video_url], per_video=_COUNT, limit=_COUNT
)
has_comment = any(it.get("id") and not it.get("errorCode") for it in items)
has_error = any(it.get("errorCode") == "no_comments" for it in items)
ok = _check(
"comments yield records or a graceful ErrorItem (never silent empty)",
has_comment or has_error,
f"{len(items)} item(s); comment={has_comment} error={has_error}",
)
return ok, items
async def stage_user_search() -> tuple[bool, list[dict[str, Any]]]:
_hr(f"STAGE 8 — user search (browser): {_PROFILE!r}")
from app.proprietary.platforms.tiktok import search_tiktok_users
# Unlike keyword *video* search, the account-search XHR serves anonymous
# headless sessions — so this asserts real records, not just degradation.
items = await search_tiktok_users([_PROFILE], per_query=_COUNT, limit=_COUNT)
real = [it for it in items if not it.get("errorCode")]
ok = _check(
"user search returns account records",
bool(real) and bool(real[0].get("uniqueId") or real[0].get("name")),
f"{len(items)} item(s); accounts={len(real)}",
)
if real:
print(f" sample: @{real[0].get('uniqueId') or real[0].get('name')}")
return ok, items
async def stage_trending() -> tuple[bool, list[dict[str, Any]]]:
_hr("STAGE 9 — trending (browser): Explore feed")
from app.proprietary.platforms.tiktok import scrape_tiktok_trending
items = await scrape_tiktok_trending(count=_COUNT)
real = [it for it in items if not it.get("errorCode")]
ok = _check(
"trending returns normalized video items",
bool(real) and bool(real[0].get("id")) and bool(real[0].get("webVideoUrl")),
f"{len(items)} item(s); videos={len(real)}",
)
if real:
print(f" sample: {real[0].get('webVideoUrl')}{real[0].get('text', '')[:60]!r}")
return ok, items
async def main() -> int:
print("TikTok scraper functional e2e — live network + proxy + browser")
results: dict[str, bool] = {}
@ -214,8 +275,10 @@ async def main() -> int:
if video_url:
ok_video, _ = await stage_blob_video(video_url)
results["Stage 3 blob video"] = ok_video
ok_comments, _ = await stage_comments(video_url)
results["Stage 7 comments"] = ok_comments
else:
print("\n [SKIP] Stage 3 — no captured struct to build a video URL")
print("\n [SKIP] Stage 3/7 — no captured struct to build a video URL")
ok_search, _ = await stage_search_listing()
results["Stage 6 search listing"] = ok_search
@ -224,6 +287,11 @@ async def main() -> int:
results["Stage 2 profile listing"] = ok_profile
results["Stage 5 pipeline"] = await stage_pipeline()
ok_users, _ = await stage_user_search()
results["Stage 8 user search"] = ok_users
ok_trending, _ = await stage_trending()
results["Stage 9 trending"] = ok_trending
_hr("SUMMARY")
for name, ok in results.items():
print(f" {'PASS' if ok else 'FAIL/SKIP'}{name}")