76 lines
No EOL
2.8 KiB
Python
76 lines
No EOL
2.8 KiB
Python
"""Database management for the LevelUp cog."""
|
|
|
|
import typing as t
|
|
import msgpack
|
|
from pathlib import Path
|
|
import logging
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
class DB:
|
|
def __init__(self):
|
|
self.configs: t.Dict[int, GuildSettings] = {}
|
|
self.cache_seconds = 300
|
|
self.render_gifs = True
|
|
self.use_skia = True
|
|
self.force_embeds = False
|
|
self.ignore_bots = True
|
|
self.auto_cleanup = False
|
|
self.ignored_guilds: t.List[int] = []
|
|
self.external_api_url = None
|
|
self.internal_api_port = None
|
|
|
|
def save(self):
|
|
"""Save the database to disk"""
|
|
data = {
|
|
"cache_seconds": self.cache_seconds,
|
|
"render_gifs": self.render_gifs,
|
|
"use_skia": self.use_skia,
|
|
"force_embeds": self.force_embeds,
|
|
"ignore_bots": self.ignore_bots,
|
|
"auto_cleanup": self.auto_cleanup,
|
|
"ignored_guilds": self.ignored_guilds,
|
|
"external_api_url": self.external_api_url,
|
|
"internal_api_port": self.internal_api_port,
|
|
"configs": {},
|
|
}
|
|
for guild_id, conf in self.configs.items():
|
|
data["configs"][guild_id] = conf.to_json()
|
|
with open(DB_PATH, "wb") as f:
|
|
msgpack.dump(data, f)
|
|
|
|
def load(self):
|
|
"""Load the database from disk"""
|
|
if not DB_PATH.exists():
|
|
self.save()
|
|
return
|
|
try:
|
|
with open(DB_PATH, "rb") as f:
|
|
data = msgpack.load(f)
|
|
self.cache_seconds = data.get("cache_seconds", 300)
|
|
self.render_gifs = data.get("render_gifs", True)
|
|
self.use_skia = data.get("use_skia", True)
|
|
self.force_embeds = data.get("force_embeds", False)
|
|
self.ignore_bots = data.get("ignore_bots", True)
|
|
self.auto_cleanup = data.get("auto_cleanup", False)
|
|
self.ignored_guilds = data.get("ignored_guilds", [])
|
|
self.external_api_url = data.get("external_api_url")
|
|
self.internal_api_port = data.get("internal_api_port")
|
|
for guild_id, conf_data in data.get("configs", {}).items():
|
|
self.configs[guild_id] = GuildSettings.from_json(conf_data)
|
|
except Exception as e:
|
|
log.error("Error loading database", exc_info=e)
|
|
if DB_PATH.exists():
|
|
backup_path = DB_PATH.with_suffix(".bak")
|
|
DB_PATH.rename(backup_path)
|
|
log.info(f"Backed up corrupt database to {backup_path}")
|
|
self.save()
|
|
|
|
def set_use_skia(self, value: bool):
|
|
"""Set whether to use Skia for image generation."""
|
|
self.use_skia = value
|
|
self.save()
|
|
|
|
def get_use_skia(self) -> bool:
|
|
"""Get whether to use Skia for image generation."""
|
|
return self.use_skia |