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

Fix for reconnection bug. Fixed typo in TwitchWebSocket.is_reconnecting. pre-commit changes.

Browse Source
This commit is contained in:
mpforce1
2023-07-15 08:48:33 +01:00
committed by Matthew Padbury
parent 586422a640
commit 29d5d8a360
4 changed files with 61 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
.pre-commit-config.yaml
View File

@@ -6,7 +6,7 @@ repos:
- id: end-of-file-fixer - id: end-of-file-fixer
- id: check-added-large-files - id: check-added-large-files
- repo: https://github.com/pycqa/isort - repo: https://github.com/pycqa/isort
rev: 5.10.1 rev: 5.12.0
hooks: hooks:
- id: isort - id: isort
files: ^TwitchChannelPointsMiner/ files: ^TwitchChannelPointsMiner/
@@ -16,7 +16,7 @@ repos:
hooks: hooks:
- id: black - id: black
files: ^TwitchChannelPointsMiner/ files: ^TwitchChannelPointsMiner/
- repo: https://gitlab.com/pycqa/flake8 - repo: https://github.com/pycqa/flake8
rev: 3.9.2 rev: 3.9.2
hooks: hooks:
- id: flake8 - id: flake8

38
TwitchChannelPointsMiner/TwitchChannelPointsMiner.py
View File

@@ -88,8 +88,7 @@ class TwitchChannelPointsMiner:
): ):
# Fixes TypeError: 'NoneType' object is not subscriptable # Fixes TypeError: 'NoneType' object is not subscriptable
if not username or username == "your-twitch-username": if not username or username == "your-twitch-username":
logger.error( logger.error("Please edit your runner file (usually run.py) and try again.")
"Please edit your runner file (usually run.py) and try again.")
logger.error("No username, exiting...") logger.error("No username, exiting...")
sys.exit(0) sys.exit(0)
@@ -122,7 +121,8 @@ class TwitchChannelPointsMiner:
if enable_analytics is True: if enable_analytics is True:
Settings.analytics_path = os.path.join( Settings.analytics_path = os.path.join(
Path().absolute(), "analytics", username) Path().absolute(), "analytics", username
)
Path(Settings.analytics_path).mkdir(parents=True, exist_ok=True) Path(Settings.analytics_path).mkdir(parents=True, exist_ok=True)
self.username = username self.username = username
@@ -161,17 +161,16 @@ class TwitchChannelPointsMiner:
current_version, github_version = check_versions() current_version, github_version = check_versions()
logger.info( logger.info(
f"Twitch Channel Points Miner v2-{current_version} (fork by rdavydov)") f"Twitch Channel Points Miner v2-{current_version} (fork by rdavydov)"
logger.info( )
"https://github.com/rdavydov/Twitch-Channel-Points-Miner-v2") logger.info("https://github.com/rdavydov/Twitch-Channel-Points-Miner-v2")
if github_version == "0.0.0": if github_version == "0.0.0":
logger.error( logger.error(
"Unable to detect if you have the latest version of this script" "Unable to detect if you have the latest version of this script"
) )
elif current_version != github_version: elif current_version != github_version:
logger.info( logger.info(f"You are running the version {current_version} of this script")
f"You are running the version {current_version} of this script")
logger.info(f"The latest version on GitHub is: {github_version}") logger.info(f"The latest version on GitHub is: {github_version}")
for sign in [signal.SIGINT, signal.SIGSEGV, signal.SIGTERM]: for sign in [signal.SIGINT, signal.SIGSEGV, signal.SIGTERM]:
@@ -195,8 +194,7 @@ class TwitchChannelPointsMiner:
http_server.name = "Analytics Thread" http_server.name = "Analytics Thread"
http_server.start() http_server.start()
else: else:
logger.error( logger.error("Can't start analytics(), please set enable_analytics=True")
"Can't start analytics(), please set enable_analytics=True")
def mine( def mine(
self, self,
@@ -242,8 +240,7 @@ class TwitchChannelPointsMiner:
streamers_dict[username] = streamer streamers_dict[username] = streamer
if followers is True: if followers is True:
followers_array = self.twitch.get_followers( followers_array = self.twitch.get_followers(order=followers_order)
order=followers_order)
logger.info( logger.info(
f"Load {len(followers_array)} followers from your profile!", f"Load {len(followers_array)} followers from your profile!",
extra={"emoji": ":clipboard:"}, extra={"emoji": ":clipboard:"},
@@ -266,8 +263,7 @@ class TwitchChannelPointsMiner:
if isinstance(streamers_dict[username], Streamer) is True if isinstance(streamers_dict[username], Streamer) is True
else Streamer(username) else Streamer(username)
) )
streamer.channel_id = self.twitch.get_channel_id( streamer.channel_id = self.twitch.get_channel_id(username)
username)
streamer.settings = set_default_settings( streamer.settings = set_default_settings(
streamer.settings, Settings.streamer_settings streamer.settings, Settings.streamer_settings
) )
@@ -309,8 +305,7 @@ class TwitchChannelPointsMiner:
# If we have at least one streamer with settings = claim_drops True # If we have at least one streamer with settings = claim_drops True
# Spawn a thread for sync inventory and dashboard # Spawn a thread for sync inventory and dashboard
if ( if (
at_least_one_value_in_settings_is( at_least_one_value_in_settings_is(self.streamers, "claim_drops", True)
self.streamers, "claim_drops", True)
is True is True
): ):
self.sync_campaigns_thread = threading.Thread( self.sync_campaigns_thread = threading.Thread(
@@ -369,14 +364,12 @@ class TwitchChannelPointsMiner:
if streamer.settings.make_predictions is True: if streamer.settings.make_predictions is True:
self.ws_pool.submit( self.ws_pool.submit(
PubsubTopic("predictions-channel-v1", PubsubTopic("predictions-channel-v1", streamer=streamer)
streamer=streamer)
) )
if streamer.settings.claim_moments is True: if streamer.settings.claim_moments is True:
self.ws_pool.submit( self.ws_pool.submit(
PubsubTopic("community-moments-channel-v1", PubsubTopic("community-moments-channel-v1", streamer=streamer)
streamer=streamer)
) )
refresh_context = time.time() refresh_context = time.time()
@@ -386,15 +379,14 @@ class TwitchChannelPointsMiner:
# Check if is not None because maybe we have already created a new connection on array+1 and now index is None # 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)): for index in range(0, len(self.ws_pool.ws)):
if ( if (
self.ws_pool.ws[index].is_reconneting is False self.ws_pool.ws[index].is_reconnecting is False
and self.ws_pool.ws[index].elapsed_last_ping() > 10 and self.ws_pool.ws[index].elapsed_last_ping() > 10
and internet_connection_available() is True and internet_connection_available() is True
): ):
logger.info( 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 10 minutes ago. Reconnecting to the WebSocket..."
) )
WebSocketsPool.handle_reconnection( WebSocketsPool.handle_reconnection(self.ws_pool.ws[index])
self.ws_pool.ws[index])
if ((time.time() - refresh_context) // 60) >= 30: if ((time.time() - refresh_context) // 60) >= 30:
refresh_context = time.time() refresh_context = time.time()

2
TwitchChannelPointsMiner/classes/TwitchWebSocket.py
View File

@@ -18,7 +18,7 @@ class TwitchWebSocket(WebSocketApp):
self.is_closed = False self.is_closed = False
self.is_opened = False self.is_opened = False
self.is_reconneting = False self.is_reconnecting = False
self.forced_close = False self.forced_close = False
# Custom attribute # Custom attribute

83
TwitchChannelPointsMiner/classes/WebSocketsPool.py
View File

@@ -52,8 +52,7 @@ class WebSocketsPool:
if self.ws[index].is_opened is False: if self.ws[index].is_opened is False:
self.ws[index].pending_topics.append(topic) self.ws[index].pending_topics.append(topic)
else: else:
self.ws[index].listen( self.ws[index].listen(topic, self.twitch.twitch_login.get_auth_token())
topic, self.twitch.twitch_login.get_auth_token())
def __new(self, index): def __new(self, index):
return TwitchWebSocket( return TwitchWebSocket(
@@ -70,8 +69,12 @@ class WebSocketsPool:
def __start(self, index): def __start(self, index):
if Settings.disable_ssl_cert_verification is True: if Settings.disable_ssl_cert_verification is True:
import ssl import ssl
thread_ws = Thread(target=lambda: self.ws[index].run_forever(
sslopt={"cert_reqs": ssl.CERT_NONE})) thread_ws = Thread(
target=lambda: self.ws[index].run_forever(
sslopt={"cert_reqs": ssl.CERT_NONE}
)
)
logger.warn("SSL certificate verification is disabled! Be aware!") logger.warn("SSL certificate verification is disabled! Be aware!")
else: else:
thread_ws = Thread(target=lambda: self.ws[index].run_forever()) thread_ws = Thread(target=lambda: self.ws[index].run_forever())
@@ -96,7 +99,7 @@ class WebSocketsPool:
while ws.is_closed is False: while ws.is_closed is False:
# Else: the ws is currently in reconnecting phase, you can't do ping or other operation. # 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 # Probably this ws will be closed very soon with ws.is_closed = True
if ws.is_reconneting is False: if ws.is_reconnecting is False:
ws.ping() # We need ping for keep the connection alive ws.ping() # We need ping for keep the connection alive
time.sleep(random.uniform(25, 30)) time.sleep(random.uniform(25, 30))
@@ -124,38 +127,40 @@ class WebSocketsPool:
@staticmethod @staticmethod
def handle_reconnection(ws): def handle_reconnection(ws):
# Close the current WebSocket. # Reconnect only if ws.is_reconnecting is False to prevent more than 1 ws from being created
ws.is_closed = True if ws.is_reconnecting is False:
ws.keep_running = False # Close the current WebSocket.
# Reconnect only if ws.forced_close is False (replace the keep_running) 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 # Set the current socket as reconnecting status
# So the external ping check will be locked # So the external ping check will be locked
ws.is_reconneting = True ws.is_reconnecting = True
if ws.forced_close is False: if ws.forced_close is False:
logger.info( logger.info(
f"#{ws.index} - Reconnecting to Twitch PubSub server in ~60 seconds" f"#{ws.index} - Reconnecting to Twitch PubSub server in ~60 seconds"
)
time.sleep(30)
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"
) )
time.sleep(random_sleep * 60) time.sleep(30)
# Why not create a new ws on the same array index? Let's try. while internet_connection_available() is False:
self = ws.parent_pool random_sleep = random.randint(1, 3)
# Create a new connection. logger.warning(
self.ws[ws.index] = self.__new(ws.index) f"#{ws.index} - No internet connection available! Retry after {random_sleep}m"
)
time.sleep(random_sleep * 60)
self.__start(ws.index) # Start a new thread. # Why not create a new ws on the same array index? Let's try.
time.sleep(30) self = ws.parent_pool
# Create a new connection.
self.ws[ws.index] = self.__new(ws.index)
for topic in ws.topics: self.__start(ws.index) # Start a new thread.
self.__submit(ws.index, topic) time.sleep(30)
for topic in ws.topics:
self.__submit(ws.index, topic)
@staticmethod @staticmethod
def on_message(ws, message): def on_message(ws, message):
@@ -179,8 +184,7 @@ class WebSocketsPool:
ws.last_message_timestamp = message.timestamp ws.last_message_timestamp = message.timestamp
ws.last_message_type_channel = message.identifier ws.last_message_type_channel = message.identifier
streamer_index = get_streamer_index( streamer_index = get_streamer_index(ws.streamers, message.channel_id)
ws.streamers, message.channel_id)
if streamer_index != -1: if streamer_index != -1:
try: try:
if message.topic == "community-points-user-v1": if message.topic == "community-points-user-v1":
@@ -239,14 +243,12 @@ class WebSocketsPool:
message.message["raid"]["id"], message.message["raid"]["id"],
message.message["raid"]["target_login"], message.message["raid"]["target_login"],
) )
ws.twitch.update_raid( ws.twitch.update_raid(ws.streamers[streamer_index], raid)
ws.streamers[streamer_index], raid)
elif message.topic == "community-moments-channel-v1": elif message.topic == "community-moments-channel-v1":
if message.type == "active": if message.type == "active":
ws.twitch.claim_moment( ws.twitch.claim_moment(
ws.streamers[streamer_index], ws.streamers[streamer_index], message.data["moment_id"]
message.data["moment_id"]
) )
elif message.topic == "predictions-channel-v1": elif message.topic == "predictions-channel-v1":
@@ -381,7 +383,9 @@ class WebSocketsPool:
if event_prediction.result["type"] != "LOSE": if event_prediction.result["type"] != "LOSE":
# Analytics switch # Analytics switch
if Settings.enable_analytics is True: if Settings.enable_analytics is True:
ws.streamers[streamer_index].persistent_annotations( ws.streamers[
streamer_index
].persistent_annotations(
event_prediction.result["type"], event_prediction.result["type"],
f"{ws.events_predictions[event_id].title}", f"{ws.events_predictions[event_id].title}",
) )
@@ -400,8 +404,7 @@ class WebSocketsPool:
) )
elif response["type"] == "RESPONSE" and len(response.get("error", "")) > 0: elif response["type"] == "RESPONSE" and len(response.get("error", "")) > 0:
raise RuntimeError( raise RuntimeError(f"Error while trying to listen for a topic: {response}")
f"Error while trying to listen for a topic: {response}")
elif response["type"] == "RECONNECT": elif response["type"] == "RECONNECT":
logger.info(f"#{ws.index} - Reconnection required") logger.info(f"#{ws.index} - Reconnection required")