137 lines
4.6 KiB
Python
137 lines
4.6 KiB
Python
import atexit
|
|
import json
|
|
import logging
|
|
from pathlib import Path
|
|
from typing import Union, Dict, Callable, Tuple
|
|
|
|
import discord
|
|
from discord.ext import tasks
|
|
|
|
from .Game import Game
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Functions used to convert between configuration versions, in a dictionary of: old_version -> converter
|
|
# Each converter takes the old configuration and returns the converted configuration and the new version
|
|
config_version_converters: Dict[Union[None, str], Callable[[Dict], Tuple[Dict, str]]] = {
|
|
}
|
|
|
|
|
|
class GamesFile:
|
|
"""
|
|
Wrapper around the configuration file for all games of all guilds
|
|
"""
|
|
version = "1.0"
|
|
|
|
@staticmethod
|
|
def empty_config() -> Dict:
|
|
"""
|
|
:return: An empty configuration, used to initialise the configuration file
|
|
"""
|
|
return {
|
|
"__version__": GamesFile.version,
|
|
}
|
|
|
|
def __init__(self, games_file_path: Union[str, Path]):
|
|
self.file_obj = None # File descriptor to the on-disk config file
|
|
self.config = None # In-memory configuration
|
|
|
|
# Loading configuration from disk
|
|
logger.debug("Loading games file")
|
|
games_file_path = Path(games_file_path).absolute()
|
|
if games_file_path.exists():
|
|
if not games_file_path.is_file():
|
|
raise ValueError(f"Games file {games_file_path} exists but is not a regular file")
|
|
|
|
self.file_obj = games_file_path.open("r+")
|
|
|
|
if games_file_path.stat().st_size == 0: # Config file is empty
|
|
logger.warning(f"games file {games_file_path} is empty, initialising an empty configuration")
|
|
self.config = self.empty_config()
|
|
self.save_to_file() # Initialise the file
|
|
else: # File is not empty
|
|
has_been_converted = self.reload_from_disk()
|
|
if has_been_converted:
|
|
self.save_to_file()
|
|
else: # File does not exist
|
|
self.file_obj = games_file_path.open("w+")
|
|
self.config = self.empty_config()
|
|
self.save_to_file() # Initialise the file
|
|
|
|
# Verifying that attributes have been initialised properly
|
|
assert self.file_obj is not None and self.config is not None
|
|
logger.debug("Games file successfully initialised")
|
|
self.save_needed = False
|
|
self.save_if_needed.start()
|
|
|
|
atexit.register(self.save_to_file)
|
|
|
|
def reload_from_disk(self) -> bool:
|
|
"""
|
|
Reload the configuration from disk
|
|
|
|
:return: Whether the configuration had to be converted to a more recent version because the one on disk was an earlier version
|
|
:except json.JSONDecodeError: if file is not valid json
|
|
:except ValueError: if the version of the configuration in the file is not known or can not be converted to a more recent one
|
|
"""
|
|
logger.debug("Loading games file from disk")
|
|
self.file_obj.seek(0) # Moving to beginning of file
|
|
try:
|
|
self.config = json.load(self.file_obj)
|
|
except json.JSONDecodeError as e:
|
|
logger.critical(f"JSON Error when parsing games file: {e}")
|
|
raise e
|
|
# Checking configuration version and converting if needed
|
|
config_version = self.config["__version__"]
|
|
has_been_converted = False
|
|
if config_version != self.version:
|
|
logger.info(f"Games file is an older version, converting (file version: {config_version}, current: {self.version})...")
|
|
|
|
# Performing the conversion
|
|
while config_version != self.version:
|
|
if config_version in config_version_converters:
|
|
logger.debug(f"Converting from {config_version}")
|
|
self.config, config_version = config_version_converters[config_version](self.config)
|
|
logger.debug(f"Converted to {config_version}")
|
|
else:
|
|
logger.critical(f"Impossible to find converter to convert from {config_version}")
|
|
raise ValueError(f"Configuration loading: impossible to convert to current version")
|
|
has_been_converted = True
|
|
|
|
assert self.config["__version__"] == self.version
|
|
|
|
logger.debug("Loaded games file from disk")
|
|
return has_been_converted
|
|
|
|
def save_to_file(self, indent = 2) -> None:
|
|
"""
|
|
Save the configuration to disk
|
|
:param indent: Indentation to use for pretty-printing json
|
|
"""
|
|
logger.debug("Writing games file to disk")
|
|
self.file_obj.seek(0)
|
|
self.file_obj.truncate()
|
|
json.dump(self.config, self.file_obj, indent = indent)
|
|
self.file_obj.flush()
|
|
logger.debug("Written games file to disk")
|
|
|
|
@tasks.loop(seconds = 10)
|
|
async def save_if_needed(self):
|
|
if self.save_needed:
|
|
self.save_to_file()
|
|
self.save_needed = False
|
|
|
|
def ask_for_save(self):
|
|
logger.debug("Asked to save configuration")
|
|
self.save_needed = True
|
|
|
|
def __getitem__(self, guild: discord.Guild) -> Game:
|
|
"""
|
|
Get the game information for one guild
|
|
"""
|
|
guild_id_str = str(guild.id)
|
|
if guild_id_str not in self.config:
|
|
self.config[guild_id_str] = Game.new_dict()
|
|
self.save_to_file()
|
|
return Game(self.config[guild_id_str], self.ask_for_save, guild)
|