mirror of
https://gitlab.com/gpvkt/twitchtts.git
synced 2025-10-31 17:17:35 +01:00
Restructured code
This commit is contained in:
gpkvt
parent
df44c29133
commit
a7aceccf4e
1 changed files with 316 additions and 227 deletions
299
tts.py
299
tts.py
|
|
@ -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()
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue