From 2ed3745b7fa65c38b13a77b1e05448edf5963ace Mon Sep 17 00:00:00 2001 From: gpkvt Date: Fri, 12 Aug 2022 17:08:02 +0200 Subject: [PATCH] Lazy formatting and some other linting --- tts.py | 178 ++++++++++++++++++++++++++++++++++++--------------------- 1 file changed, 113 insertions(+), 65 deletions(-) diff --git a/tts.py b/tts.py index 127b93e..59b9bba 100644 --- a/tts.py +++ b/tts.py @@ -1,5 +1,6 @@ #!/usr/bin/python3 # -*- coding: utf-8 -*- +# pylint: disable=line-too-long """ TwitchTTS @@ -20,7 +21,6 @@ """ import json -import yaml import logging import socket import sys @@ -35,7 +35,10 @@ from http.server import BaseHTTPRequestHandler from urllib.parse import parse_qs from collections import Counter +import yaml + class IRC: + """IRC bot""" irc = socket.socket() def __init__(self): @@ -52,7 +55,8 @@ class IRC: self.tts_allowed = conf['WHITELIST_USER'] def connect(self, server, port, channel, botnick, botpass): - logging.info("Connecting to: " + server) + """Connect to Twitch IRC servers""" + logging.info("Connecting to: %s", server) try: self.irc.connect((server, port)) except ConnectionResetError: @@ -62,7 +66,9 @@ class IRC: self.irc.settimeout(1) self.irc.send(bytes("PASS " + botpass + "\r\n", "UTF-8")) - self.irc.send(bytes("USER " + botnick + " " + botnick +" " + botnick + " :python\r\n", "UTF-8")) + self.irc.send(bytes( + "USER " + botnick + " " + botnick +" " + botnick + " :python\r\n", "UTF-8") + ) self.irc.send(bytes("NICK " + botnick + "\r\n", "UTF-8")) self.irc.send(bytes("CAP REQ :twitch.tv/commands twitch.tv/tags \r\n", "UTF-8")) time.sleep(5) @@ -75,17 +81,26 @@ class IRC: logging.warning('Please check your credentials, if this error persists.') self.irc.send(bytes("JOIN " + channel + "\r\n", "UTF-8")) - def sendpriv(self, channel, user, msg): + def sendmsg(self, channel, user, msg): + """ + 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 + """ 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: + except Exception: # pylint: disable=broad-except logging.exception('An unknown error occured while getting a IRC response.') sys.exit(255) @@ -97,7 +112,7 @@ class IRC: logging.info('CLEARMSG received') msgid = False - global msg_queue_raw + global msg_queue_raw # pylint: disable=global-statement,invalid-name filtered_msg_queue = [] tags = resp.split(';') @@ -109,7 +124,7 @@ class IRC: for msg in list(msg_queue_raw): if msg['msgid'] == msgid: - logging.info('Suppressing message '+str(msgid)) + logging.info('Suppressing message %s', msgid) else: filtered_msg_queue.append(msg) @@ -141,22 +156,21 @@ class IRC: badges = tag.rsplit('badges=',1)[1] if tag.startswith('subscriber='): subscriber = tag.rsplit('subscriber=',1)[1] - logging.debug('Subscriber: '+str(subscriber)) + logging.debug('Subscriber: %s', subscriber) if tag.startswith('id='): msgid = tag.rsplit('id=',1)[1] - logging.debug('Message ID: '+str(msgid)) + logging.debug('Message ID: %s', msgid) if tag.startswith('display-name='): user = tag.rsplit('display-name=',1)[1].lower() - logging.debug('Username: '+str(user)) + logging.debug('Username: %s', user) msg = resp.rsplit('PRIVMSG #',1)[1] msg = msg.split(':',1)[1] msg = msg.replace('\r\n','') msglen = len(msg) - logging.debug('Msg:') - logging.debug(msg) - logging.debug('Msg length: '+str(msglen)) + logging.debug('Msg: %s', msg) + logging.debug('Msg length: %s', msglen) logging.debug('Deny List:') logging.debug(self.tts_denied) @@ -169,17 +183,17 @@ class IRC: logging.debug('Removing "@" from username') user = user.replace('@', '') if user not in self.tts_denied: - logging.info("Adding "+str(user)+" to deny list") + logging.info("Adding %s to deny list", user) self.tts_denied.append(user) if user in self.tts_allowed: - logging.info("Removing "+str(user)+" from allowed list") + logging.info("Removing %s from allowed list", user) self.tts_allowed.remove(user) return True if msg.startswith('!ping'): logging.debug("Ping check received.") - self.sendpriv(conf['IRC_CHANNEL'], "@"+str(user), "Pong!") + self.sendmsg(conf['IRC_CHANNEL'], "@"+str(user), "Pong!") return True @@ -190,8 +204,8 @@ class IRC: if self.pollcount == 0: logging.info("Nobody voted") - self.sendpriv(conf['IRC_CHANNEL'], "@chat", conf['MESSAGE']['VOTEEND']) - self.sendpriv(conf['IRC_CHANNEL'], "*", conf['MESSAGE']['VOTENOBODY']) + self.sendmsg(conf['IRC_CHANNEL'], "@chat", conf['MESSAGE']['VOTEEND']) + self.sendmsg(conf['IRC_CHANNEL'], "*", conf['MESSAGE']['VOTENOBODY']) self.quickvote = False self.poll = {} return False @@ -199,12 +213,15 @@ class IRC: logging.info("Counting votes") count = 0 count = Counter(self.poll.values()).most_common(5) - self.sendpriv(conf['IRC_CHANNEL'], "@chat", conf['MESSAGE']['VOTEEND']) + self.sendmsg(conf['IRC_CHANNEL'], "@chat", conf['MESSAGE']['VOTEEND']) logging.debug(count) for key, value in count: - self.sendpriv(conf['IRC_CHANNEL'], "*", str(key)+" ("+str(value)+ " "+ conf['MESSAGE']['VOTES'] + ")") + self.sendmsg( + conf['IRC_CHANNEL'], "*", + str(key)+" ("+str(value)+ " "+ conf['MESSAGE']['VOTES'] + ")" + ) self.quickvote = False self.poll = {} @@ -215,9 +232,12 @@ class IRC: self.quickvote = True self.votemsg = resp.split('!quickvote', 1)[1].strip() if self.votemsg: - self.sendpriv(conf['IRC_CHANNEL'], "@chat", conf['MESSAGE']['VOTESTART'] + " (" + str(self.votemsg) + ")") + self.sendmsg( + conf['IRC_CHANNEL'], "@chat", + conf['MESSAGE']['VOTESTART'] + " (" + str(self.votemsg) + ")" + ) else: - self.sendpriv(conf['IRC_CHANNEL'], "@chat", conf['MESSAGE']['VOTESTART']) + self.sendmsg(conf['IRC_CHANNEL'], "@chat", conf['MESSAGE']['VOTESTART']) return True if msg.startswith('!ptts'): @@ -228,16 +248,16 @@ class IRC: logging.debug('Removing "@" from username') user = user.replace('@', '') - logging.info("Adding "+str(user)+" to whitelist") + logging.info("Adding %s to whitelist", user) self.tts_allowed.append(user) if user in self.tts_denied: - logging.info("Removing "+str(user)+" from deny list") + logging.info("Removing %s from deny list", user) self.tts_denied.remove(user) return True - if msg.startswith('#') and self.quickvote == True: + if msg.startswith('#') and self.quickvote is True: logging.info('Quickvote: Cast detected') self.pollcount += 1 self.poll[user] = msg.lower() @@ -248,7 +268,7 @@ class IRC: msg_queue.clear() msg_queue_raw.clear() self.tts_status = False - self.sendpriv(conf['IRC_CHANNEL'], "@"+str(user), conf['MESSAGE']['TOFF']) + self.sendmsg(conf['IRC_CHANNEL'], "@"+str(user), conf['MESSAGE']['TOFF']) return True @@ -257,7 +277,7 @@ class IRC: msg_queue.clear() msg_queue_raw.clear() self.tts_status = True - self.sendpriv(conf['IRC_CHANNEL'], "@"+str(user), conf['MESSAGE']['TON']) + self.sendmsg(conf['IRC_CHANNEL'], "@"+str(user), conf['MESSAGE']['TON']) return True @@ -266,20 +286,20 @@ class IRC: if msglen > conf['IRC_TTS_LEN']: logging.info('TTS message is to long') - self.sendpriv(conf['IRC_CHANNEL'], "@"+str(user), conf['MESSAGE']['TOO_LONG']) + self.sendmsg(conf['IRC_CHANNEL'], "@"+str(user), conf['MESSAGE']['TOO_LONG']) return False - logging.debug("tts status: "+str(self.tts_status)) + logging.debug("tts status: %s", self.tts_status) logging.debug(conf['TTS_STARTENABLED']) if not self.tts_status: logging.info('TTS is disabled') - self.sendpriv(conf['IRC_CHANNEL'], "@"+str(user), conf['MESSAGE']['DISABLED']) + self.sendmsg(conf['IRC_CHANNEL'], "@"+str(user), conf['MESSAGE']['DISABLED']) return False if user in self.tts_denied: - logging.info(str(user) + " is not allowed to use TTS") - self.sendpriv(conf['IRC_CHANNEL'], "@"+str(user), conf['MESSAGE']['DENIED']) + logging.info("%s is not allowed to use TTS", user) + self.sendmsg(conf['IRC_CHANNEL'], "@"+str(user), conf['MESSAGE']['DENIED']) return False if conf['IRC_SUBONLY']: @@ -287,7 +307,7 @@ class IRC: logging.debug('TTS is sub-only and user has allowance') else: logging.info('TTS is sub-only') - self.sendpriv(conf['IRC_CHANNEL'], "@"+str(user), conf['MESSAGE']['SUBONLY']) + self.sendmsg(conf['IRC_CHANNEL'], "@"+str(user), conf['MESSAGE']['SUBONLY']) return False if conf['IRC_MODONLY']: @@ -295,25 +315,41 @@ class IRC: logging.debug('TTS is mod-only and user has allowance') else: logging.info('TTS is sub-only') - self.sendpriv(conf['IRC_CHANNEL'], "@"+str(user), conf['MESSAGE']['MODONLY']) + self.sendmsg(conf['IRC_CHANNEL'], "@"+str(user), conf['MESSAGE']['MODONLY']) return False if conf['WHITELIST']: - if not user in self.tts_allowed: + if user not in self.tts_allowed: logging.info('User is not on whitelist') logging.info(self.tts_allowed) - self.sendpriv(conf['IRC_CHANNEL'], "@"+str(user), conf['MESSAGE']['WHITELISTONLY']) + self.sendmsg( + conf['IRC_CHANNEL'], + "@"+str(user), conf['MESSAGE']['WHITELISTONLY'] + ) return False else: logging.info('Nobody is on the whitelist.') - self.sendpriv(conf['IRC_CHANNEL'], "@"+str(user), conf['MESSAGE']['WHITELISTONLY']) + self.sendmsg( + conf['IRC_CHANNEL'], + "@"+str(user), conf['MESSAGE']['WHITELISTONLY'] + ) return False logging.info('Valid TTS message, adding to raw queue') tts = True now = datetime.datetime.now() msg = msg.replace('!tts','',1) - msg = {"TTS": tts, "msg": msg, "badges": badges, "subscriber": subscriber, "msgid": msgid, "user": user, "length": msglen, "queuetime": now, "timestamp": str(time.time_ns())} + msg = { + "TTS": tts, + "msg": msg, + "badges": badges, + "subscriber": subscriber, + "msgid": msgid, + "user": user, + "length": msglen, + "queuetime": now, + "timestamp": str(time.time_ns()) + } msg_queue_raw.append(msg) return True @@ -321,48 +357,52 @@ class IRC: return False class HTTPserv(BaseHTTPRequestHandler): - def log_message(self, format, *args): + """Simple HTTP Server""" + + def log_message(self, format, *args): # pylint: disable=redefined-builtin + """Suppress HTTP log messages""" return - def do_GET(self): + def do_GET(self): # pylint: disable=invalid-name + """Process GET requests""" if self.path == '/': self.send_response(200) self.send_header('Content-type', 'text/html') self.end_headers() - with open("tts.html", "rb") as fh: - html = fh.read() + with open("tts.html", "rb") as file: + html = file.read() self.wfile.write(html) elif self.path == '/favicon.ico': self.send_response(200) self.send_header('Content-type', 'image/x-icon') self.end_headers() - with open("favicon.ico", "rb") as fh: - icon = fh.read() + with open("favicon.ico", "rb") as file: + icon = file.read() self.wfile.write(icon) elif self.path == '/tts.js': self.send_response(200) self.send_header('Content-type', 'text/javascript') self.end_headers() - with open("tts.js", "rb") as fh: - html = fh.read() + with open("tts.js", "rb") as file: + html = file.read() self.wfile.write(html) elif self.path == '/jquery.js': self.send_response(200) self.send_header('Content-type', 'text/javascript') self.end_headers() - with open("jquery.js", "rb") as fh: - html = fh.read() + with open("jquery.js", "rb") as file: + html = file.read() self.wfile.write(html) elif self.path == '/bootstrap.min.css': self.send_response(200) self.send_header('Content-type', 'text/css') self.end_headers() - with open("bootstrap.min.css", "rb") as fh: - html = fh.read() + with open("bootstrap.min.css", "rb") as file: + html = file.read() self.wfile.write(html) elif self.path.startswith('/tts_queue'): @@ -378,9 +418,9 @@ class HTTPserv(BaseHTTPRequestHandler): for key in list(sorted_tts.keys()): if key not in tts_done: if msg_queue[key][0].lower() in usermap: - tts = {key: usermap[msg_queue[key][0].lower()]+" "+str(conf['MESSAGE']['SAYS'])+":"+msg_queue[key][1]} + tts = {str(key): str(usermap[msg_queue[key][0].lower()]) + " " + str(conf['MESSAGE']['SAYS']) + ":" + str(msg_queue[key][1])} else: - tts = {key: msg_queue[key][0]+" "+str(conf['MESSAGE']['SAYS'])+":"+msg_queue[key][1]} + tts = {str(key): str(msg_queue[key][0]) + " " + str(conf['MESSAGE']['SAYS']) + ":" + str(msg_queue[key][1])} tts_json = json.dumps(tts) self.wfile.write(bytes(str(tts_json)+"\n", "utf-8")) @@ -433,7 +473,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: + except Exception: # pylint: disable=broad-except logging.error('Could not fetch OAuth-URL from Twitch.') else: self.send_response(404) @@ -445,13 +485,16 @@ class HTTPserv(BaseHTTPRequestHandler): return def http_serve_forever(httpd): - httpd.serve_forever() + """httpd loop""" + httpd.serve_forever() def load_config(): + """Loading config variables""" + logging.info("Loading configfile") try: - with open("config.yml", "r") as ymlfile: + with open("config.yml", "r", encoding="UTF-8") as ymlfile: cfg = yaml.load(ymlfile, Loader=yaml.Loader) except FileNotFoundError: logging.fatal('Your config file is missing, please copy config-dist.yml to config.yml and review your settings.') @@ -459,18 +502,18 @@ def load_config(): for section in cfg: try: - logging.debug('Fetching config: '+str(section)) + logging.debug('Fetching config: %s', section) conf['IRC_CHANNEL'] = cfg['irc']['channel'] conf['IRC_USERNAME'] = cfg['irc']['username'] conf['IRC_OAUTH_TOKEN'] = cfg['irc']['oauth_token'] conf['IRC_SERVER'] = cfg['irc']['server'] or "irc.chat.twitch.tv" conf['IRC_CLEARMSG_TIMEOUT'] = cfg['irc']['clearmsg_timeout'] or 60 - + conf['IRC_SUBONLY'] = cfg['bot']['subonly'] or False conf['IRC_MODONLY'] = cfg['bot']['modonly'] or False conf['IRC_TTS_LEN'] = cfg['bot']['message_length'] or 200 conf['TTS_STARTENABLED'] = cfg['bot']['start_enabled'] or False - + conf['LOG_LEVEL'] = cfg['log']['level'] or "INFO" conf['HTTP_PORT'] = cfg['http']['port'] or 80 conf['HTTP_BIND'] = cfg['http']['bind'] or "localhost" @@ -530,12 +573,16 @@ logging.basicConfig(level=logging.DEBUG, format='%(asctime)s %(module)s %(thread sys.tracebacklimit = 0 def main(): + """Main loop""" + + global conf # pylint: disable=global-statement,invalid-name + conf = load_config() lastreload = datetime.datetime.now() logging.getLogger().setLevel(conf['LOG_LEVEL']) if conf['LOG_LEVEL'] == 'DEBUG': sys.tracebacklimit = 5 - + logging.info("Starting Webserver") httpd = socketserver.TCPServer((conf['HTTP_BIND'], conf['HTTP_PORT']), HTTPserv) httpd.allow_reuse_port = True @@ -547,15 +594,16 @@ def main(): if conf['IRC_OAUTH_TOKEN'] == "Invalid": logging.error('No OAuth Token, skipping start of IRC bot.') while True: - logging.error('Please open http://'+str(conf['HTTP_BIND'])+':'+str(conf['HTTP_PORT'])+'/token to generate your OAuth-Token.') + logging.error('Please open http://%s:%s/token to generate your OAuth-Token.', conf['HTTP_BIND'], conf['HTTP_PORT']) time.sleep(10) else: logging.info("Starting IRC bot") irc = IRC() - irc.connect(conf['IRC_SERVER'], 6667, conf['IRC_CHANNEL'], conf['IRC_USERNAME'], conf['IRC_OAUTH_TOKEN']) - irc.sendpriv(conf['IRC_CHANNEL'], 'MrDestructoid', conf['MESSAGE']['READY']) - logging.info("Please open your browser and visit: http://"+str(conf['HTTP_BIND']+":"+str(conf['HTTP_PORT'])+"/")) + irc.connect(conf['IRC_SERVER'], 6667, conf['IRC_CHANNEL'], conf['IRC_USERNAME'], conf['IRC_OAUTH_TOKEN']) + irc.sendmsg(conf['IRC_CHANNEL'], 'MrDestructoid', conf['MESSAGE']['READY']) + + logging.info("Please open your browser and visit: http://%s:%s/", conf['HTTP_BIND'], conf['HTTP_PORT']) while True: if conf['LOG_LEVEL'] == "DEBUG": @@ -574,10 +622,10 @@ def main(): if confreload - lastreload > datetime.timedelta(seconds=60): conf = load_config() lastreload = datetime.datetime.now() - + if irc.quickvote and irc.votemsg: logging.info('Quickvote is active') - irc.sendpriv(conf['IRC_CHANNEL'], "@chat", conf['MESSAGE']['VOTESTART'] + " (" + str(irc.votemsg) + ")") + irc.sendmsg(conf['IRC_CHANNEL'], "@chat", conf['MESSAGE']['VOTESTART'] + " (" + str(irc.votemsg) + ")") logging.debug('Raw message queue:') logging.debug(msg_queue_raw)