mirror of
https://github.com/MODSetter/SurfSense.git
synced 2026-07-18 23:11:12 +02:00
Merge remote-tracking branch 'upstream/dev' into feat/instagram-scraper
This commit is contained in:
Anish Sarkar
commit
e38ca19b18
119 changed files with 5948 additions and 36 deletions
|
|
@ -37,6 +37,7 @@ SUBAGENT_TO_REQUIRED_CONNECTOR_MAP: dict[str, frozenset[str]] = {
|
|||
"google_search": frozenset(),
|
||||
"reddit": frozenset(),
|
||||
"instagram": frozenset(),
|
||||
"tiktok": frozenset(),
|
||||
"mcp_discovery": frozenset(
|
||||
{
|
||||
"SLACK_CONNECTOR",
|
||||
|
|
|
|||
|
|
@ -6,9 +6,9 @@ reviews are moving, and what is being said across the open web — and to put
|
|||
that intelligence to work alongside their own knowledge base.
|
||||
|
||||
You do this by dispatching **specialist subagents** via the `task` tool:
|
||||
- **Live market data** — Reddit, YouTube, Google Maps, Google Search, and the
|
||||
web crawler return structured, current platform data (posts, comments,
|
||||
transcripts, reviews, SERPs, full page content).
|
||||
- **Live market data** — Reddit, YouTube, TikTok, Google Maps, Google Search,
|
||||
and the web crawler return structured, current platform data (posts,
|
||||
comments, transcripts, videos, reviews, SERPs, full page content).
|
||||
- **The user's own context** — their knowledge base, connected apps, and
|
||||
persistent memory.
|
||||
- **Deliverables** — reports, podcasts, and presentations built from what the
|
||||
|
|
|
|||
|
|
@ -6,9 +6,9 @@ reviews are moving, and what is being said across the open web — and to put
|
|||
that intelligence to work alongside the team's shared knowledge base.
|
||||
|
||||
You do this by dispatching **specialist subagents** via the `task` tool:
|
||||
- **Live market data** — Reddit, YouTube, Google Maps, Google Search, and the
|
||||
web crawler return structured, current platform data (posts, comments,
|
||||
transcripts, reviews, SERPs, full page content).
|
||||
- **Live market data** — Reddit, YouTube, TikTok, Google Maps, Google Search,
|
||||
and the web crawler return structured, current platform data (posts,
|
||||
comments, transcripts, videos, reviews, SERPs, full page content).
|
||||
- **The team's own context** — its shared knowledge base, connected apps, and
|
||||
persistent team memory.
|
||||
- **Deliverables** — reports, podcasts, and presentations built from what the
|
||||
|
|
|
|||
|
|
@ -1,8 +1,9 @@
|
|||
<knowledge_base_first>
|
||||
CRITICAL — ground factual answers in what you actually receive this turn:
|
||||
- **live platform data** via the market specialists —
|
||||
`task(reddit, ...)`, `task(youtube, ...)`, `task(google_maps, ...)`,
|
||||
`task(google_search, ...)`, `task(web_crawler, ...)`. Anything about
|
||||
`task(reddit, ...)`, `task(youtube, ...)`, `task(tiktok, ...)`,
|
||||
`task(google_maps, ...)`, `task(google_search, ...)`,
|
||||
`task(web_crawler, ...)`. Anything about
|
||||
competitors, markets, rankings, reviews, or audience sentiment is answered
|
||||
from what these return **this turn**, never from your training data: your
|
||||
general knowledge of companies, prices, and rankings is stale by definition,
|
||||
|
|
|
|||
|
|
@ -31,6 +31,7 @@ bounded fan-out (≤20 sites) the user already requested.
|
|||
about a brand, product, or topic is answered from the platform where they
|
||||
say it — `task(reddit, …)` for community discussion and threads,
|
||||
`task(youtube, …)` for video content, transcripts, and comment sections,
|
||||
`task(tiktok, …)` for short-form video trends by hashtag or search,
|
||||
`task(google_maps, …)` for customer reviews of physical businesses. Web
|
||||
search only finds articles *about* the conversation; the platform
|
||||
specialists return the conversation itself, structured and current. For
|
||||
|
|
|
|||
|
|
@ -0,0 +1 @@
|
|||
"""``tiktok`` builtin subagent: structured public TikTok videos and listings."""
|
||||
|
|
@ -0,0 +1,43 @@
|
|||
"""``tiktok`` route: ``SurfSenseSubagentSpec`` builder for deepagents."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
from langchain_core.language_models import BaseChatModel
|
||||
from langchain_core.tools import BaseTool
|
||||
|
||||
from app.agents.chat.multi_agent_chat.subagents.shared.md_file_reader import (
|
||||
read_md_file,
|
||||
)
|
||||
from app.agents.chat.multi_agent_chat.subagents.shared.spec import SurfSenseSubagentSpec
|
||||
from app.agents.chat.multi_agent_chat.subagents.shared.subagent_builder import (
|
||||
pack_subagent,
|
||||
)
|
||||
|
||||
from .tools.index import NAME, RULESET, load_tools
|
||||
|
||||
|
||||
def build_subagent(
|
||||
*,
|
||||
dependencies: dict[str, Any],
|
||||
model: BaseChatModel | None = None,
|
||||
middleware_stack: dict[str, Any] | None = None,
|
||||
mcp_tools: list[BaseTool] | None = None,
|
||||
) -> SurfSenseSubagentSpec:
|
||||
tools = [*load_tools(dependencies=dependencies), *(mcp_tools or [])]
|
||||
description = (
|
||||
read_md_file(__package__, "description").strip()
|
||||
or "Pulls structured data from public TikTok videos, hashtags, and searches."
|
||||
)
|
||||
system_prompt = read_md_file(__package__, "system_prompt").strip()
|
||||
return pack_subagent(
|
||||
name=NAME,
|
||||
description=description,
|
||||
system_prompt=system_prompt,
|
||||
tools=tools,
|
||||
ruleset=RULESET,
|
||||
dependencies=dependencies,
|
||||
model=model,
|
||||
middleware_stack=middleware_stack,
|
||||
)
|
||||
|
|
@ -0,0 +1,2 @@
|
|||
TikTok specialist: pulls structured public TikTok data — videos (caption/text, author, play/like/comment/share counts, music, hashtags, timestamps, web URL) from a hashtag feed, a creator profile, or a known video URL; a video's public comment thread; accounts found by keyword; and the current Explore trending-video feed. Also compares fresh TikTok results against earlier findings in this chat.
|
||||
Use whenever the task is to find what is trending or being said on TikTok about a topic, gather a creator's or hashtag's videos, read a video's comments, discover accounts by keyword, or scrape a specific video URL. Triggers include "search TikTok for X", "what's trending on TikTok", "videos with #X", "comments on this TikTok", "find TikTok accounts about X", and "scrape this TikTok video". Not for general web pages (use the web crawling specialist), Google results (use the Google Search specialist), Reddit (use the Reddit specialist), or YouTube (use the YouTube specialist).
|
||||
|
|
@ -0,0 +1,70 @@
|
|||
You are the SurfSense TikTok sub-agent.
|
||||
You receive delegated instructions from a supervisor agent and return structured results for supervisor synthesis.
|
||||
|
||||
<goal>
|
||||
Answer the delegated question from live TikTok data gathered with your verb, comparing against earlier results already in this conversation when the task calls for it.
|
||||
</goal>
|
||||
|
||||
<available_tools>
|
||||
- `tiktok_scrape` — videos from a hashtag, a profile, or a TikTok URL
|
||||
- `tiktok_comments` — a video's comment thread, from `video_urls`
|
||||
- `tiktok_user_search` — find accounts by keyword, from `queries`
|
||||
- `tiktok_trending` — the current Explore trending-video feed
|
||||
- `read_run` / `search_run` (free readers for stored scrape output)
|
||||
</available_tools>
|
||||
|
||||
<playbook>
|
||||
- Finding videos on a topic: call `tiktok_scrape` with `hashtags` (no leading '#'), or pass a TikTok URL in `urls`.
|
||||
- Scraping a specific video, profile, hashtag, or search page: pass its TikTok URL in `urls`.
|
||||
- Profiles: a creator's `profiles` feed returns the account's metadata (name, followers, bio, verification) reliably, but its video list is often withheld by TikTok — treat an empty video list as a known limit, not a failure to retry endlessly. Prefer `hashtags` or a direct video URL for videos.
|
||||
- Comments on a video: call `tiktok_comments` with the video URL(s) in `video_urls`.
|
||||
- Finding accounts by keyword: call `tiktok_user_search` with `queries`. Keyword search returns no videos, so do not use `search_queries` for videos — use `hashtags` or a video URL.
|
||||
- "What's trending now": call `tiktok_trending` (no query needed); set `max_items` for how many.
|
||||
- Controlling volume: use `max_items` for the total cap and `results_per_page` per target (per-verb equivalents: `comments_per_video`, `results_per_query`).
|
||||
- Requested counts: `max_items` defaults low — when the task asks for N items, set `max_items` (and the per-target count) above N. A call that caps below the target can never satisfy it.
|
||||
- Batch multiple hashtags, queries, or video URLs into one call rather than many single-item calls.
|
||||
<include snippet="run_reader"/>
|
||||
- Comparison requests: pull the current results, compare against prior values already in this conversation's earlier tool results, and report concrete deltas (added, removed, count changes).
|
||||
</playbook>
|
||||
|
||||
<tool_policy>
|
||||
- Use only tools in `<available_tools>`.
|
||||
- Report only results present in the tool output. Never invent captions, URLs, authors, or counts.
|
||||
</tool_policy>
|
||||
|
||||
<out_of_scope>
|
||||
- Do not read arbitrary web pages — that belongs to the web crawling specialist.
|
||||
- Do not generate deliverables or perform connector mutations; return findings for the supervisor to act on.
|
||||
- Reddit belongs to the Reddit specialist; YouTube belongs to the YouTube specialist; Google results belong to the Google Search specialist.
|
||||
</out_of_scope>
|
||||
|
||||
<safety>
|
||||
- Report uncertainty explicitly when evidence is incomplete or conflicting.
|
||||
- Never present unverified claims as facts.
|
||||
</safety>
|
||||
|
||||
<failure_policy>
|
||||
- Underspecified request — no usable hashtag, query, or URL — return `status=blocked` with the missing fields.
|
||||
- Tool failure: return `status=error` with a concise recovery `next_step`.
|
||||
- No useful evidence: return `status=blocked` with a narrower query or the scope you still need.
|
||||
</failure_policy>
|
||||
|
||||
<output_contract>
|
||||
Return **only** one JSON object (no markdown/prose):
|
||||
{
|
||||
"status": "success" | "partial" | "blocked" | "error",
|
||||
"action_summary": string,
|
||||
"evidence": {
|
||||
"findings": string[],
|
||||
"sources": string[],
|
||||
"confidence": "high" | "medium" | "low"
|
||||
},
|
||||
"next_step": string | null,
|
||||
"missing_fields": string[] | null,
|
||||
"assumptions": string[] | null
|
||||
}
|
||||
<include snippet="output_contract_base"/>
|
||||
Route-specific rules:
|
||||
- `evidence.findings`: one entry per distinct result (video, comment, or account) or delta — a single sentence each; do not paste raw payloads. Max 10 entries, unless the delegated task asks for N items: then return up to N (each backed by a real scraped result, never padded).
|
||||
- `evidence.sources`: one TikTok URL per finding when applicable, same cap as findings. List each URL once.
|
||||
</output_contract>
|
||||
|
|
@ -0,0 +1,30 @@
|
|||
"""``tiktok`` sub-agent tools: scrape, comments, user-search, and trending verbs."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
from langchain_core.tools import BaseTool
|
||||
|
||||
from app.agents.chat.multi_agent_chat.shared.permissions import Ruleset
|
||||
from app.capabilities.core.access.agent import build_capability_tools
|
||||
from app.capabilities.tiktok.comments.definition import TIKTOK_COMMENTS
|
||||
from app.capabilities.tiktok.scrape.definition import TIKTOK_SCRAPE
|
||||
from app.capabilities.tiktok.trending.definition import TIKTOK_TRENDING
|
||||
from app.capabilities.tiktok.user_search.definition import TIKTOK_USER_SEARCH
|
||||
|
||||
NAME = "tiktok"
|
||||
|
||||
RULESET = Ruleset(origin=NAME, rules=[])
|
||||
|
||||
_CI_VERBS = [TIKTOK_SCRAPE, TIKTOK_COMMENTS, TIKTOK_USER_SEARCH, TIKTOK_TRENDING]
|
||||
|
||||
|
||||
def load_tools(
|
||||
*, dependencies: dict[str, Any] | None = None, **kwargs: Any
|
||||
) -> list[BaseTool]:
|
||||
d = {**(dependencies or {}), **kwargs}
|
||||
return build_capability_tools(
|
||||
workspace_id=d.get("workspace_id"),
|
||||
capabilities=_CI_VERBS,
|
||||
)
|
||||
|
|
@ -36,6 +36,9 @@ from app.agents.chat.multi_agent_chat.subagents.builtins.memory.agent import (
|
|||
from app.agents.chat.multi_agent_chat.subagents.builtins.reddit.agent import (
|
||||
build_subagent as build_reddit_subagent,
|
||||
)
|
||||
from app.agents.chat.multi_agent_chat.subagents.builtins.tiktok.agent import (
|
||||
build_subagent as build_tiktok_subagent,
|
||||
)
|
||||
from app.agents.chat.multi_agent_chat.subagents.builtins.web_crawler.agent import (
|
||||
build_subagent as build_web_crawler_subagent,
|
||||
)
|
||||
|
|
@ -88,6 +91,7 @@ SUBAGENT_BUILDERS_BY_NAME: dict[str, SubagentBuilder] = {
|
|||
"memory": build_memory_subagent,
|
||||
"onedrive": build_onedrive_subagent,
|
||||
"reddit": build_reddit_subagent,
|
||||
"tiktok": build_tiktok_subagent,
|
||||
"web_crawler": build_web_crawler_subagent,
|
||||
"youtube": build_youtube_subagent,
|
||||
}
|
||||
|
|
|
|||
|
|
@ -37,6 +37,9 @@ _PLATFORM_RATE_KEYS: dict[BillingUnit, str] = {
|
|||
BillingUnit.YOUTUBE_COMMENT: "YOUTUBE_MICROS_PER_COMMENT",
|
||||
BillingUnit.INSTAGRAM_ITEM: "INSTAGRAM_SCRAPE_MICROS_PER_ITEM",
|
||||
BillingUnit.INSTAGRAM_COMMENT: "INSTAGRAM_SCRAPE_MICROS_PER_COMMENT",
|
||||
BillingUnit.TIKTOK_VIDEO: "TIKTOK_MICROS_PER_VIDEO",
|
||||
BillingUnit.TIKTOK_USER: "TIKTOK_MICROS_PER_USER",
|
||||
BillingUnit.TIKTOK_COMMENT: "TIKTOK_MICROS_PER_COMMENT",
|
||||
}
|
||||
|
||||
|
||||
|
|
@ -55,6 +58,9 @@ _UNIT_NOUNS: dict[BillingUnit, str] = {
|
|||
BillingUnit.YOUTUBE_COMMENT: "comment",
|
||||
BillingUnit.INSTAGRAM_ITEM: "item",
|
||||
BillingUnit.INSTAGRAM_COMMENT: "comment",
|
||||
BillingUnit.TIKTOK_VIDEO: "video",
|
||||
BillingUnit.TIKTOK_USER: "profile",
|
||||
BillingUnit.TIKTOK_COMMENT: "comment",
|
||||
}
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -27,6 +27,9 @@ class BillingUnit(StrEnum):
|
|||
YOUTUBE_COMMENT = "youtube_comment"
|
||||
INSTAGRAM_ITEM = "instagram_item"
|
||||
INSTAGRAM_COMMENT = "instagram_comment"
|
||||
TIKTOK_VIDEO = "tiktok_video"
|
||||
TIKTOK_USER = "tiktok_user"
|
||||
TIKTOK_COMMENT = "tiktok_comment"
|
||||
|
||||
|
||||
class BillableInput(Protocol):
|
||||
|
|
|
|||
8
surfsense_backend/app/capabilities/tiktok/__init__.py
Normal file
8
surfsense_backend/app/capabilities/tiktok/__init__.py
Normal file
|
|
@ -0,0 +1,8 @@
|
|||
"""``tiktok.*`` namespace: platform-native TikTok data verbs."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from app.capabilities.tiktok.comments import definition as _comments # noqa: F401
|
||||
from app.capabilities.tiktok.scrape import definition as _scrape # noqa: F401
|
||||
from app.capabilities.tiktok.trending import definition as _trending # noqa: F401
|
||||
from app.capabilities.tiktok.user_search import definition as _user_search # noqa: F401
|
||||
|
|
@ -0,0 +1,3 @@
|
|||
"""``tiktok.comments``: scrape the public comment thread of TikTok videos."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
|
@ -0,0 +1,23 @@
|
|||
"""``tiktok.comments`` capability registration (billed per comment; see config
|
||||
``TIKTOK_MICROS_PER_COMMENT``)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from app.capabilities.core import BillingUnit, Capability, register_capability
|
||||
from app.capabilities.tiktok.comments.executor import build_comments_executor
|
||||
from app.capabilities.tiktok.comments.schemas import CommentsInput, CommentsOutput
|
||||
|
||||
TIKTOK_COMMENTS = Capability(
|
||||
name="tiktok.comments",
|
||||
description=(
|
||||
"Scrape the public comments of TikTok videos. Provide video URLs; "
|
||||
"returns comment text, author, likes, and reply counts."
|
||||
),
|
||||
input_schema=CommentsInput,
|
||||
output_schema=CommentsOutput,
|
||||
executor=build_comments_executor(),
|
||||
billing_unit=BillingUnit.TIKTOK_COMMENT,
|
||||
docs_url="/docs/connectors/native/tiktok",
|
||||
)
|
||||
|
||||
register_capability(TIKTOK_COMMENTS)
|
||||
|
|
@ -0,0 +1,36 @@
|
|||
"""``tiktok.comments`` executor: video URLs -> scraper -> TikTok comment items."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Awaitable, Callable
|
||||
|
||||
from app.capabilities.core import Executor
|
||||
from app.capabilities.core.progress import emit_progress
|
||||
from app.capabilities.tiktok.comments.schemas import CommentsInput, CommentsOutput
|
||||
from app.proprietary.platforms.tiktok import scrape_tiktok_comments
|
||||
|
||||
CommentsFn = Callable[..., Awaitable[list[dict]]]
|
||||
|
||||
|
||||
def build_comments_executor(comments_fn: CommentsFn | None = None) -> Executor:
|
||||
"""Bind the executor to a comments fn (defaults to the proprietary actor)."""
|
||||
comments_fn = comments_fn or scrape_tiktok_comments
|
||||
|
||||
async def execute(payload: CommentsInput) -> CommentsOutput:
|
||||
emit_progress(
|
||||
"starting",
|
||||
"Scraping TikTok comments",
|
||||
total=payload.max_items,
|
||||
unit="item",
|
||||
)
|
||||
items = await comments_fn(
|
||||
payload.video_urls,
|
||||
per_video=payload.comments_per_video,
|
||||
limit=payload.max_items,
|
||||
)
|
||||
emit_progress(
|
||||
"done", f"Scraped {len(items)} comment(s)", current=len(items), unit="item"
|
||||
)
|
||||
return CommentsOutput(items=items)
|
||||
|
||||
return execute
|
||||
|
|
@ -0,0 +1,56 @@
|
|||
"""``tiktok.comments`` I/O contracts.
|
||||
|
||||
URL-only: given TikTok video URLs, return each video's public comment thread.
|
||||
Unlike profile-video/general-search feeds, ``/api/comment/list`` is served to
|
||||
anonymous sessions once the comments panel opens, so this verb is reliable. Each
|
||||
result is a :class:`CommentItem` (top-level comments; replies carry ``repliesToId``).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
from app.capabilities.tiktok.scrape.schemas import (
|
||||
MAX_TIKTOK_ITEMS,
|
||||
MAX_TIKTOK_SOURCES,
|
||||
)
|
||||
from app.proprietary.platforms.tiktok import CommentItem
|
||||
|
||||
|
||||
class CommentsInput(BaseModel):
|
||||
video_urls: list[str] = Field(
|
||||
min_length=1,
|
||||
max_length=MAX_TIKTOK_SOURCES,
|
||||
description="TikTok video URLs (/@<user>/video/<id>) to pull comments from.",
|
||||
)
|
||||
comments_per_video: int = Field(
|
||||
default=20,
|
||||
ge=1,
|
||||
le=MAX_TIKTOK_ITEMS,
|
||||
description="Max comments to return per video.",
|
||||
)
|
||||
max_items: int = Field(
|
||||
default=20,
|
||||
ge=1,
|
||||
le=MAX_TIKTOK_ITEMS,
|
||||
description="Max total comments to return across all videos.",
|
||||
)
|
||||
|
||||
@property
|
||||
def estimated_units(self) -> int:
|
||||
"""Worst-case billable comments for the pre-flight gate: ``max_items`` is a
|
||||
hard cross-video ceiling (le=100), so no call can exceed it."""
|
||||
return self.max_items
|
||||
|
||||
|
||||
class CommentsOutput(BaseModel):
|
||||
items: list[CommentItem] = Field(
|
||||
default_factory=list,
|
||||
description="One item per comment returned, in emission order.",
|
||||
)
|
||||
|
||||
@property
|
||||
def billable_units(self) -> int:
|
||||
"""One returned comment = one billable unit; ErrorItems (``errorCode`` set,
|
||||
for bad URLs or empty/withheld videos) are surfaced but never charged."""
|
||||
return sum(1 for item in self.items if not getattr(item, "errorCode", None))
|
||||
|
|
@ -0,0 +1 @@
|
|||
"""``tiktok.scrape``: public TikTok videos over the browser-driven scraper."""
|
||||
|
|
@ -0,0 +1,23 @@
|
|||
"""``tiktok.scrape`` capability registration (billed per video; see config
|
||||
``TIKTOK_MICROS_PER_VIDEO``)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from app.capabilities.core import BillingUnit, Capability, register_capability
|
||||
from app.capabilities.tiktok.scrape.executor import build_scrape_executor
|
||||
from app.capabilities.tiktok.scrape.schemas import ScrapeInput, ScrapeOutput
|
||||
|
||||
TIKTOK_SCRAPE = Capability(
|
||||
name="tiktok.scrape",
|
||||
description=(
|
||||
"Scrape public TikTok videos. Use urls, profiles, hashtags, or "
|
||||
"search_queries."
|
||||
),
|
||||
input_schema=ScrapeInput,
|
||||
output_schema=ScrapeOutput,
|
||||
executor=build_scrape_executor(),
|
||||
billing_unit=BillingUnit.TIKTOK_VIDEO,
|
||||
docs_url="/docs/connectors/native/tiktok",
|
||||
)
|
||||
|
||||
register_capability(TIKTOK_SCRAPE)
|
||||
48
surfsense_backend/app/capabilities/tiktok/scrape/executor.py
Normal file
48
surfsense_backend/app/capabilities/tiktok/scrape/executor.py
Normal file
|
|
@ -0,0 +1,48 @@
|
|||
"""``tiktok.scrape`` executor: verb input → scraper → TikTok video items."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Awaitable, Callable
|
||||
|
||||
from app.capabilities.core import Executor
|
||||
from app.capabilities.core.progress import emit_progress
|
||||
from app.capabilities.tiktok.scrape.schemas import ScrapeInput, ScrapeOutput
|
||||
from app.exceptions import ForbiddenError
|
||||
from app.proprietary.platforms.tiktok import (
|
||||
TikTokAccessBlockedError,
|
||||
TikTokScrapeInput,
|
||||
scrape_tiktok,
|
||||
)
|
||||
|
||||
ScrapeFn = Callable[..., Awaitable[list[dict]]]
|
||||
|
||||
|
||||
def build_scrape_executor(scrape_fn: ScrapeFn | None = None) -> Executor:
|
||||
"""Bind the executor to a scraper fn (defaults to the proprietary actor)."""
|
||||
scrape_fn = scrape_fn or scrape_tiktok
|
||||
|
||||
async def execute(payload: ScrapeInput) -> ScrapeOutput:
|
||||
actor_input = TikTokScrapeInput(
|
||||
startUrls=[{"url": url} for url in payload.urls],
|
||||
profiles=payload.profiles,
|
||||
hashtags=payload.hashtags,
|
||||
searchQueries=payload.search_queries,
|
||||
resultsPerPage=payload.results_per_page,
|
||||
)
|
||||
emit_progress(
|
||||
"starting", "Resolving TikTok targets", total=payload.max_items, unit="item"
|
||||
)
|
||||
try:
|
||||
items = await scrape_fn(actor_input, limit=payload.max_items)
|
||||
except TikTokAccessBlockedError as exc:
|
||||
# Anonymous-only scraper; a hard block can't be retried with creds.
|
||||
raise ForbiddenError(
|
||||
f"TikTok refused anonymous access: {exc}",
|
||||
code="TIKTOK_ACCESS_BLOCKED",
|
||||
) from exc
|
||||
emit_progress(
|
||||
"done", f"Scraped {len(items)} item(s)", current=len(items), unit="item"
|
||||
)
|
||||
return ScrapeOutput(items=items)
|
||||
|
||||
return execute
|
||||
87
surfsense_backend/app/capabilities/tiktok/scrape/schemas.py
Normal file
87
surfsense_backend/app/capabilities/tiktok/scrape/schemas.py
Normal file
|
|
@ -0,0 +1,87 @@
|
|||
"""``tiktok.scrape`` I/O contracts.
|
||||
|
||||
A lean, agent-friendly surface over ``TikTokScrapeInput``
|
||||
(``app/proprietary/platforms/tiktok``). The executor maps this to the full
|
||||
scraper input; the scraper's ``TikTokVideoItem`` is reused verbatim as the
|
||||
output element. Any TikTok URL kind (video, profile, hashtag, search) goes in
|
||||
``urls``; ``profiles``/``hashtags``/``search_queries`` are typed shortcuts.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pydantic import BaseModel, Field, model_validator
|
||||
|
||||
from app.proprietary.platforms.tiktok import TikTokVideoItem
|
||||
|
||||
MAX_TIKTOK_SOURCES = 20
|
||||
"""Per-call cap on each source list: bounds a synchronous request's fan-out."""
|
||||
|
||||
MAX_TIKTOK_ITEMS = 100
|
||||
"""Hard ceiling on items returned per call, regardless of the per-target count."""
|
||||
|
||||
|
||||
class ScrapeInput(BaseModel):
|
||||
urls: list[str] = Field(
|
||||
default_factory=list,
|
||||
max_length=MAX_TIKTOK_SOURCES,
|
||||
description=(
|
||||
"TikTok URLs to scrape: a video, a profile (/@<user>), a hashtag "
|
||||
"(/tag/<name>), or a search URL. Provide these OR profiles/hashtags/"
|
||||
"search_queries (at least one source is required)."
|
||||
),
|
||||
)
|
||||
profiles: list[str] = Field(
|
||||
default_factory=list,
|
||||
max_length=MAX_TIKTOK_SOURCES,
|
||||
description="Profile usernames (with or without a leading '@').",
|
||||
)
|
||||
hashtags: list[str] = Field(
|
||||
default_factory=list,
|
||||
max_length=MAX_TIKTOK_SOURCES,
|
||||
description="Hashtag names to scrape, without the leading '#'.",
|
||||
)
|
||||
search_queries: list[str] = Field(
|
||||
default_factory=list,
|
||||
max_length=MAX_TIKTOK_SOURCES,
|
||||
description="Search terms to run on TikTok.",
|
||||
)
|
||||
results_per_page: int = Field(
|
||||
default=10,
|
||||
ge=1,
|
||||
le=MAX_TIKTOK_ITEMS,
|
||||
description="Max videos to pull per profile/hashtag/search target.",
|
||||
)
|
||||
max_items: int = Field(
|
||||
default=10,
|
||||
ge=1,
|
||||
le=MAX_TIKTOK_ITEMS,
|
||||
description="Max total items to return across all sources.",
|
||||
)
|
||||
|
||||
@model_validator(mode="after")
|
||||
def _require_a_source(self) -> ScrapeInput:
|
||||
if not any((self.urls, self.profiles, self.hashtags, self.search_queries)):
|
||||
raise ValueError(
|
||||
"Provide at least one of 'urls', 'profiles', 'hashtags', or "
|
||||
"'search_queries'."
|
||||
)
|
||||
return self
|
||||
|
||||
@property
|
||||
def estimated_units(self) -> int:
|
||||
"""Worst-case billable items for the pre-flight gate: ``max_items`` is a
|
||||
hard cross-source ceiling (le=100), so no call can exceed it."""
|
||||
return self.max_items
|
||||
|
||||
|
||||
class ScrapeOutput(BaseModel):
|
||||
items: list[TikTokVideoItem] = Field(
|
||||
default_factory=list,
|
||||
description="One item per video returned, in emission order.",
|
||||
)
|
||||
|
||||
@property
|
||||
def billable_units(self) -> int:
|
||||
"""One returned video = one billable unit; ErrorItems (``errorCode`` set,
|
||||
for blocked/empty targets) are surfaced but never charged."""
|
||||
return sum(1 for item in self.items if not getattr(item, "errorCode", None))
|
||||
|
|
@ -0,0 +1,3 @@
|
|||
"""``tiktok.trending``: pull the current trending videos from the Explore feed."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
|
@ -0,0 +1,23 @@
|
|||
"""``tiktok.trending`` capability registration (billed per video on the shared
|
||||
``TIKTOK_MICROS_PER_VIDEO`` meter)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from app.capabilities.core import BillingUnit, Capability, register_capability
|
||||
from app.capabilities.tiktok.trending.executor import build_trending_executor
|
||||
from app.capabilities.tiktok.trending.schemas import TrendingInput, TrendingOutput
|
||||
|
||||
TIKTOK_TRENDING = Capability(
|
||||
name="tiktok.trending",
|
||||
description=(
|
||||
"Get the current trending TikTok videos from the Explore feed. No input "
|
||||
"needed beyond how many to return."
|
||||
),
|
||||
input_schema=TrendingInput,
|
||||
output_schema=TrendingOutput,
|
||||
executor=build_trending_executor(),
|
||||
billing_unit=BillingUnit.TIKTOK_VIDEO,
|
||||
docs_url="/docs/connectors/native/tiktok",
|
||||
)
|
||||
|
||||
register_capability(TIKTOK_TRENDING)
|
||||
|
|
@ -0,0 +1,32 @@
|
|||
"""``tiktok.trending`` executor: Explore feed -> scraper -> TikTok video items."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Awaitable, Callable
|
||||
|
||||
from app.capabilities.core import Executor
|
||||
from app.capabilities.core.progress import emit_progress
|
||||
from app.capabilities.tiktok.trending.schemas import TrendingInput, TrendingOutput
|
||||
from app.proprietary.platforms.tiktok import scrape_tiktok_trending
|
||||
|
||||
TrendingFn = Callable[..., Awaitable[list[dict]]]
|
||||
|
||||
|
||||
def build_trending_executor(trending_fn: TrendingFn | None = None) -> Executor:
|
||||
"""Bind the executor to a trending fn (defaults to the proprietary actor)."""
|
||||
trending_fn = trending_fn or scrape_tiktok_trending
|
||||
|
||||
async def execute(payload: TrendingInput) -> TrendingOutput:
|
||||
emit_progress(
|
||||
"starting",
|
||||
"Fetching TikTok trending videos",
|
||||
total=payload.max_items,
|
||||
unit="item",
|
||||
)
|
||||
items = await trending_fn(count=payload.max_items)
|
||||
emit_progress(
|
||||
"done", f"Fetched {len(items)} video(s)", current=len(items), unit="item"
|
||||
)
|
||||
return TrendingOutput(items=items)
|
||||
|
||||
return execute
|
||||
|
|
@ -0,0 +1,41 @@
|
|||
"""``tiktok.trending`` I/O contracts.
|
||||
|
||||
The Explore feed (``/api/explore/item_list``) is a single global feed of trending
|
||||
videos, served to anonymous sessions. No source input is needed — just how many
|
||||
to return. Each result reuses :class:`TikTokVideoItem`, so trending videos bill on
|
||||
the same per-video meter as ``tiktok.scrape``.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
from app.capabilities.tiktok.scrape.schemas import MAX_TIKTOK_ITEMS
|
||||
from app.proprietary.platforms.tiktok import TikTokVideoItem
|
||||
|
||||
|
||||
class TrendingInput(BaseModel):
|
||||
max_items: int = Field(
|
||||
default=20,
|
||||
ge=1,
|
||||
le=MAX_TIKTOK_ITEMS,
|
||||
description="Max trending videos to return from the Explore feed.",
|
||||
)
|
||||
|
||||
@property
|
||||
def estimated_units(self) -> int:
|
||||
"""Worst-case billable videos for the pre-flight gate (le=100 ceiling)."""
|
||||
return self.max_items
|
||||
|
||||
|
||||
class TrendingOutput(BaseModel):
|
||||
items: list[TikTokVideoItem] = Field(
|
||||
default_factory=list,
|
||||
description="One item per trending video returned, in feed order.",
|
||||
)
|
||||
|
||||
@property
|
||||
def billable_units(self) -> int:
|
||||
"""One returned video = one billable unit; an ErrorItem (``errorCode`` set,
|
||||
for an empty/withheld feed) is surfaced but never charged."""
|
||||
return sum(1 for item in self.items if not getattr(item, "errorCode", None))
|
||||
|
|
@ -0,0 +1,3 @@
|
|||
"""``tiktok.user_search``: find public TikTok accounts by keyword."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
|
@ -0,0 +1,26 @@
|
|||
"""``tiktok.user_search`` capability registration (billed per account; see config
|
||||
``TIKTOK_MICROS_PER_USER``)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from app.capabilities.core import BillingUnit, Capability, register_capability
|
||||
from app.capabilities.tiktok.user_search.executor import build_user_search_executor
|
||||
from app.capabilities.tiktok.user_search.schemas import (
|
||||
UserSearchInput,
|
||||
UserSearchOutput,
|
||||
)
|
||||
|
||||
TIKTOK_USER_SEARCH = Capability(
|
||||
name="tiktok.user_search",
|
||||
description=(
|
||||
"Find public TikTok accounts by keyword. Returns profile metadata "
|
||||
"(name, followers, bio, verification) per matching account."
|
||||
),
|
||||
input_schema=UserSearchInput,
|
||||
output_schema=UserSearchOutput,
|
||||
executor=build_user_search_executor(),
|
||||
billing_unit=BillingUnit.TIKTOK_USER,
|
||||
docs_url="/docs/connectors/native/tiktok",
|
||||
)
|
||||
|
||||
register_capability(TIKTOK_USER_SEARCH)
|
||||
|
|
@ -0,0 +1,39 @@
|
|||
"""``tiktok.user_search`` executor: queries -> scraper -> TikTok profile items."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Awaitable, Callable
|
||||
|
||||
from app.capabilities.core import Executor
|
||||
from app.capabilities.core.progress import emit_progress
|
||||
from app.capabilities.tiktok.user_search.schemas import (
|
||||
UserSearchInput,
|
||||
UserSearchOutput,
|
||||
)
|
||||
from app.proprietary.platforms.tiktok import search_tiktok_users
|
||||
|
||||
SearchFn = Callable[..., Awaitable[list[dict]]]
|
||||
|
||||
|
||||
def build_user_search_executor(search_fn: SearchFn | None = None) -> Executor:
|
||||
"""Bind the executor to a search fn (defaults to the proprietary actor)."""
|
||||
search_fn = search_fn or search_tiktok_users
|
||||
|
||||
async def execute(payload: UserSearchInput) -> UserSearchOutput:
|
||||
emit_progress(
|
||||
"starting",
|
||||
"Searching TikTok accounts",
|
||||
total=payload.max_items,
|
||||
unit="item",
|
||||
)
|
||||
items = await search_fn(
|
||||
payload.queries,
|
||||
per_query=payload.results_per_query,
|
||||
limit=payload.max_items,
|
||||
)
|
||||
emit_progress(
|
||||
"done", f"Found {len(items)} account(s)", current=len(items), unit="item"
|
||||
)
|
||||
return UserSearchOutput(items=items)
|
||||
|
||||
return execute
|
||||
|
|
@ -0,0 +1,56 @@
|
|||
"""``tiktok.user_search`` I/O contracts.
|
||||
|
||||
Account discovery over ``TikTok``'s Users tab. Where video/general search is
|
||||
login-walled for anonymous sessions, ``/api/search/user`` returns public account
|
||||
records, so this verb exposes the one reliably-unblocked search path. Each result
|
||||
is a :class:`TikTokProfileItem` (the same shape the profile verb emits).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
from app.capabilities.tiktok.scrape.schemas import (
|
||||
MAX_TIKTOK_ITEMS,
|
||||
MAX_TIKTOK_SOURCES,
|
||||
)
|
||||
from app.proprietary.platforms.tiktok import TikTokProfileItem
|
||||
|
||||
|
||||
class UserSearchInput(BaseModel):
|
||||
queries: list[str] = Field(
|
||||
min_length=1,
|
||||
max_length=MAX_TIKTOK_SOURCES,
|
||||
description="Keywords to search for TikTok accounts (e.g. names, brands).",
|
||||
)
|
||||
results_per_query: int = Field(
|
||||
default=10,
|
||||
ge=1,
|
||||
le=MAX_TIKTOK_ITEMS,
|
||||
description="Max accounts to return per query.",
|
||||
)
|
||||
max_items: int = Field(
|
||||
default=10,
|
||||
ge=1,
|
||||
le=MAX_TIKTOK_ITEMS,
|
||||
description="Max total accounts to return across all queries.",
|
||||
)
|
||||
|
||||
@property
|
||||
def estimated_units(self) -> int:
|
||||
"""Worst-case billable accounts for the pre-flight gate: ``max_items`` is a
|
||||
hard cross-query ceiling (le=100), so no call can exceed it."""
|
||||
return self.max_items
|
||||
|
||||
|
||||
class UserSearchOutput(BaseModel):
|
||||
items: list[TikTokProfileItem] = Field(
|
||||
default_factory=list,
|
||||
description="One item per account found, in emission order.",
|
||||
)
|
||||
|
||||
@property
|
||||
def billable_units(self) -> int:
|
||||
"""One returned account = one billable unit; ErrorItems (``errorCode`` set,
|
||||
for empty/withheld queries) are surfaced but never charged."""
|
||||
return sum(1 for item in self.items if not getattr(item, "errorCode", None))
|
||||
|
|
@ -727,6 +727,18 @@ class Config:
|
|||
INSTAGRAM_SCRAPE_MICROS_PER_COMMENT = int(
|
||||
os.getenv("INSTAGRAM_SCRAPE_MICROS_PER_COMMENT", "1500")
|
||||
)
|
||||
# Browser-driven listings make TikTok heavier per item than the API-backed
|
||||
# video meter, so it sits a touch above YouTube's video rate.
|
||||
TIKTOK_MICROS_PER_VIDEO = int(os.getenv("TIKTOK_MICROS_PER_VIDEO", "3500"))
|
||||
# User search returns lighter account records (name/followers/bio), priced
|
||||
# below the video meter to mirror the cheaper account-discovery market.
|
||||
TIKTOK_MICROS_PER_USER = int(os.getenv("TIKTOK_MICROS_PER_USER", "2500"))
|
||||
# Comments are the cheapest per-item TikTok data, matching the per-comment
|
||||
# market (and YouTube's comment meter).
|
||||
TIKTOK_MICROS_PER_COMMENT = int(os.getenv("TIKTOK_MICROS_PER_COMMENT", "1500"))
|
||||
# Retry an empty listing draw on a fresh rotating IP. Set to 1 for a static
|
||||
# proxy, where every retry re-hits the same exit.
|
||||
TIKTOK_LISTING_MAX_ATTEMPTS = int(os.getenv("TIKTOK_LISTING_MAX_ATTEMPTS", "3"))
|
||||
|
||||
# Low-balance WARNING threshold (micro-USD). Surfaced by the quota service
|
||||
# so the UI can nudge the user to top up / enable auto-reload. $0.50.
|
||||
|
|
@ -1146,6 +1158,11 @@ class Config:
|
|||
# round-trip => default FALSE to honor the "no speed regression" bar; flip on
|
||||
# when leak-safety outweighs the marginal latency.
|
||||
CRAWL_DNS_OVER_HTTPS = os.getenv("CRAWL_DNS_OVER_HTTPS", "FALSE").upper() == "TRUE"
|
||||
# Promises an Xvfb display so the browser can run headful (TikTok's profile
|
||||
# feed is empty to headless Chromium). Off keeps every browser headless.
|
||||
CRAWL_HEADED_XVFB_ENABLED = (
|
||||
os.getenv("CRAWL_HEADED_XVFB_ENABLED", "FALSE").upper() == "TRUE"
|
||||
)
|
||||
|
||||
# Litellm TTS Configuration
|
||||
TTS_SERVICE = os.getenv("TTS_SERVICE")
|
||||
|
|
|
|||
|
|
@ -0,0 +1,35 @@
|
|||
"""Anonymous, blob-first TikTok scraper (public interface).
|
||||
|
||||
The capability layer depends only on the names re-exported here: the input
|
||||
schema, the collector/generator, the video item shape, and the hard-block error.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from .orchestrator import (
|
||||
iter_tiktok,
|
||||
scrape_tiktok,
|
||||
scrape_tiktok_comments,
|
||||
scrape_tiktok_trending,
|
||||
search_tiktok_users,
|
||||
)
|
||||
from .schemas import (
|
||||
CommentItem,
|
||||
TikTokProfileItem,
|
||||
TikTokScrapeInput,
|
||||
TikTokVideoItem,
|
||||
)
|
||||
from .session import TikTokAccessBlockedError
|
||||
|
||||
__all__ = [
|
||||
"CommentItem",
|
||||
"TikTokAccessBlockedError",
|
||||
"TikTokProfileItem",
|
||||
"TikTokScrapeInput",
|
||||
"TikTokVideoItem",
|
||||
"iter_tiktok",
|
||||
"scrape_tiktok",
|
||||
"scrape_tiktok_comments",
|
||||
"scrape_tiktok_trending",
|
||||
"search_tiktok_users",
|
||||
]
|
||||
|
|
@ -0,0 +1,25 @@
|
|||
"""Turn raw TikTok page/API payloads into normalized items."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from .author import parse_author, parse_profile
|
||||
from .comments import comments_from_response, parse_comment
|
||||
from .hydration import extract_rehydration_data
|
||||
from .item_list import items_from_response
|
||||
from .scopes import user_info, video_item_struct
|
||||
from .user_search import parse_search_user, users_from_response
|
||||
from .video import parse_video
|
||||
|
||||
__all__ = [
|
||||
"comments_from_response",
|
||||
"extract_rehydration_data",
|
||||
"items_from_response",
|
||||
"parse_author",
|
||||
"parse_comment",
|
||||
"parse_profile",
|
||||
"parse_search_user",
|
||||
"parse_video",
|
||||
"user_info",
|
||||
"users_from_response",
|
||||
"video_item_struct",
|
||||
]
|
||||
|
|
@ -0,0 +1,38 @@
|
|||
"""Normalize TikTok author/profile payloads into an ``authorMeta`` dict."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
_PROFILE_URL = "https://www.tiktok.com/@{username}"
|
||||
|
||||
|
||||
def build_author_meta(author: dict[str, Any], stats: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Map an author object + its stats to the ``authorMeta`` output shape."""
|
||||
username = author.get("uniqueId")
|
||||
return {
|
||||
"id": author.get("id"),
|
||||
"name": username,
|
||||
"nickName": author.get("nickname"),
|
||||
"profileUrl": _PROFILE_URL.format(username=username) if username else None,
|
||||
"verified": author.get("verified"),
|
||||
"signature": author.get("signature"),
|
||||
"avatar": author.get("avatarLarger") or author.get("avatarMedium"),
|
||||
"privateAccount": author.get("privateAccount"),
|
||||
"fans": stats.get("followerCount"),
|
||||
"following": stats.get("followingCount"),
|
||||
"heart": stats.get("heartCount"),
|
||||
"video": stats.get("videoCount"),
|
||||
}
|
||||
|
||||
|
||||
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()
|
||||
|
|
@ -0,0 +1,55 @@
|
|||
"""Parse a captured ``/api/comment/list`` response into comment items.
|
||||
|
||||
The comment API returns ``{"comments": [...]}`` where each entry uses the
|
||||
mobile-API snake_case shape (``cid``, ``digg_count``, ``reply_comment_total``,
|
||||
``create_time``, and a nested ``user`` with ``uid``/``unique_id``/``nickname``/
|
||||
``avatar_thumb``). ``reply_id != "0"`` marks a reply to a parent comment.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
from .timestamps import epoch_to_iso
|
||||
|
||||
|
||||
def comments_from_response(body: Any) -> list[dict[str, Any]]:
|
||||
"""Return the raw comment records carried by one API response, or ``[]``."""
|
||||
if not isinstance(body, dict):
|
||||
return []
|
||||
comments = body.get("comments")
|
||||
if not isinstance(comments, list):
|
||||
return []
|
||||
return [c for c in comments if isinstance(c, dict)]
|
||||
|
||||
|
||||
def _avatar(user: dict[str, Any]) -> str | None:
|
||||
thumb = user.get("avatar_thumb")
|
||||
if isinstance(thumb, dict):
|
||||
urls = thumb.get("url_list")
|
||||
if isinstance(urls, list) and urls:
|
||||
return urls[0]
|
||||
return None
|
||||
|
||||
|
||||
def parse_comment(raw: dict[str, Any], video_url: str) -> dict[str, Any]:
|
||||
"""Map a raw comment record to a :class:`CommentItem` output dict."""
|
||||
from ..schemas.items import CommentItem
|
||||
|
||||
user = raw.get("user") if isinstance(raw.get("user"), dict) else {}
|
||||
reply_id = raw.get("reply_id")
|
||||
create_time = raw.get("create_time")
|
||||
return CommentItem(
|
||||
id=raw.get("cid"),
|
||||
text=raw.get("text"),
|
||||
videoWebUrl=video_url,
|
||||
diggCount=raw.get("digg_count"),
|
||||
replyCommentTotal=raw.get("reply_comment_total"),
|
||||
createTime=create_time,
|
||||
createTimeISO=epoch_to_iso(create_time),
|
||||
uid=user.get("uid"),
|
||||
uniqueId=user.get("unique_id"),
|
||||
nickName=user.get("nickname"),
|
||||
avatar=_avatar(user),
|
||||
repliesToId=reply_id if reply_id and reply_id != "0" else None,
|
||||
).to_output()
|
||||
|
|
@ -0,0 +1,28 @@
|
|||
"""Extract the ``__UNIVERSAL_DATA_FOR_REHYDRATION__`` JSON embedded in page HTML.
|
||||
|
||||
TikTok server-renders the first page of data into a single script tag; parsing
|
||||
it yields page-one items (video/profile/hashtag) without any signed API call.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import re
|
||||
from typing import Any
|
||||
|
||||
_BLOB_RE = re.compile(
|
||||
r'<script[^>]*id="__UNIVERSAL_DATA_FOR_REHYDRATION__"[^>]*>(.*?)</script>',
|
||||
re.DOTALL,
|
||||
)
|
||||
|
||||
|
||||
def extract_rehydration_data(html: str) -> dict[str, Any] | None:
|
||||
"""Return the parsed rehydration blob, or ``None`` if absent/unparseable."""
|
||||
match = _BLOB_RE.search(html)
|
||||
if not match:
|
||||
return None
|
||||
try:
|
||||
data = json.loads(match.group(1))
|
||||
except (json.JSONDecodeError, ValueError):
|
||||
return None
|
||||
return data if isinstance(data, dict) else None
|
||||
|
|
@ -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 []
|
||||
|
|
@ -0,0 +1,30 @@
|
|||
"""Navigate the rehydration blob to the scopes the flows consume."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
_DEFAULT = "__DEFAULT_SCOPE__"
|
||||
|
||||
|
||||
def _scope(data: dict[str, Any], name: str) -> dict[str, Any] | None:
|
||||
scope = (data.get(_DEFAULT) or {}).get(name)
|
||||
return scope if isinstance(scope, dict) else None
|
||||
|
||||
|
||||
def video_item_struct(data: dict[str, Any]) -> dict[str, Any] | None:
|
||||
"""The ``itemStruct`` of a video-detail page, or ``None``."""
|
||||
scope = _scope(data, "webapp.video-detail")
|
||||
if not scope:
|
||||
return None
|
||||
item = (scope.get("itemInfo") or {}).get("itemStruct")
|
||||
return item if isinstance(item, dict) else None
|
||||
|
||||
|
||||
def user_info(data: dict[str, Any]) -> dict[str, Any] | None:
|
||||
"""The ``userInfo`` (``{user, stats}``) of a profile page, or ``None``."""
|
||||
scope = _scope(data, "webapp.user-detail")
|
||||
if not scope:
|
||||
return None
|
||||
info = scope.get("userInfo")
|
||||
return info if isinstance(info, dict) else None
|
||||
|
|
@ -0,0 +1,18 @@
|
|||
"""Millisecond-ISO timestamps matching the actor's output shape."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import UTC, datetime
|
||||
|
||||
|
||||
def epoch_to_iso(seconds: int | None) -> str | None:
|
||||
"""Convert a Unix-seconds timestamp to ``YYYY-MM-DDTHH:MM:SS.000Z``."""
|
||||
if not seconds:
|
||||
return None
|
||||
stamp = datetime.fromtimestamp(seconds, tz=UTC)
|
||||
return stamp.strftime("%Y-%m-%dT%H:%M:%S.000Z")
|
||||
|
||||
|
||||
def now_iso() -> str:
|
||||
"""Current UTC time in the millisecond-ISO output shape."""
|
||||
return datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z"
|
||||
|
|
@ -0,0 +1,56 @@
|
|||
"""Parse the ``/api/search/user`` response into profile items.
|
||||
|
||||
User search returns ``{"user_list": [{"user_info": {...}}, ...]}`` where each
|
||||
``user_info`` uses the mobile-API snake_case shape (``uid``, ``unique_id``,
|
||||
``follower_count``, ``total_favorited``, ``avatar_thumb.url_list``) — distinct
|
||||
from the camelCase ``webapp.user-detail`` blob the profile flow reads, so it gets
|
||||
its own mapping into the shared :class:`TikTokProfileItem` output contract.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
_PROFILE_URL = "https://www.tiktok.com/@{username}"
|
||||
|
||||
|
||||
def users_from_response(body: Any) -> list[dict[str, Any]]:
|
||||
"""Return the ``user_info`` objects carried by one search response, or ``[]``."""
|
||||
if not isinstance(body, dict):
|
||||
return []
|
||||
user_list = body.get("user_list")
|
||||
if not isinstance(user_list, list):
|
||||
return []
|
||||
return [
|
||||
entry["user_info"]
|
||||
for entry in user_list
|
||||
if isinstance(entry, dict) and isinstance(entry.get("user_info"), dict)
|
||||
]
|
||||
|
||||
|
||||
def _avatar(user_info: dict[str, Any]) -> str | None:
|
||||
thumb = user_info.get("avatar_thumb")
|
||||
if isinstance(thumb, dict):
|
||||
urls = thumb.get("url_list")
|
||||
if isinstance(urls, list) and urls:
|
||||
return urls[0]
|
||||
return None
|
||||
|
||||
|
||||
def parse_search_user(user_info: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Map a search ``user_info`` to a :class:`TikTokProfileItem` output dict."""
|
||||
from ..schemas.items import TikTokProfileItem
|
||||
|
||||
username = user_info.get("unique_id")
|
||||
return TikTokProfileItem(
|
||||
id=user_info.get("uid"),
|
||||
name=username,
|
||||
nickName=user_info.get("nickname"),
|
||||
profileUrl=_PROFILE_URL.format(username=username) if username else None,
|
||||
verified=bool(user_info.get("enterprise_verify_reason")),
|
||||
signature=user_info.get("signature"),
|
||||
avatar=_avatar(user_info),
|
||||
fans=user_info.get("follower_count"),
|
||||
heart=user_info.get("total_favorited"),
|
||||
secUid=user_info.get("sec_uid"),
|
||||
).to_output()
|
||||
|
|
@ -0,0 +1,73 @@
|
|||
"""Normalize a raw TikTok ``itemStruct`` into a :class:`TikTokVideoItem` dict."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
from ..schemas.items import TikTokVideoItem
|
||||
from .author import build_author_meta
|
||||
from .timestamps import epoch_to_iso
|
||||
|
||||
_VIDEO_URL = "https://www.tiktok.com/@{username}/video/{video_id}"
|
||||
|
||||
|
||||
def _music_meta(music: dict[str, Any]) -> dict[str, Any]:
|
||||
return {
|
||||
"musicId": music.get("id"),
|
||||
"musicName": music.get("title"),
|
||||
"musicAuthor": music.get("authorName"),
|
||||
"musicOriginal": music.get("original"),
|
||||
"playUrl": music.get("playUrl"),
|
||||
}
|
||||
|
||||
|
||||
def _video_meta(video: dict[str, Any]) -> dict[str, Any]:
|
||||
return {
|
||||
"height": video.get("height"),
|
||||
"width": video.get("width"),
|
||||
"duration": video.get("duration"),
|
||||
"coverUrl": video.get("cover"),
|
||||
"format": video.get("format"),
|
||||
"definition": video.get("definition"),
|
||||
}
|
||||
|
||||
|
||||
def _hashtags(challenges: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
||||
return [
|
||||
{"id": c.get("id"), "name": c.get("title")}
|
||||
for c in challenges
|
||||
if isinstance(c, dict)
|
||||
]
|
||||
|
||||
|
||||
def parse_video(item: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Map an ``itemStruct`` to the flat output contract, filling known fields."""
|
||||
author = item.get("author") or {}
|
||||
author_stats = item.get("authorStats") or {}
|
||||
stats = item.get("stats") or {}
|
||||
username = author.get("uniqueId")
|
||||
video_id = item.get("id")
|
||||
|
||||
web_url = (
|
||||
_VIDEO_URL.format(username=username, video_id=video_id)
|
||||
if username and video_id
|
||||
else None
|
||||
)
|
||||
create_time = item.get("createTime")
|
||||
|
||||
return TikTokVideoItem(
|
||||
id=video_id,
|
||||
text=item.get("desc"),
|
||||
createTime=create_time,
|
||||
createTimeISO=epoch_to_iso(create_time),
|
||||
authorMeta=build_author_meta(author, author_stats),
|
||||
musicMeta=_music_meta(item.get("music") or {}),
|
||||
videoMeta=_video_meta(item.get("video") or {}),
|
||||
webVideoUrl=web_url,
|
||||
diggCount=stats.get("diggCount"),
|
||||
shareCount=stats.get("shareCount"),
|
||||
playCount=stats.get("playCount"),
|
||||
collectCount=stats.get("collectCount"),
|
||||
commentCount=stats.get("commentCount"),
|
||||
hashtags=_hashtags(item.get("challenges") or []),
|
||||
).to_output()
|
||||
|
|
@ -0,0 +1,19 @@
|
|||
"""Per-target scrape flows: resolved target -> normalized items."""
|
||||
|
||||
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."""
|
||||
|
||||
FetchUsersFn = Callable[[str, int], Awaitable[list[dict]]]
|
||||
"""Load a user-search page and return up to ``count`` captured ``user_info`` records."""
|
||||
|
||||
FetchCommentsFn = Callable[[str, int], Awaitable[list[dict]]]
|
||||
"""Load a video page and return up to ``count`` captured raw comment records."""
|
||||
|
||||
FlowResult = AsyncIterator[dict]
|
||||
|
|
@ -0,0 +1,52 @@
|
|||
"""Comments flow: a video URL -> its public comment thread.
|
||||
|
||||
Comments load over a signed ``/api/comment/list`` XHR that TikTok *does* serve to
|
||||
anonymous sessions once the comments panel is opened (unlike profile-video/search
|
||||
feeds). Records are deduped by comment id, capped, and — when a video has none or
|
||||
withholds them — degraded to one ErrorItem, mirroring the listing flow.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import AsyncIterator
|
||||
from typing import Any
|
||||
|
||||
from ..extraction import parse_comment
|
||||
from ..extraction.timestamps import now_iso
|
||||
from ..schemas import ErrorItem
|
||||
from ..targets.types import TikTokTarget
|
||||
from . import FetchCommentsFn
|
||||
|
||||
_EMPTY_MESSAGE = (
|
||||
"No comments returned. The video may have none, comments disabled, or TikTok "
|
||||
"withheld them from anonymous access."
|
||||
)
|
||||
|
||||
|
||||
async def iter_comments(
|
||||
target: TikTokTarget, *, cap: int, fetch_comments: FetchCommentsFn
|
||||
) -> AsyncIterator[dict[str, Any]]:
|
||||
if cap <= 0:
|
||||
return
|
||||
seen: set[str] = set()
|
||||
emitted = 0
|
||||
for raw in await fetch_comments(target.url, cap):
|
||||
out = parse_comment(raw, target.url)
|
||||
cid = out.get("id")
|
||||
if cid is not None:
|
||||
if cid in seen:
|
||||
continue
|
||||
seen.add(cid)
|
||||
out["scrapedAt"] = now_iso()
|
||||
yield out
|
||||
emitted += 1
|
||||
if emitted >= cap:
|
||||
return
|
||||
if emitted == 0:
|
||||
yield ErrorItem(
|
||||
url=target.url,
|
||||
input=target.value,
|
||||
error=_EMPTY_MESSAGE,
|
||||
errorCode="no_comments",
|
||||
scrapedAt=now_iso(),
|
||||
).to_output()
|
||||
|
|
@ -0,0 +1,55 @@
|
|||
"""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 ..schemas import ErrorItem
|
||||
from ..targets.types import TikTokTarget
|
||||
from . import FetchListingFn
|
||||
|
||||
# Profile and search feeds are trust-gated: an anonymous headless session gets an
|
||||
# empty body (profile) or no results XHR (search), while hashtag feeds load. We
|
||||
# can't tell "genuinely empty" from "blocked" here, so a zero-item listing emits
|
||||
# one honest ErrorItem instead of vanishing silently.
|
||||
_EMPTY_LISTING_MESSAGE = (
|
||||
"No videos returned. The target may be empty/private/removed, or TikTok "
|
||||
"withheld this feed from anonymous access (common for profiles and search)."
|
||||
)
|
||||
|
||||
|
||||
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
|
||||
if emitted == 0:
|
||||
yield ErrorItem(
|
||||
url=target.url,
|
||||
input=target.value,
|
||||
error=_EMPTY_LISTING_MESSAGE,
|
||||
errorCode="no_items",
|
||||
scrapedAt=now_iso(),
|
||||
).to_output()
|
||||
|
|
@ -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
|
||||
|
|
@ -0,0 +1,54 @@
|
|||
"""User-search flow: keyword -> public account records.
|
||||
|
||||
Unlike video/general search (login-walled for anonymous sessions), the Users tab
|
||||
hits ``/api/search/user`` and returns account records without a redirect. Each
|
||||
query's results are deduped by uid, capped, and — when a query returns nothing —
|
||||
degraded to one ErrorItem, mirroring the listing flow.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import AsyncIterator
|
||||
from typing import Any
|
||||
from urllib.parse import quote
|
||||
|
||||
from ..extraction import parse_search_user
|
||||
from ..extraction.timestamps import now_iso
|
||||
from ..schemas import ErrorItem
|
||||
from . import FetchUsersFn
|
||||
|
||||
_USER_SEARCH_URL = "https://www.tiktok.com/search/user?q={query}"
|
||||
_EMPTY_MESSAGE = (
|
||||
"No accounts returned for this query. It may have no matches, or TikTok "
|
||||
"withheld the results from anonymous access."
|
||||
)
|
||||
|
||||
|
||||
async def iter_user_search(
|
||||
query: str, *, cap: int, fetch_users: FetchUsersFn
|
||||
) -> AsyncIterator[dict[str, Any]]:
|
||||
if cap <= 0:
|
||||
return
|
||||
url = _USER_SEARCH_URL.format(query=quote(query))
|
||||
seen: set[str] = set()
|
||||
emitted = 0
|
||||
for user_info in await fetch_users(url, cap):
|
||||
out = parse_search_user(user_info)
|
||||
uid = out.get("id")
|
||||
if uid is not None:
|
||||
if uid in seen:
|
||||
continue
|
||||
seen.add(uid)
|
||||
out["scrapedAt"] = now_iso()
|
||||
yield out
|
||||
emitted += 1
|
||||
if emitted >= cap:
|
||||
return
|
||||
if emitted == 0:
|
||||
yield ErrorItem(
|
||||
url=url,
|
||||
input=query,
|
||||
error=_EMPTY_MESSAGE,
|
||||
errorCode="no_users",
|
||||
scrapedAt=now_iso(),
|
||||
).to_output()
|
||||
|
|
@ -0,0 +1,26 @@
|
|||
"""Video-URL flow: fetch a post page, read its rehydration blob, emit one item."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import AsyncIterator
|
||||
from typing import Any
|
||||
|
||||
from ..extraction import extract_rehydration_data, parse_video, video_item_struct
|
||||
from ..extraction.timestamps import now_iso
|
||||
from ..targets.types import TikTokTarget
|
||||
from . import FetchFn
|
||||
|
||||
|
||||
async def iter_video(target: TikTokTarget, *, fetch: FetchFn) -> AsyncIterator[dict[str, Any]]:
|
||||
html = await fetch(target.url)
|
||||
if not html:
|
||||
return
|
||||
data = extract_rehydration_data(html)
|
||||
if not data:
|
||||
return
|
||||
item = video_item_struct(data)
|
||||
if item is None:
|
||||
return
|
||||
out = parse_video(item)
|
||||
out["scrapedAt"] = now_iso()
|
||||
yield out
|
||||
|
|
@ -0,0 +1,195 @@
|
|||
"""Resolve a :class:`TikTokScrapeInput` into targets and stream their items.
|
||||
|
||||
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`: video URLs read the rehydration blob over HTTP,
|
||||
listings capture signed item_list XHRs through the stealth browser.
|
||||
"""
|
||||
|
||||
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
|
||||
from .flows.comments import iter_comments
|
||||
from .flows.listing import iter_listing
|
||||
from .flows.profile import iter_profile
|
||||
from .flows.user_search import iter_user_search
|
||||
from .flows.video import iter_video
|
||||
from .schemas import ErrorItem, TikTokScrapeInput
|
||||
from .session import (
|
||||
fetch_comments,
|
||||
fetch_html,
|
||||
fetch_item_list,
|
||||
fetch_trending,
|
||||
fetch_user_search,
|
||||
)
|
||||
from .targets import resolve_target
|
||||
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 every input source, dropping unresolved URLs."""
|
||||
targets: list[TikTokTarget] = []
|
||||
for entry in input_model.startUrls:
|
||||
resolved = resolve_target(entry.url)
|
||||
if resolved is not None:
|
||||
targets.append(resolved)
|
||||
for url in input_model.postURLs:
|
||||
resolved = resolve_target(url)
|
||||
if resolved is not None:
|
||||
targets.append(resolved)
|
||||
for profile in input_model.profiles:
|
||||
name = profile.lstrip("@")
|
||||
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:
|
||||
targets.append(
|
||||
TikTokTarget("search", query, _SEARCH_URL.format(query=quote(query)))
|
||||
)
|
||||
return targets
|
||||
|
||||
|
||||
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)
|
||||
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)
|
||||
|
||||
|
||||
async def iter_tiktok(
|
||||
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.
|
||||
|
||||
The video flow's ``fetch_html`` opens its own warmed proxy session per call
|
||||
when none is bound; the listing flow drives its own browser. Neither binds a
|
||||
ContextVar across these ``yield``s, so the generator stays context-safe.
|
||||
"""
|
||||
cap = input_model.resultsPerPage
|
||||
for target in _resolve_targets(input_model):
|
||||
async for item in _dispatch(
|
||||
target, cap=cap, fetch=fetch, fetch_listing=fetch_listing
|
||||
):
|
||||
yield item
|
||||
|
||||
|
||||
async def scrape_tiktok(
|
||||
input_model: TikTokScrapeInput,
|
||||
*,
|
||||
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, 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:
|
||||
break
|
||||
return results
|
||||
|
||||
|
||||
async def search_tiktok_users(
|
||||
queries: list[str],
|
||||
*,
|
||||
per_query: int,
|
||||
limit: int | None = None,
|
||||
fetch_users: FetchUsersFn = fetch_user_search,
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Collect user-search account records across queries, honoring ``limit``."""
|
||||
from app.capabilities.core.progress import emit_progress
|
||||
|
||||
results: list[dict[str, Any]] = []
|
||||
for query in queries:
|
||||
async for item in iter_user_search(
|
||||
query, cap=per_query, fetch_users=fetch_users
|
||||
):
|
||||
results.append(item)
|
||||
emit_progress("searching", current=len(results), total=limit, unit="item")
|
||||
if limit is not None and len(results) >= limit:
|
||||
return results
|
||||
return results
|
||||
|
||||
|
||||
async def scrape_tiktok_trending(
|
||||
*,
|
||||
count: int,
|
||||
fetch_trending_fn: FetchListingFn = fetch_trending,
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Collect up to ``count`` trending videos from the Explore feed.
|
||||
|
||||
A single global feed, so it reuses the listing flow (parse/dedupe/cap/empty-
|
||||
ErrorItem) over a synthetic target — no user input to resolve.
|
||||
"""
|
||||
from app.capabilities.core.progress import emit_progress
|
||||
|
||||
target = TikTokTarget(kind="trending", value="explore", url=_EXPLORE_URL)
|
||||
results: list[dict[str, Any]] = []
|
||||
async for item in iter_listing(target, cap=count, fetch_listing=fetch_trending_fn):
|
||||
results.append(item)
|
||||
emit_progress("scraping", current=len(results), total=count, unit="item")
|
||||
return results
|
||||
|
||||
|
||||
async def scrape_tiktok_comments(
|
||||
video_urls: list[str],
|
||||
*,
|
||||
per_video: int,
|
||||
limit: int | None = None,
|
||||
fetch_comments_fn: FetchCommentsFn = fetch_comments,
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Collect comments across video URLs, honoring ``limit``.
|
||||
|
||||
A non-video URL yields one ``bad_url`` ErrorItem (never a silent drop) so the
|
||||
caller can tell "wrong input" from "video has no comments".
|
||||
"""
|
||||
from app.capabilities.core.progress import emit_progress
|
||||
|
||||
results: list[dict[str, Any]] = []
|
||||
for url in video_urls:
|
||||
target = resolve_target(url)
|
||||
if target is None or target.kind != "video":
|
||||
results.append(
|
||||
ErrorItem(
|
||||
url=url,
|
||||
input=url,
|
||||
error="Not a TikTok video URL.",
|
||||
errorCode="bad_url",
|
||||
scrapedAt=now_iso(),
|
||||
).to_output()
|
||||
)
|
||||
emit_progress("scraping", current=len(results), total=limit, unit="item")
|
||||
if limit is not None and len(results) >= limit:
|
||||
return results
|
||||
continue
|
||||
async for item in iter_comments(
|
||||
target, cap=per_video, fetch_comments=fetch_comments_fn
|
||||
):
|
||||
results.append(item)
|
||||
emit_progress("scraping", current=len(results), total=limit, unit="item")
|
||||
if limit is not None and len(results) >= limit:
|
||||
return results
|
||||
return results
|
||||
|
|
@ -0,0 +1,26 @@
|
|||
"""Apify-compatible input and output contracts for the TikTok scraper."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from .input import StartUrl, TikTokScrapeInput
|
||||
from .items import (
|
||||
AuthorMeta,
|
||||
CommentItem,
|
||||
ErrorItem,
|
||||
MusicMeta,
|
||||
TikTokProfileItem,
|
||||
TikTokVideoItem,
|
||||
VideoMeta,
|
||||
)
|
||||
|
||||
__all__ = [
|
||||
"AuthorMeta",
|
||||
"CommentItem",
|
||||
"ErrorItem",
|
||||
"MusicMeta",
|
||||
"StartUrl",
|
||||
"TikTokProfileItem",
|
||||
"TikTokScrapeInput",
|
||||
"TikTokVideoItem",
|
||||
"VideoMeta",
|
||||
]
|
||||
|
|
@ -0,0 +1,59 @@
|
|||
# ruff: noqa: N815 - field names mirror the public camelCase TikTok/Apify API
|
||||
"""Input surface for the TikTok scraper, shaped to the Clockworks actor.
|
||||
|
||||
Anonymous only: no auth-shaped field exists here. ``extra="allow"`` keeps the
|
||||
contract in parity with the actor for fields the scraper does not read.
|
||||
|
||||
``resultsPerPage`` is a per-target count; the cross-target ceiling is caller
|
||||
policy applied by the collector, never baked into a flow.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Literal
|
||||
|
||||
from pydantic import BaseModel, ConfigDict, Field
|
||||
|
||||
ProfileSorting = Literal["latest", "popular", "oldest"]
|
||||
ProfileSection = Literal["videos", "reposts"]
|
||||
SearchSection = Literal["", "/video", "/user"]
|
||||
|
||||
|
||||
class StartUrl(BaseModel):
|
||||
"""A single direct URL entry (``{"url": ...}``; extra keys ignored)."""
|
||||
|
||||
model_config = ConfigDict(extra="allow")
|
||||
|
||||
url: str
|
||||
|
||||
|
||||
class TikTokScrapeInput(BaseModel):
|
||||
model_config = ConfigDict(extra="allow")
|
||||
|
||||
# Discovery
|
||||
startUrls: list[StartUrl] = Field(default_factory=list)
|
||||
hashtags: list[str] = Field(default_factory=list)
|
||||
profiles: list[str] = Field(default_factory=list)
|
||||
searchQueries: list[str] = Field(default_factory=list)
|
||||
postURLs: list[str] = Field(default_factory=list)
|
||||
|
||||
# Per-target count
|
||||
resultsPerPage: int = Field(default=1, ge=1)
|
||||
|
||||
# Profile options
|
||||
profileScrapeSections: list[ProfileSection] = Field(
|
||||
default_factory=lambda: ["videos"]
|
||||
)
|
||||
profileSorting: ProfileSorting = "latest"
|
||||
excludePinnedPosts: bool = False
|
||||
|
||||
# Search options
|
||||
searchSection: SearchSection = ""
|
||||
maxProfilesPerQuery: int = Field(default=10, ge=1)
|
||||
|
||||
# Incremental filters (ISO date or relative "<n> days" per the actor)
|
||||
oldestPostDateUnified: str | None = None
|
||||
newestPostDate: str | None = None
|
||||
|
||||
# Proxy
|
||||
proxyCountryCode: str = "None"
|
||||
|
|
@ -0,0 +1,147 @@
|
|||
# ruff: noqa: N815 - field names mirror the public camelCase TikTok/Apify API
|
||||
"""Output items keyed to the Clockworks TikTok actor's shape.
|
||||
|
||||
Every model is open (``extra="allow"``) and defaults unsourced fields to
|
||||
``None``/``[]`` so the MVP can populate a reliable subset and expand the
|
||||
contract additively without breaking consumers.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
from pydantic import BaseModel, ConfigDict, Field
|
||||
|
||||
|
||||
class AuthorMeta(BaseModel):
|
||||
model_config = ConfigDict(extra="allow")
|
||||
|
||||
id: str | None = None
|
||||
name: str | None = None
|
||||
nickName: str | None = None
|
||||
profileUrl: str | None = None
|
||||
verified: bool | None = None
|
||||
signature: str | None = None
|
||||
avatar: str | None = None
|
||||
privateAccount: bool | None = None
|
||||
fans: int | None = None
|
||||
following: int | None = None
|
||||
heart: int | None = None
|
||||
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")
|
||||
|
||||
musicId: str | None = None
|
||||
musicName: str | None = None
|
||||
musicAuthor: str | None = None
|
||||
musicOriginal: bool | None = None
|
||||
playUrl: str | None = None
|
||||
|
||||
|
||||
class VideoMeta(BaseModel):
|
||||
model_config = ConfigDict(extra="allow")
|
||||
|
||||
height: int | None = None
|
||||
width: int | None = None
|
||||
duration: int | None = None
|
||||
coverUrl: str | None = None
|
||||
format: str | None = None
|
||||
definition: str | None = None
|
||||
|
||||
|
||||
class TikTokVideoItem(BaseModel):
|
||||
"""A single scraped post (video or slideshow)."""
|
||||
|
||||
model_config = ConfigDict(extra="allow")
|
||||
|
||||
id: str | None = None
|
||||
text: str | None = None
|
||||
textLanguage: str | None = None
|
||||
createTime: int | None = None
|
||||
createTimeISO: str | None = None
|
||||
isAd: bool | None = None
|
||||
|
||||
authorMeta: AuthorMeta = Field(default_factory=AuthorMeta)
|
||||
musicMeta: MusicMeta = Field(default_factory=MusicMeta)
|
||||
videoMeta: VideoMeta = Field(default_factory=VideoMeta)
|
||||
|
||||
webVideoUrl: str | None = None
|
||||
mediaUrls: list[str] = Field(default_factory=list)
|
||||
|
||||
diggCount: int | None = None
|
||||
shareCount: int | None = None
|
||||
playCount: int | None = None
|
||||
collectCount: int | None = None
|
||||
commentCount: int | None = None
|
||||
|
||||
hashtags: list[dict[str, Any]] = Field(default_factory=list)
|
||||
mentions: list[str] = Field(default_factory=list)
|
||||
|
||||
isSlideshow: bool | None = None
|
||||
isPinned: bool | None = None
|
||||
isSponsored: bool | None = None
|
||||
|
||||
scrapedAt: str | None = None
|
||||
|
||||
def to_output(self) -> dict[str, Any]:
|
||||
return self.model_dump(exclude_none=False)
|
||||
|
||||
|
||||
class CommentItem(BaseModel):
|
||||
"""A single comment or reply under a post."""
|
||||
|
||||
model_config = ConfigDict(extra="allow")
|
||||
|
||||
id: str | None = None
|
||||
text: str | None = None
|
||||
videoWebUrl: str | None = None
|
||||
diggCount: int | None = None
|
||||
replyCommentTotal: int | None = None
|
||||
createTime: int | None = None
|
||||
createTimeISO: str | None = None
|
||||
|
||||
uid: str | None = None
|
||||
uniqueId: str | None = None
|
||||
nickName: str | None = None
|
||||
avatar: str | None = None
|
||||
|
||||
repliesToId: str | None = None
|
||||
scrapedAt: str | None = None
|
||||
|
||||
def to_output(self) -> dict[str, Any]:
|
||||
return self.model_dump(exclude_none=False)
|
||||
|
||||
|
||||
class ErrorItem(BaseModel):
|
||||
"""Per-input failure, distinguished from normal items by ``errorCode``.
|
||||
|
||||
Mirrors the actor's convention so a private/deleted/empty target surfaces
|
||||
as an item instead of silently vanishing from the results.
|
||||
"""
|
||||
|
||||
model_config = ConfigDict(extra="allow")
|
||||
|
||||
url: str | None = None
|
||||
input: str | None = None
|
||||
error: str | None = None
|
||||
errorCode: str | None = None
|
||||
|
||||
def to_output(self) -> dict[str, Any]:
|
||||
return self.model_dump(exclude_none=False)
|
||||
|
|
@ -0,0 +1,25 @@
|
|||
"""Cookie-warmed, rotate-on-block proxy session and page-fetch seam."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from .client import fetch_html
|
||||
from .errors import TikTokAccessBlockedError
|
||||
from .listing import (
|
||||
fetch_comments,
|
||||
fetch_item_list,
|
||||
fetch_trending,
|
||||
fetch_user_search,
|
||||
)
|
||||
from .proxy import bind_proxy_holder, open_proxy_holder, proxy_session
|
||||
|
||||
__all__ = [
|
||||
"TikTokAccessBlockedError",
|
||||
"bind_proxy_holder",
|
||||
"fetch_comments",
|
||||
"fetch_html",
|
||||
"fetch_item_list",
|
||||
"fetch_trending",
|
||||
"fetch_user_search",
|
||||
"open_proxy_holder",
|
||||
"proxy_session",
|
||||
]
|
||||
|
|
@ -0,0 +1,130 @@
|
|||
"""GET TikTok page HTML through a cookie-warmed, sticky-IP proxy session.
|
||||
|
||||
The warm-up mints TikTok's anonymous device cookie (``ttwid``) on the first
|
||||
homepage hit; the target page then server-renders its rehydration blob. Rotates
|
||||
the residential IP and re-warms on 403, backs off on 429, and raises
|
||||
:class:`TikTokAccessBlockedError` only when every rotated IP refuses access.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
import random
|
||||
from contextlib import suppress
|
||||
from typing import Any
|
||||
|
||||
from scrapling.fetchers import AsyncFetcher
|
||||
|
||||
from app.utils.proxy import get_proxy_url
|
||||
|
||||
from .errors import TikTokAccessBlockedError
|
||||
from .proxy import _REQUEST_TIMEOUT_S, _current_session, proxy_session
|
||||
|
||||
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
|
||||
_MAX_BACKOFFS = 4
|
||||
_BACKOFF_BASE_S = 5.0
|
||||
|
||||
_HOME_URL = "https://www.tiktok.com/"
|
||||
_TTWID_COOKIE = "ttwid"
|
||||
_HEADERS = {"Accept-Language": "en-US,en;q=0.9"}
|
||||
|
||||
|
||||
def _response_cookie_names(page: Any) -> set[str]:
|
||||
cookies = getattr(page, "cookies", None)
|
||||
return set(cookies.keys()) if isinstance(cookies, dict) else set()
|
||||
|
||||
|
||||
def _page_html(page: Any) -> str | None:
|
||||
for attr in ("text", "body"):
|
||||
val = getattr(page, attr, None)
|
||||
if isinstance(val, bytes):
|
||||
val = val.decode("utf-8", "replace")
|
||||
if isinstance(val, str) and val.strip():
|
||||
return val
|
||||
return None
|
||||
|
||||
|
||||
async def warm_session(session: Any) -> bool:
|
||||
"""Mint an anonymous ``ttwid`` cookie; ``True`` if the session can now fetch."""
|
||||
with suppress(Exception):
|
||||
page = await session.get(_HOME_URL, headers=_HEADERS)
|
||||
if _TTWID_COOKIE in _response_cookie_names(page):
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
async def _get_page(session: Any, url: str) -> Any:
|
||||
if session is not None:
|
||||
return await session.get(url, headers=_HEADERS)
|
||||
return await AsyncFetcher.get(
|
||||
url,
|
||||
headers=_HEADERS,
|
||||
proxy=get_proxy_url(),
|
||||
stealthy_headers=True,
|
||||
timeout=_REQUEST_TIMEOUT_S,
|
||||
)
|
||||
|
||||
|
||||
async def fetch_html(url: str) -> str | None:
|
||||
"""Return page HTML, or ``None`` on 404 / non-block failure."""
|
||||
holder = _current_session.get()
|
||||
if holder is None:
|
||||
async with proxy_session():
|
||||
return await fetch_html(url)
|
||||
|
||||
attempt = 0
|
||||
backoffs = 0
|
||||
while True:
|
||||
session = holder.session
|
||||
try:
|
||||
if session is not None and not holder.warmed:
|
||||
warmed_ok = await warm_session(session)
|
||||
holder.warmed = True
|
||||
if not warmed_ok:
|
||||
if attempt < _MAX_ROTATIONS:
|
||||
attempt += 1
|
||||
await holder.rotate()
|
||||
continue
|
||||
raise TikTokAccessBlockedError(
|
||||
f"could not warm session after {attempt} IP rotations: {url}"
|
||||
)
|
||||
|
||||
await holder.pace()
|
||||
page = await _get_page(session, url)
|
||||
status = page.status
|
||||
|
||||
if status == 200:
|
||||
return _page_html(page)
|
||||
if status == 404:
|
||||
return None
|
||||
if status == _BACKOFF_STATUS and backoffs < _MAX_BACKOFFS:
|
||||
backoffs += 1
|
||||
delay = _BACKOFF_BASE_S * (2 ** (backoffs - 1))
|
||||
logger.warning("[tiktok] 429 on %s; backing off %.1fs", url, delay)
|
||||
await asyncio.sleep(delay + random.uniform(0, 1))
|
||||
continue
|
||||
if status == _ROTATE_STATUS and attempt < _MAX_ROTATIONS:
|
||||
attempt += 1
|
||||
await holder.rotate()
|
||||
continue
|
||||
if status == _ROTATE_STATUS:
|
||||
raise TikTokAccessBlockedError(
|
||||
f"TikTok refused {url} on {attempt} rotated IPs (403)"
|
||||
)
|
||||
logger.warning("[tiktok] GET %s returned %s", url, status)
|
||||
return None
|
||||
except TikTokAccessBlockedError:
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.warning("[tiktok] GET %s failed: %s", url, e)
|
||||
if attempt < _MAX_ROTATIONS:
|
||||
attempt += 1
|
||||
await holder.rotate()
|
||||
continue
|
||||
return None
|
||||
|
|
@ -0,0 +1,10 @@
|
|||
"""Fetch-seam errors surfaced to the capability layer."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
|
||||
class TikTokAccessBlockedError(RuntimeError):
|
||||
"""Raised when every rotated IP is refused anonymous access.
|
||||
|
||||
Distinguishes a hard block from an empty result; the route maps it to 403.
|
||||
"""
|
||||
|
|
@ -0,0 +1,314 @@
|
|||
"""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).
|
||||
|
||||
Needs a residential proxy; datacenter IPs get empty bodies and 429s. The profile
|
||||
feed returns an empty 200 to headless sessions, so :func:`fetch_item_list` goes
|
||||
headful only when ``CRAWL_HEADED_XVFB_ENABLED`` promises an Xvfb display — else it
|
||||
stays headless and degrades to an ``ErrorItem`` instead of crashing on launch.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
from collections.abc import Callable
|
||||
from typing import Any
|
||||
|
||||
from scrapling.fetchers import StealthyFetcher
|
||||
|
||||
from app.config import config
|
||||
from app.proprietary.web_crawler.stealth import (
|
||||
build_stealthy_kwargs,
|
||||
get_stealth_config,
|
||||
)
|
||||
from app.utils.proxy import get_proxy_url
|
||||
|
||||
from ..extraction import (
|
||||
comments_from_response,
|
||||
items_from_response,
|
||||
users_from_response,
|
||||
)
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
ExtractFn = Callable[[Any], list[dict[str, Any]]]
|
||||
# Drives the page after navigation to trigger/paginate the target XHRs, filling
|
||||
# ``collected`` until it reaches ``target_count`` (or the interaction gives up).
|
||||
InteractFn = Callable[[Any, list[dict[str, Any]], int], None]
|
||||
|
||||
# XHR paths that carry itemStructs for the three listing kinds.
|
||||
_ITEM_LIST_MARKERS = (
|
||||
"/api/post/item_list",
|
||||
"/api/challenge/item_list",
|
||||
"/api/search/",
|
||||
)
|
||||
# The user-search XHR carries account records (user_list), not itemStructs.
|
||||
_USER_SEARCH_MARKERS = ("/api/search/user",)
|
||||
# The Explore feed's trending videos arrive as ordinary itemStructs.
|
||||
_EXPLORE_MARKERS = ("/api/explore/item_list",)
|
||||
# The comment feed fires only after the comments panel is opened.
|
||||
_COMMENT_MARKERS = ("/api/comment/list",)
|
||||
_COMMENT_ICON_SELECTORS = (
|
||||
'[data-e2e="comment-icon"]',
|
||||
'[data-e2e="browse-comment"]',
|
||||
)
|
||||
# The comment icon hydrates a beat after DOM-ready; wait for it before clicking.
|
||||
_COMMENT_ICON_WAIT_MS = 8000
|
||||
# First comment page lands shortly after the click — don't declare "empty" early.
|
||||
_COMMENT_FIRST_PAGE_MS = 3500
|
||||
_HOME_URL = "https://www.tiktok.com/"
|
||||
_MSTOKEN_COOKIE = "msToken"
|
||||
# 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 = 1500
|
||||
# Warm-up poll for the anonymous msToken cookie the item_list API requires.
|
||||
_WARM_POLLS = 8
|
||||
_WARM_POLL_MS = 500
|
||||
|
||||
|
||||
def _has_mstoken(page: Any) -> bool:
|
||||
try:
|
||||
return any(c.get("name") == _MSTOKEN_COOKIE for c in page.context.cookies())
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
|
||||
def _dismiss_login_modal(page: Any) -> None:
|
||||
"""Close the login modal that blocks scrolling; Escape as fallback.
|
||||
|
||||
Only the modal's own close button, never a generic dialog button (avoids
|
||||
clicking "Log in").
|
||||
"""
|
||||
try:
|
||||
closed = page.evaluate(
|
||||
"""() => {
|
||||
const btn = document.querySelector('[data-e2e="modal-close-inner-button"]');
|
||||
if (btn) { btn.click(); return true; }
|
||||
return false;
|
||||
}"""
|
||||
)
|
||||
if not closed:
|
||||
page.keyboard.press("Escape")
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
def _scroll_page(page: Any, collected: list[dict[str, Any]], target_count: int) -> None:
|
||||
"""Page down a listing feed until enough items are captured or it stops growing."""
|
||||
last_height = 0
|
||||
for _ in range(_SCROLL_MAX_ROUNDS):
|
||||
if len(collected) >= target_count:
|
||||
break
|
||||
_dismiss_login_modal(page)
|
||||
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
|
||||
|
||||
|
||||
def _open_comments(page: Any) -> None:
|
||||
"""Click the comment icon so the first ``/api/comment/list`` XHR fires.
|
||||
|
||||
The icon must be present and interactive first (the SPA hydrates it a beat
|
||||
after DOM-ready), so we wait for it, then fall back to a JS click if the
|
||||
normal click is intercepted (cookie banner / overlay).
|
||||
"""
|
||||
for selector in _COMMENT_ICON_SELECTORS:
|
||||
try:
|
||||
page.wait_for_selector(selector, timeout=_COMMENT_ICON_WAIT_MS)
|
||||
except Exception:
|
||||
continue
|
||||
try:
|
||||
page.click(selector, timeout=_COMMENT_ICON_WAIT_MS)
|
||||
return
|
||||
except Exception:
|
||||
try:
|
||||
page.eval_on_selector(selector, "el => el.click()")
|
||||
return
|
||||
except Exception:
|
||||
continue
|
||||
|
||||
|
||||
def _scroll_comments(
|
||||
page: Any, collected: list[dict[str, Any]], target_count: int
|
||||
) -> None:
|
||||
"""Open the comments panel, then scroll its last comment into view to paginate.
|
||||
|
||||
Comment XHRs fire only after the panel is opened, and paging must scroll the
|
||||
panel (not the page, which would advance the video feed), so we anchor on the
|
||||
last ``comment-level-1`` element. ponytail: naive scroll-to-last paging,
|
||||
bounded by ``_SCROLL_MAX_ROUNDS``; upgrade to container-height tracking if
|
||||
deep threads under-fetch.
|
||||
"""
|
||||
_open_comments(page)
|
||||
# The panel's first page lands a beat after the click; give it room before
|
||||
# we decide there are no comments to page through.
|
||||
page.wait_for_timeout(_COMMENT_FIRST_PAGE_MS)
|
||||
for _ in range(_SCROLL_MAX_ROUNDS):
|
||||
if len(collected) >= target_count:
|
||||
break
|
||||
moved = page.evaluate(
|
||||
"""() => {
|
||||
const items = document.querySelectorAll('[data-e2e="comment-level-1"]');
|
||||
if (!items.length) return false;
|
||||
items[items.length - 1].scrollIntoView({block: 'end'});
|
||||
return true;
|
||||
}"""
|
||||
)
|
||||
page.wait_for_timeout(_SCROLL_SETTLE_MS)
|
||||
if not moved:
|
||||
break
|
||||
|
||||
|
||||
def _build_page_action(
|
||||
collected: list[dict[str, Any]],
|
||||
url: str,
|
||||
target_count: int,
|
||||
markers: tuple[str, ...],
|
||||
extract: ExtractFn,
|
||||
interact: InteractFn,
|
||||
):
|
||||
"""A sync ``page_action`` that warms the session then captures matching XHRs.
|
||||
|
||||
A cold context returns an empty body, so we first mint the anonymous
|
||||
``msToken`` (homepage hit), then navigate to the target with the listener
|
||||
already attached so page-one fires into it; ``interact`` pages the rest.
|
||||
``markers``/``extract`` select which XHRs to keep and how to unwrap them.
|
||||
"""
|
||||
|
||||
def _on_response(response: Any) -> None:
|
||||
response_url = getattr(response, "url", "")
|
||||
if not any(marker in response_url for marker in markers):
|
||||
return
|
||||
try:
|
||||
body = response.json()
|
||||
except Exception:
|
||||
# An empty 200 (TikTok soft-block) or a body evicted before read.
|
||||
return
|
||||
collected.extend(extract(body))
|
||||
|
||||
def _warm(page: Any) -> None:
|
||||
if _has_mstoken(page):
|
||||
return
|
||||
page.goto(_HOME_URL, wait_until="domcontentloaded")
|
||||
for _ in range(_WARM_POLLS):
|
||||
page.wait_for_timeout(_WARM_POLL_MS)
|
||||
if _has_mstoken(page):
|
||||
break
|
||||
|
||||
def page_action(page: Any) -> Any:
|
||||
page.on("response", _on_response)
|
||||
try:
|
||||
_warm(page)
|
||||
# Navigate (back) to the target with the listener attached and a
|
||||
# token in hand, so the page-one XHR fires into the capture.
|
||||
page.goto(url, wait_until="domcontentloaded")
|
||||
page.wait_for_timeout(_SCROLL_SETTLE_MS)
|
||||
interact(page, collected, target_count)
|
||||
except Exception as exc:
|
||||
logger.debug("[tiktok] capture interaction aborted: %s", exc)
|
||||
return page
|
||||
|
||||
return page_action
|
||||
|
||||
|
||||
def _fetch_sync(
|
||||
url: str,
|
||||
target_count: int,
|
||||
markers: tuple[str, ...],
|
||||
extract: ExtractFn,
|
||||
interact: InteractFn,
|
||||
*,
|
||||
headless: bool = True,
|
||||
) -> list[dict[str, Any]]:
|
||||
collected: list[dict[str, Any]] = []
|
||||
kwargs = build_stealthy_kwargs(get_stealth_config())
|
||||
StealthyFetcher.fetch(
|
||||
url,
|
||||
headless=headless,
|
||||
network_idle=False,
|
||||
proxy=get_proxy_url(),
|
||||
page_action=_build_page_action(
|
||||
collected, url, target_count, markers, extract, interact
|
||||
),
|
||||
**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.
|
||||
|
||||
Headful when ``CRAWL_HEADED_XVFB_ENABLED`` promises a display (the profile feed
|
||||
is empty to headless sessions); headless otherwise so launch never fails.
|
||||
Retries an empty draw up to ``TIKTOK_LISTING_MAX_ATTEMPTS`` for a fresh exit IP.
|
||||
"""
|
||||
headless = not config.CRAWL_HEADED_XVFB_ENABLED
|
||||
attempts = max(1, config.TIKTOK_LISTING_MAX_ATTEMPTS)
|
||||
for attempt in range(1, attempts + 1):
|
||||
items = await asyncio.to_thread(
|
||||
_fetch_sync,
|
||||
page_url,
|
||||
target_count,
|
||||
_ITEM_LIST_MARKERS,
|
||||
items_from_response,
|
||||
_scroll_page,
|
||||
headless=headless,
|
||||
)
|
||||
if items or attempt == attempts:
|
||||
return items
|
||||
logger.info(
|
||||
"[tiktok] empty item_list for %s (attempt %d/%d); retrying on a fresh exit IP",
|
||||
page_url,
|
||||
attempt,
|
||||
attempts,
|
||||
)
|
||||
return []
|
||||
|
||||
|
||||
async def fetch_user_search(page_url: str, target_count: int) -> list[dict[str, Any]]:
|
||||
"""Return up to ``target_count`` ``user_info`` records from a user-search page."""
|
||||
return await asyncio.to_thread(
|
||||
_fetch_sync,
|
||||
page_url,
|
||||
target_count,
|
||||
_USER_SEARCH_MARKERS,
|
||||
users_from_response,
|
||||
_scroll_page,
|
||||
)
|
||||
|
||||
|
||||
async def fetch_comments(page_url: str, target_count: int) -> list[dict[str, Any]]:
|
||||
"""Return up to ``target_count`` raw comment records from a video page's XHRs."""
|
||||
return await asyncio.to_thread(
|
||||
_fetch_sync,
|
||||
page_url,
|
||||
target_count,
|
||||
_COMMENT_MARKERS,
|
||||
comments_from_response,
|
||||
_scroll_comments,
|
||||
)
|
||||
|
||||
|
||||
async def fetch_trending(page_url: str, target_count: int) -> list[dict[str, Any]]:
|
||||
"""Return up to ``target_count`` trending itemStructs from the Explore feed."""
|
||||
return await asyncio.to_thread(
|
||||
_fetch_sync,
|
||||
page_url,
|
||||
target_count,
|
||||
_EXPLORE_MARKERS,
|
||||
items_from_response,
|
||||
_scroll_page,
|
||||
)
|
||||
|
|
@ -0,0 +1,113 @@
|
|||
"""Rotate-on-block sticky proxy session, bound per-flow via a ContextVar.
|
||||
|
||||
Reusing one keep-alive connection pins a single residential exit IP so the
|
||||
warmed cookie jar (``ttwid``/``msToken``, bound to that IP) stays valid across
|
||||
the warm-up and every subsequent fetch. Ported from the Reddit sibling; the
|
||||
TikTok-specific warm-up lives in :mod:`client`.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
import random
|
||||
import time
|
||||
from contextlib import asynccontextmanager, suppress
|
||||
from contextvars import ContextVar
|
||||
from typing import Any
|
||||
|
||||
from scrapling.fetchers import FetcherSession
|
||||
|
||||
from app.utils.proxy import get_proxy_url
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Pace each sticky IP so a fast exit can't burst past TikTok's per-IP threshold.
|
||||
_MIN_INTERVAL_S = 0.5
|
||||
_PACE_JITTER_S = 0.25
|
||||
# A healthy fetch lands in ~1-2s; cap a dead IP at one bounded wait before it
|
||||
# falls through to a rotation.
|
||||
_REQUEST_TIMEOUT_S = 15.0
|
||||
|
||||
_current_session: ContextVar[_RotatingSession | None] = ContextVar(
|
||||
"tiktok_proxy_session", default=None
|
||||
)
|
||||
|
||||
|
||||
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.
|
||||
"""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self._cm: Any | None = None
|
||||
self.session: Any | None = None
|
||||
self.rotations = 0
|
||||
self.warmed = False
|
||||
self._last_at = 0.0
|
||||
|
||||
async def _open(self) -> None:
|
||||
proxy = get_proxy_url()
|
||||
self.warmed = False
|
||||
if proxy is None:
|
||||
self._cm = self.session = None
|
||||
return
|
||||
self._cm = FetcherSession(
|
||||
proxy=proxy,
|
||||
stealthy_headers=True,
|
||||
impersonate="chrome",
|
||||
timeout=_REQUEST_TIMEOUT_S,
|
||||
)
|
||||
self.session = await self._cm.__aenter__()
|
||||
|
||||
async def close(self) -> None:
|
||||
if self._cm is not None:
|
||||
with suppress(Exception):
|
||||
await self._cm.__aexit__(None, None, None)
|
||||
self._cm = self.session = None
|
||||
|
||||
async def rotate(self) -> Any | None:
|
||||
"""Drop the current IP and connect through a fresh one."""
|
||||
await self.close()
|
||||
self.rotations += 1
|
||||
await self._open()
|
||||
logger.info("[tiktok] rotated proxy session (rotation #%d)", self.rotations)
|
||||
return self.session
|
||||
|
||||
async def pace(self) -> None:
|
||||
"""Sleep to hold this sticky IP under TikTok's per-IP rate threshold."""
|
||||
wait = _MIN_INTERVAL_S - (time.monotonic() - self._last_at)
|
||||
if wait > 0:
|
||||
await asyncio.sleep(wait + random.uniform(0, _PACE_JITTER_S))
|
||||
self._last_at = time.monotonic()
|
||||
|
||||
|
||||
async def open_proxy_holder() -> _RotatingSession:
|
||||
"""Open a warm rotate-on-block session holder (caller owns ``close()``)."""
|
||||
holder = _RotatingSession()
|
||||
await holder._open()
|
||||
return holder
|
||||
|
||||
|
||||
@asynccontextmanager
|
||||
async def bind_proxy_holder(holder: _RotatingSession):
|
||||
"""Route this task's fetches through ``holder`` for the enclosed block."""
|
||||
token = _current_session.set(holder)
|
||||
try:
|
||||
yield holder
|
||||
finally:
|
||||
_current_session.reset(token)
|
||||
|
||||
|
||||
@asynccontextmanager
|
||||
async def proxy_session():
|
||||
"""Open one reused, rotate-on-block proxy session for a continuation chain."""
|
||||
holder = await open_proxy_holder()
|
||||
try:
|
||||
async with bind_proxy_holder(holder):
|
||||
yield holder
|
||||
finally:
|
||||
await holder.close()
|
||||
|
|
@ -0,0 +1,8 @@
|
|||
"""TikTok URL classification into scrape targets."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from .resolver import resolve_target
|
||||
from .types import TargetKind, TikTokTarget
|
||||
|
||||
__all__ = ["TargetKind", "TikTokTarget", "resolve_target"]
|
||||
|
|
@ -0,0 +1,50 @@
|
|||
"""Classify a TikTok URL into a :class:`TikTokTarget`, or ``None``."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from urllib.parse import parse_qs, unquote, urlparse
|
||||
|
||||
from .types import SearchSection, TikTokTarget
|
||||
|
||||
_TIKTOK_HOSTS = frozenset({"tiktok.com", "www.tiktok.com", "m.tiktok.com"})
|
||||
_SEARCH_SECTIONS: frozenset[SearchSection] = frozenset({"video", "user"})
|
||||
|
||||
|
||||
def _is_tiktok_host(hostname: str | None) -> bool:
|
||||
return bool(hostname) and hostname.lower() in _TIKTOK_HOSTS
|
||||
|
||||
|
||||
def resolve_target(url: str) -> TikTokTarget | None:
|
||||
parsed = urlparse(url)
|
||||
if not _is_tiktok_host(parsed.hostname):
|
||||
return None
|
||||
|
||||
segments = [s for s in (parsed.path or "").split("/") if s]
|
||||
if not segments:
|
||||
return None
|
||||
|
||||
# Profile / video live under /@username[...].
|
||||
if segments[0].startswith("@"):
|
||||
username = segments[0][1:]
|
||||
if not username:
|
||||
return None
|
||||
if len(segments) >= 3 and segments[1] == "video" and segments[2]:
|
||||
return TikTokTarget("video", segments[2], url, username=username)
|
||||
return TikTokTarget("profile", username, url)
|
||||
|
||||
if segments[0] == "tag" and len(segments) >= 2 and segments[1]:
|
||||
return TikTokTarget("hashtag", unquote(segments[1]), url)
|
||||
|
||||
if segments[0] == "search":
|
||||
query = parse_qs(parsed.query).get("q", [None])[0]
|
||||
if not query:
|
||||
return None
|
||||
section = segments[1] if len(segments) >= 2 else None
|
||||
return TikTokTarget(
|
||||
"search",
|
||||
unquote(query),
|
||||
url,
|
||||
section=section if section in _SEARCH_SECTIONS else None,
|
||||
)
|
||||
|
||||
return None
|
||||
|
|
@ -0,0 +1,25 @@
|
|||
"""Scrape-target value object produced by URL classification."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from typing import Literal
|
||||
|
||||
TargetKind = Literal["video", "profile", "hashtag", "search", "trending"]
|
||||
SearchSection = Literal["video", "user"]
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class TikTokTarget:
|
||||
"""One classified scrape target.
|
||||
|
||||
``value`` holds the kind-specific identifier: video id, username, hashtag
|
||||
name, or search query. ``username`` is set for videos (needed to build the
|
||||
canonical post URL). ``section`` narrows a search to videos or users.
|
||||
"""
|
||||
|
||||
kind: TargetKind
|
||||
value: str
|
||||
url: str
|
||||
username: str | None = None
|
||||
section: SearchSection | None = None
|
||||
|
|
@ -5,6 +5,7 @@ import app.capabilities.google_maps
|
|||
import app.capabilities.google_search
|
||||
import app.capabilities.instagram
|
||||
import app.capabilities.reddit
|
||||
import app.capabilities.tiktok
|
||||
import app.capabilities.web
|
||||
import app.capabilities.youtube # noqa: F401
|
||||
from app.automations.api import router as automations_router
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue