import json import logging import random import threading import time from dateutil import parser from millify import millify from TwitchChannelPointsMiner.classes.EventPrediction import EventPrediction from TwitchChannelPointsMiner.classes.Exceptions import TimeBasedDropNotFound from TwitchChannelPointsMiner.classes.Message import Message from TwitchChannelPointsMiner.classes.Raid import Raid from TwitchChannelPointsMiner.classes.TwitchWebSocket import TwitchWebSocket from TwitchChannelPointsMiner.constants.twitch import WEBSOCKET from TwitchChannelPointsMiner.utils import ( bet_condition, calculate_start_after, get_streamer_index, ) logger = logging.getLogger(__name__) class WebSocketsPool: def __init__( self, twitch, twitch_browser, streamers, bet_settings, events_predictions, less_printing: bool = False, ): self.ws = None self.twitch = twitch self.twitch_browser = twitch_browser self.streamers = streamers self.events_predictions = events_predictions self.bet_settings = bet_settings self.less_printing = less_printing """ API Limits - Clients can listen on up to 50 topics per connection. Trying to listen on more topics will result in an error message. - We recommend that a single client IP address establishes no more than 10 simultaneous connections. The two limits above are likely to be relaxed for approved third-party applications, as we start to better understand third-party requirements. """ def submit(self, topic): if self.ws is None or len(self.ws.topics) >= 50: self.create_new_websocket() self.ws.topics.append(topic) if not self.ws.is_opened: self.ws.pending_topics.append(topic) else: self.ws.listen(topic, self.twitch.twitch_login.get_auth_token()) def create_new_websocket(self): self.ws = TwitchWebSocket( WEBSOCKET, on_message=WebSocketsPool.on_message, on_open=WebSocketsPool.on_open, on_close=WebSocketsPool.handle_websocket_reconnection, ) self.ws.reset(self) self.thread_ws = threading.Thread(target=lambda: self.ws.run_forever()) self.thread_ws.daemon = True self.thread_ws.start() def end(self): self.ws.keep_running = False self.ws.close() @staticmethod def on_open(ws): 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 not ws.is_closed: ws.ping() time.sleep(random.uniform(25, 30)) if ws.elapsed_last_pong() > 15 and ws.is_reconneting is False: logger.info( "The last pong was received more than 15 minutes ago. Reconnect the WebSocket" ) ws.keep_running = True ws.is_reconneting = True WebSocketsPool.handle_websocket_reconnection(ws) thread_ws = threading.Thread(target=run) thread_ws.daemon = True thread_ws.start() @staticmethod def handle_websocket_reconnection(ws): ws.is_closed = True if ws.keep_running is True: logger.info("Reconnecting to Twitch PubSub server in 60 seconds") time.sleep(60) self = ws.parent_pool if self.ws == ws: self.ws = None for topic in ws.topics: self.submit(topic) @staticmethod def on_message(ws, message): logger.debug(f"Received: {message.strip()}") response = json.loads(message) if response["type"] == "MESSAGE": # We should create a Message class ... message = Message(response["data"]) # If we have more than one PubSub connection, messages may be duplicated # Check the concatenation between message_type.top.channel_id if ( ws.last_message_type_channel is not None and ws.last_message_timestamp is not None and ws.last_message_timestamp == message.timestamp and ws.last_message_type_channel == message.identifier ): return ws.last_message_timestamp = message.timestamp ws.last_message_type_channel = message.identifier streamer_index = get_streamer_index(ws.streamers, message.channel_id) if streamer_index != -1: try: if message.topic == "community-points-user-v1": if message.type == "points-earned": earned = message.data["point_gain"]["total_points"] reason_code = message.data["point_gain"]["reason_code"] balance = message.data["balance"]["balance"] ws.streamers[streamer_index].channel_points = balance logger.info( f"+{earned} → {ws.streamers[streamer_index]} - Reason: {reason_code}.", extra={"emoji": ":rocket:"}, ) ws.streamers[streamer_index].update_history( reason_code, earned ) elif message.type == "claim-available": ws.twitch.claim_bonus( ws.streamers[streamer_index], message.data["claim"]["id"], less_printing=ws.less_printing, ) elif message.topic == "video-playback-by-id": # There is stream-up message type, but it's sent earlier than the API updates if message.type == "stream-up": ws.streamers[streamer_index].stream_up = time.time() elif message.type == "stream-down": if ws.streamers[streamer_index].is_online is True: ws.streamers[streamer_index].set_offline() elif message.type == "viewcount": if ws.streamers[streamer_index].stream_up_elapsed(): ws.twitch.check_streamer_online( ws.streamers[streamer_index] ) elif message.topic == "raid": if message.type == "raid_update_v2": raid = Raid( message.message["raid"]["id"], message.message["raid"]["target_login"], ) ws.twitch.update_raid(ws.streamers[streamer_index], raid) elif message.topic == "predictions-channel-v1": event_dict = message.data["event"] event_id = event_dict["id"] event_status = event_dict["status"] current_tmsp = parser.parse(message.timestamp) if ( message.type == "event-created" and event_id not in ws.events_predictions ): if event_status == "ACTIVE": prediction_window_seconds = float( event_dict["prediction_window_seconds"] ) prediction_window_seconds -= ( 25 if prediction_window_seconds <= 180 else 60 ) event = EventPrediction( ws.streamers[streamer_index], event_id, event_dict["title"], parser.parse(event_dict["created_at"]), prediction_window_seconds, event_status, event_dict["outcomes"], bet_settings=ws.bet_settings, less_printing=ws.less_printing, ) if ( ws.streamers[streamer_index].is_online and event.closing_bet_after(current_tmsp) > 0 and bet_condition( ws.twitch_browser, event, logger, ) is True ): ws.events_predictions[event_id] = event ( start_bet_status, execution_time, ) = ws.twitch_browser.start_bet( ws.events_predictions[event_id] ) if start_bet_status is True: # place_bet_thread = threading.Timer(event.closing_bet_after(current_tmsp), ws.twitch.make_predictions, (ws.events_predictions[event_id],)) start_after = calculate_start_after( event.closing_bet_after(current_tmsp), execution_time, ) place_bet_thread = threading.Timer( start_after, ws.twitch_browser.place_bet, (ws.events_predictions[event_id],), ) place_bet_thread.daemon = True place_bet_thread.start() logger.info( f"Place the bet after: {start_after}s for: {ws.events_predictions[event_id]}", extra={"emoji": ":alarm_clock:"}, ) else: del ws.events_predictions[event_id] elif ( message.type == "event-updated" and event_id in ws.events_predictions ): ws.events_predictions[event_id].status = event_status # Game over we can't update anymore the values... The bet was placed! if ( ws.events_predictions[event_id].bet_placed is False and ws.events_predictions[event_id].bet.decision == {} ): ws.events_predictions[event_id].bet.update_outcomes( event_dict["outcomes"] ) elif message.topic == "predictions-user-v1": event_id = message.data["prediction"]["event_id"] if event_id in ws.events_predictions: if message.type == "prediction-result": event_result = message.data["prediction"]["result"] logger.info( f"{ws.events_predictions[event_id]} - Result: {event_result['type']}, Points won: {millify(event_result['points_won']) if event_result['points_won'] else 0}", extra={"emoji": ":bar_chart:"}, ) points_won = ( event_result["points_won"] if event_result["points_won"] else 0 ) ws.events_predictions[event_id].final_result = { "type": event_result["type"], "won": points_won, } elif message.type == "prediction-made": ws.events_predictions[event_id].bet_confirmed = True elif message.topic == "user-drop-events": if message.type == "drop-progress": current = message.data["current_progress_min"] required = message.data["required_progress_min"] if current >= required: try: drop = ws.twitch.search_drop_in_inventory( ws.streamers[streamer_index], message.data["drop_id"], ) if drop["dropInstanceID"] is not None: ws.twitch.claim_drop( drop["dropInstanceID"], ws.streamers[streamer_index], ) except TimeBasedDropNotFound: logger.error( f"Unable to find {message.data['drop_id']} in your inventory" ) else: # Skip 0% and 100% ... percentage_state = int((current / required) * 100) if percentage_state != 0 and percentage_state % 25 == 0: logger.info( f"Drop event {percentage_state}% for {ws.streamers[streamer_index]}!", extra={"emoji": ":package:"}, ) except Exception: logger.error( f"Exception raised for topic: {message.topic} and message: {message}", exc_info=True, ) elif response["type"] == "RESPONSE" and len(response.get("error", "")) > 0: raise RuntimeError(f"Error while trying to listen for a topic: {response}") elif response["type"] == "RECONNECT": logger.info(f"Reconnection required and keep running is: {ws.keep_running}") ws.is_reconneting = True WebSocketsPool.handle_websocket_reconnection(ws) elif response["type"] == "PONG": ws.last_pong = time.time()