mirror of
https://github.com/KenanZhu/AutoLibrary.git
synced 2026-06-18 07:23:03 +08:00
feat(utils): 添加 ConfigManager 与 JSON 配置读写，替换旧实现
add: - src/utils/ConfigManager.py - src/utils/JSONReader.py - src/utils/JSONWriter.py remove: - src/utils/ConfigReader.py - src/utils/ConfigWriter.py refactor: - 更新调用方以使用 ConfigManager / JSONReader / JSONWriter（见 ALConfigWidget.py、ALMainWindow.py、ALTimerTaskManageWidget.py、ALMainWorkers.py 等） - 统一方法命名（initlize* -> initialize*）、改进错误提示与配置路径管理 BREAKING CHANGE: 删除 ConfigReader/ConfigWriter，外部调用需改为 JSONReader/JSONWriter 或通过 ConfigManager 访问配置
This commit is contained in:
@@ -0,0 +1,233 @@
|
||||
# -*- 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 enum import Enum
|
||||
from typing import Any, Optional
|
||||
|
||||
from utils.JSONReader import JSONReader
|
||||
from utils.JSONWriter import JSONWriter
|
||||
|
||||
|
||||
# This config manager class only responsible for global and other
|
||||
# unconfigurable config files.
|
||||
|
||||
|
||||
class ConfigType(Enum):
|
||||
"""
|
||||
Config type class. Values represent the default filename.
|
||||
"""
|
||||
GLOBAL = "autolibrary.json" # Global config file.
|
||||
BULLETIN = "bulletin.json" # Bulletin board config file.
|
||||
TIMERTASK = "timer_task.json" # Timer task config file.
|
||||
|
||||
|
||||
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,
|
||||
config_type: ConfigType,
|
||||
key: str = "",
|
||||
default: Optional[Any] = None
|
||||
) -> Any:
|
||||
|
||||
with self.__config_lock:
|
||||
config_data = self.__config_data[config_type.value]
|
||||
if key == "":
|
||||
return config_data
|
||||
keys = 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,
|
||||
config_type: ConfigType,
|
||||
key: str = "",
|
||||
value: Any = None
|
||||
):
|
||||
|
||||
with self.__config_lock:
|
||||
root_data = self.__config_data[config_type.value]
|
||||
if key == "":
|
||||
self.__config_data[config_type.value] = value
|
||||
else:
|
||||
keys = 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(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 appDir(
|
||||
self
|
||||
) -> str:
|
||||
|
||||
return self.__config_dir
|
||||
|
||||
|
||||
_config_manager_instance = None
|
||||
|
||||
# Utility function to get config data (thread-safe and validated) from ConfigManager instance.
|
||||
def getValidateAutomationConfigPaths(
|
||||
) -> dict:
|
||||
"""
|
||||
Get validated automation config paths from ConfigManager instance.
|
||||
These function will validate the config paths and return the validated paths in a dict.
|
||||
|
||||
Returns:
|
||||
dict: Validated automation config paths.
|
||||
"""
|
||||
config_paths = {"run": "", "user": ""}
|
||||
auto_config = _config_manager_instance.get(ConfigType.GLOBAL, "automation", {})
|
||||
for cfg_type in ["run", "user"]:
|
||||
paths = auto_config.get(f"{cfg_type}_path", {}).get("paths", [])
|
||||
index = auto_config.get(f"{cfg_type}_path", {}).get("current", 0)
|
||||
if paths == []:
|
||||
paths.append(os.path.join(_config_manager_instance.appDir(), f"{cfg_type}.json"))
|
||||
if index < 0:
|
||||
index = 0
|
||||
if index >= len(paths):
|
||||
index = len(paths) - 1
|
||||
config_paths[cfg_type] = paths[index]
|
||||
data = {"current": index, "paths": paths}
|
||||
auto_config[f"{cfg_type}_path"] = data
|
||||
_config_manager_instance.set(ConfigType.GLOBAL, "automation", auto_config)
|
||||
return config_paths
|
||||
|
||||
def getBaseConfigDir(
|
||||
) -> str:
|
||||
|
||||
return _config_manager_instance.appDir()
|
||||
|
||||
# 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:
|
||||
_config_manager_instance = ConfigManager(config_dir)
|
||||
else:
|
||||
if config_dir == "":
|
||||
return _config_manager_instance
|
||||
if _config_manager_instance.appDir() != config_dir:
|
||||
raise ValueError(
|
||||
"ConfigManager 的实例已初始化，不能使用不同的配置目录。")
|
||||
return _config_manager_instance
|
||||
@@ -1,115 +0,0 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
Copyright (c) 2025 - 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 json
|
||||
import copy
|
||||
|
||||
from typing import Any
|
||||
|
||||
|
||||
class ConfigReader:
|
||||
"""
|
||||
Config reader class.
|
||||
|
||||
This class is used to read config file in JSON format.
|
||||
|
||||
Args:
|
||||
config_path (str): The path of config file.
|
||||
|
||||
Examples:
|
||||
>>> print(open("config.json", "r", encoding="utf-8").read())
|
||||
{
|
||||
"key1": {
|
||||
"key2": "value1"
|
||||
}
|
||||
}
|
||||
>>> config_reader = ConfigReader("config.json")
|
||||
>>> config_reader.get("key1/key2")
|
||||
"value1"
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
config_path: str
|
||||
):
|
||||
|
||||
self.__config_path = config_path
|
||||
self.__config_data = None
|
||||
self.__readConfig()
|
||||
|
||||
|
||||
def __readConfig(
|
||||
self
|
||||
):
|
||||
|
||||
try:
|
||||
with open(self.__config_path, 'r', encoding='utf-8') as file:
|
||||
self.__config_data = json.load(file)
|
||||
except FileNotFoundError as e:
|
||||
raise Exception(f"配置文件不存在: {self.__config_path}") from e
|
||||
except PermissionError as e:
|
||||
raise Exception(f"没有足够的权限读取配置文件: {self.__config_path}") from e
|
||||
except json.JSONDecodeError as e:
|
||||
raise Exception(f"JSON 解析错误: {self.__config_path}") from e
|
||||
except Exception as e:
|
||||
raise Exception(f"读取配置文件时未知错误: {e}") from e
|
||||
|
||||
|
||||
def getConfigs(
|
||||
self
|
||||
) -> dict:
|
||||
|
||||
return self.__config_data.copy()
|
||||
|
||||
|
||||
def getConfig(
|
||||
self,
|
||||
key: str
|
||||
) -> Any:
|
||||
|
||||
config = self.__config_data.get(key, {})
|
||||
return copy.deepcopy(config)
|
||||
|
||||
|
||||
def get(
|
||||
self,
|
||||
key: str,
|
||||
default: Any = None
|
||||
) -> Any:
|
||||
|
||||
keys = key.split('/')
|
||||
current = self.__config_data
|
||||
for k in keys:
|
||||
if isinstance(current, dict) and k in current:
|
||||
current = current[k]
|
||||
else:
|
||||
return default
|
||||
return copy.deepcopy(current)
|
||||
|
||||
|
||||
def hasConfig(
|
||||
self,
|
||||
key: str
|
||||
) -> bool:
|
||||
|
||||
return self.getConfig(key) != {}
|
||||
|
||||
|
||||
def reReadConfig(
|
||||
self
|
||||
) -> bool:
|
||||
|
||||
return self.__readConfig()
|
||||
|
||||
|
||||
def configPath(
|
||||
self
|
||||
) -> str:
|
||||
|
||||
return self.__config_path
|
||||
@@ -1,116 +0,0 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
Copyright (c) 2025 - 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 json
|
||||
|
||||
from typing import Any
|
||||
|
||||
|
||||
class ConfigWriter:
|
||||
"""
|
||||
Config writer class.
|
||||
|
||||
This class is used to write config file in JSON format.
|
||||
|
||||
Args:
|
||||
config_path (str): The path of config file.
|
||||
config_data (dict): The config data to be written.
|
||||
|
||||
Examples:
|
||||
>>> config_data = {
|
||||
... "key1": {
|
||||
... "key2": "value1"
|
||||
... }
|
||||
... }
|
||||
>>> config_writer = ConfigWriter("config.json", config_data)
|
||||
>>> config_writer.set("key1/key2", "value1")
|
||||
True
|
||||
>>> print(open("config.json", "r", encoding="utf-8").read())
|
||||
{
|
||||
"key1": {
|
||||
"key2": "value1"
|
||||
}
|
||||
}
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
config_path: str,
|
||||
config_data: dict
|
||||
):
|
||||
|
||||
self.__config_path = config_path
|
||||
self.__config_data = config_data.copy() if config_data is not None else {}
|
||||
self.__writeConfig()
|
||||
|
||||
|
||||
def __writeConfig(
|
||||
self
|
||||
):
|
||||
|
||||
try:
|
||||
with open(self.__config_path, "w", encoding="utf-8") as f:
|
||||
json.dump(self.__config_data, f, indent=4, sort_keys=False)
|
||||
except PermissionError as e:
|
||||
raise Exception(f"没有足够的权限写入配置文件: {self.__config_path}") from e
|
||||
except IOError as e:
|
||||
raise Exception(f"写入配置文件时发生 IO 错误: {self.__config_path}") from e
|
||||
except TypeError as e:
|
||||
raise Exception(f"配置数据包含无法 JSON 序列化的类型: {e}") from e
|
||||
except Exception as e:
|
||||
raise Exception(f"写入配置文件时未知错误: {e}") from e
|
||||
|
||||
|
||||
def setConfigs(
|
||||
self,
|
||||
configs: dict
|
||||
) -> bool:
|
||||
|
||||
self.__config_data = configs
|
||||
return self.__writeConfig()
|
||||
|
||||
|
||||
def setConfig(
|
||||
self,
|
||||
key: str,
|
||||
value: dict
|
||||
) -> bool:
|
||||
|
||||
self.__config_data[key] = value
|
||||
return self.__writeConfig()
|
||||
|
||||
|
||||
def set(
|
||||
self,
|
||||
key: str,
|
||||
value: Any
|
||||
) -> bool:
|
||||
|
||||
keys = key.replace("\\", "/").split("/")
|
||||
current = self.__config_data
|
||||
for k in keys[:-1]:
|
||||
if k not in current or not isinstance(current[k], dict):
|
||||
current[k] = {}
|
||||
current = current[k]
|
||||
current[keys[-1]] = value
|
||||
return self.__writeConfig()
|
||||
|
||||
|
||||
def reWriteConfig(
|
||||
self
|
||||
) -> bool:
|
||||
|
||||
return self.__writeConfig()
|
||||
|
||||
|
||||
def configPath(
|
||||
self
|
||||
) -> str:
|
||||
|
||||
return self.__config_path
|
||||
@@ -0,0 +1,85 @@
|
||||
# -*- 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 json
|
||||
|
||||
|
||||
class JSONReader:
|
||||
"""
|
||||
JSON reader class.
|
||||
|
||||
This class is used to read JSON file.
|
||||
|
||||
Args:
|
||||
json_path (str): The path of JSON file.
|
||||
|
||||
Examples:
|
||||
>>> print(open("config.json", "r", encoding="utf-8").read())
|
||||
{
|
||||
"key1": {
|
||||
"key2": "value1"
|
||||
}
|
||||
}
|
||||
>>> json_reader = JSONReader("config.json")
|
||||
>>> data = json_reader.data()
|
||||
>>> data["key1"]["key2"]
|
||||
"value1"
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
json_path: str
|
||||
):
|
||||
|
||||
self.__json_path = os.path.abspath(json_path)
|
||||
self.__json_data = None
|
||||
self.__read()
|
||||
|
||||
|
||||
def __read(
|
||||
self
|
||||
):
|
||||
|
||||
try:
|
||||
with open(self.__json_path, 'r', encoding='utf-8') as file:
|
||||
self.__json_data = json.load(file)
|
||||
except FileNotFoundError as e:
|
||||
raise Exception(f"文件不存在: {self.__json_path}") from e
|
||||
except PermissionError as e:
|
||||
raise Exception(f"没有足够的权限读取文件: {self.__json_path}") from e
|
||||
except json.JSONDecodeError as e:
|
||||
raise Exception(f"JSON 解析错误: {self.__json_path}") from e
|
||||
except Exception as e:
|
||||
raise Exception(f"读取文件时发生未知错误: {e}") from e
|
||||
|
||||
|
||||
def read(
|
||||
self
|
||||
) -> bool:
|
||||
|
||||
try:
|
||||
self.__read()
|
||||
except:
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def data(
|
||||
self
|
||||
) -> dict:
|
||||
|
||||
return self.__json_data.copy()
|
||||
|
||||
|
||||
def path(
|
||||
self
|
||||
) -> str:
|
||||
|
||||
return self.__json_path
|
||||
@@ -0,0 +1,82 @@
|
||||
# -*- 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 json
|
||||
|
||||
|
||||
class JSONWriter:
|
||||
"""
|
||||
JSON writer class.
|
||||
|
||||
This class is used to write JSON file.
|
||||
|
||||
Args:
|
||||
json_path (str): The path of JSON file.
|
||||
json_data (dict): The JSON data to be written.
|
||||
|
||||
Examples:
|
||||
>>> json_data = {
|
||||
... "key1": {
|
||||
... "key2": "value1"
|
||||
... }
|
||||
... }
|
||||
>>> json_writer = JSONWriter("config.json", json_data)
|
||||
>>> print(open("config.json", "r", encoding="utf-8").read())
|
||||
{
|
||||
"key1": {
|
||||
"key2": "value1"
|
||||
}
|
||||
}
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
json_path: str,
|
||||
json_data: dict
|
||||
):
|
||||
|
||||
self.__json_path = os.path.abspath(json_path)
|
||||
self.__json_data = json_data.copy() if json_data is not None else {}
|
||||
self.__write()
|
||||
|
||||
|
||||
def __write(
|
||||
self
|
||||
):
|
||||
|
||||
try:
|
||||
with open(self.__json_path, "w", encoding="utf-8") as f:
|
||||
json.dump(self.__json_data, f, indent=4, sort_keys=False)
|
||||
except PermissionError as e:
|
||||
raise Exception(f"没有足够的权限写入文件: {self.__json_path}") from e
|
||||
except IOError as e:
|
||||
raise Exception(f"写入文件时发生 IO 错误: {self.__json_path}") from e
|
||||
except TypeError as e:
|
||||
raise Exception(f"JSON 数据包含无法 JSON 序列化的类型: {e}") from e
|
||||
except Exception as e:
|
||||
raise Exception(f"写入文件时发生未知错误: {e}") from e
|
||||
|
||||
|
||||
def write(
|
||||
self
|
||||
) -> bool:
|
||||
|
||||
try:
|
||||
self.__write()
|
||||
except:
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def path(
|
||||
self
|
||||
) -> str:
|
||||
|
||||
return self.__json_path
|
||||
@@ -2,6 +2,7 @@
|
||||
Utils module for the AutoLibrary project.
|
||||
|
||||
Here are the classes and modules in this package:
|
||||
- ConfigReader: Configuration reader class for the AutoLibrary project.
|
||||
- ConfigWriter: Configuration writer class for the AutoLibrary project.
|
||||
- ConfigManager: Configuration manager class for the AutoLibrary project.
|
||||
- JSONReader: JSON reader class for the AutoLibrary project.
|
||||
- JSONWriter: JSON writer class for the AutoLibrary project.
|
||||
"""
|
||||
Reference in New Issue
Block a user