Merge branch 'master' into master

instagrapi/__init__.py

@@ -34,6 +34,7 @@
     TopSearchesPublicMixin,
 )
 from instagrapi.mixins.share import ShareMixin
+from instagrapi.mixins.signup import SignUpMixin
 from instagrapi.mixins.story import StoryMixin
 from instagrapi.mixins.timeline import ReelsMixin
 from instagrapi.mixins.totp import TOTPMixin
@@ -78,6 +79,7 @@ class Client(
     CommentMixin,
     StoryMixin,
     PasswordMixin,
+    SignUpMixin,
     DownloadClipMixin,
     UploadClipMixin,
     ReelsMixin,

instagrapi/extractors.py

@@ -1,7 +1,7 @@
+import datetime
 import html
 import json
 import re
-import datetime
 from copy import deepcopy
 
 from .types import (
@@ -23,9 +23,9 @@
     ReplyMessage,
     Resource,
     Story,
+    StoryHashtag,
     StoryLink,
     StoryLocation,
-    StoryHashtag,
     StoryMedia,
     StoryMention,
     Track,

instagrapi/mixins/account.py

@@ -7,7 +7,7 @@
 from instagrapi.exceptions import ClientError, ClientLoginRequired
 from instagrapi.extractors import extract_account, extract_user_short
 from instagrapi.types import Account, UserShort
-from instagrapi.utils import dumps, gen_token
+from instagrapi.utils import dumps, gen_token, generate_signature
 
 
 class AccountMixin:
@@ -34,7 +34,11 @@ def reset_password(self, username: str) -> Dict:
                 "Accept": "*/*",
                 "Accept-Encoding": "gzip,deflate",
                 "Accept-Language": "en-US",
-                "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_13_6) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/11.1.2 Safari/605.1.15",
+                "User-Agent": (
+                    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_13_6) "
+                    "AppleWebKit/605.1.15 (KHTML, like Gecko) "
+                    "Version/11.1.2 Safari/605.1.15"
+                ),
             },
             proxies=self.public.proxies,
             timeout=self.request_timeout,
@@ -57,7 +61,7 @@ def account_info(self) -> Account:
         """
         result = self.private_request("accounts/current_user/?edit=true")
         return extract_account(result["user"])
-    
+
     def change_password(
         self,
         old_password: str,
@@ -96,15 +100,26 @@ def change_password(
             result = self.private_request("accounts/change_password/", data=data)
             return result
         except Exception as e:
+            self.logger.exception(e)
             return False
+
     def set_external_url(self, external_url) -> dict:
         """
         Set new biography
         """
-
-        signed_body = f"signed_body=SIGNATURE.%7B%22updated_links%22%3A%22%5B%7B%5C%22url%5C%22%3A%5C%22{external_url}%5C%22%2C%5C%22title%5C%22%3A%5C%22%5C%22%2C%5C%22link_type%5C%22%3A%5C%22external%5C%22%7D%5D%22%2C%22_uid%22%3A%22{self.user_id}%22%2C%22_uuid%22%3A%22{self.uuid}%22%7D"
+        data = dumps(
+            {
+                "updated_links": dumps(
+                    [{"url": external_url, "title": "", "link_type": "external"}]
+                ),
+                "_uid": self.user_id,
+                "_uuid": self.uuid,
+            }
+        )
         return self.private_request(
-            "accounts/update_bio_links/", data=signed_body, with_signature=False
+            "accounts/update_bio_links/",
+            data=generate_signature(data),
+            with_signature=False,
         )
 
     def account_set_private(self) -> bool:

instagrapi/mixins/public.py

@@ -139,11 +139,19 @@ def _send_public_request(
         try:
             if data is not None:  # POST
                 response = self.public.data(
-                    url, data=data, params=params, proxies=self.public.proxies, timeout=timeout
+                    url,
+                    data=data,
+                    params=params,
+                    proxies=self.public.proxies,
+                    timeout=timeout,
                 )
             else:  # GET
                 response = self.public.get(
-                    url, params=params, proxies=self.public.proxies, stream=stream, timeout=timeout
+                    url,
+                    params=params,
+                    proxies=self.public.proxies,
+                    stream=stream,
+                    timeout=timeout,
                 )
 
             if stream:

instagrapi/mixins/signup.py

@@ -0,0 +1,212 @@
+import base64
+import random
+import time
+from datetime import datetime
+from uuid import uuid4
+
+from instagrapi.extractors import extract_user_short
+from instagrapi.types import UserShort
+
+CHOICE_EMAIL = 1
+
+
+class SignUpMixin:
+    waterfall_id = str(uuid4())
+    adid = str(uuid4())
+    wait_seconds = 5
+
+    def signup(
+        self,
+        username: str,
+        password: str,
+        email: str,
+        phone_number: str,
+        full_name: str = "",
+        year: int = None,
+        month: int = None,
+        day: int = None,
+    ) -> UserShort:
+        self.get_signup_config()
+        check = self.check_email(email)
+        assert check.get("valid"), f"Email not valid ({check})"
+        assert check.get("available"), f"Email not available ({check})"
+        sent = self.send_verify_email(email)
+        assert sent.get("email_sent"), "Email not sent ({sent})"
+        # send code confirmation
+        code = ""
+        for attempt in range(1, 11):
+            code = self.challenge_code_handler(username, CHOICE_EMAIL)
+            if code:
+                break
+            time.sleep(self.wait_seconds * attempt)
+        print(
+            f'Enter code "{code}" for {username} '
+            f"({attempt} attempts, by {self.wait_seconds} seconds)"
+        )
+        signup_code = self.check_confirmation_code(email, code).get("signup_code")
+        retries = 0
+        kwargs = {
+            "username": username,
+            "password": password,
+            "email": email,
+            "signup_code": signup_code,
+            "full_name": full_name,
+            "year": year,
+            "month": month,
+            "day": day,
+        }
+        while retries < 3:
+            data = self.accounts_create(**kwargs)
+            if data.get("message") != "challenge_required":
+                break
+            if self.challenge_flow(data["challenge"]):
+                kwargs.update({"suggestedUsername": "", "sn_result": "MLA"})
+            retries += 1
+        return extract_user_short(data["created_user"])
+
+    def get_signup_config(self) -> dict:
+        return self.private_request(
+            "consent/get_signup_config/",
+            params={"guid": self.uuid, "main_account_selected": False},
+        )
+
+    def check_email(self, email) -> dict:
+        """Check available (free, not registred) email"""
+        return self.private_request(
+            "users/check_email/",
+            {
+                "android_device_id": self.device_id,
+                "login_nonce_map": "{}",
+                "login_nonces": "[]",
+                "email": email,
+                "qe_id": str(uuid4()),
+                "waterfall_id": self.waterfall_id,
+            },
+        )
+
+    def send_verify_email(self, email) -> dict:
+        """Send request to receive code to email"""
+        return self.private_request(
+            "accounts/send_verify_email/",
+            {
+                "phone_id": self.phone_id,
+                "device_id": self.device_id,
+                "email": email,
+                "waterfall_id": self.waterfall_id,
+                "auto_confirm_only": "false",
+            },
+        )
+
+    def check_confirmation_code(self, email, code) -> dict:
+        """Enter code from email"""
+        return self.private_request(
+            "accounts/check_confirmation_code/",
+            {
+                "code": code,
+                "device_id": self.device_id,
+                "email": email,
+                "waterfall_id": self.waterfall_id,
+            },
+        )
+
+    def check_age_eligibility(self, year, month, day):
+        return self.private.post(
+            "consent/check_age_eligibility/",
+            data={"_csrftoken": self.token, "day": day, "year": year, "month": month},
+        ).json()
+
+    def accounts_create(
+        self,
+        username: str,
+        password: str,
+        email: str,
+        signup_code: str,
+        full_name: str = "",
+        year: int = None,
+        month: int = None,
+        day: int = None,
+        **kwargs,
+    ) -> dict:
+        timestamp = datetime.now().strftime("%s")
+        nonce = f'{username}|{timestamp}|\xb9F"\x8c\xa2I\xaaz|\xf6xz\x86\x92\x91Y\xa5\xaa#f*o%\x7f'
+        sn_nonce = base64.encodebytes(nonce.encode()).decode().strip()
+        data = {
+            "is_secondary_account_creation": "true",
+            "jazoest": str(int(random.randint(22300, 22399))),  # "22341",
+            "tos_version": "row",
+            "suggestedUsername": "sn_result",
+            "do_not_auto_login_if_credentials_match": "false",
+            "phone_id": self.phone_id,
+            "enc_password": self.password_encrypt(password),
+            "username": str(username),
+            "first_name": str(full_name),
+            "day": str(day),
+            "adid": self.adid,
+            "guid": self.uuid,
+            "year": str(year),
+            "device_id": self.device_id,
+            "_uuid": self.uuid,
+            "email": email,
+            "month": str(month),
+            "sn_nonce": sn_nonce,
+            "force_sign_up_code": signup_code,
+            "waterfall_id": self.waterfall_id,
+            "password": password,
+            "one_tap_opt_in": "true",
+            **kwargs,
+        }
+        return self.private_request("accounts/create/", data)
+
+    def challenge_flow(self, data):
+        data = self.challenge_api(data)
+        while True:
+            if data.get("message") == "challenge_required":
+                data = self.challenge_captcha(data["challenge"])
+                continue
+            elif data.get("challengeType") == "SubmitPhoneNumberForm":
+                data = self.challenge_submit_phone_number(data)
+                continue
+            elif data.get("challengeType") == "VerifySMSCodeFormForSMSCaptcha":
+                data = self.challenge_verify_sms_captcha(data)
+                continue
+
+    def challenge_api(self, data):
+        resp = self.private.get(
+            f"https://i.instagram.com/api/v1{data['api_path']}",
+            params={
+                "guid": self.uuid,
+                "device_id": self.device_id,
+                "challenge_context": data["challenge_context"],
+            },
+        )
+        return resp.json()
+
+    def challenge_captcha(self, data):
+        g_recaptcha_response = self.captcha_resolve()
+        resp = self.private.post(
+            f"https://i.instagram.com{data['api_path']}",
+            data={"g-recaptcha-response": g_recaptcha_response},
+        )
+        return resp.json()
+
+    def challenge_submit_phone_number(self, data, phone_number):
+        api_path = data.get("navigation", {}).get("forward")
+        resp = self.private.post(
+            f"https://i.instagram.com{api_path}",
+            data={
+                "phone_number": phone_number,
+                "challenge_context": data["challenge_context"],
+            },
+        )
+        return resp.json()
+
+    def challenge_verify_sms_captcha(self, data, security_code):
+        api_path = data.get("navigation", {}).get("forward")
+        resp = self.private.post(
+            f"https://i.instagram.com{api_path}",
+            data={
+                "security_code": security_code,
+                "challenge_context": data["challenge_context"],
+            },
+        )
+        return resp.json()

instagrapi/mixins/story.py

@@ -5,8 +5,6 @@
 from typing import List
 from urllib.parse import urlparse
 
-import requests
-
 from instagrapi import config
 from instagrapi.exceptions import ClientNotFoundError, StoryNotFound, UserNotFound
 from instagrapi.extractors import (
@@ -132,7 +130,8 @@ def _userid_chunks():
             assert user_ids is not None
             user_ids_per_query = 50
             for i in range(0, len(user_ids), user_ids_per_query):
-                yield user_ids[i : i + user_ids_per_query]
+                end = i + user_ids_per_query
+                yield user_ids[i:end]
 
         stories_un = {}
         for userid_chunk in _userid_chunks():
@@ -296,7 +295,9 @@ def story_download_by_url(
         filename = "%s.%s" % (filename, fname.rsplit(".", 1)[1]) if filename else fname
         path = Path(folder) / filename
 
-        response = self._send_public_request(url, stream=True, timeout=self.request_timeout)
+        response = self._send_public_request(
+            url, stream=True, timeout=self.request_timeout
+        )
         response.raise_for_status()
 
         with open(path, "wb") as f:

instagrapi/types.py

@@ -99,10 +99,12 @@ class Account(TypesBaseModel):
 class UserShort(TypesBaseModel):
     def __hash__(self):
         return hash(self.pk)
+
     def __eq__(self, other):
         if isinstance(other, UserShort):
             return self.pk == other.pk
         return NotImplemented
+
     pk: str
     username: Optional[str] = None
     full_name: Optional[str] = ""

requirements-test.txt

@@ -4,10 +4,10 @@ isort==5.13.2
 bandit==1.7.8
 mike==2.0.0
 markdown-include==0.8.1
-mkdocs-material==9.5.17
+mkdocs-material==9.5.18
 mkdocs-minify-plugin==0.8.0
 mkdocstrings==0.24.3
 ./util/mkdocs-redirects
 ./util/mkdocstrings_patch_type_aliases
-pytest-xdist==3.5.0
+pytest-xdist==3.6.0
 pytest~=8.1.1

requirements.txt

@@ -1,5 +1,5 @@
 requests==2.31.0
 PySocks==1.7.1
-pydantic==2.6.4
+pydantic==2.7.1
 moviepy==1.0.3
 pycryptodomex==3.20.0