114 lines
3.1 KiB
Python
114 lines
3.1 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
|
|
from discord.ext import commands
|
|
from dotenv import load_dotenv
|
|
|
|
load_dotenv(override=True)
|
|
log = logging.getLogger(__name__)
|
|
|
|
BASE_DIR = Path(__file__).resolve().parent
|
|
|
|
|
|
class DiscordBot(commands.Bot):
|
|
"""
|
|
Represents a Discord bot.
|
|
Contains controls to interact with the bot via the Discord API.
|
|
"""
|
|
rcon_details: dict
|
|
|
|
def __init__(self):
|
|
super().__init__(command_prefix="-", intents=Intents.all())
|
|
self.rcon_details = {
|
|
"host": getenv("SPIFFO__RCON_HOST"),
|
|
"port": getenv("SPIFFO__RCON_PORT"),
|
|
"passwd": getenv("SPIFFO__RCON_PASSWORD")
|
|
}
|
|
|
|
async def sync_app_commands(self):
|
|
"""
|
|
Sync application commands between Discord and the bot.
|
|
"""
|
|
await self.wait_until_ready()
|
|
await self.tree.sync()
|
|
log.info("Application commands successfully synced")
|
|
|
|
async def on_ready(self):
|
|
"""
|
|
Execute init operations that require the bot to be ready.
|
|
Ideally should not be manually called, this is handled by discord.py
|
|
"""
|
|
|
|
await self.sync_app_commands()
|
|
log.info("Discord Bot is ready")
|
|
|
|
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}")
|
|
|
|
|
|
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():
|
|
"""
|
|
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)
|
|
|
|
# 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.
|
|
"""
|
|
setup_logging_config()
|
|
bot_token = get_bot_token()
|
|
|
|
async with DiscordBot() as bot:
|
|
await bot.load_cogs()
|
|
await bot.start(bot_token, reconnect=True)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|