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