AutoLibrary/src/managers/config/ConfigManager.py

# -*- coding: utf-8 -*-
"""
Copyright (c) 2026 KenanZhu.
All rights reserved.

This software is provided "as is", without any warranty of any kind.
You may use, modify, and distribute this file under the terms of the MIT License.
See the LICENSE file for details.
"""
import os
import threading

from typing import Any, Optional

from utils.JSONReader import JSONReader
from utils.JSONWriter import JSONWriter
from interfaces.ConfigProvider import ConfigType, ConfigPath


# This config manager class only responsible for global and other
# unconfigurable config files.


class ConfigTemplate:
    """
        Config template class.
    """

    def __init__(
        self,
        config_type: ConfigType
    ):

        self.__config_type = config_type

    def template(
        self
    ) -> dict:
        """
            Get config template.

            Returns:
                dict: Config template.
        """
        match self.__config_type:
            case ConfigType.GLOBAL:
                return {
                    "automation": {
                        "run_path": {
                            "current": 0,
                            "paths": []
                        },
                        "user_path": {
                            "current": 0,
                            "paths": []
                        }
                    }
                }
            case ConfigType.BULLETIN:
                return {
                    "bulletin": [],
                    "last_sync_time": None
                }
            case ConfigType.TIMERTASK:
                return {
                    "timer_tasks": []
                }
            case _:
                return {}


class ConfigManager:

    def __init__(
        self,
        config_dir: str
    ):

        self.__config_dir = os.path.abspath(config_dir)
        self.__config_lock = threading.Lock()
        self.__config_data = {}

        self.initialize()

    def initialize(
        self
    ):

        for config_type in ConfigType:
            self.load(config_type)

    def load(
        self,
        config_type: ConfigType
    ):

        config_path = os.path.join(self.__config_dir, config_type.value)
        if os.path.exists(config_path):
            try:
                config_data = JSONReader(config_path).data()
                self.__config_data[config_type.value] = config_data
                return
            except:
                pass
        self.__config_data[config_type.value] = ConfigTemplate(config_type).template()
        JSONWriter(config_path, self.__config_data[config_type.value])

    def get(
        self,
        key: ConfigPath,
        default: Optional[Any] = None
    ) -> Any:

        with self.__config_lock:
            config_data = self.__config_data[key.config_type.value]
            if key.key == "":
                return config_data
            keys = key.key.split('.')
            for k in keys[:-1]:
                config_data = config_data.get(k, None)
                if config_data is None:
                    return default
            return config_data.get(keys[-1], default)

    def set(
        self,
        key: ConfigPath,
        value: Any = None
    ):

        with self.__config_lock:
            root_data = self.__config_data[key.config_type.value]
            if key.key == "":
                self.__config_data[key.config_type.value] = value
            else:
                keys = key.key.split('.')
                config_data = root_data
                for k in keys[:-1]:
                    if k not in config_data:
                        config_data[k] = {}
                    config_data = config_data[k]
                config_data[keys[-1]] = value
        self.save(key.config_type)

    def save(
        self,
        config_type: ConfigType
    ):

        config_path = os.path.join(self.__config_dir, config_type.value)
        JSONWriter(config_path, self.__config_data[config_type.value])

    def configDir(
        self
    ) -> str:

        return self.__config_dir


# ConfigManager singleton instance.
_config_manager_instance : ConfigManager | None = None

# Singleton instance of ConfigManager.
_instance_lock = threading.Lock()

def instance(
    config_dir: str = ""
) -> ConfigManager:
    """
        Initialize ConfigManager singleton instance.

        Args:
            config_dir (str): Config directory.
    """
    global _config_manager_instance
    with _instance_lock:
        if _config_manager_instance is None:
            if not config_dir:
                raise ValueError("ConfigManager 需要配置目录参数")
            _config_manager_instance = ConfigManager(config_dir)
        else:
            if config_dir == "":
                return _config_manager_instance
            if _config_manager_instance.configDir() != config_dir:
                raise ValueError("ConfigManager 的实例已初始化,不能使用不同的配置目录。")
    return _config_manager_instance