gpkvt/twitchtts
1
0
Fork 0
mirror of https://gitlab.com/gpvkt/twitchtts.git synced 2025-10-31 00:57:35 +01:00
Code Issues Projects Releases Packages Wiki Activity

Added/Improved docstrings

Browse source
This commit is contained in:
gpkvt 2022-08-27 15:19:44 +02:00
parent c65e3b86b1
commit 1cf77afbcc
1 changed files with 74 additions and 36 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

110
tts.py
View file

@ -1,6 +1,6 @@
#!/usr/bin/python3 #!/usr/bin/python3
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
# pylint: disable=bare-except # pylint: disable=global-statement,broad-except,line-too-long,too-many-lines
""" """
TwitchTTS TwitchTTS
@ -67,7 +67,14 @@ class IRC:
self.tts_allowed = CONF['WHITELIST_USER'] self.tts_allowed = CONF['WHITELIST_USER']
def connect(self, server, port, channel, botnick, botpass): def connect(self, server, port, channel, botnick, botpass):
""" Connect to Twitch IRC servers """ """ Connect to Twitch IRC servers
:param str server: Servername or IP
:param int port: Server Port
:param str channel: Channel to connect to
:param str botnick: Username
:param str botpass: OAuth Token
"""
logging.info("Connecting to: %s", server) logging.info("Connecting to: %s", server)
logging.info('Waiting...') logging.info('Waiting...')
try: try:
@ -95,23 +102,27 @@ class IRC:
self.irc.send(bytes("JOIN " + channel + "\r\n", "UTF-8")) self.irc.send(bytes("JOIN " + channel + "\r\n", "UTF-8"))
def sendmsg(self, channel, user, msg): def sendmsg(self, channel, user, msg):
""" """ Send (a public) message to IRC channel
Send (a public) message to IRC channel
Parameters: :param str channel: Channel to post msg to
channel (str): Channel to post msg to :param str user: User to address message to
user (str): User to address message to :param str msg: Message to send
msg (str): Message to send
""" """
self.irc.send(bytes("PRIVMSG "+channel+" :"+user+" "+msg+"\r\n", "UTF-8")) self.irc.send(bytes("PRIVMSG "+channel+" :"+user+" "+msg+"\r\n", "UTF-8"))
def resp_ping(self): def resp_ping(self):
""" Respond to PING """ """ Respond to PING
Respond to PING messages, to keep the connection alive.
"""
logging.debug('PING received') logging.debug('PING received')
self.irc.send(bytes('PONG :tmi.twitch.tv\r\n', "UTF-8")) self.irc.send(bytes('PONG :tmi.twitch.tv\r\n', "UTF-8"))
def resp_notice(self, resp): def resp_notice(self, resp):
""" Respond to NOTICE """ """ Respond to NOTICE
:param str resp: IRC Server message (in response to command)
"""
if 'Login authentication failed' in resp: if 'Login authentication failed' in resp:
try: try:
raise RuntimeError() raise RuntimeError()
@ -124,7 +135,7 @@ class IRC:
logging.info('CLEARMSG received') logging.info('CLEARMSG received')
msgid = False msgid = False
global msg_queue_raw # pylint: disable=global-statement,invalid-name global MSG_QUEUE_RAW
filtered_msg_queue = [] filtered_msg_queue = []
tags = resp.split(';') tags = resp.split(';')
@ -134,13 +145,13 @@ class IRC:
logging.debug('Trying to suppress message') logging.debug('Trying to suppress message')
logging.debug(msgid) logging.debug(msgid)
for msg in list(msg_queue_raw): for msg in list(MSG_QUEUE_RAW):
if msg['msgid'] == msgid: if msg['msgid'] == msgid:
logging.info('Suppressing message %s', msgid) logging.info('Suppressing message %s', msgid)
else: else:
filtered_msg_queue.append(msg) filtered_msg_queue.append(msg)
msg_queue_raw = filtered_msg_queue MSG_QUEUE_RAW = filtered_msg_queue
def resp_privmsg(self, resp): def resp_privmsg(self, resp):
""" Respond to PRIVMSG """ """ Respond to PRIVMSG """
@ -276,14 +287,14 @@ class IRC:
elif msg.startswith('!toff'): elif msg.startswith('!toff'):
logging.info('TTS is now turned off') logging.info('TTS is now turned off')
msg_queue.clear() msg_queue.clear()
msg_queue_raw.clear() MSG_QUEUE_RAW.clear()
self.tts_status = False self.tts_status = False
self.sendmsg(CONF['IRC_CHANNEL'], "@"+str(user), CONF['MESSAGE']['TOFF']) self.sendmsg(CONF['IRC_CHANNEL'], "@"+str(user), CONF['MESSAGE']['TOFF'])
elif msg.startswith('!ton'): elif msg.startswith('!ton'):
logging.info('TTS is now turned on') logging.info('TTS is now turned on')
msg_queue.clear() msg_queue.clear()
msg_queue_raw.clear() MSG_QUEUE_RAW.clear()
self.tts_status = True self.tts_status = True
self.sendmsg(CONF['IRC_CHANNEL'], "@"+str(user), CONF['MESSAGE']['TON']) self.sendmsg(CONF['IRC_CHANNEL'], "@"+str(user), CONF['MESSAGE']['TON'])
@ -342,7 +353,13 @@ class IRC:
return False return False
def check_user_denied(self, user): def check_user_denied(self, user):
""" Check if user is on denied list """ """ Check if user is on denied list
Check if the given user is on the TTS blacklist
:param str user: Username
"""
if user in self.tts_denied: if user in self.tts_denied:
logging.info("%s is not allowed to use TTS", user) logging.info("%s is not allowed to use TTS", user)
self.sendmsg(CONF['IRC_CHANNEL'], "@"+str(user), CONF['MESSAGE']['DENIED']) self.sendmsg(CONF['IRC_CHANNEL'], "@"+str(user), CONF['MESSAGE']['DENIED'])
@ -352,7 +369,12 @@ class IRC:
return False return False
def check_whitelist(self, user): def check_whitelist(self, user):
""" Check Whitelist """ """ Check Whitelist
Checks if the given user is on the TTS whitelist
:param str user: Username
"""
if CONF['WHITELIST']: if CONF['WHITELIST']:
if user not in self.tts_allowed: if user not in self.tts_allowed:
@ -366,7 +388,11 @@ class IRC:
return False return False
def send_tts_msg(self, message, tags): def send_tts_msg(self, message, tags):
""" Send message to TTS queue """ """ Send message to TTS queue
:param dict tags: Message metadata (tags)
:param str message: IRC message
"""
logging.info('Valid TTS message, adding to raw queue') logging.info('Valid TTS message, adding to raw queue')
tts = True tts = True
@ -394,7 +420,7 @@ class IRC:
"timestamp": timestamp "timestamp": timestamp
} }
msg_queue_raw.append(msg) MSG_QUEUE_RAW.append(msg)
def get_response(self): def get_response(self):
"""Get and process response from IRC""" """Get and process response from IRC"""
@ -404,7 +430,7 @@ class IRC:
logging.debug(resp) logging.debug(resp)
except socket.timeout: except socket.timeout:
return return
except Exception: # pylint: disable=broad-except except Exception:
logging.exception('An unknown error occured while getting a IRC response.') logging.exception('An unknown error occured while getting a IRC response.')
sys.exit(255) sys.exit(255)
@ -466,6 +492,9 @@ class IRC:
""" !addquote command """ !addquote command
Adds a newline to quotes.txt Adds a newline to quotes.txt
:param dict tags: Message metadata (tags)
:param str msg: IRC message triggering the command
""" """
user = tags['user'] user = tags['user']
@ -514,7 +543,7 @@ class IRC:
game = data['data'][0]['game_name'] game = data['data'][0]['game_name']
quote = f"#{nol}: \"{quote[1]}\" -{username}/{game} ({date})\n" quote = f"#{nol}: \"{quote[1]}\" -{username}/{game} ({date})\n"
except: except Exception:
logging.warning('Could not get metadata for quote') logging.warning('Could not get metadata for quote')
quote = f"#{nol}: \"{quote[1]}\" -{username} ({date})\n" quote = f"#{nol}: \"{quote[1]}\" -{username} ({date})\n"
@ -539,7 +568,13 @@ class IRC:
IRC.sendmsg(self, CONF['IRC_CHANNEL'], "@"+str(user), msg) IRC.sendmsg(self, CONF['IRC_CHANNEL'], "@"+str(user), msg)
def wiki(self, tags, msg): def wiki(self, tags, msg):
""" !wiki command """ """ !wiki command
Search for wikipedia articles and return the first 3 sentences
:param dict tags: Message metadata (tags)
:param str msg: IRC message triggering the command
"""
try: try:
user = tags['user'] user = tags['user']
@ -567,16 +602,19 @@ class IRC:
except wikipedia.exceptions.DisambiguationError: except wikipedia.exceptions.DisambiguationError:
IRC.sendmsg(self, CONF['IRC_CHANNEL'], "@"+str(user), CONF['MESSAGE']['WIKI_TOO_MANY']) IRC.sendmsg(self, CONF['IRC_CHANNEL'], "@"+str(user), CONF['MESSAGE']['WIKI_TOO_MANY'])
except: except Exception:
IRC.sendmsg(self, CONF['IRC_CHANNEL'], "@"+str(user), CONF['MESSAGE']['WIKI_NO_RESULT']) IRC.sendmsg(self, CONF['IRC_CHANNEL'], "@"+str(user), CONF['MESSAGE']['WIKI_NO_RESULT'])
def quote(self, tags, msg = False): def quote(self, tags, msg = False):
""" !smartquote command """ !smartquote command
Gets a line from quotes.txt. If a number if given as msg Gets a line from quotes.txt. If a number if given as msg
it fetch the given line number. If a string is given it fetches the given line number. If a string is given
it fetch the best matching line. If nothing is given it fetches the best matching line. If nothing is given
it fetch a random line. it fetches a random line.
:param dict tags: Message metadata (tags)
:param str msg: IRC message triggering the command
""" """
try: try:
@ -676,7 +714,7 @@ class IRC:
try: try:
delay = msg.split(' ')[1] delay = msg.split(' ')[1]
except: except Exception:
delay = False delay = False
if delay: if delay:
@ -703,7 +741,7 @@ class IRC:
splitmsg = msg.split(" ") splitmsg = msg.split(" ")
username, *mappingname = splitmsg username, *mappingname = splitmsg
mappingname = ' '.join(mappingname) mappingname = ' '.join(mappingname)
except: except Exception:
username = False username = False
mappingname = False mappingname = False
@ -734,7 +772,7 @@ class IRC:
converted_picks = [str(element) for element in picks] converted_picks = [str(element) for element in picks]
joined_picks = " ".join(converted_picks) joined_picks = " ".join(converted_picks)
except: except Exception:
logging.error("There was an error during picking.") logging.error("There was an error during picking.")
joined_picks = False joined_picks = False
@ -1085,7 +1123,7 @@ class HTTPserv(BaseHTTPRequestHandler):
self.send_header('Content-type', 'text/plain') self.send_header('Content-type', 'text/plain')
self.end_headers() self.end_headers()
self.wfile.write(bytes("Could not get OAuth-URL from Twitch\n", "utf-8")) self.wfile.write(bytes("Could not get OAuth-URL from Twitch\n", "utf-8"))
except Exception: # pylint: disable=broad-except except Exception:
logging.error('Could not fetch OAuth-URL from Twitch.') logging.error('Could not fetch OAuth-URL from Twitch.')
else: else:
self.send_response(404) self.send_response(404)
@ -1200,8 +1238,8 @@ def load_config():
def send_tts_queue(): def send_tts_queue():
""" Send messages to TTS """ """ Send messages to TTS """
for raw_msg in msg_queue_raw: for raw_msg in MSG_QUEUE_RAW:
logging.debug('Raw msg: %s', msg_queue_raw) logging.debug('Raw msg: %s', MSG_QUEUE_RAW)
now = datetime.datetime.now() now = datetime.datetime.now()
if now - raw_msg['queuetime'] > datetime.timedelta(seconds=CONF['IRC_CLEARMSG_TIMEOUT']): if now - raw_msg['queuetime'] > datetime.timedelta(seconds=CONF['IRC_CLEARMSG_TIMEOUT']):
@ -1229,7 +1267,7 @@ def get_url(path=False):
def check_oauth_token(): def check_oauth_token():
""" Check for valid authentication via Twitch API """ """ Check for valid authentication via Twitch API """
global CONF # pylint: disable=global-statement global CONF
logging.debug('Checking OAuth Token') logging.debug('Checking OAuth Token')
try: try:
@ -1255,7 +1293,7 @@ def check_oauth_token():
def main(): def main():
"""Main loop""" """Main loop"""
global CONF # pylint: disable=global-statement global CONF
CONF = load_config() CONF = load_config()
lastreload = datetime.datetime.now() lastreload = datetime.datetime.now()
@ -1300,7 +1338,7 @@ def main():
if not irc.tts_status: if not irc.tts_status:
continue continue
logging.debug('msg_queue_raw: %s', msg_queue_raw) logging.debug('MSG_QUEUE_RAW: %s', MSG_QUEUE_RAW)
send_tts_queue() send_tts_queue()
except KeyboardInterrupt: except KeyboardInterrupt:
@ -1315,7 +1353,7 @@ if __name__ == "__main__":
VERSION = "1.7.1" VERSION = "1.7.1"
CONF = {} CONF = {}
tts_done = [] tts_done = []
msg_queue_raw = [] MSG_QUEUE_RAW = []
msg_queue = {} msg_queue = {}
if sys.argv[1:]: if sys.argv[1:]: