diff --git a/surfsense_backend/app/capabilities/instagram/scrape/__init__.py b/surfsense_backend/app/capabilities/instagram/scrape/__init__.py new file mode 100644 index 000000000..de8b3c7c4 --- /dev/null +++ b/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 diff --git a/surfsense_backend/app/capabilities/instagram/scrape/definition.py b/surfsense_backend/app/capabilities/instagram/scrape/definition.py new file mode 100644 index 000000000..e84ca4938 --- /dev/null +++ b/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) diff --git a/surfsense_backend/app/capabilities/instagram/scrape/executor.py b/surfsense_backend/app/capabilities/instagram/scrape/executor.py new file mode 100644 index 000000000..a76722dcd --- /dev/null +++ b/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 diff --git a/surfsense_backend/app/capabilities/instagram/scrape/schemas.py b/surfsense_backend/app/capabilities/instagram/scrape/schemas.py new file mode 100644 index 000000000..0e5646dc7 --- /dev/null +++ b/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)