diff --git a/.sponsor-logos/socialapis.png b/.sponsor-logos/socialapis.png new file mode 100644 index 0000000..f2d400c Binary files /dev/null and b/.sponsor-logos/socialapis.png differ diff --git a/README.rst b/README.rst index dd81329..3d4d061 100644 --- a/README.rst +++ b/README.rst @@ -124,11 +124,30 @@ Supporters .. current-sponsors-start | Instaloader is proudly sponsored by -| `@rocketapi-io `__ + +.. list-table:: + :align: center + + * - |socialapis| + * - `SocialAPIs `__ + +| + +.. list-table:: + :align: center + + * - `@rocketapi-io `__ + +| See `Alex' GitHub Sponsors `__ 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 diff --git a/docs/.sponsor-logos b/docs/.sponsor-logos new file mode 120000 index 0000000..0650adf --- /dev/null +++ b/docs/.sponsor-logos @@ -0,0 +1 @@ +../.sponsor-logos \ No newline at end of file diff --git a/instaloader/__init__.py b/instaloader/__init__.py index 02f552d..dced6cc 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.1' +__version__ = '4.15.3' try: diff --git a/instaloader/__main__.py b/instaloader/__main__.py index e76dfea..7b2670d 100644 --- a/instaloader/__main__.py +++ b/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.get(cookie_name, {}) - else: - return cookies + 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) diff --git a/instaloader/instaloadercontext.py b/instaloader/instaloadercontext.py index 045408a..07b5378 100644 --- a/instaloader/instaloadercontext.py +++ b/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) diff --git a/instaloader/structures.py b/instaloader/structures.py index 1f4ca60..3d2342c 100644 --- a/instaloader/structures.py +++ b/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( - "8845758582119845", {"shortcode": self.shortcode} - )["data"]["xdt_shortcode_media"] - if pic_json is None: + resp = self._context.doc_id_graphql_query( + "27128499623469141", + { + "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: - 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 + self._full_metadata_dict = Post._normalize_post_data(items[0], self._context) 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"] - if data: - for user in data["xdt_api__v1__fbsearch__non_profiled_serp"]["users"]: - if user["username"].lower() == username.lower(): - return cls(context, user) + # Resolve the profile through the web_profile_info endpoint, which works both + # anonymously and when logged in and returns the complete profile node + # (including the first page of posts). The GraphQL fbsearch query previously + # used here started responding with HTTP 400. + 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', - {'user_id': str(profile_id), - 'include_chaining': False, - 'include_reel': True, - 'include_suggested_users': False, - 'include_logged_out_extras': False, - 'include_highlight_reels': False})['data']['user'] - if data: - profile = cls(context, data['reel']['owner']) + # 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') + except QueryReturnedNotFoundException: + user = None + 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') diff --git a/test/instaloader_unittests.py b/test/instaloader_unittests.py index 0cb7c40..87d5990 100644 --- a/test/instaloader_unittests.py +++ b/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()