Init

dom5game.py

@@ -0,0 +1,59 @@
+import json
+import pandas as pd
+
+
+class Dom5game:
+    def __init__(self, name, channelId, members, turn, players):
+        self.name = name
+        self.channelId = channelId
+        self.members = members
+        self.htmlPath = "games/" + name + "/turnstats.html"
+        self.turn = turn
+        self.players = players
+        self.jsonPath = "games/" + self.name + "/" + self.name + ".json"
+
+    def to_json(self):
+        with open(self.jsonPath, "w") as file:
+            json.dump(self.__dict__, file)
+
+    @staticmethod
+    def load_json(filepath):
+        with open(filepath, "r") as file:
+            jsondata = json.load(file)
+        return Dom5game(
+            jsondata["name"],
+            jsondata["channelId"],
+            jsondata["members"],
+            jsondata["turn"],
+            jsondata["players"],
+        )
+
+    @staticmethod
+    def get_game_by_name(name, list):
+        for game in list:
+            if game.name == name:
+                return game
+        return None
+
+    def get_turn(self):
+        df = pd.read_html(self.htmlPath)[0]
+        return df.iloc[0].to_string().split(" ")[-1]
+
+    def update_turn(self):
+        self.turn = self.get_turn()
+
+    @staticmethod
+    def html_to_dict(htmlPath):
+        df = pd.read_html(htmlPath)[0]
+        return dict(zip(df.iloc[1:, 0], df.iloc[1:, 1]))
+
+    def update_players(self):
+        self.players = self.html_to_dict(self.htmlPath)
+
+    def get_new_turns(self):
+        newdict = self.html_to_dict(self.htmlPath)
+        return {
+            key: newdict[key]
+            for key in self.players.keys() & newdict
+            if self.players[key] != newdict[key]
+        }

main.py

@@ -0,0 +1,96 @@
+import discord
+import os
+from discord.ext import tasks, commands
+from dotenv import load_dotenv
+from dom5game import Dom5game
+from servermanager import create_server
+
+intents = discord.Intents.default()
+bot = commands.Bot(command_prefix="!", intents=intents)
+load_dotenv()
+TOKEN = os.getenv("TOKEN")
+
+bot.tracked_games = []
+
+
+def reload_games():
+    bot.tracked_games = []
+    for _, _, files in os.walk("games"):
+        for name in files:
+            if name.endswith("json"):
+                bot.tracked_games.append(
+                    Dom5game.load_json("games/" + name[:-5] + "/" + name)
+                )
+
+
+create_server("Amogus", 7777, 831955362646851617)
+reload_games()
+
+
+@bot.event
+async def on_ready():
+    print(f"Logged in as {bot.user}")
+    await bot.load_extension("cogs.slash_commands")
+    # await bot.tree.sync()
+    reload_games()
+    task_loop.start()
+
+
+@tasks.loop(seconds=2)
+async def task_loop():
+    await bot.change_presence(
+        activity=discord.Activity(
+            type=discord.ActivityType.watching,
+            name=str(len(bot.tracked_games)) + " Servers.",
+        )
+    )
+    for game in bot.tracked_games:
+        channel = bot.get_channel(game.channelId)
+        # handle turn ticks
+        if game.get_turn() != game.turn and game.get_turn().isdigit():
+            game.update_turn()
+            pingstr = ""
+            if game.members:
+                pingstr += "-# "
+            for member in game.members:
+                pingstr += "<@" + str(member) + "> "
+            await channel.send(
+                'Game **"'
+                + game.name
+                + '"** has advanced to turn '
+                + game.turn
+                + " \n"
+                + pingstr
+            )
+            game.to_json()
+
+        # handle uploaded turns
+        turns = game.get_new_turns()
+        if turns != {}:
+            for key in turns:
+                value = turns[key]
+                if value == "Turn played":
+                    await channel.send(
+                        'Game **"'
+                        + game.name
+                        + '"**: '
+                        + key
+                        + " has played their turn."
+                    )
+                if value == "Eliminated":
+                    await channel.send(
+                        'Game **"' + game.name + '"**: ' + key + " has been eliminated."
+                    )
+                if value == "Turn unfinished":
+                    await channel.send(
+                        'Game **"'
+                        + game.name
+                        + '"**: '
+                        + key
+                        + " has marked their turn as unfinished."
+                    )
+                game.update_players()
+                game.to_json()
+
+
+bot.run(TOKEN)

servermanager.py

@@ -0,0 +1,46 @@
+import subprocess
+from dotenv import load_dotenv
+import os
+from dom5game import Dom5game
+
+load_dotenv()
+SERVER_PATH = os.getenv("SERVERPATH")
+
+
+def create_server(name, port, channel):
+    if is_port_in_use(port):
+        return "ERROR_PORT_IN_USE"
+    # os.mkdir("games/" + name)
+    # print(SERVER_PATH)
+    # print(server_command_builder(name, port, channel))
+    try:
+        subprocess.Popen(
+            server_command_builder(name, port, channel),
+            stdin=None,
+            stdout=None,
+            stderr=None,
+        )
+    except Exception as e:
+        return "EXCEPTION OCCURED: " + str(e)
+
+
+def server_command_builder(name, port, channel):
+    command = [
+        SERVER_PATH,
+        "-TS",
+        name,
+        "--port",
+        str(port),
+        "--statuspage",
+        os.getcwd() + "/games/" + name + "/turnstats.html",
+    ]
+    game = Dom5game(name, channel, [], 0, {})
+    game.to_json()
+    return command
+
+
+def is_port_in_use(port: int) -> bool:
+    import socket
+
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+        return s.connect_ex(("localhost", port)) == 0