refactor(instagram): streamline scraper, fetch, and parsing pipeline

This commit is contained in:
Anish Sarkar 2026-07-11 02:20:51 +05:30
parent de1990f9f6
commit 4813dc96e3
6 changed files with 304 additions and 290 deletions

View file

@ -2,10 +2,7 @@
from .fetch import InstagramAccessBlockedError from .fetch import InstagramAccessBlockedError
from .schemas import ( from .schemas import (
InstagramComment,
InstagramHashtag,
InstagramMediaItem, InstagramMediaItem,
InstagramPlace,
InstagramProfile, InstagramProfile,
InstagramScrapeInput, InstagramScrapeInput,
) )
@ -13,10 +10,7 @@ from .scraper import iter_instagram, scrape_instagram
__all__ = [ __all__ = [
"InstagramAccessBlockedError", "InstagramAccessBlockedError",
"InstagramComment",
"InstagramHashtag",
"InstagramMediaItem", "InstagramMediaItem",
"InstagramPlace",
"InstagramProfile", "InstagramProfile",
"InstagramScrapeInput", "InstagramScrapeInput",
"iter_instagram", "iter_instagram",

View file

@ -32,6 +32,7 @@ import json
import logging import logging
import random import random
import time import time
from collections.abc import Callable
from contextlib import asynccontextmanager, suppress from contextlib import asynccontextmanager, suppress
from contextvars import ContextVar from contextvars import ContextVar
from datetime import UTC, datetime from datetime import UTC, datetime
@ -75,12 +76,6 @@ _MAX_ROTATIONS = 3
_MAX_BACKOFFS = 4 _MAX_BACKOFFS = 4
_BACKOFF_BASE_S = 5.0 _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 # 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; # 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. # 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 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: def _is_login_redirect(page: Any) -> bool:
"""True if IG redirected this request to the anonymous login wall. """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: 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. 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 See :func:`_fetch` for the warm/rotate/backoff resilience contract.
401/403; backs off on 429. Raises :class:`InstagramAccessBlockedError` only """
when every rotated IP refuses anonymous access (the login-wall branch, which return await _fetch(path, params, _parse_json)
we cannot satisfy).
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/<shortcode>/`` 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() holder = _current_session.get()
if holder is None: if holder is None:
# No bound session (e.g. a direct call outside fan_out): open a # 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. # short-lived warmed session for this one fetch, then tear it down.
async with proxy_session(): async with proxy_session():
return await fetch_json(path, params) return await _fetch(path, params, extract)
url = _build_url(path, params) url = _build_url(path, params)
attempt = 0 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)" f"Instagram login wall on {path} (endpoint requires auth)"
) )
if status == 200: if status == 200:
return _parse_json(page) return extract(page)
if status == 404: if status == 404:
return None return None
if status == _BACKOFF_STATUS and backoffs < _MAX_BACKOFFS: 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)) await asyncio.sleep(delay + random.uniform(0, 1))
continue continue
if status in _ROTATE_STATUSES: if status in _ROTATE_STATUSES:
# Bare 401/403 on a login-gated endpoint: rotating never clears an # Bare 401/403: a per-IP block that a fresh exit IP recovers, so
# endpoint auth wall, so fail fast (mirrors the login-redirect # rotate and re-warm. (The endpoint-level auth wall is caught by
# branch above). Other endpoints rotate — a per-IP 401 recovers. # the login-redirect branch above and fails fast without rotating.)
if any(p in path for p in _AUTH_WALLED_PATHS):
raise InstagramAccessBlockedError(
f"Instagram login wall on {path} (endpoint requires auth)"
)
if attempt < _MAX_ROTATIONS: if attempt < _MAX_ROTATIONS:
attempt += 1 attempt += 1
await holder.rotate() await holder.rotate()

View file

@ -15,6 +15,7 @@ parity is additive.
from __future__ import annotations from __future__ import annotations
import json
import re import re
from datetime import UTC, datetime from datetime import UTC, datetime
from typing import Any 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]: def parse_profile(user: dict[str, Any]) -> dict[str, Any]:
"""Map a raw ``web_profile_info`` ``data.user`` to a flat profile item dict.""" """Map a raw ``web_profile_info`` ``data.user`` to a flat profile item dict."""
username = user.get("username") 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]: # Anonymous single-post extraction (/p/<shortcode>/, /reel/<shortcode>/) --------
"""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 # Instagram serves logged-out visitors the post's public metadata inside the
name = node.get("name") # document itself, not via a JSON XHR (the ``?__a=1`` API 404s / login-walls for
top = _edges(node.get("edge_hashtag_to_top_posts")) # anonymous callers). Two durable, anonymous surfaces carry it:
recent = _edges(node.get("edge_hashtag_to_media")) # 1. ``<script type="application/ld+json">`` — schema.org VideoObject/ImageObject
return { # with author, caption (articleBody/caption), uploadDate, interactionStatistic
"detailKind": "hashtag", # (likes/comments), and the media URL.
"id": node.get("id"), # 2. Open Graph ``<meta property="og:*">`` — a lossy fallback (og:description
"name": name, # packs "N likes, M comments - author on DATE: caption").
"url": f"{_BASE}/explore/tags/{name}/" if name else None, # ld+json is preferred; og fills gaps. ponytail: pinned to these two surfaces —
"postsCount": _edge_count(node, "edge_hashtag_to_media"), # if a live probe shows a different embedded blob (e.g. a PolarisPost JSON), add a
"topPosts": [parse_media(n) for n in top], # branch here; the wiring in scraper._media_flow stays the same.
"posts": [parse_media(n) for n in recent],
} _LDJSON_RE = re.compile(
r'<script[^>]+type="application/ld\+json"[^>]*>(.*?)</script>',
re.IGNORECASE | re.DOTALL,
)
_OG_RE = re.compile(
r'<meta\s+property="og:([^"]+)"\s+content="([^"]*)"', re.IGNORECASE
)
# og:description shape: "1,234 likes, 56 comments - author on 2024-01-02 ..."
_OG_COUNTS_RE = re.compile(
r"([\d.,]+)\s+likes?,\s*([\d.,]+)\s+comments?", re.IGNORECASE
)
def parse_place(data: dict[str, Any]) -> dict[str, Any]: def _html_int(value: Any) -> int | None:
"""Map a raw ``locations/web_info`` payload to a flat place item dict.""" """Coerce a string/number (``"1,234"``) to int, or ``None``."""
loc = data.get("location") if isinstance(data.get("location"), dict) else data if isinstance(value, bool):
recent = _edges(loc.get("edge_location_to_media")) 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:<key>`` -> 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/<code>/`` (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 { return {
"detailKind": "place", "id": node.get("identifier") if isinstance(node.get("identifier"), str) else None,
"name": loc.get("name"), "type": "Video" if is_video else "Image",
"location_id": str(loc.get("id")) if loc.get("id") is not None else None, "shortCode": shortcode,
"slug": loc.get("slug"), "caption": caption,
"lat": loc.get("lat"), "hashtags": _HASHTAG_RE.findall(caption) if caption else [],
"lng": loc.get("lng"), "mentions": _MENTION_RE.findall(caption) if caption else [],
"location_address": loc.get("address_json") or loc.get("address"), "url": url,
"location_city": loc.get("city"), "commentsCount": counts.get("comments"),
"phone": loc.get("phone"), "displayUrl": display_url,
"website": loc.get("website"), "videoUrl": video_url if is_video else None,
"category": loc.get("category"), "likesCount": counts.get("likes"),
"media_count": _edge_count(loc, "edge_location_to_media"), "timestamp": node.get("uploadDate") or node.get("datePublished"),
"posts": [parse_media(n) for n in recent], "ownerUsername": _ld_author_username(node),
} }

View file

@ -19,11 +19,8 @@ from typing import Any, Literal
from pydantic import BaseModel, ConfigDict, Field from pydantic import BaseModel, ConfigDict, Field
InstagramResultsType = Literal[ InstagramResultsType = Literal["posts", "details", "reels", "mentions"]
"posts", "details", "comments", "reels", "mentions", "stories" InstagramSearchType = Literal["profile", "user"]
]
InstagramSearchType = Literal["hashtag", "profile", "place", "user"]
InstagramDetailKind = Literal["profile", "hashtag", "place"]
class InstagramScrapeInput(BaseModel): class InstagramScrapeInput(BaseModel):
@ -41,12 +38,10 @@ class InstagramScrapeInput(BaseModel):
resultsLimit: int | None = Field(default=None, ge=1) resultsLimit: int | None = Field(default=None, ge=1)
onlyPostsNewerThan: str | None = None onlyPostsNewerThan: str | None = None
search: str | None = None search: str | None = None
searchType: InstagramSearchType = "hashtag" searchType: InstagramSearchType = "profile"
searchLimit: int | None = Field(default=None, ge=1, le=250) searchLimit: int | None = Field(default=None, ge=1, le=250)
addParentData: bool = False addParentData: bool = False
skipPinnedPosts: bool = False skipPinnedPosts: bool = False
isNewestComments: bool = False
includeNestedComments: bool = False
addProfileStatistics: bool = False addProfileStatistics: bool = False
@ -109,22 +104,6 @@ class InstagramMediaItem(_ItemBase):
dataSource: dict[str, Any] | None = None 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): class InstagramProfile(_ItemBase):
"""A profile detail item (``detailKind = "profile"``).""" """A profile detail item (``detailKind = "profile"``)."""
@ -149,39 +128,3 @@ class InstagramProfile(_ItemBase):
relatedProfiles: list[dict[str, Any]] = Field(default_factory=list) relatedProfiles: list[dict[str, Any]] = Field(default_factory=list)
latestPosts: list[dict[str, Any]] = Field(default_factory=list) latestPosts: list[dict[str, Any]] = Field(default_factory=list)
statistics: dict[str, Any] | None = None 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

View file

@ -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 to *this* module's proxy holders so every worker warms its own session once and
reuses it. 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``: Flows are selected by ``resultsType``:
- ``posts`` / ``reels`` / ``mentions`` -> media items (profile / hashtag feeds, - ``posts`` / ``reels`` / ``mentions`` -> media items (profile feed, or a single
or discovery search) ``/p/``/``/reel/`` page, or discovery search)
- ``comments`` -> comment items for post/reel URLs - ``details`` -> profile metadata (by URL or discovery search)
- ``details`` -> profile / hashtag / place metadata (by URL or discovery search)
ponytail: deep feed pagination (past the first web page of media) needs the 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. 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 datetime import UTC, datetime, timedelta
from typing import Any 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 ( from .fetch import (
InstagramAccessBlockedError, InstagramAccessBlockedError,
bind_proxy_holder, bind_proxy_holder,
fetch_html,
fetch_json, fetch_json,
now_iso, now_iso,
open_proxy_holder, open_proxy_holder,
) )
from .parsers import ( from .parsers import parse_media, parse_post, parse_profile
parse_comment,
parse_hashtag,
parse_media,
parse_place,
parse_profile,
)
from .schemas import InstagramScrapeInput from .schemas import InstagramScrapeInput
from .url_resolver import ResolvedUrl, resolve_url from .url_resolver import ResolvedUrl, resolve_url
@ -62,14 +64,7 @@ __all__ = [
# residential pool with parallel login walls. # residential pool with parallel login walls.
_FANOUT_CONCURRENCY = 8 _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/" _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 # Instagram usernames: 1-30 chars of letters/digits/period/underscore. Used to
# treat a profile/user discovery query as a direct (anonymous) handle lookup. # treat a profile/user discovery query as a direct (anonymous) handle lookup.
@ -235,7 +230,7 @@ async def _media_flow(
cutoff: datetime | None, cutoff: datetime | None,
per_target: int, per_target: int,
) -> AsyncIterator[dict[str, Any]]: ) -> 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 from .parsers import _edges
result_type = input_model.resultsType result_type = input_model.resultsType
@ -258,128 +253,95 @@ async def _media_flow(
if emitted >= per_target: if emitted >= per_target:
return return
return return
if resolved.kind == "hashtag": if resolved.kind in ("post", "reel"):
data = await fetch_json(_HASHTAG_PATH, {"tag_name": resolved.value}) # Single-post extraction: the anonymous ``?__a=1`` JSON API 404s/login-
if isinstance(data, dict): # walls, but the public /p/<code>/ document embeds the post's og-meta +
parsed = parse_hashtag(data) # ld+json, which parse_post reads. Numeric-ID URLs can't be keyed this
emitted = 0 # way (the page needs the shortCode), so they're skipped upstream.
for node in [*parsed.get("topPosts", []), *parsed.get("posts", [])]: if resolved.numeric_post_id:
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:
return 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( async def _details_flow(
resolved: ResolvedUrl, *, input_model: InstagramScrapeInput resolved: ResolvedUrl, *, input_model: InstagramScrapeInput
) -> AsyncIterator[dict[str, Any]]: ) -> 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": if resolved.kind == "profile":
user = await _profile_user(resolved.value) user = await _profile_user(resolved.value)
if user is not None: if user is not None:
yield _emit(parse_profile(user), input_url=resolved.url) yield _emit(parse_profile(user), input_url=resolved.url)
return
if resolved.kind == "hashtag":
data = await fetch_json(_HASHTAG_PATH, {"tag_name": resolved.value}) def _kind_matches(resolved: ResolvedUrl, search_type: str) -> bool:
if isinstance(data, dict): """True if a resolved IG URL is the kind the discovery query asked for.
yield _emit(parse_hashtag(data), input_url=resolved.url)
return Discovery is profile-only now (hashtag/place feeds are login-walled), so
if resolved.kind == "place": every supported ``search_type`` maps to a profile target.
data = await fetch_json(_LOCATION_PATH, {"location_id": resolved.value}) """
if isinstance(data, dict): return resolved.kind == "profile"
yield _emit(parse_place(data), input_url=resolved.url)
return
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( async def _discover(
query: str, *, search_type: str, limit: int query: str, *, search_type: str, limit: int
) -> list[ResolvedUrl]: ) -> 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 A query that is a valid handle resolves directly against the anonymous
it. A profile/user query is resolved as a direct handle lookup against the profile endpoint ("messi" -> instagram.com/messi/). A non-handle query (e.g.
anonymous profile endpoint ("messi" -> instagram.com/messi/). Hashtag/place "national geographic") goes through Google ``site:instagram.com`` since IG's
keyword discovery has NO anonymous endpoint (topsearch and the tag/location native keyword search is login-walled.
feeds all require a session), so we surface a clear block instead of a
misleading empty success.
""" """
if search_type in ("profile", "user"): handle = query.strip().lstrip("@")
handle = query.strip().lstrip("@") if _HANDLE_RE.match(handle):
if _HANDLE_RE.match(handle): url = f"https://www.instagram.com/{handle}/"
url = f"https://www.instagram.com/{handle}/" return [ResolvedUrl("profile", handle, url)][:limit]
return [ResolvedUrl("profile", handle, url)][:limit] return await _discover_via_google(query, search_type=search_type, limit=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"
)
def _resolve_inputs(input_model: InstagramScrapeInput) -> list[ResolvedUrl]: def _resolve_inputs(input_model: InstagramScrapeInput) -> list[ResolvedUrl]:
@ -425,17 +387,6 @@ async def iter_instagram(
cutoff = _parse_newer_than(input_model.onlyPostsNewerThan) cutoff = _parse_newer_than(input_model.onlyPostsNewerThan)
per_target = input_model.resultsLimit or 10 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": if result_type == "details":
jobs = [_details_flow(r, input_model=input_model) for r in targets] jobs = [_details_flow(r, input_model=input_model) for r in targets]
async with aclosing(fan_out(jobs)) as stream: async with aclosing(fan_out(jobs)) as stream:

View file

@ -1,18 +1,18 @@
"""Classify and normalize an Instagram URL into a scrape job. """Classify and normalize an Instagram URL into a scrape job.
Covers the supported ``directUrls`` shapes: a profile, a post (``/p/``), a reel Covers the anonymously-scrapable ``directUrls`` shapes: a profile, a post
(``/reel/``), a hashtag (``/explore/tags/``), and a place (``/p/``), and a reel (``/reel/``), plus bare profile IDs. Hashtag and place
(``/explore/locations/``), plus bare profile IDs. Non-Instagram hosts resolve to URLs are deliberately unsupported their feeds are login-walled for anonymous
``None`` so the orchestrator can skip them. Mirrors the frozen ``ResolvedUrl`` callers (use Google-backed discovery + single-post extraction instead).
dataclass shape of ``../reddit/url_resolver.py``. 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): Normalization rules (from the reference spec):
- ``_u/`` and ``/profilecard/`` segments are stripped. - ``_u/`` and ``/profilecard/`` segments are stripped.
- Story URLs (``/stories/<user>/...``) reduce to the profile. - Story URLs (``/stories/<user>/...``) reduce to the profile.
- Location URLs are valid with the numeric ID alone (no trailing slug). - Numeric post-ID URLs cannot be single-post-extracted anonymously (the HTML
- Numeric post-ID URLs are only valid for the ``comments`` flow; elsewhere the page keys on the shortCode), so they resolve with ``numeric_post_id`` set and
shortCode form is required, so a numeric-ID URL resolves with the media flow skips them.
``numeric_post_id`` set and callers reject it outside comments.
- ``share/`` redirect resolution is handled at fetch time (network), not here. - ``share/`` redirect resolution is handled at fetch time (network), not here.
""" """
@ -22,7 +22,7 @@ from dataclasses import dataclass
from typing import Literal from typing import Literal
from urllib.parse import urlparse from urllib.parse import urlparse
ResolvedKind = Literal["profile", "post", "reel", "hashtag", "place"] ResolvedKind = Literal["profile", "post", "reel"]
_INSTAGRAM_HOSTS = frozenset( _INSTAGRAM_HOSTS = frozenset(
{"m.instagram.com", "www.instagram.com", "instagram.com"} {"m.instagram.com", "www.instagram.com", "instagram.com"}
@ -75,11 +75,6 @@ def resolve_url(url: str) -> ResolvedUrl | None:
if not segments: if not segments:
return None return None
head = segments[0] 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: if head == "p" and len(segments) >= 2:
code = segments[1] code = segments[1]
return ResolvedUrl("post", code, url, numeric_post_id=code.isdigit()) return ResolvedUrl("post", code, url, numeric_post_id=code.isdigit())