chorus-client/Other tries/client.py

352 lines
10 KiB
Python

#
# Project OBS Twitch Chat Spam Script
# @author David Madison
# @link github.com/dmadison/OBS-ChatSpam
# @license GPLv3 - Copyright (c) 2018 David Madison
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
import obspython as obs
import socket
import time
class TwitchIRC:
def __init__(self, chan="", nick="", passw="", host="irc.twitch.tv", port=6667):
self.channel = chan
self.nickname = nick
self.password = passw
self.host = host
self.port = port
self.rate_num_msgs = 19 # Number of messages allowed...
self.rate_timeframe = 30 # ...in timeframe of x seconds
self.__message_timestamps = []
self.__connected = False
self.__last_message = None # Last connection timestamp
self.timeout = 10.0 # Time before open connection is closed, in seconds
self.__sock = socket.socket()
def connect(self, suppress_warnings=True):
connection_result = self.__connect()
if connection_result is not True:
self.__connected = False
if suppress_warnings:
print("Connection Error:", connection_result)
return False
else:
raise UserWarning(connection_result)
self.__connected = True
return True
def __connect(self):
if self.__connected:
return True # Already connected, nothing to see here
self.__sock = socket.socket()
self.__sock.settimeout(1) # One second to connect
try:
self.__sock.connect((self.host, self.port))
except socket.gaierror:
return "Cannot find server"
except (TimeoutError, socket.timeout):
return "No response from server (connection timed out)"
if self.password is not "":
self.__sock.send("PASS {}\r\n".format(self.password).encode("utf-8"))
self.__sock.send("NICK {}\r\n".format(self.nickname).encode("utf-8"))
self.__sock.send("JOIN #{}\r\n".format(self.channel).encode("utf-8"))
auth_response = self.read()
if "Welcome, GLHF!" not in auth_response:
return "Bad Authentication! Check your Oauth key"
try:
self.read() # Wait for "JOIN" response
except socket.timeout:
return "Channel not found!"
return True
def disconnect(self):
if self.__connected:
self.__sock.shutdown(socket.SHUT_RDWR)
self.__sock.close()
self.__connected = False
def connection_timeout(self):
if self.__connected and time.time() >= self.__last_message + self.timeout:
self.disconnect()
def test_authentication(self):
if self.connect(False):
self.disconnect()
print("Authentication successful!")
def chat(self, msg, suppress_warnings=True):
if not self.check_rates() or not self.connect(suppress_warnings):
return
# Store timestamps for rate limit and connection timeout
message_time = time.time()
self.__message_timestamps.append(message_time + self.rate_timeframe)
self.__last_message = message_time
self.__chat_direct(msg)
print("Sent \'" + msg + "\'", "as", self.nickname, "in #" + self.channel)
def __chat_direct(self, msg):
self.__sock.send("PRIVMSG #{} :{}\r\n".format(self.channel, msg).encode("utf-8"))
def read(self):
response = self.__read_socket()
while self.__ping(response):
response = self.__read_socket()
return response.rstrip()
def __read_socket(self):
return self.__sock.recv(1024).decode("utf-8")
def __ping(self, msg):
if msg[:4] == "PING":
self.__pong(msg[4:])
return True
return False
def __pong(self, host):
self.__sock.send(("PONG" + host).encode("utf-8"))
def check_rates(self):
index = 0
# Remove timestamps that have passed
for index, timestamp in enumerate(self.__message_timestamps):
if time.time() <= timestamp:
break
self.__message_timestamps = self.__message_timestamps[index:]
# If at max rate, throw an error
if len(self.__message_timestamps) >= self.rate_num_msgs:
next_clear = int(self.__message_timestamps[0] - time.time())
msg_plural = "s"
if next_clear <= 1:
next_clear = 1 # Avoiding "wait 0 more seconds" messages
msg_plural = ""
print("Error: Rate limit reached. Please wait " + str(next_clear) + " more second" + msg_plural)
return False
return True
twitch = TwitchIRC()
class ChatMessage:
messages = []
max_description_length = 32
def __init__(self, msg, position, obs_settings, irc=twitch):
self.text = msg
self.irc = irc
self.obs_data = obs_settings
self.position = position
self.hotkey_id = obs.OBS_INVALID_HOTKEY_ID
self.hotkey_saved_key = None
self.load_hotkey()
self.register_hotkey()
self.save_hotkey()
def __del__(self):
self.cleanup()
def cleanup(self):
self.deregister_hotkey()
self.release_memory()
def release_memory(self):
obs.obs_data_array_release(self.hotkey_saved_key)
def new_text(self, msg):
self.text = msg
self.deregister_hotkey()
self.register_hotkey()
def new_position(self, pos):
self.deregister_hotkey()
self.unsave_hotkey()
self.position = pos
self.register_hotkey()
def register_hotkey(self):
if len(self.text) > ChatMessage.max_description_length:
key_description = self.text[:ChatMessage.max_description_length - 3] + "..."
else:
key_description = self.text
key_description = "Chat \'" + key_description + "\'"
self.callback = lambda pressed: self.key_passthrough(pressed) # Small hack to get around the callback signature reqs.
self.hotkey_id = obs.obs_hotkey_register_frontend("chat_hotkey", key_description, self.callback)
obs.obs_hotkey_load(self.hotkey_id, self.hotkey_saved_key)
def deregister_hotkey(self):
obs.obs_hotkey_unregister(self.callback)
def load_hotkey(self):
self.hotkey_saved_key = obs.obs_data_get_array(self.obs_data, "chat_hotkey_" + str(self.position))
def save_hotkey(self):
self.hotkey_saved_key = obs.obs_hotkey_save(self.hotkey_id)
obs.obs_data_set_array(self.obs_data, "chat_hotkey_" + str(self.position), self.hotkey_saved_key)
def unsave_hotkey(self):
obs.obs_data_erase(self.obs_data, "chat_hotkey_" + str(self.position))
def key_passthrough(self, pressed):
if pressed:
self.send()
def send(self, suppress_warnings=True):
self.irc.chat(self.text, suppress_warnings)
@staticmethod
def check_messages(new_msgs, settings):
# Check if list hasn't changed
if len(new_msgs) == len(ChatMessage.messages):
num_diff = 0
diff_index = None
for index, msg in enumerate(ChatMessage.messages):
if new_msgs[index] != msg.text:
num_diff += 1
diff_index = index
if num_diff > 1:
break
else:
if num_diff != 0:
ChatMessage.messages[diff_index].new_text(new_msgs[diff_index]) # single entry modified
return # Lists identical
# Check if objects already exist, otherwise create them
new_list = []
for pos, msg in enumerate(new_msgs):
for msg_obj in ChatMessage.messages:
if msg == msg_obj.text:
new_list.append(msg_obj)
break
else:
new_list.append(ChatMessage(msg, pos, settings))
# Clean up old objects
for msg in ChatMessage.messages:
for msg_new in new_msgs:
if msg.text == msg_new:
break
else:
msg.cleanup()
msg.unsave_hotkey()
# Assign to master array and reindex
ChatMessage.messages = new_list
ChatMessage.__reindex_messages()
@staticmethod
def __reindex_messages():
for index, msg in enumerate(ChatMessage.messages):
msg.new_position(index)
for msg in ChatMessage.messages: # Separate loop as to avoid memory overwrites
msg.save_hotkey()
# ------------------------------------------------------------
# OBS Script Functions
def check_connection():
twitch.connection_timeout()
def test_authentication(prop, props):
twitch.test_authentication()
def test_message(prop, props):
ChatMessage.messages[0].send(False)
def script_description():
return "<b>Twitch Chat Spam</b>" + \
"<hr>" + \
"Python script for sending messages to Twitch chat using OBS hotkeys." + \
"<br/><br/>" + \
"Made by David Madison, © 2018" + \
"<br/><br/>" + \
"github.com/dmadison/OBS-ChatSpam" + \
"<br/>" + \
"partsnotincluded.com"
def script_update(settings):
twitch.channel = obs.obs_data_get_string(settings, "channel").lower()
twitch.nickname = obs.obs_data_get_string(settings, "user").lower()
new_oauth = obs.obs_data_get_string(settings, "oauth").lower()
if new_oauth != twitch.password:
twitch.disconnect() # Disconnect old oauth connection, if it exists
twitch.password = new_oauth
obs_messages = obs.obs_data_get_array(settings, "messages")
num_messages = obs.obs_data_array_count(obs_messages)
messages = []
for i in range(num_messages): # Convert C array to Python list
message_object = obs.obs_data_array_item(obs_messages, i)
messages.append(obs.obs_data_get_string(message_object, "value"))
ChatMessage.check_messages(messages, settings)
obs.obs_data_array_release(obs_messages)
#print("Settings JSON", obs.obs_data_get_json(settings))
def script_properties():
props = obs.obs_properties_create()
obs.obs_properties_add_text(props, "channel", "Channel", obs.OBS_TEXT_DEFAULT)
obs.obs_properties_add_text(props, "user", "User", obs.OBS_TEXT_DEFAULT)
obs.obs_properties_add_text(props, "oauth", "Oauth", obs.OBS_TEXT_PASSWORD)
obs.obs_properties_add_editable_list(props, "messages", "Messages", obs.OBS_EDITABLE_LIST_TYPE_STRINGS, "", "")
obs.obs_properties_add_button(props, "test_auth", "Test Authentication", test_authentication)
obs.obs_properties_add_button(props, "test_message", "Test Message #1", test_message)
return props
def script_save(settings):
for message in ChatMessage.messages:
message.save_hotkey()
def script_load(settings):
obs.timer_add(check_connection, 1000) # Check for timeout every second
def script_unload():
obs.timer_remove(check_connection)
for message in ChatMessage.messages:
message.cleanup()
cargo