diff --git a/surfsense_backend/app/proprietary/platforms/instagram/__init__.py b/surfsense_backend/app/proprietary/platforms/instagram/__init__.py index e3a2a122a..2302ca9fc 100644 --- a/surfsense_backend/app/proprietary/platforms/instagram/__init__.py +++ b/surfsense_backend/app/proprietary/platforms/instagram/__init__.py @@ -2,10 +2,7 @@ from .fetch import InstagramAccessBlockedError from .schemas import ( - InstagramComment, - InstagramHashtag, InstagramMediaItem, - InstagramPlace, InstagramProfile, InstagramScrapeInput, ) @@ -13,10 +10,7 @@ from .scraper import iter_instagram, scrape_instagram __all__ = [ "InstagramAccessBlockedError", - "InstagramComment", - "InstagramHashtag", "InstagramMediaItem", - "InstagramPlace", "InstagramProfile", "InstagramScrapeInput", "iter_instagram", diff --git a/surfsense_backend/app/proprietary/platforms/instagram/fetch.py b/surfsense_backend/app/proprietary/platforms/instagram/fetch.py index 73320eddb..e2b7c5818 100644 --- a/surfsense_backend/app/proprietary/platforms/instagram/fetch.py +++ b/surfsense_backend/app/proprietary/platforms/instagram/fetch.py @@ -32,6 +32,7 @@ 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 @@ -75,12 +76,6 @@ _MAX_ROTATIONS = 3 _MAX_BACKOFFS = 4 _BACKOFF_BASE_S = 5.0 -# Endpoints Instagram serves only to logged-in clients (confirmed live). A bare -# 401/403 here is an endpoint auth wall, not a per-IP block, so every rotated IP -# hits the same wall — fail fast instead of burning the pool, exactly like the -# /accounts/login/ redirect branch. Content endpoints (profiles) still rotate. -_AUTH_WALLED_PATHS = ("web/search/topsearch/", "api/v1/tags/web_info/") - # 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. @@ -149,6 +144,20 @@ def _parse_json(page: Any) -> Any | 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. @@ -308,20 +317,44 @@ async def resolve_redirect(url: str) -> str | None: async def fetch_json(path: str, params: dict[str, Any] | None = None) -> Any | None: - """GET an Instagram web endpoint through a warmed HTTP session. + """GET an Instagram web endpoint through a warmed session; parse JSON. Returns parsed JSON (dict or list), or ``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). + 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//`` og-meta / ld+json) 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_json(path, params) + return await _fetch(path, params, extract) url = _build_url(path, params) attempt = 0 @@ -357,7 +390,7 @@ async def fetch_json(path: str, params: dict[str, Any] | None = None) -> Any | N f"Instagram login wall on {path} (endpoint requires auth)" ) if status == 200: - return _parse_json(page) + return extract(page) if status == 404: return None if status == _BACKOFF_STATUS and backoffs < _MAX_BACKOFFS: @@ -369,13 +402,9 @@ async def fetch_json(path: str, params: dict[str, Any] | None = None) -> Any | N await asyncio.sleep(delay + random.uniform(0, 1)) continue if status in _ROTATE_STATUSES: - # Bare 401/403 on a login-gated endpoint: rotating never clears an - # endpoint auth wall, so fail fast (mirrors the login-redirect - # branch above). Other endpoints rotate — a per-IP 401 recovers. - if any(p in path for p in _AUTH_WALLED_PATHS): - raise InstagramAccessBlockedError( - f"Instagram login wall on {path} (endpoint requires auth)" - ) + # 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() diff --git a/surfsense_backend/app/proprietary/platforms/instagram/parsers.py b/surfsense_backend/app/proprietary/platforms/instagram/parsers.py index 96d66a2c5..55884537d 100644 --- a/surfsense_backend/app/proprietary/platforms/instagram/parsers.py +++ b/surfsense_backend/app/proprietary/platforms/instagram/parsers.py @@ -15,6 +15,7 @@ parity is additive. from __future__ import annotations +import json import re from datetime import UTC, datetime from typing import Any @@ -136,28 +137,6 @@ def parse_media(node: dict[str, Any]) -> dict[str, Any]: } -def parse_comment(node: dict[str, Any], *, post_url: str | None) -> dict[str, Any]: - """Map a raw comment node to a flat comment item dict.""" - owner = node.get("owner") if isinstance(node.get("owner"), dict) else {} - code = _shortcode(node) - return { - "id": node.get("id"), - "postUrl": post_url, - "commentUrl": f"{_BASE}/p/{code}/c/{node.get('id')}/" if code else None, - "text": node.get("text"), - "ownerUsername": owner.get("username"), - "ownerProfilePicUrl": owner.get("profile_pic_url"), - "timestamp": _utc_from_sec(node.get("created_at")), - "repliesCount": _edge_count(node, "edge_threaded_comments") - or _int(node.get("child_comment_count")), - "likesCount": _edge_count(node, "edge_liked_by") - or _int(node.get("comment_like_count")), - "owner": {"id": owner.get("id"), "username": owner.get("username")} - if owner - else None, - } - - 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") @@ -185,39 +164,162 @@ def parse_profile(user: dict[str, Any]) -> dict[str, Any]: } -def parse_hashtag(data: dict[str, Any]) -> dict[str, Any]: - """Map a raw ``tags/web_info`` payload to a flat hashtag item dict.""" - node = data.get("data") if isinstance(data.get("data"), dict) else data - name = node.get("name") - top = _edges(node.get("edge_hashtag_to_top_posts")) - recent = _edges(node.get("edge_hashtag_to_media")) - return { - "detailKind": "hashtag", - "id": node.get("id"), - "name": name, - "url": f"{_BASE}/explore/tags/{name}/" if name else None, - "postsCount": _edge_count(node, "edge_hashtag_to_media"), - "topPosts": [parse_media(n) for n in top], - "posts": [parse_media(n) for n in recent], - } +# Anonymous single-post extraction (/p//, /reel//) -------- +# +# Instagram serves logged-out visitors the post's public metadata inside the +# document itself, not via a JSON XHR (the ``?__a=1`` API 404s / login-walls for +# anonymous callers). Two durable, anonymous surfaces carry it: +# 1. ``', + re.IGNORECASE | re.DOTALL, +) +_OG_RE = re.compile( + r' dict[str, Any]: - """Map a raw ``locations/web_info`` payload to a flat place item dict.""" - loc = data.get("location") if isinstance(data.get("location"), dict) else data - recent = _edges(loc.get("edge_location_to_media")) +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 _ldjson_blocks(html: str) -> list[dict[str, Any]]: + """Parse every ``application/ld+json`` script block into dicts.""" + out: list[dict[str, Any]] = [] + for raw in _LDJSON_RE.findall(html): + try: + data = json.loads(raw.strip()) + except (ValueError, TypeError): + continue + for node in data if isinstance(data, list) else [data]: + if isinstance(node, dict): + out.append(node) + return out + + +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 _ld_interaction(node: dict[str, Any]) -> dict[str, int]: + """Pull like/comment counts out of schema.org ``interactionStatistic``.""" + stats = node.get("interactionStatistic") + items = stats if isinstance(stats, list) else [stats] if stats else [] + out: dict[str, int] = {} + for stat in items: + if not isinstance(stat, dict): + continue + itype = str(stat.get("interactionType") or "") + count = _html_int(stat.get("userInteractionCount")) + if count is None: + continue + if "Like" in itype: + out["likes"] = count + elif "Comment" in itype: + out["comments"] = count + return out + + +def _ld_author_username(node: dict[str, Any]) -> str | None: + """Owner handle from a schema.org ``author`` (alternateName / identifier).""" + author = node.get("author") + author = author[0] if isinstance(author, list) and author else author + if not isinstance(author, dict): + return None + for key in ("alternateName", "identifier", "name"): + val = author.get(key) + if isinstance(val, dict): + val = val.get("value") + if isinstance(val, str) and val.strip(): + return val.strip().lstrip("@") or None + return None + + +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 schema.org ``ld+json`` block, falling back to Open Graph + meta tags for whatever it omits. 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 + blocks = _ldjson_blocks(html) + og = _og_tags(html) + if not blocks and not og: + return None + + node = next( + (b for b in blocks if str(b.get("@type", "")).endswith(("Object", "Post"))), + blocks[0] if blocks else {}, + ) + counts = _ld_interaction(node) + if "likes" not in counts or "comments" not in counts: + m = _OG_COUNTS_RE.search(og.get("description", "")) + if m: + counts.setdefault("likes", _html_int(m.group(1)) or 0) + counts.setdefault("comments", _html_int(m.group(2)) or 0) + + caption = ( + node.get("articleBody") + or node.get("caption") + or node.get("description") + or og.get("description") + ) + caption = caption if isinstance(caption, str) else None + + video = node.get("video") + video = video[0] if isinstance(video, list) and video else video + video_url = ( + video.get("contentUrl") if isinstance(video, dict) else None + ) or og.get("video") + is_video = bool(video_url) or og.get("type") == "video.other" + + image = node.get("image") + image = image[0] if isinstance(image, list) and image else image + display_url = ( + image.get("url") if isinstance(image, dict) else image + if isinstance(image, str) + else None + ) or og.get("image") + return { - "detailKind": "place", - "name": loc.get("name"), - "location_id": str(loc.get("id")) if loc.get("id") is not None else None, - "slug": loc.get("slug"), - "lat": loc.get("lat"), - "lng": loc.get("lng"), - "location_address": loc.get("address_json") or loc.get("address"), - "location_city": loc.get("city"), - "phone": loc.get("phone"), - "website": loc.get("website"), - "category": loc.get("category"), - "media_count": _edge_count(loc, "edge_location_to_media"), - "posts": [parse_media(n) for n in recent], + "id": node.get("identifier") if isinstance(node.get("identifier"), str) else None, + "type": "Video" if is_video else "Image", + "shortCode": shortcode, + "caption": caption, + "hashtags": _HASHTAG_RE.findall(caption) if caption else [], + "mentions": _MENTION_RE.findall(caption) if caption else [], + "url": url, + "commentsCount": counts.get("comments"), + "displayUrl": display_url, + "videoUrl": video_url if is_video else None, + "likesCount": counts.get("likes"), + "timestamp": node.get("uploadDate") or node.get("datePublished"), + "ownerUsername": _ld_author_username(node), } diff --git a/surfsense_backend/app/proprietary/platforms/instagram/schemas.py b/surfsense_backend/app/proprietary/platforms/instagram/schemas.py index 2175744a2..76ad81d12 100644 --- a/surfsense_backend/app/proprietary/platforms/instagram/schemas.py +++ b/surfsense_backend/app/proprietary/platforms/instagram/schemas.py @@ -19,11 +19,8 @@ from typing import Any, Literal from pydantic import BaseModel, ConfigDict, Field -InstagramResultsType = Literal[ - "posts", "details", "comments", "reels", "mentions", "stories" -] -InstagramSearchType = Literal["hashtag", "profile", "place", "user"] -InstagramDetailKind = Literal["profile", "hashtag", "place"] +InstagramResultsType = Literal["posts", "details", "reels", "mentions"] +InstagramSearchType = Literal["profile", "user"] class InstagramScrapeInput(BaseModel): @@ -41,12 +38,10 @@ class InstagramScrapeInput(BaseModel): resultsLimit: int | None = Field(default=None, ge=1) onlyPostsNewerThan: str | None = None search: str | None = None - searchType: InstagramSearchType = "hashtag" + searchType: InstagramSearchType = "profile" searchLimit: int | None = Field(default=None, ge=1, le=250) addParentData: bool = False skipPinnedPosts: bool = False - isNewestComments: bool = False - includeNestedComments: bool = False addProfileStatistics: bool = False @@ -109,22 +104,6 @@ class InstagramMediaItem(_ItemBase): dataSource: dict[str, Any] | None = None -class InstagramComment(_ItemBase): - """A comment on a post / reel.""" - - id: str | None = None - postUrl: str | None = None - commentUrl: str | None = None - text: str | None = None - ownerUsername: str | None = None - ownerProfilePicUrl: str | None = None - timestamp: str | None = None - repliesCount: int | None = None - replies: list[dict[str, Any]] = Field(default_factory=list) - likesCount: int | None = None - owner: dict[str, Any] | None = None - - class InstagramProfile(_ItemBase): """A profile detail item (``detailKind = "profile"``).""" @@ -149,39 +128,3 @@ class InstagramProfile(_ItemBase): relatedProfiles: list[dict[str, Any]] = Field(default_factory=list) latestPosts: list[dict[str, Any]] = Field(default_factory=list) statistics: dict[str, Any] | None = None - - -class InstagramHashtag(_ItemBase): - """A hashtag detail item (``detailKind = "hashtag"``).""" - - detailKind: Literal["hashtag"] = "hashtag" - id: str | None = None - name: str | None = None - url: str | None = None - postsCount: int | None = None - topPosts: list[dict[str, Any]] = Field(default_factory=list) - posts: list[dict[str, Any]] = Field(default_factory=list) - related: list[dict[str, Any]] = Field(default_factory=list) - searchTerm: str | None = None - searchSource: str | None = None - - -class InstagramPlace(_ItemBase): - """A place detail item (``detailKind = "place"``).""" - - detailKind: Literal["place"] = "place" - name: str | None = None - location_id: str | None = None - slug: str | None = None - lat: float | None = None - lng: float | None = None - location_address: str | None = None - location_city: str | None = None - location_zip: str | None = None - phone: str | None = None - website: str | None = None - category: str | None = None - media_count: int | None = None - posts: list[dict[str, Any]] = Field(default_factory=list) - searchTerm: str | None = None - searchSource: str | None = None diff --git a/surfsense_backend/app/proprietary/platforms/instagram/scraper.py b/surfsense_backend/app/proprietary/platforms/instagram/scraper.py index f748d5026..381b40d5b 100644 --- a/surfsense_backend/app/proprietary/platforms/instagram/scraper.py +++ b/surfsense_backend/app/proprietary/platforms/instagram/scraper.py @@ -10,11 +10,15 @@ 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`` / ``mentions`` -> media items (profile / hashtag feeds, - or discovery search) -- ``comments`` -> comment items for post/reel URLs -- ``details`` -> profile / hashtag / place metadata (by URL or discovery search) +- ``posts`` / ``reels`` / ``mentions`` -> 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. @@ -32,20 +36,18 @@ 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_comment, - parse_hashtag, - parse_media, - parse_place, - parse_profile, -) +from .parsers import parse_media, parse_post, parse_profile from .schemas import InstagramScrapeInput from .url_resolver import ResolvedUrl, resolve_url @@ -62,14 +64,7 @@ __all__ = [ # residential pool with parallel login walls. _FANOUT_CONCURRENCY = 8 -# Per-post comment fetches fan across their own warm sessions; kept below the -# top-level width so N concurrent targets x this can't explode the IP count. -_COMMENT_CONCURRENCY = 4 - _PROFILE_PATH = "api/v1/users/web_profile_info/" -_HASHTAG_PATH = "api/v1/tags/web_info/" -_LOCATION_PATH = "api/v1/locations/web_info/" -_SEARCH_PATH = "web/search/topsearch/" # Instagram usernames: 1-30 chars of letters/digits/period/underscore. Used to # treat a profile/user discovery query as a direct (anonymous) handle lookup. @@ -235,7 +230,7 @@ async def _media_flow( cutoff: datetime | None, per_target: int, ) -> AsyncIterator[dict[str, Any]]: - """Emit media items for a profile / hashtag / place URL.""" + """Emit media items for a profile feed, or a single ``/p/``/``/reel/`` page.""" from .parsers import _edges result_type = input_model.resultsType @@ -258,128 +253,95 @@ async def _media_flow( if emitted >= per_target: return return - if resolved.kind == "hashtag": - data = await fetch_json(_HASHTAG_PATH, {"tag_name": resolved.value}) - if isinstance(data, dict): - parsed = parse_hashtag(data) - emitted = 0 - for node in [*parsed.get("topPosts", []), *parsed.get("posts", [])]: - if not _media_matches(node, result_type): - continue - if not _is_after(node.get("timestamp"), cutoff): - continue - yield _emit(node, input_url=resolved.url) - emitted += 1 - if emitted >= per_target: - return - return - if resolved.kind == "place": - data = await fetch_json(_LOCATION_PATH, {"location_id": resolved.value}) - if isinstance(data, dict): - parsed = parse_place(data) - emitted = 0 - for node in parsed.get("posts", []): - if not _is_after(node.get("timestamp"), cutoff): - continue - yield _emit(node, input_url=resolved.url) - emitted += 1 - if emitted >= per_target: - return - return - - -async def _comments_flow( - resolved: ResolvedUrl, - *, - input_model: InstagramScrapeInput, - per_target: int, -) -> AsyncIterator[dict[str, Any]]: - """Emit comment items for a post / reel URL. - - ponytail: the anonymous comment page uses a GraphQL cursor whose doc-id - drifts; v1 sources the comments embedded in the media info payload and caps - at the actor's 50/post ceiling. Deeper paging is the upgrade path in - ``fetch.py``. - """ - from .parsers import _edges - - path = f"p/{resolved.value}/" - data = await fetch_json(path, {"__a": 1, "__d": "dis"}) - node = None - if isinstance(data, dict): - items = data.get("items") - if isinstance(items, list) and items: - node = items[0] - else: - gql = data.get("graphql") - node = gql.get("shortcode_media") if isinstance(gql, dict) else None - if not isinstance(node, dict): - return - comment_nodes = _edges(node.get("edge_media_to_parent_comment")) or _edges( - node.get("edge_media_to_comment") - ) - cap = min(per_target, 50) - emitted = 0 - for cnode in comment_nodes: - item = parse_comment(cnode, post_url=resolved.url) - yield _emit(item, input_url=resolved.url) - emitted += 1 - if input_model.includeNestedComments: - for reply in _edges(cnode.get("edge_threaded_comments")): - if emitted >= cap: - return - yield _emit( - parse_comment(reply, post_url=resolved.url), - input_url=resolved.url, - ) - emitted += 1 - if emitted >= cap: + 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 post's og-meta + + # ld+json, 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 / hashtag / place detail item for a URL.""" + """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) - return - if resolved.kind == "hashtag": - data = await fetch_json(_HASHTAG_PATH, {"tag_name": resolved.value}) - if isinstance(data, dict): - yield _emit(parse_hashtag(data), input_url=resolved.url) - return - if resolved.kind == "place": - data = await fetch_json(_LOCATION_PATH, {"location_id": resolved.value}) - if isinstance(data, dict): - yield _emit(parse_place(data), input_url=resolved.url) - return + + +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 targets - anonymously. + """Resolve a discovery query into profile targets - anonymously. - Instagram's keyword search (``topsearch``) is login-walled, so we never call - it. A profile/user query is resolved as a direct handle lookup against the - anonymous profile endpoint ("messi" -> instagram.com/messi/). Hashtag/place - keyword discovery has NO anonymous endpoint (topsearch and the tag/location - feeds all require a session), so we surface a clear block instead of a - misleading empty success. + 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. """ - if search_type in ("profile", "user"): - handle = query.strip().lstrip("@") - if _HANDLE_RE.match(handle): - url = f"https://www.instagram.com/{handle}/" - return [ResolvedUrl("profile", handle, url)][:limit] - return [] # not a handle, and no anonymous fuzzy search -> nothing to do - raise InstagramAccessBlockedError( - f"Instagram {search_type} search requires login and is unavailable in " - "anonymous mode - pass a profile URL or handle via directUrls instead" - ) + 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]: @@ -425,17 +387,6 @@ async def iter_instagram( cutoff = _parse_newer_than(input_model.onlyPostsNewerThan) per_target = input_model.resultsLimit or 10 - if result_type == "comments": - jobs = [ - _comments_flow(r, input_model=input_model, per_target=per_target) - for r in targets - if r.kind in ("post", "reel") - ] - async with aclosing(fan_out(jobs, concurrency=_COMMENT_CONCURRENCY)) as stream: - async for item in stream: - yield item - return - if result_type == "details": jobs = [_details_flow(r, input_model=input_model) for r in targets] async with aclosing(fan_out(jobs)) as stream: diff --git a/surfsense_backend/app/proprietary/platforms/instagram/url_resolver.py b/surfsense_backend/app/proprietary/platforms/instagram/url_resolver.py index 596b85c2c..aec46a678 100644 --- a/surfsense_backend/app/proprietary/platforms/instagram/url_resolver.py +++ b/surfsense_backend/app/proprietary/platforms/instagram/url_resolver.py @@ -1,18 +1,18 @@ """Classify and normalize an Instagram URL into a scrape job. -Covers the supported ``directUrls`` shapes: a profile, a post (``/p/``), a reel -(``/reel/``), a hashtag (``/explore/tags/``), and a place -(``/explore/locations/``), plus bare profile IDs. Non-Instagram hosts resolve to -``None`` so the orchestrator can skip them. Mirrors the frozen ``ResolvedUrl`` -dataclass shape of ``../reddit/url_resolver.py``. +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. -- Location URLs are valid with the numeric ID alone (no trailing slug). -- Numeric post-ID URLs are only valid for the ``comments`` flow; elsewhere the - shortCode form is required, so a numeric-ID URL resolves with - ``numeric_post_id`` set and callers reject it outside comments. +- 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/`` redirect resolution is handled at fetch time (network), not here. """ @@ -22,7 +22,7 @@ from dataclasses import dataclass from typing import Literal from urllib.parse import urlparse -ResolvedKind = Literal["profile", "post", "reel", "hashtag", "place"] +ResolvedKind = Literal["profile", "post", "reel"] _INSTAGRAM_HOSTS = frozenset( {"m.instagram.com", "www.instagram.com", "instagram.com"} @@ -75,11 +75,6 @@ def resolve_url(url: str) -> ResolvedUrl | None: if not segments: return None head = segments[0] - if head == "explore" and len(segments) >= 3 and segments[1] == "tags": - return ResolvedUrl("hashtag", segments[2], url) - if head == "explore" and len(segments) >= 3 and segments[1] == "locations": - slug = segments[3] if len(segments) >= 4 else None - return ResolvedUrl("place", segments[2], url, slug=slug) if head == "p" and len(segments) >= 2: code = segments[1] return ResolvedUrl("post", code, url, numeric_post_id=code.isdigit())