From e61a3f0fab2c5a76ab9048a55c081a2231640305 Mon Sep 17 00:00:00 2001 From: subzeroid <143403577+subzeroid@users.noreply.github.com> Date: Tue, 30 Jun 2026 17:52:10 +0300 Subject: [PATCH] fix: mirror public media info web fallback --- CHANGELOG.md | 8 ++ aiograpi/__init__.py | 2 +- aiograpi/extractors.py | 8 +- aiograpi/mixins/media.py | 57 ++++++++- aiograpi/mixins/public.py | 53 ++++++-- pyproject.toml | 2 +- tests/regression/test_media_pagination.py | 145 +++++++++++++++++++++- tests/regression/test_public.py | 32 +++++ tests/regression/test_upstream_sync.py | 2 +- uv.lock | 2 +- 10 files changed, 287 insertions(+), 24 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 606077be..44f9cfd3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,14 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html) starting with 1.0.0. +## [1.11.1] - 2026-06-30 + +### Fixed + +- Mirrored the `instagrapi` 2.17.1 public media info fallback: current web `doc_id` payloads are posted to `/api/graphql` with LSD web headers and normalized from `xdt_api__v1__media__shortcode__web_info.items`. +- Handled Instagram media payloads where `video_versions` is present but `null`, including sidecar children, without falling back to private media info. +- Synced the recorded upstream baseline to `instagrapi` 2.17.1. + ## [1.11.0] - 2026-06-30 ### Added diff --git a/aiograpi/__init__.py b/aiograpi/__init__.py index 4a28c2dc..728231c2 100644 --- a/aiograpi/__init__.py +++ b/aiograpi/__init__.py @@ -46,7 +46,7 @@ # Used as fallback logger if another is not provided. DEFAULT_LOGGER = logging.getLogger("aiograpi") -__upstream_instagrapi_version__ = "2.17.0" +__upstream_instagrapi_version__ = "2.17.1" class Client( diff --git a/aiograpi/extractors.py b/aiograpi/extractors.py index cca0a0ae..4df791ea 100644 --- a/aiograpi/extractors.py +++ b/aiograpi/extractors.py @@ -60,7 +60,7 @@ def _normalize_media_gql_typename(data): def extract_media_v1(data): """Extract media from Private API""" media = deepcopy(data) - if "video_versions" in media: + if media.get("video_versions"): # Select Best Quality by Resolutiuon media["video_url"] = sorted(media["video_versions"], key=lambda o: o["height"] * o["width"])[-1]["url"] if media["media_type"] == 2 and not media.get("product_type"): @@ -221,7 +221,7 @@ def extract_media_inline_comment_gql(data, replied_to_comment_id=None): def extract_resource_v1(data): data = deepcopy(data) - if "video_versions" in data: + if data.get("video_versions"): data["video_url"] = sorted(data["video_versions"], key=lambda o: o["height"] * o["width"])[-1]["url"] candidates = data.get("image_versions2", {}).get("candidates", []) data["thumbnail_url"] = ( @@ -559,7 +559,7 @@ def extract_direct_message(data): def extract_direct_media(data): media = deepcopy(data) - if "video_versions" in media: + if media.get("video_versions"): # Select Best Quality by Resolutiuon media["video_url"] = sorted(media["video_versions"], key=lambda o: o["height"] * o["width"])[-1]["url"] if "image_versions2" in media: @@ -596,7 +596,7 @@ def extract_story_v1(data): """Extract story from Private API""" story = deepcopy(data) story["pk"] = str(story.get("pk")) - if "video_versions" in story: + if story.get("video_versions"): # Select Best Quality by Resolutiuon story["video_url"] = sorted(story["video_versions"], key=lambda o: o["height"] * o["width"])[-1]["url"] if story["media_type"] == 2 and not story.get("product_type"): diff --git a/aiograpi/mixins/media.py b/aiograpi/mixins/media.py index 1ea06c6d..37358149 100644 --- a/aiograpi/mixins/media.py +++ b/aiograpi/mixins/media.py @@ -37,7 +37,7 @@ from aiograpi.utils.serialization import dumps, json_value IG_PROFILE_TIMELINE_DOC_ID = "56030350814417327502004290437" -MEDIA_INFO_DOC_ID = "8845758582119845" +MEDIA_INFO_DOC_ID = "27128499623469141" class MediaMixin(ClientMixin): @@ -88,6 +88,46 @@ def _normalize_xdt_profile_media(media: Dict) -> Dict: media["taken_at"] = media["1ltaken_at"] return media + @staticmethod + def _normalize_xdt_media_info(media: Dict) -> Dict: + media = deepcopy(media) + user = media.get("user") or {} + if "pk" not in user and user.get("id"): + user["pk"] = user["id"] + media["user"] = user + media_pk = str(media.get("pk") or "") + media_id = str(media.get("id") or "") + if not media_pk and media_id: + media_pk = media_id.split("_", 1)[0] + media["pk"] = media_pk + elif "_" in media_pk: + media_pk = media_pk.split("_", 1)[0] + media["pk"] = media_pk + if media_id and "_" not in media_id and user.get("pk"): + media["id"] = f"{media_id}_{user['pk']}" + elif not media_id and media_pk and user.get("pk"): + media["id"] = f"{media_pk}_{user['pk']}" + if "code" not in media and media.get("shortcode"): + media["code"] = media["shortcode"] + if "taken_at" not in media and "taken_at_timestamp" in media: + media["taken_at"] = media["taken_at_timestamp"] + if not media.get("media_type"): + if media.get("carousel_media"): + media["media_type"] = 8 + elif media.get("video_versions"): + media["media_type"] = 2 + else: + media["media_type"] = 1 + if isinstance(media.get("caption"), str): + media["caption"] = {"text": media["caption"]} + elif "caption_text" in media and "caption" not in media: + media["caption"] = {"text": media.pop("caption_text") or ""} + elif "caption_text" in media: + media.pop("caption_text", None) + if media.get("carousel_media"): + media["carousel_media"] = [MediaMixin._normalize_xdt_media_info(item) for item in media["carousel_media"]] + return media + async def _user_medias_chunk_app_gql( self, user_id: int, amount: int = 0, end_cursor=None ) -> Tuple[List[Media], str]: @@ -564,13 +604,22 @@ async def media_info_gql(self, media_pk: str) -> Media: ): data = await self.public_doc_id_graphql_request( MEDIA_INFO_DOC_ID, - {"shortcode": shortcode}, + { + "shortcode": shortcode, + "__relay_internal__pv__PolarisAIGMMediaWebLabelEnabledrelayprovider": False, + }, referer=f"https://www.instagram.com/p/{shortcode}/", + url=self.GRAPHQL_PUBLIC_WEB_API_URL, + include_lsd=True, + headers={"X-FB-Friendly-Name": "PolarisPostRootQuery"}, ) media = data.get("xdt_shortcode_media") or data.get("shortcode_media") - if not media: + if media: + return extract_media_gql(media) + media_items = json_value(data, "xdt_api__v1__media__shortcode__web_info", "items", default=[]) + if not media_items: raise MediaNotFound(media_pk=media_pk, **data) - return extract_media_gql(media) + return extract_media_v1(self._normalize_xdt_media_info(media_items[0])) if not data.get("shortcode_media"): raise MediaNotFound(media_pk=media_pk, **data) if data["shortcode_media"]["location"] and self.authorization: diff --git a/aiograpi/mixins/public.py b/aiograpi/mixins/public.py index f51a2a95..14283fe6 100644 --- a/aiograpi/mixins/public.py +++ b/aiograpi/mixins/public.py @@ -1,6 +1,7 @@ import asyncio import json import logging +import re import time from pathlib import Path from typing import Any, Dict, Literal, Optional @@ -31,12 +32,15 @@ from aiograpi.utils.timing import random_delay PublicTransport = Literal["requests", "curl"] +PUBLIC_WEB_APP_ID = "936619743392459" +PUBLIC_WEB_ASBD_ID = "129477" class PublicRequestMixin(ClientMixin): public_requests_count = 0 PUBLIC_API_URL = "https://www.instagram.com/" GRAPHQL_PUBLIC_API_URL = "https://www.instagram.com/graphql/query/" + GRAPHQL_PUBLIC_WEB_API_URL = "https://www.instagram.com/api/graphql" last_public_response = None last_public_json = {} public_request_logger = logging.getLogger("public_request") @@ -410,19 +414,31 @@ async def public_graphql_request( pass raise ClientGraphqlError("Error: '{}'. Message: '{}'".format(e, message), response=e.response) + @staticmethod + def _extract_public_lsd_token(html: str) -> Optional[str]: + if not html: + return None + match = re.search(r'\["LSD",\[\],\{"token":"([^"]+)"\}', html) + if match: + return match.group(1) + match = re.search(r'"LSD",\[\],\{"token":"([^"]+)"', html) + return match.group(1) if match else None + async def public_doc_id_graphql_request( self, doc_id: str, variables: Dict[str, Any], referer: Optional[str] = None, headers: Optional[Dict[str, str]] = None, + url: Optional[str] = None, + include_lsd: bool = False, ) -> Dict[str, Any]: """ - POST a doc_id-based GraphQL query to www.instagram.com/graphql/query/. + POST a doc_id-based GraphQL query to Instagram's public web endpoints. - Newer IG GraphQL endpoints (e.g. PolarisProfilePageContentQuery) are - addressed by ``doc_id`` rather than the legacy ``query_hash`` / - ``query_id`` scheme. Returns the parsed ``data`` payload. + Newer Instagram web GraphQL endpoints use ``doc_id`` instead of the + legacy ``query_hash`` / ``query_id`` scheme. Returns the parsed + ``data`` payload. Parameters ---------- @@ -444,31 +460,48 @@ async def public_doc_id_graphql_request( inject_sessionid = getattr(self, "inject_sessionid_to_public", None) if inject_sessionid: inject_sessionid() - # IG rejects bare /graphql/query/ POSTs — needs the iPhone web-app - # signalling headers it sees from m.instagram.com (instaloader does - # the same: see _default_http_header(empty_session_only=True)). + query_url = url or self.GRAPHQL_PUBLIC_API_URL + referer_url = referer or "https://www.instagram.com/" + lsd = None + if include_lsd: + html = await self.public_request(referer_url, return_json=False) + lsd = self._extract_public_lsd_token(html) + if lsd: + data["lsd"] = lsd merged_headers = { "Accept": "*/*", "Accept-Encoding": "gzip, deflate", "Accept-Language": "en-US,en;q=0.8", - "Referer": referer or "https://www.instagram.com/", + "Referer": referer_url, "User-Agent": ( "Instagram 273.0.0.16.70 (iPhone15,2; iOS 17_5_1; en_US; en-US; scale=3.00; 1290x2796; 470085518)" ), } + if include_lsd: + merged_headers.update( + { + "Content-Type": "application/x-www-form-urlencoded", + "Origin": "https://www.instagram.com", + "User-Agent": self.public_user_agent, + "X-ASBD-ID": PUBLIC_WEB_ASBD_ID, + "X-IG-App-ID": PUBLIC_WEB_APP_ID, + } + ) + if lsd: + merged_headers["X-FB-LSD"] = lsd csrftoken = self.public.cookies_dict().get("csrftoken") if csrftoken: merged_headers["X-CSRFToken"] = csrftoken if headers: merged_headers.update(headers) body_json = await self.public_request( - self.GRAPHQL_PUBLIC_API_URL, + query_url, data=data, headers=merged_headers, update_headers=False, return_json=True, ) - if "data" not in body_json: + if body_json.get("data") is None: errors = body_json.get("errors") or [] summary = errors[0].get("summary") if errors else None description = errors[0].get("description") if errors else None diff --git a/pyproject.toml b/pyproject.toml index d232cd2d..3bea530b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "aiograpi" -version = "1.11.0" +version = "1.11.1" authors = [ { name = "Mark Subzeroid", email = "143403577+subzeroid@users.noreply.github.com" }, ] diff --git a/tests/regression/test_media_pagination.py b/tests/regression/test_media_pagination.py index 607ffc9d..a1130e57 100644 --- a/tests/regression/test_media_pagination.py +++ b/tests/regression/test_media_pagination.py @@ -259,12 +259,153 @@ async def test_media_info_gql_falls_back_to_doc_id_endpoint(self): media = await client.media_info_gql("1") client.public_doc_id_graphql_request.assert_awaited_once_with( - "8845758582119845", - {"shortcode": "B"}, + "27128499623469141", + { + "shortcode": "B", + "__relay_internal__pv__PolarisAIGMMediaWebLabelEnabledrelayprovider": False, + }, referer="https://www.instagram.com/p/B/", + url=client.GRAPHQL_PUBLIC_WEB_API_URL, + include_lsd=True, + headers={"X-FB-Friendly-Name": "PolarisPostRootQuery"}, ) assert media.pk == "1" + async def test_media_info_gql_extracts_current_web_info_payload(self): + client = Client() + media_payload = { + "pk": "3929128837042014584", + "id": "3929128837042014584_1713591624", + "code": "DaHEdwgogl4", + "taken_at": 1782608696, + "media_type": 2, + "product_type": "clips", + "user": { + "id": "1713591624", + "username": "example", + "profile_pic_url": "https://example.com/profile.jpg", + "is_private": False, + }, + "image_versions2": { + "candidates": [ + { + "url": "https://example.com/thumbnail.jpg", + "width": 720, + "height": 1280, + } + ] + }, + "video_versions": [ + { + "url": "https://example.com/video.mp4", + "width": 720, + "height": 1280, + "type": 101, + } + ], + "caption": {"text": "#suomi"}, + "comment_count": 1114, + "like_count": 113000, + "view_count": 200000, + "play_count": 210000, + "has_liked": False, + } + client.public_graphql_request = AsyncMock(side_effect=ClientForbiddenError("blocked")) + client.public_doc_id_graphql_request = AsyncMock( + return_value={"xdt_api__v1__media__shortcode__web_info": {"items": [media_payload]}} + ) + + media = await client.media_info_gql("3929128837042014584") + + client.public_doc_id_graphql_request.assert_awaited_once_with( + "27128499623469141", + { + "shortcode": "DaHEdwgogl4", + "__relay_internal__pv__PolarisAIGMMediaWebLabelEnabledrelayprovider": False, + }, + referer="https://www.instagram.com/p/DaHEdwgogl4/", + url=client.GRAPHQL_PUBLIC_WEB_API_URL, + include_lsd=True, + headers={"X-FB-Friendly-Name": "PolarisPostRootQuery"}, + ) + assert media.pk == "3929128837042014584" + assert media.id == "3929128837042014584_1713591624" + assert media.code == "DaHEdwgogl4" + assert media.media_type == 2 + assert media.product_type == "clips" + assert media.user.pk == "1713591624" + assert str(media.thumbnail_url) == "https://example.com/thumbnail.jpg" + assert str(media.video_url) == "https://example.com/video.mp4" + assert media.caption_text == "#suomi" + assert media.comment_count == 1114 + assert media.like_count == 113000 + assert media.view_count == 200000 + assert media.play_count == 210000 + assert media.has_liked is False + + async def test_media_info_handles_web_info_payload_with_null_video_versions(self): + client = Client() + user = { + "id": "1713591624", + "username": "example", + "profile_pic_url": "https://example.com/profile.jpg", + "is_private": False, + } + media_payload = { + "pk": "3908029358833535861", + "id": "3908029358833535861_1713591624", + "code": "DY8G_8JEgt1", + "taken_at": 1782608696, + "media_type": 8, + "user": user, + "image_versions2": { + "candidates": [ + { + "url": "https://example.com/thumbnail.jpg", + "width": 720, + "height": 1280, + } + ] + }, + "video_versions": None, + "carousel_media": [ + { + "pk": "3908029358833535862", + "id": "3908029358833535862_1713591624", + "media_type": 1, + "image_versions2": { + "candidates": [ + { + "url": "https://example.com/child.jpg", + "width": 720, + "height": 1280, + } + ] + }, + "video_versions": None, + } + ], + "caption": {"text": "sidecar"}, + "comment_count": 0, + "like_count": -1, + "has_liked": False, + } + client.public_graphql_request = AsyncMock(side_effect=ClientForbiddenError("blocked")) + client.public_doc_id_graphql_request = AsyncMock( + return_value={"xdt_api__v1__media__shortcode__web_info": {"items": [media_payload]}} + ) + client.media_info_v1 = AsyncMock(side_effect=AssertionError("private fallback")) + + media = await client.media_info("3908029358833535861") + + assert media.pk == "3908029358833535861" + assert media.code == "DY8G_8JEgt1" + assert media.media_type == 8 + assert media.video_url is None + assert len(media.resources) == 1 + assert media.resources[0].media_type == 1 + assert media.resources[0].video_url is None + async def test_media_info_gql_normalizes_xdt_sidecar_children(self): client = Client() child_payload = { diff --git a/tests/regression/test_public.py b/tests/regression/test_public.py index ba87086b..9efcf15a 100644 --- a/tests/regression/test_public.py +++ b/tests/regression/test_public.py @@ -34,3 +34,35 @@ async def test_public_doc_id_graphql_request_injects_logged_in_public_cookies(se self.assertEqual(client.public.cookies_dict()["sessionid"], "123:session") headers = client.public_request.await_args.kwargs["headers"] self.assertEqual(headers["X-CSRFToken"], "csrf-token") + + async def test_public_doc_id_graphql_request_posts_web_api_with_lsd(self): + client = Client() + client.public.set_cookies({"csrftoken": "csrf-token"}) + html = '' + client.public_request = AsyncMock(side_effect=[html, {"data": {"ok": True}}]) + + result = await client.public_doc_id_graphql_request( + "27128499623469141", + {"shortcode": "DaHEdwgogl4"}, + referer="https://www.instagram.com/p/DaHEdwgogl4/", + url=client.GRAPHQL_PUBLIC_WEB_API_URL, + include_lsd=True, + headers={"X-FB-Friendly-Name": "PolarisPostRootQuery"}, + ) + + self.assertEqual(result, {"ok": True}) + bootstrap_call, query_call = client.public_request.await_args_list + self.assertEqual(bootstrap_call.args[0], "https://www.instagram.com/p/DaHEdwgogl4/") + self.assertFalse(bootstrap_call.kwargs["return_json"]) + self.assertEqual(query_call.args[0], client.GRAPHQL_PUBLIC_WEB_API_URL) + kwargs = query_call.kwargs + self.assertEqual(kwargs["data"]["doc_id"], "27128499623469141") + self.assertEqual(kwargs["data"]["variables"], '{"shortcode":"DaHEdwgogl4"}') + self.assertEqual(kwargs["data"]["lsd"], "lsd-token") + self.assertEqual(kwargs["headers"]["X-FB-LSD"], "lsd-token") + self.assertEqual(kwargs["headers"]["X-CSRFToken"], "csrf-token") + self.assertEqual(kwargs["headers"]["X-FB-Friendly-Name"], "PolarisPostRootQuery") + self.assertEqual(kwargs["headers"]["X-ASBD-ID"], "129477") + self.assertEqual(kwargs["headers"]["X-IG-App-ID"], "936619743392459") + self.assertFalse(kwargs["update_headers"]) + self.assertTrue(kwargs["return_json"]) diff --git a/tests/regression/test_upstream_sync.py b/tests/regression/test_upstream_sync.py index 23a70de9..4ff26f51 100644 --- a/tests/regression/test_upstream_sync.py +++ b/tests/regression/test_upstream_sync.py @@ -2,4 +2,4 @@ def test_upstream_instagrapi_baseline_is_recorded(): - assert aiograpi.__upstream_instagrapi_version__ == "2.17.0" + assert aiograpi.__upstream_instagrapi_version__ == "2.17.1" diff --git a/uv.lock b/uv.lock index a2a6bc66..884b5a65 100644 --- a/uv.lock +++ b/uv.lock @@ -12,7 +12,7 @@ resolution-markers = [ [[package]] name = "aiograpi" -version = "1.11.0" +version = "1.11.1" source = { editable = "." } dependencies = [ { name = "httpx" },