Restructured code

This commit is contained in:
gpkvt 2022-08-13 19:35:02 +02:00
parent 2b4da04127
commit bd5e884a17
1 changed files with 316 additions and 227 deletions

299
tts.py
View File

@ -95,23 +95,22 @@ class IRC:
"""
self.irc.send(bytes("PRIVMSG "+channel+" :"+user+" "+msg+"\r\n", "UTF-8"))
def get_response(self):
"""Get and process response from IRC"""
try:
resp = self.irc.recv(2048).decode("UTF-8")
logging.debug('resp:')
logging.debug(resp)
except socket.timeout:
return False
except Exception: # pylint: disable=broad-except
logging.exception('An unknown error occured while getting a IRC response.')
sys.exit(255)
if resp.find('PING') != -1:
def resp_ping(self):
""" Respond to PING """
logging.debug('PING received')
self.irc.send(bytes('PONG :tmi.twitch.tv\r\n', "UTF-8"))
if resp.find('CLEARMSG') != -1:
def resp_notice(self, resp):
""" Respond to NOTICE """
if 'Login authentication failed' in resp:
try:
raise RuntimeError()
except RuntimeError:
logging.exception('Login failed, please check your credentials and try again.')
sys.exit(251)
def resp_clearmsg(self, resp):
""" Respond to CLEARMSG """
logging.info('CLEARMSG received')
msgid = False
@ -133,25 +132,35 @@ class IRC:
msg_queue_raw = filtered_msg_queue
return True
if resp.find('NOTICE') != -1:
if 'Login authentication failed' in resp:
try:
raise RuntimeError()
except RuntimeError:
logging.exception('Login failed, please check your credentials and try again.')
sys.exit(251)
if resp.find('PRIVMSG') != -1:
def resp_privmsg(self, resp):
""" Respond to PRIVMSG """
logging.debug('PRIVMSG received')
badges = False
subscriber = False
msgid = False
msg = False
msglen = False
user = False
tts = False
tags = self.get_tags(resp)
message = self.get_message(resp)
user = tags['user']
msg = message['message']
msglen = message['length']
logging.debug('Msg: %s', msg)
logging.debug('Msg length: %s', msglen)
logging.debug('Deny List:')
logging.debug(self.tts_denied)
self.priviledged_commands(message, tags)
if msg.startswith('#') and self.quickvote_status is True:
logging.info('Quickvote: Cast detected')
self.pollcount += 1
self.poll[user] = msg.lower()
logging.debug(self.poll)
if msg.startswith('!tts'):
self.tts_command(message, tags)
def get_tags(self, resp):
""" Strip tags from response """
tags = resp.split(';')
for tag in tags:
@ -167,100 +176,128 @@ class IRC:
user = tag.rsplit('display-name=',1)[1].lower()
logging.debug('Username: %s', user)
msg = resp.rsplit('PRIVMSG #',1)[1]
msg = msg.split(':',1)[1]
msg = msg.replace('\r\n','')
msglen = len(msg)
tags = {}
tags['badges'] = badges
tags['subscriber'] = subscriber
tags['msgid'] = msgid
tags['user'] = user
logging.debug('Msg: %s', msg)
logging.debug('Msg length: %s', msglen)
logging.debug('Deny List:')
logging.debug(self.tts_denied)
return tags
if msg.startswith('#') and self.quickvote_status is True:
logging.info('Quickvote: Cast detected')
self.pollcount += 1
self.poll[user] = msg.lower()
logging.debug(self.poll)
return True
def get_message(self, resp):
""" Process message """
msg = {}
msg['message'] = resp.rsplit('PRIVMSG #',1)[1].split(':',1)[1].replace('\r\n','')
msg['length'] = len(msg['message'])
return msg
def priviledged_commands(self, message, tags):
""" Process priviledged commands """
msg = message['message']
badges = tags['badges']
user = tags['user']
if 'broadcaster' in badges or 'moderator' in badges:
if msg.startswith('!ping'):
logging.debug("Ping check received.")
self.sendmsg(conf['IRC_CHANNEL'], "@"+str(user), "Pong!")
return True
if msg.startswith('!dtts'):
elif msg.startswith('!dtts'):
logging.debug("!dtts command detected")
self.Commands.dtts(self, msg)
return True
if msg.startswith('!random'):
elif msg.startswith('!random'):
logging.info('!random command detected')
self.Commands.random(self, msg)
return True
if msg.startswith('!quickvote'):
elif msg.startswith('!quickvote'):
logging.info("!quickvote command detected")
self.Commands.quickvote(self, msg)
return True
if msg.startswith('!ptts'):
elif msg.startswith('!ptts'):
logging.debug("!ptts command detected")
self.Commands.ptts(self, msg)
return True
if msg.startswith('!toff'):
elif msg.startswith('!toff'):
logging.info('TTS is now turned off')
msg_queue.clear()
msg_queue_raw.clear()
self.tts_status = False
self.sendmsg(conf['IRC_CHANNEL'], "@"+str(user), conf['MESSAGE']['TOFF'])
return True
if msg.startswith('!ton'):
elif msg.startswith('!ton'):
logging.info('TTS is now turned on')
msg_queue.clear()
msg_queue_raw.clear()
self.tts_status = True
self.sendmsg(conf['IRC_CHANNEL'], "@"+str(user), conf['MESSAGE']['TON'])
def check_subonly(self, tags):
""" subonly """
subscriber = tags['subscriber']
badges = tags['badges']
user = tags['user']
if subscriber != "0" or 'moderator' in badges or 'broadcaster' in badges:
logging.debug('TTS is sub-only and user has allowance')
return False
logging.info('TTS is sub-only')
self.sendmsg(conf['IRC_CHANNEL'], "@"+str(user), conf['MESSAGE']['SUBONLY'])
return True
if msg.startswith('!tts'):
logging.debug('!tts command detected')
logging.debug("tts status: %s", self.tts_status)
def check_modonly(self, tags):
""" modonly """
badges = tags['badges']
user = tags['user']
if 'moderator' in badges or 'broadcaster' in badges:
logging.debug('TTS is mod-only and user has allowance')
return False
logging.info('TTS is sub-only')
self.sendmsg(conf['IRC_CHANNEL'], "@"+str(user), conf['MESSAGE']['MODONLY'])
return True
def check_tts_enabled(self, user):
""" Check if TTS is enabled """
if not self.tts_status:
logging.info('TTS is disabled')
self.sendmsg(conf['IRC_CHANNEL'], "@"+str(user), conf['MESSAGE']['DISABLED'])
return False
if msglen > conf['IRC_TTS_LEN']:
logging.debug('TTS is enabled')
return True
def check_msg_too_long(self, message, user):
""" Check if message is too long """
if message['length'] > conf['IRC_TTS_LEN']:
logging.info('TTS message is to long')
self.sendmsg(conf['IRC_CHANNEL'], "@"+str(user), conf['MESSAGE']['TOO_LONG'])
return True
logging.info('Check length: Message is ok')
return False
if conf['IRC_SUBONLY']:
if subscriber != "0" or 'moderator' in badges or 'broadcaster' in badges:
logging.debug('TTS is sub-only and user has allowance')
else:
logging.info('TTS is sub-only')
self.sendmsg(conf['IRC_CHANNEL'], "@"+str(user), conf['MESSAGE']['SUBONLY'])
return False
if conf['IRC_MODONLY']:
if 'moderator' in badges or 'broadcaster' in badges:
logging.debug('TTS is mod-only and user has allowance')
else:
logging.info('TTS is sub-only')
self.sendmsg(conf['IRC_CHANNEL'], "@"+str(user), conf['MESSAGE']['MODONLY'])
return False
def check_user_denied(self, user):
""" Check if user is on denied list """
if user in self.tts_denied:
logging.info("%s is not allowed to use TTS", user)
self.sendmsg(conf['IRC_CHANNEL'], "@"+str(user), conf['MESSAGE']['DENIED'])
return True
logging.debug("%s is allowed to use TTS", user)
return False
def check_whitelist(self, user):
""" Check Whitelist """
if conf['WHITELIST']:
if user not in self.tts_allowed:
logging.info('User is not on whitelist')
@ -270,18 +307,26 @@ class IRC:
"@"+str(user), conf['MESSAGE']['WHITELISTONLY']
)
return False
logging.warning('Nobody is on the whitelist.')
self.sendmsg(
conf['IRC_CHANNEL'],
"@"+str(user), conf['MESSAGE']['WHITELISTONLY']
)
return True
return False
def send_tts_msg(self, message, tags):
""" Send message to TTS queue """
logging.info('Valid TTS message, adding to raw queue')
tts = True
now = datetime.datetime.now()
timestamp = str(time.time_ns())
user = tags['user']
msgid = tags['msgid']
badges = tags['badges']
subscriber = tags['subscriber']
msg = message['message']
msglen = message['length']
msg = msg.replace('!tts','',1)
msg = {
"TTS": tts,
"msg": msg,
@ -291,13 +336,56 @@ class IRC:
"user": user,
"length": msglen,
"queuetime": now,
"timestamp": str(time.time_ns())
"timestamp": timestamp
}
msg_queue_raw.append(msg)
return True
def tts_command(self, message, tags):
""" Process !tts command """
logging.debug('!tts command detected')
logging.debug("tts status: %s", self.tts_status)
user = tags['user']
if self.check_tts_enabled(user):
return False
elif self.check_msg_too_long(message, user):
return False
elif self.check_user_denied(user):
return False
elif self.check_subonly(tags):
return False
elif self.check_modonly(tags):
return False
elif self.check_whitelist(user):
return False
self.send_tts_msg(message, tags)
def get_response(self):
"""Get and process response from IRC"""
try:
resp = self.irc.recv(2048).decode("UTF-8")
logging.debug('resp:')
logging.debug(resp)
except socket.timeout:
return
except Exception: # pylint: disable=broad-except
logging.exception('An unknown error occured while getting a IRC response.')
sys.exit(255)
if resp.find('PING') != -1:
self.resp_ping()
if resp.find('CLEARMSG') != -1:
self.resp_clearmsg(resp)
if resp.find('NOTICE') != -1:
self.resp_notice(resp)
if resp.find('PRIVMSG') != -1:
self.resp_privmsg(resp)
class Commands():
""" Bot commands """
@ -345,7 +433,7 @@ class IRC:
self.quickvote_status = False
self.poll = {}
return False
return
logging.info("Counting votes")
count = 0
@ -707,6 +795,22 @@ if sys.argv[1:]:
print('Version 1.2.1')
sys.exit(1)
def send_tts_queue():
""" Send messages to TTS """
for raw_msg in msg_queue_raw:
logging.debug('Raw msg: %s', msg_queue_raw)
now = datetime.datetime.now()
if now - raw_msg['queuetime'] > datetime.timedelta(seconds=conf['IRC_CLEARMSG_TIMEOUT']):
logging.debug('clearmsg_timeout reached')
if raw_msg['timestamp'] not in msg_queue:
logging.info('Sending TTS message')
msg_queue[raw_msg['timestamp']] = [raw_msg['user'], raw_msg['msg']]
logging.debug("msg_queue: %s", msg_queue)
else:
logging.debug('Msg is already in queue')
def main():
"""Main loop"""
@ -734,7 +838,7 @@ def main():
logging.info('Please complete the OAuth process within the next 15 minutes.')
time.sleep(900)
sys.exit(250)
else:
logging.info("Starting IRC bot")
irc = IRC()
@ -762,27 +866,12 @@ def main():
irc.sendmsg(conf['IRC_CHANNEL'], "@chat", conf['MESSAGE']['VOTESTART'] + " (" + str(irc.votemsg) + ")")
if not irc.tts_status:
logging.debug("TTS is disabled")
if conf['LOG_LEVEL'] == "DEBUG":
time.sleep(1)
continue
else:
logging.debug('Raw message queue:')
logging.debug(msg_queue_raw)
send_tts_queue()
for raw_msg in msg_queue_raw:
logging.debug('Raw msg:')
logging.debug(msg_queue_raw)
now = datetime.datetime.now()
if now - raw_msg['queuetime'] > datetime.timedelta(seconds=conf['IRC_CLEARMSG_TIMEOUT']):
logging.debug('clearmsg_timeout reached')
if raw_msg['timestamp'] not in msg_queue:
logging.info('Sending TTS message')
msg_queue[raw_msg['timestamp']] = [raw_msg['user'], raw_msg['msg']]
logging.debug(msg_queue)
else:
logging.debug('Msg is already in queue')
except KeyboardInterrupt:
logging.info('Exiting...')
sys.exit()