feat(tiktok): tiktok.scrape capability + billing wire-up

This commit is contained in:
CREDO23 2026-07-08 18:21:44 +02:00
parent 9dd39faa50
commit 2943d8b23c
14 changed files with 317 additions and 0 deletions

View file

@ -35,6 +35,7 @@ _PLATFORM_RATE_KEYS: dict[BillingUnit, str] = {
BillingUnit.GOOGLE_MAPS_REVIEW: "GOOGLE_MAPS_MICROS_PER_REVIEW",
BillingUnit.YOUTUBE_VIDEO: "YOUTUBE_MICROS_PER_VIDEO",
BillingUnit.YOUTUBE_COMMENT: "YOUTUBE_MICROS_PER_COMMENT",
BillingUnit.TIKTOK_VIDEO: "TIKTOK_MICROS_PER_VIDEO",
}
@ -51,6 +52,7 @@ _UNIT_NOUNS: dict[BillingUnit, str] = {
BillingUnit.GOOGLE_MAPS_REVIEW: "review",
BillingUnit.YOUTUBE_VIDEO: "video",
BillingUnit.YOUTUBE_COMMENT: "comment",
BillingUnit.TIKTOK_VIDEO: "video",
}

View file

@ -25,6 +25,7 @@ class BillingUnit(StrEnum):
GOOGLE_MAPS_REVIEW = "google_maps_review"
YOUTUBE_VIDEO = "youtube_video"
YOUTUBE_COMMENT = "youtube_comment"
TIKTOK_VIDEO = "tiktok_video"
class BillableInput(Protocol):

View file

@ -0,0 +1,5 @@
"""``tiktok.*`` namespace: platform-native TikTok data verbs."""
from __future__ import annotations
from app.capabilities.tiktok.scrape import definition as _scrape # noqa: F401

View file

@ -0,0 +1 @@
"""``tiktok.scrape``: public TikTok videos over the browser-driven scraper."""

View file

@ -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)

View 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

View file

@ -0,0 +1,86 @@
"""``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 item = one billable unit."""
return len(self.items)