101 lines
2.8 KiB
Python
101 lines
2.8 KiB
Python
|
from helpers.TwitchIrc import TwitchIrc
|
||
|
|
from utils.MessageQueue import MessageQueue
|
||
|
|
|
||
|
|
class ChatBot:
|
||
|
|
|
||
|
|
twitch_irc = TwitchIrc()
|
||
|
|
queue = MessageQueue()
|
||
|
|
|
||
|
|
services = []
|
||
|
|
users = {}
|
||
|
|
|
||
|
|
msg_help = ""
|
||
|
|
msg_unknown_cmd = ""
|
||
|
|
msg_hi = ""
|
||
|
|
msg_bye = ""
|
||
|
|
|
||
|
|
running = False
|
||
|
|
|
||
|
|
# Settings -----
|
||
|
|
|
||
|
|
def set_twitch_irc_settings(self, nickname, password, channel):
|
||
|
|
self.twitch_irc.join(nickname, password, channel)
|
||
|
|
|
||
|
|
def add_service(self, service):
|
||
|
|
service.set_message_queue(self.queue)
|
||
|
|
self.services.append(service)
|
||
|
|
|
||
|
|
def clear_services(self):
|
||
|
|
self.services = []
|
||
|
|
|
||
|
|
def set_help_message(self, message):
|
||
|
|
self.msg_help = message
|
||
|
|
|
||
|
|
def set_unknown_command_message(self, message):
|
||
|
|
self.msg_unknown_cmd = message
|
||
|
|
|
||
|
|
def set_welcome_message(self, message):
|
||
|
|
self.msg_hi = message
|
||
|
|
|
||
|
|
def set_farewell_message(self, message):
|
||
|
|
self.msg_bye = message
|
||
|
|
|
||
|
|
# Commands -----
|
||
|
|
def parse_response(self, response):
|
||
|
|
command = response["message"].split(" ", 1)[0][1:]
|
||
|
|
if command == "help":
|
||
|
|
self.display_help()
|
||
|
|
return
|
||
|
|
elif command == "stop":
|
||
|
|
self.stop_command(response["tags"]["badges"])
|
||
|
|
return
|
||
|
|
result = None
|
||
|
|
for service in self.services:
|
||
|
|
if not service.knows(command):
|
||
|
|
continue
|
||
|
|
result = service.eval(command, response, self.users)
|
||
|
|
return
|
||
|
|
self.queue.append(f"{self.msg_unknown_cmd}, @{response['username']}")
|
||
|
|
|
||
|
|
def display_help(self):
|
||
|
|
self.queue.append(f"{self.msg_help}{self.list_commands()}")
|
||
|
|
|
||
|
|
def list_commands(self):
|
||
|
|
commands = ["help"]
|
||
|
|
for service in self.services:
|
||
|
|
commands.extend(service.list_commands())
|
||
|
|
return f" !{' !'.join(commands)}"
|
||
|
|
|
||
|
|
def stop_command(self, badges):
|
||
|
|
for badge in badges:
|
||
|
|
if badge["name"] == "broadcaster":
|
||
|
|
self.stop()
|
||
|
|
return
|
||
|
|
|
||
|
|
# Listening -----
|
||
|
|
|
||
|
|
def start(self):
|
||
|
|
for service in self.services:
|
||
|
|
service.start()
|
||
|
|
self.twitch_irc.request_tags()
|
||
|
|
self.twitch_irc.start()
|
||
|
|
if self.msg_hi != "":
|
||
|
|
self.queue.append(self.msg_hi)
|
||
|
|
self.running = True
|
||
|
|
while self.running:
|
||
|
|
response = self.twitch_irc.receive()
|
||
|
|
if response and response["message"][:1] == "!":
|
||
|
|
self.parse_response(response)
|
||
|
|
while self.queue.pending():
|
||
|
|
self.twitch_irc.send(self.queue.shift())
|
||
|
|
|
||
|
|
def stop(self):
|
||
|
|
if self.msg_bye != "":
|
||
|
|
self.queue.append(self.msg_bye)
|
||
|
|
while self.queue.pending():
|
||
|
|
continue
|
||
|
|
self.running = False
|
||
|
|
self.twitch_irc.stop()
|
||
|
|
for service in self.services:
|
||
|
|
service.stop()
|