gpkvt
/
twitchtts
mirror of https://gitlab.com/gpvkt/twitchtts.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Added OAuth check

This commit is contained in:
gpkvt 2022-08-14 11:41:05 +02:00
parent 26e81f7aeb
commit f5ae18d0ea
2 changed files with 101 additions and 54 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
CHANGELOG.md
View File

@ -2,9 +2,19 @@
All notable changes to this project will be documented in this file. If there is a `Changed` section please read carefully, as this often means that you will need to adapt your `config.yml`, otherwise the bot might fail to start.
## [1.2.4] - 2022-08-15
### Added 1.2.4
* Check OAuth Token via Twitch API
### Fixed 1.2.4
* Internal URL when using special HTTP_BIND values
## [1.2.3] - 2022-08-14
### Fixed 1.2.2
### Fixed 1.2.3
* Message sort order

143
tts.py
View File

@ -36,12 +36,13 @@ import urllib.request
from threading import Thread
from collections import Counter
from urllib.parse import parse_qs
from urllib.error import HTTPError
from http.server import BaseHTTPRequestHandler
import yaml
class IRC:
"""IRC bot"""
""" IRC bot """
irc = socket.socket()
def __init__(self):
@ -58,7 +59,7 @@ 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 """
logging.info("Connecting to: %s", server)
try:
self.irc.connect((server, port))
@ -154,11 +155,11 @@ class IRC:
logging.info('Quickvote: Cast detected')
self.pollcount += 1
self.poll[user] = msg.lower()
logging.debug(self.poll)
logging.debug("poll: %s", self.poll)
if msg.startswith('!tts'):
logging.info('!tts command detected')
self.tts_command(message, tags)
self.Commands.tts(self, message, tags)
def get_tags(self, resp):
""" Strip tags from response """
@ -345,27 +346,6 @@ class IRC:
msg_queue_raw.append(msg)
def tts_command(self, message, tags):
""" Process !tts command """
user = tags['user']
if self.check_tts_disabled(user):
logging.info('TTS is disabled')
elif self.check_msg_too_long(message, user):
logging.info('TTS message is too long')
elif self.check_user_denied(user):
logging.info('User is not allowed to use TTS')
elif self.check_subonly(tags):
logging.info('TTS is sub-only')
elif self.check_modonly(tags):
logging.info('TTS is mod-only')
elif self.check_whitelist(user):
logging.info('User is not on whitelist')
else:
logging.info('Sending TTS message to raw_queue')
self.send_tts_msg(message, tags)
def get_response(self):
"""Get and process response from IRC"""
try:
@ -401,6 +381,33 @@ class IRC:
self.poll = self.poll
self.pollcount = self.pollcount
def tts(self, msg, tags):
""" !tts command
Check if message is valid and send it to queue
:param str msg: The IRC message triggering the command
:param dict tags: The message metadata
"""
user = tags['user']
if IRC.check_tts_disabled(self, user):
logging.info('TTS is disabled')
elif IRC.check_msg_too_long(self, msg, user):
logging.info('TTS message is too long')
elif IRC.check_user_denied(self, user):
logging.info('User is not allowed to use TTS')
elif IRC.check_subonly(self, tags):
logging.info('TTS is sub-only')
elif IRC.check_modonly(self, tags):
logging.info('TTS is mod-only')
elif IRC.check_whitelist(self, user):
logging.info('User is not on whitelist')
else:
logging.info('Sending TTS message to raw_queue')
IRC.send_tts_msg(self, msg, tags)
def quickvote(self, msg):
""" !quickvote command
@ -659,7 +666,7 @@ class HTTPserv(BaseHTTPRequestHandler):
self.end_headers()
self.wfile.write(bytes("Internal Server error\n", "utf-8"))
elif self.path.startswith('/token') and conf['IRC_OAUTH_TOKEN'] == "Invalid":
elif self.path.startswith('/token'):
data = {}
data['client_id'] = "ebo548vs6tq54c9zlrgin2yfzzlrrs"
data['response_type'] = "token"
@ -784,20 +791,6 @@ def load_config():
return conf
conf = {}
tts_done = []
msg_queue_raw = []
msg_queue = {}
logging.basicConfig(level=logging.DEBUG, format='%(asctime)s %(module)s %(threadName)s %(levelname)s: %(message)s')
sys.tracebacklimit = 0
if sys.argv[1:]:
if sys.argv[1] == "--version":
print('Simple TTS Bot')
print('Version 1.2.3')
sys.exit(1)
def send_tts_queue():
""" Send messages to TTS """
@ -814,12 +807,49 @@ def send_tts_queue():
else:
logging.debug('Msg is already in queue')
def get_url(path=False):
""" Generate a valid URL from config values """
if conf['HTTP_BIND'] == "0.0.0.0":
url = "localhost"
else:
url = conf['HTTP_BIND']
url = "http://"+str(url)+":"+str(conf['HTTP_PORT'])+"/"
if path:
url = url+str(path)
return url
def check_oauth_token(conf):
""" Check for valid authentication via Twitch API """
logging.debug('Checking OAuth Token')
try:
url = 'https://id.twitch.tv/oauth2/validate'
oauth = "OAuth %s" % conf['IRC_OAUTH_TOKEN'].replace('oauth:','')
request = urllib.request.Request(url)
request.add_header('Authorization', oauth)
urllib.request.urlopen(request)
except HTTPError:
conf['IRC_OAUTH_TOKEN'] == "Invalid"
logging.fatal('Twitch rejected your OAuth Token. Please check and generate a new one.')
logging.info('Please open http://%s:%s/token to generate your OAuth-Token.', conf['HTTP_BIND'], conf['HTTP_PORT'])
url = get_url("token")
webbrowser.open_new_tab(url)
logging.info('Please complete the OAuth process and add the token into your "config.yml" within the next 5 minutes.')
time.sleep(300)
conf = load_config()
check_oauth_token()
logging.info('OAuth Token is valid \o/')
return
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':
@ -832,15 +862,7 @@ def main():
http_thread = Thread(target=http_serve_forever, daemon=True, args=(httpd, ))
http_thread.start()
if conf['IRC_OAUTH_TOKEN'] == "Invalid":
logging.error('No OAuth Token, skipping start of IRC bot.')
logging.error('Please open http://%s:%s/token to generate your OAuth-Token.', conf['HTTP_BIND'], conf['HTTP_PORT'])
url = 'http://'+str(conf['HTTP_BIND'])+':'+str(conf['HTTP_PORT'])+'/token'
webbrowser.open_new_tab(url)
logging.info('Please complete the OAuth process within the next 15 minutes.')
time.sleep(900)
sys.exit(250)
check_oauth_token(conf)
logging.info("Starting IRC bot")
irc = IRC()
@ -848,8 +870,8 @@ def main():
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'])
url = 'http://'+str(conf['HTTP_BIND'])+':'+str(conf['HTTP_PORT'])
url = get_url()
logging.info("Please open your browser and visit: %s", url)
webbrowser.open_new_tab(url)
while True:
@ -879,4 +901,19 @@ def main():
sys.exit()
if __name__ == "__main__":
conf = {}
tts_done = []
msg_queue_raw = []
msg_queue = {}
logging.basicConfig(level=logging.DEBUG, format='%(asctime)s %(module)s %(threadName)s %(levelname)s: %(message)s')
sys.tracebacklimit = 3
if sys.argv[1:]:
if sys.argv[1] == "--version":
print('Simple TTS Bot')
print('Version 1.2.4')
sys.exit(1)
main()