feat(google_maps): add scrape + reviews capability verbs

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

@@ -0,0 +1,3 @@
+"""``google_maps.scrape`` verb: search queries / Maps URLs / place IDs → place items."""
+
+from __future__ import annotations

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

@@ -0,0 +1,25 @@
+"""``google_maps.scrape`` capability registration (free — see 04-capabilities open item)."""
+
+from __future__ import annotations
+
+from app.capabilities.core import Capability, register_capability
+from app.capabilities.google_maps.scrape.executor import build_scrape_executor
+from app.capabilities.google_maps.scrape.schemas import ScrapeInput, ScrapeOutput
+
+GOOGLE_MAPS_SCRAPE = Capability(
+    name="google_maps.scrape",
+    description=(
+        "Scrape public Google Maps places. Give it search queries (optionally "
+        "scoped by location), Google Maps URLs, or place IDs, and it returns "
+        "structured place items — name, address, category, phone, website, "
+        "rating, review count, coordinates, and opening hours. Set "
+        "include_details for richer detail-page fields, or max_reviews/"
+        "max_images to attach reviews and photos per place."
+    ),
+    input_schema=ScrapeInput,
+    output_schema=ScrapeOutput,
+    executor=build_scrape_executor(),
+    billing_unit=None,
+)
+
+register_capability(GOOGLE_MAPS_SCRAPE)

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

@@ -0,0 +1,43 @@
+"""``google_maps.scrape`` executor: verb input → scraper → place items."""
+
+from __future__ import annotations
+
+from collections.abc import Awaitable, Callable
+
+from app.capabilities.core import Executor
+from app.capabilities.google_maps.scrape.schemas import ScrapeInput, ScrapeOutput
+from app.exceptions import ForbiddenError
+from app.proprietary.platforms.google_maps import (
+    GoogleMapsScrapeInput,
+    scrape_places,
+)
+from app.proprietary.platforms.google_maps.scraper import SignInRequiredError
+
+ScrapeFn = Callable[[GoogleMapsScrapeInput], 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_places
+
+    async def execute(payload: ScrapeInput) -> ScrapeOutput:
+        actor_input = GoogleMapsScrapeInput(
+            searchStringsArray=payload.search_queries,
+            startUrls=[{"url": url} for url in payload.urls],
+            placeIds=payload.place_ids,
+            locationQuery=payload.location,
+            maxCrawledPlacesPerSearch=payload.max_places,
+            language=payload.language,
+            scrapePlaceDetailPage=payload.include_details,
+            maxReviews=payload.max_reviews,
+            maxImages=payload.max_images,
+        )
+        try:
+            items = await scrape_fn(actor_input)
+        except SignInRequiredError as exc:
+            raise ForbiddenError(
+                f"Google sign in required: {exc}", code="GOOGLE_SIGNIN_REQUIRED"
+            ) from exc
+        return ScrapeOutput(items=items)
+
+    return execute

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

@@ -0,0 +1,91 @@
+"""``google_maps.scrape`` I/O contracts.
+
+A lean, agent-friendly surface over ``GoogleMapsScrapeInput``
+(``app/proprietary/platforms/google_maps``). The executor maps this to the full
+scraper input; the scraper's ``PlaceItem`` is reused verbatim as the output
+element.
+"""
+
+from __future__ import annotations
+
+from pydantic import BaseModel, Field, model_validator
+
+from app.proprietary.platforms.google_maps import PlaceItem
+
+MAX_MAPS_SOURCES = 20
+"""Per-call cap on queries + urls + place_ids: bounds a sync request's fan-out."""
+
+
+class ScrapeInput(BaseModel):
+    search_queries: list[str] = Field(
+        default_factory=list,
+        max_length=MAX_MAPS_SOURCES,
+        description=(
+            "Google Maps search terms (e.g. 'coffee shops', 'dentist'); each "
+            "returns up to max_places. Provide these OR urls OR place_ids "
+            "(at least one is required). Pair with location to scope a search."
+        ),
+    )
+    urls: list[str] = Field(
+        default_factory=list,
+        max_length=MAX_MAPS_SOURCES,
+        description=(
+            "Google Maps URLs — a place page (/maps/place/...) or a search "
+            "results URL. Provide these OR search_queries OR place_ids."
+        ),
+    )
+    place_ids: list[str] = Field(
+        default_factory=list,
+        max_length=MAX_MAPS_SOURCES,
+        description=(
+            "Known Google place IDs (ChIJ...) to fetch directly. Provide these "
+            "OR search_queries OR urls."
+        ),
+    )
+    location: str | None = Field(
+        default=None,
+        description="Location to scope search_queries to, e.g. 'New York, USA'.",
+    )
+    max_places: int = Field(
+        default=10,
+        ge=1,
+        le=1000,
+        description="Max places to return per search query.",
+    )
+    language: str = Field(
+        default="en",
+        description="Result language code, e.g. 'en', 'fr'.",
+    )
+    include_details: bool = Field(
+        default=False,
+        description=(
+            "Also fetch each place's detail page — opening hours, popular "
+            "times, extra contact info (slower; more requests)."
+        ),
+    )
+    max_reviews: int = Field(
+        default=0,
+        ge=0,
+        le=100_000,
+        description="Reviews to attach per place (0 = none).",
+    )
+    max_images: int = Field(
+        default=0,
+        ge=0,
+        description="Images to attach per place (0 = none).",
+    )
+
+    @model_validator(mode="after")
+    def _require_a_source(self) -> ScrapeInput:
+        if not (self.search_queries or self.urls or self.place_ids):
+            raise ValueError(
+                "Provide at least one of 'search_queries', 'urls', or 'place_ids'."
+            )
+        return self
+
+
+class ScrapeOutput(BaseModel):
+    items: list[PlaceItem] = Field(
+        default_factory=list,
+        description="One place item per result, in the scraper's emission order.",
+    )