From dd8141028a9809d37b5154eb68949816b73416be Mon Sep 17 00:00:00 2001 From: "Mikhail Andreev (adw0rd)" Date: Wed, 2 Dec 2020 03:24:37 +0300 Subject: [PATCH] Account registration [#23] Init SignUpMixin and SignUpTestCase --- instagrapi/__init__.py | 4 +- instagrapi/extractors.py | 2 +- instagrapi/signup.py | 205 +++++++++++++++++++++++++++++++++++++++ tests.py | 25 +++++ 4 files changed, 234 insertions(+), 2 deletions(-) create mode 100644 instagrapi/signup.py diff --git a/instagrapi/__init__.py b/instagrapi/__init__.py index 28bf3973..ce41fae2 100644 --- a/instagrapi/__init__.py +++ b/instagrapi/__init__.py @@ -21,6 +21,7 @@ from .account import Account from .direct import Direct from .location import LocationMixin +from .signup import SignUpMixin class Client( @@ -45,7 +46,8 @@ class Client( Collection, Account, Direct, - LocationMixin + LocationMixin, + SignUpMixin ): proxy = None logger = logging.getLogger("instagrapi") diff --git a/instagrapi/extractors.py b/instagrapi/extractors.py index acea22ea..daafecc6 100644 --- a/instagrapi/extractors.py +++ b/instagrapi/extractors.py @@ -138,7 +138,7 @@ def extract_user_short(data): """Extract User Short info """ data['pk'] = data.get("id", data.get("pk", None)) - assert data['pk'], 'User without pk "%s"' % data + assert data['pk'], f'User without pk "{data}"' return UserShort(**data) diff --git a/instagrapi/signup.py b/instagrapi/signup.py new file mode 100644 index 00000000..5b0c4942 --- /dev/null +++ b/instagrapi/signup.py @@ -0,0 +1,205 @@ +import time +import random +from uuid import uuid4 + +from .types import UserShort +from .extractors import extract_user_short + + +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 _ in range(5): + 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, password, email, signup_code, full_name, + year, month, 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, + **data + ) -> dict: + return self.private_request("accounts/create", { + "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": "#PWD_INSTAGRAM:4:1605642001:...", + "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": "cnJydHRnZ2dn...........qGkpFZpaojUi1vJX8=", + "force_sign_up_code": signup_code, + "waterfall_id": self.waterfall_id, + "password": password, + "one_tap_opt_in": "true", + **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() diff --git a/tests.py b/tests.py index 956afa8d..a487b383 100644 --- a/tests.py +++ b/tests.py @@ -9,6 +9,7 @@ from json.decoder import JSONDecodeError from instagrapi import Client +from instagrapi.utils import gen_password from instagrapi.types import ( User, UserShort, Media, MediaOembed, Comment, Collection, DirectThread, DirectMessage, Usertag, Location, Account @@ -868,5 +869,29 @@ def test_location_info_without_lat_lng(self): self.assertEqual(loc.name, 'In The Clouds') +class SignUpTestCase(unittest.TestCase): + + def test_signup(self): + cl = Client() + username = gen_password() + password = gen_password(12, symbols=True) + email = f'{username}@gmail.com' + phone_number = os.environ.get("IG_PHONE_NUMBER") + full_name = f'John {username}' + user = cl.signup( + username, password, email, phone_number, full_name, + year=random.randint(1980, 1990), + month=random.randint(1, 12), + day=random.randint(1, 30) + ) + self.assertIsInstance(user, UserShort) + for key, val in { + "username": username, + "full_name": full_name + }.items(): + self.assertEqual(getattr(user, key), val) + self.assertTrue(user.profile_pic_url.startswith("https://")) + + if __name__ == '__main__': unittest.main()