Merge branch 'master' into upcoming/v4.16

README.rst

@@ -124,11 +124,30 @@ Supporters
 .. current-sponsors-start
 
 | Instaloader is proudly sponsored by
-|  `@rocketapi-io <https://github.com/rocketapi-io>`__
+
+.. list-table::
+  :align: center
+
+  * - |socialapis|
+  * - `SocialAPIs <https://socialapis.io/?utm_source=github&utm_medium=sponsor&utm_campaign=instaloader>`__
+
+|
+
+.. list-table::
+  :align: center
+
+  * - `@rocketapi-io <https://github.com/rocketapi-io>`__
+
+|
 
 See `Alex' GitHub Sponsors <https://github.com/sponsors/aandergr>`__ page for
 how you can sponsor the development of Instaloader!
 
+.. |socialapis| image:: ./.sponsor-logos/socialapis.png
+   :alt: SocialAPIs
+   :target: https://socialapis.io/?utm_source=github&utm_medium=sponsor&utm_campaign=instaloader
+   :height: 96px
+
 .. current-sponsors-end
 
 It is a pleasure for us to share our Instaloader to the world, and we are proud

docs/.sponsor-logos

@@ -0,0 +1 @@
+../.sponsor-logos

instaloader/__init__.py

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

instaloader/__main__.py

@@ -82,7 +82,7 @@ def filterstr_to_filterfunc(filter_str: str, item_type: type):
     return filterfunc
 
 
-def get_cookies_from_instagram(domain, browser, cookie_file='', cookie_name=''):
+def get_cookies_from_instagram(domain: str, browser: str, cookie_file: Optional[str] = None):
     supported_browsers = {
         "brave": browser_cookie3.brave,
         "chrome": browser_cookie3.chrome,
@@ -114,13 +114,10 @@ def get_cookies_from_instagram(domain, browser, cookie_file='', cookie_name=''):
         raise LoginException(f"No cookies found for Instagram in {browser}, "
                              f"Are you logged in successfully in {browser}?")
 
-    if cookie_name:
+    return cookies
-        return cookies.get(cookie_name, {})
-    else:
-        return cookies
 
 
-def import_session(browser, instaloader, cookiefile):
+def import_session(browser: str, instaloader: Instaloader, cookiefile: Optional[str]):
     cookie = get_cookies_from_instagram('instagram', browser, cookiefile)
     if cookie is not None:
         instaloader.context.update_cookies(cookie)

instaloader/instaloadercontext.py

@@ -191,7 +191,8 @@ class InstaloaderContext:
                   'Referer': 'https://www.instagram.com/',
                   'User-Agent': self.user_agent,
                   'X-Instagram-AJAX': '1',
-                  'X-Requested-With': 'XMLHttpRequest'}
+                  'X-Requested-With': 'XMLHttpRequest',
+                  'x-ig-app-id': '936619743392459'}
         if empty_session_only:
             del header['Host']
             del header['Origin']
@@ -540,6 +541,13 @@ class InstaloaderContext:
         :param referer: HTTP Referer, or None.
         :return: The server's response dictionary.
         """
+        csrf = next((c.value for c in self._session.cookies
+                     if c.name == 'csrftoken' and c.value), None)
+        if not csrf:
+            self._session.get('https://www.instagram.com/', timeout=self.request_timeout)
+            csrf = next((c.value for c in self._session.cookies
+                         if c.name == 'csrftoken' and c.value), '')
+
         with copy_session(self._session, self.request_timeout) as tmpsession:
             tmpsession.headers.update(self._default_http_header(empty_session_only=True))
             del tmpsession.headers['Connection']
@@ -547,6 +555,7 @@ class InstaloaderContext:
             tmpsession.headers['authority'] = 'www.instagram.com'
             tmpsession.headers['scheme'] = 'https'
             tmpsession.headers['accept'] = '*/*'
+            tmpsession.headers['x-csrftoken'] = csrf
             if referer is not None:
                 tmpsession.headers['referer'] = urllib.parse.quote(referer)
 

instaloader/structures.py

@@ -152,7 +152,7 @@ _hashtag_regex = re.compile(r"(?:#)((?:\w){1,150})")
 # 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)
+_mention_regex = re.compile(r"(?:^|\W|_)(?:@)(\w(?:(?:\w|(?:\.(?!\.))){0,28}(?:\w))?)", re.ASCII)
 
 
 def _optional_normalize(string: Optional[str]) -> Optional[str]:
@@ -258,6 +258,100 @@ class Post:
             fake_node["video_url"] = iphone_node["video_versions"][0]["url"]
         return fake_node
 
+    @staticmethod
+    def _fetch_play_count_from_clips(context: 'InstaloaderContext', user_id: str,
+                                     shortcode: str) -> Optional[int]:
+        """Fetch play_count for a reel via the clips connection endpoint as a fallback."""
+        try:
+            resp = context.doc_id_graphql_query(
+                "27234427476213202",
+                {"data": {"include_feed_video": True, "page_size": 12,
+                           "target_user_id": str(user_id)}},
+            )
+            edges = ((resp.get("data") or {})
+                     .get("xdt_api__v1__clips__user__connection_v2") or {})
+            for edge in edges.get("edges") or []:
+                media = (edge.get("node") or {}).get("media") or {}
+                if media.get("code") == shortcode:
+                    return media.get("play_count")
+        except (ConnectionException, BadResponseException):
+            pass
+        return None
+
+    @staticmethod
+    def _normalize_post_data(media: Dict[str, Any], context: 'InstaloaderContext') -> Dict[str, Any]:
+        """Normalize a Polaris media item to a legacy-compatible node, preserving the source structure."""
+        media_types = {1: "GraphImage", 2: "GraphVideo", 8: "GraphSidecar"}
+        media_type = media.get("media_type")
+        typename = None
+        if isinstance(media_type, int):
+            typename = media_types.get(media_type)
+        if not typename:
+            raise BadResponseException(f"Unknown media_type in metadata: {media_type}.")
+        pic_json: Dict[str, Any] = media.copy()
+        pic_json["shortcode"] = media["code"]
+        pic_json["id"] = media["pk"]
+        pic_json["__typename"] = typename
+        pic_json["is_video"] = media_type == 2
+        pic_json["taken_at_timestamp"] = media["taken_at"]
+        pic_json["owner"] = {
+            "id": media["user"]["pk"],
+            "username": media["user"].get("username", ""),
+            "full_name": media["user"].get("full_name", ""),
+        }
+        candidates = (media.get("image_versions2") or {}).get("candidates") or []
+        pic_json["display_url"] = candidates[0]["url"] if candidates else None
+        video_versions = media.get("video_versions") or []
+        pic_json["video_url"] = video_versions[0]["url"] if video_versions else None
+        pic_json["video_duration"] = media.get("video_duration")
+        pic_json["video_view_count"] = media.get("view_count")
+        pic_json["video_play_count"] = media.get("play_count")
+        if media_type == 2 and pic_json["video_view_count"] is None:
+            pic_json["video_play_count"] = Post._fetch_play_count_from_clips(
+                context, media["user"]["pk"], media["code"]
+            )
+        caption = media.get("caption")
+        caption_text = caption.get("text") if isinstance(caption, dict) else None
+        pic_json["edge_media_to_caption"] = (
+            {"edges": [{"node": {"text": caption_text}}]} if caption_text is not None
+            else {"edges": []}
+        )
+        pic_json["edge_media_preview_like"] = {"count": media.get("like_count") or 0}
+        pic_json["edge_media_to_parent_comment"] = {
+            "count": media.get("comment_count") or 0,
+            "edges": [],
+        }
+        if media.get("has_liked") is not None:
+            pic_json["viewer_has_liked"] = media["has_liked"]
+        carousel = media.get("carousel_media") or []
+        if carousel:
+            carousel_nodes = []
+            for item in carousel:
+                item_type = item.get("media_type", 1)
+                node: Dict[str, Any] = {
+                    "shortcode": item.get("code", ""),
+                    "__typename": media_types.get(item_type, "GraphImage"),
+                    "is_video": item_type == 2,
+                }
+                item_candidates = (item.get("image_versions2") or {}).get("candidates") or []
+                node["display_url"] = item_candidates[0]["url"] if item_candidates else ""
+                item_videos = item.get("video_versions") or []
+                node["video_url"] = item_videos[0]["url"] if item_videos else None
+                if item.get("accessibility_caption") is not None:
+                    node["accessibility_caption"] = item["accessibility_caption"]
+                carousel_nodes.append({"node": node})
+            pic_json["edge_sidecar_to_children"] = {"edges": carousel_nodes}
+        tagged = (media.get("usertags") or {}).get("in") or []
+        if tagged:
+            pic_json["edge_media_to_tagged_user"] = {
+                "edges": [
+                    {"node": {"user": {"username": t["user"]["username"].lower()}}}
+                    for t in tagged
+                    if (t.get("user") or {}).get("username")
+                ]
+            }
+        return pic_json
+
     @staticmethod
     def shortcode_to_mediaid(code: str) -> int:
         if len(code) > 11:
@@ -319,23 +413,18 @@ class Post:
 
     def _obtain_metadata(self):
         if not self._full_metadata_dict:
-            pic_json = self._context.doc_id_graphql_query(
+            resp = self._context.doc_id_graphql_query(
-                "8845758582119845", {"shortcode": self.shortcode}
+                "27128499623469141",
-            )["data"]["xdt_shortcode_media"]
+                {
-            if pic_json is None:
+                    "shortcode": self.shortcode,
+                    "__relay_internal__pv__PolarisAIGMMediaWebLabelEnabledrelayprovider": False,
+                },
+            )
+            web_info = (resp.get("data") or {}).get("xdt_api__v1__media__shortcode__web_info") or {}
+            items = web_info.get("items")
+            if not items:
                 raise BadResponseException("Fetching Post metadata failed.")
-            try:
+            self._full_metadata_dict = Post._normalize_post_data(items[0], self._context)
-                xdt_types = {
-                    "XDTGraphImage": "GraphImage",
-                    "XDTGraphVideo": "GraphVideo",
-                    "XDTGraphSidecar": "GraphSidecar",
-                }
-                pic_json["__typename"] = xdt_types[pic_json["__typename"]]
-            except KeyError as exc:
-                raise BadResponseException(
-                    f"Unknown __typename in metadata: {pic_json['__typename']}."
-                ) from exc
-            self._full_metadata_dict = pic_json
             if self.shortcode != self._full_metadata_dict['shortcode']:
                 self._node.update(self._full_metadata_dict)
                 raise PostChangedException
@@ -910,11 +999,20 @@ class Profile:
         :param username: Username
         :raises: :class:`ProfileNotExistsException`
         """
-        data = context.doc_id_graphql_query("26347858941511777", {"hasQuery": True, "query": username})["data"]
+        # Resolve the profile through the web_profile_info endpoint, which works both
-        if data:
+        # anonymously and when logged in and returns the complete profile node
-            for user in data["xdt_api__v1__fbsearch__non_profiled_serp"]["users"]:
+        # (including the first page of posts). The GraphQL fbsearch query previously
-                if user["username"].lower() == username.lower():
+        # used here started responding with HTTP 400.
-                    return cls(context, user)
+        try:
+            data = context.get_json(
+                "api/v1/users/web_profile_info/", params={"username": username.lower()}
+            ).get("data")
+        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))
 
@@ -929,15 +1027,16 @@ class Profile:
         """
         if profile_id in context.profile_id_cache:
             return context.profile_id_cache[profile_id]
-        data = context.graphql_query('7c16654f22c819fb63d1183034a5162f',
+        # Resolve the current username from the user id, then load the full profile.
-                                     {'user_id': str(profile_id),
+        # The GraphQL user query previously used here started responding with HTTP 400.
-                                      'include_chaining': False,
+        try:
-                                      'include_reel': True,
+            user = context.get_json(
-                                      'include_suggested_users': False,
+                "api/v1/users/{0}/info/".format(profile_id), params={}
-                                      'include_logged_out_extras': False,
+            ).get('user')
-                                      'include_highlight_reels': False})['data']['user']
+        except QueryReturnedNotFoundException:
-        if data:
+            user = None
-            profile = cls(context, data['reel']['owner'])
+        if user and user.get('username'):
+            profile = cls.from_username(context, user['username'])
         else:
             raise ProfileNotExistsException("No profile found, the user may have blocked you (ID: " +
                                             str(profile_id) + ").")
@@ -983,6 +1082,24 @@ class Profile:
     def _obtain_metadata(self):
         try:
             if not self._has_full_metadata:
+                if not self._context.is_logged_in:
+                    # Anonymous access: the web_profile_info endpoint returns the full
+                    # node in the legacy format and still works, unlike the GraphQL
+                    # profile query.
+                    try:
+                        data = self._context.get_json(
+                            "api/v1/users/web_profile_info/",
+                            params={"username": self.username},
+                        ).get('data')
+                    except QueryReturnedNotFoundException as err:
+                        raise ProfileNotExistsException(
+                            'Profile {} does not exist.'.format(self.username)) from err
+                    user_data = data.get('user') if data else None
+                    if user_data is None:
+                        raise ProfileNotExistsException('Profile {} does not exist.'.format(self.username))
+                    self._node = user_data
+                    self._has_full_metadata = True
+                    return
                 user_id = self._node.get('id') or self._node.get('pk')
                 variables = {
                     "id": str(user_id),
@@ -990,8 +1107,10 @@ class 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,
                 }
-                data = self._context.doc_id_graphql_query('25980296051578533', variables)
+                data = self._context.doc_id_graphql_query('27937681195819736', variables)
                 if data is None:
                     raise QueryReturnedNotFoundException('GraphQL query returned None')
                 user_data = data.get('data', {}).get('user')

test/instaloader_unittests.py

@@ -210,5 +210,28 @@ class TestInstaloaderLoggedIn(TestInstaloaderAnonymously):
                 break
 
 
+class TestCaptionMentions(unittest.TestCase):
+    """Caption parsing, which requires no network access."""
+
+    @staticmethod
+    def _post_with_caption(caption: str) -> instaloader.Post:
+        return instaloader.Post(None, {"shortcode": "BhrEy4nBFVU", "caption": caption})
+
+    def test_caption_mentions(self):
+        for caption, expected in [
+                ("@alice at the beginning", ["alice"]),
+                ("preceded by a space @bob", ["bob"]),
+                ("preceded by a newline\n@carol", ["carol"]),
+                ("multiple lines\n@dave\n@erin", ["dave", "erin"]),
+                ("no mentions here", []),
+        ]:
+            with self.subTest(caption=caption):
+                self.assertEqual(self._post_with_caption(caption).caption_mentions, expected)
+
+    def test_caption_mentions_ignores_email_addresses(self):
+        post = self._post_with_caption("write to alice@example.com\ncontact @bob instead")
+        self.assertEqual(post.caption_mentions, ["bob"])
+
+
 if __name__ == '__main__':
     unittest.main()