From 6dd77e7b56acdf0a0017d473678b15d3cd7a2c72 Mon Sep 17 00:00:00 2001 From: Elias Date: Sun, 26 Jul 2026 11:47:46 +0300 Subject: [PATCH] Fix Post metadata fetching (#2706) Use doc_id 27128499623469141 (PolarisPostRootQuery / xdt_api__v1__media__shortcode__web_info) which returns v1/iPhone API format. Converts the new format to the legacy GraphQL field names that the rest of the codebase expects. --- instaloader/instaloadercontext.py | 8 ++ instaloader/structures.py | 121 +++++++++++++++++++++++++----- 2 files changed, 112 insertions(+), 17 deletions(-) diff --git a/instaloader/instaloadercontext.py b/instaloader/instaloadercontext.py index a6641b1..07b5378 100644 --- a/instaloader/instaloadercontext.py +++ b/instaloader/instaloadercontext.py @@ -541,6 +541,13 @@ class InstaloaderContext: :param referer: HTTP Referer, or None. :return: The server's response dictionary. """ + csrf = next((c.value for c in self._session.cookies + if c.name == 'csrftoken' and c.value), None) + if not csrf: + self._session.get('https://www.instagram.com/', timeout=self.request_timeout) + csrf = next((c.value for c in self._session.cookies + if c.name == 'csrftoken' and c.value), '') + with copy_session(self._session, self.request_timeout) as tmpsession: tmpsession.headers.update(self._default_http_header(empty_session_only=True)) del tmpsession.headers['Connection'] @@ -548,6 +555,7 @@ class InstaloaderContext: tmpsession.headers['authority'] = 'www.instagram.com' tmpsession.headers['scheme'] = 'https' tmpsession.headers['accept'] = '*/*' + tmpsession.headers['x-csrftoken'] = csrf if referer is not None: tmpsession.headers['referer'] = urllib.parse.quote(referer) diff --git a/instaloader/structures.py b/instaloader/structures.py index 8cee76b..3d2342c 100644 --- a/instaloader/structures.py +++ b/instaloader/structures.py @@ -258,6 +258,100 @@ class Post: fake_node["video_url"] = iphone_node["video_versions"][0]["url"] return fake_node + @staticmethod + def _fetch_play_count_from_clips(context: 'InstaloaderContext', user_id: str, + shortcode: str) -> Optional[int]: + """Fetch play_count for a reel via the clips connection endpoint as a fallback.""" + try: + resp = context.doc_id_graphql_query( + "27234427476213202", + {"data": {"include_feed_video": True, "page_size": 12, + "target_user_id": str(user_id)}}, + ) + edges = ((resp.get("data") or {}) + .get("xdt_api__v1__clips__user__connection_v2") or {}) + for edge in edges.get("edges") or []: + media = (edge.get("node") or {}).get("media") or {} + if media.get("code") == shortcode: + return media.get("play_count") + except (ConnectionException, BadResponseException): + pass + return None + + @staticmethod + def _normalize_post_data(media: Dict[str, Any], context: 'InstaloaderContext') -> Dict[str, Any]: + """Normalize a Polaris media item to a legacy-compatible node, preserving the source structure.""" + media_types = {1: "GraphImage", 2: "GraphVideo", 8: "GraphSidecar"} + media_type = media.get("media_type") + typename = None + if isinstance(media_type, int): + typename = media_types.get(media_type) + if not typename: + raise BadResponseException(f"Unknown media_type in metadata: {media_type}.") + pic_json: Dict[str, Any] = media.copy() + pic_json["shortcode"] = media["code"] + pic_json["id"] = media["pk"] + pic_json["__typename"] = typename + pic_json["is_video"] = media_type == 2 + pic_json["taken_at_timestamp"] = media["taken_at"] + pic_json["owner"] = { + "id": media["user"]["pk"], + "username": media["user"].get("username", ""), + "full_name": media["user"].get("full_name", ""), + } + candidates = (media.get("image_versions2") or {}).get("candidates") or [] + pic_json["display_url"] = candidates[0]["url"] if candidates else None + video_versions = media.get("video_versions") or [] + pic_json["video_url"] = video_versions[0]["url"] if video_versions else None + pic_json["video_duration"] = media.get("video_duration") + pic_json["video_view_count"] = media.get("view_count") + pic_json["video_play_count"] = media.get("play_count") + if media_type == 2 and pic_json["video_view_count"] is None: + pic_json["video_play_count"] = Post._fetch_play_count_from_clips( + context, media["user"]["pk"], media["code"] + ) + caption = media.get("caption") + caption_text = caption.get("text") if isinstance(caption, dict) else None + pic_json["edge_media_to_caption"] = ( + {"edges": [{"node": {"text": caption_text}}]} if caption_text is not None + else {"edges": []} + ) + pic_json["edge_media_preview_like"] = {"count": media.get("like_count") or 0} + pic_json["edge_media_to_parent_comment"] = { + "count": media.get("comment_count") or 0, + "edges": [], + } + if media.get("has_liked") is not None: + pic_json["viewer_has_liked"] = media["has_liked"] + carousel = media.get("carousel_media") or [] + if carousel: + carousel_nodes = [] + for item in carousel: + item_type = item.get("media_type", 1) + node: Dict[str, Any] = { + "shortcode": item.get("code", ""), + "__typename": media_types.get(item_type, "GraphImage"), + "is_video": item_type == 2, + } + item_candidates = (item.get("image_versions2") or {}).get("candidates") or [] + node["display_url"] = item_candidates[0]["url"] if item_candidates else "" + item_videos = item.get("video_versions") or [] + node["video_url"] = item_videos[0]["url"] if item_videos else None + if item.get("accessibility_caption") is not None: + node["accessibility_caption"] = item["accessibility_caption"] + carousel_nodes.append({"node": node}) + pic_json["edge_sidecar_to_children"] = {"edges": carousel_nodes} + tagged = (media.get("usertags") or {}).get("in") or [] + if tagged: + pic_json["edge_media_to_tagged_user"] = { + "edges": [ + {"node": {"user": {"username": t["user"]["username"].lower()}}} + for t in tagged + if (t.get("user") or {}).get("username") + ] + } + return pic_json + @staticmethod def shortcode_to_mediaid(code: str) -> int: if len(code) > 11: @@ -319,25 +413,18 @@ class Post: def _obtain_metadata(self): if not self._full_metadata_dict: - response = self._context.doc_id_graphql_query( - "8845758582119845", {"shortcode": self.shortcode} + resp = self._context.doc_id_graphql_query( + "27128499623469141", + { + "shortcode": self.shortcode, + "__relay_internal__pv__PolarisAIGMMediaWebLabelEnabledrelayprovider": False, + }, ) - data = (response or {}).get("data") - pic_json = data.get("xdt_shortcode_media") if data else None - if pic_json is None: + web_info = (resp.get("data") or {}).get("xdt_api__v1__media__shortcode__web_info") or {} + items = web_info.get("items") + if not items: raise BadResponseException("Fetching Post metadata failed.") - try: - xdt_types = { - "XDTGraphImage": "GraphImage", - "XDTGraphVideo": "GraphVideo", - "XDTGraphSidecar": "GraphSidecar", - } - pic_json["__typename"] = xdt_types[pic_json["__typename"]] - except KeyError as exc: - raise BadResponseException( - f"Unknown __typename in metadata: {pic_json['__typename']}." - ) from exc - self._full_metadata_dict = pic_json + self._full_metadata_dict = Post._normalize_post_data(items[0], self._context) if self.shortcode != self._full_metadata_dict['shortcode']: self._node.update(self._full_metadata_dict) raise PostChangedException