Get thread_id by user_id [#137] Added cl.direct_thread_by_participants

instagrapi/mixins/auth.py

@@ -13,7 +13,7 @@
 import requests
 
 from instagrapi import config
-from instagrapi.exceptions import PleaseWaitFewMinutes, ReloginAttemptExceeded
+from instagrapi.exceptions import PrivateError, PleaseWaitFewMinutes, ReloginAttemptExceeded
 from instagrapi.utils import generate_jazoest
 from instagrapi.zones import CET
 
@@ -286,7 +286,10 @@ def login_by_sessionid(self, sessionid: str) -> bool:
  self.settings = {"cookies": {"sessionid": sessionid}}
  self.init()
  user_id = re.search(r"^\d+", sessionid).group()
- user = self.user_info_v1(int(user_id))
+ try:
+ user = self.user_info_v1(int(user_id))
+ except PrivateError:
+ user = self.user_short_gql(int(user_id))
  self.username = user.username
  return True
 

instagrapi/mixins/direct.py

@@ -160,10 +160,10 @@ def direct_send(
  String to be posted on the thread
 
  user_ids: List[int]
- List of unique identifier of Users thread
+ List of unique identifier of Users id
 
  thread_ids: List[int]
- List of unique identifier of Direct Message thread
+ List of unique identifier of Direct Message thread id
 
  Returns
  -------
@@ -203,10 +203,10 @@ def direct_send_photo(
  Path to photo that will be posted on the thread
 
  user_ids: List[int]
- List of unique identifier of Users thread
+ List of unique identifier of Users id
 
  thread_ids: List[int]
- List of unique identifier of Direct Message thread
+ List of unique identifier of Direct Message thread id
 
  Returns
  -------
@@ -271,3 +271,30 @@ def direct_search(self, query: str) -> List[DirectShortThread]:
  for item in result.get('ranked_recipients', [])
  if 'thread' in item
  ]
+
+ def direct_thread_by_participants(self, user_ids: List[int]) -> DirectThread:
+ """
+ Get direct thread by participants
+
+ Parameters
+ ----------
+ user_ids: List[int]
+ List of unique identifier of Users id
+
+ Returns
+ -------
+ DirectThread
+ An object of DirectThread
+ """
+ recipient_users = dumps([int(uid) for uid in user_ids])
+ result = self.private_request(
+ "direct_v2/threads/get_by_participants/",
+ params={"recipient_users": recipient_users, "seq_id": 2580572, "limit": 20}
+ )
+ if 'thread' not in result:
+ raise DirectThreadNotFound(
+ f'Thread not found by recipient_users={recipient_users}',
+ user_ids=user_ids,
+ **self.last_json
+ )
+ return extract_direct_thread(result['thread'])

setup.py

@@ -30,7 +30,7 @@
 
 setup(
  name='instagrapi',
- version='1.6.3',
+ version='1.6.4',
  author='Mikhail Andreev',
  author_email='[email protected]',
  license='MIT',

tests.py

@@ -31,6 +31,7 @@
 )
 from instagrapi.zones import UTC
 from instagrapi.utils import generate_jazoest
+from instagrapi.exceptions import DirectThreadNotFound
 
 ACCOUNT_USERNAME = os.environ.get("IG_USERNAME", "instagrapi2")
 ACCOUNT_PASSWORD = os.environ.get("IG_PASSWORD", "yoa5af6deeRujeec")
@@ -967,6 +968,12 @@ def test_direct_thread(self):
  seen = self.api.direct_send_seen(thread_id=thread.id)
  self.assertEqual(seen.status, 'ok')
 
+ def test_direct_thread_by_participants(self):
+ try:
+ self.api.direct_thread_by_participants([12345])
+ except DirectThreadNotFound:
+ pass
+
 
 class ClientAccountTestCase(ClientPrivateTestCase):