diff --git a/TwitchChannelPointsMiner/classes/TwitchLogin.py b/TwitchChannelPointsMiner/classes/TwitchLogin.py index d257a856..bfafd724 100644 --- a/TwitchChannelPointsMiner/classes/TwitchLogin.py +++ b/TwitchChannelPointsMiner/classes/TwitchLogin.py @@ -3,26 +3,45 @@ # The MIT License (MIT) import copy -import getpass +# import getpass import logging import os import pickle -import browser_cookie3 +# import webbrowser +# import browser_cookie3 + import requests from TwitchChannelPointsMiner.classes.Exceptions import ( BadCredentialsException, WrongCookiesException, ) -from TwitchChannelPointsMiner.constants import GQLOperations +from TwitchChannelPointsMiner.constants import CLIENT_ID, GQLOperations, USER_AGENTS, CLIENT_VERSION + +from datetime import datetime, timedelta, timezone +from time import sleep logger = logging.getLogger(__name__) +"""def interceptor(request) -> str: + if ( + request.method == 'POST' + and request.url == 'https://passport.twitch.tv/protected_login' + ): + import json + body = request.body.decode('utf-8') + data = json.loads(body) + data['client_id'] = CLIENT_ID + request.body = json.dumps(data).encode('utf-8') + del request.headers['Content-Length'] + request.headers['Content-Length'] = str(len(request.body))""" + class TwitchLogin(object): __slots__ = [ "client_id", + "device_id", "token", "login_check_result", "session", @@ -32,15 +51,18 @@ class TwitchLogin(object): "user_id", "email", "cookies", + "shared_cookies" ] - def __init__(self, client_id, username, user_agent, password=None): + def __init__(self, client_id, device_id, username, user_agent, password=None): self.client_id = client_id + self.device_id = device_id self.token = None self.login_check_result = False self.session = requests.session() self.session.headers.update( - {"Client-ID": self.client_id, "User-Agent": user_agent} + {"Client-ID": self.client_id, + "X-Device-Id": self.device_id, "User-Agent": user_agent} ) self.username = username self.password = password @@ -48,96 +70,109 @@ def __init__(self, client_id, username, user_agent, password=None): self.email = None self.cookies = [] + self.shared_cookies = [] def login_flow(self): logger.info("You'll have to login to Twitch!") post_data = { "client_id": self.client_id, - "undelete_user": False, - "remember_me": True, + "scopes": ( + "channel_read chat:read user_blocks_edit " + "user_blocks_read user_follows_edit user_read" + ) } - + # login-fix use_backup_flow = False + # use_backup_flow = True + while True: + logger.info("Trying the TV login method..") + + login_response = self.send_oauth_request( + "https://id.twitch.tv/oauth2/device", post_data) + + # { + # "device_code": "40 chars [A-Za-z0-9]", + # "expires_in": 1800, + # "interval": 5, + # "user_code": "8 chars [A-Z]", + # "verification_uri": "https://www.twitch.tv/activate" + # } + + if login_response.status_code != 200: + logger.error("TV login response is not 200. Try again") + break - for attempt in range(0, 25): - password = ( - getpass.getpass(f"Enter Twitch password for {self.username}: ") - if self.password in [None, ""] - else self.password - ) - - post_data["username"] = self.username - post_data["password"] = password - - while True: - # Try login without 2FA - login_response = self.send_login_request(post_data) - - if "captcha_proof" in login_response: - post_data["captcha"] = dict(proof=login_response["captcha_proof"]) - - if "error_code" in login_response: - err_code = login_response["error_code"] - if err_code in [3011, 3012]: # missing 2fa token - if err_code == 3011: - logger.info( - "Two factor authentication enabled, please enter token below." - ) - else: - logger.info("Invalid two factor token, please try again.") - - twofa = input("2FA token: ") - post_data["authy_token"] = twofa.strip() - continue - - elif err_code in [3022, 3023]: # missing 2fa token - if err_code == 3022: - logger.info("Login Verification code required.") - self.email = login_response["obscured_email"] - else: - logger.info( - "Invalid Login Verification code entered, please try again." - ) - - twofa = input( - f"Please enter the 6-digit code sent to {self.email}: " - ) - post_data["twitchguard_code"] = twofa.strip() - continue - - # invalid password, or password not provided - elif err_code in [3001, 3003]: - logger.info("Invalid username or password, please try again.") - - # If the password is loaded from run.py, require the user to fix it there. - if self.password not in [None, ""]: - raise BadCredentialsException( - "Username or password is incorrect." - ) - - # If the user didn't load the password from run.py we can just ask for it again. - break - elif err_code == 1000: - logger.info( - "Console login unavailable (CAPTCHA solving required)." - ) - use_backup_flow = True + login_response_json = login_response.json() + + if "user_code" in login_response_json: + user_code: str = login_response_json["user_code"] + now = datetime.now(timezone.utc) + device_code: str = login_response_json["device_code"] + interval: int = login_response_json["interval"] + expires_at = now + \ + timedelta(seconds=login_response_json["expires_in"]) + logger.info( + "Open https://www.twitch.tv/activate" + ) + logger.info( + f"and enter this code: {user_code}" + ) + logger.info( + f"Hurry up! It will expire in {int(login_response_json['expires_in'] / 60)} minutes!" + ) + # twofa = input("2FA token: ") + # webbrowser.open_new_tab("https://www.twitch.tv/activate") + + post_data = { + "client_id": CLIENT_ID, + "device_code": device_code, + "grant_type": "urn:ietf:params:oauth:grant-type:device_code", + } + + while True: + # sleep first, not like the user is gonna enter the code *that* fast + sleep(interval) + login_response = self.send_oauth_request( + "https://id.twitch.tv/oauth2/token", post_data) + if now == expires_at: + logger.error("Code expired. Try again") break + # 200 means success, 400 means the user haven't entered the code yet + if login_response.status_code != 200: + continue + # { + # "access_token": "40 chars [A-Za-z0-9]", + # "refresh_token": "40 chars [A-Za-z0-9]", + # "scope": [...], + # "token_type": "bearer" + # } + login_response_json = login_response.json() + if "access_token" in login_response_json: + self.set_token(login_response_json["access_token"]) + return self.check_login() + # except RequestInvalid: + # the device_code has expired, request a new code + # continue + # invalidate_after is not None + # account for the expiration landing during the request + # and datetime.now(timezone.utc) >= (invalidate_after - session_timeout) + # ): + # raise RequestInvalid() else: + if "error_code" in login_response: + err_code = login_response["error_code"] + logger.error(f"Unknown error: {login_response}") raise NotImplementedError( f"Unknown TwitchAPI error code: {err_code}" ) - if "access_token" in login_response: - self.set_token(login_response["access_token"]) - return self.check_login() - if use_backup_flow: break if use_backup_flow: + # self.set_token(self.login_flow_backup(password)) self.set_token(self.login_flow_backup()) return self.check_login() @@ -145,13 +180,92 @@ def login_flow(self): def set_token(self, new_token): self.token = new_token - self.session.headers.update({"Authorization": f"Bearer {self.token}"}) + self.session.headers.update({"Authorization": f"OAuth {self.token}"}) + + # def send_login_request(self, json_data): + def send_oauth_request(self, url, json_data): + # response = self.session.post("https://passport.twitch.tv/protected_login", json=json_data) + """response = self.session.post("https://passport.twitch.tv/login", json=json_data, headers={ + 'Accept': 'application/vnd.twitchtv.v3+json', + 'Accept-Encoding': 'gzip', + 'Accept-Language': 'en-US', + 'Content-Type': 'application/json; charset=UTF-8', + 'Host': 'passport.twitch.tv' + },)""" + response = self.session.post(url, data=json_data, headers={ + 'Accept': 'application/json', + 'Accept-Encoding': 'gzip', + 'Accept-Language': 'en-US', + "Cache-Control": "no-cache", + "Client-Id": CLIENT_ID, + "Host": "id.twitch.tv", + "Origin": "https://android.tv.twitch.tv", + "Pragma": "no-cache", + "Referer": "https://android.tv.twitch.tv/", + "User-Agent": USER_AGENTS["Android"]["TV"], + "X-Device-Id": self.device_id + },) + return response + + def login_flow_backup(self, password=None): + """Backup OAuth Selenium login + from undetected_chromedriver import ChromeOptions + import seleniumwire.undetected_chromedriver.v2 as uc + from selenium.webdriver.common.by import By + from time import sleep + + HEADLESS = False + + options = uc.ChromeOptions() + if HEADLESS is True: + options.add_argument('--headless') + options.add_argument('--log-level=3') + options.add_argument('--disable-web-security') + options.add_argument('--allow-running-insecure-content') + options.add_argument('--lang=en') + options.add_argument('--no-sandbox') + options.add_argument('--disable-gpu') + # options.add_argument("--user-agent=\"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/107.0.0.0 Safari/537.36\"") + # options.add_argument("--window-size=1920,1080") + # options.set_capability("detach", True) + + logger.info( + 'Now a browser window will open, it will login with your data.') + driver = uc.Chrome( + options=options, use_subprocess=True # , executable_path=EXECUTABLE_PATH + ) + driver.request_interceptor = interceptor + driver.get('https://www.twitch.tv/login') + + driver.find_element(By.ID, 'login-username').send_keys(self.username) + driver.find_element(By.ID, 'password-input').send_keys(password) + sleep(0.3) + driver.execute_script( + 'document.querySelector("#root > div > div.scrollable-area > div.simplebar-scroll-content > div > div > div > div.Layout-sc-nxg1ff-0.gZaqky > form > div > div:nth-child(3) > button > div > div").click()' + ) - def send_login_request(self, json_data): - response = self.session.post("https://passport.twitch.tv/login", json=json_data) - return response.json() + logger.info( + 'Enter your verification code in the browser and wait for the Twitch website to load, then press Enter here.' + ) + input() + + logger.info("Extracting cookies...") + self.cookies = driver.get_cookies() + # print(self.cookies) + # driver.close() + driver.quit() + self.username = self.get_cookie_value("login") + # print(f"self.username: {self.username}") + + if not self.username: + logger.error("Couldn't extract login, probably bad cookies.") + return False + + return self.get_cookie_value("auth-token")""" + + # logger.error("Backup login flow is not available. Use a VPN or wait a while to avoid the CAPTCHA.") + # return False - def login_flow_backup(self): """Backup OAuth login flow in case manual captcha solving is required""" browser = input( "What browser do you use? Chrome (1), Firefox (2), Other (3): " @@ -169,8 +283,11 @@ def login_flow_backup(self): cookie_jar = browser_cookie3.chrome(domain_name=twitch_domain) else: cookie_jar = browser_cookie3.firefox(domain_name=twitch_domain) + # logger.info(f"cookie_jar: {cookie_jar}") cookies_dict = requests.utils.dict_from_cookiejar(cookie_jar) + # logger.info(f"cookies_dict: {cookies_dict}") self.username = cookies_dict.get("login") + self.shared_cookies = cookies_dict return cookies_dict.get("auth-token") def check_login(self): @@ -183,14 +300,20 @@ def check_login(self): return self.login_check_result def save_cookies(self, cookies_file): + logger.info("Saving cookies to your computer..") cookies_dict = self.session.cookies.get_dict() + # print(f"cookies_dict2pickle: {cookies_dict}") cookies_dict["auth-token"] = self.token if "persistent" not in cookies_dict: # saving user id cookies cookies_dict["persistent"] = self.user_id + # old way saves only 'auth-token' and 'persistent' self.cookies = [] + # cookies_dict = self.shared_cookies + # print(f"cookies_dict2pickle: {cookies_dict}") for cookie_name, value in cookies_dict.items(): self.cookies.append({"name": cookie_name, "value": value}) + # print(f"cookies2pickle: {self.cookies}") pickle.dump(self.cookies, open(cookies_file, "wb")) def get_cookie_value(self, key): @@ -208,28 +331,40 @@ def load_cookies(self, cookies_file): def get_user_id(self): persistent = self.get_cookie_value("persistent") - user_id = ( - int(persistent.split("%")[0]) if persistent is not None else self.user_id - ) + user_id = self.user_id + if persistent is not None: + try: + user_id = int(persistent.split("%")[0]) + except ValueError: + logger.debug(f"Could not parse persistent cookie value: {persistent}") + if user_id is None: if self.__set_user_id() is True: return self.user_id return user_id def __set_user_id(self): - json_data = copy.deepcopy(GQLOperations.ReportMenuItem) - json_data["variables"] = {"channelLogin": self.username} - response = self.session.post(GQLOperations.url, json=json_data) + json_data = copy.deepcopy(GQLOperations.GetIDFromLogin) + json_data["variables"]["login"] = self.username + headers = { + "Client-Version": CLIENT_VERSION + } + response = self.session.post(GQLOperations.url, json=json_data, headers=headers) if response.status_code == 200: json_response = response.json() if ( "data" in json_response and "user" in json_response["data"] + and json_response["data"]["user"] is not None and json_response["data"]["user"]["id"] is not None ): self.user_id = json_response["data"]["user"]["id"] return True + else: + logger.error(f"Failed to get user_id for {self.username}. Response: {json_response}") + else: + logger.error(f"Failed to get user_id. Status: {response.status_code}, Content: {response.text}") return False def get_auth_token(self): diff --git a/example.py b/example.py index 9e4393a8..17060bab 100644 --- a/example.py +++ b/example.py @@ -6,7 +6,11 @@ from TwitchChannelPointsMiner.logger import LoggerSettings, ColorPalette from TwitchChannelPointsMiner.classes.Chat import ChatPresence from TwitchChannelPointsMiner.classes.Discord import Discord +from TwitchChannelPointsMiner.classes.Webhook import Webhook from TwitchChannelPointsMiner.classes.Telegram import Telegram +from TwitchChannelPointsMiner.classes.Matrix import Matrix +from TwitchChannelPointsMiner.classes.Pushover import Pushover +from TwitchChannelPointsMiner.classes.Gotify import Gotify from TwitchChannelPointsMiner.classes.Settings import Priority, Events, FollowersOrder from TwitchChannelPointsMiner.classes.entities.Bet import Strategy, BetSettings, Condition, OutcomeKeys, FilterCondition, DelayMode from TwitchChannelPointsMiner.classes.entities.Streamer import Streamer, StreamerSettings @@ -18,11 +22,17 @@ priority=[ # Custom priority in this case for example: Priority.STREAK, # - We want first of all to catch all watch streak from all streamers Priority.DROPS, # - When we don't have anymore watch streak to catch, wait until all drops are collected over the streamers - Priority.ORDER # - When we have all of the drops claimed and no watch-streak available, use the order priority (POINTS_ASCENDING, POINTS_DESCEDING) + Priority.ORDER # - When we have all of the drops claimed and no watch-streak available, use the order priority (POINTS_ASCENDING, POINTS_DESCENDING) ], + enable_analytics=False, # Disables Analytics if False. Disabling it significantly reduces memory consumption + disable_ssl_cert_verification=False, # Set to True at your own risk and only to fix SSL: CERTIFICATE_VERIFY_FAILED error + disable_at_in_nickname=False, # Set to True if you want to check for your nickname mentions in the chat even without @ sign logger_settings=LoggerSettings( save=True, # If you want to save logs in a file (suggested) console_level=logging.INFO, # Level of logs - use logging.DEBUG for more info + console_username=False, # Adds a username to every console log line if True. Also adds it to Telegram, Discord, etc. Useful when you have several accounts + auto_clear=True, # Create a file rotation handler with interval = 1D and backupCount = 7 if True (default) + time_zone="", # Set a specific time zone for console and file loggers. Use tz database names. Example: "America/Denver" file_level=logging.DEBUG, # Level of logs - If you think the log file it's too big, use logging.INFO emoji=True, # On Windows, we have a problem printing emoji. Set to false if you have a problem less=False, # If you think that the logs are too verbose, set this to True @@ -32,22 +42,52 @@ streamer_offline="red", # Read more in README.md BET_wiN=Fore.MAGENTA # Color allowed are: [BLACK, RED, GREEN, YELLOW, BLUE, MAGENTA, CYAN, WHITE, RESET]. ), - telegram=Telegram( # You can omit or leave None if you don't want to receive updates on Telegram - chat_id=123456789, # Chat ID to send messages @GiveChatId + telegram=Telegram( # You can omit or set to None if you don't want to receive updates on Telegram + chat_id=123456789, # Chat ID to send messages @getmyid_bot token="123456789:shfuihreuifheuifhiu34578347", # Telegram API token @BotFather - events=[Events.STREAMER_ONLINE, Events.STREAMER_OFFLINE, "BET_LOSE"], # Only these events will be sent to the chat + events=[Events.STREAMER_ONLINE, Events.STREAMER_OFFLINE, + Events.BET_LOSE, Events.CHAT_MENTION], # Only these events will be sent to the chat disable_notification=True, # Revoke the notification (sound/vibration) ), discord=Discord( webhook_api="https://discord.com/api/webhooks/0123456789/0a1B2c3D4e5F6g7H8i9J", # Discord Webhook URL - events=[Events.STREAMER_ONLINE, Events.STREAMER_OFFLINE, Events.BET_LOSE], # Only these events will be sent to the chat + events=[Events.STREAMER_ONLINE, Events.STREAMER_OFFLINE, + Events.BET_LOSE, Events.CHAT_MENTION], # Only these events will be sent to the chat + ), + webhook=Webhook( + endpoint="https://example.com/webhook", # Webhook URL + method="GET", # GET or POST + events=[Events.STREAMER_ONLINE, Events.STREAMER_OFFLINE, + Events.BET_LOSE, Events.CHAT_MENTION], # Only these events will be sent to the endpoint + ), + matrix=Matrix( + username="twitch_miner", # Matrix username (without homeserver) + password="...", # Matrix password + homeserver="matrix.org", # Matrix homeserver + room_id="...", # Room ID + events=[Events.STREAMER_ONLINE, Events.STREAMER_OFFLINE, Events.BET_LOSE], # Only these events will be sent + ), + pushover=Pushover( + userkey="YOUR-ACCOUNT-TOKEN", # Login to https://pushover.net/, the user token is on the main page + token="YOUR-APPLICATION-TOKEN", # Create a application on the website, and use the token shown in your application + priority=0, # Read more about priority here: https://pushover.net/api#priority + sound="pushover", # A list of sounds can be found here: https://pushover.net/api#sounds + events=[Events.CHAT_MENTION, Events.DROP_CLAIM], # Only these events will be sent + ), + gotify=Gotify( + endpoint="https://example.com/message?token=TOKEN", + priority=8, + events=[Events.STREAMER_ONLINE, Events.STREAMER_OFFLINE, + Events.BET_LOSE, Events.CHAT_MENTION], ) ), streamer_settings=StreamerSettings( make_predictions=True, # If you want to Bet / Make prediction follow_raid=True, # Follow raid to obtain more points claim_drops=True, # We can't filter rewards base on stream. Set to False for skip viewing counter increase and you will never obtain a drop reward from this script. Issue #21 + claim_moments=True, # If set to True, https://help.twitch.tv/s/article/moments will be claimed when available watch_streak=True, # If a streamer go online change the priority of streamers array and catch the watch screak. Issue #11 + community_goals=False, # If True, contributes the max channel points per stream to the streamers' community challenge goals chat=ChatPresence.ONLINE, # Join irc chat to increase watch-time [ALWAYS, NEVER, ONLINE, OFFLINE] bet=BetSettings( strategy=Strategy.SMART, # Choose you strategy! @@ -75,13 +115,15 @@ # For example, if in the mine function you don't provide any value for 'make_prediction' but you have set it on TwitchChannelPointsMiner instance, the script will take the value from here. # If you haven't set any value even in the instance the default one will be used +#twitch_miner.analytics(host="127.0.0.1", port=5000, refresh=5, days_ago=7) # Start the Analytics web-server + twitch_miner.mine( [ - Streamer("streamer-username01", settings=StreamerSettings(make_predictions=True , follow_raid=False , claim_drops=True , watch_streak=True , bet=BetSettings(strategy=Strategy.SMART , percentage=5 , stealth_mode=True, percentage_gap=20 , max_points=234 , filter_condition=FilterCondition(by=OutcomeKeys.TOTAL_USERS, where=Condition.LTE, value=800 ) ) )), - Streamer("streamer-username02", settings=StreamerSettings(make_predictions=False , follow_raid=True , claim_drops=False , bet=BetSettings(strategy=Strategy.PERCENTAGE , percentage=5 , stealth_mode=False, percentage_gap=20 , max_points=1234 , filter_condition=FilterCondition(by=OutcomeKeys.TOTAL_POINTS, where=Condition.GTE, value=250 ) ) )), - Streamer("streamer-username03", settings=StreamerSettings(make_predictions=True , follow_raid=False , watch_streak=True , bet=BetSettings(strategy=Strategy.SMART , percentage=5 , stealth_mode=False, percentage_gap=30 , max_points=50000 , filter_condition=FilterCondition(by=OutcomeKeys.ODDS, where=Condition.LT, value=300 ) ) )), - Streamer("streamer-username04", settings=StreamerSettings(make_predictions=False , follow_raid=True , watch_streak=True )), - Streamer("streamer-username05", settings=StreamerSettings(make_predictions=True , follow_raid=True , claim_drops=True , watch_streak=True , bet=BetSettings(strategy=Strategy.HIGH_ODDS , percentage=7 , stealth_mode=True, percentage_gap=20 , max_points=90 , filter_condition=FilterCondition(by=OutcomeKeys.PERCENTAGE_USERS, where=Condition.GTE, value=300 ) ) )), + Streamer("streamer-username01", settings=StreamerSettings(make_predictions=True , follow_raid=False , claim_drops=True , watch_streak=True , community_goals=False , bet=BetSettings(strategy=Strategy.SMART , percentage=5 , stealth_mode=True, percentage_gap=20 , max_points=234 , filter_condition=FilterCondition(by=OutcomeKeys.TOTAL_USERS, where=Condition.LTE, value=800 ) ) )), + Streamer("streamer-username02", settings=StreamerSettings(make_predictions=False , follow_raid=True , claim_drops=False , bet=BetSettings(strategy=Strategy.PERCENTAGE , percentage=5 , stealth_mode=False, percentage_gap=20 , max_points=1234 , filter_condition=FilterCondition(by=OutcomeKeys.TOTAL_POINTS, where=Condition.GTE, value=250 ) ) )), + Streamer("streamer-username03", settings=StreamerSettings(make_predictions=True , follow_raid=False , watch_streak=True , community_goals=True , bet=BetSettings(strategy=Strategy.SMART , percentage=5 , stealth_mode=False, percentage_gap=30 , max_points=50000 , filter_condition=FilterCondition(by=OutcomeKeys.ODDS, where=Condition.LT, value=300 ) ) )), + Streamer("streamer-username04", settings=StreamerSettings(make_predictions=False , follow_raid=True , watch_streak=True , )), + Streamer("streamer-username05", settings=StreamerSettings(make_predictions=True , follow_raid=True , claim_drops=True , watch_streak=True , community_goals=True , bet=BetSettings(strategy=Strategy.HIGH_ODDS , percentage=7 , stealth_mode=True, percentage_gap=20 , max_points=90 , filter_condition=FilterCondition(by=OutcomeKeys.PERCENTAGE_USERS, where=Condition.GTE, value=300 ) ) )), Streamer("streamer-username06"), Streamer("streamer-username07"), Streamer("streamer-username08"),