Release of Version 4.10

Fixes #1905.

.github/workflows/documentation.yml

@@ -17,10 +17,10 @@ jobs:
           python-version: 3.8
       - name: "Install Dependencies"
         run: |
-          python -m pip install pipenv==2022.1.8
-          pipenv sync --dev
+          python -m pip install pipenv==2023.2.4
+          pipenv --python `python --version | grep -Eo '3\.[0-9]+'` sync --dev
       - name: "Build Documentation"
-        run: pipenv run make -C docs html SPHINXOPTS="-W -n"
+        run: pipenv --python `python --version | grep -Eo '3\.[0-9]+'` run make -C docs html SPHINXOPTS="-W -n"
       - name: "Deploy Documentation"
         if: github.event_name == 'push' && github.ref == 'refs/heads/master'
         uses: JamesIves/github-pages-deploy-action@v4.2.5

.github/workflows/lint.yml

@@ -7,8 +7,9 @@ jobs:
     name: PyLint and MyPy
     runs-on: ubuntu-latest
     strategy:
+      fail-fast: false
       matrix:
-        python-version: ["3.6", "3.7", "3.8", "3.9", "3.10"]
+        python-version: ["3.8", "3.9", "3.10", "3.11"]
     steps:
       - name: Checkout Instaloader Repository
         uses: actions/checkout@v2
@@ -18,9 +19,9 @@ jobs:
           python-version: ${{ matrix.python-version }}
       - name: Install Dependencies
         run: |
-          python -m pip install pipenv==2022.1.8
-          pipenv sync --dev
+          python -m pip install pipenv==2023.2.4
+          pipenv --python `python --version | grep -Eo '3\.[0-9]+'` sync --dev
       - name: PyLint
-        run: pipenv run pylint instaloader
+        run: pipenv --python `python --version | grep -Eo '3\.[0-9]+'` run pylint instaloader
       - name: MyPy
-        run: pipenv run mypy -m instaloader
+        run: pipenv --python `python --version | grep -Eo '3\.[0-9]+'` run mypy -m instaloader

Pipfile

@@ -14,6 +14,10 @@ psutil = "*"
 typing-extensions = "*"
 types-requests = "*"
 typed-ast = "*"
+jinja2 = "~=3.0.1"
+wrapt = "*"  # Needed for Python<3.11
+tomli = "*"  # Needed for Python<3.11
+dill = "*"  # Needed for Python<3.11
 
 [packages]
 requests = "*"

README.rst

@@ -2,11 +2,7 @@
 
 .. badges-start
 
-|travis| |pypi| |pyversion| |license| |aur| |contributors| |downloads|
-
-.. |travis| image:: https://img.shields.io/travis/instaloader/instaloader/master.svg
-   :alt: Travis-CI Build Status
-   :target: https://travis-ci.org/instaloader/instaloader
+|pypi| |pyversion| |license| |aur| |contributors| |downloads|
 
 .. |pypi| image:: https://img.shields.io/pypi/v/instaloader.svg
    :alt: Instaloader PyPI Project Page
@@ -128,6 +124,7 @@ Supporters
 .. current-sponsors-start
 
 | Instaloader is proudly sponsored by
+|  `@rocketapi-io <https://github.com/rocketapi-io>`__
 |  `@socialmethod <https://github.com/socialmethod>`__
 
 See `Alex' GitHub Sponsors <https://github.com/sponsors/aandergr>`__ page for

deploy/arch/.SRCINFO.template

@@ -6,7 +6,7 @@ pkgbase = instaloader
 	arch = any
 	license = MIT
 	makedepends = python-setuptools
-	depends = python>=3.6
+	depends = python>=3.8
 	depends = python-requests>=2.4
 	options = !emptydirs
 	source = instaloader-{{version}}.tar.gz::https://codeload.github.com/instaloader/instaloader/tar.gz/v{{version}}

deploy/arch/PKGBUILD.template

@@ -8,7 +8,7 @@ arch=('any')
 url="https://instaloader.github.io/"
 license=('MIT')
 groups=()
-depends=('python>=3.6' 'python-requests>=2.4')
+depends=('python>=3.8' 'python-requests>=2.4')
 makedepends=('python-setuptools')
 options=('!emptydirs')
 source=($pkgname-$pkgver.tar.gz::https://codeload.github.com/instaloader/instaloader/tar.gz/v$pkgver)

deploy/windows/create_exe.py

@@ -44,7 +44,7 @@ with open('__main__.py', 'w+') as f:
     f.writelines(lines)
 
 # install dependencies and invoke PyInstaller
-commands = ["pip install pipenv==2022.1.8",
+commands = ["pip install pipenv==2023.2.4",
             f"pipenv --python {sys.version_info.major}.{sys.version_info.minor} sync --dev",
             "pipenv run pyinstaller --log-level=DEBUG instaloader.spec"]
 

docs/installation.rst

@@ -8,7 +8,7 @@ Install Instaloader
 To **install Instaloader**,
 
 #. Ensure that you have `Python <https://www.python.org/>`__, at least
-   version 3.6, and `pip <https://pypi.python.org/pypi/pip>`__
+   version 3.8, and `pip <https://pypi.python.org/pypi/pip>`__
    installed.
 
 #. Then, install Instaloader using::

docs/module/nodeiterator.rst

@@ -30,7 +30,7 @@ Iterator :class:`NodeIterator` and a context manager
 
    It can be serialized and deserialized with :func:`save_structure_to_file`
    and :func:`load_structure_from_file`, as well as with :mod:`json` and
-   :mod:`pickle` thanks to being a :func:`~collections.namedtuple`.
+   :mod:`pickle` thanks to being a :class:`~typing.NamedTuple`.
 
 ``resumable_iteration``
 """""""""""""""""""""""

instaloader/__init__.py

@@ -1,7 +1,7 @@
 """Download pictures (or videos) along with their captions and other metadata from Instagram."""
 
 
-__version__ = '4.9.5'
+__version__ = '4.10'
 
 
 try:

instaloader/__main__.py

@@ -45,7 +45,6 @@ def filterstr_to_filterfunc(filter_str: str, item_type: type):
 
     class TransformFilterAst(ast.NodeTransformer):
         def visit_Name(self, node: ast.Name):
-            # pylint:disable=no-self-use
             if not isinstance(node.ctx, ast.Load):
                 raise InvalidArgumentException("Invalid filter: Modifying variables ({}) not allowed.".format(node.id))
             if node.id == "datetime":

instaloader/instaloader.py

@@ -221,8 +221,8 @@ class Instaloader:
                  download_comments: bool = False,
                  save_metadata: bool = True,
                  compress_json: bool = True,
-                 post_metadata_txt_pattern: str = None,
-                 storyitem_metadata_txt_pattern: str = None,
+                 post_metadata_txt_pattern: Optional[str] = None,
+                 storyitem_metadata_txt_pattern: Optional[str] = None,
                  max_connection_attempts: int = 3,
                  request_timeout: float = 300.0,
                  rate_controller: Optional[Callable[[InstaloaderContext], RateController]] = None,
@@ -447,7 +447,7 @@ class Instaloader:
         """Updates picture caption / Post metadata info"""
         def _elliptify(caption):
             pcaption = caption.replace('\n', ' ').strip()
-            return '[' + ((pcaption[:29] + u"\u2026") if len(pcaption) > 31 else pcaption) + ']'
+            return '[' + ((pcaption[:29] + "\u2026") if len(pcaption) > 31 else pcaption) + ']'
         filename += '.txt'
         caption += '\n'
         pcaption = _elliptify(caption)
@@ -524,7 +524,7 @@ class Instaloader:
         .. versionadded:: 4.3"""
 
         http_response = self.context.get_raw(url)
-        date_object = None  # type: Optional[datetime]
+        date_object: Optional[datetime] = None
         if 'Last-Modified' in http_response.headers:
             date_object = datetime.strptime(http_response.headers["Last-Modified"], '%a, %d %b %Y %H:%M:%S GMT')
             date_object = date_object.replace(tzinfo=timezone.utc)
@@ -585,6 +585,23 @@ class Instaloader:
         .. versionadded:: 4.4"""
         self.download_title_pic(hashtag.profile_pic_url, '#' + hashtag.name, 'profile_pic', None)
 
+    @_requires_login
+    def save_session(self) -> dict:
+        """Saves internally stored :class:`requests.Session` object to :class:`dict`.
+
+        :raises LoginRequiredException: If called without being logged in.
+
+        .. versionadded:: 4.10
+        """
+        return self.context.save_session()
+
+    def load_session(self, username: str, session_data: dict) -> None:
+        """Internally stores :class:`requests.Session` object from :class:`dict`.
+
+        .. versionadded:: 4.10
+        """
+        self.context.load_session(username, session_data)
+
     @_requires_login
     def save_session_to_file(self, filename: Optional[str] = None) -> None:
         """Saves internally stored :class:`requests.Session` object.
@@ -712,7 +729,7 @@ class Instaloader:
                             post.get_sidecar_nodes(self.slide_start, self.slide_end),
                             start=self.slide_start % post.mediacount + 1
                     ):
-                        suffix = str(edge_number)  # type: Optional[str]
+                        suffix: Optional[str] = str(edge_number)
                         if '{filename}' in self.filename_pattern:
                             suffix = None
                         if self.download_pictures and (not sidecar_node.is_video or self.download_video_thumbnails):
@@ -945,11 +962,11 @@ class Instaloader:
         """
         for user_highlight in self.get_highlights(user):
             name = user_highlight.owner_username
-            highlight_target = (filename_target
+            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)))  # type: Union[str, Path]
+                                                                       self.sanitize_paths)))
             self.context.log("Retrieving highlights \"{}\" from profile {}".format(user_highlight.title, name))
             self.download_highlight_cover(user_highlight, highlight_target)
             totalcount = user_highlight.itemcount
@@ -1080,7 +1097,7 @@ class Instaloader:
                                                'has_stories': False})["data"]
 
     @_requires_login
-    def download_feed_posts(self, max_count: int = None, fast_update: bool = False,
+    def download_feed_posts(self, max_count: Optional[int] = None, fast_update: bool = False,
                             post_filter: Optional[Callable[[Post], bool]] = None) -> None:
         """
         Download pictures from the user's feed.
@@ -1101,7 +1118,7 @@ class Instaloader:
         self.posts_download_loop(self.get_feed_posts(), ":feed", fast_update, post_filter, max_count=max_count)
 
     @_requires_login
-    def download_saved_posts(self, max_count: int = None, fast_update: bool = False,
+    def download_saved_posts(self, max_count: Optional[int] = None, fast_update: bool = False,
                              post_filter: Optional[Callable[[Post], bool]] = None) -> None:
         """Download user's saved pictures.
 

instaloader/instaloadercontext.py

@@ -9,6 +9,7 @@ import sys
 import textwrap
 import time
 import urllib.parse
+import uuid
 from contextlib import contextmanager
 from datetime import datetime, timedelta
 from functools import partial
@@ -24,10 +25,10 @@ def copy_session(session: requests.Session, request_timeout: Optional[float] = N
     """Duplicates a requests.Session."""
     new = requests.Session()
     new.cookies = requests.utils.cookiejar_from_dict(requests.utils.dict_from_cookiejar(session.cookies))
-    new.headers = session.headers.copy()
+    new.headers = session.headers.copy()  # type: ignore
     # Override default timeout behavior.
     # Need to silence mypy bug for this. See: https://github.com/python/mypy/issues/2427
-    new.request = partial(new.request, timeout=request_timeout) # type: ignore
+    new.request = partial(new.request, timeout=request_timeout)  # type: ignore
     return new
 
 
@@ -36,6 +37,34 @@ def default_user_agent() -> str:
            '(KHTML, like Gecko) Chrome/92.0.4515.131 Safari/537.36'
 
 
+def default_iphone_headers() -> Dict[str, Any]:
+    return {'User-Agent': 'Instagram 273.0.0.16.70 (iPad13,8; iOS 16_3; en_US; en-US; ' \
+                          'scale=2.00; 2048x2732; 452417278) AppleWebKit/420+',
+            'x-ads-opt-out': '1',
+            'x-bloks-is-panorama-enabled': 'true',
+            'x-bloks-version-id': '01507c21540f73e2216b6f62a11a5b5e51aa85491b72475c080da35b1228ddd6',
+            'x-fb-client-ip': 'True',
+            'x-fb-connection-type': 'wifi',
+            'x-fb-http-engine': 'Liger',
+            'x-fb-server-cluster': 'True',
+            'x-fb': '1',
+            'x-ig-abr-connection-speed-kbps': '2',
+            'x-ig-app-id': '124024574287414',
+            'x-ig-app-locale': 'en-US',
+            'x-ig-app-startup-country': 'US',
+            'x-ig-bandwidth-speed-kbps': '0.000',
+            'x-ig-capabilities': '36r/F/8=',
+            'x-ig-connection-speed': '{}kbps'.format(random.randint(1000, 20000)),
+            'x-ig-connection-type': 'WiFi',
+            'x-ig-device-locale': 'en-US',
+            'x-ig-mapped-locale': 'en-US',
+            'x-ig-timezone-offset': str((datetime.now().astimezone().utcoffset() or timedelta(seconds=0)).seconds),
+            'x-ig-www-claim': '0',
+            'x-pigeon-session-id': str(uuid.uuid4()),
+            'x-tigon-is-retry': 'False',
+            'x-whatsapp': '0'}
+
+
 class InstaloaderContext:
     """Class providing methods for (error) logging and low-level communication with Instagram.
 
@@ -61,6 +90,7 @@ class InstaloaderContext:
         self.request_timeout = request_timeout
         self._session = self.get_anonymous_session()
         self.username = None
+        self.user_id = None
         self.sleep = sleep
         self.quiet = quiet
         self.max_connection_attempts = max_connection_attempts
@@ -68,9 +98,10 @@ class InstaloaderContext:
         self._root_rhx_gis = None
         self.two_factor_auth_pending = None
         self.iphone_support = iphone_support
+        self.iphone_headers = default_iphone_headers()
 
         # error log, filled with error() and printed at the end of Instaloader.main()
-        self.error_log = []                      # type: List[str]
+        self.error_log: List[str] = []
 
         self._rate_controller = rate_controller(self) if rate_controller is not None else RateController(self)
 
@@ -81,20 +112,26 @@ class InstaloaderContext:
         self.fatal_status_codes = fatal_status_codes or []
 
         # Cache profile from id (mapping from id to Profile)
-        self.profile_id_cache = dict()           # type: Dict[int, Any]
+        self.profile_id_cache: Dict[int, Any] = dict()
 
     @contextmanager
     def anonymous_copy(self):
         session = self._session
         username = self.username
+        user_id = self.user_id
+        iphone_headers = self.iphone_headers
         self._session = self.get_anonymous_session()
         self.username = None
+        self.user_id = None
+        self.iphone_headers = default_iphone_headers()
         try:
             yield self
         finally:
             self._session.close()
             self.username = username
             self._session = session
+            self.user_id = user_id
+            self.iphone_headers = iphone_headers
 
     @property
     def is_logged_in(self) -> bool:
@@ -170,22 +207,30 @@ class InstaloaderContext:
         session.request = partial(session.request, timeout=self.request_timeout) # type: ignore
         return session
 
-    def save_session_to_file(self, sessionfile):
-        """Not meant to be used directly, use :meth:`Instaloader.save_session_to_file`."""
-        pickle.dump(requests.utils.dict_from_cookiejar(self._session.cookies), sessionfile)
+    def save_session(self):
+        """Not meant to be used directly, use :meth:`Instaloader.save_session`."""
+        return requests.utils.dict_from_cookiejar(self._session.cookies)
 
-    def load_session_from_file(self, username, sessionfile):
-        """Not meant to be used directly, use :meth:`Instaloader.load_session_from_file`."""
+    def load_session(self, username, sessiondata):
+        """Not meant to be used directly, use :meth:`Instaloader.load_session`."""
         session = requests.Session()
-        session.cookies = requests.utils.cookiejar_from_dict(pickle.load(sessionfile))
+        session.cookies = requests.utils.cookiejar_from_dict(sessiondata)
         session.headers.update(self._default_http_header())
         session.headers.update({'X-CSRFToken': session.cookies.get_dict()['csrftoken']})
         # Override default timeout behavior.
         # Need to silence mypy bug for this. See: https://github.com/python/mypy/issues/2427
-        session.request = partial(session.request, timeout=self.request_timeout) # type: ignore
+        session.request = partial(session.request, timeout=self.request_timeout)  # type: ignore
         self._session = session
         self.username = username
 
+    def save_session_to_file(self, sessionfile):
+        """Not meant to be used directly, use :meth:`Instaloader.save_session_to_file`."""
+        pickle.dump(self.save_session(), sessionfile)
+
+    def load_session_from_file(self, username, sessionfile):
+        """Not meant to be used directly, use :meth:`Instaloader.load_session_from_file`."""
+        self.load_session(username, pickle.load(sessionfile))
+
     def test_login(self) -> Optional[str]:
         """Not meant to be used directly, use :meth:`Instaloader.test_login`."""
         data = self.graphql_query("d6f4427fbe92d846298cf93df0b937d3", {})
@@ -265,6 +310,7 @@ class InstaloaderContext:
         session.headers.update({'X-CSRFToken': login.cookies['csrftoken']})
         self._session = session
         self.username = user
+        self.user_id = resp_json['userId']
 
     def two_factor_login(self, two_factor_code):
         """Second step of login if 2FA is enabled.
@@ -298,7 +344,8 @@ class InstaloaderContext:
             time.sleep(min(random.expovariate(0.6), 15.0))
 
     def get_json(self, path: str, params: Dict[str, Any], host: str = 'www.instagram.com',
-                 session: Optional[requests.Session] = None, _attempt=1) -> Dict[str, Any]:
+                 session: Optional[requests.Session] = None, _attempt=1,
+                 response_headers: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
         """JSON request to Instagram.
 
         :param path: URL, relative to the given domain which defaults to www.instagram.com/
@@ -325,22 +372,29 @@ class InstaloaderContext:
             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 ""
-                raise AbortDownloadException("Query to https://{}/{} responded with \"{} {}\"{}".format(
-                    host, path, resp.status_code, resp.reason, redirect
+                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'):
+                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.")
-                    # alternate rate limit exceeded behavior
-                    raise TooManyRequestsException("Redirected to login")
+                    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:
                 raise QueryReturnedBadRequestException("400 Bad Request")
             if resp.status_code == 404:
@@ -391,7 +445,8 @@ class InstaloaderContext:
                         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)
+                return self.get_json(path=path, params=params, host=host, session=sess, _attempt=_attempt + 1,
+                                     response_headers=response_headers)
             except KeyboardInterrupt:
                 self.error("[skipped by user]", repeat_at_end=False)
                 raise ConnectionException(error_string) from err
@@ -481,11 +536,57 @@ class InstaloaderContext:
 
         .. versionadded:: 4.2.1"""
         with copy_session(self._session, self.request_timeout) as tempsession:
-            tempsession.headers['User-Agent'] = 'Instagram 146.0.0.27.125 (iPhone12,1; iOS 13_3; en_US; en-US; ' \
-                                                'scale=2.00; 1656x3584; 190542906)'
-            for header in ['Host', 'Origin', 'X-Instagram-AJAX', 'X-Requested-With']:
+            # Set headers to simulate an API request from iPad
+            tempsession.headers['ig-intended-user-id'] = str(self.user_id)
+            tempsession.headers['x-pigeon-rawclienttime'] = '{:.6f}'.format(time.time())
+
+            # Add headers obtained from previous iPad request
+            tempsession.headers.update(self.iphone_headers)
+
+            # Extract key information from cookies if we haven't got it already from a previous request
+            header_cookies_mapping = {'x-mid': 'mid',
+                                     'ig-u-ds-user-id': 'ds_user_id',
+                                     'x-ig-device-id': 'ig_did',
+                                     'x-ig-family-device-id': 'ig_did',
+                                     'family_device_id': 'ig_did'}
+
+            # Map the cookie value to the matching HTTP request header
+            cookies = tempsession.cookies.get_dict().copy()
+            for key, value in header_cookies_mapping.items():
+                if value in cookies:
+                    if key not in tempsession.headers:
+                        tempsession.headers[key] = cookies[value]
+                    else:
+                        # Remove the cookie value if it's already specified as a header
+                        tempsession.cookies.pop(value, None)
+
+            # Edge case for ig-u-rur header due to special string encoding in cookie
+            if 'rur' in cookies:
+                if 'ig-u-rur' not in tempsession.headers:
+                    tempsession.headers['ig-u-rur'] = cookies['rur'].strip('\"').encode('utf-8') \
+                                                                    .decode('unicode_escape')
+                else:
+                    tempsession.cookies.pop('rur', None)
+
+            # Remove headers specific to Desktop version
+            for header in ['Host', 'Origin', 'X-Instagram-AJAX', 'X-Requested-With', 'Referer']:
                 tempsession.headers.pop(header, None)
-            return self.get_json(path, params, 'i.instagram.com', tempsession)
+
+            # No need for cookies if we have a bearer token
+            if 'authorization' in tempsession.headers:
+                tempsession.cookies.clear()
+
+            response_headers = dict()    # type: Dict[str, Any]
+            response = self.get_json(path, params, 'i.instagram.com', tempsession, response_headers=response_headers)
+
+            # Extract the ig-set-* headers and use them in the next request
+            for key, value in response_headers.items():
+                if key.startswith('ig-set-'):
+                    self.iphone_headers[key.replace('ig-set-', '')] = value
+                elif key.startswith('x-ig-set-'):
+                    self.iphone_headers[key.replace('x-ig-set-', 'x-ig-')] = value
+
+            return response
 
     def write_raw(self, resp: Union[bytes, requests.Response], filename: str) -> None:
         """Write raw response data into a file.
@@ -581,7 +682,7 @@ class RateController:
 
     def __init__(self, context: InstaloaderContext):
         self._context = context
-        self._query_timestamps = dict()  # type: Dict[str, List[float]]
+        self._query_timestamps: Dict[str, List[float]] = dict()
         self._earliest_next_request_time = 0.0
         self._iphone_earliest_next_request_time = 0.0
 
@@ -589,7 +690,6 @@ class RateController:
         """Wait given number of seconds."""
         # Not static, to allow for the behavior of this method to depend on context-inherent properties, such as
         # whether we are logged in.
-        # pylint:disable=no-self-use
         time.sleep(secs)
 
     def _dump_query_timestamps(self, current_time: float, failed_query_type: str):
@@ -613,7 +713,6 @@ class RateController:
         control."""
         # Not static, to allow for the count_per_sliding_window to depend on context-inherent properties, such as
         # whether we are logged in.
-        # pylint:disable=no-self-use
         return 75 if query_type == 'other' else 200
 
     def _reqs_in_sliding_window(self, query_type: Optional[str], current_time: float, window: float) -> List[float]:

instaloader/lateststamps.py

@@ -1,6 +1,8 @@
 import configparser
 from datetime import datetime, timezone
 from typing import Optional
+from os.path import dirname
+from os import makedirs
 
 
 class LatestStamps:
@@ -25,6 +27,7 @@ class LatestStamps:
         self.data.read(latest_stamps_file)
 
     def _save(self):
+        makedirs(dirname(self.file), exist_ok=True)
         with open(self.file, 'w') as f:
             self.data.write(f)
 

instaloader/nodeiterator.py

@@ -10,15 +10,15 @@ from typing import Any, Callable, Dict, Iterable, Iterator, NamedTuple, Optional
 from .exceptions import AbortDownloadException, InvalidArgumentException, QueryReturnedBadRequestException
 from .instaloadercontext import InstaloaderContext
 
-FrozenNodeIterator = NamedTuple('FrozenNodeIterator',
-                                [('query_hash', str),
-                                 ('query_variables', Dict),
-                                 ('query_referer', Optional[str]),
-                                 ('context_username', Optional[str]),
-                                 ('total_index', int),
-                                 ('best_before', Optional[float]),
-                                 ('remaining_data', Optional[Dict]),
-                                 ('first_node', Optional[Dict])])
+class FrozenNodeIterator(NamedTuple):
+    query_hash: str
+    query_variables: Dict
+    query_referer: Optional[str]
+    context_username: Optional[str]
+    total_index: int
+    best_before: Optional[float]
+    remaining_data: Optional[Dict]
+    first_node: Optional[Dict]
 FrozenNodeIterator.query_hash.__doc__ = """The GraphQL ``query_hash`` parameter."""
 FrozenNodeIterator.query_variables.__doc__ = """The GraphQL ``query_variables`` parameter."""
 FrozenNodeIterator.query_referer.__doc__ = """The HTTP referer used for the GraphQL query."""
@@ -95,7 +95,7 @@ class NodeIterator(Iterator[T]):
         self._is_first = is_first
 
     def _query(self, after: Optional[str] = None) -> Dict:
-        pagination_variables = {'first': NodeIterator._graphql_page_length}  # type: Dict[str, Any]
+        pagination_variables: Dict[str, Any] = {'first': NodeIterator._graphql_page_length}
         if after is not None:
             pagination_variables['after'] = after
         try:
@@ -137,7 +137,7 @@ class NodeIterator(Iterator[T]):
                 if self._first_node is None:
                     self._first_node = node
             return item
-        if self._data['page_info']['has_next_page']:
+        if self._data.get('page_info', {}).get('has_next_page'):
             query_response = self._query(self._data['page_info']['end_cursor'])
             if self._data['edges'] != query_response['edges'] and len(query_response['edges']) > 0:
                 page_index, data = self._page_index, self._data
@@ -289,7 +289,7 @@ def resumable_iteration(context: InstaloaderContext,
             is_resuming = True
             start_index = iterator.total_index
             context.log("Resuming from {}.".format(resume_file_path))
-        except (InvalidArgumentException, LZMAError, json.decoder.JSONDecodeError) as exc:
+        except (InvalidArgumentException, LZMAError, json.decoder.JSONDecodeError, EOFError) as exc:
             context.error("Warning: Not resuming from {}: {}".format(resume_file_path, exc))
     try:
         yield is_resuming, start_index

instaloader/sectioniterator.py

@@ -29,7 +29,7 @@ class SectionIterator(Iterator[T]):
     def _query(self, max_id: Optional[str] = None) -> Dict[str, Any]:
         pagination_variables = {"max_id": max_id} if max_id is not None else {}
         return self._sections_extractor(
-            self._context.get_json(self._query_path, params={"__a": 1, **pagination_variables})
+            self._context.get_json(self._query_path, params={"__a": 1, "__d": "dis", **pagination_variables})
         )
 
     def __next__(self) -> T:

instaloader/structures.py

@@ -2,12 +2,11 @@ import json
 import lzma
 import re
 from base64 import b64decode, b64encode
-from collections import namedtuple
 from contextlib import suppress
 from datetime import datetime
 from itertools import islice
 from pathlib import Path
-from typing import Any, Callable, Dict, Iterable, Iterator, List, Optional, Tuple, Union
+from typing import Any, Callable, Dict, Iterable, Iterator, List, NamedTuple, Optional, Tuple, Union
 from unicodedata import normalize
 
 from . import __version__
@@ -16,25 +15,57 @@ from .instaloadercontext import InstaloaderContext
 from .nodeiterator import FrozenNodeIterator, NodeIterator
 from .sectioniterator import SectionIterator
 
-PostSidecarNode = namedtuple('PostSidecarNode', ['is_video', 'display_url', 'video_url'])
-PostSidecarNode.__doc__ = "Item of a Sidecar Post."
+
+class PostSidecarNode(NamedTuple):
+    """Item of a Sidecar Post."""
+    is_video: bool
+    display_url: str
+    video_url: str
+
+
 PostSidecarNode.is_video.__doc__ = "Whether this node is a video."
 PostSidecarNode.display_url.__doc__ = "URL of image or video thumbnail."
 PostSidecarNode.video_url.__doc__ = "URL of video or None."
 
-PostCommentAnswer = namedtuple('PostCommentAnswer', ['id', 'created_at_utc', 'text', 'owner', 'likes_count'])
+
+class PostCommentAnswer(NamedTuple):
+    id: int
+    created_at_utc: datetime
+    text: str
+    owner: 'Profile'
+    likes_count: int
+
+
 PostCommentAnswer.id.__doc__ = "ID number of comment."
 PostCommentAnswer.created_at_utc.__doc__ = ":class:`~datetime.datetime` when comment was created (UTC)."
 PostCommentAnswer.text.__doc__ = "Comment text."
 PostCommentAnswer.owner.__doc__ = "Owner :class:`Profile` of the comment."
 PostCommentAnswer.likes_count.__doc__ = "Number of likes on comment."
 
-PostComment = namedtuple('PostComment', (*PostCommentAnswer._fields, 'answers')) # type: ignore
+
+class PostComment(NamedTuple):
+    id: int
+    created_at_utc: datetime
+    text: str
+    owner: 'Profile'
+    likes_count: int
+    answers: Iterator[PostCommentAnswer]
+
+
 for field in PostCommentAnswer._fields:
     getattr(PostComment, field).__doc__ = getattr(PostCommentAnswer, field).__doc__  # pylint: disable=no-member
-PostComment.answers.__doc__ = r"Iterator which yields all :class:`PostCommentAnswer`\ s for the comment." # type: ignore
+PostComment.answers.__doc__ = r"Iterator which yields all :class:`PostCommentAnswer`\ s for the comment."
+
+
+class PostLocation(NamedTuple):
+    id: int
+    name: str
+    slug: str
+    has_public_page: Optional[bool]
+    lat: Optional[float]
+    lng: Optional[float]
+
 
-PostLocation = namedtuple('PostLocation', ['id', 'name', 'slug', 'has_public_page', 'lat', 'lng'])
 PostLocation.id.__doc__ = "ID number of location."
 PostLocation.name.__doc__ = "Location name."
 PostLocation.slug.__doc__ = "URL friendly variant of location name."
@@ -42,6 +73,22 @@ PostLocation.has_public_page.__doc__ = "Whether location has a public page."
 PostLocation.lat.__doc__ = "Latitude (:class:`float` or None)."
 PostLocation.lng.__doc__ = "Longitude (:class:`float` or None)."
 
+# This regular expression is by MiguelX413
+_hashtag_regex = re.compile(r"(?:#)((?:\w){1,150})")
+
+# This regular expression is modified from jStassen, adjusted to use Python's \w to
+# support Unicode and a word/beginning of string delimiter at the beginning to ensure
+# that no email addresses join the list of mentions.
+# http://blog.jstassen.com/2016/03/code-regex-for-instagram-username-and-hashtags/
+_mention_regex = re.compile(r"(?:^|[^\w\n]|_)(?:@)(\w(?:(?:\w|(?:\.(?!\.))){0,28}(?:\w))?)", re.ASCII)
+
+
+def _optional_normalize(string: Optional[str]) -> Optional[str]:
+    if string is not None:
+        return normalize("NFC", string)
+    else:
+        return None
+
 
 class Post:
     """
@@ -73,8 +120,8 @@ class Post:
         self._context = context
         self._node = node
         self._owner_profile = owner_profile
-        self._full_metadata_dict = None  # type: Optional[Dict[str, Any]]
-        self._location = None            # type: Optional[PostLocation]
+        self._full_metadata_dict: Optional[Dict[str, Any]] = None
+        self._location: Optional[PostLocation] = None
         self._iphone_struct_ = None
         if 'iphone_struct' in node:
             # if loaded from JSON with load_structure_from_file()
@@ -364,16 +411,10 @@ class Post:
     @property
     def caption(self) -> Optional[str]:
         """Caption."""
-        def _normalize(string: Optional[str]) -> Optional[str]:
-            if string is not None:
-                return normalize("NFC", string)
-            else:
-                return None
-
         if "edge_media_to_caption" in self._node and self._node["edge_media_to_caption"]["edges"]:
-            return _normalize(self._node["edge_media_to_caption"]["edges"][0]["node"]["text"])
+            return _optional_normalize(self._node["edge_media_to_caption"]["edges"][0]["node"]["text"])
         elif "caption" in self._node:
-            return _normalize(self._node["caption"])
+            return _optional_normalize(self._node["caption"])
         return None
 
     @property
@@ -381,22 +422,14 @@ class Post:
         """List of all lowercased hashtags (without preceeding #) that occur in the Post's caption."""
         if not self.caption:
             return []
-        # This regular expression is from jStassen, adjusted to use Python's \w to support Unicode
-        # http://blog.jstassen.com/2016/03/code-regex-for-instagram-username-and-hashtags/
-        hashtag_regex = re.compile(r"(?:#)(\w(?:(?:\w|(?:\.(?!\.))){0,28}(?:\w))?)")
-        return re.findall(hashtag_regex, self.caption.lower())
+        return _hashtag_regex.findall(self.caption.lower())
 
     @property
     def caption_mentions(self) -> List[str]:
         """List of all lowercased profiles that are mentioned in the Post's caption, without preceeding @."""
         if not self.caption:
             return []
-        # This regular expression is modified from jStassen, adjusted to use Python's \w to
-        # support Unicode and a word/beginning of string delimiter at the beginning to ensure
-        # that no email addresses join the list of mentions.
-        # http://blog.jstassen.com/2016/03/code-regex-for-instagram-username-and-hashtags/
-        mention_regex = re.compile(r"(?:^|\W|_)(?:@)(\w(?:(?:\w|(?:\.(?!\.))){0,28}(?:\w))?)", re.ASCII)
-        return re.findall(mention_regex, self.caption.lower())
+        return _mention_regex.findall(self.caption.lower())
 
     @property
     def pcaption(self) -> str:
@@ -405,7 +438,7 @@ class Post:
         .. versionadded:: 4.2.6"""
         def _elliptify(caption):
             pcaption = ' '.join([s.replace('/', '\u2215') for s in caption.splitlines() if s]).strip()
-            return (pcaption[:30] + u"\u2026") if len(pcaption) > 31 else pcaption
+            return (pcaption[:30] + "\u2026") if len(pcaption) > 31 else pcaption
         return _elliptify(self.caption) if self.caption else ''
 
     @property
@@ -515,7 +548,7 @@ class Post:
     def get_comments(self) -> Iterable[PostComment]:
         r"""Iterate over all comments of the post.
 
-        Each comment is represented by a PostComment namedtuple with fields text (string), created_at (datetime),
+        Each comment is represented by a PostComment NamedTuple with fields text (string), created_at (datetime),
         id (int), owner (:class:`Profile`) and answers (:class:`~typing.Iterator`\ [:class:`PostCommentAnswer`])
         if available.
 
@@ -558,7 +591,7 @@ class Post:
             return []
 
         comment_edges = self._field('edge_media_to_comment', 'edges')
-        answers_count = sum([edge['node'].get('edge_threaded_comments', {}).get('count', 0) for edge in comment_edges])
+        answers_count = sum(edge['node'].get('edge_threaded_comments', {}).get('count', 0) for edge in comment_edges)
 
         if self.comments == len(comment_edges) + answers_count:
             # If the Post's metadata already contains all parent comments, don't do GraphQL requests to obtain them
@@ -625,7 +658,7 @@ class Post:
     @property
     def location(self) -> Optional[PostLocation]:
         """
-        If the Post has a location, returns PostLocation namedtuple with fields 'id', 'lat' and 'lng' and 'name'.
+        If the Post has a location, returns PostLocation NamedTuple with fields 'id', 'lat' and 'lng' and 'name'.
 
         .. versionchanged:: 4.2.9
            Require being logged in (as required by Instagram), return None if not logged-in.
@@ -638,7 +671,7 @@ class Post:
         location_id = int(loc['id'])
         if any(k not in loc for k in ('name', 'slug', 'has_public_page', 'lat', 'lng')):
             loc.update(self._context.get_json("explore/locations/{0}/".format(location_id),
-                                              params={'__a': 1})['native_location_data']['location_info'])
+                                              params={'__a': 1, '__d': 'dis'})['native_location_data']['location_info'])
         self._location = PostLocation(location_id, loc['name'], loc['slug'], loc['has_public_page'],
                                       loc.get('lat'), loc.get('lng'))
         return self._location
@@ -681,7 +714,7 @@ class Profile:
     def __init__(self, context: InstaloaderContext, node: Dict[str, Any]):
         assert 'username' in node
         self._context = context
-        self._has_public_story = None  # type: Optional[bool]
+        self._has_public_story: Optional[bool] = None
         self._node = node
         self._has_full_metadata = False
         self._iphone_struct_ = None
@@ -874,7 +907,29 @@ class Profile:
 
     @property
     def biography(self) -> str:
-        return self._metadata('biography')
+        return normalize("NFC", self._metadata('biography'))
+
+    @property
+    def biography_hashtags(self) -> List[str]:
+        """
+        List of all lowercased hashtags (without preceeding #) that occur in the Profile's biography.
+
+        .. versionadded:: 4.10
+        """
+        if not self.biography:
+            return []
+        return _hashtag_regex.findall(self.biography.lower())
+
+    @property
+    def biography_mentions(self) -> List[str]:
+        """
+        List of all lowercased profiles that are mentioned in the Profile's biography, without preceeding @.
+
+        .. versionadded:: 4.10
+        """
+        if not self.biography:
+            return []
+        return _mention_regex.findall(self.biography.lower())
 
     @property
     def blocked_by_viewer(self) -> bool:
@@ -1042,6 +1097,27 @@ class Profile:
     def _make_is_newest_checker() -> Callable[[Post, Optional[Post]], bool]:
         return lambda post, first: first is None or post.date_local > first.date_local
 
+    def get_followed_hashtags(self) -> NodeIterator['Hashtag']:
+        """
+        Retrieve list of hashtags followed by given profile.
+        To use this, one needs to be logged in and private profiles has to be followed.
+
+        :rtype: NodeIterator[Hashtag]
+
+        .. versionadded:: 4.10
+        """
+        if not self._context.is_logged_in:
+            raise LoginRequiredException("--login required to get a profile's followers.")
+        self._obtain_metadata()
+        return NodeIterator(
+            self._context,
+            'e6306cc3dbe69d6a82ef8b5f8654c50b',
+            lambda d: d["data"]["user"]["edge_following_hashtag"],
+            lambda n: Hashtag(self._context, n),
+            {'id': str(self.userid)},
+            'https://www.instagram.com/{0}/'.format(self.username),
+        )
+
     def get_followers(self) -> NodeIterator['Profile']:
         """
         Retrieve list of followers of given profile.
@@ -1248,6 +1324,53 @@ class StoryItem:
         """Type of post, GraphStoryImage or GraphStoryVideo"""
         return self._node['__typename']
 
+    @property
+    def caption(self) -> Optional[str]:
+        """
+        Caption.
+
+        .. versionadded:: 4.10
+        """
+        if "edge_media_to_caption" in self._node and self._node["edge_media_to_caption"]["edges"]:
+            return _optional_normalize(self._node["edge_media_to_caption"]["edges"][0]["node"]["text"])
+        elif "caption" in self._node:
+            return _optional_normalize(self._node["caption"])
+        return None
+
+    @property
+    def caption_hashtags(self) -> List[str]:
+        """
+        List of all lowercased hashtags (without preceeding #) that occur in the StoryItem's caption.
+
+        .. versionadded:: 4.10
+        """
+        if not self.caption:
+            return []
+        return _hashtag_regex.findall(self.caption.lower())
+
+    @property
+    def caption_mentions(self) -> List[str]:
+        """
+        List of all lowercased profiles that are mentioned in the StoryItem's caption, without preceeding @.
+
+        .. versionadded:: 4.10
+        """
+        if not self.caption:
+            return []
+        return _mention_regex.findall(self.caption.lower())
+
+    @property
+    def pcaption(self) -> str:
+        """
+        Printable caption, useful as a format specifier for --filename-pattern.
+
+        .. versionadded:: 4.10
+        """
+        def _elliptify(caption):
+            pcaption = ' '.join([s.replace('/', '\u2215') for s in caption.splitlines() if s]).strip()
+            return (pcaption[:30] + "\u2026") if len(pcaption) > 31 else pcaption
+        return _elliptify(self.caption) if self.caption else ''
+
     @property
     def is_video(self) -> bool:
         """True if the StoryItem is a video."""
@@ -1313,9 +1436,9 @@ class Story:
     def __init__(self, context: InstaloaderContext, node: Dict[str, Any]):
         self._context = context
         self._node = node
-        self._unique_id = None      # type: Optional[str]
-        self._owner_profile = None  # type: Optional[Profile]
-        self._iphone_struct_ = None  # type: Optional[Dict[str, Any]]
+        self._unique_id: Optional[str] = None
+        self._owner_profile: Optional[Profile] = None
+        self._iphone_struct_: Optional[Dict[str, Any]] = None
 
     def __repr__(self):
         return '<Story by {} changed {:%Y-%m-%d_%H-%M-%S_UTC}>'.format(self.owner_username, self.latest_media_utc)
@@ -1431,8 +1554,8 @@ class Highlight(Story):
     def __init__(self, context: InstaloaderContext, node: Dict[str, Any], owner: Optional[Profile] = None):
         super().__init__(context, node)
         self._owner_profile = owner
-        self._items = None  # type: Optional[List[Dict[str, Any]]]
-        self._iphone_struct_ = None  # type: Optional[Dict[str, Any]]
+        self._items: Optional[List[Dict[str, Any]]] = None
+        self._iphone_struct_: Optional[Dict[str, Any]] = None
 
     def __repr__(self):
         return '<Highlight by {}: {}>'.format(self.owner_username, self.title)

setup.py

@@ -18,8 +18,8 @@ def get_version():
     raise SystemExit("Could not find version string.")
 
 
-if sys.version_info < (3, 6):
-    sys.exit('Instaloader requires Python >= 3.6.')
+if sys.version_info < (3, 8):
+    sys.exit('Instaloader requires Python >= 3.8.')
 
 requirements = ['requests>=2.4']
 
@@ -43,7 +43,7 @@ setup(
                 'from Instagram.',
     long_description=open(os.path.join(SRC, 'README.rst')).read(),
     install_requires=requirements,
-    python_requires='>=3.6',
+    python_requires='>=3.8',
     entry_points={'console_scripts': ['instaloader=instaloader.__main__:main']},
     zip_safe=False,
     keywords=keywords,
@@ -54,10 +54,10 @@ setup(
         'Intended Audience :: Developers',
         'License :: OSI Approved :: MIT License',
         'Operating System :: OS Independent',
-        'Programming Language :: Python :: 3.6',
-        'Programming Language :: Python :: 3.7',
         'Programming Language :: Python :: 3.8',
         'Programming Language :: Python :: 3.9',
+        'Programming Language :: Python :: 3.10',
+        'Programming Language :: Python :: 3.11',
         'Programming Language :: Python :: 3 :: Only',
         'Topic :: Internet',
         'Topic :: Multimedia :: Graphics'

test/instaloader_unittests.py

@@ -23,7 +23,7 @@ PRIVATE_PROFILE_ID = 1706625676
 EMPTY_PROFILE = "not_public"
 EMPTY_PROFILE_ID = 1928659031
 
-ratecontroller = None  # type: Optional[instaloader.RateController]
+ratecontroller: Optional[instaloader.RateController] = None
 
 
 class TestInstaloaderAnonymously(unittest.TestCase):