from pathlib import Path from typing import Union, Dict, List import logging import json import atexit import discord import utils logger = logging.getLogger(__name__) class ChannelsConfigFile: """ Wrapper for the channels configuration file, telling which channels to watch and which channels were created by the bot. """ def __init__(self, config_file_path: Union[str, Path]): self.config_file = None # File descriptor to the on-disk config file self.config = None # In-memory configuration # Loading configuration from disk logger.debug("Loading channels configuration file") config_file_path = Path(config_file_path).absolute() if config_file_path.exists(): if not config_file_path.is_file(): raise ValueError(f"Config file {config_file_path} exists but is not a regular file") self.config_file = config_file_path.open("r+") if config_file_path.stat().st_size == 0: # Config file is empty logger.warning(f"Config file {config_file_path} is empty, using empty config") self.config = {} self.save_to_file() # So that the file is proper json else: # Config file is not empty self.reload_from_disk() else: # Config file does not exist self.config_file = config_file_path.open("w+") self.config = {} self.save_to_file() # So that the file is proper json # Verifying that attributes have been initialised properly assert self.config_file is not None and self.config is not None logger.debug("Channels configuration file successfully initialised") atexit.register(self.save_to_file) def reload_from_disk(self) -> None: """ Reload the configuration from the on-disk file """ logger.debug("Loading channels configuration file from disk") self.config_file.seek(0) # Moving to beginning of file try: self.config = json.load(self.config_file) except json.JSONDecodeError as e: logger.critical(f"JSON Error when parsing channels config file: {e}") raise e logger.debug("Loaded channels configuration file from disk") def save_to_file(self, indent = 2) -> None: """ Save the configuration to the on-fisk file :param indent: Indentation to use for pretty-printing json """ logger.debug("Writing channels configuration file to disk") self.config_file.seek(0) self.config_file.truncate() json.dump(self.config, self.config_file, indent = indent) self.config_file.flush() logger.debug("Written channels configuration file to disk") def _get_guild_configuration(self, guild: discord.Guild) -> Dict: str_guild_id = str(guild.id) if str_guild_id not in self.config: self.config[str_guild_id] = { "_name": guild.name, "watched_channels": [], "created_channels": [], } return self.config[str_guild_id] def get_watched_channels_ids(self, guild: discord.Guild) -> List[int]: """ :return: The list of IDs of watched channels for the given guild. Beware that the returned list is not a copy, it should not be modified. """ guild_configuration = self._get_guild_configuration(guild) if "watched_channels" not in guild_configuration: guild_configuration["watched_channels"] = [] return guild_configuration["watched_channels"] def add_channel_to_watched(self, guild: discord.Guild, channel: Union[discord.VoiceChannel, discord.Object]) -> bool: """ Add a voice channel to the list of channels that the bot should watch :return: True if the channel has been added, False if it was already present """ watched_channels = self.get_watched_channels_ids(guild) if channel.id in watched_channels: return False else: watched_channels.append(channel.id) self.save_to_file() return True def remove_channel_from_watched(self, guild: discord.Guild, channel: Union[discord.VoiceChannel, discord.Object]) -> bool: """ Attempt to remove a voice channel from the list of channels that the bot should watch :return: True if the channel was in the list and has been removed, False if it was not in the list """ watched_channels = self.get_watched_channels_ids(guild) size_before = len(watched_channels) utils.list_remove_all_occurrences(watched_channels, channel.id) # Removing all occurrences of the channel, just in case if len(watched_channels) != size_before: # Size differs: something must have been removed self.save_to_file() return True else: # Same size: list has not been modified return False def clear_watched_channels(self, guild: discord.Guild) -> None: """ Remove all channels from the watched channels list """ self.get_watched_channels_ids(guild).clear() self.save_to_file() def is_watched(self, guild: discord.Guild, channel: discord.VoiceChannel) -> bool: """ :return: Whether the given channel is watched """ return channel.id in self.get_watched_channels_ids(guild) def _get_created_channels(self, guild: discord.Guild) -> List[int]: guild_configuration = self._get_guild_configuration(guild) if "created_channels" not in guild_configuration: guild_configuration["created_channels"] = [] return guild_configuration["created_channels"] def add_channel_to_created(self, guild: discord.Guild, channel: discord.VoiceChannel) -> None: """ Add a voice channel to the list of channels that the bot has created and manages. """ self._get_created_channels(guild).append(channel.id) self.save_to_file() def remove_channel_from_created(self, guild: discord.Guild, channel: Union[discord.VoiceChannel, discord.Object]): """ Remove a voice channel from the list of channels that have been created by the bot """ created_channels = self._get_created_channels(guild) utils.list_remove_all_occurrences(created_channels, channel.id) # We could update the on-disk file, but we avoid it to limit the number of writes to disk. # It is not critical if this information is lost, and it will probably be updated properly anyway the next time the file is written # self.save_to_file() pass def is_created(self, guild: discord.Guild, channel: Union[discord.VoiceChannel, discord.Object]): """ :return: Whether the given channel is one of the channels that have been created by the bot """ return channel.id in self._get_created_channels(guild)