Rotating proxy

[Wathfea]

Hi,

theoretically, how can we implement a proxy rotation?
For example I have a list of proxies in txt.
When I start the program I login using the first proxy from the list.
I do some stuff then log out and log in again using an other proxy for doing other stuffs again?
Isn't this behaviour raise problems? To many logins from different places for example?

Or I misunderstand how to use this option @adw0rd ?

[adw0rd]

#299 and handle_exception

[Wathfea]

If I add this handle_exceptions part to my code it means in other parts for example where I follow a user can't catch any exceptions am I right?

So for clear view an exemple:

Bot.py

self.session = Client(delay_range=[min, max])
self.session.handle_exception = self.handle_exception
self.session.challenge_code_handler = self.challenge_code_handler
self.session.change_password_handler = self.change_password_handler

Follow.py

    def follow(self, user: Union[int, UserShort]) -> Union[tuple[None, Any], tuple[Exception, bool]]:
        try:
            followed = self.session.user_follow(
                user_id=self.session.user_id_from_username(username=user.username) if isinstance(user,
                                                                                                 UserShort) else user)
            return None, followed
        except Exception as error:
            print(f'[ERROR]: follow.py {error}.')
            self.logger.error(error)
            return error, False

And I'm not sure it is good to use a try catch in this part.
Yesterday I faced a situation when I got an error [ERROR]: challenge_required. But the self.session.challenge_code_handler = self.challenge_code_handler part of my code never triggered. And I think because I catched the exception before.
Am I understad the behavior right? I not need to try catch anything anywhere else just in the handle_exception method?

[Wathfea]

I think I miss something

Here is the implementation:

import email
import imaplib
import os
import random
import re
import tkinter as tk
import chime
import time

from typing import Tuple
from instagrapi import Client
from instagrapi.exceptions import ChallengeRequired, SelectContactPointRecoveryForm, RecaptchaChallengeForm, \
    PleaseWaitFewMinutes
from instagrapi.mixins.challenge import ChallengeChoice
from affybot.common.gmail_database_helper import GmailDatabaseHelper
from affybot.instagram.actions import Actions


class InstagramBase:
    def __init__(self, username: str = None, password: str = None, verification_code: str = None,
                 delay_range: Tuple[float, float] = None):
        if username is None:
            print('[ERROR]: Username has not been set.')
            return

        if password is None:
            print('[ERROR]: Password has not been set.')
            return

        if delay_range is not None:
            min, max = delay_range
        else:
            min, max = (1, 5)

        self.username = username
        self.session = Client(delay_range=[min, max])
        self.session.handle_exception = self.handle_exception
        self.session.challenge_code_handler = self.challenge_code_handler
        self.session.change_password_handler = self.change_password_handler

        if not os.path.exists(path=os.getcwd() + f'{os.path.sep}affybot{os.path.sep}files'):
            os.mkdir(path=os.getcwd() + f'{os.path.sep}affybot{os.path.sep}files')

        settings_path = os.getcwd() + f'{os.path.sep}affybot{os.path.sep}files{os.path.sep}settings.json'

        if os.path.exists(path=settings_path):
            print('Read stored settings')
            self.session.load_settings(path=settings_path)

            print('Login')
            logged_in = self.session.login(username=username, password=password)
            print(logged_in)
        else:
            print('Cretae session settings')
            if verification_code:
                print('Login with 2FA')
                logged_in = self.session.login(username=username, password=password,
                                               verification_code=verification_code)
            else:
                print('Login')
                logged_in = self.session.login(username=username, password=password)

            self.session.dump_settings(path=settings_path)

        print(
            f'[INFO]: Successfully logged in as: {self.session.username}.' if logged_in else f'[ERROR]: Failed to log in.')
        self.actions = Actions(session=self.session)

    def freeze(self, message, days=0, hours=0):
        chime.warning()
        tk.messagebox.showwarning(title='Error', message=message)

        if days > 0:
            time.sleep(days * 24 * 60 * 60)
        elif hours > 0:
            time.sleep(hours * 60)

    def handle_exception(self, client, e):
        print('Exception handler triggered')

        if isinstance(e, ChallengeRequired):
            try:
                client.challenge_resolve(client.last_json)
            except ChallengeRequired as e:
                self.freeze('Manual Challenge Required', days=2)
                raise e
            except (ChallengeRequired, SelectContactPointRecoveryForm, RecaptchaChallengeForm) as e:
                self.freeze(str(e), days=4)
                raise e

            return True
        elif isinstance(e, PleaseWaitFewMinutes):
            self.freeze(str(e), hours=1)
        raise e

    def challenge_code_handler(self, username, choice):
        print(f'challenge_code_handler triggered for {username} with {choice}')
        if choice == ChallengeChoice.SMS:
            chime.warning()
            tk.messagebox.showwarning(title='Error', message="Can't resolve SMS type challenge")
        elif choice == ChallengeChoice.EMAIL:
            return self.get_code_from_email()
        return False

    def get_code_from_email(self):
        gdh = GmailDatabaseHelper()
        (gemail, gpassword) = gdh.get_stored_settings()

        mail = imaplib.IMAP4_SSL("imap.gmail.com")
        mail.login(gemail, gpassword)
        mail.select("inbox")
        result, data = mail.search(None, "(UNSEEN)")
        assert result == "OK", "Error1 during get_code_from_email: %s" % result
        ids = data.pop().split()
        for num in reversed(ids):
            mail.store(num, "+FLAGS", "\\Seen")  # mark as read
            result, data = mail.fetch(num, "(RFC822)")
            assert result == "OK", "Error2 during get_code_from_email: %s" % result
            msg = email.message_from_string(data[0][1].decode())
            payloads = msg.get_payload()
            if not isinstance(payloads, list):
                payloads = [msg]
            code = None
            for payload in payloads:
                body = payload.get_payload(decode=True).decode()
                if "<div" not in body:
                    continue
                match = re.search(">([^>]*?({u})[^<]*?)<".format(u=self.username), body)
                if not match:
                    continue
                print("Match from email:", match.group(1))
                match = re.search(r">(\d{6})<", body)
                if not match:
                    print('Skip this email, "code" not found')
                    continue
                code = match.group(1)
                if code:
                    return code
        return False

    def change_password_handler(self):
        # Simple way to generate a random string
        chars = list("abcdefghijklmnopqrstuvwxyz1234567890!&£@#")
        password = "".join(random.sample(chars, 10))

        f = open("passwords.txt", "a")
        f.write(f"New instagram password: {password} \n")
        f.close()

        return password

I added handle_exceptions and challenge_code_handler too.
When I run the bot got the following error:

Exception handler triggered
Exception in thread Thread-2:
Traceback (most recent call last):
  File "/Users/davidperlusz/code/affy/.venv/lib/python3.9/site-packages/instagrapi/mixins/private.py", line 298, in _send_private_request
    response.raise_for_status()
  File "/Users/davidperlusz/code/affy/.venv/lib/python3.9/site-packages/requests/models.py", line 1021, in raise_for_status
    raise HTTPError(http_error_msg, response=self)
requests.exceptions.HTTPError: 400 Client Error: Bad Request for url: https://i.instagram.com/api/v1/friendships/create/6948444210/

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/Users/davidperlusz/code/affy/.venv/lib/python3.9/site-packages/instagrapi/mixins/private.py", line 434, in private_request
    self._send_private_request(endpoint, **kwargs)
  File "/Users/davidperlusz/code/affy/.venv/lib/python3.9/site-packages/instagrapi/mixins/private.py", line 330, in _send_private_request
    raise ChallengeRequired(**last_json)
instagrapi.exceptions.ChallengeRequired: challenge_required

So the exception handler triggered since the print worked, but the challange_handler seems to did nothing. What am I missing?