From 07515e92c9929f49ba005cf4dd35d5b5bf9d269e Mon Sep 17 00:00:00 2001 From: Alessandro Maggio Date: Sun, 7 Feb 2021 15:54:31 +0100 Subject: [PATCH 01/10] If the user is online but the last updated was performed more than 10 minutes ago attempt to perform a force update. #54 --- TwitchChannelPointsMiner/classes/Twitch.py | 13 ++++++++++++- TwitchChannelPointsMiner/classes/entities/Stream.py | 5 ++++- 2 files changed, 16 insertions(+), 2 deletions(-) diff --git a/TwitchChannelPointsMiner/classes/Twitch.py b/TwitchChannelPointsMiner/classes/Twitch.py index b9595a0..abe3d8f 100644 --- a/TwitchChannelPointsMiner/classes/Twitch.py +++ b/TwitchChannelPointsMiner/classes/Twitch.py @@ -234,6 +234,12 @@ class Twitch(object): ) ] + for index in streamers_index: + if (streamers[index].stream.update_elapsed() / 60) > 10: + # Why this user It's currently online but the last updated was more than 10minutes ago? + # Please perform a manually update and check if the user it's online + self.check_streamer_online(streamers[index]) + """ Check if we need need to change priority based on watch streak Viewers receive points for returning for x consecutive streams. @@ -266,7 +272,12 @@ class Twitch(object): while len(streamers_watching) < 2 and len(streamers_index) > 1: another_streamer_index = streamers_index.pop(0) if another_streamer_index not in streamers_watching: - streamers_watching.append(another_streamer_index) + try: + streamers_watching.append(another_streamer_index) + except requests.exceptions.ConnectionError as e: + logger.error( + f"Error while trying to perform a force for streamer's update: {e}" + ) """ Twitch has a limit - you can't watch more than 2 channels at one time. diff --git a/TwitchChannelPointsMiner/classes/entities/Stream.py b/TwitchChannelPointsMiner/classes/entities/Stream.py index affce68..59ddb1f 100644 --- a/TwitchChannelPointsMiner/classes/entities/Stream.py +++ b/TwitchChannelPointsMiner/classes/entities/Stream.py @@ -63,7 +63,10 @@ class Stream(object): return None if self.game in [{}, None] else self.game["name"] def update_required(self): - return self.__last_update == 0 or (time.time() - self.__last_update) >= 120 + return self.__last_update == 0 or self.update_elapsed() >= 120 + + def update_elapsed(self): + return 0 if self.__last_update == 0 else (time.time() - self.__last_update) def init_watch_streak(self): self.watch_streak_missing = True From fb36fa3dbbf67be5b1ad79ea05d6c52459d41f96 Mon Sep 17 00:00:00 2001 From: Alessandro Maggio Date: Sun, 7 Feb 2021 16:19:39 +0100 Subject: [PATCH 02/10] On Exception is_closed=True. Rety after x random minutes if the internet connection It's not available --- .../classes/TwitchWebSocket.py | 10 +++++++--- TwitchChannelPointsMiner/classes/WebSocketsPool.py | 14 +++++++++++++- TwitchChannelPointsMiner/utils.py | 10 ++++++++++ 3 files changed, 30 insertions(+), 4 deletions(-) diff --git a/TwitchChannelPointsMiner/classes/TwitchWebSocket.py b/TwitchChannelPointsMiner/classes/TwitchWebSocket.py index aaf595c..f87aff1 100644 --- a/TwitchChannelPointsMiner/classes/TwitchWebSocket.py +++ b/TwitchChannelPointsMiner/classes/TwitchWebSocket.py @@ -3,6 +3,7 @@ import logging import time from websocket import WebSocketApp +from websocket.exceptions import WebSocketConnectionClosedException from TwitchChannelPointsMiner.utils import create_nonce @@ -52,9 +53,12 @@ class TwitchWebSocket(WebSocketApp): self.last_ping = time.time() def send(self, request): - request_str = json.dumps(request, separators=(",", ":")) - logger.debug(f"#{self.index} - Send: {request_str}") - super().send(request_str) + try: + request_str = json.dumps(request, separators=(",", ":")) + logger.debug(f"#{self.index} - Send: {request_str}") + super().send(request_str) + except WebSocketConnectionClosedException: + self.is_closed = True def elapsed_last_pong(self): return (time.time() - self.last_pong) // 60 diff --git a/TwitchChannelPointsMiner/classes/WebSocketsPool.py b/TwitchChannelPointsMiner/classes/WebSocketsPool.py index 881e2d4..289cef8 100644 --- a/TwitchChannelPointsMiner/classes/WebSocketsPool.py +++ b/TwitchChannelPointsMiner/classes/WebSocketsPool.py @@ -12,7 +12,11 @@ from TwitchChannelPointsMiner.classes.entities.Raid import Raid from TwitchChannelPointsMiner.classes.Exceptions import TimeBasedDropNotFound from TwitchChannelPointsMiner.classes.TwitchWebSocket import TwitchWebSocket from TwitchChannelPointsMiner.constants import WEBSOCKET -from TwitchChannelPointsMiner.utils import _millify, get_streamer_index +from TwitchChannelPointsMiner.utils import ( + _millify, + check_internet_connection, + get_streamer_index, +) logger = logging.getLogger(__name__) @@ -32,6 +36,7 @@ class WebSocketsPool: """ def submit(self, topic): + # Check if we need to create a new WebSocket instance if self.ws == [] or self.ws[-1] is None or len(self.ws[-1].topics) >= 50: self.append_new_websocket() @@ -112,6 +117,13 @@ class WebSocketsPool: ) time.sleep(30) + while check_internet_connection() is False: + random_sleep = random.randint(1, 3) + logger.warning( + f"#{ws.index} - No internet connection available! Retry after {random_sleep}m" + ) + time.sleep(random_sleep * 60) + self = ws.parent_pool self.ws[ws.index] = None for topic in ws.topics: diff --git a/TwitchChannelPointsMiner/utils.py b/TwitchChannelPointsMiner/utils.py index 26d2483..eb97098 100644 --- a/TwitchChannelPointsMiner/utils.py +++ b/TwitchChannelPointsMiner/utils.py @@ -1,5 +1,6 @@ import platform import re +import socket import time from copy import deepcopy from datetime import datetime, timezone @@ -130,3 +131,12 @@ def set_default_settings(settings, defaults): def char_decision_as_index(char): return 0 if char == "A" else 1 + + +def check_internet_connection(host="8.8.8.8", port=53, timeout=3): + try: + socket.setdefaulttimeout(timeout) + socket.socket(socket.AF_INET, socket.SOCK_STREAM).connect((host, port)) + return True + except socket.error: + return False From 86c28673d13ccb15682634426ddced1b65f4d895 Mon Sep 17 00:00:00 2001 From: Alessandro Maggio Date: Mon, 8 Feb 2021 20:59:36 +0100 Subject: [PATCH 03/10] WebSocketConnectionClosedException is inside websocket and not inside websocket.exceptions --- TwitchChannelPointsMiner/classes/TwitchWebSocket.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/TwitchChannelPointsMiner/classes/TwitchWebSocket.py b/TwitchChannelPointsMiner/classes/TwitchWebSocket.py index f87aff1..2e85bc5 100644 --- a/TwitchChannelPointsMiner/classes/TwitchWebSocket.py +++ b/TwitchChannelPointsMiner/classes/TwitchWebSocket.py @@ -2,8 +2,7 @@ import json import logging import time -from websocket import WebSocketApp -from websocket.exceptions import WebSocketConnectionClosedException +from websocket import WebSocketApp, WebSocketConnectionClosedException from TwitchChannelPointsMiner.utils import create_nonce From 696a78197ce0792f071ff8b9e9463c47585cff37 Mon Sep 17 00:00:00 2001 From: Alessandro Maggio Date: Tue, 9 Feb 2021 10:44:35 +0100 Subject: [PATCH 04/10] Attempt to handle all connection error/exception. Return {} in case of exception and check correctly with If statement if we are right or not. Don't recconect with lost PING if connection in not available --- .../TwitchChannelPointsMiner.py | 7 +- TwitchChannelPointsMiner/classes/Twitch.py | 105 +++++++++++------- .../classes/WebSocketsPool.py | 7 +- TwitchChannelPointsMiner/utils.py | 2 +- 4 files changed, 74 insertions(+), 47 deletions(-) diff --git a/TwitchChannelPointsMiner/TwitchChannelPointsMiner.py b/TwitchChannelPointsMiner/TwitchChannelPointsMiner.py index 941b637..1f1d10c 100644 --- a/TwitchChannelPointsMiner/TwitchChannelPointsMiner.py +++ b/TwitchChannelPointsMiner/TwitchChannelPointsMiner.py @@ -25,6 +25,7 @@ from TwitchChannelPointsMiner.utils import ( _millify, at_least_one_value_in_settings_is, get_user_agent, + internet_connection_available, set_default_settings, ) @@ -172,6 +173,7 @@ class TwitchChannelPointsMiner: ), ), ) + self.minute_watcher_thread.name = "Minute watcher" self.minute_watcher_thread.start() self.ws_pool = WebSocketsPool( @@ -230,10 +232,11 @@ class TwitchChannelPointsMiner: for index in range(0, len(self.ws_pool.ws)): if ( self.ws_pool.ws[index] is not None - and self.ws_pool.ws[index].elapsed_last_ping() > 10 + and self.ws_pool.ws[index].elapsed_last_ping() > 15 + and internet_connection_available() is True ): logger.info( - f"#{index} - The last PING was sent more than 10 minutes ago. Reconnecting to the WebSocket..." + f"#{index} - The last PING was sent more than 15 minutes ago. Reconnecting to the WebSocket..." ) WebSocketsPool.handle_reconnection(self.ws_pool.ws[index]) diff --git a/TwitchChannelPointsMiner/classes/Twitch.py b/TwitchChannelPointsMiner/classes/Twitch.py index abe3d8f..886f1b4 100644 --- a/TwitchChannelPointsMiner/classes/Twitch.py +++ b/TwitchChannelPointsMiner/classes/Twitch.py @@ -23,7 +23,7 @@ from TwitchChannelPointsMiner.classes.Exceptions import ( from TwitchChannelPointsMiner.classes.Settings import Settings from TwitchChannelPointsMiner.classes.TwitchLogin import TwitchLogin from TwitchChannelPointsMiner.constants import API, CLIENT_ID, GQLOperations -from TwitchChannelPointsMiner.utils import _millify +from TwitchChannelPointsMiner.utils import _millify, internet_connection_available logger = logging.getLogger(__name__) @@ -74,50 +74,63 @@ class Twitch(object): ] def get_spade_url(self, streamer): - headers = {"User-Agent": self.user_agent} - main_page_request = requests.get(streamer.streamer_url, headers=headers) - response = main_page_request.text - settings_url = re.search( - "(https://static.twitchcdn.net/config/settings.*?js)", response - ).group(1) + try: + headers = {"User-Agent": self.user_agent} + main_page_request = requests.get(streamer.streamer_url, headers=headers) + response = main_page_request.text + settings_url = re.search( + "(https://static.twitchcdn.net/config/settings.*?js)", response + ).group(1) - settings_request = requests.get(settings_url, headers=headers) - response = settings_request.text - streamer.stream.spade_url = re.search('"spade_url":"(.*?)"', response).group(1) + settings_request = requests.get(settings_url, headers=headers) + response = settings_request.text + streamer.stream.spade_url = re.search( + '"spade_url":"(.*?)"', response + ).group(1) + except requests.exceptions.RequestException as e: + logger.error(f"Something went wrong during extraction of 'spade_url': {e}") def post_gql_request(self, json_data): - response = requests.post( - GQLOperations.url, - json=json_data, - headers={ - "Authorization": f"OAuth {self.twitch_login.get_auth_token()}", - "Client-Id": CLIENT_ID, - "User-Agent": self.user_agent, - }, - ) - logger.debug( - f"Data: {json_data}, Status code: {response.status_code}, Content: {response.text}" - ) - return response.json() + try: + response = requests.post( + GQLOperations.url, + json=json_data, + headers={ + "Authorization": f"OAuth {self.twitch_login.get_auth_token()}", + "Client-Id": CLIENT_ID, + "User-Agent": self.user_agent, + }, + ) + logger.debug( + f"Data: {json_data}, Status code: {response.status_code}, Content: {response.text}" + ) + return response.json() + except requests.exceptions.RequestException as e: + logger.error( + f"Error with GQLOperations ({json_data['operationName']}): {e}" + ) + return {} def get_broadcast_id(self, streamer): json_data = copy.deepcopy(GQLOperations.WithIsStreamLiveQuery) json_data["variables"] = {"id": streamer.channel_id} response = self.post_gql_request(json_data) - stream = response["data"]["user"]["stream"] - if stream is not None: - return stream["id"] - else: - raise StreamerIsOfflineException + if response != {}: + stream = response["data"]["user"]["stream"] + if stream is not None: + return stream["id"] + else: + raise StreamerIsOfflineException def get_stream_info(self, streamer): json_data = copy.deepcopy(GQLOperations.VideoPlayerStreamInfoOverlayChannel) json_data["variables"] = {"channel": streamer.username} response = self.post_gql_request(json_data) - if response["data"]["user"]["stream"] is None: - raise StreamerIsOfflineException - else: - return response["data"]["user"] + if response != {}: + if response["data"]["user"]["stream"] is None: + raise StreamerIsOfflineException + else: + return response["data"]["user"] def check_streamer_online(self, streamer): if time.time() < streamer.offline_at + 60: @@ -181,7 +194,7 @@ class Twitch(object): def __get_inventory(self): response = self.post_gql_request(GQLOperations.Inventory) - return response["data"]["currentUser"]["inventory"] + return response["data"]["currentUser"]["inventory"] if response != {} else {} # Load the amount of current points for a channel, check if a bonus is available def load_channel_points_context(self, streamer): @@ -189,14 +202,15 @@ class Twitch(object): json_data["variables"] = {"channelLogin": streamer.username} response = self.post_gql_request(json_data) - if response["data"]["community"] is None: - raise StreamerDoesNotExistException - channel = response["data"]["community"]["channel"] - community_points = channel["self"]["communityPoints"] - streamer.channel_points = community_points["balance"] + if response != {}: + if response["data"]["community"] is None: + raise StreamerDoesNotExistException + channel = response["data"]["community"]["channel"] + community_points = channel["self"]["communityPoints"] + streamer.channel_points = community_points["balance"] - if community_points["availableClaim"] is not None: - self.claim_bonus(streamer, community_points["availableClaim"]["id"]) + if community_points["availableClaim"] is not None: + self.claim_bonus(streamer, community_points["availableClaim"]["id"]) def make_predictions(self, event): decision = event.bet.calculate(event.streamer.channel_points) @@ -299,9 +313,18 @@ class Twitch(object): ) if response.status_code == 204: streamers[index].stream.update_minute_watched() - except requests.exceptions.ConnectionError as e: + except requests.exceptions.RequestException as e: logger.error(f"Error while trying to watch a minute: {e}") + # The success rate It's very hight usually. Why we have failed? + # Check internet connection ... + while internet_connection_available() is False: + random_sleep = random.randint(1, 3) + logger.warning( + f"No internet connection available! Retry after {random_sleep}m" + ) + time.sleep(random_sleep * 60) + # Create chunk of sleep of speed-up the break loop after CTRL+C sleep_time = max(next_iteration - time.time(), 0) / chunk_size for i in range(0, chunk_size): diff --git a/TwitchChannelPointsMiner/classes/WebSocketsPool.py b/TwitchChannelPointsMiner/classes/WebSocketsPool.py index 289cef8..5bfa7d4 100644 --- a/TwitchChannelPointsMiner/classes/WebSocketsPool.py +++ b/TwitchChannelPointsMiner/classes/WebSocketsPool.py @@ -14,8 +14,8 @@ from TwitchChannelPointsMiner.classes.TwitchWebSocket import TwitchWebSocket from TwitchChannelPointsMiner.constants import WEBSOCKET from TwitchChannelPointsMiner.utils import ( _millify, - check_internet_connection, get_streamer_index, + internet_connection_available, ) logger = logging.getLogger(__name__) @@ -63,6 +63,7 @@ class WebSocketsPool: self.thread_ws = threading.Thread(target=lambda: self.ws[-1].run_forever()) self.thread_ws.daemon = True + self.thread_ws.name = f"WebSocket #{self.ws[-1].index}" self.thread_ws.start() def end(self): @@ -85,7 +86,7 @@ class WebSocketsPool: if ws.elapsed_last_pong() > 10 and ws.is_reconneting is False: logger.info( - f"#{ws.index} - The last PONG was received more than 10 minutes ago. Reconnect the WebSocket" + f"#{ws.index} - The last PONG was received more than 10 minutes ago" ) ws.is_reconneting = True WebSocketsPool.handle_reconnection(ws) @@ -117,7 +118,7 @@ class WebSocketsPool: ) time.sleep(30) - while check_internet_connection() is False: + while internet_connection_available() is False: random_sleep = random.randint(1, 3) logger.warning( f"#{ws.index} - No internet connection available! Retry after {random_sleep}m" diff --git a/TwitchChannelPointsMiner/utils.py b/TwitchChannelPointsMiner/utils.py index eb97098..da8de53 100644 --- a/TwitchChannelPointsMiner/utils.py +++ b/TwitchChannelPointsMiner/utils.py @@ -133,7 +133,7 @@ def char_decision_as_index(char): return 0 if char == "A" else 1 -def check_internet_connection(host="8.8.8.8", port=53, timeout=3): +def internet_connection_available(host="8.8.8.8", port=53, timeout=3): try: socket.setdefaulttimeout(timeout) socket.socket(socket.AF_INET, socket.SOCK_STREAM).connect((host, port)) From ca3c5037afb5025a8093bd0105897fc524057c35 Mon Sep 17 00:00:00 2001 From: Alessandro Maggio Date: Wed, 10 Feb 2021 15:17:02 +0100 Subject: [PATCH 05/10] Prevent multiple subscription to the same topics from multiple ws. https://github.com/Tkd-Alex/Twitch-Channel-Points-Miner-v2/issues/54\#issuecomment-776696545 --- .../classes/WebSocketsPool.py | 20 +++++++++++++++---- TwitchChannelPointsMiner/utils.py | 8 ++++++++ 2 files changed, 24 insertions(+), 4 deletions(-) diff --git a/TwitchChannelPointsMiner/classes/WebSocketsPool.py b/TwitchChannelPointsMiner/classes/WebSocketsPool.py index 5bfa7d4..3aa606b 100644 --- a/TwitchChannelPointsMiner/classes/WebSocketsPool.py +++ b/TwitchChannelPointsMiner/classes/WebSocketsPool.py @@ -14,6 +14,7 @@ from TwitchChannelPointsMiner.classes.TwitchWebSocket import TwitchWebSocket from TwitchChannelPointsMiner.constants import WEBSOCKET from TwitchChannelPointsMiner.utils import ( _millify, + currently_connected_topics, get_streamer_index, internet_connection_available, ) @@ -40,12 +41,17 @@ class WebSocketsPool: if self.ws == [] or self.ws[-1] is None or len(self.ws[-1].topics) >= 50: self.append_new_websocket() - self.ws[-1].topics.append(topic) - if self.ws[-1].is_opened is False: self.ws[-1].pending_topics.append(topic) else: - self.ws[-1].listen(topic, self.twitch.twitch_login.get_auth_token()) + if topic not in currently_connected_topics(self.ws): + self.ws[-1].listen(topic, self.twitch.twitch_login.get_auth_token()) + else: + logger.warning( + f"#{self.ws[-1].index} - Another WebSocket It's currently connected to: {topic}" + ) + + self.ws[-1].topics.append(topic) def append_new_websocket(self): self.ws.append( @@ -78,7 +84,13 @@ class WebSocketsPool: ws.is_opened = True ws.ping() for topic in ws.pending_topics: - ws.listen(topic, ws.twitch.twitch_login.get_auth_token()) + # I know: ws.parent_pool.ws it's really strange + if topic not in currently_connected_topics(ws.parent_pool.ws): + ws.listen(topic, ws.twitch.twitch_login.get_auth_token()) + else: + logger.warning( + f"#{ws.index} - Another WebSocket It's currently connected to: {topic}" + ) while not ws.is_closed: ws.ping() diff --git a/TwitchChannelPointsMiner/utils.py b/TwitchChannelPointsMiner/utils.py index da8de53..992ea58 100644 --- a/TwitchChannelPointsMiner/utils.py +++ b/TwitchChannelPointsMiner/utils.py @@ -140,3 +140,11 @@ def internet_connection_available(host="8.8.8.8", port=53, timeout=3): return True except socket.error: return False + + +def currently_connected_topics(ws_pool): + merged_topics = [] + for ws in ws_pool: + if ws is not None: + merged_topics += ws.topics + return merged_topics From 137d5d5e3451b0ad274e01c04e6505956bf97dab Mon Sep 17 00:00:00 2001 From: Alessandro Maggio Date: Thu, 11 Feb 2021 10:42:14 +0100 Subject: [PATCH 06/10] I don't know why but I've already done this commit yesterday but It was lost :confused: --- TwitchChannelPointsMiner/classes/WebSocketsPool.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/TwitchChannelPointsMiner/classes/WebSocketsPool.py b/TwitchChannelPointsMiner/classes/WebSocketsPool.py index 87e5aa9..884ffc3 100644 --- a/TwitchChannelPointsMiner/classes/WebSocketsPool.py +++ b/TwitchChannelPointsMiner/classes/WebSocketsPool.py @@ -50,8 +50,7 @@ class WebSocketsPool: logger.warning( f"#{self.ws[-1].index} - Another WebSocket It's currently connected to: {topic}" ) - - self.ws[-1].topics.append(topic) + self.ws[-1].topics.append(topic) def append_new_websocket(self): self.ws.append( @@ -91,6 +90,7 @@ class WebSocketsPool: logger.warning( f"#{ws.index} - Another WebSocket It's currently connected to: {topic}" ) + ws.topics.append(topic) while not ws.is_closed: ws.ping() From 7c2dd5395172fb9131e71e577f686bde0623dc6f Mon Sep 17 00:00:00 2001 From: Alessandro Maggio Date: Thu, 11 Feb 2021 16:58:48 +0100 Subject: [PATCH 07/10] Appen new ws also when the pending_topics are gte 50 --- TwitchChannelPointsMiner/classes/WebSocketsPool.py | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/TwitchChannelPointsMiner/classes/WebSocketsPool.py b/TwitchChannelPointsMiner/classes/WebSocketsPool.py index 884ffc3..d4a92c6 100644 --- a/TwitchChannelPointsMiner/classes/WebSocketsPool.py +++ b/TwitchChannelPointsMiner/classes/WebSocketsPool.py @@ -38,7 +38,12 @@ class WebSocketsPool: def submit(self, topic): # Check if we need to create a new WebSocket instance - if self.ws == [] or self.ws[-1] is None or len(self.ws[-1].topics) >= 50: + if ( + self.ws == [] + or self.ws[-1] is None + or len(self.ws[-1].topics) >= 50 + or len(self.ws[-1].pending_topics) >= 50 + ): self.append_new_websocket() if self.ws[-1].is_opened is False: From 175d6abe126dfba8b1942e4787f449c21d724f0e Mon Sep 17 00:00:00 2001 From: Alessandro Maggio Date: Fri, 12 Feb 2021 00:15:47 +0100 Subject: [PATCH 08/10] Instead of appending every time a new ws at the end of array replace at same index. Seems to be a good solution. --- .../TwitchChannelPointsMiner.py | 6 +- TwitchChannelPointsMiner/classes/Twitch.py | 72 +++++++------- .../classes/WebSocketsPool.py | 96 +++++++++---------- TwitchChannelPointsMiner/utils.py | 8 -- 4 files changed, 85 insertions(+), 97 deletions(-) diff --git a/TwitchChannelPointsMiner/TwitchChannelPointsMiner.py b/TwitchChannelPointsMiner/TwitchChannelPointsMiner.py index d1e8710..e6c8560 100644 --- a/TwitchChannelPointsMiner/TwitchChannelPointsMiner.py +++ b/TwitchChannelPointsMiner/TwitchChannelPointsMiner.py @@ -236,12 +236,12 @@ class TwitchChannelPointsMiner: # Check if is not None because maybe we have already created a new connection on array+1 and now index is None for index in range(0, len(self.ws_pool.ws)): if ( - self.ws_pool.ws[index] is not None - and self.ws_pool.ws[index].elapsed_last_ping() > 15 + self.ws_pool.ws[index].is_reconneting is False + and self.ws_pool.ws[index].elapsed_last_ping() > 10 and internet_connection_available() is True ): logger.info( - f"#{index} - The last PING was sent more than 15 minutes ago. Reconnecting to the WebSocket..." + f"#{index} - The last PING was sent more than 10 minutes ago. Reconnecting to the WebSocket..." ) WebSocketsPool.handle_reconnection(self.ws_pool.ws[index]) diff --git a/TwitchChannelPointsMiner/classes/Twitch.py b/TwitchChannelPointsMiner/classes/Twitch.py index 2034c59..347ea82 100644 --- a/TwitchChannelPointsMiner/classes/Twitch.py +++ b/TwitchChannelPointsMiner/classes/Twitch.py @@ -48,30 +48,31 @@ class Twitch(object): def update_stream(self, streamer): if streamer.stream.update_required() is True: stream_info = self.get_stream_info(streamer) - streamer.stream.update( - broadcast_id=stream_info["stream"]["id"], - title=stream_info["broadcastSettings"]["title"], - game=stream_info["broadcastSettings"]["game"], - tags=stream_info["stream"]["tags"], - viewers_count=stream_info["stream"]["viewersCount"], - ) + if stream_info is not None: + streamer.stream.update( + broadcast_id=stream_info["stream"]["id"], + title=stream_info["broadcastSettings"]["title"], + game=stream_info["broadcastSettings"]["game"], + tags=stream_info["stream"]["tags"], + viewers_count=stream_info["stream"]["viewersCount"], + ) - event_properties = { - "channel_id": streamer.channel_id, - "broadcast_id": streamer.stream.broadcast_id, - "player": "site", - "user_id": self.twitch_login.get_user_id(), - } + event_properties = { + "channel_id": streamer.channel_id, + "broadcast_id": streamer.stream.broadcast_id, + "player": "site", + "user_id": self.twitch_login.get_user_id(), + } - if ( - streamer.stream.game_name() is not None - and streamer.settings.claim_drops is True - ): - event_properties["game"] = streamer.stream.game_name() + if ( + streamer.stream.game_name() is not None + and streamer.settings.claim_drops is True + ): + event_properties["game"] = streamer.stream.game_name() - streamer.stream.payload = [ - {"event": "minute-watched", "properties": event_properties} - ] + streamer.stream.payload = [ + {"event": "minute-watched", "properties": event_properties} + ] def get_spade_url(self, streamer): try: @@ -231,21 +232,22 @@ class Twitch(object): extra={"emoji": ":pushpin:"}, ) else: - logger.info( - f"Place {_millify(decision['amount'])} channel points on: {event.bet.get_outcome(selector_index)}", - extra={"emoji": ":four_leaf_clover:"}, - ) + if decision["amount"] > 0: + logger.info( + f"Place {_millify(decision['amount'])} channel points on: {event.bet.get_outcome(selector_index)}", + extra={"emoji": ":four_leaf_clover:"}, + ) - json_data = copy.deepcopy(GQLOperations.MakePrediction) - json_data["variables"] = { - "input": { - "eventID": event.event_id, - "outcomeID": decision["id"], - "points": decision["amount"], - "transactionID": token_hex(16), + json_data = copy.deepcopy(GQLOperations.MakePrediction) + json_data["variables"] = { + "input": { + "eventID": event.event_id, + "outcomeID": decision["id"], + "points": decision["amount"], + "transactionID": token_hex(16), + } } - } - return self.post_gql_request(json_data) + return self.post_gql_request(json_data) else: logger.info( f"Oh no! The event is not active anymore! Current status: {event.status}", @@ -330,7 +332,7 @@ class Twitch(object): if response.status_code == 204: streamers[index].stream.update_minute_watched() except requests.exceptions.RequestException as e: - logger.error(f"Error while trying to watch a minute: {e}") + logger.error(f"Error while trying to send minute watched: {e}") # The success rate It's very hight usually. Why we have failed? # Check internet connection ... diff --git a/TwitchChannelPointsMiner/classes/WebSocketsPool.py b/TwitchChannelPointsMiner/classes/WebSocketsPool.py index d4a92c6..7d1193d 100644 --- a/TwitchChannelPointsMiner/classes/WebSocketsPool.py +++ b/TwitchChannelPointsMiner/classes/WebSocketsPool.py @@ -14,7 +14,6 @@ from TwitchChannelPointsMiner.classes.TwitchWebSocket import TwitchWebSocket from TwitchChannelPointsMiner.constants import WEBSOCKET from TwitchChannelPointsMiner.utils import ( _millify, - currently_connected_topics, get_streamer_index, internet_connection_available, ) @@ -38,49 +37,44 @@ class WebSocketsPool: def submit(self, topic): # Check if we need to create a new WebSocket instance - if ( - self.ws == [] - or self.ws[-1] is None - or len(self.ws[-1].topics) >= 50 - or len(self.ws[-1].pending_topics) >= 50 - ): - self.append_new_websocket() + if self.ws == [] or len(self.ws[-1].topics) >= 50: + self.ws.append(self.__new(len(self.ws))) + self.__start(-1) - if self.ws[-1].is_opened is False: - self.ws[-1].pending_topics.append(topic) + self.__submit(-1, topic) + + def __submit(self, index, topic): + # Topic in topics should never happen. Anyway prevent any types of duplicates + if topic not in self.ws[index].topics: + self.ws[index].topics.append(topic) + + if self.ws[index].is_opened is False: + self.ws[index].pending_topics.append(topic) else: - if topic not in currently_connected_topics(self.ws): - self.ws[-1].listen(topic, self.twitch.twitch_login.get_auth_token()) - else: - logger.warning( - f"#{self.ws[-1].index} - Another WebSocket It's currently connected to: {topic}" - ) - self.ws[-1].topics.append(topic) + self.ws[index].listen(topic, self.twitch.twitch_login.get_auth_token()) - def append_new_websocket(self): - self.ws.append( - TwitchWebSocket( - index=len(self.ws), - parent_pool=self, - url=WEBSOCKET, - on_message=WebSocketsPool.on_message, - on_open=WebSocketsPool.on_open, - on_error=WebSocketsPool.on_error, - on_close=WebSocketsPool.on_close - # on_close=WebSocketsPool.handle_reconnection, # Do nothing. - ) + def __new(self, index): + return TwitchWebSocket( + index=index, + parent_pool=self, + url=WEBSOCKET, + on_message=WebSocketsPool.on_message, + on_open=WebSocketsPool.on_open, + on_error=WebSocketsPool.on_error, + on_close=WebSocketsPool.on_close + # on_close=WebSocketsPool.handle_reconnection, # Do nothing. ) - self.thread_ws = threading.Thread(target=lambda: self.ws[-1].run_forever()) - self.thread_ws.daemon = True - self.thread_ws.name = f"WebSocket #{self.ws[-1].index}" - self.thread_ws.start() + def __start(self, index): + thread_ws = threading.Thread(target=lambda: self.ws[index].run_forever()) + thread_ws.daemon = True + thread_ws.name = f"WebSocket #{self.ws[index].index}" + thread_ws.start() def end(self): for index in range(0, len(self.ws)): - if self.ws[index] is not None: - self.ws[index].forced_close = True - self.ws[index].close() + self.ws[index].forced_close = True + self.ws[index].close() @staticmethod def on_open(ws): @@ -88,22 +82,15 @@ class WebSocketsPool: ws.is_opened = True ws.ping() for topic in ws.pending_topics: - # I know: ws.parent_pool.ws it's really strange - if topic not in currently_connected_topics(ws.parent_pool.ws): - ws.listen(topic, ws.twitch.twitch_login.get_auth_token()) - else: - logger.warning( - f"#{ws.index} - Another WebSocket It's currently connected to: {topic}" - ) - ws.topics.append(topic) + ws.listen(topic, ws.twitch.twitch_login.get_auth_token()) - while not ws.is_closed: + while ws.is_closed is False: ws.ping() time.sleep(random.uniform(25, 30)) - if ws.elapsed_last_pong() > 10 and ws.is_reconneting is False: + if ws.elapsed_last_pong() > 5 and ws.is_reconneting is False: logger.info( - f"#{ws.index} - The last PONG was received more than 10 minutes ago" + f"#{ws.index} - The last PONG was received more than 5 minutes ago" ) ws.is_reconneting = True WebSocketsPool.handle_reconnection(ws) @@ -114,6 +101,8 @@ class WebSocketsPool: @staticmethod def on_error(ws, error): + # Connection lost | [WinError 10054] An existing connection was forcibly closed by the remote host + # Connection already closed | Connection is already closed (raise WebSocketConnectionClosedException) logger.error(f"#{ws.index} - WebSocket error: {error}") @staticmethod @@ -125,13 +114,12 @@ class WebSocketsPool: @staticmethod def handle_reconnection(ws): # Close the current WebSocket. - # anyway, we replace the ws with None ws.is_closed = True ws.keep_running = False # Reconnect only if ws.forced_close is False (replace the keep_running) if ws.forced_close is False: logger.info( - f"#{ws.index} - Reconnecting to Twitch PubSub server in 30 seconds" + f"#{ws.index} - Reconnecting to Twitch PubSub server in ~60 seconds" ) time.sleep(30) @@ -142,10 +130,16 @@ class WebSocketsPool: ) time.sleep(random_sleep * 60) + # Why not create a new ws on the same array index? Let's try. self = ws.parent_pool - self.ws[ws.index] = None + self.ws[ws.index] = self.__new(ws.index) # Create a new connection. + # self.ws[ws.index].topics = ws.topics + + self.__start(ws.index) # Start a new thread. + time.sleep(30) + for topic in ws.topics: - self.submit(topic) + self.__submit(ws.index, topic) @staticmethod def on_message(ws, message): diff --git a/TwitchChannelPointsMiner/utils.py b/TwitchChannelPointsMiner/utils.py index 992ea58..da8de53 100644 --- a/TwitchChannelPointsMiner/utils.py +++ b/TwitchChannelPointsMiner/utils.py @@ -140,11 +140,3 @@ def internet_connection_available(host="8.8.8.8", port=53, timeout=3): return True except socket.error: return False - - -def currently_connected_topics(ws_pool): - merged_topics = [] - for ws in ws_pool: - if ws is not None: - merged_topics += ws.topics - return merged_topics From 9a36dbd862b455fcecb49b4399410359f6401923 Mon Sep 17 00:00:00 2001 From: Alessandro Maggio Date: Fri, 12 Feb 2021 22:12:56 +0100 Subject: [PATCH 09/10] Use reconnect flag for manage ws reconnection and main threads --- .../classes/WebSocketsPool.py | 25 ++++++++++++------- 1 file changed, 16 insertions(+), 9 deletions(-) diff --git a/TwitchChannelPointsMiner/classes/WebSocketsPool.py b/TwitchChannelPointsMiner/classes/WebSocketsPool.py index 7d1193d..f1f6a8f 100644 --- a/TwitchChannelPointsMiner/classes/WebSocketsPool.py +++ b/TwitchChannelPointsMiner/classes/WebSocketsPool.py @@ -81,19 +81,22 @@ class WebSocketsPool: def run(): ws.is_opened = True ws.ping() + for topic in ws.pending_topics: ws.listen(topic, ws.twitch.twitch_login.get_auth_token()) while ws.is_closed is False: - ws.ping() - time.sleep(random.uniform(25, 30)) + # Else: the ws is currently in reconnecting phase, you can't do ping or other operation. + # Probably this ws will be closed very soon with ws.is_closed = True + if ws.is_reconneting is False: + ws.ping() # We need ping for keep the connection alive + time.sleep(random.uniform(25, 30)) - if ws.elapsed_last_pong() > 5 and ws.is_reconneting is False: - logger.info( - f"#{ws.index} - The last PONG was received more than 5 minutes ago" - ) - ws.is_reconneting = True - WebSocketsPool.handle_reconnection(ws) + if ws.elapsed_last_pong() > 5: + logger.info( + f"#{ws.index} - The last PONG was received more than 5 minutes ago" + ) + WebSocketsPool.handle_reconnection(ws) thread_ws = threading.Thread(target=run) thread_ws.daemon = True @@ -117,6 +120,11 @@ class WebSocketsPool: ws.is_closed = True ws.keep_running = False # Reconnect only if ws.forced_close is False (replace the keep_running) + + # Set the current socket as reconnecting status + # So the exsternal ping check will be locked + ws.is_reconneting = True + if ws.forced_close is False: logger.info( f"#{ws.index} - Reconnecting to Twitch PubSub server in ~60 seconds" @@ -365,7 +373,6 @@ class WebSocketsPool: elif response["type"] == "RECONNECT": logger.info(f"#{ws.index} - Reconnection required") - ws.is_reconneting = True WebSocketsPool.handle_reconnection(ws) elif response["type"] == "PONG": From a8baad782c6b66a0eb45e8a6a6111a61eb7002e4 Mon Sep 17 00:00:00 2001 From: Alessandro Maggio Date: Sat, 13 Feb 2021 11:04:57 +0100 Subject: [PATCH 10/10] use to_login instead of to_name, this will fix: https://github.com/Tkd-Alex/Twitch-Channel-Points-Miner-v2/pull/68\#issuecomment-778518970 --- TwitchChannelPointsMiner/classes/Twitch.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/TwitchChannelPointsMiner/classes/Twitch.py b/TwitchChannelPointsMiner/classes/Twitch.py index 13cbad4..9723d33 100644 --- a/TwitchChannelPointsMiner/classes/Twitch.py +++ b/TwitchChannelPointsMiner/classes/Twitch.py @@ -374,7 +374,7 @@ class Twitch(object): json_response = self.__do_helix_request(query) pagination = json_response["pagination"] - followers += [fw["to_name"].lower() for fw in json_response["data"]] + followers += [fw["to_login"].lower() for fw in json_response["data"]] time.sleep(random.uniform(0.3, 0.7)) if pagination == {}: