diff --git a/examples_threads.py b/examples_threads.py new file mode 100644 index 0000000..ee33aaa --- /dev/null +++ b/examples_threads.py @@ -0,0 +1,14 @@ +from rocketapi import ThreadsAPI +from rocketapi.exceptions import NotFoundException, BadResponseException + +api = ThreadsAPI(token="put your token here") + +# Get user feed by id +user_id = 35670846775 +try: + user = api.get_user_feed(user_id) + print(user) +except NotFoundException: + print(f"User {user_id} not found") +except BadResponseException: + print(f"Can't get {user_id} feed from API") diff --git a/rocketapi/__init__.py b/rocketapi/__init__.py index ecf79a2..f3fc4f1 100644 --- a/rocketapi/__init__.py +++ b/rocketapi/__init__.py @@ -1 +1,2 @@ from .instagramapi import InstagramAPI +from .threadsapi import ThreadsAPI diff --git a/rocketapi/instagramapi.py b/rocketapi/instagramapi.py index d141bfa..3809a21 100644 --- a/rocketapi/instagramapi.py +++ b/rocketapi/instagramapi.py @@ -3,13 +3,12 @@ class InstagramAPI(RocketAPI): - def __init__(self, token, threads=1, max_timeout=30): + def __init__(self, token, max_timeout=30): """ Instagram API client. Args: token (str): Your RocketAPI token (https://rocketapi.io/dashboard/) - threads (int): Number of threads to use for requests (it's beta, use with caution, every thread charges 1 request) max_timeout (int): Maximum timeout for requests. Please, don't use values lower than 15 seconds, it may cause problems with API. For debugging purposes you can use the following variables: @@ -20,13 +19,16 @@ def __init__(self, token, threads=1, max_timeout=30): """ self.last_response = None self.counter = 0 - super().__init__(token, threads=threads, max_timeout=max_timeout) + super().__init__(token, max_timeout=max_timeout) def request(self, method, data): response = super().request(method, data) self.last_response = response self.counter += 1 if response["status"] == "done": + if method in ["instagram/media/get_shortcode_by_id", "instagram/media/get_id_by_shortcode"]: + return response + if ( response["response"]["status_code"] == 200 and response["response"]["content_type"] == "application/json" @@ -35,13 +37,22 @@ def request(self, method, data): elif response["response"]["status_code"] == 404: raise NotFoundException("Instagram resource not found") else: - raise BadResponseException("Bad response from Instagram") - raise BadResponseException("Bad response from RocketAPI") + raise BadResponseException( + f"Bad response from Instagram ({method}: {response['response']['status_code']})" + ) + raise BadResponseException(f"Bad response from RocketAPI ({method})") def search(self, query): """ Search for a specific user, hashtag or place. + As of September 2024, we no longer recommend using this method, as it now only returns a maximum of 5 users and leaves the places and hashtags arrays empty. Instead, please use the separate methods: + - `search_users` + - `search_hashtags` + - `search_locations` + - `search_audios` + - `search_clips` + Args: query (str): The search query @@ -49,16 +60,30 @@ def search(self, query): """ return self.request("instagram/search", {"query": query}) + def get_web_profile_info(self, username): + """ + Retrieve user web profile information by username. + + Args: + username (str): Username + + For more information, see documentation: https://docs.rocketapi.io/api/instagram/user/get_web_profile_info + """ + return self.request( + "instagram/user/get_web_profile_info", {"username": username} + ) + def get_user_info(self, username): """ Retrieve user information by username. + This is an alias for get_web_profile_info. Args: username (str): Username For more information, see documentation: https://docs.rocketapi.io/api/instagram/user/get_info """ - return self.request("instagram/user/get_info", {"username": username}) + return self.get_web_profile_info(username) def get_user_info_by_id(self, user_id): """ @@ -77,7 +102,7 @@ def get_user_media(self, user_id, count=12, max_id=None): Args: user_id (int): User id - count (int): Number of media to retrieve (max: 50) + count (int): Number of media to retrieve (max: 12) max_id (str): Use for pagination You can use the `max_id` parameter to paginate through the media (take from the `next_max_id` field of the response). @@ -89,19 +114,38 @@ def get_user_media(self, user_id, count=12, max_id=None): payload["max_id"] = max_id return self.request("instagram/user/get_media", payload) - def get_user_clips(self, user_id, max_id=None): + def get_user_media_by_username(self, username, count=12, max_id=None): + """ + Retrieve user media by username. + + Args: + username (str): Username + count (int): Number of media to retrieve (max: 12) + max_id (str): Use for pagination + + You can use the `max_id` parameter to paginate through the media (take from the `next_max_id` field of the response). + + For more information, see documentation: https://docs.rocketapi.io/api/instagram/user/get_media_by_username + """ + payload = {"username": username, "count": count} + if max_id is not None: + payload["max_id"] = max_id + return self.request("instagram/user/get_media_by_username", payload) + + def get_user_clips(self, user_id, count=12, max_id=None): """ Retrieve user clips (videos from "Reels" section) by id. Args: user_id (int): User id + count (int): Number of media to retrieve (max: 12) max_id (str): Use for pagination You can use the `max_id` parameter to paginate through the media (take from the `max_id` (!) field of the response). For more information, see documentation: https://docs.rocketapi.io/api/instagram/user/get_clips """ - payload = {"id": user_id} + payload = {"id": user_id, "count": count} if max_id is not None: payload["max_id"] = max_id return self.request("instagram/user/get_clips", payload) @@ -179,7 +223,7 @@ def get_user_followers(self, user_id, count=12, max_id=None): Args: user_id (int): User id - count (int): Number of users to return (max: 100) + count (int): Number of users to return (max: 50) max_id (str): Use for pagination You can use the `max_id` parameter to paginate through followers (take from the `next_max_id` field of the response). @@ -285,24 +329,52 @@ def get_media_info_by_shortcode(self, shortcode): "instagram/media/get_info_by_shortcode", {"shortcode": shortcode} ) - def get_media_likes(self, shortcode, count=12, max_id=None): + def get_media_likes_by_shortcode(self, shortcode): """ - Retrieve media likes by media shortcode. + Retrieve up to 1000 media likes by media shortcode. Args: shortcode (str): Media shortcode - count (int): Number of likers to return (max: 50) - max_id (str): Use for pagination - You can use the `max_id` parameter to paginate through likers (take from the `next_max_id` field of the response). + Pagination is not supported for this endpoint. For more information, see documentation: https://docs.rocketapi.io/api/instagram/media/get_likes """ - payload = {"shortcode": shortcode, "count": count} - if max_id is not None: - payload["max_id"] = max_id + payload = {"shortcode": shortcode} return self.request("instagram/media/get_likes", payload) + def get_media_likes(self, shortcode, count=12, max_id=None): + """ + Retrieve up to 1000 media likes by media shortcode. + This is an alias for get_media_likes_by_shortcode. + + Note: The parameters count and max_id are kept for backward compatibility but are no longer supported. + + Args: + shortcode (str): Media shortcode + count (int): DEPRECATED - No longer supported + max_id (str): DEPRECATED - No longer supported + + Pagination is not supported for this endpoint. + + For more information, see documentation: https://docs.rocketapi.io/api/instagram/media/get_likes + """ + # Ignoring count and max_id parameters as they're no longer supported + return self.get_media_likes_by_shortcode(shortcode) + + def get_media_likes_by_id(self, media_id): + """ + Retrieve up to 1000 media likes by media id. + + Args: + media_id (int): Media id + + Pagination is not supported for this endpoint. + + For more information, see documentation: https://docs.rocketapi.io/api/instagram/media/get_likes_by_id + """ + return self.request("instagram/media/get_likes_by_id", {"id": media_id}) + def get_media_comments(self, media_id, can_support_threading=True, min_id=None): """ Retrieve media comments by media id. @@ -345,6 +417,17 @@ def get_media_id_by_shortcode(self, shortcode): "instagram/media/get_id_by_shortcode", {"shortcode": shortcode} ) + def get_media_id_by_share(self, share): + """ + Get media id by share code (for links like https://www.instagram.com/share/XXXxx356, where XXXxx356 is the share code). + + Args: + share (str): Share code + + For more information, see documentation: https://docs.rocketapi.io/api/instagram/media/get_id_by_share + """ + return self.request("instagram/media/get_id_by_share", {"share": share}) + def get_guide_info(self, guide_id): """ Retrieve guide information by guide id. @@ -367,7 +450,7 @@ def get_location_info(self, location_id): """ return self.request("instagram/location/get_info", {"id": location_id}) - def get_location_media(self, location_id, page=None, max_id=None): + def get_location_media(self, location_id, page=None, max_id=None, tab=None): """ Retrieve location media by location id. @@ -375,6 +458,7 @@ def get_location_media(self, location_id, page=None, max_id=None): location_id (int): Location id page (int): Page number max_id (str): Use for pagination + tab (str): Tab name: recent, ranked (default: recent) In order to use pagination, you need to use both the `max_id` and `page` parameters. You can obtain these values from the response's `next_page` and `next_max_id` fields. @@ -385,6 +469,8 @@ def get_location_media(self, location_id, page=None, max_id=None): payload["page"] = page if max_id is not None: payload["max_id"] = max_id + if tab is not None: + payload["tab"] = tab return self.request("instagram/location/get_media", payload) def get_hashtag_info(self, name): @@ -398,7 +484,7 @@ def get_hashtag_info(self, name): """ return self.request("instagram/hashtag/get_info", {"name": name}) - def get_hashtag_media(self, name, page=None, max_id=None): + def get_hashtag_media(self, name, page=None, max_id=None, tab=None): """ Retrieve hashtag media by hashtag name. @@ -406,6 +492,7 @@ def get_hashtag_media(self, name, page=None, max_id=None): name (str): Hashtag name page (int): Page number max_id (str): Use for pagination + tab (str): Tab name: recent, top, or clips (default: recent) In order to use pagination, you need to use both the `max_id` and `page` parameters. You can obtain these values from the response's `next_page` and `next_max_id` fields. @@ -416,6 +503,8 @@ def get_hashtag_media(self, name, page=None, max_id=None): payload["page"] = page if max_id is not None: payload["max_id"] = max_id + if tab is not None: + payload["tab"] = tab return self.request("instagram/hashtag/get_media", payload) def get_highlight_stories_bulk(self, highlight_ids): @@ -492,16 +581,102 @@ def get_audio_media(self, audio_id, max_id=None): payload["max_id"] = max_id return self.request("instagram/audio/get_media", payload) + def get_audio_media_by_canonical_id(self, audio_canonical_id, max_id=None): + """ + Retrieve audio media by audio canonical id. + + Args: + audio_canonical_id (int): Audio canonical id + max_id (str): Use for pagination + + You can use the `max_id` parameter to paginate through media (take from the `next_max_id` field of the response). + + For more information, see documentation: https://docs.rocketapi.io/api/instagram/audio/get_media_by_canonical_id + """ + payload = {"id": audio_canonical_id} + if max_id is not None: + payload["max_id"] = max_id + return self.request("instagram/audio/get_media_by_canonical_id", payload) + + def get_live_info(self, broadcast_id): + """ + Retrieve live information by broadcast id. + + Args: + broadcast_id (int): Broadcast id + + For more information, see documentation: https://docs.rocketapi.io/api/instagram/live/get_info + """ + return self.request("instagram/live/get_info", {"id": broadcast_id}) + def get_user_about(self, user_id): """ Obtain user details from «About this Account» section. - ⭐️ This method is exclusively available to our Enterprise+ clients. - If you wish to enable it for your account, please get in touch with our support team: https://t.me/rocketapi - Args: user_id (int): User id For more information, see documentation: https://docs.rocketapi.io/api/instagram/user/get_about """ return self.request("instagram/user/get_about", {"id": user_id}) + + def search_users(self, query): + """ + Search for a specific user (max 50 results) + + Args: + query (str): The search query + + For more information, see documentation: https://docs.rocketapi.io/api/instagram/user/search + """ + return self.request("instagram/user/search", {"query": query}) + + def search_hashtags(self, query): + """ + Search for a specific hashtag (max 20 results) + + Args: + query (str): The search query + + For more information, see documentation: https://docs.rocketapi.io/api/instagram/hashtag/search + """ + return self.request("instagram/hashtag/search", {"query": query}) + + def search_locations(self, query): + """ + Search for a specific location (max 20 results) + + Args: + query (str): The search query + + For more information, see documentation: https://docs.rocketapi.io/api/instagram/location/search + """ + return self.request("instagram/location/search", {"query": query}) + + def search_audios(self, query): + """ + Search for a specific audio (max 10 results) + + Args: + query (str): The search query + + For more information, see documentation: https://docs.rocketapi.io/api/instagram/audio/search + """ + return self.request("instagram/audio/search", {"query": query}) + + def search_clips(self, query, max_id=None): + """ + Search for a specific clip with a caption that includes the query (max 12 results) + + Args: + query (str): The search query + max_id (str): Use for pagination + + You can use the max_id parameter to paginate through following (take from the reels_max_id field of the response). + + For more information, see documentation: https://docs.rocketapi.io/api/instagram/media/search_clips + """ + payload = {"query": query} + if max_id is not None: + payload["max_id"] = max_id + return self.request("instagram/media/search_clips", payload) diff --git a/rocketapi/rocketapi.py b/rocketapi/rocketapi.py index 5f1db31..02c776e 100644 --- a/rocketapi/rocketapi.py +++ b/rocketapi/rocketapi.py @@ -2,7 +2,7 @@ class RocketAPI: - def __init__(self, token, threads=1, max_timeout=30): + def __init__(self, token, max_timeout=30): """ RocketAPI client. @@ -11,15 +11,17 @@ def __init__(self, token, threads=1, max_timeout=30): For more information, see documentation: https://docs.rocketapi.io/api/ """ self.base_url = "https://v1.rocketapi.io/" + self.version = "1.0.12" self.token = token - self.threads = threads self.max_timeout = max_timeout def request(self, method, data): - data["_threads"] = self.threads return requests.post( url=self.base_url + method, json=data, - headers={"Authorization": f"Token {self.token}"}, + headers={ + "Authorization": f"Token {self.token}", + "User-Agent": f"RocketAPI Python SDK/{self.version}", + }, timeout=self.max_timeout, ).json() diff --git a/rocketapi/threadsapi.py b/rocketapi/threadsapi.py new file mode 100644 index 0000000..d281e8b --- /dev/null +++ b/rocketapi/threadsapi.py @@ -0,0 +1,185 @@ +from rocketapi.exceptions import NotFoundException, BadResponseException +from rocketapi.rocketapi import RocketAPI + + +class ThreadsAPI(RocketAPI): + def __init__(self, token, max_timeout=30): + """ + Threads API client. + + Args: + token (str): Your RocketAPI token (https://rocketapi.io/dashboard/) + max_timeout (int): Maximum timeout for requests. Please, don't use values lower than 15 seconds, it may cause problems with API. + + For debugging purposes you can use the following variables: + last_response (dict): contains the last response from the API. + counter (int): contains the number of requests made in the current session. + + For more information, see documentation: https://docs.rocketapi.io/api/ + """ + self.last_response = None + self.counter = 0 + super().__init__(token, max_timeout=max_timeout) + + def request(self, method, data): + response = super().request(method, data) + self.last_response = response + self.counter += 1 + if response["status"] == "done": + if ( + response["response"]["status_code"] == 200 + and response["response"]["content_type"] == "application/json" + ): + return response["response"]["body"] + elif response["response"]["status_code"] == 404: + raise NotFoundException("Instagram resource not found") + else: + raise BadResponseException("Bad response from Threads") + raise BadResponseException("Bad response from RocketAPI") + + def search_users(self, query): + """ + Search for a specific user in Threads + + Args: + query (str): Username to search for + + For more information, see documentation: https://docs.rocketapi.io/api/threads/search_users + """ + return self.request("threads/search_users", {"query": query}) + + def get_user_info(self, user_id): + """ + Retrieve Threads user information by id. + + Args: + user_id (int): User id + + For more information, see documentation: https://docs.rocketapi.io/api/threads/user/get_info + """ + return self.request("threads/user/get_info", {"id": user_id}) + + def get_user_feed(self, user_id, max_id=None): + """ + Retrieve Threads user feed by id. + + Args: + user_id (int): User id + max_id (str): Use for pagination + + You can use the `max_id` parameter to paginate through the media (take from the `next_max_id` field of the response). + + For more information, see documentation: https://docs.rocketapi.io/api/threads/user/get_feed + """ + payload = {"id": user_id} + if max_id is not None: + payload["max_id"] = max_id + return self.request("threads/user/get_feed", payload) + + def get_user_replies(self, user_id, max_id=None): + """ + Retrieve Threads user replies by id. + + Args: + user_id (int): User id + max_id (str): Use for pagination + + You can use the `max_id` parameter to paginate through the media (take from the `next_max_id` field of the response). + + For more information, see documentation: https://docs.rocketapi.io/api/threads/user/get_replies + """ + payload = {"id": user_id} + if max_id is not None: + payload["max_id"] = max_id + return self.request("threads/user/get_replies", payload) + + def get_user_followers(self, user_id, max_id=None): + """ + Retrieve Threads user followers by id. + + Args: + user_id (int): User id + max_id (str): Use for pagination + + You can use the `max_id` parameter to paginate through followers (take from the `next_max_id` field of the response). + + For more information, see documentation: https://docs.rocketapi.io/api/threads/user/get_followers + """ + payload = {"id": user_id} + if max_id is not None: + payload["max_id"] = max_id + return self.request("threads/user/get_followers", payload) + + def search_user_followers(self, user_id, query): + """ + Search Threads user followers by user id. + + Args: + user_id (int): User id + query (str): Search query + + For more information, see documentation: https://docs.rocketapi.io/api/threads/user/get_followers + """ + return self.request( + "threads/user/get_followers", {"id": user_id, "query": query} + ) + + def get_user_following(self, user_id, max_id=None): + """ + Retrieve Threads user following by id. + + Args: + user_id (int): User id + max_id (str): Use for pagination + + You can use the `max_id` parameter to paginate through followers (take from the `next_max_id` field of the response). + + For more information, see documentation: https://docs.rocketapi.io/api/threads/user/get_following + """ + payload = {"id": user_id} + if max_id is not None: + payload["max_id"] = max_id + return self.request("threads/user/get_following", payload) + + def search_user_following(self, user_id, query): + """ + Search Threads user following by user id. + + Args: + user_id (int): User id + query (str): Search query + + For more information, see documentation: https://docs.rocketapi.io/api/threads/user/get_following + """ + return self.request( + "threads/user/get_following", {"id": user_id, "query": query} + ) + + def get_thread_replies(self, thread_id, max_id=None): + """ + Retrieve thread replies by id. + + Args: + thread_id (int): Thread id + max_id (str): Use for pagination + + You can use the `max_id` parameter to paginate through the media (take from the `paging_tokens["downwards"]` field of the response). + + For more information, see documentation: https://docs.rocketapi.io/api/threads/thread/get_replies + """ + payload = {"id": thread_id} + if max_id is not None: + payload["max_id"] = max_id + return self.request("threads/thread/get_replies", payload) + + def get_thread_likes(self, thread_id): + """ + Retrieve thread likes by id. + + Args: + thread_id (int): Thread id + + For more information, see documentation: https://docs.rocketapi.io/api/threads/thread/get_likes + """ + payload = {"id": thread_id} + return self.request("threads/thread/get_likes", payload) diff --git a/setup.py b/setup.py index 4063754..1dbcf2b 100644 --- a/setup.py +++ b/setup.py @@ -3,12 +3,12 @@ setuptools.setup( name="rocketapi", - version="1.0.4", + version="1.0.12", author="RocketAPI", author_email="developer@rocketapi.io", description="RocketAPI Python SDK", packages=["rocketapi"], url="https://github.com/rocketapi-io/rocketapi-python", - download_url="https://github.com/rocketapi-io/rocketapi-python/archive/refs/tags/v1.0.4.tar.gz", + download_url="https://github.com/rocketapi-io/rocketapi-python/archive/refs/tags/v1.0.12.tar.gz", install_requires=["requests"], )