ameliabot/core/owncast.py

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"]
    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