130 lines
3.8 KiB
Python
130 lines
3.8 KiB
Python
import importlib
|
|
import os
|
|
import random
|
|
import requests
|
|
from requests.structures import CaseInsensitiveDict
|
|
import sys
|
|
from core import __version__, config, aliases, commands
|
|
from core.logger import logging
|
|
from flask import request
|
|
|
|
|
|
# This is gross and needs to be replaced
|
|
plugin_path = "%s/../plugins" % os.path.dirname(os.path.realpath(__file__))
|
|
sys.path.append(plugin_path)
|
|
modules = os.listdir(plugin_path)
|
|
|
|
for module in modules:
|
|
if module.endswith(".py") and not module.startswith("__"):
|
|
globals()[module[:-3]] = importlib.import_module(module[:-3])
|
|
|
|
# prepare the header for the bot posts
|
|
headers = CaseInsensitiveDict()
|
|
headers["Content-Type"] = "application/json"
|
|
headers["Authorization"] = "Bearer {}".format(config["owncast"]["AccessToken"])
|
|
|
|
post_url = "%s/api/integrations/chat/send" % config["owncast"]["ServerURL"]
|
|
|
|
|
|
def parse_webhook():
|
|
hook_type = request.json["type"]
|
|
data = request.json["eventData"]
|
|
|
|
if "CHAT" in hook_type:
|
|
data = process_chat(data)
|
|
|
|
if data:
|
|
send_message(data)
|
|
|
|
|
|
def send_message(data):
|
|
data = '{"body": "%s"}' % data
|
|
resp = requests.post(
|
|
post_url,
|
|
headers=headers,
|
|
data=data.encode('utf-8'))
|
|
|
|
if resp.status_code == 200:
|
|
logging.debug("RESP: {}".format(resp.__dict__))
|
|
else:
|
|
logging.error("Status code {} returned".format(
|
|
str(resp.status_code)))
|
|
|
|
|
|
def get_command(text):
|
|
first_word = text.split()[0].lower()
|
|
|
|
if first_word in commands:
|
|
return commands[first_word]
|
|
elif first_word in aliases:
|
|
return commands[first_word]
|
|
|
|
|
|
def get_quote(num):
|
|
try:
|
|
num = int(num)
|
|
except TypeError:
|
|
pass
|
|
except ValueError:
|
|
pass
|
|
return quote.get(num) # NOQA should work even if linter complains
|
|
|
|
|
|
def process_placeholders(sender, param, text):
|
|
# Move this dictionary to a configuration file or something
|
|
# This is so bad, it runs everything every command :pikajoy:
|
|
logging.debug("Processing placeholders")
|
|
placeholders = {
|
|
"sender": sender,
|
|
"botowner": config["DEFAULT"]["BotOwner"],
|
|
"target": param,
|
|
"random": str(random.randrange(1, 100, 1)),
|
|
"commands": ", ".join(list(commands.keys())),
|
|
"aliases": ", ".join(list(aliases.keys())),
|
|
"bot_version": __version__,
|
|
"quote": get_quote(param),
|
|
"quote_parameters": "Not implemented yet",
|
|
}
|
|
|
|
for placeholder in placeholders:
|
|
findme = "{%s}" % placeholder
|
|
if findme in text:
|
|
logging.debug("Found {}".format(findme))
|
|
if placeholders[placeholder] == "":
|
|
logging.debug("Placeholder empty")
|
|
text = "%s required" % placeholder.upper()
|
|
try:
|
|
logging.debug(
|
|
"Replacing {} with {}".format(
|
|
findme, placeholders[placeholder]))
|
|
text = text.replace(
|
|
findme, placeholders[placeholder])
|
|
except TypeError:
|
|
text = "Target not specified."
|
|
|
|
return text
|
|
|
|
|
|
def process_chat(data):
|
|
sender = data["user"]["displayName"]
|
|
text = data["body"][3:-4]
|
|
command_reply = get_command(text)
|
|
logging.info("<{}> {}".format(sender, text))
|
|
|
|
if command_reply:
|
|
logging.debug("Command found")
|
|
if command_reply.startswith("{plugins."):
|
|
plugin = command_reply[1:-1].split(".")
|
|
logging.debug("Attempting to run plugin: %s" % plugin[1])
|
|
reply = globals()[plugin[1]].run()
|
|
return reply
|
|
try:
|
|
first_parameter = text.split(" ")[1]
|
|
except IndexError:
|
|
first_parameter = None
|
|
|
|
command_reply = process_placeholders(
|
|
sender, first_parameter, command_reply)
|
|
logging.debug("Reply: {}".format(command_reply))
|
|
return command_reply
|