diff --git a/.pylintrc b/.pylintrc
index 9609887..5aed06b 100644
--- a/.pylintrc
+++ b/.pylintrc
@@ -93,10 +93,6 @@ py-version=3.8
# Discover python modules and packages in the file system subtree.
recursive=no
-# When enabled, pylint would attempt to guess common misconfiguration and emit
-# user-friendly hints instead of false-positive error messages.
-suggestion-mode=yes
-
# Allow loading of arbitrary C extensions. Extensions are imported into the
# active Python interpreter and may run arbitrary code.
unsafe-load-any-extension=no
diff --git a/instaloader/__init__.py b/instaloader/__init__.py
index 4c72039..c00ea98 100644
--- a/instaloader/__init__.py
+++ b/instaloader/__init__.py
@@ -1,7 +1,7 @@
"""Download pictures (or videos) along with their captions and other metadata from Instagram."""
-__version__ = '4.15.5'
+__version__ = '4.16rc1'
try:
diff --git a/instaloader/__main__.py b/instaloader/__main__.py
index 7b2670d..efe2f2f 100644
--- a/instaloader/__main__.py
+++ b/instaloader/__main__.py
@@ -16,7 +16,7 @@ from .instaloader import (get_default_session_filename, get_default_stamps_filen
from .instaloadercontext import default_user_agent
from .lateststamps import LatestStamps
try:
- import browser_cookie3
+ import browser_cookie3 # type: ignore[import-untyped]
bc3_library = True
except ImportError:
bc3_library = False
diff --git a/instaloader/instaloader.py b/instaloader/instaloader.py
index 2cbd606..40f9837 100644
--- a/instaloader/instaloader.py
+++ b/instaloader/instaloader.py
@@ -85,31 +85,36 @@ def _requires_login(func: Callable) -> Callable:
def _retry_on_connection_error(func: Callable) -> Callable:
"""Decorator to retry the function max_connection_attempts number of times.
- Herewith-decorated functions need an ``_attempt`` keyword argument.
-
This is to decorate functions that do network requests that may fail. Note that
:meth:`.get_json`, :meth:`.get_iphone_json`, :meth:`.graphql_query` and :meth:`.graphql_node_list` already have
their own logic for retrying, hence functions that only use these for network access must not be decorated with this
- decorator."""
+ decorator.
+
+ .. versionchanged:: 4.16
+ Decorated functions no longer need to accept an ``_attempt`` keyword argument.
+ """
@wraps(func)
- def call(instaloader, *args, **kwargs):
- try:
- return func(instaloader, *args, **kwargs)
- except (urllib3.exceptions.HTTPError, requests.exceptions.RequestException, ConnectionException) as err:
- error_string = "{}({}): {}".format(func.__name__, ', '.join([repr(arg) for arg in args]), err)
- if (kwargs.get('_attempt') or 1) == instaloader.context.max_connection_attempts:
- raise ConnectionException(error_string) from None
- instaloader.context.error(error_string + " [retrying; skip with ^C]", repeat_at_end=False)
+ def call(instaloader, *args, **kwargs) -> Any:
+ # The failure we are currently retrying, if any. It tells us whether a ^C means "skip this
+ # download" rather than "abort everything", and whether to sleep before the next attempt.
+ pending_error_string: Optional[str] = None
+ for attempt in range(1, instaloader.context.max_connection_attempts + 1):
try:
- if kwargs.get('_attempt'):
- kwargs['_attempt'] += 1
- else:
- kwargs['_attempt'] = 2
- instaloader.context.do_sleep()
- return call(instaloader, *args, **kwargs)
+ if pending_error_string is not None:
+ instaloader.context.do_sleep()
+ return func(instaloader, *args, **kwargs)
+ except (urllib3.exceptions.HTTPError, requests.exceptions.RequestException, ConnectionException) as err:
+ error_string = "{}({}): {}".format(func.__name__, ', '.join([repr(arg) for arg in args]), err)
+ if attempt == instaloader.context.max_connection_attempts:
+ raise ConnectionException(error_string) from None
+ instaloader.context.error(error_string + " [retrying; skip with ^C]", repeat_at_end=False)
+ pending_error_string = error_string
except KeyboardInterrupt:
+ if pending_error_string is None:
+ raise
instaloader.context.error("[skipped by user]", repeat_at_end=False)
- raise ConnectionException(error_string) from None
+ raise ConnectionException(pending_error_string) from None
+ raise AssertionError("_retry_on_connection_error() retry loop exited without result")
return call
@@ -335,7 +340,7 @@ class Instaloader:
@_retry_on_connection_error
def download_pic(self, filename: str, url: str, mtime: datetime,
- filename_suffix: Optional[str] = None, _attempt: int = 1) -> bool:
+ filename_suffix: Optional[str] = None) -> bool:
"""Downloads and saves picture with given url under given directory with given timestamp.
Returns true, if file was actually downloaded, i.e. updated."""
if filename_suffix is not None:
@@ -519,8 +524,9 @@ class Instaloader:
'{0}_{1}_{2}.{3}'.format(target, identifier, name_suffix, extension))
@_retry_on_connection_error
- def download_title_pic(self, url: str, target: Union[str, Path], name_suffix: str, owner_profile: Optional[Profile],
- _attempt: int = 1) -> None:
+ def download_title_pic(
+ self, url: str, target: Union[str, Path], name_suffix: str, owner_profile: Optional[Profile]
+ ) -> None:
"""Downloads and saves a picture that does not have an association with a Post or StoryItem, such as a
Profile picture or a Highlight cover picture. Modification time is taken from the HTTP response headers.
@@ -963,6 +969,11 @@ class Instaloader:
.. versionchanged:: 4.3
Also downloads and saves the Highlight's cover pictures.
+ .. versionchanged:: 4.16
+ Add progress output for highlight retrieval, displaying a counter of the current
+ highlight number and the total number of highlights being processed. This provides
+ feedback similar to the per-item download counter already present in this function.
+
:param user: ID or Profile of the user whose highlights should get downloaded.
:param fast_update: If true, abort when first already-downloaded picture is encountered
:param filename_target: Replacement for {target} in dirname_pattern and filename_pattern
@@ -970,14 +981,20 @@ class Instaloader:
:param storyitem_filter: function(storyitem), which returns True if given StoryItem should be downloaded
:raises LoginRequiredException: If called without being logged in.
"""
- for user_highlight in self.get_highlights(user):
+ user_highlights = list(self.get_highlights(user))
+ hl_size = len(user_highlights)
+ for hl_number, user_highlight in enumerate(user_highlights, start=1):
name = user_highlight.owner_username
highlight_target: Union[str, Path] = (filename_target
if filename_target
else (Path(_PostPathFormatter.sanitize_path(name, self.sanitize_paths)) /
_PostPathFormatter.sanitize_path(user_highlight.title,
self.sanitize_paths)))
- self.context.log("Retrieving highlights \"{}\" from profile {}".format(user_highlight.title, name))
+ self.context.log(
+ "[{0:{w}d}/{1:{w}d}] Retrieving highlights \"{2}\" from profile {3}".format(
+ hl_number, hl_size, user_highlight.title, name, w=len(str(hl_size))
+ )
+ )
self.download_highlight_cover(user_highlight, highlight_target)
totalcount = user_highlight.itemcount
count = 1
diff --git a/instaloader/instaloadercontext.py b/instaloader/instaloadercontext.py
index 591f9d9..ac54868 100644
--- a/instaloader/instaloadercontext.py
+++ b/instaloader/instaloadercontext.py
@@ -102,7 +102,7 @@ class InstaloaderContext:
self.user_agent = user_agent if user_agent is not None else default_user_agent()
self.request_timeout = request_timeout
self._session = self.get_anonymous_session()
- self.username = None
+ self.username: Optional[str] = None
self.sleep = sleep
self.quiet = quiet
self.max_connection_attempts = max_connection_attempts
@@ -125,6 +125,9 @@ class InstaloaderContext:
# Cache profile from id (mapping from id to Profile)
self.profile_id_cache: Dict[int, Any] = dict()
+ # Tokens of the logged-in web page for api/graphql queries, by sessionid cookie
+ self._web_api_tokens_cache: Dict[str, Dict[str, str]] = dict()
+
@contextmanager
def anonymous_copy(self):
session = self._session
@@ -411,8 +414,102 @@ class InstaloaderContext:
f" when accessing {resp.url}"
)
+ @staticmethod
+ def _rate_limit_key(path: str, params: Dict[str, Any], host: str) -> str:
+ """Determine the :class:`RateController` bucket that a request is accounted to."""
+ if 'graphql/query' in path or path == 'api/graphql':
+ if 'query_hash' in params:
+ return params['query_hash']
+ if 'doc_id' in params:
+ return params['doc_id']
+ if host == 'i.instagram.com':
+ return 'iphone'
+ return 'other'
+
+ def _raise_if_fatal_status(self, resp: requests.Response, host: str, path: str) -> None:
+ if resp.status_code not in self.fatal_status_codes:
+ return
+ redirect = " redirect to {}".format(resp.headers['location']) if 'location' in resp.headers else ""
+ body = ""
+ if resp.headers.get('Content-Type', '').startswith('application/json'):
+ body = ': ' + resp.text[:500] + ('…' if len(resp.text) > 501 else '')
+ raise AbortDownloadException("Query to https://{}/{} responded with \"{} {}\"{}{}".format(
+ host, path, resp.status_code, resp.reason, redirect, body
+ ))
+
+ def _raise_if_login_redirect(self, resp: requests.Response) -> None:
+ """Turn a redirect to the login page into a meaningful exception.
+
+ Instagram answers requests it will not serve anonymously with a redirect to the login page
+ rather than with an error status, so this needs to be checked wherever we do not follow
+ redirects automatically.
+ """
+ if not resp.is_redirect:
+ return
+ redirect_url = resp.headers.get('location', '')
+ if not (redirect_url.startswith('https://www.instagram.com/accounts/login') or
+ redirect_url.startswith('https://i.instagram.com/accounts/login')):
+ return
+ if not self.is_logged_in:
+ raise LoginRequiredException("Redirected to login page. Use --login or --load-cookies.")
+ raise AbortDownloadException("Redirected to login page. You've been logged out, please wait " +
+ "some time, recreate the session and try again")
+
+ def _follow_redirects(self, sess: requests.Session, resp: requests.Response,
+ host: str, path: str, params: Dict[str, Any], use_post: bool) -> requests.Response:
+ """Follow redirects that stay on *host*, and reject those to the login page."""
+ # Redirects are followed with GET. For a POST request, `params` is the request body and must not be
+ # turned into query arguments.
+ redirect_params = None if use_post else params
+ while resp.is_redirect:
+ redirect_url = resp.headers['location']
+ self.log('\nHTTP redirect from https://{0}/{1} to {2}'.format(host, path, redirect_url))
+ self._raise_if_login_redirect(resp)
+ if not redirect_url.startswith('https://{}/'.format(host)):
+ break
+ resp = sess.get(redirect_url if redirect_url.endswith('/') else redirect_url + '/',
+ params=redirect_params, allow_redirects=False)
+ return resp
+
+ def _raise_for_status(self, resp: requests.Response) -> None:
+ """Translate an unsuccessful HTTP status code into the corresponding exception."""
+ if resp.status_code == 400:
+ with suppress(json.decoder.JSONDecodeError):
+ if resp.json().get("message") in [
+ "feedback_required",
+ "checkpoint_required",
+ "challenge_required",
+ ]:
+ # Raise AbortDownloadException in case of substantial Instagram
+ # requirements to stop producing more requests
+ raise AbortDownloadException(self._response_error(resp))
+ raise QueryReturnedBadRequestException(self._response_error(resp))
+ if resp.status_code == 404:
+ raise QueryReturnedNotFoundException(self._response_error(resp))
+ if resp.status_code == 429:
+ raise TooManyRequestsException(self._response_error(resp))
+ if resp.status_code != 200:
+ raise ConnectionException(self._response_error(resp))
+
+ def _raise_for_download_status(self, resp: requests.Response) -> None:
+ """Translate an unsuccessful status code of an anonymous download request into an exception.
+
+ This is deliberately a different policy from :meth:`_raise_for_status`, which applies to the
+ JSON API: a download may be answered with 403 for an invalid URL signature, and is never
+ answered with the 400 or 429 that the API uses.
+ """
+ if resp.status_code == 200:
+ return
+ if resp.status_code == 403:
+ # suspected invalid URL signature
+ raise QueryReturnedForbiddenException(self._response_error(resp))
+ if resp.status_code == 404:
+ # 404 not worth retrying.
+ raise QueryReturnedNotFoundException(self._response_error(resp))
+ raise ConnectionException(self._response_error(resp))
+
def get_json(self, path: str, params: Dict[str, Any], host: str = 'www.instagram.com',
- session: Optional[requests.Session] = None, _attempt=1,
+ session: Optional[requests.Session] = None,
response_headers: Optional[Dict[str, Any]] = None,
use_post: bool = False) -> Dict[str, Any]:
"""JSON request to Instagram.
@@ -432,96 +529,112 @@ class InstaloaderContext:
.. versionchanged:: 4.13
Added `use_post` parameter.
+
+ .. versionchanged:: 4.16
+ Removed the internal `_attempt` parameter; retries are no longer implemented by recursion.
"""
- is_graphql_query = 'query_hash' in params and 'graphql/query' in path
- is_doc_id_query = 'doc_id' in params and 'graphql/query' in path
- is_iphone_query = host == 'i.instagram.com'
- is_other_query = not is_graphql_query and not is_doc_id_query and host == "www.instagram.com"
- sess = session if session else self._session
- try:
- self.do_sleep()
- if is_graphql_query:
- self._rate_controller.wait_before_query(params['query_hash'])
- if is_doc_id_query:
- self._rate_controller.wait_before_query(params['doc_id'])
- if is_iphone_query:
- self._rate_controller.wait_before_query('iphone')
- if is_other_query:
- self._rate_controller.wait_before_query('other')
- if use_post:
- resp = sess.post('https://{0}/{1}'.format(host, path), data=params, allow_redirects=False)
- else:
- resp = sess.get('https://{0}/{1}'.format(host, path), params=params, allow_redirects=False)
- if resp.status_code in self.fatal_status_codes:
- redirect = " redirect to {}".format(resp.headers['location']) if 'location' in resp.headers else ""
- body = ""
- if resp.headers['Content-Type'].startswith('application/json'):
- body = ': ' + resp.text[:500] + ('…' if len(resp.text) > 501 else '')
- raise AbortDownloadException("Query to https://{}/{} responded with \"{} {}\"{}{}".format(
- host, path, resp.status_code, resp.reason, redirect, body
- ))
- while resp.is_redirect:
- redirect_url = resp.headers['location']
- self.log('\nHTTP redirect from https://{0}/{1} to {2}'.format(host, path, redirect_url))
- if (redirect_url.startswith('https://www.instagram.com/accounts/login') or
- redirect_url.startswith('https://i.instagram.com/accounts/login')):
- if not self.is_logged_in:
- raise LoginRequiredException("Redirected to login page. Use --login or --load-cookies.")
- raise AbortDownloadException("Redirected to login page. You've been logged out, please wait " +
- "some time, recreate the session and try again")
- if redirect_url.startswith('https://{}/'.format(host)):
- resp = sess.get(redirect_url if redirect_url.endswith('/') else redirect_url + '/',
- params=params, allow_redirects=False)
- else:
- break
- if response_headers is not None:
- response_headers.clear()
- response_headers.update(resp.headers)
- if resp.status_code == 400:
- with suppress(json.decoder.JSONDecodeError):
- if resp.json().get("message") in [
- "feedback_required",
- "checkpoint_required",
- "challenge_required",
- ]:
- # Raise AbortDownloadException in case of substantial Instagram
- # requirements to stop producing more requests
- raise AbortDownloadException(self._response_error(resp))
- raise QueryReturnedBadRequestException(self._response_error(resp))
- if resp.status_code == 404:
- raise QueryReturnedNotFoundException(self._response_error(resp))
- if resp.status_code == 429:
- raise TooManyRequestsException(self._response_error(resp))
- if resp.status_code != 200:
- raise ConnectionException(self._response_error(resp))
- else:
- resp_json = resp.json()
- if 'status' in resp_json and resp_json['status'] != "ok":
- raise ConnectionException(self._response_error(resp))
- return resp_json
- except (ConnectionException, json.decoder.JSONDecodeError, requests.exceptions.RequestException) as err:
- error_string = "JSON Query to {}: {}".format(path, err)
- if _attempt == self.max_connection_attempts:
- if isinstance(err, QueryReturnedNotFoundException):
- raise QueryReturnedNotFoundException(error_string) from err
- else:
- raise ConnectionException(error_string) from err
- self.error(error_string + " [retrying; skip with ^C]", repeat_at_end=False)
+ sess = session or self._session
+ rate_limit_key = self._rate_limit_key(path, params, host)
+ url = 'https://{0}/{1}'.format(host, path)
+
+ # The failure we are currently retrying, if any. It drives the 429 backoff performed before the
+ # next attempt, and tells us whether a ^C means "skip this query" rather than "abort everything".
+ pending_error: Optional[BaseException] = None
+ pending_error_string = ""
+
+ for attempt in range(1, self.max_connection_attempts + 1):
try:
- if isinstance(err, TooManyRequestsException):
- if is_graphql_query:
- self._rate_controller.handle_429(params['query_hash'])
- if is_doc_id_query:
- self._rate_controller.handle_429(params['doc_id'])
- if is_iphone_query:
- self._rate_controller.handle_429('iphone')
- if is_other_query:
- self._rate_controller.handle_429('other')
- return self.get_json(path=path, params=params, host=host, session=sess, _attempt=_attempt + 1,
- response_headers=response_headers)
+ if isinstance(pending_error, TooManyRequestsException):
+ self._rate_controller.handle_429(rate_limit_key)
+ self.do_sleep()
+ self._rate_controller.wait_before_query(rate_limit_key)
+
+ if use_post:
+ resp = sess.post(url, data=params, allow_redirects=False)
+ else:
+ resp = sess.get(url, params=params, allow_redirects=False)
+
+ self._raise_if_fatal_status(resp, host, path)
+ resp = self._follow_redirects(sess, resp, host, path, params, use_post)
+
+ if response_headers is not None:
+ response_headers.clear()
+ response_headers.update(resp.headers)
+
+ self._raise_for_status(resp)
+ resp_json = resp.json()
+ if 'status' in resp_json and resp_json['status'] != "ok":
+ raise ConnectionException(self._response_error(resp))
+ return resp_json
+ except (ConnectionException, json.decoder.JSONDecodeError, requests.exceptions.RequestException) as err:
+ error_string = "JSON Query to {}: {}".format(path, err)
+ if attempt == self.max_connection_attempts:
+ if isinstance(err, QueryReturnedNotFoundException):
+ raise QueryReturnedNotFoundException(error_string) from err
+ raise ConnectionException(error_string) from err
+ self.error(error_string + " [retrying; skip with ^C]", repeat_at_end=False)
+ pending_error, pending_error_string = err, error_string
except KeyboardInterrupt:
+ if pending_error is None:
+ raise
self.error("[skipped by user]", repeat_at_end=False)
- raise ConnectionException(error_string) from err
+ raise ConnectionException(pending_error_string) from pending_error
+
+ raise AssertionError("get_json() retry loop exited without result")
+
+ @contextmanager
+ def _graphql_session(self, referer: Optional[str], csrf: Optional[str] = None,
+ lsd: Optional[str] = None) -> Iterator[requests.Session]:
+ """Short-lived session carrying the headers that the GraphQL endpoint expects.
+
+ Pass the page's `lsd` token for the web app's ``api/graphql`` endpoint."""
+ 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']
+ del tmpsession.headers['Content-Length']
+ tmpsession.headers['authority'] = 'www.instagram.com'
+ tmpsession.headers['scheme'] = 'https'
+ tmpsession.headers['accept'] = '*/*'
+ if csrf is not None:
+ tmpsession.headers['x-csrftoken'] = csrf
+ if referer is not None:
+ tmpsession.headers['referer'] = urllib.parse.quote(referer)
+ if lsd is not None:
+ tmpsession.headers.update({'x-fb-lsd': lsd, 'sec-fetch-dest': 'empty', 'sec-fetch-mode': 'cors',
+ 'sec-fetch-site': 'same-origin'})
+ yield tmpsession
+
+ def _csrf_token(self) -> str:
+ """The csrftoken cookie value, requesting the start page once if we do not have one yet."""
+ def token_from_cookies() -> Optional[str]:
+ return next((c.value for c in self._session.cookies if c.name == 'csrftoken' and c.value), None)
+ csrf = token_from_cookies()
+ if csrf:
+ return csrf
+ self._session.get('https://www.instagram.com/', timeout=self.request_timeout)
+ return token_from_cookies() or ''
+
+ def _web_api_tokens(self) -> Dict[str, str]:
+ """The tokens that the web app sends along with its api/graphql queries, from the start page."""
+ sessionid = self._session.cookies.get_dict().get("sessionid", "")
+ if sessionid not in self._web_api_tokens_cache:
+ html = self._get_page(self._session, "")
+ tokens = {}
+ for name, pattern in [
+ ("fb_dtsg", r'"DTSGInitialData",\[\],\{[^{}]*"token":"([^"]+)"'),
+ ("lsd", r'"LSD",\[\],\{[^{}]*"token":"([^"]+)"'),
+ ("av", r'"IG_USER_EIMU":"?(\d+)'),
+ ]:
+ match = re.search(pattern, html)
+ if match is not None:
+ tokens[name] = match.group(1)
+ # Logged out, the start page has no fb_dtsg, and the web app sends its queries without.
+ elif name != "fb_dtsg" or self.is_logged_in:
+ raise BadResponseException(
+ "The start page did not contain the token {}.".format(name)
+ )
+ self._web_api_tokens_cache[sessionid] = tokens
+ return self._web_api_tokens_cache[sessionid]
def graphql_query(self, query_hash: str, variables: Dict[str, Any],
referer: Optional[str] = None) -> Dict[str, Any]:
@@ -536,18 +649,8 @@ class InstaloaderContext:
.. versionchanged:: 4.13.1
Removed the `rhx_gis` parameter.
"""
- 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']
- del tmpsession.headers['Content-Length']
- tmpsession.headers['authority'] = 'www.instagram.com'
- tmpsession.headers['scheme'] = 'https'
- tmpsession.headers['accept'] = '*/*'
- if referer is not None:
- tmpsession.headers['referer'] = urllib.parse.quote(referer)
-
- variables_json = json.dumps(variables, separators=(',', ':'))
-
+ variables_json = json.dumps(variables, separators=(',', ':'))
+ with self._graphql_session(referer) as tmpsession:
resp_json = self.get_json('graphql/query',
params={'query_hash': query_hash,
'variables': variables_json},
@@ -557,7 +660,7 @@ class InstaloaderContext:
return resp_json
def doc_id_graphql_query(self, doc_id: str, variables: Dict[str, Any],
- referer: Optional[str] = None) -> Dict[str, Any]:
+ referer: Optional[str] = None, web_api: bool = False) -> Dict[str, Any]:
"""
Do a doc_id-based GraphQL Query using method POST.
@@ -566,35 +669,23 @@ class InstaloaderContext:
:param doc_id: doc_id for the query.
:param variables: Variables for the Query.
:param referer: HTTP Referer, or None.
+ :param web_api: Query the web app's ``api/graphql`` endpoint, as the web app does, rather than
+ ``graphql/query``.
:return: The server's response dictionary.
+
+ .. versionchanged:: 4.16
+ Added `web_api` parameter.
"""
- 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']
- del tmpsession.headers['Content-Length']
- 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)
-
- variables_json = json.dumps(variables, separators=(',', ':'))
-
- resp_json = self.get_json('graphql/query',
- params={'variables': variables_json,
- 'doc_id': doc_id,
- 'server_timestamps': 'true'},
- session=tmpsession,
- use_post=True)
- if 'status' not in resp_json:
+ variables_json = json.dumps(variables, separators=(',', ':'))
+ params = {'variables': variables_json,
+ 'doc_id': doc_id,
+ 'server_timestamps': 'true',
+ **(self._web_api_tokens() if web_api else {})}
+ with self._graphql_session(referer, csrf=self._csrf_token(), lsd=params.get('lsd')) as tmpsession:
+ resp_json = self.get_json('api/graphql' if web_api else 'graphql/query', params=params,
+ session=tmpsession, use_post=True)
+ # api/graphql does not respond with a "status" field.
+ if not web_api and 'status' not in resp_json:
self.error("GraphQL response did not contain a \"status\" field.")
return resp_json
@@ -719,6 +810,25 @@ class InstaloaderContext:
return response
+ def _get_page(self, session: requests.Session, path: str) -> str:
+ """Load a page of ``www.instagram.com`` and return its HTML."""
+ 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:
+ resp = session.get(url, headers=headers, allow_redirects=False)
+ except requests.exceptions.RequestException as err:
+ raise ConnectionException("Page request to {}: {}".format(url, err)) from err
+ self._raise_if_login_redirect(resp)
+ if resp.status_code == 404:
+ raise QueryReturnedNotFoundException(self._response_error(resp))
+ if resp.status_code != 200:
+ raise ConnectionException(self._response_error(resp))
+ return resp.text
+
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.
@@ -730,23 +840,10 @@ class InstaloaderContext:
: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))
+ with self.get_anonymous_session() as anonymous_session:
+ html = self._get_page(anonymous_session, path)
data: List[Dict[str, Any]] = []
- for script in re.finditer(r'', resp.text, re.DOTALL):
+ for script in re.finditer(r'', html, re.DOTALL):
with suppress(json.decoder.JSONDecodeError):
data.extend(_embedded_query_data(json.loads(script.group(1))))
return data
@@ -763,27 +860,26 @@ class InstaloaderContext:
file.write(resp)
os.replace(filename + '.temp', filename)
- def get_raw(self, url: str, _attempt=1) -> requests.Response:
+ def get_raw(self, url: str) -> requests.Response:
"""Downloads a file anonymously.
+ This does not retry on its own; callers that need retrying are decorated with
+ ``_retry_on_connection_error``.
+
:raises QueryReturnedNotFoundException: When the server responds with a 404.
:raises QueryReturnedForbiddenException: When the server responds with a 403.
:raises ConnectionException: When download failed.
- .. versionadded:: 4.2.1"""
+ .. versionadded:: 4.2.1
+
+ .. versionchanged:: 4.16
+ Removed the `_attempt` parameter.
+ """
with self.get_anonymous_session() as anonymous_session:
resp = anonymous_session.get(url, stream=True)
- if resp.status_code == 200:
- resp.raw.decode_content = True
- return resp
- else:
- if resp.status_code == 403:
- # suspected invalid URL signature
- raise QueryReturnedForbiddenException(self._response_error(resp))
- if resp.status_code == 404:
- # 404 not worth retrying.
- raise QueryReturnedNotFoundException(self._response_error(resp))
- raise ConnectionException(self._response_error(resp))
+ self._raise_for_download_status(resp)
+ resp.raw.decode_content = True
+ return resp
def get_and_write_raw(self, url: str, filename: str) -> None:
"""Downloads and writes anonymously-requested raw data into a file.
@@ -804,16 +900,8 @@ class InstaloaderContext:
"""
with self.get_anonymous_session() as anonymous_session:
resp = anonymous_session.head(url, allow_redirects=allow_redirects)
- if resp.status_code == 200:
- return resp
- else:
- if resp.status_code == 403:
- # suspected invalid URL signature
- raise QueryReturnedForbiddenException(self._response_error(resp))
- if resp.status_code == 404:
- # 404 not worth retrying.
- raise QueryReturnedNotFoundException(self._response_error(resp))
- raise ConnectionException(self._response_error(resp))
+ self._raise_for_download_status(resp)
+ return resp
class RateController:
diff --git a/instaloader/nodeiterator.py b/instaloader/nodeiterator.py
index 94decae..5a01af8 100644
--- a/instaloader/nodeiterator.py
+++ b/instaloader/nodeiterator.py
@@ -69,6 +69,9 @@ class NodeIterator(Iterator[T]):
.. versionchanged: 4.13
Included support for `doc_id`-based queries (using POST method).
+
+ .. versionchanged: 4.16
+ Added `web_api` parameter, see :meth:`InstaloaderContext.doc_id_graphql_query`.
"""
_graphql_page_length = 12
@@ -83,10 +86,12 @@ class NodeIterator(Iterator[T]):
query_referer: Optional[str] = None,
first_data: Optional[Dict[str, Any]] = None,
is_first: Optional[Callable[[T, Optional[T]], bool]] = None,
- doc_id: Optional[str] = None):
+ doc_id: Optional[str] = None,
+ web_api: bool = False):
self._context = context
self._query_hash = query_hash
self._doc_id = doc_id
+ self._web_api = web_api
self._edge_extractor = edge_extractor
self._node_wrapper = node_wrapper
self._query_variables = query_variables if query_variables is not None else {}
@@ -110,15 +115,15 @@ class NodeIterator(Iterator[T]):
return self._query_query_hash(self._query_hash, after)
def _query_doc_id(self, doc_id: str, after: Optional[str] = None) -> Dict:
- pagination_variables: Dict[str, Any] = {'__relay_internal__pv__PolarisFeedShareMenurelayprovider': False}
+ pagination_variables: Dict[str, Any] = {}
if after is not None:
pagination_variables['after'] = after
pagination_variables['before'] = None
- pagination_variables['first'] = 12
+ pagination_variables['first'] = NodeIterator._graphql_page_length
pagination_variables['last'] = None
data = self._edge_extractor(
self._context.doc_id_graphql_query(
- doc_id, {**self._query_variables, **pagination_variables}, self._query_referer
+ doc_id, {**self._query_variables, **pagination_variables}, self._query_referer, self._web_api
)
)
self._best_before = datetime.now() + NodeIterator._shelf_life
diff --git a/instaloader/structures.py b/instaloader/structures.py
index d3ceb9d..2324869 100644
--- a/instaloader/structures.py
+++ b/instaloader/structures.py
@@ -279,7 +279,7 @@ class Post:
return None
@staticmethod
- def _normalize_post_data(media: Dict[str, Any], context: 'InstaloaderContext') -> Dict[str, Any]:
+ def _normalize_post_data(media: Dict[str, Any]) -> 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")
@@ -306,10 +306,6 @@ class Post:
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"] = (
@@ -419,12 +415,14 @@ class Post:
"shortcode": self.shortcode,
"__relay_internal__pv__PolarisAIGMMediaWebLabelEnabledrelayprovider": False,
},
+ # As the logged-out web app does.
+ web_api=not self._context.is_logged_in,
)
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.")
- self._full_metadata_dict = Post._normalize_post_data(items[0], self._context)
+ self._full_metadata_dict = Post._normalize_post_data(items[0])
if self.shortcode != self._full_metadata_dict['shortcode']:
self._node.update(self._full_metadata_dict)
raise PostChangedException
@@ -697,7 +695,11 @@ class Post:
.. versionadded:: 4.14.3"""
if self.is_video:
- return self._field('video_play_count')
+ play_count = self._field('video_play_count')
+ # Not logged in, the clips connection is null.
+ if play_count is None and self.video_view_count is None and self._context.is_logged_in:
+ play_count = Post._fetch_play_count_from_clips(self._context, str(self.owner_id), self.shortcode)
+ return play_count
return None
@property
@@ -1025,16 +1027,24 @@ class Profile:
"""
if profile_id in context.profile_id_cache:
return context.profile_id_cache[profile_id]
- # Resolve the current username from the user id, then load the full profile.
- # The GraphQL user query previously used here started responding with HTTP 400.
try:
- user = context.get_json(
- "api/v1/users/{0}/info/".format(profile_id), params={}
- ).get('user')
+ user = context.doc_id_graphql_query(
+ '27937681195819736',
+ {
+ "id": str(profile_id),
+ "render_surface": "PROFILE",
+ "__relay_internal__pv__PolarisCannesGuardianExperienceEnabledrelayprovider": True,
+ "__relay_internal__pv__PolarisCASB976ProfileEnabledrelayprovider": False,
+ "__relay_internal__pv__PolarisRepostsConsumptionEnabledrelayprovider": False,
+ "__relay_internal__pv__PolarisWebSchoolsEnabledrelayprovider": False,
+ "enable_integrity_filters": True,
+ },
+ web_api=True,
+ ).get('data', {}).get('user')
except QueryReturnedNotFoundException:
user = None
if user and user.get('username'):
- profile = cls.from_username(context, user['username'])
+ profile = cls(context, cls._normalize_profile_data(user))
else:
raise ProfileNotExistsException("No profile found, the user may have blocked you (ID: " +
str(profile_id) + ").")
@@ -1137,6 +1147,9 @@ class Profile:
normalized['id'] = normalized['pk']
if 'edge_owner_to_timeline_media' not in normalized and 'media_count' in normalized:
normalized['edge_owner_to_timeline_media'] = {'count': normalized['media_count']}
+ # The logged-out profile page has all_media_count instead.
+ if 'edge_owner_to_timeline_media' not in normalized and 'all_media_count' in normalized:
+ normalized['edge_owner_to_timeline_media'] = {'count': normalized['all_media_count']}
if 'edge_felix_video_timeline' not in normalized:
normalized['edge_felix_video_timeline'] = {'count': 0}
if 'edge_followed_by' not in normalized and 'follower_count' in normalized:
@@ -1222,7 +1235,11 @@ class Profile:
return self._metadata('followed_by_viewer')
@property
- def mediacount(self) -> int:
+ def mediacount(self) -> Optional[int]:
+ """Number of posts, or None if unavailable (e.g. when not logged in).
+
+ .. versionchanged:: 4.16
+ Returns None if unavailable (e.g. when not logged in)."""
return self._metadata('edge_owner_to_timeline_media', 'count')
@property
@@ -1374,35 +1391,53 @@ class Profile:
def get_posts(self) -> NodeIterator[Post]:
"""Retrieve all posts from a profile.
- :rtype: NodeIterator[Post]"""
- self._obtain_metadata()
+ :rtype: NodeIterator[Post]
+ :raises PrivateProfileNotFollowedException: When logged in and the profile is private and not followed.
+ :raises LoginRequiredException: When not logged in and the profile is private."""
if not self._context.is_logged_in:
+ # As the logged-out web app, take the first posts from the profile page, and query further ones
+ # (PolarisLoggedOutDesktopWWWProfilePostsTabContentQuery_connection) by the page's "id" of the
+ # profile, which is not its user ID.
+ page = self._context.get_page_data(f"{self.username}/")
+ # The page shows no posts of a private profile, rather than an error.
+ if any((data.get("xig_user_by_username") or {}).get("is_private") for data in page):
+ raise LoginRequiredException(f"Login required to access the posts of private profile {self.username}.")
+ user = next((data["xig_user_by_username"] for data in page
+ if "polaris_ordered_timeline_connection" in (data.get("xig_user_by_username") or {})),
+ None)
+ if user is None:
+ raise BadResponseException(f"The page of profile {self.username} did not contain its posts.")
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),
+ edge_extractor=lambda d: d["data"]["node"]["polaris_ordered_timeline_connection"],
+ # The nodes lack most of the post's metadata, as in get_reels().
+ node_wrapper=lambda n: Post.from_shortcode(self._context, n["code"]),
query_variables={
- "data": {
- "count": 12,
- "include_relationship_info": True,
- "latest_besties_reel_media": True,
- "latest_reel_media": True,
- },
- "id": self.userid,
+ "id": user["id"],
+ "__relay_internal__pv__PolarisSEOIsCrawlerrelayprovider": False,
},
query_referer="https://www.instagram.com/{0}/".format(self.username),
is_first=Profile._make_is_newest_checker(),
- doc_id="7950326061742207",
+ doc_id="28904005159232689",
query_hash=None,
- first_data=self._metadata("edge_owner_to_timeline_media"),
+ first_data=user["polaris_ordered_timeline_connection"],
+ web_api=True,
)
+ self._obtain_metadata()
+ # Instagram answers with an empty timeline rather than an error in that case. Unlike followed_by_viewer,
+ # which defaults to False, friendship_status tells whether the follow state is actually known.
+ if (self.is_private and self.username != (self._context.username or '').lower() and
+ (self._node.get('friendship_status') or {}).get('following') is False):
+ raise PrivateProfileNotFollowedException("Profile {} is private but not followed.".format(self.username))
return NodeIterator(
context=self._context,
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".
+ # previous queries, 7898261790222653 ("execution error") and 28975909992013618
+ # (400 "feedback_required"), with errors.
+ # It must be sent to the web app's api/graphql endpoint.
query_variables={
"data": {
"count": 12,
@@ -1415,13 +1450,16 @@ class Profile:
"include_multi_captions": True,
"username": self.username,
"__relay_internal__pv__PolarisMultiCaptionCarouselEnabledrelayprovider": True,
- "__relay_internal__pv__PolarisShortDramaEnabledrelayprovider": False,
+ "__relay_internal__pv__PolarisShortDramaEnabledrelayprovider": True,
+ "__relay_internal__pv__PolarisSEOIsCrawlerrelayprovider": False,
+ "__relay_internal__pv__PolarisThreadsStyleCommentsEnabledrelayprovider": 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="28975909992013618",
+ doc_id="29104886249145907",
query_hash=None,
+ web_api=True,
)
def get_saved_posts(self) -> NodeIterator[Post]: