147 lines
4.4 KiB
Python
147 lines
4.4 KiB
Python
import json
|
||
import logging
|
||
from pathlib import Path
|
||
from typing import Any, TypeVar
|
||
|
||
import aiofiles
|
||
|
||
from .config import Config
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
CONFIG_PATH = Path.cwd() / ".nyahome" / "config.json"
|
||
|
||
T = TypeVar("T", str, int, bool, list)
|
||
|
||
|
||
class ConfigManager:
|
||
"""
|
||
ConfigManager 携带一个初始化的 Config 实例。在 Config 初始化时，所有的默认设置键的值就都已经加载。
|
||
因此，如果不 load_config，ConfigManager 也将持有一套默认设置。
|
||
"""
|
||
|
||
def __init__(self) -> None:
|
||
CONFIG_PATH.parent.mkdir(exist_ok=True)
|
||
self._config = Config()
|
||
|
||
def _parse(self, config: dict) -> None:
|
||
"""
|
||
解析给定的字典作为配置。
|
||
|
||
Args:
|
||
config: 配置字典
|
||
"""
|
||
for key, value in config.items():
|
||
setattr(self._config, key, value)
|
||
|
||
def _dumps(self) -> str:
|
||
"""
|
||
将配置项序列化为 json 字符串，包含格式化缩进。
|
||
|
||
Returns:
|
||
json 字符串。
|
||
"""
|
||
config = {}
|
||
for attr in dir(self._config):
|
||
if not attr.startswith("_"):
|
||
value = getattr(self._config, attr)
|
||
config[attr] = value
|
||
return json.dumps(config, ensure_ascii=False, indent=2)
|
||
|
||
async def async_load_config(self) -> None:
|
||
async with aiofiles.open(CONFIG_PATH, "r", encoding="utf-8") as f:
|
||
self._parse(json.loads(await f.read()))
|
||
logger.info("异步从 config.json 读取设置完成。")
|
||
|
||
def sync_load_config(self) -> None:
|
||
with open(CONFIG_PATH, "r", encoding="utf-8") as f:
|
||
self._parse(json.load(f))
|
||
logger.info("同步从 config.json 读取设置完成。")
|
||
|
||
async def async_save_config(self) -> None:
|
||
async with aiofiles.open(CONFIG_PATH, "w", encoding="utf-8") as f:
|
||
await f.write(self._dumps())
|
||
logger.info("异步保存设置到 config.json 完成。")
|
||
|
||
def sync_save_config(self) -> None:
|
||
with open(CONFIG_PATH, "w", encoding="utf-8") as f:
|
||
f.write(self._dumps())
|
||
logger.info("同步保存设置到 config.json 完成。")
|
||
|
||
def get(self, key: str, default: T | None = None) -> T:
|
||
"""
|
||
获取配置项
|
||
|
||
Args:
|
||
key: 配置键
|
||
default: 默认值，如果不提供则会在获取配置项失败时报错
|
||
|
||
Returns:
|
||
返回配置值，返回类型根据提供的默认值进行推断。
|
||
"""
|
||
return getattr(self._config, key, default) # type: ignore[return-value]
|
||
|
||
def set(self, key: str, value: T) -> None:
|
||
"""
|
||
设置配置项。
|
||
|
||
Args:
|
||
key: 配置键名
|
||
value: 配置键的新值，可以是（且仅支持）字符串、整型以及列表。
|
||
|
||
Raises:
|
||
AttributeError: 配置键名错误
|
||
TypeError: 配置键值类型错误
|
||
"""
|
||
try:
|
||
old_value = self.get(key)
|
||
except AttributeError as e:
|
||
raise e
|
||
|
||
match old_value:
|
||
case str():
|
||
new_value = str(value)
|
||
case int():
|
||
new_value = int(value)
|
||
case bool():
|
||
new_value = bool(value)
|
||
case list():
|
||
new_value = list(value)
|
||
case _:
|
||
raise TypeError(f"不支持 {type(old_value).__name__} 类型的设置项。（{key}）")
|
||
setattr(self._config, key, new_value)
|
||
|
||
def reset(self, key: str) -> None:
|
||
"""
|
||
将配置项恢复至默认值。
|
||
|
||
Args:
|
||
key: 配置键名
|
||
|
||
Raises:
|
||
AttributeError: 配置键名错误
|
||
"""
|
||
ci = Config()
|
||
try:
|
||
default_value = getattr(ci, key)
|
||
except AttributeError as e:
|
||
raise e
|
||
setattr(self._config, key, default_value)
|
||
|
||
def get_config(self) -> dict[str, Any]:
|
||
config = {}
|
||
for attr in dir(self._config):
|
||
if not attr.startswith("_"):
|
||
value = getattr(self._config, attr)
|
||
config[attr] = value
|
||
return config
|
||
|
||
def set_config(self, config: dict[str, Any]) -> dict[str, Any]:
|
||
for attr in dir(self._config):
|
||
if not attr.startswith("_"):
|
||
setattr(self._config, attr, config[attr])
|
||
return self.get_config()
|
||
|
||
|
||
config_manager = ConfigManager()
|