/")` → `parse_post`, which
+ reads the embedded mobile-v1 `PolarisMedia` JSON (full fidelity) and falls
+ back to Open Graph meta only if that blob is absent. Numeric-ID post URLs are
+ skipped (the page keys on the shortCode).
+3. `fetch_json` / `fetch_html` warm the session on first use, rotate the IP +
+ re-warm on 401/403, back off on 429, return `None` on 404, and raise
+ `InstagramAccessBlockedError` on a `/accounts/login/` redirect.
+4. Parsers map raw web JSON/HTML to flat dicts; the orchestrator stamps
+ `scrapedAt` and applies `resultsLimit` / `onlyPostsNewerThan` as request-time
+ policy.
+
+## Profile discovery (Google-backed)
+
+Instagram's native keyword search is login-walled, so `_discover` resolves a
+query that is a valid handle directly (`"messi"` → `instagram.com/messi/`) and
+routes any other query (e.g. `"national geographic"`) through
+`_discover_via_google`, which calls the `google_search` platform with
+`site:instagram.com`, classifies each organic URL with `resolve_url`, keeps the
+**profile** hits (discovery is profile-only), de-dupes, and caps at `searchLimit`.
+
+Caveats:
+
+- **Coupling**: Instagram depends on the `google_search` platform. The dependency
+ is one-directional and lives behind `_discover_via_google` so it stays testable.
+- **Quality**: results reflect Google's index/ranking of `instagram.com`, not
+ IG's own relevance. This is discovery, not search parity.
+
+## Observed limits & calibration caveats
+
+- Anonymous web JSON/HTML is rate-limited per IP; the sticky-session pool keeps
+ each IP's request rate modest but a hot pool will still hit login walls — that's
+ the `InstagramAccessBlockedError` path, not a bug.
+- `likesCount` is frequently withheld on anonymous responses (surfaces as `-1` or
+ absent upstream); treat it as best-effort.
+- **Single-post extraction** reads the mobile-v1 `PolarisMedia` object embedded in
+ the public `/p/` document (og-meta is a lossy fallback). If Instagram strips both
+ for a given post (private, taken down, or a login interstitial), `parse_post`
+ returns `None` — an honest empty, never a fabricated item. ponytail: the
+ embedded-blob shape can drift; a live probe that dumps the raw HTML pins it (see
+ Testing) and any change is contained to `_find_media` / `parse_post`.
+- The `$3.50 / 1k items` default meter assumes the proxy-bytes-per-item measured
+ on the reference targets; re-measure with the scale harness before high-volume use.
+
+## Testing
+
+- Offline unit tests: `tests/unit/platforms/instagram/` — `test_skeleton.py`
+ (schema + URL resolver), `test_parsers.py` (mapping incl. `parse_post`
+ relay-JSON/og shapes; fixture-pinned tests skip when the fixture is absent),
+ `test_discovery.py` (Google-backed profile discovery with a fake `scrape_serps`),
+ `test_fetch_resilience.py` (warm / rotate / backoff loop + fan-out with fake
+ sessions, no network), `test_budget.py` (fair-share caps + de-dup).
+- Stress / accuracy harness (live, needs network + residential proxy):
+ `scripts/stress/stress_instagram_scraper.py` — `--mode live-discovery` (profile
+ discovery accuracy), `--mode probe-post` (dumps a real anonymous `/p/` payload
+ to `fixtures/post.json` and shows what `parse_post` extracted), `--mode
+ probe-mentions` (settles that the tagged/`mentions` feed is login-walled), and
+ `--mode accuracy` (field coverage across the profile + single-post flows).
+
+```bash
+cd surfsense_backend
+uv run pytest tests/unit/platforms/instagram/
+# Live single-post probe: confirms /p/ is anonymously extractable + pins the shape
+uv run python scripts/stress/stress_instagram_scraper.py --mode probe-post \
+ --post https://www.instagram.com/p//
+```
+
+## TODO / out of scope (v1)
+
+- Deep feed pagination past the first web page of profile media (GraphQL cursor
+ doc-id).
+- Sticky-IP provider parity (same `__sid` caveat as the Reddit sibling).
diff --git a/surfsense_backend/app/proprietary/platforms/instagram/__init__.py b/surfsense_backend/app/proprietary/platforms/instagram/__init__.py
new file mode 100644
index 000000000..2302ca9fc
--- /dev/null
+++ b/surfsense_backend/app/proprietary/platforms/instagram/__init__.py
@@ -0,0 +1,18 @@
+"""Platform-native Instagram scraper (anonymous, no browser)."""
+
+from .fetch import InstagramAccessBlockedError
+from .schemas import (
+ InstagramMediaItem,
+ InstagramProfile,
+ InstagramScrapeInput,
+)
+from .scraper import iter_instagram, scrape_instagram
+
+__all__ = [
+ "InstagramAccessBlockedError",
+ "InstagramMediaItem",
+ "InstagramProfile",
+ "InstagramScrapeInput",
+ "iter_instagram",
+ "scrape_instagram",
+]
diff --git a/surfsense_backend/app/proprietary/platforms/instagram/fetch.py b/surfsense_backend/app/proprietary/platforms/instagram/fetch.py
new file mode 100644
index 000000000..41f692bf1
--- /dev/null
+++ b/surfsense_backend/app/proprietary/platforms/instagram/fetch.py
@@ -0,0 +1,407 @@
+"""Proxy-aware fetch seam for the Instagram scraper (no browser).
+
+All network I/O flows through :func:`fetch_json` and always egresses through the
+residential proxy (a direct hit would expose and risk-block the server IP).
+
+Instagram's public web app exposes anonymous JSON endpoints that a logged-out
+browser calls, guarded by the ``X-IG-App-ID`` web app id and a warmed
+``csrftoken``/``mid`` cookie pair:
+
+ warm one anonymous session (plain GET to ``www.instagram.com/`` mints
+ ``csrftoken``/``mid``), then GET the ``api/v1/*/web_info`` /
+ ``web_profile_info`` endpoints through that same Chrome-impersonated,
+ sticky-IP session with the ``X-IG-App-ID`` header.
+
+This is a direct port of ``../reddit/fetch.py``'s rotate-on-block sticky-session
+pattern (``_RotatingSession`` + ``_current_session`` ContextVar +
+``open_proxy_holder``/``bind_proxy_holder``/``proxy_session``), with an
+Instagram-specific :func:`warm_session` and header set.
+
+Honest ceiling: anonymous Instagram access is the most hostile of our platforms.
+Login walls appear as 401/403 and rotate the exit IP; 429 backs off on the same
+IP. Observed per-IP/session limits are documented in ``README.md``; the safe
+``_FANOUT_CONCURRENCY`` is deliberately low. ponytail: the pacing/rotation
+constants are calibrated to residential exits and may need retuning per pool —
+watch for 401/403/429 log spam and adjust.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import random
+import time
+from collections.abc import Callable
+from contextlib import asynccontextmanager, suppress
+from contextvars import ContextVar
+from datetime import UTC, datetime
+from typing import Any
+from urllib.parse import urlencode
+
+from scrapling.fetchers import AsyncFetcher, FetcherSession
+
+from app.utils.proxy import get_proxy_url
+
+logger = logging.getLogger(__name__)
+
+
+class InstagramAccessBlockedError(RuntimeError):
+ """Raised when every rotated IP is refused anonymous access.
+
+ This is the Instagram login-wall branch: after warming and rotating, the
+ exit IPs still 401/403. We are anonymous-only and cannot log in, so instead
+ of silently returning nothing we surface it as a clear error (mirrors
+ reddit's ``RedditAccessBlockedError`` / google_maps' ``SignInRequiredError``).
+ The executor turns it into a 403 for REST callers.
+ """
+
+
+# Per-flow proxy session, set by ``bind_proxy_holder`` around one continuation
+# chain. Reusing one keep-alive connection pins a single sticky exit IP so the
+# warmed ``csrftoken``/``mid`` cookies (bound to that IP) stay valid across the
+# warm-up + every subsequent web-endpoint fetch. A ContextVar keeps each
+# concurrent fan-out flow on its own session/IP without threading a param
+# through every call.
+_current_session: ContextVar[_RotatingSession | None] = ContextVar(
+ "instagram_proxy_session", default=None
+)
+
+# 401/403 => this IP hit the login wall; rotate to a fresh one and re-warm.
+# 429 => rate limited; back off on the SAME IP (rotating wouldn't help and burns
+# the pool).
+_ROTATE_STATUSES = frozenset({401, 403})
+_BACKOFF_STATUS = 429
+_MAX_ROTATIONS = 3
+_MAX_BACKOFFS = 4
+_BACKOFF_BASE_S = 5.0
+
+# Instagram 429s hard on bursts. Pace each sticky session so a fast IP can't
+# burst past the per-IP threshold. ponytail: 1.5s is tuned to residential exits;
+# a pool with a stricter per-IP cap may need it raised — watch for 429 log spam.
+_MIN_INTERVAL_S = 1.5
+_PACE_JITTER_S = 0.5
+
+# A healthy fetch lands in ~1-2s; cap at 15s so a dead sticky IP costs one
+# bounded wait, then the timeout falls into the generic exception branch of
+# fetch_json and rotates to a fresh IP — same treatment as a 403.
+_REQUEST_TIMEOUT_S = 15.0
+
+# The anonymous web app id every logged-out instagram.com XHR carries. Without
+# it the api/v1/*/web_info and web_profile_info endpoints 403 outright.
+_IG_APP_ID = "936619743392459"
+_HEADERS = {
+ "Accept-Language": "en-US,en;q=0.9",
+ "X-IG-App-ID": _IG_APP_ID,
+ "X-Requested-With": "XMLHttpRequest",
+ "Referer": "https://www.instagram.com/",
+}
+
+# A plain GET to the home page mints the anonymous csrftoken/mid cookie pair.
+_WARM_URL = "https://www.instagram.com/"
+_BASE = "https://www.instagram.com"
+_CSRF_COOKIE = "csrftoken"
+
+# Soft login wall: instead of a 401/403, IG answers an api/v1/* request with a
+# 302 to /accounts/login/ that the impersonated client follows to a 200 login
+# page. The status is 200 but the body is login HTML, so this evades the
+# status-code rotate check — detect it via the response's final URL and treat
+# it exactly like a 403.
+_LOGIN_PATH = "/accounts/login"
+
+
+def now_iso() -> str:
+ """UTC timestamp in the millisecond ISO shape used by scraper output."""
+ return datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z"
+
+
+def _response_cookie_names(page: Any) -> set[str]:
+ """Cookie names set by a response (best-effort across scrapling shapes)."""
+ cookies = getattr(page, "cookies", None)
+ if isinstance(cookies, dict):
+ return set(cookies.keys())
+ return set()
+
+
+def _parse_json(page: Any) -> Any | None:
+ """Parse a scrapling response body into JSON, or ``None``.
+
+ Prefers ``page.json()``; falls back to ``json.loads`` on the raw body when
+ the impersonated response hands back text.
+ """
+ fn = getattr(page, "json", None)
+ if callable(fn):
+ with suppress(Exception):
+ return fn()
+ for attr in ("body", "text"):
+ val = getattr(page, attr, None)
+ if isinstance(val, bytes):
+ val = val.decode("utf-8", "replace")
+ if isinstance(val, str) and val.strip():
+ with suppress(Exception):
+ return json.loads(val)
+ return None
+ return None
+
+
+def _page_text(page: Any) -> str | None:
+ """Best-effort HTML/text body of a scrapling response, or ``None``."""
+ for attr in ("text", "body", "content"):
+ val = getattr(page, attr, None)
+ if callable(val):
+ with suppress(Exception):
+ val = val()
+ if isinstance(val, bytes):
+ val = val.decode("utf-8", "replace")
+ if isinstance(val, str) and val.strip():
+ return val
+ return None
+
+
+def _is_login_redirect(page: Any) -> bool:
+ """True if IG redirected this request to the anonymous login wall.
+
+ A soft block: the final URL lands on ``/accounts/login/`` (served 200), so
+ the status check never fires. Best-effort — returns ``False`` when the
+ response exposes no URL.
+ """
+ final = getattr(page, "url", None)
+ return isinstance(final, str) and _LOGIN_PATH in final
+
+
+def _build_url(path: str, params: dict[str, Any] | None) -> str:
+ """Absolute URL for an instagram.com path (accepts already-absolute URLs)."""
+ base = path if path.startswith("http") else f"{_BASE}/{path.strip('/')}/"
+ if not params:
+ return base
+ qs = urlencode({k: v for k, v in params.items() if v is not None})
+ sep = "&" if "?" in base else "?"
+ return f"{base}{sep}{qs}" if qs else base
+
+
+class _RotatingSession:
+ """Owns one live ``FetcherSession`` (sticky IP) and can swap it for a fresh one.
+
+ ``rotate()`` closes the current keep-alive connection and opens a new one, so
+ the rotating gateway hands out a different residential exit IP. Because the
+ warmed cookies bind to the exit IP, ``rotate()`` also drops the warmed state
+ — the next fetch re-warms on the new IP. Used sequentially within a single
+ flow (never shared across concurrent tasks), so no locking is needed.
+ ``session`` is ``None`` only when no proxy is configured.
+ """
+
+ def __init__(self) -> None:
+ self._cm: Any | None = None
+ self.session: Any | None = None
+ self.rotations = 0
+ self.warmed = False
+ self._last_at = 0.0
+
+ async def _open(self) -> None:
+ proxy = get_proxy_url()
+ self.warmed = False
+ if proxy is None:
+ self._cm = self.session = None
+ return
+ self._cm = FetcherSession(
+ proxy=proxy,
+ stealthy_headers=True,
+ impersonate="chrome",
+ timeout=_REQUEST_TIMEOUT_S,
+ )
+ self.session = await self._cm.__aenter__()
+
+ async def close(self) -> None:
+ if self._cm is not None:
+ with suppress(Exception): # best-effort teardown
+ await self._cm.__aexit__(None, None, None)
+ self._cm = self.session = None
+
+ async def rotate(self) -> Any | None:
+ """Drop the current IP and connect through a fresh one. Returns new session."""
+ await self.close()
+ self.rotations += 1
+ await self._open()
+ logger.info(
+ "[instagram] rotated proxy session (rotation #%d)", self.rotations
+ )
+ return self.session
+
+ async def pace(self) -> None:
+ """Sleep to hold this sticky IP under Instagram's per-IP rate threshold."""
+ wait = _MIN_INTERVAL_S - (time.monotonic() - self._last_at)
+ if wait > 0:
+ await asyncio.sleep(wait + random.uniform(0, _PACE_JITTER_S))
+ self._last_at = time.monotonic()
+
+
+async def open_proxy_holder() -> _RotatingSession:
+ """Open a warm rotate-on-block session holder (caller owns ``close()``)."""
+ holder = _RotatingSession()
+ await holder._open()
+ return holder
+
+
+@asynccontextmanager
+async def bind_proxy_holder(holder: _RotatingSession):
+ """Route this task's fetches through ``holder`` for the enclosed block.
+
+ Does NOT close the holder — enables pooling warm sessions across sequential
+ jobs so each job skips the proxy handshake AND the cookie warm-up.
+ """
+ token = _current_session.set(holder)
+ try:
+ yield holder
+ finally:
+ _current_session.reset(token)
+
+
+@asynccontextmanager
+async def proxy_session():
+ """Open one reused, rotate-on-block proxy session for a continuation chain."""
+ holder = await open_proxy_holder()
+ try:
+ async with bind_proxy_holder(holder):
+ yield holder
+ finally:
+ await holder.close()
+
+
+async def warm_session(session: Any) -> bool:
+ """Mint anonymous ``csrftoken``/``mid`` cookies on a freshly opened session.
+
+ Returns ``True`` when a ``csrftoken`` was issued (the session can now reach
+ the web endpoints), else ``False`` (caller rotates the IP and retries).
+
+ Takes an already-open ``session`` (never constructs one) so tests can drive
+ warm/rotate deterministically with a fake session, exactly like the reddit
+ sibling's fetch-resilience tests.
+ """
+ seen: set[str] = set()
+ with suppress(Exception):
+ page = await session.get(_WARM_URL, headers=_HEADERS)
+ seen |= _response_cookie_names(page)
+ return _CSRF_COOKIE in seen
+
+
+async def _get_page(session: Any, url: str) -> Any:
+ """GET through the warmed sticky session, or a one-shot proxied fetch."""
+ if session is not None:
+ return await session.get(url, headers=_HEADERS)
+ return await AsyncFetcher.get(
+ url,
+ headers=_HEADERS,
+ proxy=get_proxy_url(),
+ stealthy_headers=True,
+ timeout=_REQUEST_TIMEOUT_S,
+ )
+
+
+async def fetch_json(path: str, params: dict[str, Any] | None = None) -> Any | None:
+ """GET an Instagram web endpoint through a warmed session; parse JSON.
+
+ Returns parsed JSON (dict or list), or ``None`` on 404 / non-block failure.
+ See :func:`_fetch` for the warm/rotate/backoff resilience contract.
+ """
+ return await _fetch(path, params, _parse_json)
+
+
+async def fetch_html(path: str, params: dict[str, Any] | None = None) -> str | None:
+ """GET an Instagram web page through a warmed session; return its HTML text.
+
+ Same warm/rotate/backoff resilience as :func:`fetch_json` (a login-wall
+ redirect still raises :class:`InstagramAccessBlockedError`), but hands back
+ the raw HTML body for the pages that embed their data in the document
+ (``/p//`` embedded PolarisMedia JSON / og-meta) instead of a JSON
+ XHR endpoint.
+ """
+ return await _fetch(path, params, _page_text)
+
+
+async def _fetch(
+ path: str,
+ params: dict[str, Any] | None,
+ extract: Callable[[Any], Any | None],
+) -> Any | None:
+ """GET an Instagram web endpoint through a warmed HTTP session.
+
+ Applies ``extract`` to the 200 response (JSON parse or HTML text); returns
+ ``None`` on 404 / non-block failure. Warms cookies once per session; rotates
+ the residential IP and re-warms on 401/403; backs off on 429. Raises
+ :class:`InstagramAccessBlockedError` only when every rotated IP refuses
+ anonymous access (the login-wall branch, which we cannot satisfy).
+ """
+ holder = _current_session.get()
+ if holder is None:
+ # No bound session (e.g. a direct call outside fan_out): open a
+ # short-lived warmed session for this one fetch, then tear it down.
+ async with proxy_session():
+ return await _fetch(path, params, extract)
+
+ url = _build_url(path, params)
+ attempt = 0
+ backoffs = 0
+ while True:
+ session = holder.session
+ try:
+ if session is not None and not holder.warmed:
+ warmed_ok = await warm_session(session)
+ holder.warmed = True # attempted; don't re-warm this IP
+ if not warmed_ok:
+ if attempt < _MAX_ROTATIONS:
+ attempt += 1
+ await holder.rotate()
+ continue
+ raise InstagramAccessBlockedError(
+ f"could not warm session after {attempt} IP rotations for {path}"
+ )
+
+ await holder.pace()
+ page = await _get_page(session, url)
+ status = page.status
+
+ # Endpoint-level login wall (302 -> /accounts/login/, served as 200
+ # login HTML): fail fast, do NOT rotate. Unlike the per-IP 401/403
+ # below — which recovers on a fresh exit IP, so it still rotates —
+ # every rotated IP hits this same wall (observed live), so rotating
+ # only burns the pool and re-warms for an unwinnable block. Raising
+ # (vs returning None) keeps a blocked target distinguishable from an
+ # empty one; fan_out swallows it per-target for partial results.
+ if _is_login_redirect(page):
+ raise InstagramAccessBlockedError(
+ f"Instagram login wall on {path} (endpoint requires auth)"
+ )
+ if status == 200:
+ return extract(page)
+ if status == 404:
+ return None
+ if status == _BACKOFF_STATUS and backoffs < _MAX_BACKOFFS:
+ backoffs += 1
+ delay = _BACKOFF_BASE_S * (2 ** (backoffs - 1))
+ logger.warning(
+ "[instagram] 429 on %s; backing off %.1fs", path, delay
+ )
+ await asyncio.sleep(delay + random.uniform(0, 1))
+ continue
+ if status in _ROTATE_STATUSES:
+ # Bare 401/403: a per-IP block that a fresh exit IP recovers, so
+ # rotate and re-warm. (The endpoint-level auth wall is caught by
+ # the login-redirect branch above and fails fast without rotating.)
+ if attempt < _MAX_ROTATIONS:
+ attempt += 1
+ await holder.rotate()
+ continue
+ raise InstagramAccessBlockedError(
+ f"Instagram refused {path} on {attempt} rotated IPs ({status})"
+ )
+ logger.warning("[instagram] GET %s returned %s", path, status)
+ return None
+ except InstagramAccessBlockedError:
+ raise
+ except Exception as e:
+ logger.warning("[instagram] GET %s failed: %s", path, e)
+ if attempt < _MAX_ROTATIONS:
+ attempt += 1
+ await holder.rotate()
+ continue
+ return None
diff --git a/surfsense_backend/app/proprietary/platforms/instagram/parsers.py b/surfsense_backend/app/proprietary/platforms/instagram/parsers.py
new file mode 100644
index 000000000..5c4e8ea03
--- /dev/null
+++ b/surfsense_backend/app/proprietary/platforms/instagram/parsers.py
@@ -0,0 +1,548 @@
+"""Pure JSON -> item mapping for the Instagram scraper.
+
+Framework-agnostic and I/O-free so it can be unit-tested against captured
+fixtures. Every function takes a raw Instagram web-JSON node and returns a plain
+dict shaped like the public actor spec — no network, no proxy, no ``scrapedAt``
+stamp (the orchestrator adds provenance so these stay deterministic under
+fixture tests).
+
+Instagram's web JSON nests media under GraphQL-style ``edge_*`` containers
+(``edge_media_to_caption``, ``edge_liked_by``, ...) with ``taken_at_timestamp``
+epoch seconds. These parsers flatten that into the actor's camelCase item shape.
+Fields the anonymous endpoints don't expose are left unset (``None``/``[]``) so
+parity is additive.
+"""
+
+from __future__ import annotations
+
+import html
+import json
+import re
+from datetime import UTC, datetime
+from typing import Any
+
+_BASE = "https://www.instagram.com"
+_HASHTAG_RE = re.compile(r"#(\w+)")
+# Instagram handles are letters/digits/period/underscore but never start or end
+# with a period, so anchor both ends to alphanumerics/underscore — otherwise
+# trailing sentence punctuation ("@hulu.") leaks into the handle.
+_MENTION_RE = re.compile(r"@([A-Za-z0-9_](?:[A-Za-z0-9_.]*[A-Za-z0-9_])?)")
+_TYPE_MAP = {
+ "GraphImage": "Image",
+ "GraphVideo": "Video",
+ "GraphSidecar": "Sidecar",
+ "XDTGraphImage": "Image",
+ "XDTGraphVideo": "Video",
+ "XDTGraphSidecar": "Sidecar",
+}
+# Mobile v1 ``media_type``: 1 = image, 2 = video, 8 = carousel/sidecar. Used by
+# the single-post relay parser (the embedded PolarisMedia blob uses this int, not
+# the GraphQL ``__typename`` the profile feed uses).
+_MEDIA_TYPE = {1: "Image", 2: "Video", 8: "Sidecar"}
+
+
+def _int(value: Any) -> int | None:
+ """Coerce to int, or ``None`` (never coerces bools)."""
+ if isinstance(value, bool):
+ return None
+ if isinstance(value, int):
+ return value
+ if isinstance(value, float):
+ return int(value)
+ return None
+
+
+def _utc_from_sec(value: Any) -> str | None:
+ """Epoch seconds -> millisecond ISO string, or ``None``."""
+ if not isinstance(value, int | float) or isinstance(value, bool):
+ return None
+ dt = datetime.fromtimestamp(float(value), tz=UTC)
+ return dt.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z"
+
+
+def _edge_count(node: dict[str, Any], key: str) -> int | None:
+ """``node[key].count`` for a GraphQL ``edge_*`` container."""
+ container = node.get(key)
+ if isinstance(container, dict):
+ return _int(container.get("count"))
+ return None
+
+
+def _edges(container: Any) -> list[dict[str, Any]]:
+ """``container.edges[].node`` list for a GraphQL ``edge_*`` container."""
+ if not isinstance(container, dict):
+ return []
+ out = []
+ for edge in container.get("edges") or []:
+ node = edge.get("node") if isinstance(edge, dict) else None
+ if not isinstance(node, dict):
+ continue
+ out.append(node)
+ return out
+
+
+def _caption_text(node: dict[str, Any]) -> str | None:
+ """First caption edge's text (web feed) or a flat ``caption`` fallback."""
+ edges = _edges(node.get("edge_media_to_caption"))
+ if edges:
+ text = edges[0].get("text")
+ if isinstance(text, str):
+ return text
+ cap = node.get("caption")
+ if isinstance(cap, dict):
+ cap = cap.get("text")
+ if isinstance(cap, str):
+ return cap
+ return None
+
+
+def _likes_count(node: dict[str, Any]) -> int | None:
+ """Like count. ``-1`` (creator hid it) is passed through, never coerced."""
+ for key in ("edge_liked_by", "edge_media_preview_like"):
+ count = _edge_count(node, key)
+ if count is not None:
+ return count
+ return _int(node.get("like_count"))
+
+
+def _shortcode(node: dict[str, Any]) -> str | None:
+ code = node.get("shortcode") or node.get("code")
+ if isinstance(code, str):
+ return code
+ return None
+
+
+def _user_ref(user: Any) -> dict[str, Any] | None:
+ """A trimmed public-user dict (tagged users / coauthor producers), or None.
+
+ Normalizes the two anonymous dialects: the profile feed nests the handle
+ under ``edge_media_to_tagged_user...node.user`` / ``coauthor_producers`` while
+ the single-post relay blob uses ``usertags.in[].user`` — both carry the same
+ scalar user fields, so this trims them to one shape.
+ """
+ if not isinstance(user, dict):
+ return None
+ ref = {
+ "username": user.get("username"),
+ "fullName": user.get("full_name"),
+ "id": user.get("id") or user.get("pk"),
+ "isVerified": user.get("is_verified"),
+ "profilePicUrl": user.get("profile_pic_url"),
+ }
+ return ref if ref["username"] or ref["id"] else None
+
+
+def _iv2_url(iv2: Any) -> str | None:
+ """First candidate URL from a mobile ``image_versions2`` container, or None."""
+ if isinstance(iv2, dict):
+ cands = iv2.get("candidates")
+ if isinstance(cands, list) and cands and isinstance(cands[0], dict):
+ url = cands[0].get("url")
+ return url if isinstance(url, str) else None
+ return None
+
+
+def _location_ref(loc: Any) -> tuple[str | None, str | None]:
+ """``(name, id)`` from a location node, or ``(None, None)``."""
+ if isinstance(loc, dict):
+ lid = loc.get("id") or loc.get("pk")
+ return loc.get("name"), (str(lid) if lid is not None else None)
+ return None, None
+
+
+def _feed_child(node: dict[str, Any]) -> dict[str, Any]:
+ """Map a profile-feed ``edge_sidecar_to_children`` child to a childPost dict."""
+ dims = node.get("dimensions") if isinstance(node.get("dimensions"), dict) else {}
+ is_video = bool(node.get("is_video"))
+ return {
+ "id": node.get("id"),
+ "shortCode": node.get("shortcode"),
+ "type": "Video" if is_video else "Image",
+ "displayUrl": node.get("display_url"),
+ "videoUrl": node.get("video_url") if is_video else None,
+ "alt": node.get("accessibility_caption"),
+ "dimensionsHeight": _int(dims.get("height")),
+ "dimensionsWidth": _int(dims.get("width")),
+ }
+
+
+def _relay_child(node: dict[str, Any]) -> dict[str, Any]:
+ """Map a single-post relay ``carousel_media`` child to a childPost dict."""
+ mt = node.get("media_type")
+ vv = node.get("video_versions")
+ video_url = (
+ vv[0].get("url") if isinstance(vv, list) and vv and isinstance(vv[0], dict) else None
+ )
+ is_video = mt == 2 or bool(video_url)
+ return {
+ "id": node.get("id"),
+ "shortCode": node.get("code"),
+ "type": _MEDIA_TYPE.get(mt) or ("Video" if is_video else "Image"),
+ "displayUrl": _iv2_url(node.get("image_versions2")) or node.get("display_uri"),
+ "videoUrl": video_url,
+ "alt": node.get("accessibility_caption"),
+ "dimensionsHeight": _int(node.get("original_height")),
+ "dimensionsWidth": _int(node.get("original_width")),
+ }
+
+
+def parse_media(node: dict[str, Any]) -> dict[str, Any]:
+ """Map a raw timeline/feed media node to a flat media item dict."""
+ code = _shortcode(node)
+ caption = _caption_text(node)
+ typename = node.get("__typename")
+ owner = node.get("owner") if isinstance(node.get("owner"), dict) else {}
+ dims = node.get("dimensions") if isinstance(node.get("dimensions"), dict) else {}
+ is_video = bool(node.get("is_video"))
+ children = _edges(node.get("edge_sidecar_to_children"))
+ tagged = [
+ ref
+ for n in _edges(node.get("edge_media_to_tagged_user"))
+ if (ref := _user_ref(n.get("user")))
+ ]
+ coauthors = [
+ ref for c in (node.get("coauthor_producers") or []) if (ref := _user_ref(c))
+ ]
+ loc_name, loc_id = _location_ref(node.get("location"))
+ return {
+ "id": node.get("id"),
+ "type": _TYPE_MAP.get(typename) or ("Video" if is_video else "Image"),
+ "shortCode": code,
+ "caption": caption,
+ "hashtags": _HASHTAG_RE.findall(caption) if caption else [],
+ "mentions": _MENTION_RE.findall(caption) if caption else [],
+ "url": f"{_BASE}/p/{code}/" if code else None,
+ "commentsCount": _edge_count(node, "edge_media_to_comment")
+ or _int(node.get("comment_count")),
+ "dimensionsHeight": _int(dims.get("height")),
+ "dimensionsWidth": _int(dims.get("width")),
+ "displayUrl": node.get("display_url"),
+ "images": [c.get("display_url") for c in children if c.get("display_url")],
+ "childPosts": [_feed_child(c) for c in children],
+ "videoUrl": node.get("video_url") if is_video else None,
+ "alt": node.get("accessibility_caption"),
+ "likesCount": _likes_count(node),
+ "videoViewCount": _int(node.get("video_view_count")) if is_video else None,
+ "videoDuration": node.get("video_duration") if is_video else None,
+ "timestamp": _utc_from_sec(node.get("taken_at_timestamp")),
+ "ownerUsername": owner.get("username"),
+ "ownerId": owner.get("id") or node.get("owner_id"),
+ "ownerFullName": owner.get("full_name"),
+ "isPinned": bool(node.get("pinned_for_users")),
+ "productType": node.get("product_type"),
+ "paidPartnership": node.get("is_paid_partnership"),
+ "taggedUsers": tagged,
+ "coauthorProducers": coauthors,
+ "musicInfo": node.get("clips_music_attribution_info"),
+ "locationName": loc_name,
+ "locationId": loc_id,
+ "isCommentsDisabled": node.get("comments_disabled"),
+ }
+
+
+def parse_profile(user: dict[str, Any]) -> dict[str, Any]:
+ """Map a raw ``web_profile_info`` ``data.user`` to a flat profile item dict."""
+ username = user.get("username")
+ latest = [parse_media(n) for n in _edges(user.get("edge_owner_to_timeline_media"))]
+ related = [
+ ref for n in _edges(user.get("edge_related_profiles")) if (ref := _user_ref(n))
+ ]
+ return {
+ "detailKind": "profile",
+ "id": user.get("id"),
+ "username": username,
+ "url": f"{_BASE}/{username}/" if username else None,
+ "fullName": user.get("full_name"),
+ "biography": user.get("biography"),
+ "externalUrl": user.get("external_url"),
+ "followersCount": _edge_count(user, "edge_followed_by"),
+ "followsCount": _edge_count(user, "edge_follow"),
+ "postsCount": _edge_count(user, "edge_owner_to_timeline_media"),
+ "highlightReelCount": _int(user.get("highlight_reel_count")),
+ "igtvVideoCount": _edge_count(user, "edge_felix_video_timeline"),
+ "isBusinessAccount": user.get("is_business_account"),
+ "businessCategoryName": user.get("business_category_name"),
+ "private": user.get("is_private"),
+ "verified": user.get("is_verified"),
+ "profilePicUrl": user.get("profile_pic_url"),
+ "profilePicUrlHD": user.get("profile_pic_url_hd"),
+ "relatedProfiles": related,
+ "latestPosts": latest,
+ }
+
+
+# Anonymous single-post extraction (/p//, /reel//) --------
+#
+# Instagram serves logged-out visitors the post's full metadata inside the
+# document itself, not via a JSON XHR (the ``?__a=1`` API 404s / login-walls for
+# anonymous callers). Two anonymous surfaces carry it, in order of fidelity:
+# 1. An inline ``', re.DOTALL
+)
+_OG_RE = re.compile(
+ r' on Instagram: ".
+_OG_TITLE_RE = re.compile(r"^(.+?)\s+on Instagram:\s*(.*)$", re.DOTALL)
+# The numeric media id (pk) rides in the App Link deep-link meta tags
+# (al:ios:url / al:android:url = "instagram://media?id=") on anonymous pages,
+# even though the og:* tags omit it.
+_MEDIA_ID_RE = re.compile(r"instagram://media\?id=(\d+)")
+
+
+def _og_date_to_iso(value: str) -> str | None:
+ """``"July 9, 2026"`` -> ``"2026-07-09"`` (date-only; og carries no time)."""
+ try:
+ return datetime.strptime(value, "%B %d, %Y").replace(tzinfo=UTC).date().isoformat()
+ except ValueError:
+ return None
+
+
+def _clean_caption(raw: str) -> str | None:
+ """HTML-unescape and unwrap the surrounding quotes off an og caption preview."""
+ return html.unescape(raw).strip().strip('"').strip() or None
+
+
+def _parse_og_meta(og: dict[str, str]) -> dict[str, Any]:
+ """Lift post fields out of Instagram's Open Graph tags (see module notes above).
+
+ Caption + full name come from ``og:title``; counts + username + date from
+ ``og:description``. Every field is optional and independently guarded, so a
+ shape we don't recognise yields a partial (or empty) dict instead of raising.
+ """
+ out: dict[str, Any] = {}
+ desc = og.get("description", "")
+ title = og.get("title", "")
+
+ counts = _OG_COUNTS_RE.search(desc)
+ if counts:
+ out["likes"] = _html_int(counts.group(1))
+ out["comments"] = _html_int(counts.group(2))
+
+ owner_date = _OG_OWNER_DATE_RE.search(desc)
+ if owner_date:
+ out["ownerUsername"] = owner_date.group(1).strip().lstrip("@") or None
+ out["timestamp"] = _og_date_to_iso(owner_date.group(2))
+
+ tm = _OG_TITLE_RE.match(title)
+ if tm:
+ out["ownerFullName"] = tm.group(1).strip() or None
+ out["caption"] = _clean_caption(tm.group(2))
+ elif owner_date:
+ # No usable og:title: fall back to the caption after og:description's
+ # date prefix — still clean (the counts/username/date are stripped).
+ out["caption"] = _clean_caption(desc[owner_date.end():])
+ return out
+
+
+def _html_int(value: Any) -> int | None:
+ """Coerce a string/number (``"1,234"``) to int, or ``None``."""
+ if isinstance(value, bool):
+ return None
+ if isinstance(value, int | float):
+ return int(value)
+ if isinstance(value, str):
+ digits = value.replace(",", "").strip()
+ if digits.isdigit():
+ return int(digits)
+ return None
+
+
+def _og_tags(html: str) -> dict[str, str]:
+ """Map ``og:`` -> content for the post document."""
+ return {k.lower(): v for k, v in _OG_RE.findall(html)}
+
+
+def _find_media(root: Any, shortcode: str | None) -> dict[str, Any] | None:
+ """Depth-first search a JSON tree for the post's mobile-v1 media object.
+
+ Matches on ``code == shortcode`` (so a carousel *child* or a related post
+ can't be picked instead of the target) plus ``taken_at`` and an id, which
+ together uniquely identify the top-level ``PolarisMedia`` node.
+ """
+ stack = [root]
+ while stack:
+ cur = stack.pop()
+ if isinstance(cur, dict):
+ if (
+ cur.get("taken_at") is not None
+ and ("pk" in cur or "id" in cur)
+ and (shortcode is None or cur.get("code") == shortcode)
+ ):
+ return cur
+ stack.extend(cur.values())
+ elif isinstance(cur, list):
+ stack.extend(cur)
+ return None
+
+
+def _relay_media(html: str, shortcode: str | None) -> dict[str, Any] | None:
+ """Locate the embedded ``PolarisMedia`` object for this post, or ``None``.
+
+ The logged-out media payload is inlined as one of ~40 ``application/json``
+ script blocks. We only ``json.loads`` blocks that mention ``taken_at`` (and
+ the shortcode when known) so a single post fetch doesn't parse every blob.
+ """
+ for raw in _APP_JSON_RE.findall(html):
+ if "taken_at" not in raw:
+ continue
+ if shortcode and shortcode not in raw:
+ continue
+ try:
+ data = json.loads(raw)
+ except (ValueError, TypeError):
+ continue
+ media = _find_media(data, shortcode)
+ if media is not None:
+ return media
+ return None
+
+
+def _media_from_relay(
+ media: dict[str, Any], *, url: str, shortcode: str | None
+) -> dict[str, Any]:
+ """Map an embedded mobile-v1 ``PolarisMedia`` object to a flat media item.
+
+ Same output shape as :func:`parse_media` (so it flows through
+ ``InstagramMediaItem`` unchanged), sourced from the relay dialect
+ (``user``/``taken_at``/``usertags.in``/``carousel_media``/flat counts).
+ """
+ mt = media.get("media_type")
+ cap = media.get("caption")
+ caption = (
+ cap.get("text") if isinstance(cap, dict) else (cap if isinstance(cap, str) else None)
+ )
+ carousel = media.get("carousel_media")
+ carousel = [c for c in carousel if isinstance(c, dict)] if isinstance(carousel, list) else []
+ vv = media.get("video_versions")
+ video_url = (
+ vv[0].get("url") if isinstance(vv, list) and vv and isinstance(vv[0], dict) else None
+ )
+ is_video = mt == 2 or bool(video_url)
+ owner = media.get("user") if isinstance(media.get("user"), dict) else {}
+ tagged = [
+ ref
+ for t in ((media.get("usertags") or {}).get("in") or [])
+ if isinstance(t, dict) and (ref := _user_ref(t.get("user")))
+ ]
+ coauthors = [
+ ref for c in (media.get("coauthor_producers") or []) if (ref := _user_ref(c))
+ ]
+ loc_name, loc_id = _location_ref(media.get("location"))
+ # The relay ``id`` is ``POLARIS_``; strip the prefix so single-post ids
+ # match the numeric pk that og-fallback + the al:ios meta also yield.
+ ident = media.get("id")
+ if isinstance(ident, str) and ident.startswith("POLARIS_"):
+ ident = ident[len("POLARIS_") :]
+ pk = media.get("pk")
+ media_id = ident or (str(pk) if pk is not None else None)
+ return {
+ "id": media_id,
+ "type": _MEDIA_TYPE.get(mt) or ("Video" if is_video else "Image"),
+ "shortCode": media.get("code") or shortcode,
+ "caption": caption,
+ "hashtags": list(dict.fromkeys(_HASHTAG_RE.findall(caption))) if caption else [],
+ "mentions": list(dict.fromkeys(_MENTION_RE.findall(caption))) if caption else [],
+ "url": url,
+ "commentsCount": _int(media.get("comment_count")),
+ "dimensionsHeight": _int(media.get("original_height")),
+ "dimensionsWidth": _int(media.get("original_width")),
+ "displayUrl": _iv2_url(media.get("image_versions2")) or media.get("display_uri"),
+ "images": [
+ u
+ for c in carousel
+ if (u := _iv2_url(c.get("image_versions2")) or c.get("display_uri"))
+ ],
+ "childPosts": [_relay_child(c) for c in carousel],
+ "videoUrl": video_url,
+ "alt": media.get("accessibility_caption"),
+ "likesCount": _int(media.get("like_count")),
+ "videoViewCount": _int(media.get("view_count") or media.get("play_count"))
+ if is_video
+ else None,
+ "videoDuration": media.get("video_duration") if is_video else None,
+ "timestamp": _utc_from_sec(media.get("taken_at")),
+ "ownerUsername": owner.get("username"),
+ "ownerId": owner.get("id") or owner.get("pk"),
+ "ownerFullName": owner.get("full_name"),
+ "productType": media.get("product_type"),
+ "taggedUsers": tagged,
+ "coauthorProducers": coauthors,
+ "locationName": loc_name,
+ "locationId": loc_id,
+ }
+
+
+def parse_post(
+ html: str | None, *, url: str, shortcode: str | None = None
+) -> dict[str, Any] | None:
+ """Map an anonymous ``/p//`` (or ``/reel/``) HTML page to a media dict.
+
+ Prefers the embedded mobile-v1 ``PolarisMedia`` relay JSON (full fidelity),
+ falling back to the lossy Open Graph meta tags only if that blob is absent.
+ Returns a dict shaped like :func:`parse_media` (so it flows through
+ ``InstagramMediaItem`` unchanged), or ``None`` when the document carries
+ neither surface (e.g. a login interstitial slipped past the fetch-layer
+ redirect check — the caller treats ``None`` as "nothing to emit", never a
+ silent success).
+ """
+ if not isinstance(html, str) or not html.strip():
+ return None
+
+ media = _relay_media(html, shortcode)
+ if media is not None:
+ return _media_from_relay(media, url=url, shortcode=shortcode)
+
+ # Fallback: no embedded relay blob -> Open Graph meta only.
+ og = _og_tags(html)
+ if not og:
+ return None
+ og_meta = _parse_og_meta(og)
+ caption = og_meta.get("caption")
+ video_url = og.get("video")
+ is_video = bool(video_url) or og.get("type") == "video.other"
+ id_match = _MEDIA_ID_RE.search(html)
+ return {
+ "id": id_match.group(1) if id_match else None,
+ "type": "Video" if is_video else "Image",
+ "shortCode": shortcode,
+ "caption": caption,
+ "hashtags": list(dict.fromkeys(_HASHTAG_RE.findall(caption))) if caption else [],
+ "mentions": list(dict.fromkeys(_MENTION_RE.findall(caption))) if caption else [],
+ "url": url,
+ "commentsCount": og_meta.get("comments"),
+ "displayUrl": og.get("image"),
+ "videoUrl": video_url if is_video else None,
+ "likesCount": og_meta.get("likes"),
+ "timestamp": og_meta.get("timestamp"),
+ "ownerUsername": og_meta.get("ownerUsername"),
+ "ownerFullName": og_meta.get("ownerFullName"),
+ }
diff --git a/surfsense_backend/app/proprietary/platforms/instagram/schemas.py b/surfsense_backend/app/proprietary/platforms/instagram/schemas.py
new file mode 100644
index 000000000..930a2acb9
--- /dev/null
+++ b/surfsense_backend/app/proprietary/platforms/instagram/schemas.py
@@ -0,0 +1,136 @@
+# ruff: noqa: N815
+"""Input/output models for the Instagram scraper.
+
+The models mirror the public Instagram scraper actor spec so the endpoint can be
+a drop-in: the input accepts the full documented surface, and every output field
+is emitted (``None``/``[]`` when the anonymous web endpoints cannot source it
+yet) so the contract expands additively — the same rule the Google Search and
+YouTube models follow.
+
+**Anonymous only.** There is deliberately **no** authentication field on the
+input (no username/password/token/cookie/``login*``) — the scraper holds only
+Instagram's anonymous web-session cookies (``csrftoken``/``mid``) and can never
+log in. Anything auth-shaped a caller sends lands in ``extra`` and is ignored.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Literal
+
+from pydantic import BaseModel, ConfigDict, Field
+
+InstagramResultsType = Literal["posts", "details", "reels"]
+# Kept as distinct values for actor-spec parity, but under anonymous Google-backed
+# discovery ``user`` and ``profile`` are aliases: both resolve to profile targets
+# (IG's own hashtag/place/keyword search is login-walled).
+InstagramSearchType = Literal["profile", "user"]
+
+
+class InstagramScrapeInput(BaseModel):
+ """Instagram scraper input surface (anonymous, no auth fields).
+
+ Field names mirror the public actor spec verbatim. ``resultsLimit`` /
+ ``searchLimit`` are collector policy applied by :func:`scrape_instagram`,
+ NOT ceilings baked into the streaming flows. Fields the acquisition layer
+ doesn't source yet are still accepted via ``extra="allow"`` for parity.
+ """
+
+ model_config = ConfigDict(extra="allow")
+ resultsType: InstagramResultsType = "posts"
+ directUrls: list[str] = Field(default_factory=list)
+ resultsLimit: int | None = Field(default=None, ge=1)
+ onlyPostsNewerThan: str | None = None
+ search: str | None = None
+ searchType: InstagramSearchType = "profile"
+ searchLimit: int | None = Field(default=None, ge=1, le=250)
+ addParentData: bool = False
+ skipPinnedPosts: bool = False
+ addProfileStatistics: bool = False
+
+
+class _ItemBase(BaseModel):
+ """Common error / provenance fields carried on every output item.
+
+ Errors surface as item-level fields (never exceptions) so a partial run
+ still returns the items it could source, mirroring the actor's shape.
+ """
+
+ model_config = ConfigDict(extra="allow")
+ inputUrl: str | None = None
+ error: str | None = None
+ errorDescription: str | None = None
+ requestErrorMessages: list[str] = Field(default_factory=list)
+
+ def to_output(self) -> dict[str, Any]:
+ """Serialize to the flat output dict shape (keeps extras)."""
+ return self.model_dump(exclude_none=False)
+
+
+class InstagramMediaItem(_ItemBase):
+ """A post or reel. One flat schema per the actor FAQ.
+
+ ``firstComment``/``latestComments`` are intentionally absent: comment
+ *content* is login-walled (only the anonymous comment *count* is exposed, as
+ ``commentsCount``), so this scraper can never source them.
+ """
+
+ id: str | None = None
+ type: Literal["Image", "Video", "Sidecar"] | None = None
+ shortCode: str | None = None
+ caption: str | None = None
+ hashtags: list[str] = Field(default_factory=list)
+ mentions: list[str] = Field(default_factory=list)
+ url: str | None = None
+ commentsCount: int | None = None
+ dimensionsHeight: int | None = None
+ dimensionsWidth: int | None = None
+ displayUrl: str | None = None
+ images: list[str] = Field(default_factory=list)
+ videoUrl: str | None = None
+ alt: str | None = None
+ likesCount: int | None = None
+ videoViewCount: int | None = None
+ videoPlayCount: int | None = None
+ reshareCount: int | None = None
+ timestamp: str | None = None
+ childPosts: list[dict[str, Any]] = Field(default_factory=list)
+ ownerUsername: str | None = None
+ ownerId: str | None = None
+ ownerFullName: str | None = None
+ isPinned: bool | None = None
+ productType: str | None = None
+ videoDuration: float | None = None
+ paidPartnership: bool | None = None
+ taggedUsers: list[dict[str, Any]] = Field(default_factory=list)
+ musicInfo: dict[str, Any] | None = None
+ coauthorProducers: list[dict[str, Any]] = Field(default_factory=list)
+ locationName: str | None = None
+ locationId: str | None = None
+ isCommentsDisabled: bool | None = None
+ dataSource: dict[str, Any] | None = None
+
+
+class InstagramProfile(_ItemBase):
+ """A profile detail item (``detailKind = "profile"``)."""
+
+ detailKind: Literal["profile"] = "profile"
+ id: str | None = None
+ username: str | None = None
+ url: str | None = None
+ fullName: str | None = None
+ biography: str | None = None
+ externalUrl: str | None = None
+ followersCount: int | None = None
+ followsCount: int | None = None
+ postsCount: int | None = None
+ highlightReelCount: int | None = None
+ igtvVideoCount: int | None = None
+ isBusinessAccount: bool | None = None
+ businessCategoryName: str | None = None
+ private: bool | None = None
+ verified: bool | None = None
+ profilePicUrl: str | None = None
+ profilePicUrlHD: str | None = None
+ relatedProfiles: list[dict[str, Any]] = Field(default_factory=list)
+ latestPosts: list[dict[str, Any]] = Field(default_factory=list)
+ statistics: dict[str, Any] | None = None
diff --git a/surfsense_backend/app/proprietary/platforms/instagram/scraper.py b/surfsense_backend/app/proprietary/platforms/instagram/scraper.py
new file mode 100644
index 000000000..a4c22bde7
--- /dev/null
+++ b/surfsense_backend/app/proprietary/platforms/instagram/scraper.py
@@ -0,0 +1,433 @@
+"""Orchestrator for the Instagram scraper.
+
+The core is the async generator :func:`iter_instagram` (unbounded);
+:func:`scrape_instagram` is a thin collector with a caller-supplied ``limit``
+guard. Any cap is caller policy, never baked into flow logic.
+
+Independent targets (one per ``directUrl`` / discovered entity) fan out
+concurrently on a pool of warm sessions (sticky IPs); each target's own paging
+stays sequential. ``fan_out`` is ported from ``../reddit/scraper.py`` but bound
+to *this* module's proxy holders so every worker warms its own session once and
+reuses it.
+
+Anonymous-only. Every surface here is reachable without a login: profile web
+info, the media embedded in a profile page, single-post/reel pages, and
+Google-backed handle discovery. Login-walled surfaces (hashtag/place feeds,
+comment threads, IG's native keyword search) are deliberately absent.
+
+Flows are selected by ``resultsType``:
+- ``posts`` / ``reels`` -> media items (profile feed, or a single
+ ``/p/``/``/reel/`` page, or discovery search)
+- ``details`` -> profile metadata (by URL or discovery search)
+
+ponytail: deep feed pagination (past the first web page of media) needs the
+GraphQL cursor endpoint whose doc-id drifts; v1 emits the first page and stops.
+The upgrade path is a ``_paginate_feed`` helper in this file plus a doc-id in
+``fetch.py`` — contained to these two files, per the acquisition-seam rule.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import re
+from collections.abc import AsyncIterator
+from contextlib import aclosing
+from datetime import UTC, datetime, timedelta
+from typing import Any
+
+from app.proprietary.platforms.google_search.schemas import GoogleSearchScrapeInput
+from app.proprietary.platforms.google_search.scraper import scrape_serps
+
+from .fetch import (
+ InstagramAccessBlockedError,
+ bind_proxy_holder,
+ fetch_html,
+ fetch_json,
+ now_iso,
+ open_proxy_holder,
+)
+from .parsers import parse_media, parse_post, parse_profile
+from .schemas import InstagramScrapeInput
+from .url_resolver import ResolvedUrl, resolve_url
+
+logger = logging.getLogger(__name__)
+
+__all__ = [
+ "InstagramAccessBlockedError",
+ "iter_instagram",
+ "scrape_instagram",
+]
+
+# Independent jobs run concurrently on a pool of warm proxy sessions. Anonymous
+# Instagram is the most hostile platform, so this stays low to avoid burning the
+# residential pool with parallel login walls.
+_FANOUT_CONCURRENCY = 8
+
+_PROFILE_PATH = "api/v1/users/web_profile_info/"
+
+# Instagram usernames: 1-30 chars of letters/digits/period/underscore. Used to
+# treat a profile/user discovery query as a direct (anonymous) handle lookup.
+_HANDLE_RE = re.compile(r"[A-Za-z0-9._]{1,30}\Z")
+
+
+def _parse_newer_than(value: str | None) -> datetime | None:
+ """Parse ``onlyPostsNewerThan`` (ISO, YYYY-MM-DD, or relative) to UTC.
+
+ Relative forms: ``" "`` where unit is minute/hour/day/week/month/
+ year (singular or plural). Anything unparseable returns ``None`` (no filter).
+ """
+ if not value:
+ return None
+ text = value.strip().lower()
+ parts = text.split()
+ if len(parts) == 2 and parts[0].isdigit():
+ n = int(parts[0])
+ unit = parts[1].rstrip("s")
+ days = {
+ "minute": n / 1440,
+ "hour": n / 24,
+ "day": n,
+ "week": n * 7,
+ "month": n * 30,
+ "year": n * 365,
+ }.get(unit)
+ if days is None:
+ return None
+ return datetime.now(UTC) - timedelta(days=days)
+ try:
+ dt = datetime.fromisoformat(value.replace("Z", "+00:00"))
+ if dt.tzinfo:
+ return dt
+ return dt.replace(tzinfo=UTC)
+ except ValueError:
+ return None
+
+
+def _is_after(timestamp: str | None, cutoff: datetime | None) -> bool:
+ """True if the item ``timestamp`` (ISO) is at/after the cutoff (or no cutoff)."""
+ if cutoff is None:
+ return True
+ if not timestamp:
+ return True
+ try:
+ dt = datetime.fromisoformat(timestamp.replace("Z", "+00:00"))
+ return dt >= cutoff
+ except ValueError:
+ return True
+
+
+async def fan_out(
+ jobs: list[AsyncIterator[dict[str, Any]]], *, concurrency: int = _FANOUT_CONCURRENCY
+) -> AsyncIterator[dict[str, Any]]:
+ """Stream items from independent async-iterator jobs via a warm worker pool.
+
+ Each worker opens ONE proxy session and reuses it across the sequential jobs
+ it pulls, so only the first job per worker pays the proxy handshake + the
+ cookie warm-up. Partial results (matches the reddit sibling): one blocked or
+ failed target yields nothing rather than aborting the batch — Instagram is
+ an aggregation, not an atomic transaction, so 4/5 good targets beat 0/5. But
+ if EVERY target was refused (zero items AND a hard block seen), the whole run
+ couldn't reach anonymous data, so we surface ``InstagramAccessBlockedError``
+ (-> 403) instead of a misleading empty success. Workers are cancelled and
+ their sessions closed if the consumer stops early.
+ """
+ if not jobs:
+ return
+ job_queue: asyncio.Queue[AsyncIterator[dict[str, Any]]] = asyncio.Queue()
+ for job in jobs:
+ job_queue.put_nowait(job)
+ results: asyncio.Queue[list[dict[str, Any]]] = asyncio.Queue()
+ blocked = False # set if any target hit a hard login/auth wall
+
+ async def worker() -> None:
+ nonlocal blocked
+ holder = None
+ try:
+ holder = await open_proxy_holder()
+ except Exception as e: # no session: jobs still run via one-shot fetches
+ logger.warning("[instagram] proxy session open failed: %s", e)
+ try:
+ while True:
+ try:
+ job = job_queue.get_nowait()
+ except asyncio.QueueEmpty:
+ return
+ items: list[dict[str, Any]] = []
+ try:
+ if holder is not None:
+ async with bind_proxy_holder(holder):
+ items = [item async for item in job]
+ else:
+ items = [item async for item in job]
+ except InstagramAccessBlockedError as e:
+ # Partial results: a blocked target must not kill the batch.
+ # Record it so a fully-blocked run can still surface the 403.
+ blocked = True
+ logger.warning("[instagram] target blocked: %s", e)
+ except Exception as e: # one bad target must not kill the run
+ logger.warning("[instagram] fan-out job failed: %s", e)
+ await results.put(items)
+ finally:
+ if holder is not None:
+ await holder.close()
+
+ tasks = [asyncio.create_task(worker()) for _ in range(min(concurrency, len(jobs)))]
+ emitted = 0
+ try:
+ for _ in range(len(jobs)):
+ for item in await results.get():
+ emitted += 1
+ yield item
+ finally:
+ for task in tasks:
+ if not task.done():
+ task.cancel()
+ await asyncio.gather(*tasks, return_exceptions=True)
+ # Reached only on natural exhaustion (an early-stop close raises GeneratorExit
+ # inside the loop and skips this). Nothing came back AND a wall was hit ->
+ # the run was fully refused, so fail loud rather than return empty.
+ if emitted == 0 and blocked:
+ raise InstagramAccessBlockedError(
+ "Instagram refused anonymous access to every target"
+ )
+
+
+def _emit(partial: dict[str, Any], *, input_url: str | None) -> dict[str, Any]:
+ """Stamp provenance and serialize (parsers return plain dicts)."""
+ out = {**partial, "scrapedAt": now_iso()}
+ if input_url is not None:
+ out.setdefault("inputUrl", input_url)
+ return out
+
+
+async def _profile_user(username: str) -> dict[str, Any] | None:
+ """Fetch a profile's ``data.user`` node, or ``None``."""
+ data = await fetch_json(_PROFILE_PATH, {"username": username})
+ if isinstance(data, dict):
+ user = (
+ data.get("data", {}).get("user")
+ if isinstance(data.get("data"), dict)
+ else None
+ )
+ if isinstance(user, dict):
+ return user
+ return None
+ return None
+
+
+def _media_matches(item: dict[str, Any], result_type: str) -> bool:
+ """Filter a media item by feed type. ``reels`` keeps clips/videos only."""
+ if result_type == "reels":
+ return item.get("type") == "Video" or item.get("productType") == "clips"
+ return True
+
+
+async def _media_flow(
+ resolved: ResolvedUrl,
+ *,
+ input_model: InstagramScrapeInput,
+ cutoff: datetime | None,
+ per_target: int,
+) -> AsyncIterator[dict[str, Any]]:
+ """Emit media items for a profile feed, or a single ``/p/``/``/reel/`` page."""
+ from .parsers import _edges
+
+ result_type = input_model.resultsType
+ if resolved.kind == "profile":
+ user = await _profile_user(resolved.value)
+ if user is None:
+ return
+ nodes = _edges(user.get("edge_owner_to_timeline_media"))
+ emitted = 0
+ for node in nodes:
+ item = parse_media(node)
+ if input_model.skipPinnedPosts and item.get("isPinned"):
+ continue
+ if not _media_matches(item, result_type):
+ continue
+ if not _is_after(item.get("timestamp"), cutoff):
+ continue
+ yield _emit(item, input_url=resolved.url)
+ emitted += 1
+ if emitted >= per_target:
+ return
+ return
+ if resolved.kind in ("post", "reel"):
+ # Single-post extraction: the anonymous ``?__a=1`` JSON API 404s/login-
+ # walls, but the public /p// document embeds the mobile-v1
+ # PolarisMedia JSON (og-meta fallback), which parse_post reads. Numeric-ID
+ # URLs can't be keyed this way (the page needs the shortCode), so they're
+ # skipped upstream.
+ if resolved.numeric_post_id:
+ return
+ html = await fetch_html(f"p/{resolved.value}/")
+ item = parse_post(html, url=resolved.url, shortcode=resolved.value)
+ if item is None:
+ return
+ if not _media_matches(item, result_type):
+ return
+ if not _is_after(item.get("timestamp"), cutoff):
+ return
+ yield _emit(item, input_url=resolved.url)
+ return
+
+
+async def _details_flow(
+ resolved: ResolvedUrl, *, input_model: InstagramScrapeInput
+) -> AsyncIterator[dict[str, Any]]:
+ """Emit one profile detail item for a URL (anonymous web_profile_info)."""
+ if resolved.kind == "profile":
+ user = await _profile_user(resolved.value)
+ if user is not None:
+ yield _emit(parse_profile(user), input_url=resolved.url)
+
+
+def _kind_matches(resolved: ResolvedUrl, search_type: str) -> bool:
+ """True if a resolved IG URL is the kind the discovery query asked for.
+
+ Discovery is profile-only now (hashtag/place feeds are login-walled), so
+ every supported ``search_type`` maps to a profile target.
+ """
+ return resolved.kind == "profile"
+
+
+async def _discover_via_google(
+ query: str, *, search_type: str, limit: int
+) -> list[ResolvedUrl]:
+ """Discover IG profile targets via Google ``site:instagram.com`` (anonymous).
+
+ Instagram's own keyword search is login-walled, so we reuse the existing
+ ``google_search`` platform, classify each organic URL with ``resolve_url``,
+ keep the profile hits, de-dup, and cap at ``limit``.
+
+ Quality caveat: results reflect Google's index/ranking of instagram.com, not
+ IG's own relevance. This is discovery, not search parity (see README).
+ """
+ serps = await scrape_serps(
+ GoogleSearchScrapeInput(
+ queries=query, site="instagram.com", maxPagesPerQuery=2
+ ),
+ limit=2,
+ )
+ resolved: list[ResolvedUrl] = []
+ seen: set[tuple[str, str]] = set()
+ for serp in serps:
+ for org in serp.get("organicResults") or []:
+ url = org.get("url", "") if isinstance(org, dict) else ""
+ r = resolve_url(url)
+ if r is None or not _kind_matches(r, search_type):
+ continue
+ key = (r.kind, r.value)
+ if key in seen:
+ continue
+ seen.add(key)
+ resolved.append(r)
+ if len(resolved) >= limit:
+ return resolved
+ return resolved
+
+
+async def _discover(
+ query: str, *, search_type: str, limit: int
+) -> list[ResolvedUrl]:
+ """Resolve a discovery query into profile targets - anonymously.
+
+ A query that is a valid handle resolves directly against the anonymous
+ profile endpoint ("messi" -> instagram.com/messi/). A non-handle query (e.g.
+ "national geographic") goes through Google ``site:instagram.com`` since IG's
+ native keyword search is login-walled.
+ """
+ handle = query.strip().lstrip("@")
+ if _HANDLE_RE.match(handle):
+ url = f"https://www.instagram.com/{handle}/"
+ return [ResolvedUrl("profile", handle, url)][:limit]
+ return await _discover_via_google(query, search_type=search_type, limit=limit)
+
+
+def _resolve_inputs(input_model: InstagramScrapeInput) -> list[ResolvedUrl]:
+ """Resolve ``directUrls`` (URLs take priority over ``search``)."""
+ resolved: list[ResolvedUrl] = []
+ for url in input_model.directUrls:
+ r = resolve_url(url)
+ if r is None:
+ logger.warning("[instagram] unrecognized URL: %s", url)
+ continue
+ resolved.append(r)
+ return resolved
+
+
+async def _targets(input_model: InstagramScrapeInput) -> list[ResolvedUrl]:
+ """The resolved targets for this run: direct URLs, else discovery search."""
+ if input_model.directUrls:
+ return _resolve_inputs(input_model)
+ if not input_model.search:
+ return []
+ limit = input_model.searchLimit or 10
+ queries = [q.strip() for q in input_model.search.split(",") if q.strip()]
+ targets: list[ResolvedUrl] = []
+ for query in queries:
+ targets.extend(
+ await _discover(query, search_type=input_model.searchType, limit=limit)
+ )
+ return targets
+
+
+async def iter_instagram(
+ input_model: InstagramScrapeInput,
+) -> AsyncIterator[dict[str, Any]]:
+ """Yield flat Instagram items. ``directUrls`` override ``search``.
+
+ Independent targets fan out concurrently; each target's paging stays
+ sequential. De-dupes media by ``id`` across targets.
+ """
+ targets = await _targets(input_model)
+ if not targets:
+ return
+ result_type = input_model.resultsType
+ cutoff = _parse_newer_than(input_model.onlyPostsNewerThan)
+ per_target = input_model.resultsLimit or 10
+
+ if result_type == "details":
+ jobs = [_details_flow(r, input_model=input_model) for r in targets]
+ async with aclosing(fan_out(jobs)) as stream:
+ async for item in stream:
+ yield item
+ return
+
+ # posts / reels -> media feeds, de-duped by id across targets.
+ jobs = [
+ _media_flow(
+ r, input_model=input_model, cutoff=cutoff, per_target=per_target
+ )
+ for r in targets
+ ]
+ seen: set[str] = set()
+ async with aclosing(fan_out(jobs)) as stream:
+ async for item in stream:
+ item_id = item.get("id")
+ if isinstance(item_id, str):
+ if item_id in seen:
+ continue
+ seen.add(item_id)
+ yield item
+
+
+async def scrape_instagram(
+ input_model: InstagramScrapeInput, *, limit: int | None = None
+) -> list[dict[str, Any]]:
+ """Collect :func:`iter_instagram` into a list, honoring an optional ``limit``.
+
+ ``limit`` is a request-time policy guard, NOT a ceiling in the streaming
+ core.
+ """
+ from app.capabilities.core.progress import emit_progress
+
+ results: list[dict[str, Any]] = []
+ async with aclosing(iter_instagram(input_model)) as stream:
+ async for item in stream:
+ results.append(item)
+ emit_progress("scraping", current=len(results), total=limit, unit="item")
+ if limit is not None and len(results) >= limit:
+ break
+ return results
diff --git a/surfsense_backend/app/proprietary/platforms/instagram/url_resolver.py b/surfsense_backend/app/proprietary/platforms/instagram/url_resolver.py
new file mode 100644
index 000000000..08f5c6553
--- /dev/null
+++ b/surfsense_backend/app/proprietary/platforms/instagram/url_resolver.py
@@ -0,0 +1,91 @@
+"""Classify and normalize an Instagram URL into a scrape job.
+
+Covers the anonymously-scrapable ``directUrls`` shapes: a profile, a post
+(``/p/``), and a reel (``/reel/``), plus bare profile IDs. Hashtag and place
+URLs are deliberately unsupported — their feeds are login-walled for anonymous
+callers (use Google-backed discovery + single-post extraction instead).
+Non-Instagram hosts resolve to ``None`` so the orchestrator can skip them.
+Mirrors the frozen ``ResolvedUrl`` dataclass shape of ``../reddit/url_resolver.py``.
+
+Normalization rules (from the reference spec):
+- ``_u/`` and ``/profilecard/`` segments are stripped.
+- Story URLs (``/stories//...``) reduce to the profile.
+- Numeric post-ID URLs cannot be single-post-extracted anonymously (the HTML
+ page keys on the shortCode), so they resolve with ``numeric_post_id`` set and
+ the media flow skips them.
+- ``share/`` links are unsupported (they need a network redirect to resolve to a
+ canonical post/profile URL); pass the resolved ``/p/`` or profile URL instead.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Literal
+from urllib.parse import urlparse
+
+ResolvedKind = Literal["profile", "post", "reel"]
+
+_INSTAGRAM_HOSTS = frozenset(
+ {"m.instagram.com", "www.instagram.com", "instagram.com"}
+)
+_STRIP_SEGMENTS = frozenset({"_u", "profilecard"})
+_RESERVED = frozenset(
+ {"p", "s", "tv", "reel", "reels", "share", "explore", "stories", "accounts"}
+)
+
+
+@dataclass(frozen=True)
+class ResolvedUrl:
+ """A classified Instagram target: the kind, its key, and the source URL."""
+
+ kind: ResolvedKind
+ value: str
+ url: str
+ numeric_post_id: bool = False
+
+
+def _is_instagram_host(hostname: str | None) -> bool:
+ if not hostname:
+ return False
+ return hostname.lower() in _INSTAGRAM_HOSTS
+
+
+def _segments(url: str) -> list[str]:
+ parsed = urlparse(url)
+ if not _is_instagram_host(parsed.hostname):
+ return []
+ if not parsed.path:
+ return []
+ return [s for s in parsed.path.split("/") if s and s not in _STRIP_SEGMENTS]
+
+
+def resolve_url(url: str) -> ResolvedUrl | None:
+ """Classify an Instagram URL into a scrape job, or ``None`` if unrecognized.
+
+ A bare token with no ``http`` prefix and no ``/`` is treated as a profile ID
+ (the reference accepts bare profile handles alongside full URLs).
+ """
+ if "instagram.com" not in url.lower():
+ token = url.strip().lstrip("@")
+ if token and "/" not in token and "." not in token:
+ return ResolvedUrl(
+ "profile", token, f"https://www.instagram.com/{token}/"
+ )
+ segments = _segments(url)
+ if not segments:
+ return None
+ head = segments[0]
+ if head == "p" and len(segments) >= 2:
+ code = segments[1]
+ return ResolvedUrl("post", code, url, numeric_post_id=code.isdigit())
+ if head in ("reel", "reels") and len(segments) >= 2:
+ code = segments[1]
+ return ResolvedUrl("reel", code, url, numeric_post_id=code.isdigit())
+ if head == "stories" and len(segments) >= 2:
+ user = segments[1]
+ return ResolvedUrl(
+ "profile", user, f"https://www.instagram.com/{user}/"
+ )
+ if head not in _RESERVED:
+ return ResolvedUrl("profile", head, url)
+ return None
diff --git a/surfsense_backend/app/routes/__init__.py b/surfsense_backend/app/routes/__init__.py
index 3c28fd65c..85450d4b7 100644
--- a/surfsense_backend/app/routes/__init__.py
+++ b/surfsense_backend/app/routes/__init__.py
@@ -3,6 +3,7 @@ from fastapi import APIRouter, Depends
# Import verb namespaces for their registration side effects before the door builds.
import app.capabilities.google_maps
import app.capabilities.google_search
+import app.capabilities.instagram
import app.capabilities.reddit
import app.capabilities.tiktok
import app.capabilities.web
diff --git a/surfsense_backend/scripts/e2e_instagram_scraper.py b/surfsense_backend/scripts/e2e_instagram_scraper.py
new file mode 100644
index 000000000..c81075932
--- /dev/null
+++ b/surfsense_backend/scripts/e2e_instagram_scraper.py
@@ -0,0 +1,238 @@
+"""Manual functional e2e for the Instagram scraper (app/proprietary/platforms/instagram).
+
+Run from the backend directory:
+ cd surfsense_backend
+ uv run python scripts/e2e_instagram_scraper.py
+ # or: .venv/bin/python scripts/e2e_instagram_scraper.py
+
+This is NOT a pytest test (it needs live network + a residential/custom proxy).
+It:
+
+ Step 0 — go/no-go probe: open a proxy session, mint the anonymous
+ ``csrftoken``/``mid`` cookies, then fetch ``web_profile_info`` on the SAME
+ sticky IP and assert it returns a profile. If this fails the whole
+ approach is invalid — later steps are skipped.
+ Step 1 — scrape a profile's posts.
+ Step 2 — scrape a profile's reels.
+ Step 3 — anonymous single-post extraction for a discovered ``/p/`` URL.
+ Step 4 — fetch profile details.
+ Step 5 — run a profile discovery search (Google-backed).
+ Step 6 — dump trimmed, PII-anonymized raw fixtures into
+ tests/unit/platforms/instagram/fixtures/ for the offline parser tests.
+
+Anonymous-only: hashtag/place feeds and comment threads are login-walled and are
+not part of the scraper, so there are no steps for them.
+"""
+
+import asyncio
+import json
+import sys
+from pathlib import Path
+
+from dotenv import load_dotenv
+
+# --- bootstrap: load .env and put the backend root on sys.path before app.* ---
+_BACKEND_ROOT = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(_BACKEND_ROOT))
+for _candidate in (_BACKEND_ROOT / ".env", _BACKEND_ROOT.parent / ".env"):
+ if _candidate.exists():
+ load_dotenv(_candidate)
+ break
+
+from app.proprietary.platforms.instagram import ( # noqa: E402
+ InstagramScrapeInput,
+ scrape_instagram,
+)
+from app.proprietary.platforms.instagram.fetch import ( # noqa: E402
+ fetch_html,
+ fetch_json,
+ proxy_session,
+ warm_session,
+)
+from app.proprietary.platforms.instagram.url_resolver import resolve_url # noqa: E402
+
+_PROFILE = "natgeo"
+_SEARCH_TERM = "national geographic"
+
+_FIXTURE_DIR = (
+ _BACKEND_ROOT / "tests" / "unit" / "platforms" / "instagram" / "fixtures"
+)
+
+# Fields to strip from dumped fixtures so we never commit PII / volatile tokens.
+_PII_KEYS = frozenset(
+ {"profile_pic_url", "profile_pic_url_hd", "display_url", "video_url", "biography"}
+)
+
+
+def _hr(title: str) -> None:
+ print(f"\n{'=' * 70}\n{title}\n{'=' * 70}")
+
+
+def _check(label: str, ok: bool, detail: str = "") -> bool:
+ print(f" [{'PASS' if ok else 'FAIL'}] {label}{f' — {detail}' if detail else ''}")
+ return ok
+
+
+def _anonymize(obj):
+ """Recursively blank PII-ish string values so fixtures are safe to commit."""
+ if isinstance(obj, dict):
+ return {
+ k: ("" if k in _PII_KEYS and v else _anonymize(v))
+ for k, v in obj.items()
+ }
+ if isinstance(obj, list):
+ return [_anonymize(x) for x in obj]
+ return obj
+
+
+async def step0_probe() -> bool:
+ _hr("STEP 0 — go/no-go: csrftoken warm-up + sticky web_profile_info")
+ async with proxy_session() as holder:
+ if holder.session is None:
+ return _check(
+ "proxy configured", False, "no proxy -> set PROXY_PROVIDER + creds"
+ )
+ minted = await warm_session(holder.session)
+ holder.warmed = True # don't let fetch_json re-warm; we just warmed it
+ _check("csrftoken warm-up minted a session", minted)
+ data = await fetch_json(
+ "api/v1/users/web_profile_info/", {"username": _PROFILE}
+ )
+ user = (data or {}).get("data", {}).get("user") if isinstance(data, dict) else None
+ print(f" web_profile_info({_PROFILE}) -> user={'yes' if user else 'no'}")
+ return _check("sticky web_profile_info", minted and bool(user))
+
+
+async def step1_posts() -> bool:
+ _hr("STEP 1 — profile posts")
+ items = await scrape_instagram(
+ InstagramScrapeInput(
+ resultsType="posts",
+ directUrls=[f"https://www.instagram.com/{_PROFILE}/"],
+ resultsLimit=5,
+ ),
+ limit=5,
+ )
+ for it in items[:5]:
+ print(f" - {it.get('shortCode')} | likes={it.get('likesCount')}")
+ return _check("profile returned posts", len(items) > 0, f"{len(items)} posts")
+
+
+async def step2_reels() -> bool:
+ _hr("STEP 2 — profile reels")
+ items = await scrape_instagram(
+ InstagramScrapeInput(
+ resultsType="reels",
+ directUrls=[f"https://www.instagram.com/{_PROFILE}/"],
+ resultsLimit=5,
+ ),
+ limit=5,
+ )
+ print(f" {len(items)} reels for {_PROFILE}")
+ return _check("reels returned items", len(items) >= 0, f"{len(items)} reels")
+
+
+async def step3_single_post(post_url: str | None) -> bool:
+ _hr("STEP 3 — single-post extraction for a /p/ URL")
+ if not post_url:
+ return _check("had a post URL", False, "step 1 found no post")
+ items = await scrape_instagram(
+ InstagramScrapeInput(
+ resultsType="posts", directUrls=[post_url], resultsLimit=1
+ ),
+ limit=1,
+ )
+ got = items[0] if items else {}
+ print(f" {len(items)} item for {post_url} | owner={got.get('ownerUsername')}")
+ return _check("single post returned an item", len(items) > 0, post_url)
+
+
+async def step4_details() -> bool:
+ _hr("STEP 4 — profile details")
+ items = await scrape_instagram(
+ InstagramScrapeInput(
+ resultsType="details",
+ directUrls=[f"https://www.instagram.com/{_PROFILE}/"],
+ ),
+ limit=10,
+ )
+ kinds = sorted({i.get("detailKind") for i in items})
+ print(f" detail kinds={kinds}")
+ return _check("details returned", len(items) > 0, f"{len(items)} items {kinds}")
+
+
+async def step5_search() -> bool:
+ _hr("STEP 5 — profile discovery search (Google-backed)")
+ items = await scrape_instagram(
+ InstagramScrapeInput(
+ resultsType="posts",
+ search=_SEARCH_TERM,
+ searchType="profile",
+ searchLimit=3,
+ resultsLimit=3,
+ ),
+ limit=9,
+ )
+ print(f" {len(items)} items for search '{_SEARCH_TERM}'")
+ return _check("search returned results", len(items) >= 0, f"{len(items)} items")
+
+
+async def step6_dump_fixtures(post_url: str | None) -> bool:
+ _hr("STEP 6 — dump trimmed, anonymized fixtures for offline tests")
+ profile = await fetch_json(
+ "api/v1/users/web_profile_info/", {"username": _PROFILE}
+ )
+ _FIXTURE_DIR.mkdir(parents=True, exist_ok=True)
+ wrote = []
+ if isinstance(profile, dict) and profile.get("data", {}).get("user"):
+ (_FIXTURE_DIR / "profile.json").write_text(
+ json.dumps(_anonymize(profile)), encoding="utf-8"
+ )
+ wrote.append("profile.json")
+ resolved = resolve_url(post_url) if post_url else None
+ if resolved is not None and resolved.kind in ("post", "reel"):
+ html = await fetch_html(f"p/{resolved.value}/")
+ if html:
+ (_FIXTURE_DIR / "post.json").write_text(
+ json.dumps(
+ {"url": post_url, "shortcode": resolved.value, "html": html}
+ ),
+ encoding="utf-8",
+ )
+ wrote.append("post.json")
+ return _check("dumped fixtures", bool(wrote), f"{wrote} -> {_FIXTURE_DIR}")
+
+
+async def _first_post_url() -> str | None:
+ """Discover a live post URL from the target profile's first page."""
+ items = await scrape_instagram(
+ InstagramScrapeInput(
+ resultsType="posts",
+ directUrls=[f"https://www.instagram.com/{_PROFILE}/"],
+ resultsLimit=1,
+ ),
+ limit=1,
+ )
+ return items[0].get("url") if items else None
+
+
+async def main() -> int:
+ results = [await step0_probe()]
+ if not results[-1]:
+ print("\ncookie probe failed — the approach is invalid on this IP/proxy.")
+ print("Aborting remaining steps.")
+ return 1
+ results.append(await step1_posts())
+ results.append(await step2_reels())
+ post_url = await _first_post_url()
+ results.append(await step3_single_post(post_url))
+ results.append(await step4_details())
+ results.append(await step5_search())
+ results.append(await step6_dump_fixtures(post_url))
+ _hr("SUMMARY")
+ print(f" {sum(results)}/{len(results)} steps passed")
+ return 0 if all(results) else 1
+
+
+if __name__ == "__main__":
+ raise SystemExit(asyncio.run(main()))
diff --git a/surfsense_backend/tests/unit/agents/multi_agent_chat/test_subagent_composition.py b/surfsense_backend/tests/unit/agents/multi_agent_chat/test_subagent_composition.py
index 66e8f28db..ee889cd17 100644
--- a/surfsense_backend/tests/unit/agents/multi_agent_chat/test_subagent_composition.py
+++ b/surfsense_backend/tests/unit/agents/multi_agent_chat/test_subagent_composition.py
@@ -32,6 +32,7 @@ _EXPECTED_SUBAGENTS = frozenset(
"google_drive",
"google_maps",
"google_search",
+ "instagram",
"knowledge_base",
"mcp_discovery",
"memory",
diff --git a/surfsense_backend/tests/unit/capabilities/instagram/__init__.py b/surfsense_backend/tests/unit/capabilities/instagram/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/surfsense_backend/tests/unit/capabilities/instagram/test_executor.py b/surfsense_backend/tests/unit/capabilities/instagram/test_executor.py
new file mode 100644
index 000000000..a19e597c5
--- /dev/null
+++ b/surfsense_backend/tests/unit/capabilities/instagram/test_executor.py
@@ -0,0 +1,82 @@
+"""Executor tests: lean verb input → ``InstagramScrapeInput`` mapping + wrapping.
+
+A fake scraper captures the actor input the executor built (no network), so the
+snake_case→camelCase mapping and the ``InstagramAccessBlockedError`` →
+``ForbiddenError`` translation are asserted deterministically.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from app.capabilities.instagram.details.executor import build_details_executor
+from app.capabilities.instagram.details.schemas import DetailsInput, DetailsOutput
+from app.capabilities.instagram.scrape.executor import build_scrape_executor
+from app.capabilities.instagram.scrape.schemas import ScrapeInput, ScrapeOutput
+from app.exceptions import ForbiddenError
+from app.proprietary.platforms.instagram import (
+ InstagramAccessBlockedError,
+ InstagramScrapeInput,
+)
+
+pytestmark = pytest.mark.unit
+
+
+class _FakeScraper:
+ def __init__(self, items: list[dict]) -> None:
+ self.items = items
+ self.calls: list[tuple[InstagramScrapeInput, int | None]] = []
+
+ async def __call__(self, actor_input, *, limit=None):
+ self.calls.append((actor_input, limit))
+ return self.items
+
+
+async def test_scrape_maps_urls_and_wraps_items():
+ fake = _FakeScraper([{"id": "1", "shortCode": "abc", "caption": "hi"}])
+ execute = build_scrape_executor(fake)
+ out = await execute(ScrapeInput(urls=["https://www.instagram.com/natgeo/"]))
+ assert isinstance(out, ScrapeOutput)
+ assert out.items[0].shortCode == "abc"
+ actor_input, limit = fake.calls[0]
+ assert actor_input.resultsType == "posts"
+ assert actor_input.directUrls == ["https://www.instagram.com/natgeo/"]
+ assert actor_input.search == ""
+ assert limit == 10 # default max_items forwarded as the collector limit
+
+
+async def test_scrape_joins_search_queries():
+ fake = _FakeScraper([])
+ execute = build_scrape_executor(fake)
+ await execute(ScrapeInput(search_queries=["natgeo", "nasa"], search_type="profile"))
+ actor_input, _ = fake.calls[0]
+ assert actor_input.search == "natgeo,nasa"
+ assert actor_input.searchType == "profile"
+ assert actor_input.directUrls == []
+
+
+async def test_scrape_access_blocked_maps_to_forbidden():
+ async def _blocked(actor_input, *, limit=None):
+ raise InstagramAccessBlockedError("login wall")
+
+ execute = build_scrape_executor(_blocked)
+ with pytest.raises(ForbiddenError):
+ await execute(ScrapeInput(urls=["x"]))
+
+
+async def test_details_maps_and_wraps_discriminated_items():
+ fake = _FakeScraper(
+ [
+ {
+ "detailKind": "profile",
+ "username": "natgeo",
+ "url": "https://www.instagram.com/natgeo/",
+ }
+ ]
+ )
+ execute = build_details_executor(fake)
+ out = await execute(DetailsInput(urls=["https://www.instagram.com/natgeo/"]))
+ assert isinstance(out, DetailsOutput)
+ assert out.items[0].username == "natgeo"
+ actor_input, _ = fake.calls[0]
+ assert actor_input.resultsType == "details"
diff --git a/surfsense_backend/tests/unit/capabilities/instagram/test_registry.py b/surfsense_backend/tests/unit/capabilities/instagram/test_registry.py
new file mode 100644
index 000000000..503feedff
--- /dev/null
+++ b/surfsense_backend/tests/unit/capabilities/instagram/test_registry.py
@@ -0,0 +1,29 @@
+"""The instagram namespace registers its verbs for the doors/agent to read.
+
+Unlike the stale reddit assertion (``billing_unit is None``), these assert the
+real meters — the Capability definitions are the source of truth.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from app.capabilities import (
+ instagram, # noqa: F401 — importing the namespace registers its verbs
+)
+from app.capabilities.core.store import get_capability
+from app.capabilities.core.types import BillingUnit
+
+pytestmark = pytest.mark.unit
+
+
+def test_instagram_scrape_registered_with_item_meter():
+ cap = get_capability("instagram.scrape")
+ assert cap.name == "instagram.scrape"
+ assert cap.billing_unit is BillingUnit.INSTAGRAM_ITEM
+
+
+def test_instagram_details_registered_with_item_meter():
+ cap = get_capability("instagram.details")
+ assert cap.name == "instagram.details"
+ assert cap.billing_unit is BillingUnit.INSTAGRAM_ITEM
diff --git a/surfsense_backend/tests/unit/capabilities/instagram/test_schemas.py b/surfsense_backend/tests/unit/capabilities/instagram/test_schemas.py
new file mode 100644
index 000000000..afabbf833
--- /dev/null
+++ b/surfsense_backend/tests/unit/capabilities/instagram/test_schemas.py
@@ -0,0 +1,71 @@
+"""``instagram.*`` input guards: source exclusivity and bounded batches."""
+
+from __future__ import annotations
+
+import pytest
+from pydantic import ValidationError
+
+from app.capabilities.instagram.details.schemas import DetailsInput, DetailsOutput
+from app.capabilities.instagram.scrape.schemas import (
+ MAX_INSTAGRAM_ITEMS,
+ MAX_INSTAGRAM_SOURCES,
+ ScrapeInput,
+)
+
+pytestmark = pytest.mark.unit
+
+
+def test_scrape_rejects_no_source():
+ with pytest.raises(ValidationError):
+ ScrapeInput()
+
+
+def test_scrape_rejects_both_sources():
+ with pytest.raises(ValidationError):
+ ScrapeInput(urls=["https://www.instagram.com/natgeo/"], search_queries=["fit"])
+
+
+def test_scrape_accepts_urls_only():
+ payload = ScrapeInput(urls=["https://www.instagram.com/natgeo/"])
+ assert payload.search_queries == []
+ assert payload.estimated_units == payload.max_items
+
+
+def test_scrape_bounds():
+ with pytest.raises(ValidationError):
+ ScrapeInput(
+ urls=["https://www.instagram.com/x/"],
+ max_items=MAX_INSTAGRAM_ITEMS + 1,
+ )
+ with pytest.raises(ValidationError):
+ ScrapeInput(
+ urls=[
+ f"https://www.instagram.com/u{i}/"
+ for i in range(MAX_INSTAGRAM_SOURCES + 1)
+ ]
+ )
+
+
+def test_scrape_rejects_walled_search_type():
+ # Discovery is profile-only; hashtag/place are login-walled and rejected.
+ with pytest.raises(ValidationError):
+ ScrapeInput(search_queries=["travel"], search_type="hashtag")
+
+
+def test_details_wraps_profile_items():
+ out = DetailsOutput(
+ items=[
+ {"detailKind": "profile", "username": "natgeo"},
+ {"detailKind": "profile", "username": "nasa"},
+ ]
+ )
+ kinds = [type(i).__name__ for i in out.items]
+ assert kinds == ["InstagramProfile", "InstagramProfile"]
+ assert out.billable_units == 2
+
+
+def test_details_rejects_both_sources():
+ with pytest.raises(ValidationError):
+ DetailsInput(
+ urls=["https://www.instagram.com/natgeo/"], search_queries=["x"]
+ )
diff --git a/surfsense_backend/tests/unit/capabilities/test_billing.py b/surfsense_backend/tests/unit/capabilities/test_billing.py
index 36d9a2978..85187b5f5 100644
--- a/surfsense_backend/tests/unit/capabilities/test_billing.py
+++ b/surfsense_backend/tests/unit/capabilities/test_billing.py
@@ -427,3 +427,54 @@ async def test_platform_gate_disabled_is_noop(monkeypatch):
)
session.execute.assert_not_called()
+
+
+# ===================================================================
+# Instagram per-item / per-comment billing
+# ===================================================================
+
+
+async def test_instagram_item_charges_owner_per_item(
+ monkeypatch, record_usage, _enable_platform_billing
+):
+ monkeypatch.setattr(config, "INSTAGRAM_SCRAPE_MICROS_PER_ITEM", 3500)
+ session, user = _make_session(_OWNER, balance_micros=1_000_000)
+
+ charged = await charge_capability(
+ _FakePlatformOutput(4), BillingUnit.INSTAGRAM_ITEM, _ctx(session)
+ )
+
+ assert charged == 4 * 3500
+ assert user.credit_micros_balance == 1_000_000 - 4 * 3500
+ kwargs = record_usage.await_args.kwargs
+ assert kwargs["usage_type"] == "instagram_item"
+
+
+async def test_instagram_comment_charges_owner_per_comment(
+ monkeypatch, record_usage, _enable_platform_billing
+):
+ monkeypatch.setattr(config, "INSTAGRAM_SCRAPE_MICROS_PER_COMMENT", 1500)
+ session, user = _make_session(_OWNER, balance_micros=1_000_000)
+
+ charged = await charge_capability(
+ _FakePlatformOutput(6), BillingUnit.INSTAGRAM_COMMENT, _ctx(session)
+ )
+
+ assert charged == 6 * 1500
+ assert user.credit_micros_balance == 1_000_000 - 6 * 1500
+ kwargs = record_usage.await_args.kwargs
+ assert kwargs["usage_type"] == "instagram_comment"
+
+
+async def test_instagram_gate_blocks_when_worst_case_exceeds_balance(
+ monkeypatch, _enable_platform_billing
+):
+ monkeypatch.setattr(config, "INSTAGRAM_SCRAPE_MICROS_PER_ITEM", 3500)
+ session = _gate_session(_OWNER, balance_micros=5000) # affords 1 item, not 2
+
+ with pytest.raises(InsufficientCreditsError):
+ await gate_capability(
+ _FakePlatformInput(estimated_units=2),
+ BillingUnit.INSTAGRAM_ITEM,
+ _ctx(session),
+ )
diff --git a/surfsense_backend/tests/unit/platforms/instagram/__init__.py b/surfsense_backend/tests/unit/platforms/instagram/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/surfsense_backend/tests/unit/platforms/instagram/test_budget.py b/surfsense_backend/tests/unit/platforms/instagram/test_budget.py
new file mode 100644
index 000000000..ea500be40
--- /dev/null
+++ b/surfsense_backend/tests/unit/platforms/instagram/test_budget.py
@@ -0,0 +1,97 @@
+"""Offline budget tests: per-target caps, cross-target de-dup, and the limit guard.
+
+No network. ``fetch_json`` is stubbed with a synthetic profile payload and the
+fan-out proxy holders are replaced with no-ops, so the orchestrator's paging and
+de-dup policy is exercised deterministically.
+"""
+
+from __future__ import annotations
+
+from contextlib import asynccontextmanager
+
+import pytest
+
+from app.proprietary.platforms.instagram import scraper
+from app.proprietary.platforms.instagram.schemas import InstagramScrapeInput
+
+
+class _NoopHolder:
+ async def close(self) -> None:
+ return None
+
+
+@pytest.fixture
+def _stub_proxy(monkeypatch):
+ async def _open():
+ return _NoopHolder()
+
+ @asynccontextmanager
+ async def _bind(_holder):
+ yield _holder
+
+ monkeypatch.setattr(scraper, "open_proxy_holder", _open)
+ monkeypatch.setattr(scraper, "bind_proxy_holder", _bind)
+
+
+def _profile_payload(n: int) -> dict:
+ return {
+ "data": {
+ "user": {
+ "id": "9",
+ "username": "acct",
+ "edge_owner_to_timeline_media": {
+ "count": n,
+ "edges": [
+ {"node": {"id": str(i), "shortcode": f"S{i}"}}
+ for i in range(n)
+ ],
+ },
+ }
+ }
+ }
+
+
+async def test_per_target_cap_limits_media(_stub_proxy, monkeypatch):
+ async def _fetch(path, params=None):
+ return _profile_payload(50)
+
+ monkeypatch.setattr(scraper, "fetch_json", _fetch)
+ model = InstagramScrapeInput(
+ resultsType="posts",
+ directUrls=["https://www.instagram.com/acct/"],
+ resultsLimit=5,
+ )
+ items = [i async for i in scraper.iter_instagram(model)]
+ assert len(items) == 5
+
+
+async def test_cross_target_dedup_by_id(_stub_proxy, monkeypatch):
+ async def _fetch(path, params=None):
+ return _profile_payload(3) # both targets return ids 0,1,2
+
+ monkeypatch.setattr(scraper, "fetch_json", _fetch)
+ model = InstagramScrapeInput(
+ resultsType="posts",
+ directUrls=[
+ "https://www.instagram.com/one/",
+ "https://www.instagram.com/two/",
+ ],
+ resultsLimit=10,
+ )
+ items = [i async for i in scraper.iter_instagram(model)]
+ ids = sorted(i["id"] for i in items)
+ assert ids == ["0", "1", "2"]
+
+
+async def test_scrape_instagram_honors_limit(_stub_proxy, monkeypatch):
+ async def _fetch(path, params=None):
+ return _profile_payload(50)
+
+ monkeypatch.setattr(scraper, "fetch_json", _fetch)
+ model = InstagramScrapeInput(
+ resultsType="posts",
+ directUrls=["https://www.instagram.com/acct/"],
+ resultsLimit=100,
+ )
+ items = await scraper.scrape_instagram(model, limit=7)
+ assert len(items) == 7
diff --git a/surfsense_backend/tests/unit/platforms/instagram/test_discovery.py b/surfsense_backend/tests/unit/platforms/instagram/test_discovery.py
new file mode 100644
index 000000000..c95a80315
--- /dev/null
+++ b/surfsense_backend/tests/unit/platforms/instagram/test_discovery.py
@@ -0,0 +1,78 @@
+"""Offline tests for Google-backed Instagram discovery.
+
+Discovery is profile-only (hashtag/place feeds are login-walled). A valid handle
+resolves directly; any other query falls back to the ``google_search`` platform
+(``site:instagram.com``), classifying organic results with ``resolve_url`` and
+keeping only profile hits. These tests inject a fake ``scrape_serps`` so there is
+no network: they pin the classification, de-dup, and ``limit`` cap.
+"""
+
+from __future__ import annotations
+
+from app.proprietary.platforms.instagram import scraper
+
+
+def _fake_serps(*organic_urls: str):
+ async def _scrape_serps(input_model, *, limit=None):
+ assert input_model.site == "instagram.com"
+ return [{"organicResults": [{"url": u} for u in organic_urls]}]
+
+ return _scrape_serps
+
+
+async def test_google_discovery_keeps_only_profiles(monkeypatch):
+ # A non-handle query goes to Google; only profile URLs survive (hashtag /
+ # post / non-instagram results are dropped since discovery is profile-only).
+ monkeypatch.setattr(
+ scraper,
+ "scrape_serps",
+ _fake_serps(
+ "https://www.instagram.com/natgeo/",
+ "https://www.instagram.com/explore/tags/travel/",
+ "https://www.instagram.com/p/ABC123/",
+ "https://example.com/not-instagram",
+ ),
+ )
+ targets = await scraper._discover(
+ "nat geo photos", search_type="profile", limit=10
+ )
+ assert [(t.kind, t.value) for t in targets] == [("profile", "natgeo")]
+
+
+async def test_google_discovery_dedupes(monkeypatch):
+ monkeypatch.setattr(
+ scraper,
+ "scrape_serps",
+ _fake_serps(
+ "https://www.instagram.com/natgeo/",
+ "https://www.instagram.com/natgeo/",
+ ),
+ )
+ targets = await scraper._discover(
+ "nat geo photos", search_type="profile", limit=10
+ )
+ assert len(targets) == 1
+
+
+async def test_google_discovery_respects_limit(monkeypatch):
+ monkeypatch.setattr(
+ scraper,
+ "scrape_serps",
+ _fake_serps(
+ "https://www.instagram.com/a_a/",
+ "https://www.instagram.com/b_b/",
+ "https://www.instagram.com/c_c/",
+ ),
+ )
+ targets = await scraper._discover("some brand name", search_type="profile", limit=2)
+ assert [t.value for t in targets] == ["a_a", "b_b"]
+
+
+async def test_discover_profile_handle_fast_path_skips_google(monkeypatch):
+ # A valid handle resolves directly without touching Google.
+ async def _boom(input_model, *, limit=None):
+ raise AssertionError("Google should not be called for a valid handle")
+
+ monkeypatch.setattr(scraper, "scrape_serps", _boom)
+ targets = await scraper._discover("messi", search_type="user", limit=10)
+ assert [(t.kind, t.value) for t in targets] == [("profile", "messi")]
diff --git a/surfsense_backend/tests/unit/platforms/instagram/test_fetch_resilience.py b/surfsense_backend/tests/unit/platforms/instagram/test_fetch_resilience.py
new file mode 100644
index 000000000..859c70443
--- /dev/null
+++ b/surfsense_backend/tests/unit/platforms/instagram/test_fetch_resilience.py
@@ -0,0 +1,420 @@
+"""Offline resilience tests for the Instagram fetch seam and fan-out worker pool.
+
+No network. Fake sessions drive the ``csrftoken`` warm-up + rotate-on-block +
+backoff paths deterministically (in live runs the first IP warms and returns
+200s, so these branches rarely fire). Mirrors the reddit sibling's
+``test_fetch_resilience.py`` shape, adapted to Instagram's cookie warm-up.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+from collections.abc import AsyncIterator
+from contextlib import asynccontextmanager
+
+from app.proprietary.platforms.instagram import fetch, scraper
+from app.proprietary.platforms.instagram.fetch import (
+ InstagramAccessBlockedError,
+ _current_session,
+ fetch_json,
+)
+
+_PAYLOAD = {"data": {"user": {"username": "natgeo"}}}
+
+
+class _FakePage:
+ def __init__(
+ self, status: int, *, cookies: dict | None = None, payload=None, url=None
+ ):
+ self.status = status
+ self.cookies = cookies or {}
+ self.url = url
+ self._payload = payload if payload is not None else _PAYLOAD
+
+ def json(self):
+ return self._payload
+
+ @property
+ def body(self) -> str:
+ return json.dumps(self._payload)
+
+
+class _FakeSession:
+ """One 'IP': the warm-up GET mints csrftoken per flag; endpoint GETs return ``status``."""
+
+ def __init__(
+ self,
+ status: int = 200,
+ *,
+ csrftoken: bool = True,
+ payload=None,
+ login_wall: bool = False,
+ ) -> None:
+ self.status = status
+ self.csrftoken = csrftoken
+ self.payload = payload
+ self.login_wall = login_wall
+ self.json_calls = 0
+ self.warm_calls = 0
+
+ async def get(self, url, headers=None, cookies=None):
+ if url.rstrip("/") == "https://www.instagram.com":
+ self.warm_calls += 1
+ ck = {"csrftoken": "x", "mid": "y"} if self.csrftoken else {}
+ return _FakePage(200, cookies=ck)
+ self.json_calls += 1
+ # A soft login wall: 200, but the final URL is the login page.
+ final = "https://www.instagram.com/accounts/login/" if self.login_wall else url
+ return _FakePage(self.status, payload=self.payload, url=final)
+
+
+class _FakeHolder:
+ """Holder whose ``rotate()`` advances to the next fake session (a new IP)."""
+
+ def __init__(self, sessions: list[_FakeSession]) -> None:
+ self._sessions = sessions
+ self.session = sessions[0]
+ self.rotations = 0
+ self.warmed = False
+
+ async def rotate(self):
+ self.rotations += 1
+ self.session = self._sessions[min(self.rotations, len(self._sessions) - 1)]
+ self.warmed = False # cookies bind to the IP: re-warm on the fresh one
+ return self.session
+
+ async def pace(self) -> None:
+ return None
+
+ async def close(self) -> None:
+ return None
+
+
+def _no_sleep(monkeypatch) -> None:
+ async def _noop(_seconds):
+ return None
+
+ monkeypatch.setattr(fetch.asyncio, "sleep", _noop)
+
+
+async def test_warms_then_returns_json():
+ holder = _FakeHolder([_FakeSession(200, csrftoken=True)])
+ token = _current_session.set(holder)
+ try:
+ result = await fetch_json("api/v1/users/web_profile_info/", {"username": "natgeo"})
+ finally:
+ _current_session.reset(token)
+ assert result == _PAYLOAD
+ assert holder.rotations == 0
+ assert holder.session.warm_calls == 1 # warmed exactly once
+
+
+async def test_rotates_when_warm_fails_then_succeeds():
+ holder = _FakeHolder(
+ [_FakeSession(200, csrftoken=False), _FakeSession(200, csrftoken=True)]
+ )
+ token = _current_session.set(holder)
+ try:
+ result = await fetch_json("api/v1/users/web_profile_info/")
+ finally:
+ _current_session.reset(token)
+ assert result == _PAYLOAD
+ assert holder.rotations == 1
+
+
+async def test_raises_when_no_ip_can_warm():
+ holder = _FakeHolder(
+ [_FakeSession(200, csrftoken=False) for _ in range(fetch._MAX_ROTATIONS + 1)]
+ )
+ token = _current_session.set(holder)
+ try:
+ raised = False
+ try:
+ await fetch_json("api/v1/users/web_profile_info/")
+ except InstagramAccessBlockedError:
+ raised = True
+ finally:
+ _current_session.reset(token)
+ assert raised
+ assert holder.rotations == fetch._MAX_ROTATIONS
+
+
+async def test_rotates_and_rewarms_on_403():
+ holder = _FakeHolder([_FakeSession(403), _FakeSession(200)])
+ token = _current_session.set(holder)
+ try:
+ result = await fetch_json("api/v1/users/web_profile_info/")
+ finally:
+ _current_session.reset(token)
+ assert result == _PAYLOAD
+ assert holder.rotations == 1
+ assert holder.session.warm_calls == 1 # re-warmed on the fresh IP
+
+
+async def test_rotates_on_401_login_wall():
+ holder = _FakeHolder([_FakeSession(401), _FakeSession(200)])
+ token = _current_session.set(holder)
+ try:
+ result = await fetch_json("api/v1/users/web_profile_info/")
+ finally:
+ _current_session.reset(token)
+ assert result == _PAYLOAD
+ assert holder.rotations == 1
+
+
+async def test_login_redirect_fails_fast_without_rotating():
+ # A 302 -> /accounts/login/ (served 200) is an endpoint-level wall: rotating
+ # never clears it, so we raise immediately instead of burning IP rotations.
+ # A second healthy session is present to prove we do NOT fall through to it.
+ holder = _FakeHolder([_FakeSession(200, login_wall=True), _FakeSession(200)])
+ token = _current_session.set(holder)
+ try:
+ raised = False
+ try:
+ await fetch_json("api/v1/users/web_profile_info/", {"username": "natgeo"})
+ except InstagramAccessBlockedError:
+ raised = True
+ finally:
+ _current_session.reset(token)
+ assert raised
+ assert holder.rotations == 0 # fail fast: no rotation burned
+
+
+async def test_404_returns_none_without_rotating():
+ holder = _FakeHolder([_FakeSession(404), _FakeSession(200)])
+ token = _current_session.set(holder)
+ try:
+ result = await fetch_json("api/v1/users/web_profile_info/")
+ finally:
+ _current_session.reset(token)
+ assert result is None
+ assert holder.rotations == 0
+
+
+async def test_429_backs_off_without_rotating(monkeypatch):
+ _no_sleep(monkeypatch)
+ session = _FakeSession(429)
+
+ async def _get(url, headers=None, cookies=None):
+ if url.rstrip("/") == "https://www.instagram.com":
+ session.warm_calls += 1
+ return _FakePage(200, cookies={"csrftoken": "x"})
+ session.json_calls += 1
+ return _FakePage(429 if session.json_calls == 1 else 200)
+
+ session.get = _get # type: ignore[method-assign]
+ holder = _FakeHolder([session])
+ token = _current_session.set(holder)
+ try:
+ result = await fetch_json("api/v1/users/web_profile_info/")
+ finally:
+ _current_session.reset(token)
+ assert result == _PAYLOAD
+ assert holder.rotations == 0
+
+
+async def test_persistent_403_raises_blocked(monkeypatch):
+ _no_sleep(monkeypatch)
+ holder = _FakeHolder([_FakeSession(403) for _ in range(fetch._MAX_ROTATIONS + 1)])
+ token = _current_session.set(holder)
+ try:
+ raised = False
+ try:
+ await fetch_json("api/v1/users/web_profile_info/")
+ except InstagramAccessBlockedError:
+ raised = True
+ finally:
+ _current_session.reset(token)
+ assert raised
+ assert holder.rotations == fetch._MAX_ROTATIONS
+
+
+class _TrackingHolder:
+ """Fake fan-out session holder that records whether it was closed."""
+
+ def __init__(self) -> None:
+ self.closed = False
+
+ async def close(self) -> None:
+ self.closed = True
+
+
+async def test_fan_out_closes_all_sessions_on_early_stop(monkeypatch):
+ holders: list[_TrackingHolder] = []
+
+ async def _fake_open():
+ h = _TrackingHolder()
+ holders.append(h)
+ return h
+
+ @asynccontextmanager
+ async def _fake_bind(_holder):
+ yield _holder
+
+ monkeypatch.setattr(scraper, "open_proxy_holder", _fake_open)
+ monkeypatch.setattr(scraper, "bind_proxy_holder", _fake_bind)
+
+ async def _job(n: int) -> AsyncIterator[dict]:
+ for i in range(5):
+ yield {"job": n, "i": i}
+
+ jobs = [_job(n) for n in range(20)]
+ gen = scraper.fan_out(jobs, concurrency=4)
+ collected = []
+ async for item in gen:
+ collected.append(item)
+ if len(collected) >= 3:
+ break
+ await gen.aclose()
+
+ assert len(collected) >= 3
+ assert holders, "workers should have opened sessions"
+ assert all(h.closed for h in holders), "a worker leaked its proxy session"
+
+
+async def test_fan_out_empty_jobs_is_noop():
+ out = [x async for x in scraper.fan_out([])]
+ assert out == []
+
+
+async def _install_fake_holders(monkeypatch) -> None:
+ async def _fake_open():
+ return _TrackingHolder()
+
+ @asynccontextmanager
+ async def _fake_bind(_holder):
+ yield _holder
+
+ monkeypatch.setattr(scraper, "open_proxy_holder", _fake_open)
+ monkeypatch.setattr(scraper, "bind_proxy_holder", _fake_bind)
+
+
+async def _blocked_job() -> AsyncIterator[dict]:
+ raise InstagramAccessBlockedError("login wall")
+ yield {} # unreachable; makes this an async generator
+
+
+async def test_fan_out_all_blocked_raises_without_deadlock(monkeypatch):
+ # Regression: a blocked worker used to strand its exception on a dead task
+ # and deadlock the consumer on results.get(). When EVERY target is blocked
+ # (zero items), it must surface InstagramAccessBlockedError, not hang.
+ await _install_fake_holders(monkeypatch)
+
+ raised = False
+ try:
+ async with asyncio.timeout(5): # fail fast if the deadlock regresses
+ async for _ in scraper.fan_out([_blocked_job()], concurrency=1):
+ pass
+ except InstagramAccessBlockedError:
+ raised = True
+ assert raised, "fully-blocked run must surface the 403, not deadlock"
+
+
+async def test_fan_out_partial_results_survive_one_blocked_target(monkeypatch):
+ # Q2: one blocked target must NOT abort the batch — the good target's items
+ # come through and no exception is raised (aggregation, not a transaction).
+ await _install_fake_holders(monkeypatch)
+
+ async def _good_job() -> AsyncIterator[dict]:
+ for i in range(3):
+ yield {"id": f"good:{i}"}
+
+ async with asyncio.timeout(5):
+ items = [
+ item
+ async for item in scraper.fan_out(
+ [_blocked_job(), _good_job()], concurrency=2
+ )
+ ]
+ assert [it["id"] for it in items] == ["good:0", "good:1", "good:2"]
+
+
+def _profile_payload(username: str, n: int) -> dict:
+ # IDs namespaced per target so cross-target de-dup doesn't collapse them.
+ return {
+ "data": {
+ "user": {
+ "id": f"u_{username}",
+ "username": username,
+ "edge_owner_to_timeline_media": {
+ "count": n,
+ "edges": [{"node": {"id": f"{username}:{i}"}} for i in range(n)],
+ },
+ }
+ }
+ }
+
+
+async def test_scrape_instagram_closes_sessions_when_limit_stops_inflight_workers(
+ monkeypatch,
+):
+ """Hitting ``limit`` must tear down the whole fan-out chain deterministically.
+
+ Regression: closing the outer ``iter_instagram`` generator does NOT
+ synchronously close the inner ``fan_out`` it loops over — CPython defers that
+ to async-gen GC — so without an explicit ``aclosing`` the collector's early
+ ``break`` leaked every warm proxy session that was still mid-fetch. The
+ ``fan_out``-direct test misses this because instant jobs self-drain before
+ cancellation ever runs; here each fetch yields to the loop so workers are
+ genuinely in-flight when the limit trips.
+ """
+ holders: list[_TrackingHolder] = []
+
+ async def _fake_open():
+ h = _TrackingHolder()
+ holders.append(h)
+ return h
+
+ @asynccontextmanager
+ async def _fake_bind(_holder):
+ yield _holder
+
+ async def _fetch(path, params=None):
+ await asyncio.sleep(0) # yield control: keep sibling workers in-flight
+ username = (params or {}).get("username", "acct")
+ return _profile_payload(username, 5)
+
+ monkeypatch.setattr(scraper, "open_proxy_holder", _fake_open)
+ monkeypatch.setattr(scraper, "bind_proxy_holder", _fake_bind)
+ monkeypatch.setattr(scraper, "fetch_json", _fetch)
+
+ model = scraper.InstagramScrapeInput(
+ resultsType="posts",
+ directUrls=[f"https://www.instagram.com/acct{i}/" for i in range(50)],
+ resultsLimit=5,
+ )
+ items = await scraper.scrape_instagram(model, limit=3)
+
+ assert len(items) == 3
+ assert holders, "workers should have opened sessions"
+ assert all(h.closed for h in holders), "early stop leaked a proxy session"
+
+
+async def test_discover_profile_is_anonymous_handle_lookup():
+ # keyword search (topsearch) is login-walled, so a profile/user query resolves
+ # as a DIRECT handle lookup against the anonymous profile endpoint — no network
+ # here, just the URL resolution, so no session/monkeypatch needed.
+ targets = await scraper._discover("messi", search_type="user", limit=10)
+ assert [(t.kind, t.value) for t in targets] == [("profile", "messi")]
+
+
+async def test_discover_nonhandle_routes_through_google(monkeypatch):
+ # A non-handle profile query goes through Google (site:instagram.com) and
+ # classifies the organic results into profile targets (the only kind now).
+ async def _fake_scrape_serps(input_model, *, limit=None):
+ assert input_model.site == "instagram.com"
+ return [
+ {
+ "organicResults": [
+ {"url": "https://www.instagram.com/natgeo/"},
+ {"url": "https://www.instagram.com/p/Cabc/"}, # wrong kind
+ ]
+ }
+ ]
+
+ monkeypatch.setattr(scraper, "scrape_serps", _fake_scrape_serps)
+ targets = await scraper._discover(
+ "national geographic", search_type="profile", limit=10
+ )
+ assert [(t.kind, t.value) for t in targets] == [("profile", "natgeo")]
diff --git a/surfsense_backend/tests/unit/platforms/instagram/test_parsers.py b/surfsense_backend/tests/unit/platforms/instagram/test_parsers.py
new file mode 100644
index 000000000..8cba74b3a
--- /dev/null
+++ b/surfsense_backend/tests/unit/platforms/instagram/test_parsers.py
@@ -0,0 +1,298 @@
+"""Offline parser tests: raw web JSON -> flat item dicts.
+
+Synthetic nodes cover the GraphQL ``edge_*`` flattening the anonymous web
+payloads use. A fixture-pinned test runs only when a captured fixture is present
+(the live e2e script dumps trimmed, PII-anonymized fixtures), so the suite stays
+green offline.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+
+from app.proprietary.platforms.instagram.parsers import (
+ parse_media,
+ parse_post,
+ parse_profile,
+)
+
+_FIXTURES = Path(__file__).parent / "fixtures"
+
+
+def _edge(nodes: list[dict]) -> dict:
+ return {"edges": [{"node": n} for n in nodes]}
+
+
+def test_parse_media_flattens_edges_and_extracts_tags():
+ node = {
+ "id": "1",
+ "shortcode": "Cabc",
+ "__typename": "GraphImage",
+ "taken_at_timestamp": 1_600_000_000,
+ "edge_media_to_caption": _edge([{"text": "love #nasa shot @buzz"}]),
+ "edge_media_to_comment": {"count": 7},
+ "edge_liked_by": {"count": 42},
+ "owner": {"username": "natgeo", "id": "9"},
+ }
+ item = parse_media(node)
+ assert item["shortCode"] == "Cabc"
+ assert item["type"] == "Image"
+ assert item["hashtags"] == ["nasa"]
+ assert item["mentions"] == ["buzz"]
+ assert item["commentsCount"] == 7
+ assert item["likesCount"] == 42
+ assert item["ownerUsername"] == "natgeo"
+ assert item["url"] == "https://www.instagram.com/p/Cabc/"
+
+
+def test_parse_media_passes_through_hidden_like_count():
+ # Instagram reports -1 when the creator hid likes; never coerce it away.
+ item = parse_media({"id": "1", "edge_liked_by": {"count": -1}})
+ assert item["likesCount"] == -1
+
+
+def test_parse_media_marks_video_type():
+ item = parse_media({"id": "1", "is_video": True, "video_view_count": 99})
+ assert item["type"] == "Video"
+ assert item["videoViewCount"] == 99
+
+
+def test_parse_media_extracts_sidecar_tags_location_pinned():
+ # The anonymous profile feed node carries far more than the core fields:
+ # sidecar children, tagged users, coauthors, location, product type and pin
+ # state — all populated here from the real GraphQL key shapes.
+ node = {
+ "id": "1",
+ "shortcode": "Cabc",
+ "__typename": "GraphSidecar",
+ "taken_at_timestamp": 1_704_164_645,
+ "edge_media_to_caption": _edge([{"text": "x #tag @me"}]),
+ "pinned_for_users": [{"id": "9"}],
+ "product_type": "feed",
+ "location": {"id": "55", "name": "Paris"},
+ "coauthor_producers": [{"username": "co", "id": "7"}],
+ "edge_media_to_tagged_user": _edge(
+ [{"user": {"username": "tg", "id": "3"}, "x": 0.1, "y": 0.2}]
+ ),
+ "edge_sidecar_to_children": _edge(
+ [
+ {
+ "id": "c1",
+ "shortcode": "s1",
+ "display_url": "https://cdn/1.jpg",
+ "dimensions": {"height": 10, "width": 20},
+ },
+ {
+ "id": "c2",
+ "shortcode": "s2",
+ "is_video": True,
+ "video_url": "https://cdn/2.mp4",
+ "display_url": "https://cdn/2.jpg",
+ },
+ ]
+ ),
+ }
+ item = parse_media(node)
+ assert item["type"] == "Sidecar"
+ assert item["isPinned"] is True
+ assert item["productType"] == "feed"
+ assert item["locationName"] == "Paris"
+ assert item["locationId"] == "55"
+ assert item["taggedUsers"][0]["username"] == "tg"
+ assert item["coauthorProducers"][0]["username"] == "co"
+ assert item["images"] == ["https://cdn/1.jpg", "https://cdn/2.jpg"]
+ assert len(item["childPosts"]) == 2
+ assert item["childPosts"][1]["type"] == "Video"
+ assert item["childPosts"][1]["videoUrl"] == "https://cdn/2.mp4"
+
+
+def test_parse_media_unpinned_is_false():
+ assert parse_media({"id": "1"})["isPinned"] is False
+
+
+def test_parse_profile_flattens_counts_and_latest_posts():
+ user = {
+ "id": "9",
+ "username": "natgeo",
+ "full_name": "Nat Geo",
+ "edge_followed_by": {"count": 1000},
+ "edge_follow": {"count": 50},
+ "edge_owner_to_timeline_media": {
+ "count": 2,
+ "edges": [{"node": {"id": "p1", "shortcode": "A"}}],
+ },
+ "edge_related_profiles": _edge(
+ [{"username": "similar1", "id": "11"}, {"username": "similar2", "id": "12"}]
+ ),
+ }
+ item = parse_profile(user)
+ assert item["detailKind"] == "profile"
+ assert item["username"] == "natgeo"
+ assert item["followersCount"] == 1000
+ assert item["followsCount"] == 50
+ assert item["postsCount"] == 2
+ assert len(item["latestPosts"]) == 1
+ assert [p["username"] for p in item["relatedProfiles"]] == ["similar1", "similar2"]
+
+
+_POST_URL = "https://www.instagram.com/p/Cabc/"
+
+
+def test_parse_post_prefers_relay_json():
+ # Anonymous /p/ pages inline the mobile-v1 PolarisMedia object in an
+ # application/json script. It's the full-fidelity source (carousel children,
+ # tagged users, coauthors, location, precise timestamp), preferred over og.
+ media = {
+ "pk": "3938367641542741384",
+ "id": "POLARIS_3938367641542741384",
+ "code": "Cabc",
+ "taken_at": 1_704_164_645,
+ "media_type": 8,
+ "product_type": "carousel_container",
+ "like_count": 4200,
+ "comment_count": 37,
+ "accessibility_caption": "alt text",
+ "caption": {"text": "sunset over #bali with @friend @friend"},
+ "user": {"username": "natgeo", "full_name": "Nat Geo", "id": "9"},
+ "image_versions2": {"candidates": [{"url": "https://cdn/i.jpg"}]},
+ "carousel_media": [
+ {
+ "id": "m1",
+ "code": "c1",
+ "media_type": 1,
+ "image_versions2": {"candidates": [{"url": "https://cdn/c1.jpg"}]},
+ "original_height": 1080,
+ "original_width": 1080,
+ },
+ {
+ "id": "m2",
+ "code": "c2",
+ "media_type": 2,
+ "video_versions": [{"url": "https://cdn/c2.mp4"}],
+ "image_versions2": {"candidates": [{"url": "https://cdn/c2.jpg"}]},
+ },
+ ],
+ "usertags": {"in": [{"position": [0.5, 0.5], "user": {"username": "tagged1", "id": "77"}}]},
+ "coauthor_producers": [{"username": "coauthor1", "id": "88", "is_verified": True}],
+ "location": {"id": "123", "name": "Bali"},
+ }
+ html = (
+ '"
+ )
+ item = parse_post(html, url=_POST_URL, shortcode="Cabc")
+ assert item is not None
+ assert item["id"] == "3938367641542741384" # POLARIS_ prefix stripped
+ assert item["type"] == "Sidecar" # media_type 8
+ assert item["shortCode"] == "Cabc"
+ assert item["url"] == _POST_URL
+ assert item["caption"] == "sunset over #bali with @friend @friend"
+ assert item["hashtags"] == ["bali"]
+ assert item["mentions"] == ["friend"] # deduped
+ assert item["likesCount"] == 4200
+ assert item["commentsCount"] == 37
+ assert item["displayUrl"] == "https://cdn/i.jpg"
+ assert item["timestamp"].startswith("2024-01-02T") # real epoch -> ISO w/ time
+ assert item["ownerUsername"] == "natgeo"
+ assert item["ownerFullName"] == "Nat Geo"
+ assert item["images"] == ["https://cdn/c1.jpg", "https://cdn/c2.jpg"]
+ assert len(item["childPosts"]) == 2
+ assert item["childPosts"][1]["type"] == "Video"
+ assert item["childPosts"][1]["videoUrl"] == "https://cdn/c2.mp4"
+ assert item["taggedUsers"][0]["username"] == "tagged1"
+ assert item["coauthorProducers"][0]["username"] == "coauthor1"
+ assert item["locationName"] == "Bali"
+ assert item["locationId"] == "123"
+ assert item["productType"] == "carousel_container"
+
+
+def test_parse_post_falls_back_to_og_meta():
+ # Anonymous /p/ pages carry no ld+json; everything is lifted from the og
+ # tags. og:description gives counts + username + date; og:title gives the
+ # clean caption + full name. Entities in the caption are deduped.
+ html = """
+
+
+
+
+
+
+
+ """
+ item = parse_post(html, url=_POST_URL, shortcode="Cabc")
+ assert item is not None
+ assert item["id"] == "3938367641542741384" # numeric pk from al:ios:url meta
+ assert item["likesCount"] == 1234
+ assert item["commentsCount"] == 56
+ assert item["displayUrl"] == "https://cdn/i.jpg"
+ assert item["type"] == "Video"
+ assert item["ownerUsername"] == "natgeo"
+ assert item["ownerFullName"] == "Nat Geo"
+ assert item["timestamp"] == "2024-01-02" # og carries date only, no time
+ assert item["caption"] == "a caption #wow #wow @buzz." # @ -> @, unescaped
+ assert item["hashtags"] == ["wow"] # deduped, no @-as-#064 pollution
+ assert item["mentions"] == ["buzz"] # trailing sentence dot stripped
+
+
+def test_parse_post_og_degrades_without_crashing():
+ # A shape we don't recognise (hidden likes / a non-English locale that
+ # slipped the en-US header) must yield a partial item with None fields,
+ # never an exception or a caption polluted with the counts/date prefix.
+ html = """
+
+
+
+
+
+
+ """
+ item = parse_post(html, url=_POST_URL, shortcode="Cabc")
+ assert item is not None # og:image present -> still emits
+ assert item["displayUrl"] == "https://cdn/i.jpg"
+ assert item["likesCount"] is None
+ assert item["commentsCount"] is None
+ assert item["ownerUsername"] is None
+ assert item["timestamp"] is None
+ assert item["caption"] is None # unrecognised prefix -> no pollution
+
+
+def test_parse_post_returns_none_without_surfaces():
+ # A login interstitial / empty doc carries neither ld+json nor og -> None,
+ # never a silent empty-success item.
+ assert parse_post("login", url=_POST_URL) is None
+ assert parse_post(None, url=_POST_URL) is None
+ assert parse_post("", url=_POST_URL) is None
+
+
+@pytest.mark.skipif(
+ not (_FIXTURES / "profile.json").exists(),
+ reason="captured fixture absent (run scripts/e2e_instagram_scraper.py to dump)",
+)
+def test_fixture_profile_maps():
+ raw = json.loads((_FIXTURES / "profile.json").read_text())
+ user = raw.get("data", {}).get("user", raw)
+ item = parse_profile(user)
+ assert item["detailKind"] == "profile"
+ assert item["username"]
+
+
+@pytest.mark.skipif(
+ not (_FIXTURES / "post.json").exists(),
+ reason="captured fixture absent (run the single-post probe to dump /p/ HTML)",
+)
+def test_fixture_post_maps():
+ raw = json.loads((_FIXTURES / "post.json").read_text())
+ item = parse_post(raw["html"], url=raw["url"], shortcode=raw.get("shortcode"))
+ assert item is not None, "captured /p/ HTML produced no media item"
+ assert item["url"] == raw["url"]
+ # The relay blob (not og-meta) should drive extraction: numeric id + a
+ # precise timestamp with a time component (og-only would be date-only).
+ assert item["id"] and item["id"].isdigit()
+ assert item["ownerUsername"]
+ assert item["timestamp"] and "T" in item["timestamp"]
diff --git a/surfsense_backend/tests/unit/platforms/instagram/test_skeleton.py b/surfsense_backend/tests/unit/platforms/instagram/test_skeleton.py
new file mode 100644
index 000000000..f609883eb
--- /dev/null
+++ b/surfsense_backend/tests/unit/platforms/instagram/test_skeleton.py
@@ -0,0 +1,102 @@
+"""Offline skeleton tests: input surface parity + URL classification.
+
+No network. Locks the two invariants the reference-compatible surface promises —
+no auth fields ever, and additive ``extra="allow"`` parity — plus the
+``url_resolver`` classification/normalization table (``_u/`` and profilecard
+stripping, story→profile, numeric post-ID flagging). Hashtag/place URLs are
+login-walled and deliberately resolve to ``None``.
+"""
+
+from __future__ import annotations
+
+from app.proprietary.platforms.instagram.schemas import (
+ InstagramMediaItem,
+ InstagramScrapeInput,
+)
+from app.proprietary.platforms.instagram.url_resolver import resolve_url
+
+
+def test_input_has_no_auth_fields():
+ # Anonymous-only: the input surface must never expose a login/credential seam.
+ forbidden = {
+ "sessionid",
+ "username",
+ "password",
+ "cookies",
+ "authorization",
+ "proxyConfiguration",
+ "loginCredentials",
+ }
+ assert forbidden.isdisjoint(InstagramScrapeInput.model_fields)
+
+
+def test_input_defaults():
+ model = InstagramScrapeInput()
+ assert model.resultsType == "posts"
+ assert model.searchType == "profile"
+ assert model.directUrls == []
+ assert model.addParentData is False
+
+
+def test_input_allows_extra_inert_fields():
+ # A reference field the acquisition layer doesn't source is accepted, not rejected.
+ model = InstagramScrapeInput(enhanceUserSearchWithFacebookPage="x")
+ assert model.model_dump().get("enhanceUserSearchWithFacebookPage") == "x"
+
+
+def test_media_item_to_output_keeps_none_keys():
+ out = InstagramMediaItem(id="123", shortCode="abc").to_output()
+ assert out["id"] == "123"
+ assert out["shortCode"] == "abc"
+ # Unsourced fields stay present as None / [] for additive parity.
+ assert out["likesCount"] is None
+ assert out["requestErrorMessages"] == []
+
+
+def test_resolve_profile():
+ r = resolve_url("https://www.instagram.com/natgeo/")
+ assert r.kind == "profile"
+ assert r.value == "natgeo"
+
+
+def test_resolve_bare_profile_id():
+ r = resolve_url("natgeo")
+ assert r.kind == "profile" and r.value == "natgeo"
+
+
+def test_resolve_post_and_reel():
+ r = resolve_url("https://www.instagram.com/p/Cabc123/")
+ assert r.kind == "post" and r.value == "Cabc123" and r.numeric_post_id is False
+ r = resolve_url("https://www.instagram.com/reel/Cxyz/")
+ assert r.kind == "reel" and r.value == "Cxyz"
+
+
+def test_resolve_hashtag_and_place_unsupported():
+ # Login-walled surfaces: they must resolve to None so the orchestrator skips
+ # them rather than building a target that can only return a login wall.
+ assert resolve_url("https://www.instagram.com/explore/tags/crossfit/") is None
+ assert (
+ resolve_url("https://www.instagram.com/explore/locations/7538318/copenhagen/")
+ is None
+ )
+
+
+def test_resolve_strips_u_and_profilecard():
+ stripped_u = resolve_url("https://www.instagram.com/_u/natgeo/")
+ assert stripped_u.kind == "profile" and stripped_u.value == "natgeo"
+ card = resolve_url("https://www.instagram.com/natgeo/profilecard/")
+ assert card.kind == "profile" and card.value == "natgeo"
+
+
+def test_resolve_story_reduces_to_profile():
+ r = resolve_url("https://www.instagram.com/stories/natgeo/12345/")
+ assert r.kind == "profile" and r.value == "natgeo"
+
+
+def test_resolve_numeric_post_id_flagged():
+ r = resolve_url("https://www.instagram.com/p/12345/")
+ assert r.kind == "post" and r.numeric_post_id is True
+
+
+def test_resolve_rejects_non_instagram_host():
+ assert resolve_url("https://example.com/natgeo/") is None
diff --git a/surfsense_mcp/README.md b/surfsense_mcp/README.md
index 5bac5f8b2..95ac293d9 100644
--- a/surfsense_mcp/README.md
+++ b/surfsense_mcp/README.md
@@ -21,7 +21,9 @@ Connect it two ways:
**Scrapers (all platforms)**
- `surfsense_web_crawl`, `surfsense_google_search`, `surfsense_reddit_scrape`,
`surfsense_youtube_scrape`, `surfsense_youtube_comments`,
- `surfsense_tiktok_scrape`,
+ `surfsense_instagram_scrape`, `surfsense_instagram_details`,
+ `surfsense_tiktok_scrape`, `surfsense_tiktok_comments`,
+ `surfsense_tiktok_user_search`, `surfsense_tiktok_trending`,
`surfsense_google_maps_scrape`, `surfsense_google_maps_reviews`
- `surfsense_list_scraper_runs`, `surfsense_get_scraper_run` — retrieve past
results in full (useful when a large result was truncated inline)
diff --git a/surfsense_mcp/mcp_server/features/scrapers/__init__.py b/surfsense_mcp/mcp_server/features/scrapers/__init__.py
index c0e7dea07..eb5f72ba5 100644
--- a/surfsense_mcp/mcp_server/features/scrapers/__init__.py
+++ b/surfsense_mcp/mcp_server/features/scrapers/__init__.py
@@ -13,9 +13,26 @@ from mcp.server.fastmcp import FastMCP
from ...core.client import SurfSenseClient
from ...core.workspace_context import WorkspaceContext
from . import run_history
-from .platforms import google_maps, google_search, reddit, tiktok, web, youtube
+from .platforms import (
+ google_maps,
+ google_search,
+ instagram,
+ reddit,
+ tiktok,
+ web,
+ youtube,
+)
-_REGISTRARS = (web, google_search, reddit, youtube, tiktok, google_maps, run_history)
+_REGISTRARS = (
+ web,
+ google_search,
+ reddit,
+ youtube,
+ instagram,
+ tiktok,
+ google_maps,
+ run_history,
+)
def register(
diff --git a/surfsense_mcp/mcp_server/features/scrapers/platforms/instagram.py b/surfsense_mcp/mcp_server/features/scrapers/platforms/instagram.py
new file mode 100644
index 000000000..28ad34e66
--- /dev/null
+++ b/surfsense_mcp/mcp_server/features/scrapers/platforms/instagram.py
@@ -0,0 +1,138 @@
+"""Instagram scraper tools: posts/reels and profile details (anonymous-only)."""
+
+from __future__ import annotations
+
+from typing import Annotated, Literal
+
+from mcp.server.fastmcp import FastMCP
+from pydantic import Field
+
+from ....core.client import SurfSenseClient
+from ....core.rendering import ResponseFormatParam
+from ....core.workspace_context import WorkspaceContext, WorkspaceParam
+from ..annotations import SCRAPE
+from ..capability import run_scraper
+
+ResultType = Literal["posts", "reels"]
+SearchType = Literal["profile", "user"]
+
+
+def register(
+ mcp: FastMCP, client: SurfSenseClient, context: WorkspaceContext
+) -> None:
+ """Register the Instagram scrape and details tools (anonymous-only)."""
+
+ @mcp.tool(
+ name="surfsense_instagram_scrape",
+ title="Scrape Instagram posts or reels",
+ annotations=SCRAPE,
+ structured_output=False,
+ )
+ async def instagram_scrape(
+ urls: Annotated[
+ list[str] | None,
+ Field(
+ description="Instagram URLs: a profile, post (/p/), or reel "
+ "(/reel/). Hashtag/place URLs are unsupported (login-walled). "
+ "Provide urls OR search_queries."
+ ),
+ ] = None,
+ search_queries: Annotated[
+ list[str] | None,
+ Field(
+ description="Terms to discover public profiles for (resolved via "
+ "Google). Provide search_queries OR urls."
+ ),
+ ] = None,
+ search_type: Annotated[
+ SearchType,
+ Field(description="Discovery kind (profile-only)."),
+ ] = "profile",
+ result_type: Annotated[
+ ResultType,
+ Field(description="Which feed to return: 'posts' or 'reels'."),
+ ] = "posts",
+ max_items: Annotated[
+ int, Field(ge=1, description="Maximum items to return across sources.")
+ ] = 10,
+ workspace: WorkspaceParam = None,
+ response_format: ResponseFormatParam = "markdown",
+ ) -> str:
+ """Scrape public Instagram posts or reels from URLs or search queries.
+
+ Use this for Instagram content research: a creator's recent posts, a
+ single post/reel, or discovering public profiles by keyword. For a
+ profile's metadata use surfsense_instagram_details. Returns per-item
+ caption, likes, comments count, media URLs, and owner. Anonymous-only:
+ hashtag/place feeds and comment threads are login-walled and unavailable.
+ Example: urls=['https://www.instagram.com/natgeo/'], result_type='reels'.
+ """
+ return await run_scraper(
+ client,
+ context,
+ platform="instagram",
+ verb="scrape",
+ payload={
+ "urls": urls,
+ "search_queries": search_queries,
+ "search_type": search_type,
+ "result_type": result_type,
+ "max_items": max_items,
+ },
+ workspace=workspace,
+ response_format=response_format,
+ )
+
+ @mcp.tool(
+ name="surfsense_instagram_details",
+ title="Fetch Instagram profile details",
+ annotations=SCRAPE,
+ structured_output=False,
+ )
+ async def instagram_details(
+ urls: Annotated[
+ list[str] | None,
+ Field(
+ description="Profile URLs (or bare profile IDs). Provide urls OR "
+ "search_queries."
+ ),
+ ] = None,
+ search_queries: Annotated[
+ list[str] | None,
+ Field(
+ description="Terms to discover public profiles for. Provide "
+ "search_queries OR urls."
+ ),
+ ] = None,
+ search_type: Annotated[
+ SearchType,
+ Field(description="Discovery kind (profile-only)."),
+ ] = "profile",
+ max_items: Annotated[
+ int, Field(ge=1, description="Max detail items to return.")
+ ] = 10,
+ workspace: WorkspaceParam = None,
+ response_format: ResponseFormatParam = "markdown",
+ ) -> str:
+ """Fetch Instagram profile metadata.
+
+ Use this for entity lookups: a profile's follower/post counts and bio.
+ For a feed of posts use surfsense_instagram_scrape instead. Each item
+ carries a detailKind field (always "profile"). Anonymous-only: hashtag
+ and place details are login-walled and unavailable.
+ Example: urls=['https://www.instagram.com/natgeo/'].
+ """
+ return await run_scraper(
+ client,
+ context,
+ platform="instagram",
+ verb="details",
+ payload={
+ "urls": urls,
+ "search_queries": search_queries,
+ "search_type": search_type,
+ "max_items": max_items,
+ },
+ workspace=workspace,
+ response_format=response_format,
+ )
diff --git a/surfsense_mcp/mcp_server/selfcheck.py b/surfsense_mcp/mcp_server/selfcheck.py
index 126981e3c..bb61151bc 100644
--- a/surfsense_mcp/mcp_server/selfcheck.py
+++ b/surfsense_mcp/mcp_server/selfcheck.py
@@ -29,6 +29,8 @@ EXPECTED_TOOLS = {
"surfsense_tiktok_trending",
"surfsense_google_maps_scrape",
"surfsense_google_maps_reviews",
+ "surfsense_instagram_scrape",
+ "surfsense_instagram_details",
"surfsense_list_scraper_runs",
"surfsense_get_scraper_run",
# knowledge-base management
diff --git a/surfsense_mcp/mcp_server/server.py b/surfsense_mcp/mcp_server/server.py
index c0398ed50..f801aeb15 100644
--- a/surfsense_mcp/mcp_server/server.py
+++ b/surfsense_mcp/mcp_server/server.py
@@ -36,9 +36,10 @@ def build_server(settings: Settings) -> tuple[FastMCP, SurfSenseClient]:
"SurfSense gives you live scrapers and a personal knowledge base. "
"Prefer these tools over generic/built-in web search whenever the "
"task involves Reddit (posts, comments, finding subreddits or "
- "communities), YouTube (videos, transcripts, comments), TikTok "
- "(videos by hashtag, search, or URL), Google Maps (places, "
- "reviews), Google Search results, or reading "
+ "communities), YouTube (videos, transcripts, comments), Instagram "
+ "(posts, reels, profile details), TikTok (videos by hashtag, "
+ "search, or URL), Google Maps (places, reviews), Google Search "
+ "results, or reading "
"specific web pages. Scraper results are persisted as runs; if an "
"inline result is truncated, fetch it in full with "
"surfsense_get_scraper_run."
diff --git a/surfsense_web/app/(home)/mcp-server/page.tsx b/surfsense_web/app/(home)/mcp-server/page.tsx
index 22c867c73..ce4caf992 100644
--- a/surfsense_web/app/(home)/mcp-server/page.tsx
+++ b/surfsense_web/app/(home)/mcp-server/page.tsx
@@ -16,7 +16,7 @@ import type { FaqItem } from "@/lib/connectors-marketing/types";
const canonicalUrl = "https://www.surfsense.com/mcp-server";
const metaDescription =
- "The SurfSense MCP server gives Claude, Cursor, and any MCP client native tools for your workspace: scrape Reddit, YouTube, TikTok, Google Maps, Google Search, and the web, plus full knowledge base access. One API key.";
+ "The SurfSense MCP server gives Claude, Cursor, and any MCP client native tools for your workspace: scrape Reddit, YouTube, Instagram, TikTok, Google Maps, Google Search, and the web, plus full knowledge base access. One API key.";
export const metadata: Metadata = {
title: "SurfSense MCP Server: Scraper APIs and Knowledge Base as Agent Tools",
@@ -93,7 +93,12 @@ const TOOL_GROUPS = [
"surfsense_reddit_scrape",
"surfsense_youtube_scrape",
"surfsense_youtube_comments",
+ "surfsense_instagram_scrape",
+ "surfsense_instagram_details",
"surfsense_tiktok_scrape",
+ "surfsense_tiktok_comments",
+ "surfsense_tiktok_user_search",
+ "surfsense_tiktok_trending",
"surfsense_google_maps_scrape",
"surfsense_google_maps_reviews",
"surfsense_google_search",
@@ -128,7 +133,7 @@ const FAQ: FaqItem[] = [
{
question: "What is the SurfSense MCP server?",
answer:
- "It is a Model Context Protocol server that exposes your SurfSense workspace to MCP clients like Claude Code, Cursor, and Claude Desktop. Your agents get native tools for every scraper API (Reddit, YouTube, TikTok, Google Maps, Google Search, web crawl) and for searching, reading, and writing your knowledge base.",
+ "It is a Model Context Protocol server that exposes your SurfSense workspace to MCP clients like Claude Code, Cursor, and Claude Desktop. Your agents get native tools for every scraper API (Reddit, YouTube, Instagram, TikTok, Google Maps, Google Search, web crawl) and for searching, reading, and writing your knowledge base.",
},
{
question: "Which MCP clients does it work with?",
@@ -216,9 +221,9 @@ export default function McpServerPage() {
The SurfSense MCP server hands Claude, Cursor, or any MCP client the whole platform:
- scrape Reddit, YouTube, TikTok, Google Maps, Google Search, and the open web, and
- search, read, and write your knowledge base. One API key, typed tools, pay as you
- go.
+ scrape Reddit, YouTube, Instagram, TikTok, Google Maps, Google Search, and the open
+ web, and search, read, and write your knowledge base. One API key, typed tools, pay
+ as you go.