feat(instagram): add scrape capability verb

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

@@ -0,0 +1,3 @@
+"""``instagram.scrape`` verb: Instagram URLs / search terms → posts, reels, mentions."""
+
+from __future__ import annotations

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

@@ -0,0 +1,23 @@
+"""``instagram.scrape`` capability registration (billed per item; see config
+``INSTAGRAM_SCRAPE_MICROS_PER_ITEM``)."""
+
+from __future__ import annotations
+
+from app.capabilities.core import BillingUnit, Capability, register_capability
+from app.capabilities.instagram.scrape.executor import build_scrape_executor
+from app.capabilities.instagram.scrape.schemas import ScrapeInput, ScrapeOutput
+
+INSTAGRAM_SCRAPE = Capability(
+    name="instagram.scrape",
+    description=(
+        "Scrape public Instagram posts, reels, or mentions from "
+        "profile/post/hashtag/place URLs, or discover content via search queries."
+    ),
+    input_schema=ScrapeInput,
+    output_schema=ScrapeOutput,
+    executor=build_scrape_executor(),
+    billing_unit=BillingUnit.INSTAGRAM_ITEM,
+    docs_url="/docs/connectors/native/instagram",
+)
+
+register_capability(INSTAGRAM_SCRAPE)

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

@@ -0,0 +1,54 @@
+"""``instagram.scrape`` executor: verb input → scraper → media 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.instagram.scrape.schemas import ScrapeInput, ScrapeOutput
+from app.exceptions import ForbiddenError
+from app.proprietary.platforms.instagram import (
+    InstagramAccessBlockedError,
+    InstagramScrapeInput,
+    scrape_instagram,
+)
+
+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_instagram
+
+    async def execute(payload: ScrapeInput) -> ScrapeOutput:
+        actor_input = InstagramScrapeInput(
+            resultsType=payload.result_type,
+            directUrls=payload.urls,
+            search=",".join(payload.search_queries),
+            searchType=payload.search_type,
+            resultsLimit=payload.max_per_target,
+            onlyPostsNewerThan=payload.newer_than,
+            skipPinnedPosts=payload.skip_pinned_posts,
+            addParentData=payload.add_parent_data,
+        )
+        emit_progress(
+            "starting",
+            "Resolving Instagram targets",
+            total=payload.max_items,
+            unit="item",
+        )
+        try:
+            items = await scrape_fn(actor_input, limit=payload.max_items)
+        except InstagramAccessBlockedError as exc:
+            # Anonymous-only scraper; a hard block can't be retried with creds.
+            raise ForbiddenError(
+                f"Instagram refused anonymous access: {exc}",
+                code="INSTAGRAM_ACCESS_BLOCKED",
+            ) from exc
+        emit_progress(
+            "done", f"Scraped {len(items)} item(s)", current=len(items), unit="item"
+        )
+        return ScrapeOutput(items=items)
+
+    return execute

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

@@ -0,0 +1,105 @@
+"""``instagram.scrape`` I/O contracts.
+
+A lean, agent-friendly surface over ``InstagramScrapeInput``
+(``app/proprietary/platforms/instagram``). The executor maps this to the full
+scraper input; the scraper's ``InstagramMediaItem`` is reused verbatim as the
+output element.
+"""
+
+from __future__ import annotations
+
+from typing import Literal
+
+from pydantic import BaseModel, Field, model_validator
+
+from app.proprietary.platforms.instagram import InstagramMediaItem
+
+MAX_INSTAGRAM_SOURCES = 20
+"""Per-call cap on urls + search_queries: bounds a synchronous request's fan-out."""
+
+MAX_INSTAGRAM_ITEMS = 100
+"""Hard ceiling on items returned per call, regardless of the per-target caps."""
+
+
+class ScrapeInput(BaseModel):
+    urls: list[str] = Field(
+        default_factory=list,
+        max_length=MAX_INSTAGRAM_SOURCES,
+        description=(
+            "Instagram URLs or bare profile IDs: profile, post (/p/), reel "
+            "(/reel/), hashtag (/explore/tags/), or place (/explore/locations/). "
+            "Provide these OR search_queries (never both)."
+        ),
+    )
+    search_queries: list[str] = Field(
+        default_factory=list,
+        max_length=MAX_INSTAGRAM_SOURCES,
+        description=(
+            "Discovery keywords (hashtags as plaintext, no '#'). Provide these "
+            "OR urls (never both)."
+        ),
+    )
+    search_type: Literal["hashtag", "profile", "place", "user"] = Field(
+        default="hashtag",
+        description="What to discover from search_queries. Only used with search_queries.",
+    )
+    result_type: Literal["posts", "reels", "mentions"] = Field(
+        default="posts",
+        description="Which feed to return. 'mentions' requires profile URLs.",
+    )
+    newer_than: str | None = Field(
+        default=None,
+        description=(
+            "Only return posts newer than this: YYYY-MM-DD, ISO timestamp, or "
+            "relative ('1 day', '2 months'); UTC."
+        ),
+    )
+    skip_pinned_posts: bool = Field(
+        default=False,
+        description="Exclude pinned posts (posts mode).",
+    )
+    max_per_target: int = Field(
+        default=10,
+        ge=1,
+        description="Max results per URL or per discovered target.",
+    )
+    max_items: int = Field(
+        default=10,
+        ge=1,
+        le=MAX_INSTAGRAM_ITEMS,
+        description="Max total items to return across all sources.",
+    )
+    add_parent_data: bool = Field(
+        default=False,
+        description="Attach a dataSource block to each feed item.",
+    )
+
+    @model_validator(mode="after")
+    def _exactly_one_source(self) -> ScrapeInput:
+        if not self.urls and not self.search_queries:
+            raise ValueError(
+                "Provide at least one of 'urls' or 'search_queries'."
+            )
+        if self.urls and self.search_queries:
+            raise ValueError(
+                "Provide 'urls' OR 'search_queries', not both (they cannot be combined)."
+            )
+        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[InstagramMediaItem] = Field(
+        default_factory=list,
+        description="One media item per result (post/reel/mention), in emission order.",
+    )
+
+    @property
+    def billable_units(self) -> int:
+        """One returned item = one billable unit."""
+        return len(self.items)