from collections.abc import Iterable from typing import Any, Literal, overload from cru import CruException from cru.config import Configuration, ConfigItem from cru.value import ( INTEGER_VALUE_TYPE, TEXT_VALUE_TYPE, CruValueTypeError, RandomStringValueGenerator, UuidValueGenerator, ) from cru.parsing import ParseError, SimpleLineConfigParser from ._base import AppFeaturePath, AppCommandFeatureProvider class AppConfigError(CruException): def __init__( self, message: str, configuration: Configuration, *args, **kwargs ) -> None: super().__init__(message, *args, **kwargs) self._configuration = configuration @property def configuration(self) -> Configuration: return self._configuration class AppConfigFileError(AppConfigError): def __init__( self, message: str, configuration: Configuration, *args, **kwargs, ) -> None: super().__init__(message, configuration, *args, **kwargs) class AppConfigFileNotFoundError(AppConfigFileError): def __init__( self, message: str, configuration: Configuration, file_path: str, *args, **kwargs, ) -> None: super().__init__(message, configuration, *args, **kwargs) self._file_path = file_path @property def file_path(self) -> str: return self._file_path class AppConfigFileParseError(AppConfigFileError): def __init__( self, message: str, configuration: Configuration, file_content: str, *args, **kwargs, ) -> None: super().__init__(message, configuration, *args, **kwargs) self._file_content = file_content self.__cause__: ParseError @property def file_content(self) -> str: return self._file_content def get_user_message(self) -> str: return f"Error while parsing config file at line {self.__cause__.line_number}." class AppConfigFileEntryError(AppConfigFileError): def __init__( self, message: str, configuration: Configuration, entries: Iterable[SimpleLineConfigParser.Entry], *args, **kwargs, ) -> None: super().__init__(message, configuration, *args, **kwargs) self._entries = list(entries) @property def error_entries(self) -> list[SimpleLineConfigParser.Entry]: return self._entries @staticmethod def entries_to_friendly_message( entries: Iterable[SimpleLineConfigParser.Entry], ) -> str: return "\n".join( f"line {entry.line_number}: {entry.key}={entry.value}" for entry in entries ) @property def friendly_message_head(self) -> str: return "Error entries found in config file" def get_user_message(self) -> str: return ( f"{self.friendly_message_head}:\n" f"{self.entries_to_friendly_message(self.error_entries)}" ) class AppConfigDuplicateEntryError(AppConfigFileEntryError): @property def friendly_message_head(self) -> str: return "Duplicate entries found in config file" class AppConfigEntryValueFormatError(AppConfigFileEntryError): @property def friendly_message_head(self) -> str: return "Invalid value format for entries" class AppConfigItemNotSetError(AppConfigError): def __init__( self, message: str, configuration: Configuration, items: list[ConfigItem], *args, **kwargs, ) -> None: super().__init__(message, configuration, *args, **kwargs) self._items = items class ConfigManager(AppCommandFeatureProvider): def __init__(self) -> None: super().__init__("config-manager") configuration = Configuration() self._configuration = configuration self._loaded: bool = False self._init_app_defined_items() def _init_app_defined_items(self) -> None: prefix = self.config_name_prefix def _add_text(name: str, description: str) -> ConfigItem: item = ConfigItem(f"{prefix}_{name}", description, TEXT_VALUE_TYPE) self.configuration.add(item) return item def _add_uuid(name: str, description: str) -> ConfigItem: item = ConfigItem( f"{prefix}_{name}", description, TEXT_VALUE_TYPE, default=UuidValueGenerator(), ) self.configuration.add(item) return item def _add_random_string( name: str, description: str, length: int = 32, secure: bool = True ) -> ConfigItem: item = ConfigItem( f"{prefix}_{name}", description, TEXT_VALUE_TYPE, default=RandomStringValueGenerator(length, secure), ) self.configuration.add(item) return item def _add_int(name: str, description: str) -> ConfigItem: item = ConfigItem(f"{prefix}_{name}", description, INTEGER_VALUE_TYPE) self.configuration.add(item) return item self._domain = _add_text("DOMAIN", "domain name") self._email = _add_text("EMAIL", "admin email address") _add_text( "AUTO_BACKUP_COS_SECRET_ID", "access key id for Tencent COS, used for auto backup", ) _add_text( "AUTO_BACKUP_COS_SECRET_KEY", "access key secret for Tencent COS, used for auto backup", ) _add_text( "AUTO_BACKUP_COS_ENDPOINT", "endpoint (cos.*.myqcloud.com) for Tencent COS, used for auto backup", ) _add_text( "AUTO_BACKUP_COS_BUCKET", "bucket name for Tencent COS, used for auto backup", ) _add_uuid("V2RAY_TOKEN", "v2ray user id") _add_uuid("V2RAY_PATH", "v2ray path, which will be prefixed by _") _add_random_string("2FAUTH_APP_KEY", "2FAuth App Key") _add_text("2FAUTH_MAIL_USERNAME", "2FAuth SMTP user") _add_text("2FAUTH_MAIL_PASSWORD", "2FAuth SMTP password") _add_text("GIT_SERVER_USERNAME", "Git server username") _add_text("GIT_SERVER_PASSWORD", "Git server password") def setup(self) -> None: self._config_file_path = self.app.data_dir.add_subpath( "config", False, description="Configuration file path." ) @property def config_name_prefix(self) -> str: return self.app.app_id.upper() @property def configuration(self) -> Configuration: return self._configuration @property def config_file_path(self) -> AppFeaturePath: return self._config_file_path @property def all_set(self) -> bool: return self.configuration.all_set def get_item(self, name: str) -> ConfigItem[Any]: if not name.startswith(self.config_name_prefix + "_"): name = f"{self.config_name_prefix}_{name}" item = self.configuration.get_or(name, None) if item is None: raise AppConfigError(f"Config item '{name}' not found.", self.configuration) return item @overload def get_item_value_str(self, name: str) -> str: ... @overload def get_item_value_str(self, name: str, ensure_set: Literal[True]) -> str: ... @overload def get_item_value_str(self, name: str, ensure_set: bool = True) -> str | None: ... def get_item_value_str(self, name: str, ensure_set: bool = True) -> str | None: self.load_config_file() item = self.get_item(name) if not item.is_set: if ensure_set: raise AppConfigItemNotSetError( f"Config item '{name}' is not set.", self.configuration, [item] ) return None return item.value_str def get_str_dict(self, ensure_all_set: bool = True) -> dict[str, str]: self.load_config_file() if ensure_all_set and not self.configuration.all_set: raise AppConfigItemNotSetError( "Some config items are not set.", self.configuration, self.configuration.get_unset_items(), ) return self.configuration.to_str_dict() @property def domain_item_name(self) -> str: return self._domain.name def get_domain_value_str(self) -> str: return self.get_item_value_str(self._domain.name) def get_email_value_str_optional(self) -> str | None: return self.get_item_value_str(self._email.name, ensure_set=False) def _set_with_default(self) -> None: if not self.configuration.all_not_set: raise AppConfigError( "Config is not clean. " "Some config items are already set. " "Can't set again with default value.", self.configuration, ) for item in self.configuration: if item.can_generate_default: item.set_value(item.generate_default_value()) def _to_config_file_content(self) -> str: content = "".join( [ f"{item.name}={item.value_str if item.is_set else ''}\n" for item in self.configuration ] ) return content def _create_init_config_file(self) -> None: if self.config_file_path.check_self(): raise AppConfigError( "Config file already exists.", self.configuration, user_message=f"The config file at " f"{self.config_file_path.full_path_str} already exists.", ) self._set_with_default() self.config_file_path.ensure() with open( self.config_file_path.full_path, "w", encoding="utf-8", newline="\n" ) as file: file.write(self._to_config_file_content()) def _parse_config_file(self) -> SimpleLineConfigParser.Result: if not self.config_file_path.check_self(): raise AppConfigFileNotFoundError( "Config file not found.", self.configuration, self.config_file_path.full_path_str, user_message=f"The config file at " f"{self.config_file_path.full_path_str} does not exist. " f"You can create an initial one with 'init' command.", ) text = self.config_file_path.full_path.read_text() try: parser = SimpleLineConfigParser() return parser.parse(text) except ParseError as e: raise AppConfigFileParseError( "Failed to parse config file.", self.configuration, text ) from e def _parse_and_print_config_file(self) -> None: parse_result = self._parse_config_file() for entry in parse_result: print(f"{entry.key}={entry.value}") def _check_duplicate( self, parse_result: dict[str, list[SimpleLineConfigParser.Entry]], ) -> dict[str, SimpleLineConfigParser.Entry]: entry_dict: dict[str, SimpleLineConfigParser.Entry] = {} duplicate_entries: list[SimpleLineConfigParser.Entry] = [] for key, entries in parse_result.items(): entry_dict[key] = entries[0] if len(entries) > 1: duplicate_entries.extend(entries) if len(duplicate_entries) > 0: raise AppConfigDuplicateEntryError( "Duplicate entries found.", self.configuration, duplicate_entries ) return entry_dict def _check_type( self, entry_dict: dict[str, SimpleLineConfigParser.Entry] ) -> dict[str, Any]: value_dict: dict[str, Any] = {} error_entries: list[SimpleLineConfigParser.Entry] = [] errors: list[CruValueTypeError] = [] for key, entry in entry_dict.items(): config_item = self.configuration.get(key) try: if entry.value == "": value_dict[key] = None else: value_dict[key] = config_item.value_type.convert_str_to_value( entry.value ) except CruValueTypeError as e: error_entries.append(entry) errors.append(e) if len(error_entries) > 0: raise AppConfigEntryValueFormatError( "Entry value format is not correct.", self.configuration, error_entries, ) from ExceptionGroup("Multiple format errors occurred.", errors) return value_dict def _read_config_file(self) -> dict[str, Any]: parsed = self._parse_config_file() entry_groups = parsed.cru_iter().group_by(lambda e: e.key) entry_dict = self._check_duplicate(entry_groups) value_dict = self._check_type(entry_dict) return value_dict def _real_load_config_file(self) -> None: self.configuration.reset_all() value_dict = self._read_config_file() for key, value in value_dict.items(): if value is None: continue self.configuration.set_config_item(key, value) def load_config_file(self, force=False) -> None: if force or not self._loaded: self._real_load_config_file() self._loaded = True def _print_app_config_info(self): for item in self.configuration: print(item.description_str) def get_command_info(self): return "config", "Manage configuration." def setup_arg_parser(self, arg_parser) -> None: subparsers = arg_parser.add_subparsers( dest="config_command", required=True, metavar="CONFIG_COMMAND" ) _init_parser = subparsers.add_parser( "init", help="create an initial config file" ) _print_app_parser = subparsers.add_parser( "print-app", help="print information of the config items defined by app", ) _print_parser = subparsers.add_parser("print", help="print current config") _check_config_parser = subparsers.add_parser( "check", help="check the validity of the config file", ) _check_config_parser.add_argument( "-f", "--format-only", action="store_true", help="only check content format, not app config item requirements.", ) def run_command(self, args) -> None: if args.config_command == "init": self._create_init_config_file() elif args.config_command == "print-app": self._print_app_config_info() elif args.config_command == "print": self._parse_and_print_config_file() elif args.config_command == "check": if args.format_only: self._parse_config_file() else: self._read_config_file()