majjo/Twitch-Drops-Miner
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Instead of appending every time a new ws at the end of array replace at same index. Seems to be a good solution.

Browse Source
This commit is contained in:
Alessandro Maggio
2021-02-12 00:15:47 +01:00
parent 7c2dd53951
commit 175d6abe12
4 changed files with 85 additions and 97 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
TwitchChannelPointsMiner/TwitchChannelPointsMiner.py
View File

@@ -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])

72
TwitchChannelPointsMiner/classes/Twitch.py
View File

@@ -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 ...

96
TwitchChannelPointsMiner/classes/WebSocketsPool.py
View File

@@ -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):

8
TwitchChannelPointsMiner/utils.py
View File

@@ -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