import html import json import random import requests from requests.structures import CaseInsensitiveDict from ameliabot.logger import logging from ameliabot.quote import Quote from flask import request bot_version = "0.0.1" # Get basic owncast bot config with open("config/owncast.json", "r") as f: config = json.load(f) logging.info("Configuration loaded") with open("commands.json", "r") as f: commands = json.load(f) logging.info("Commands loaded") with open("alias.json", "r") as f: aliases = json.load(f) logging.info("Aliases loaded") quote = Quote( database=config["quote_db_name"], user=config["quote_db_user"], password=config["quote_db_pass"], ) # prepare the header for the bot posts headers = CaseInsensitiveDict() headers["Content-Type"] = "application/json" headers["Authorization"] = "Bearer {}".format(config["access_token"]) def parse_webhook(): hook_type = request.json["type"] data = request.json["eventData"] if "CHAT" in hook_type: data = process_chat(data) if data: logging.debug("Processed: {}".format(data)) data = '{"body": "%s"}' % data resp = requests.post( config["owncast_server"], headers=headers, data=data.encode('utf-8')) if resp.status_code == 200: logging.debug("RESP: {}".format(data)) 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 q = quote.get(num) return html.escape(q) def process_chat(data): timestamp = "{} {}".format( data["timestamp"][0:10], data["timestamp"][11:19]) sender = data["user"]["displayName"] text = data["body"] command_reply = get_command(text) logging.info("{} <{}> {}".format(timestamp, sender, text)) if command_reply: try: first_parameter = text.split(" ")[1] except IndexError: first_parameter = None # Move this dictionary to a configuration file or something # This is so bad, it runs everything every command :pikajoy: placeholders = { "sender": sender, "streamer": config["streamer_name"], "target": first_parameter, "random": str(random.randrange(1, 100, 1)), "commands": ", ".join(list(commands.keys())), "aliases": ", ".join(list(aliases.keys())), "bot_version": bot_version, "quote": get_quote(first_parameter), "quote_parameters": "Not implemented yet", } for placeholder in placeholders: findme = "{%s}" % placeholder if findme in command_reply: if placeholders[placeholder] == "": command_reply = "%s required" % placeholder.upper() command_reply = command_reply.replace( findme, placeholders[placeholder]) return command_reply