From 1cf77afbccc2ca04f6a23f0d8b8228da8fb309a5 Mon Sep 17 00:00:00 2001 From: gpkvt Date: Sat, 27 Aug 2022 15:19:44 +0200 Subject: [PATCH] Added/Improved docstrings --- tts.py | 110 ++++++++++++++++++++++++++++++++++++++------------------- 1 file changed, 74 insertions(+), 36 deletions(-) diff --git a/tts.py b/tts.py index 073b382..c479726 100644 --- a/tts.py +++ b/tts.py @@ -1,6 +1,6 @@ #!/usr/bin/python3 # -*- coding: utf-8 -*- -# pylint: disable=bare-except +# pylint: disable=global-statement,broad-except,line-too-long,too-many-lines """ TwitchTTS @@ -67,7 +67,14 @@ class IRC: self.tts_allowed = CONF['WHITELIST_USER'] 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('Waiting...') try: @@ -95,23 +102,27 @@ class IRC: self.irc.send(bytes("JOIN " + channel + "\r\n", "UTF-8")) def sendmsg(self, channel, user, msg): - """ - Send (a public) message to IRC channel + """ Send (a public) message to IRC channel - Parameters: - channel (str): Channel to post msg to - user (str): User to address message to - msg (str): Message to send + :param str channel: Channel to post msg to + :param str user: User to address message to + :param str msg: Message to send """ self.irc.send(bytes("PRIVMSG "+channel+" :"+user+" "+msg+"\r\n", "UTF-8")) def resp_ping(self): - """ Respond to PING """ + """ Respond to PING + + Respond to PING messages, to keep the connection alive. + """ logging.debug('PING received') self.irc.send(bytes('PONG :tmi.twitch.tv\r\n', "UTF-8")) 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: try: raise RuntimeError() @@ -124,7 +135,7 @@ class IRC: logging.info('CLEARMSG received') msgid = False - global msg_queue_raw # pylint: disable=global-statement,invalid-name + global MSG_QUEUE_RAW filtered_msg_queue = [] tags = resp.split(';') @@ -134,13 +145,13 @@ class IRC: logging.debug('Trying to suppress message') logging.debug(msgid) - for msg in list(msg_queue_raw): + for msg in list(MSG_QUEUE_RAW): if msg['msgid'] == msgid: logging.info('Suppressing message %s', msgid) else: filtered_msg_queue.append(msg) - msg_queue_raw = filtered_msg_queue + MSG_QUEUE_RAW = filtered_msg_queue def resp_privmsg(self, resp): """ Respond to PRIVMSG """ @@ -276,14 +287,14 @@ class IRC: elif msg.startswith('!toff'): logging.info('TTS is now turned off') msg_queue.clear() - msg_queue_raw.clear() + MSG_QUEUE_RAW.clear() self.tts_status = False self.sendmsg(CONF['IRC_CHANNEL'], "@"+str(user), CONF['MESSAGE']['TOFF']) elif msg.startswith('!ton'): logging.info('TTS is now turned on') msg_queue.clear() - msg_queue_raw.clear() + MSG_QUEUE_RAW.clear() self.tts_status = True self.sendmsg(CONF['IRC_CHANNEL'], "@"+str(user), CONF['MESSAGE']['TON']) @@ -342,7 +353,13 @@ class IRC: return False 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: logging.info("%s is not allowed to use TTS", user) self.sendmsg(CONF['IRC_CHANNEL'], "@"+str(user), CONF['MESSAGE']['DENIED']) @@ -352,7 +369,12 @@ class IRC: return False 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 user not in self.tts_allowed: @@ -366,7 +388,11 @@ class IRC: return False 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') tts = True @@ -394,7 +420,7 @@ class IRC: "timestamp": timestamp } - msg_queue_raw.append(msg) + MSG_QUEUE_RAW.append(msg) def get_response(self): """Get and process response from IRC""" @@ -404,7 +430,7 @@ class IRC: logging.debug(resp) except socket.timeout: return - except Exception: # pylint: disable=broad-except + except Exception: logging.exception('An unknown error occured while getting a IRC response.') sys.exit(255) @@ -466,6 +492,9 @@ class IRC: """ !addquote command Adds a newline to quotes.txt + + :param dict tags: Message metadata (tags) + :param str msg: IRC message triggering the command """ user = tags['user'] @@ -514,7 +543,7 @@ class IRC: game = data['data'][0]['game_name'] quote = f"#{nol}: \"{quote[1]}\" -{username}/{game} ({date})\n" - except: + except Exception: logging.warning('Could not get metadata for quote') quote = f"#{nol}: \"{quote[1]}\" -{username} ({date})\n" @@ -539,7 +568,13 @@ class IRC: IRC.sendmsg(self, CONF['IRC_CHANNEL'], "@"+str(user), 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: user = tags['user'] @@ -567,16 +602,19 @@ class IRC: except wikipedia.exceptions.DisambiguationError: 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']) def quote(self, tags, msg = False): """ !smartquote command 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 fetch the best matching line. If nothing is given - it fetch a random line. + it fetches the given line number. If a string is given + it fetches the best matching line. If nothing is given + it fetches a random line. + + :param dict tags: Message metadata (tags) + :param str msg: IRC message triggering the command """ try: @@ -676,7 +714,7 @@ class IRC: try: delay = msg.split(' ')[1] - except: + except Exception: delay = False if delay: @@ -703,7 +741,7 @@ class IRC: splitmsg = msg.split(" ") username, *mappingname = splitmsg mappingname = ' '.join(mappingname) - except: + except Exception: username = False mappingname = False @@ -734,7 +772,7 @@ class IRC: converted_picks = [str(element) for element in picks] joined_picks = " ".join(converted_picks) - except: + except Exception: logging.error("There was an error during picking.") joined_picks = False @@ -1085,7 +1123,7 @@ class HTTPserv(BaseHTTPRequestHandler): self.send_header('Content-type', 'text/plain') self.end_headers() 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.') else: self.send_response(404) @@ -1200,8 +1238,8 @@ def load_config(): def send_tts_queue(): """ Send messages to TTS """ - for raw_msg in msg_queue_raw: - logging.debug('Raw msg: %s', msg_queue_raw) + 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']): @@ -1229,7 +1267,7 @@ def get_url(path=False): def check_oauth_token(): """ Check for valid authentication via Twitch API """ - global CONF # pylint: disable=global-statement + global CONF logging.debug('Checking OAuth Token') try: @@ -1255,7 +1293,7 @@ def check_oauth_token(): def main(): """Main loop""" - global CONF # pylint: disable=global-statement + global CONF CONF = load_config() lastreload = datetime.datetime.now() @@ -1300,7 +1338,7 @@ def main(): if not irc.tts_status: continue - logging.debug('msg_queue_raw: %s', msg_queue_raw) + logging.debug('MSG_QUEUE_RAW: %s', MSG_QUEUE_RAW) send_tts_queue() except KeyboardInterrupt: @@ -1315,7 +1353,7 @@ if __name__ == "__main__": VERSION = "1.7.1" CONF = {} tts_done = [] - msg_queue_raw = [] + MSG_QUEUE_RAW = [] msg_queue = {} if sys.argv[1:]: