path: root/tools/cru-py/cru/service/_config.py

from collections.abc import Iterable
from typing import Any, Literal, overload

from cru import CruException, CruNotFound
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():
            try:
                if entry.value == "":
                    value_dict[key] = None
                else:
                    value = entry.value
                    config_item = self.configuration.get_or(key)
                    if config_item is not CruNotFound.VALUE:
                        value = config_item.value_type.convert_str_to_value(value)
                    value_dict[key] = 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()