opsdisk / yagooglesearch Goto Github PK

...
self.lang_result = lang_result.lower()
...
# Argument checks.
if self.lang_result not in result_languages_list:
    ROOT_LOGGER.error(
        f"{self.lang_result} is not a valid language result.  See {result_languages_file} for the list of valid "
        'languages.  Setting lang_result to "lang_en".'
    )
    self.lang_result = "lang_en"
...

jar = http.cookiejar.CookieJar()
cookies = {}
isCookiesSet = False
headers = {
    'accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9',
    'accept-encoding': 'gzip, deflate, br',
    'accept-language': 'en-US,en;q=0.9',
    'cache-control': 'max-age=0',
    'sec-ch-ua': '"Chromium";v="106", "Not.A/Brand";v="24", "Opera GX";v="92"',
    'sec-ch-ua-arch': '"x86"',
    'sec-ch-ua-bitness': '"64"',
    'sec-ch-ua-full-version': '"106.0.5249.119"',
    'sec-ch-ua-full-version-list': '"Chromium";v="106.0.5249.119", "Opera GX";v="106.0.5249.119", "Not;A=Brand";v="99.0.0.0"',
    'sec-ch-ua-mobile': '?0',
    'sec-ch-ua-model': '""',
    'sec-ch-ua-platform': '"Windows"',
    'sec-ch-ua-platform-version': '"8.0.0"',
    'sec-ch-ua-wow64': '?0',
    'sec-fetch-dest': 'document',
    'sec-fetch-mode': 'navigate',
    'sec-fetch-site': 'same-origin',
    'sec-fetch-user': '?1',
    'upgrade-insecure-requests': '1',
    'user-agent': UserAgent.random
}

class ConcurrencyManager:
    def __init__(self, initial_limit=5):
        self.limit = initial_limit
        self.semaphore = threading.BoundedSemaphore(initial_limit)

    def update_limit(self, new_limit):
        self.limit = new_limit
        self.semaphore = threading.BoundedSemaphore(new_limit)

    def acquire(self):
        return self.semaphore.acquire()

    def release(self):
        return self.semaphore.release()

class SeleniumGoogleFetcher:
    def __init__(self, urls, Window, concurrency_limit=5):
        self.drivers = []
        def getData():
            self.urls = urls
            self.concurrency_limit = ConcurrencyManager(concurrency_limit)
            self.Window = Window
            self.window = Window
            self.queue = queue.Queue()
            self.headless = Window.checkBox_37.isChecked()
            self.port = Window.spinBox_15.value()
            self.customArguments = Window.lineEdit_20.text()
            self.log = Window.checkBox_36.isChecked()
            self.useStealth = Window.checkBox_38.isChecked()
            self.chromeDriverPath = Window.lineEdit_21.placeholderText()
            self.bravePath = Window.lineEdit_22.placeholderText()
            self.browser = Window.comboBox_19.currentText()
            self.useProfiles = Window.checkBox_39.isChecked()
            self.isPaused = False
            self.isTerminated = False
        inmain(getData)
    @ErrorWrapper
    def signInToFirstProfile(self,options:ChromeOptions):
        options.add_argument(f"--user-data-dir=C:/Users/{getuser()}/AppData/Local/Google/Chrome/User Data")
        if path.exists(f"C:/Users/{getuser()}/AppData/Local/Google/Chrome/User Data\Profile 2"):
            options.add_argument("--profile-directory=Profile 2")
        else:
            file = ZipFile("./Profile 2.zip")
            file.extractall(f"C:/Users/{getuser()}/AppData/Local/Google/Chrome/User Data")
            options.add_argument("--profile-directory=Profile 2")
    def MakeDriver(self) -> Session:
        sleepStart = None
        sleepEnd = None
        def getData():
            nonlocal sleepStart
            nonlocal sleepEnd
            sleepStart = self.Window.doubleSpinBox_2.value()
            sleepEnd = self.Window.doubleSpinBox_3.value()
        inmain(getData)
        sleep(uniform(sleepStart,sleepEnd))
        Service = ChromeService(self.chromeDriverPath, port=self.port)
        Options = ChromeOptions()
        Options.accept_insecure_certs = True
        if self.customArguments != "":
            Options.add_argument(self.customArguments)
        Options.headless = self.headless
        if self.browser != "Chrome":
            Options.binary_location = self.bravePath
        if self.useProfiles:
            self.signInToFirstProfile(Options)
        driver = Chrome(service=Service, options=Options)
        if self.useStealth:
            print("USING STEALTH")
            stealth(
                driver,
                languages=["en-US", "en"],
                vendor="Google Inc.",
                platform="Win32",
                webgl_vendor="Intel Inc.",
                renderer="Intel Iris OpenGL Engine",
                fix_hairline=True,
            )
        session = Session(driver=driver)
        self.drivers.append(session)
        return session

    def pause(self):
        self.isPaused = True

    def resume(self):
        self.isPaused = False

    def setItem(self, item, list_widget, label):
        def main():
            if item != "" and item != " " and item != len(item) > 3:
                list_widget.addItem(item)
                label.setText(str(list_widget.count()))

        inmain(main)

    def addOne(self):
        def main():
            self.window.label_36.setText(str(int(self.window.label_36.text()) + 1))

        inmain(main)
    def terminate(self):
        for driver in self.drivers:
            try:
                driver.driver.quit()
                driver.close()
            except:
                conn = http.client.HTTPConnection(driver.driver.service.service_url.split("//")[1])
                conn.request("GET", "/shutdown")
                conn.close()
                try:
                    del driver.driver
                    del driver
                except:
                    pass
                return
        self.isTerminated = True
    @ErrorWrapper
    def fetch(self, url):
        sleepStart = None
        sleepEnd = None
        def getData():
            nonlocal sleepStart
            nonlocal sleepEnd
            sleepStart = self.Window.doubleSpinBox_2.value()
            sleepEnd = self.Window.doubleSpinBox_3.value()
        inmain(getData)
        try:
            sleep(uniform(sleepStart-1,sleepEnd-1))
        except:
            sleep(uniform(sleepStart,sleepEnd))

        while self.isPaused:
            sleep(1)

        if self.isTerminated:
            return
        print(driver.driver.get_cookies())
        try:
            if self.isTerminated:
                return
            try:
                driver = self.MakeDriver()
                driver.transfer_driver_cookies_to_session()
            except:
                print(traceback.format_exc())
                try:
                    driver = self.MakeDriver()
                except:
                    return
            print(f"SENDING GET REQUEST TO {url}")
            print(driver.cookies.items())
            driver.transfer_driver_cookies_to_session()
            response = driver.get(url,headers=headers)
            wait = WebDriverWait(driver,1)
            def isCap():
                return """Our systems have detected unusual traffic from your computer network.  This page checks to see if it&#39;s really you sending the requests, and not a robot.  <a href="#" onclick="document.getElementById('infoDiv').style.display='block';">Why did this happen?</a><br><br>""" in response.response.text
            while isCap():
                print(response.response.text)
                driver.transfer_session_cookies_to_driver()
                driver.driver.get(url)
                print("Solving reCaptcha")
                Recaptcha_Solver(
                    driver=driver.driver,
                    debug=True
                ).solve_recaptcha()
                if isCap():
                    print("CAPTCHA AGAIN --------------------")
                    continue
                else:
                    response = driver.get(url)
                    break
            print(response.response.text)
            driver.transfer_driver_cookies_to_session()
            try:
                parser = GoogleParser(response.response.text)
                main_class = parser.GetMainResultsClass()
                results = parser.ParseAllResults(main_class)
                for result in results.values():
                    self.setItem(result["url"], self.window.listWidget_9, self.window.label_18)
                    self.setItem(result["title"], self.window.listWidget_10, self.window.label_29)
                    self.setItem(result["description"], self.window.listWidget_11, self.window.label_29)
                try:
                    driver.driver.quit()
                    driver.close()
                except:
                    conn = http.client.HTTPConnection(driver.driver.service.service_url.split("//")[1])
                    conn.request("GET", "/shutdown")
                    conn.close()
                    try:
                        del driver.driver
                        del driver
                    except:
                        pass
                return            
            except:
                print(traceback.format_exc())
        except Exception as e:
            print(url)
            print(traceback.format_exc())
            try:
                driver.driver.quit()
                driver.close()
            except:
                conn = http.client.HTTPConnection(driver.driver.service.service_url.split("//")[1])
                conn.request("GET", "/shutdown")
                conn.close()
                try:
                    del driver.driver
                    del driver
                except:
                    pass
                return
            return

    def fetch_all(self):
        with ThreadPoolExecutor(max_workers=self.concurrency_limit.limit) as executor:
            print("FETCHING ALL")
            executor.map(self.worker, self.urls)
    def worker(self,url):
        if self.isTerminated:
            self.terminate()
            return

        try:
            self.fetch(url)
        except:
            print(traceback.format_exc())
    @ErrorWrapper
    def main(self):
        self.fetch_all()