mirror/instaloader
1
0
Fork 0
mirror of https://github.com/instaloader/instaloader.git synced 2026-10-16 13:48:03 -07:00
Code Issues Packages Projects Releases Wiki Activity

Fix profile resolution and logged-in post fetching (#2765)

Browse Source 
- `InstaloaderContext.get_page_data()` loads a page as a logged-out visitor and returns the query results that the page embeds. Instagram only embeds them when the request looks like a page navigation, so the request sends the `Sec-Fetch-*` headers of one.
- `Profile.from_username()` reads the profile from its public page. Anonymously, `_normalize_profile_data()` completes the node.
- Logged in, `Profile.get_posts()` uses query `28975909992013618` (`PolarisProfilePostsTabContentQuery_connection`) with the variables that the web app sends. The response has the same shape as before.

When logged in, `from_username()` does not normalize the page's node. The page shows the profile to a logged-out visitor, so normalizing would set `followed_by_viewer` to `False`, and `download_profiles()` would refuse private profiles that the session follows. Instead, the GraphQL profile query in `_obtain_metadata()` fills in the missing fields when they are first read.
This commit is contained in:
Markus Magnuson
2026-10-12 21:39:03 +02:00
committed by GitHub
parent 91e30cf33b
commit 0cf00a6079
2 changed files with 103 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
instaloader/instaloadercontext.py
+46
View File
@@ -2,6 +2,7 @@ import json
import os
import pickle
import random
import re
import shutil
import sys
import textwrap
@@ -30,6 +31,19 @@ def copy_session(session: requests.Session, request_timeout: Optional[float] = N
return new
def _embedded_query_data(obj: Any) -> Iterator[Dict[str, Any]]:
"""Yield the data of each GraphQL query result in JSON that a page of www.instagram.com embeds."""
if isinstance(obj, dict):
result = obj['__bbox'].get('result') if isinstance(obj.get('__bbox'), dict) else None
if isinstance(result, dict) and isinstance(result.get('data'), dict):
yield result['data']
for value in obj.values():
yield from _embedded_query_data(value)
elif isinstance(obj, list):
for value in obj:
yield from _embedded_query_data(value)
def default_user_agent() -> str:
return ('Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 '
'(KHTML, like Gecko) Chrome/142.0.0.0 Safari/537.36')
@@ -674,6 +688,38 @@ class InstaloaderContext:
return response
def get_page_data(self, path: str) -> List[Dict[str, Any]]:
"""Load a page of ``www.instagram.com`` as a logged-out visitor and return the data it embeds.
Instagram renders public profile and post pages with the results of their GraphQL queries
embedded in the HTML, even when it refuses anonymous API requests for the same data.
:param path: URL, relative to ``www.instagram.com/``
:return: The ``data`` object of each query result that the page embeds, in page order.
:raises QueryReturnedNotFoundException: When the server responds with a 404.
:raises ConnectionException: When the request failed.
"""
self.do_sleep()
self._rate_controller.wait_before_query('other')
url = 'https://www.instagram.com/{0}'.format(path)
# Instagram only embeds the data when the request looks like a browser's page navigation.
headers = {'Accept': 'text/html', 'Sec-Fetch-Dest': 'document', 'Sec-Fetch-Mode': 'navigate',
'Sec-Fetch-Site': 'none'}
try:
with self.get_anonymous_session() as anonymous_session:
resp = anonymous_session.get(url, headers=headers, allow_redirects=False)
except requests.exceptions.RequestException as err:
raise ConnectionException("Page request to {}: {}".format(url, err)) from err
if resp.status_code == 404:
raise QueryReturnedNotFoundException(self._response_error(resp))
if resp.status_code != 200:
raise ConnectionException(self._response_error(resp))
data: List[Dict[str, Any]] = []
for script in re.finditer(r'<script type="application/json"[^>]*>(.*?)</script>', resp.text, re.DOTALL):
with suppress(json.decoder.JSONDecodeError):
data.extend(_embedded_query_data(json.loads(script.group(1))))
return data
def write_raw(self, resp: Union[bytes, requests.Response], filename: str) -> None:
"""Write raw response data into a file.
instaloader/structures.py
+57 -29
View File
@@ -999,22 +999,30 @@ class Profile:
:param username: Username
:raises: :class:`ProfileNotExistsException`
"""
# Resolve the profile through the web_profile_info endpoint, which works both
# anonymously and when logged in and returns the complete profile node
# (including the first page of posts). The GraphQL fbsearch query previously
# used here started responding with HTTP 400.
# Resolve the profile through its public web page, which Instagram renders with the
# profile embedded. The web_profile_info endpoint previously used here is refused
# with HTTP 400 for some professional profiles, and with HTTP 401 or 429 for some
# sessions, both anonymously and when logged in.
try:
data = context.get_json(
"api/v1/users/web_profile_info/", params={"username": username.lower()}
).get("data")
page = context.get_page_data("{0}/".format(username.lower()))
except QueryReturnedNotFoundException:
data = None
if data and data.get("user"):
profile = cls(context, data["user"])
profile._has_full_metadata = True
return profile
raise ProfileNotExistsException("Profile {} does not exist.".format(username))
page = []
# The page embeds two queries for the profile: one for its details and one for its
# posts.
user: Dict[str, Any] = {}
for data in page:
user.update(data.get("xig_user_by_username") or {})
if not user.get("pk"):
raise ProfileNotExistsException("Profile {} does not exist.".format(username))
user.pop("polaris_ordered_timeline_connection", None)
# The page's "id" is not the user ID, which is "pk".
node = dict(user, id=user["pk"])
if context.is_logged_in:
# The page shows the profile as a logged-out visitor sees it. Leave the missing
# fields, notably the viewer's relationship to the profile, such as
# followed_by_viewer, to the GraphQL profile query in _obtain_metadata().
return cls(context, node)
return cls(context, cls._normalize_profile_data(node))
@classmethod
def from_id(cls, context: InstaloaderContext, profile_id: int):
@@ -1131,7 +1139,8 @@ class Profile:
', '.join(similar_profiles[0:5]))) from err
raise ProfileNotExistsException('Profile {} does not exist.'.format(self.username)) from err
def _normalize_profile_data(self, user_data: Dict[str, Any]) -> Dict[str, Any]:
@staticmethod
def _normalize_profile_data(user_data: Dict[str, Any]) -> Dict[str, Any]:
"""Normalize PolarisProfilePageContentQuery response to match legacy format."""
normalized = user_data.copy()
if 'id' not in normalized and 'pk' in normalized:
@@ -1377,33 +1386,52 @@ class Profile:
:rtype: NodeIterator[Post]"""
self._obtain_metadata()
logged_in = self._context.is_logged_in
if not self._context.is_logged_in:
return NodeIterator(
context=self._context,
edge_extractor=lambda d: d["data"]["user"]["edge_owner_to_timeline_media"],
node_wrapper=lambda n: Post(self._context, n, self),
query_variables={
"data": {
"count": 12,
"include_relationship_info": True,
"latest_besties_reel_media": True,
"latest_reel_media": True,
},
"id": self.userid,
},
query_referer="https://www.instagram.com/{0}/".format(self.username),
is_first=Profile._make_is_newest_checker(),
doc_id="7950326061742207",
query_hash=None,
first_data=self._metadata("edge_owner_to_timeline_media"),
)
return NodeIterator(
context=self._context,
edge_extractor=(
(lambda d: d["data"]["xdt_api__v1__feed__user_timeline_graphql_connection"])
if logged_in
else (lambda d: d["data"]["user"]["edge_owner_to_timeline_media"])
),
node_wrapper=(
(lambda n: Post.from_iphone_struct(self._context, n))
if logged_in
else (lambda n: Post(self._context, n, self))
),
edge_extractor=lambda d: d["data"]["xdt_api__v1__feed__user_timeline_graphql_connection"],
node_wrapper=lambda n: Post.from_iphone_struct(self._context, n),
# The query and variables that the web app sends for a profile's posts tab
# (PolarisProfilePostsTabContentQuery_connection). Instagram answers the
# previous query, 7898261790222653, with "execution error".
query_variables={
"data": {
"count": 12,
"include_reel_media_seen_timestamp": True,
"include_relationship_info": True,
"latest_besties_reel_media": True,
"latest_reel_media": True,
},
**({"username": self.username} if logged_in else {"id": self.userid}),
"first": 12,
"include_multi_captions": True,
"username": self.username,
"__relay_internal__pv__PolarisMultiCaptionCarouselEnabledrelayprovider": True,
"__relay_internal__pv__PolarisShortDramaEnabledrelayprovider": False,
"__relay_internal__pv__PolarisReelsRecoDebugOverlayEnabledrelayprovider": False,
},
query_referer="https://www.instagram.com/{0}/".format(self.username),
is_first=Profile._make_is_newest_checker(),
doc_id="7898261790222653" if logged_in else "7950326061742207",
doc_id="28975909992013618",
query_hash=None,
first_data=(None if logged_in else self._metadata("edge_owner_to_timeline_media")),
)
def get_saved_posts(self) -> NodeIterator[Post]: