Merge branch 'master' into upcoming/v4.16

This commit is contained in:
Alexander Graf
2026-10-12 22:07:09 +02:00
5 changed files with 122 additions and 34 deletions
+10 -4
View File
@@ -2,17 +2,23 @@
<!--
Thanks for your willingness to contribute to Instaloader! Please briefly
describe
describe:
- A motivation for this change, e.g.
- Fixes # .
- More general: What problem does the pull request solve?
- What problem does the pull request solve?
- The changes proposed in this pull request
- Keep the PR focused: avoid bundling independent fixes or refactorings.
- The completeness of this change
- Is it just a proof of concept?
- Is the documentation updated (if appropriate)?
- Do you consider it ready to be merged or is it a draft?
- How has it been tested?
- Do you consider it ready to be merged or is it a draft / proof of concept?
- Can we help you at some point?
- Further considerations
- Is it based on master (bugfix), upcoming/v4.X (new feature), or another pull request?
- Does it preserve backward compatibility and existing behavior?
-->
+8
View File
@@ -74,6 +74,14 @@ Improvements of the Instaloader source or its documentation can be proposed as a
- a bug fix that does require thorough testing before being released to a
final version.
- Keep Pull Requests focused: address a single problem or feature in a
self-contained change. Avoid bundling multiple independent fixes, refactorings,
or architectural changes into one pull request, as this makes it hard to
review, validate, or selectively merge them.
- Be mindful of backward compatibility: changes should preserve existing intended
behavior whenever possible.
- All Pull Requests are analyzed by `Pylint <https://www.pylint.org/>`__ for
error and syntax checking of the source and
`Mypy <https://github.com/python/mypy>`__ for type checking. You can run them
+1 -1
View File
@@ -1,7 +1,7 @@
"""Download pictures (or videos) along with their captions and other metadata from Instagram."""
__version__ = '4.15.3'
__version__ = '4.15.4'
try:
+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.
+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]: