diff --git a/instagrapi/mixins/direct.py b/instagrapi/mixins/direct.py index 52d12849..08c5f4c5 100644 --- a/instagrapi/mixins/direct.py +++ b/instagrapi/mixins/direct.py @@ -26,6 +26,13 @@ SELECTED_FILTERS = ("flagged", "unread") SEARCH_MODES = ("raven", "universal") SEND_ATTRIBUTES = ("message_button", "inbox_search") +SEND_ATTRIBUTES_MEDIA = ( + "feed_timeline", + "feed_contextual_chain", + "feed_short_url", + "feed_contextual_self_profile", + "feed_contextual_profile" +) BOXES = ("general", "primary") try: @@ -34,6 +41,7 @@ SELECTED_FILTER = Literal[SELECTED_FILTERS] SEARCH_MODE = Literal[SEARCH_MODES] SEND_ATTRIBUTE = Literal[SEND_ATTRIBUTES] + SEND_ATTRIBUTE_MEDIA = Literal[SEND_ATTRIBUTES_MEDIA] BOX = Literal[BOXES] except ImportError: # python <= 3.8 @@ -813,7 +821,12 @@ def direct_thread_hide(self, thread_id: int, move_to_spam: bool = False) -> bool ) return result.get("status", "") == "ok" - def direct_media_share(self, media_id: str, user_ids: List[int]) -> DirectMessage: + def direct_media_share( + self, + media_id: str, + user_ids: List[int], + send_attribute: SEND_ATTRIBUTES_MEDIA = "feed_timeline", + media_type: str = "photo",) -> DirectMessage: """ Share a media to list of users @@ -823,6 +836,10 @@ def direct_media_share(self, media_id: str, user_ids: List[int]) -> DirectMessag Unique Media ID user_ids: List[int] List of unique identifier of Users id + send_attribute: str, optional + Sending option. Default is "feed_timeline" + media_type: str, optional + Type of the shared media. Default is "photo", also can be "video" Returns ------- @@ -833,26 +850,37 @@ def direct_media_share(self, media_id: str, user_ids: List[int]) -> DirectMessag token = self.generate_mutation_token() media_id = self.media_id(media_id) recipient_users = dumps([[int(uid) for uid in user_ids]]) - data = { + kwargs = { "recipient_users": recipient_users, "action": "send_item", - "is_shh_mode": 0, - "send_attribution": "feed_timeline", + "is_shh_mode": "0", + "send_attribution": send_attribute, "client_context": token, "media_id": media_id, + "device_id": self.android_device_id, "mutation_token": token, + "_uuid": self.uuid, + "btt_dual_send": "false", "nav_chain": ( - "1VL:feed_timeline:1,1VL:feed_timeline:2,1VL:feed_timeline:5," - "DirectShareSheetFragment:direct_reshare_sheet:6" + "1qT:feed_timeline:1,1qT:feed_timeline:2,1qT:feed_timeline:3," + "7Az:direct_inbox:4,7Az:direct_inbox:5,5rG:direct_thread:7" ), + "is_ae_dual_send": "false", "offline_threading_id": token, } + if send_attribute in ["feed_contextual_chain", "feed_short_url"]: + kwargs["inventory_source"] = "recommended_explore_grid_cover_model" + if send_attribute == "feed_timeline": + kwargs["inventory_source"] = "media_or_ad" + result = self.private_request( "direct_v2/threads/broadcast/media_share/", - # params={'media_type': 'video'}, - data=self.with_default_data(data), + params={'media_type': media_type}, + data=self.with_default_data(kwargs), with_signature=False, ) + assert result.get("status", "") == "ok" + return extract_direct_message(result["payload"]) def direct_story_share( @@ -1058,28 +1086,34 @@ def direct_profile_share( user_ids and thread_ids ), "Specify user_ids or thread_ids, but not both" token = self.generate_mutation_token() - data = { + kwargs = { + "profile_user_id": user_id, "action": "send_item", "is_shh_mode": "0", "send_attribution": "profile", - "client_context": token, + "client_context": token, + "device_id": self.android_device_id, "mutation_token": token, + "_uuid": self.uuid, + "btt_dual_send": "false", "nav_chain": ( - "1qT:feed_timeline:1,ReelViewerFragment:reel_feed_timeline:4," - "DirectShareSheetFragment:direct_reshare_sheet:5" + "1qT:feed_timeline:1,1qT:feed_timeline:2,1qT:feed_timeline:3," + "7Az:direct_inbox:4,7Az:direct_inbox:5,5rG:direct_thread:7" ), - "profile_user_id": user_id, + "is_ae_dual_send": "false", "offline_threading_id": token, } if user_ids: - data["recipient_users"] = dumps([[int(uid) for uid in user_ids]]) + kwargs["recipient_users"] = dumps([[int(uid) for uid in user_ids]]) if thread_ids: - data["thread_ids"] = dumps([int(tid) for tid in thread_ids]) + kwargs["thread_ids"] = dumps([int(tid) for tid in thread_ids]) result = self.private_request( "direct_v2/threads/broadcast/profile/", - data=self.with_default_data(data), + data=self.with_default_data(kwargs), with_signature=False, ) + assert result.get("status", "") == "ok" + return extract_direct_message(result["payload"]) def direct_media(self, thread_id: int, amount: int = 20) -> List[Media]: diff --git a/instagrapi/mixins/user.py b/instagrapi/mixins/user.py index de619853..87ca0c72 100644 --- a/instagrapi/mixins/user.py +++ b/instagrapi/mixins/user.py @@ -10,7 +10,7 @@ UserNotFound, ) from instagrapi.extractors import extract_user_gql, extract_user_short, extract_user_v1 -from instagrapi.types import Relationship, User, UserShort +from instagrapi.types import Relationship, RelationshipShort, User, UserShort from instagrapi.utils import json_value @@ -297,18 +297,19 @@ def new_feed_exist(self) -> bool: results = self.private_request("feed/new_feed_posts_exist/") return results.get("new_feed_posts_exist", False) - def user_friendships_v1(self, user_ids: List[str]) -> dict: + def user_friendships_v1(self, user_ids: List[str]) -> List[RelationshipShort]: """ Get user friendship status Parameters ---------- user_ids: List[str] - List of user id of an instagram account + List of user ID of an instagram account Returns ------- - dict + List[RelationshipShort] + List of RelationshipShorts with requested user_ids """ user_ids_str = ",".join(user_ids) result = self.private_request( @@ -316,7 +317,13 @@ def user_friendships_v1(self, user_ids: List[str]) -> dict: data={"user_ids": user_ids_str, "_uuid": self.uuid}, with_signature=False, ) - return result["friendship_statuses"] + assert result.get("status", "") == "ok" + + relationships = [] + for user_id, status in result.get("friendship_statuses", {}).items(): + relationships.append(RelationshipShort(user_id=user_id, **status)) + + return relationships def user_friendship_v1(self, user_id: str) -> Relationship: """ @@ -334,8 +341,10 @@ def user_friendship_v1(self, user_id: str) -> Relationship: """ try: - results = self.private_request(f"friendships/show/{user_id}/") - return Relationship(**results) + result = self.private_request(f"friendships/show/{user_id}/") + assert result.get("status", "") == "ok" + + return Relationship(user_id=user_id, **result) except ClientError as e: self.logger.exception(e) return None @@ -837,6 +846,69 @@ def user_unfollow(self, user_id: str) -> bool: self._users_following[self.user_id].pop(user_id, None) return result["friendship_status"]["following"] is False + def user_block(self, user_id: str, surface: str = "profile") -> bool: + """ + Block a User + + Parameters + ---------- + user_id: str + User ID of an Instagram account + surface: str, (optional) + Surface of block (deafult "profile", also can be "direct_thread_info") + + Returns + ------- + bool + A boolean value + """ + data = { + "surface": surface, + "is_auto_block_enabled": "false", + "user_id": user_id, + "_uid": self.user_id, + "_uuid": self.uuid, + } + if surface == "direct_thread_info": + data["client_request_id"] = self.request_id + + result = self.private_request( + f"friendships/block/{user_id}/", data) + assert result.get("status", "") == "ok" + + return result.get("friendship_status", {}).get("blocking") is True + + def user_unblock(self, user_id: str, surface: str = "profile") -> bool: + """ + Unlock a User + + Parameters + ---------- + user_id: str + User ID of an Instagram account + surface: str, (optional) + Surface of block (deafult "profile", also can be "direct_thread_info") + + Returns + ------- + bool + A boolean value + """ + data = { + "container_module": surface, + "user_id": user_id, + "_uid": self.user_id, + "_uuid": self.uuid, + } + if surface == "direct_thread_info": + data["client_request_id"] = self.request_id + + result = self.private_request( + f"friendships/unblock/{user_id}/", data) + assert result.get("status", "") == "ok" + + return result.get("friendship_status", {}).get("blocking") is False + def user_remove_follower(self, user_id: str) -> bool: """ Remove a follower diff --git a/instagrapi/types.py b/instagrapi/types.py index 02a20012..f768612c 100644 --- a/instagrapi/types.py +++ b/instagrapi/types.py @@ -418,6 +418,7 @@ def is_seen(self, user_id: str): class Relationship(BaseModel): + user_id: str blocking: bool followed_by: bool following: bool @@ -429,8 +430,16 @@ class Relationship(BaseModel): is_restricted: bool muting: bool outgoing_request: bool - status: str +class RelationshipShort(BaseModel): + user_id: str + following: bool + incoming_request: bool + is_bestie: bool + is_feed_favorite: bool + is_private: bool + is_restricted: bool + outgoing_request: bool class Highlight(BaseModel): pk: str # 17895485401104052