139 lines
4.0 KiB
Python
139 lines
4.0 KiB
Python
"""
|
|
Spiffo
|
|
A Discord Bot for integration with a Project Zomboid game server.
|
|
"""
|
|
|
|
import json
|
|
import atexit
|
|
import asyncio
|
|
import logging
|
|
import logging.config
|
|
from os import getenv
|
|
from pathlib import Path
|
|
|
|
from discord import Intents, TextChannel
|
|
from discord.ext import commands
|
|
from dotenv import load_dotenv
|
|
from tortoise import Tortoise
|
|
|
|
load_dotenv(override=True)
|
|
log = logging.getLogger(__name__)
|
|
|
|
BASE_DIR = Path(__file__).resolve().parent
|
|
DATA_DIR = getenv("SPIFFO__DATA_FOLDER_PATH")
|
|
TORTOISE_ORM = {
|
|
"connections": { "default": f"sqlite://{str(DATA_DIR)}/db.sqlite" },
|
|
"apps": {
|
|
"models": {
|
|
"models": ["utils.models"],
|
|
"default_connection": "default"
|
|
}
|
|
}
|
|
}
|
|
|
|
|
|
class DiscordBot(commands.Bot):
|
|
"""
|
|
Represents a Discord bot.
|
|
Contains controls to interact with the bot via the Discord API.
|
|
"""
|
|
def __init__(self, debug_mode: bool):
|
|
super().__init__(command_prefix="-", intents=Intents.all())
|
|
self.debug_mode = debug_mode
|
|
self.in_game_channel_id = int(getenv("SPIFFO__DISCORD_CHANNEL_ID"))
|
|
self.steam_api_key = getenv("SPIFFO__STEAM_API_KEY")
|
|
self.rcon_details = {
|
|
"host": getenv("SPIFFO__RCON_HOST"),
|
|
"port": getenv("SPIFFO__RCON_PORT"),
|
|
"passwd": getenv("SPIFFO__RCON_PASSWORD")
|
|
}
|
|
|
|
async def on_ready(self):
|
|
"""
|
|
Execute initial operations that require the bot to be ready.
|
|
Ideally should not be manually called, this is handled by discord.py
|
|
"""
|
|
|
|
# Sync app commands
|
|
await self.wait_until_ready()
|
|
await self.tree.sync()
|
|
log.info("Discord Bot is ready")
|
|
|
|
async def close(self):
|
|
await Tortoise.close_connections()
|
|
await super().close();
|
|
log.info("Shutdown successfully and safely")
|
|
|
|
async def load_cogs(self):
|
|
"""
|
|
Load any extensions found in the cogs dictionary.
|
|
"""
|
|
|
|
for path in (BASE_DIR / "cogs").iterdir():
|
|
if path.suffix == ".py":
|
|
await self.load_extension(f"cogs.{path.stem}")
|
|
|
|
async def get_ingame_channel(self) -> TextChannel:
|
|
"""
|
|
"""
|
|
channel = self.get_channel(self.in_game_channel_id)
|
|
return channel or await self.fetch_channel(self.in_game_channel_id)
|
|
|
|
|
|
def get_bot_token() -> str:
|
|
"""
|
|
Retrieve the access token of the bot from the environment.
|
|
Raises `ValueError` if token is missing or blank.
|
|
"""
|
|
bot_token = getenv("SPIFFO__BOT_TOKEN")
|
|
if not bot_token:
|
|
raise ValueError("'SPIFFO__BOT_TOKEN' environment variable is missing or empty.")
|
|
|
|
return bot_token
|
|
|
|
def setup_logging_config(debug_mode: bool):
|
|
"""
|
|
Loads the logging configuration and creates an asynchronous queue handler.
|
|
"""
|
|
log_config_path = BASE_DIR / "logging.json"
|
|
if not log_config_path.exists():
|
|
raise FileNotFoundError(log_config_path)
|
|
|
|
with open(log_config_path, "r", encoding="utf-8") as file:
|
|
log_config = json.load(file)
|
|
|
|
# Ensure the logging directory exists
|
|
(BASE_DIR / "logs").mkdir(exist_ok=True)
|
|
|
|
# update the log level dependent on debug mode
|
|
log_config["loggers"]["root"]["level"] = "DEBUG" if debug_mode else "INFO"
|
|
|
|
# Load the config
|
|
logging.config.dictConfig(log_config) # This 'logging.config' path is jank.
|
|
|
|
# Register the queue handler for asynchronous logging
|
|
queue_handler = logging.getHandlerByName("queue_handler")
|
|
if queue_handler is not None:
|
|
queue_handler.listener.start()
|
|
atexit.register(queue_handler.listener.stop)
|
|
|
|
async def main():
|
|
"""
|
|
The entrypoint function, initialises the application.
|
|
"""
|
|
debug_mode = getenv("SPIFFO__DEBUG").lower() == "true"
|
|
setup_logging_config(debug_mode)
|
|
bot_token = get_bot_token()
|
|
|
|
# Open database connection
|
|
await Tortoise.init(config=TORTOISE_ORM)
|
|
await Tortoise.generate_schemas()
|
|
|
|
async with DiscordBot(debug_mode) as bot:
|
|
await bot.load_cogs()
|
|
await bot.start(bot_token, reconnect=True)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|