Add files via upload

hidden_chrome_driver.py

@@ -0,0 +1,106 @@
+__author__ = "https://stackoverflow.com/a/52883069/6120487"
+
+import errno
+import os
+import platform
+import subprocess
+import sys
+import time
+import warnings
+
+from selenium.common.exceptions import WebDriverException
+from selenium.webdriver.common import utils
+from selenium.webdriver.remote.webdriver import WebDriver as RemoteWebDriver
+from selenium.webdriver.chrome import service, webdriver, remote_connection
+
+
+class HiddenChromeService(service.Service):
+    def start(self):
+        try:
+            cmd = [self.path]
+            cmd.extend(self.command_line_args())
+
+            if platform.system() == 'Windows':
+                info = subprocess.STARTUPINFO()
+                info.dwFlags = subprocess.STARTF_USESHOWWINDOW
+                info.wShowWindow = 0  # SW_HIDE (6 == SW_MINIMIZE)
+            else:
+                info = None
+
+            self.process = subprocess.Popen(
+                cmd, env=self.env,
+                close_fds=platform.system() != 'Windows',
+                startupinfo=info,
+                stdout=self.log_file,
+                stderr=self.log_file,
+                stdin=subprocess.PIPE)
+        except TypeError:
+            raise
+        except OSError as err:
+            if err.errno == errno.ENOENT:
+                raise WebDriverException(
+                    "'%s' executable needs to be in PATH. %s" % (
+                        os.path.basename(self.path), self.start_error_message)
+                )
+            elif err.errno == errno.EACCES:
+                raise WebDriverException(
+                    "'%s' executable may have wrong permissions. %s" % (
+                        os.path.basename(self.path), self.start_error_message)
+                )
+            else:
+                raise
+        except Exception as e:
+            raise WebDriverException(
+                "Executable %s must be in path. %s\n%s" % (
+                    os.path.basename(self.path), self.start_error_message,
+                    str(e)))
+        count = 0
+        while True:
+            self.assert_process_still_running()
+            if self.is_connectable():
+                break
+            count += 1
+            time.sleep(1)
+            if count == 30:
+                raise WebDriverException("Can't connect to the Service %s" % (
+                    self.path,))
+
+
+class HiddenChromeWebDriver(webdriver.WebDriver):
+    def __init__(self, executable_path="chromedriver", port=0,
+                options=None, service_args=None,
+                desired_capabilities=None, service_log_path=None,
+                chrome_options=None, keep_alive=True):
+        if chrome_options:
+            warnings.warn('use options instead of chrome_options',
+                        DeprecationWarning, stacklevel=2)
+            options = chrome_options
+
+        if options is None:
+            # desired_capabilities stays as passed in
+            if desired_capabilities is None:
+                desired_capabilities = self.create_options().to_capabilities()
+        else:
+            if desired_capabilities is None:
+                desired_capabilities = options.to_capabilities()
+            else:
+                desired_capabilities.update(options.to_capabilities())
+
+        self.service = HiddenChromeService(
+            executable_path,
+            port=port,
+            service_args=service_args,
+            log_path=service_log_path)
+        self.service.start()
+
+        try:
+            RemoteWebDriver.__init__(
+                self,
+                command_executor=remote_connection.ChromeRemoteConnection(
+                    remote_server_addr=self.service.service_url,
+                    keep_alive=keep_alive),
+                desired_capabilities=desired_capabilities)
+        except Exception:
+            self.quit()
+            raise
+        self._is_remote = False

main.py

@@ -0,0 +1,159 @@
+from selenium import webdriver
+from selenium.webdriver.chrome.options import Options
+import logging
+from time import sleep
+import re
+import pickle
+import os
+import traceback
+from hidden_chrome_driver import HiddenChromeWebDriver
+
+
+PAGE_LOAD_WAIT_SECONDS = 3
+COOKIES_FILENAME = "./twitch-cookies.pkl"
+
+
+# проверка что чел стримит
+def main():
+    global twitch_streamer
+    twitch_streamer = input("Enter the streamer name: ")
+    print("Loading, please wait...")
+    create_webdriver(True)
+    check_login()
+    
+    if not streamer_exists():
+        print("This streamer does not exist.")
+        driver.quit()
+        return
+    
+    check_mature_content()
+    
+    try:
+        set_lowest_quality()
+        go_fullscreen()
+        while True:
+            check_for_credits_update()
+            check_for_bonus()
+            sleep(5)
+    except:
+        print("The streamer is offline currently.")
+        driver.quit()
+        return
+      
+
+
+def create_webdriver(headless):
+    global driver
+    
+    chrome_options = Options()
+    if headless:
+        chrome_options.add_argument("--headless")
+        chrome_options.add_argument("--window-size=1280,800");
+        #chrome_options.add_argument("--disable-gpu")
+    chrome_options.add_experimental_option('excludeSwitches', ['enable-logging'])
+    chrome_options.add_argument("--mute-audio")
+
+    
+    driver = HiddenChromeWebDriver(service_log_path='NUL', options=chrome_options)    
+    if headless:
+        url = f"https://www.twitch.com/{twitch_streamer}"
+    else:
+        url = "https://www.twitch.com"
+    driver.get(url)
+    load_cookies()
+    sleep(PAGE_LOAD_WAIT_SECONDS)
+    return driver
+
+
+def save_cookies():
+    pickle.dump(driver.get_cookies(), open(COOKIES_FILENAME, "wb"))
+
+
+def load_cookies():
+    if os.path.isfile(COOKIES_FILENAME):
+        cookies = pickle.load(open(COOKIES_FILENAME, "rb"))
+        for cookie in cookies:
+            driver.add_cookie(cookie)
+
+        driver.refresh()
+
+    
+
+def check_login():
+    while True:
+        login_button_search = driver.find_elements_by_css_selector("button[data-a-target='login-button']")
+        if len(login_button_search) > 0:
+            print("You'll have to login to Twitch, please wait.")
+            driver.quit()
+            create_webdriver(False)
+            input("Please login to Twitch and press Enter when you're done...")
+            save_cookies()
+            driver.quit()
+            print("Loading, please wait...")
+            create_webdriver(True)
+        else:
+            break
+
+
+def streamer_exists():
+    error_search = driver.find_elements_by_css_selector("a[data-test-selector='page-not-found__browse-channels-button']")
+    return len(error_search) == 0
+
+
+def check_mature_content():
+    accept_button_search = driver.find_elements_by_css_selector("button[data-a-target='player-overlay-mature-accept']")
+    if accept_button_search:
+        accept_button = accept_button_search[0]
+        accept_button.click()
+        sleep(1)
+
+    
+def set_lowest_quality():
+    settings_button = driver.find_element_by_css_selector("button[data-a-target='player-settings-button']")
+    settings_button.click()
+    sleep(0.1)
+    quality_button = driver.find_element_by_css_selector("button[data-a-target='player-settings-menu-item-quality']")
+    quality_button.click()
+
+    qualities = driver.find_elements_by_css_selector("div[data-a-target='player-settings-submenu-quality-option']")
+    qualities[-1].click()
+
+
+def go_fullscreen():
+    fullscreen_button = driver.find_element_by_css_selector("button[data-a-target='player-fullscreen-button']")
+    fullscreen_button.click()    
+
+
+def check_for_bonus():
+    bonus_button_search = driver.find_elements_by_css_selector("button[class='tw-button tw-button--success tw-interactive']")
+    if bonus_button_search:
+        bonus_button = bonus_button_search[0]
+        bonus_button.click()
+        print("Clicked the bonus button!")
+
+
+current_credits = -1
+
+def check_for_credits_update():
+    global current_credits
+    new_credits = get_credits()
+    if new_credits != current_credits:
+        current_credits = new_credits
+        print(f"Now you have {current_credits} credits!")
+
+
+def get_credits():
+    balance_div = driver.find_element_by_css_selector("div[data-test-selector='balance-string']")
+    balance_text_span = balance_div.find_element_by_class_name("tw-animated-number")
+    balance_text = balance_text_span.text
+    # убираем пробелы
+    balance_text = re.sub(r"\s+", "", balance_text, flags=re.UNICODE)
+    if len(balance_text) == 0:
+        return -1
+    else:
+        return int(balance_text)
+
+
+if __name__ == "__main__":
+    main()
+