main.py

import os.path
import time
import uuid

import cloudscraper

import lib
from emailprovider.kopeechka import KopeechkaService
from emailprovider.base import EmailException, EmailErrorType, EmailMatch
from lib import capcha
from lib.capcha.capsolver import Capsolver
from lib.com import readconf
from lib.com.reqbase import BaseReClient

from lib.const import LocalConf
from lib.emailclient import EmailClientCollector
from lib.exceptions import InvalidEmail, FailedToCreate, BadProxy
from lib.core import TwitterAccount
from lib.models import Proxy
from configparser import ConfigParser
from threading import Lock, Event
from concurrent.futures import ThreadPoolExecutor, as_completed
from os import path

import requests
import random
import re
import typing as tp


class TwitterMane(BaseReClient):
    def __init__(self):
        """cfg = ConfigParser()
        cfg.SECTCRE = re.compile(r"\[ *(?P<header>[^]]+?) *\]")
        cfg.read(os.path.join(LocalConf.ASSETS_DIR, "config.toml"))"""
        """
            threads = cfg.get("app", "threads")
            ANYCAP_APIKEY = cfg.getboolean("capsolver", "enabled")
            KS_APIKEY = cfg.get("yescap", "ks_apikey")
            THREADS = cfg.getint("yescap", "threads")
            THREADS = cfg.getint("yhapi", "threads")
            CAPTCHA_SERVICE = cfg.get("run-settings", "captcha_service").lower()
            EMAIL_SERVICE = cfg.get("run-settings", "email_service").lower()
            AVAILABLE_CAPTCHA_SERVICES = ["anycaptcha"]
            AVAILABLE_EMAIL_SERVICES = ["kopeechka", "hotmailbox"]
        """
        self.configurations = readconf(
            file_name="resources.json",
            required_values=['situation', 'clash_proxy', 'capsolver']
        )

        # self._ks = KopeechkaService(self.configurations)
        self._cap = Capsolver(self.configurations["capsolver"])
        """
        self._cap_task = BaseTask(
        TaskType.FUNCAPTCHA,
        "2CB16598-CB82-4CF7-B332-5990DB66F3AB",
        "https://twitter.com/i/flow/signup")
        """

        self._threads = self.configurations["thread"]
        self._lock = Lock()
        self._event = Event()
        self._jobs = 500
        self._proxies: "tp.List[Proxy]" = list()
        self._counter = 0
        self.urlreq__init__()

    def _load_proxies(self):
        with open(os.path.join(LocalConf.ASSETS_DIR, "proxies.txt")) as fp:
            for proxy in fp.read().split("\n"):
                user, pswd, host, port = re.search(r"(.+):(.+)@(.+):(.+)", proxy.strip()).groups()
                self._proxies.append(Proxy(user, pswd, host, port))

    def _remove_proxy(self, proxy: "Proxy"):
        with self._lock:
            if proxy in self._proxies:
                self._proxies.remove(proxy)

    def _get_proxy(self) -> "Proxy":
        with self._lock:
            if len(self._proxies) > 0:
                return random.choice(self._proxies)

            self._event.set()
            print("[*] Exiting because proxy balance is not enough.")

    @staticmethod
    def _append_success_file(line: str):
        file_path = os.path.join(LocalConf.CACHE_PATH, "accounts_twitter.txt")
        with open(file_path, "a") as fp:
            fp.write(line + "\n")

    def _handle_capcha_exception(self, exception: capcha.base.CaptchaException):
        self._event.set()
        print("[*] Exiting because captcha balance is not enough.")

    def _handle_cloudf_capcha_exception(self, exception: cloudscraper.exceptions.CaptchaException):
        self._event.set()
        print("[*] Exiting because captcha balance is not enough.")

    def _handle_email_exception(self, exception: "EmailException"):
        if exception.error_type == EmailErrorType.NO_BALANCE:
            self._event.set()
            print("[*] Exiting because email balance is not enough.")

    def _worker(self):
        email = ""
        emailid = ""
        done = False
        mail = EmailClientCollector(self.session, self.agent_x, self.configurations)
        if not self._event.is_set():
            proxy = self._get_proxy()

            while True:
                try:
                    email, emailid = mail.get_email_address()
                    print(email)
                except lib.emailclient.tempemailbase.NewEmailCodeNeeded:
                    print('mail server down. try different email system...')
                    time.sleep(5)
                    continue
                break

            while not self._event.is_set() and not done:
                try:
                    tw = TwitterAccount(email.email, email.password, proxy)
                    tw.name += f"{random.randint(100, 500)}.{random.choice(['eth', 'sol'])}"
                    tw.init()

                    captcha = self._cap.get_task_result(self._cap.create_task(self._cap_task), 100.0)
                    print("[*] Captcha fetched successfully: %s..." % captcha[:18])
                    code = self._ks.get_otp(email, EmailMatch(r"\s(\d{6})\s"), 120.0)

                    tw.submit_verification_code(code, captcha)
                    tw.finalize()

                    self._append_success_file(f"{tw.username}:{email.password}:{email.email}:{tw.get_auth_token()}")
                    print("[*] New: %s" % tw.username)

                    with self._lock:
                        self._counter += 1
                        print("[*] Counter is at: %d" % self._counter)
                    done = True

                except capcha.base.CaptchaException as exception:
                    self._handle_capcha_exception(exception)
                except cloudscraper.exceptions.CaptchaException as exception:
                    self._handle_cloudf_capcha_exception(exception)
                except (InvalidEmail, FailedToCreate):
                    done = True

                except (BadProxy, requests.exceptions.ProxyError, requests.exceptions.ConnectionError):
                    self._remove_proxy(proxy)
                    proxy = self._get_proxy()

    def run(self):
        self._load_proxies()

        jobs = []

        while (
                not self._event.is_set() and
                self._cap.get_balance() > 0
        ):
            with ThreadPoolExecutor(self._threads) as executor:
                for _ in range(self._jobs):
                    job = executor.submit(self._worker)
                    jobs.append(job)

                for job in as_completed(jobs):
                    job.result()

                jobs.clear()


if __name__ == "__main__":
    main = TwitterMane()

    if (
            not (CAPTCHA_SERVICE in AVAILABLE_CAPTCHA_SERVICES) or
            not (EMAIL_SERVICE in AVAILABLE_EMAIL_SERVICES)
    ):
        print("[*] Something went wrong, cannot start with these settings... please check config.ini file.")
    else:
        main.run()