api.py

import json
import os
import glob
from random import uniform, randint
from time import sleep
from urllib.request import urlretrieve

from pydub import AudioSegment
from pydub.effects import normalize
from selenium import webdriver
from selenium.webdriver.common.action_chains import ActionChains
import undetected_chromedriver as uc
from selenium.common.exceptions import NoSuchElementException, TimeoutException
from selenium.webdriver.common.by import By
from selenium.webdriver.common.keys import Keys
from selenium.webdriver.support.expected_conditions import \
    presence_of_element_located as poel
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.chrome.service import Service
# Firefox support (Camoufox)
from selenium.webdriver.firefox.options import Options as FirefoxOptions
from selenium.webdriver.firefox.service import Service as GeckoService
from webdriver_manager.chrome import ChromeDriverManager
from webdriver_manager.firefox import GeckoDriverManager
from speech_recognition import AudioFile, Recognizer
from faster_whisper import WhisperModel
from user_agent import generate_user_agent

from colorize import Colors
from proxier import Proxier


class RecaptchaBypass:
    def __init__(self, debug=False, use_proxy=False, use_agent=False, headless=True):
        self.debug = debug
        self.use_proxy = use_proxy
        self.use_agent = use_agent
        self.headless = headless
        self.current_proxy = None
        self.proxier = Proxier() if use_proxy else None
        self._print_message(Colors.INFO, f'{Colors.blue}Loading Whisper model (tiny.en)...{Colors.reset}')
        self.whisper_model = WhisperModel("tiny.en", device="cpu", compute_type="int8")
        self.browser = self._get_chromedriver(use_proxy, use_agent, headless)
        self.wait = WebDriverWait(self.browser, 30)

    def _get_chromedriver(self, use_proxy, use_agent, headless):
        camoufox_cache = os.path.expanduser('~/.cache/camoufox')
        camoufox_bin = None
        if os.path.isdir(camoufox_cache):
            candidates = [
                os.path.join(camoufox_cache, 'camoufox-bin'),
                os.path.join(camoufox_cache, 'camoufox'),
                os.path.join(camoufox_cache, 'browser', 'camoufox')
            ]
            for c in candidates:
                if os.path.isfile(c) and os.access(c, os.X_OK):
                    camoufox_bin = c
                    break
        
        if camoufox_bin:
            self._print_message(Colors.INFO, f'{Colors.purple}Camoufox binary detected {Colors.red}>{Colors.reset} {camoufox_bin}')
            gecko_path = '/usr/bin/geckodriver'
            if not os.path.exists(gecko_path):
                gecko_path = GeckoDriverManager().install()

            fox_opts = FirefoxOptions()
            os.environ['MOZ_NODEJS'] = '1'
            if headless:
                fox_opts.add_argument('-headless')
            
            fox_opts.binary_location = camoufox_bin
            fox_opts.set_preference('security.sandbox.content.level', 0)
            fox_opts.set_preference('dom.webdriver.enabled', False)
            fox_opts.set_preference('useprivacy', True)
            fox_opts.set_preference('general.useragent.override', generate_user_agent())
            fox_opts.set_preference('intl.accept_languages', 'en-US, en')
            fox_opts.set_preference('media.volume_scale', '0.0') 
            
            if use_proxy:
                proxy_str = self._setup_proxy()
                if '://' in proxy_str:
                    addr = proxy_str.split('://')[1]
                    host, port = addr.split(':')
                    fox_opts.set_preference('network.proxy.type', 1)
                    fox_opts.set_preference('network.proxy.http', host)
                    fox_opts.set_preference('network.proxy.http_port', int(port))
                    fox_opts.set_preference('network.proxy.ssl', host)
                    fox_opts.set_preference('network.proxy.ssl_port', int(port))

            service = GeckoService(gecko_path)
            return webdriver.Firefox(service=service, options=fox_opts)

        chrome_options = Options()
        if headless: chrome_options.add_argument('--headless=new')
        return uc.Chrome(options=chrome_options)

    def _print_message(self, level, message):
        print(f'{level} {message}')

    def _quit_browser(self):
        try: self.browser.quit()
        except: pass

    def _human_mouse_movement(self):
        try:
            actions = ActionChains(self.browser)
            for _ in range(randint(2, 4)):
                actions.move_by_offset(randint(-20, 20), randint(-20, 20)).pause(uniform(0.1, 0.2))
            actions.perform()
        except: pass

    def _solve_logic(self, i):
        if self._is_block():
            return {'status': 'blocked'}

        # Click Checkbox
        self._print_message(Colors.INFO, f'{Colors.blue}Getting recaptcha checkbox..{Colors.reset}')
        iframe = self.wait.until(poel((By.CSS_SELECTOR, "iframe[src*='recaptcha/api2/anchor']")))
        self.browser.switch_to.frame(iframe)
        check_box = self.wait.until(poel((By.ID, "recaptcha-anchor")))
        self._human_mouse_movement()
        ActionChains(self.browser).move_to_element(check_box).pause(uniform(0.5, 1.0)).click().perform()
        self.browser.switch_to.default_content()

        sleep(uniform(2, 4))
        
        if self._is_block():
            return {'status': 'blocked'}

        # Check if solved automatically
        token = self._get_token_from_page()
        if token: return {'status': 'solved', 'token': token}

        # Click Audio
        self._print_message(Colors.INFO, f'{Colors.blue}Getting recaptcha challenge..{Colors.reset}')
        self.wait.until(lambda d: len(d.find_elements(By.CSS_SELECTOR, "iframe[src*='recaptcha/api2/bframe']")) > 0)
        iframe = self.browser.find_element(By.CSS_SELECTOR, "iframe[src*='recaptcha/api2/bframe']")
        self.browser.switch_to.frame(iframe)
        
        if self._is_block():
            self.browser.switch_to.default_content()
            return {'status': 'blocked'}

        audio_button = self.wait.until(poel((By.ID, 'recaptcha-audio-button')))
        ActionChains(self.browser).move_to_element(audio_button).pause(uniform(0.5, 1.0)).click().perform()
        
        # Get Audio URL
        download_link = self.wait.until(poel((By.CLASS_NAME, 'rc-audiochallenge-tdownload-link')))
        audio_url = download_link.get_attribute('href')
        self.browser.switch_to.default_content()

        # Transcribe
        phrase_text, _ = self._audio_to_text(audio_url)
        self._print_message(Colors.INFO, f'{Colors.purple}Phrase Text {Colors.red}> {Colors.green}{phrase_text}{Colors.reset}')
        
        # Enter Phrase
        self.browser.switch_to.frame(self.browser.find_element(By.CSS_SELECTOR, "iframe[src*='recaptcha/api2/bframe']"))
        input_field = self.wait.until(poel((By.ID, 'audio-response')))
        for char in phrase_text:
            input_field.send_keys(char)
            sleep(uniform(0.05, 0.15))
        sleep(uniform(0.5, 1))
        input_field.send_keys(Keys.ENTER)
        self.browser.switch_to.default_content()
        
        sleep(uniform(3, 5))
        token = self._get_token_from_page()
        if token:
            return {'status': 'solved', 'token': token}
        return {'status': 'failed'}

    def get_token(self, url, max_retries=5):
        for i in range(max_retries):
            self._print_message(Colors.INFO, f'{Colors.blue}Token retrieval attempt {i+1}/{max_retries}..{Colors.reset}')
            try:
                if i > 0:
                    self._quit_browser()
                    self.browser = self._get_chromedriver(self.use_proxy, self.use_agent, self.headless)
                    self.wait = WebDriverWait(self.browser, 30)

                self.browser.get(url)
                sleep(uniform(5, 7))
                result = self._solve_logic(i)
                
                if result['status'] == 'solved':
                    self._print_message(Colors.SUCCESS, f'Token retrieved! 📍\n{Colors.white}{result["token"]}{Colors.reset}\n')
                    token = result['token']
                    self._quit_browser()
                    return {'success': True, 'token': token}
                elif result['status'] == 'blocked':
                    if self.use_proxy and self.current_proxy: self.proxier.remove_proxy(self.current_proxy)
                    continue
            except Exception as e:
                self._print_message(Colors.FAILED, f'Error: {str(e)}')
                continue
        self._quit_browser()
        return {'success': False, 'error': 'Max retries reached'}

    def bypass_recaptcha_v2(self, url, max_retries=5):
        for i in range(max_retries):
            self._print_message(Colors.INFO, f'{Colors.blue}Bypass attempt {i+1}/{max_retries}..{Colors.reset}')
            try:
                if i > 0:
                    self._quit_browser()
                    self.browser = self._get_chromedriver(self.use_proxy, self.use_agent, self.headless)
                    self.wait = WebDriverWait(self.browser, 30)

                self.browser.get(url)
                sleep(uniform(5, 7))
                solve_res = self._solve_logic(i)
                
                if solve_res['status'] == 'solved':
                    token = solve_res['token']
                    self._print_message(Colors.SUCCESS, 'V2 Bypassed! 📍')
                    self._print_message(Colors.INFO, f'Token: {token[:100]}...')
                    
                    # Try to click submit if it's the demo page
                    if 'google.com/recaptcha/api2/demo' in self.browser.current_url:
                        try:
                            submit_btn = self.browser.find_element(By.ID, 'recaptcha-demo-submit')
                            self.browser.execute_script("arguments[0].click();", submit_btn)
                            sleep(3)
                        except: pass
                    
                    self._quit_browser()
                    return {'success': True, 'token': token}
                
                if solve_res['status'] == 'blocked' and self.use_proxy:
                    self.proxier.remove_proxy(self.current_proxy)
                self._quit_browser()
            except Exception as e:
                self._print_message(Colors.FAILED, f'Error: {str(e)}')
                self._quit_browser()
        return {'success': False}

    def _get_token_from_page(self):
        try:
            textarea = self.browser.find_element(By.ID, "g-recaptcha-response")
            token = textarea.get_attribute("value")
            if token and len(token) > 100: return token
        except: pass
        return None

    def _audio_to_text(self, audio_url):
        filename = 'audio.mp3'
        wav_output = 'audio.wav'
        import urllib.request
        opener = urllib.request.build_opener()
        opener.addheaders = [('User-Agent', generate_user_agent())]
        urllib.request.install_opener(opener)
        urlretrieve(audio_url, filename)
        AudioSegment.from_mp3(filename).export(wav_output, format='wav')
        segments, _ = self.whisper_model.transcribe(wav_output)
        phrase = " ".join([s.text for s in segments]).strip()
        if not self.debug:
            for f in [filename, wav_output]:
                if os.path.exists(f): os.remove(f)
        return phrase, wav_output

    def _is_block(self):
        try:
            txt = self.browser.page_source.lower()
            if "try again later" in txt or "automated queries" in txt: return True
        except: pass
        return None