feat(capabilities): add youtube.scrape and youtube.comments verbs

surfsense_backend/app/capabilities/youtube/__init__.py

@@ -0,0 +1,6 @@
+"""``youtube.*`` namespace: platform-native YouTube data verbs."""
+
+from __future__ import annotations
+
+from app.capabilities.youtube.comments import definition as _comments  # noqa: F401
+from app.capabilities.youtube.scrape import definition as _scrape  # noqa: F401

surfsense_backend/app/capabilities/youtube/comments/__init__.py

@@ -0,0 +1,3 @@
+"""``youtube.comments`` verb: video URLs → comment items (+ replies)."""
+
+from __future__ import annotations

surfsense_backend/app/capabilities/youtube/comments/definition.py

@@ -0,0 +1,23 @@
+"""``youtube.comments`` capability registration (free — see 04-capabilities open item)."""
+
+from __future__ import annotations
+
+from app.capabilities.core import Capability, register_capability
+from app.capabilities.youtube.comments.executor import build_comments_executor
+from app.capabilities.youtube.comments.schemas import CommentsInput, CommentsOutput
+
+YOUTUBE_COMMENTS = Capability(
+    name="youtube.comments",
+    description=(
+        "Fetch public comments (and their replies) for one or more YouTube "
+        "videos. Give it the video URLs; returns structured comment items with "
+        "author, text, like count, reply relationships, and timestamps. Use it "
+        "to gauge sentiment or pull discussion on specific videos."
+    ),
+    input_schema=CommentsInput,
+    output_schema=CommentsOutput,
+    executor=build_comments_executor(),
+    billing_unit=None,
+)
+
+register_capability(YOUTUBE_COMMENTS)

surfsense_backend/app/capabilities/youtube/comments/executor.py

@@ -0,0 +1,30 @@
+"""``youtube.comments`` executor: verb input → Apify actor → comment items."""
+
+from __future__ import annotations
+
+from collections.abc import Awaitable, Callable
+
+from app.capabilities.core import Executor
+from app.capabilities.youtube.comments.schemas import CommentsInput, CommentsOutput
+from app.proprietary.scrapers.youtube import (
+    YouTubeCommentsInput,
+    scrape_comments,
+)
+
+CommentsFn = Callable[[YouTubeCommentsInput], Awaitable[list[dict]]]
+
+
+def build_comments_executor(scrape_fn: CommentsFn | None = None) -> Executor:
+    """Bind the executor to a comments scraper fn (defaults to the proprietary actor)."""
+    scrape_fn = scrape_fn or scrape_comments
+
+    async def execute(payload: CommentsInput) -> CommentsOutput:
+        actor_input = YouTubeCommentsInput(
+            startUrls=[{"url": url} for url in payload.urls],
+            maxComments=payload.max_comments,
+            sortCommentsBy=payload.sort_by,
+        )
+        items = await scrape_fn(actor_input)
+        return CommentsOutput(items=items)
+
+    return execute

surfsense_backend/app/capabilities/youtube/comments/schemas.py

@@ -0,0 +1,44 @@
+"""``youtube.comments`` I/O contracts.
+
+A lean surface over the Apify-compatible ``YouTubeCommentsInput``; the actor's
+``CommentItem`` is reused verbatim as the output element for parity.
+"""
+
+from __future__ import annotations
+
+from typing import Literal
+
+from pydantic import BaseModel, Field
+
+from app.proprietary.scrapers.youtube import CommentItem
+
+MAX_COMMENT_VIDEOS = 20
+"""Per-call cap on how many video URLs one request may harvest comments from."""
+
+
+class CommentsInput(BaseModel):
+    urls: list[str] = Field(
+        min_length=1,
+        max_length=MAX_COMMENT_VIDEOS,
+        description="YouTube video URLs to fetch comments (and replies) for (1-20).",
+    )
+    max_comments: int = Field(
+        default=20,
+        ge=1,
+        le=100_000,
+        description=(
+            "Max items returned per video, counting both top-level comments and "
+            "their replies."
+        ),
+    )
+    sort_by: Literal["TOP_COMMENTS", "NEWEST_FIRST"] = Field(
+        default="NEWEST_FIRST",
+        description="Comment ordering: most-liked first, or most-recent first.",
+    )
+
+
+class CommentsOutput(BaseModel):
+    items: list[CommentItem] = Field(
+        default_factory=list,
+        description="One item per comment or reply, in the scraper's emission order.",
+    )

surfsense_backend/app/capabilities/youtube/scrape/__init__.py

@@ -0,0 +1,3 @@
+"""``youtube.scrape`` verb: YouTube URLs / search queries → video items."""
+
+from __future__ import annotations

surfsense_backend/app/capabilities/youtube/scrape/definition.py

@@ -0,0 +1,24 @@
+"""``youtube.scrape`` capability registration (free — see 04-capabilities open item)."""
+
+from __future__ import annotations
+
+from app.capabilities.core import Capability, register_capability
+from app.capabilities.youtube.scrape.executor import build_scrape_executor
+from app.capabilities.youtube.scrape.schemas import ScrapeInput, ScrapeOutput
+
+YOUTUBE_SCRAPE = Capability(
+    name="youtube.scrape",
+    description=(
+        "Scrape public YouTube data. Give it YouTube URLs (video, channel, "
+        "playlist, shorts, or hashtag) and/or search queries, and it returns "
+        "structured video items — title, views, likes, publish date, channel "
+        "info, description, and optionally subtitles. Use search_queries to "
+        "discover videos, or urls to pull a known video/channel/playlist."
+    ),
+    input_schema=ScrapeInput,
+    output_schema=ScrapeOutput,
+    executor=build_scrape_executor(),
+    billing_unit=None,
+)
+
+register_capability(YOUTUBE_SCRAPE)

surfsense_backend/app/capabilities/youtube/scrape/executor.py

@@ -0,0 +1,36 @@
+"""``youtube.scrape`` executor: verb input → Apify actor → video items."""
+
+from __future__ import annotations
+
+from collections.abc import Awaitable, Callable
+
+from app.capabilities.core import Executor
+from app.capabilities.youtube.scrape.schemas import ScrapeInput, ScrapeOutput
+from app.proprietary.scrapers.youtube import (
+    YouTubeScrapeInput,
+    scrape_youtube,
+)
+
+ScrapeFn = Callable[[YouTubeScrapeInput], 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_youtube
+
+    async def execute(payload: ScrapeInput) -> ScrapeOutput:
+        # Channels emit three content types; cap each at the caller's max_results
+        # so a channel scrape isn't silently limited to plain videos only.
+        actor_input = YouTubeScrapeInput(
+            startUrls=[{"url": url} for url in payload.urls],
+            searchQueries=payload.search_queries,
+            maxResults=payload.max_results,
+            maxResultsShorts=payload.max_results,
+            maxResultStreams=payload.max_results,
+            downloadSubtitles=payload.download_subtitles,
+            subtitlesLanguage=payload.subtitles_language,
+        )
+        items = await scrape_fn(actor_input)
+        return ScrapeOutput(items=items)
+
+    return execute

surfsense_backend/app/capabilities/youtube/scrape/schemas.py

@@ -0,0 +1,66 @@
+"""``youtube.scrape`` I/O contracts.
+
+A lean, agent-friendly surface over the Apify-compatible ``YouTubeScrapeInput``
+(``app/proprietary/scrapers/youtube``). The executor maps this to the full actor
+input; the actor's ``VideoItem`` is reused verbatim as the output element so the
+verb stays parity-faithful with the underlying Apify shape.
+"""
+
+from __future__ import annotations
+
+from pydantic import BaseModel, Field, model_validator
+
+from app.proprietary.scrapers.youtube import VideoItem
+
+MAX_YOUTUBE_SOURCES = 20
+"""Per-call cap on URLs + queries: bounds a synchronous request's fan-out (05)."""
+
+
+class ScrapeInput(BaseModel):
+    urls: list[str] = Field(
+        default_factory=list,
+        max_length=MAX_YOUTUBE_SOURCES,
+        description=(
+            "YouTube URLs to scrape: video, channel (/@handle or /channel/UC...), "
+            "playlist (?list=...), shorts, or hashtag pages. Provide these OR "
+            "search_queries (at least one is required)."
+        ),
+    )
+    search_queries: list[str] = Field(
+        default_factory=list,
+        max_length=MAX_YOUTUBE_SOURCES,
+        description=(
+            "Search terms to run on YouTube; each returns up to max_results videos. "
+            "Provide these OR urls (at least one is required)."
+        ),
+    )
+    max_results: int = Field(
+        default=10,
+        ge=1,
+        le=1000,
+        description=(
+            "Max items to return per source and per content type (videos, shorts, "
+            "streams are capped independently for a channel)."
+        ),
+    )
+    download_subtitles: bool = Field(
+        default=False,
+        description="Also fetch each video's subtitle track (slower; more requests).",
+    )
+    subtitles_language: str = Field(
+        default="en",
+        description="Subtitle language code (e.g. 'en', 'fr'). Used when download_subtitles is true.",
+    )
+
+    @model_validator(mode="after")
+    def _require_a_source(self) -> ScrapeInput:
+        if not self.urls and not self.search_queries:
+            raise ValueError("Provide at least one of 'urls' or 'search_queries'.")
+        return self
+
+
+class ScrapeOutput(BaseModel):
+    items: list[VideoItem] = Field(
+        default_factory=list,
+        description="One video item per result, in the scraper's emission order.",
+    )