diff --git a/starbot_datasource/__init__.py b/starbot_datasource/__init__.py index 0a877d5..2f7524d 100644 --- a/starbot_datasource/__init__.py +++ b/starbot_datasource/__init__.py @@ -1,3 +1,3 @@ +from .core.config import config from .core.event import EventType, DataSourceEvent -from .core.datasource import DataSource, DictDataSource, JsonDataSource -from .core.config import config \ No newline at end of file +from .core.datasource import DataSource, JsonDataSource diff --git a/starbot_datasource/core/config.py b/starbot_datasource/core/config.py new file mode 100644 index 0000000..04650da --- /dev/null +++ b/starbot_datasource/core/config.py @@ -0,0 +1,74 @@ +import json +from json import JSONDecodeError +from typing import Dict, Any, Type, Optional + +from loguru import logger + +from ..exception.DataSourceException import DataSourceException + + +class Config: + """ + 全局配置类 + """ + + __config: Dict[str, Any] + + def __init__(self): + try: + with open("config.json", encoding="utf-8") as file: + self.__config = json.loads(file.read()) + except Exception as ex: + if isinstance(ex, FileNotFoundError): + logger.error("配置文件 config.json 不存在") + elif isinstance(ex, UnicodeDecodeError): + logger.error("配置文件 config.json 编码不正确, 请将其转换为 UTF-8 格式编码") + elif isinstance(ex, JSONDecodeError): + logger.error("配置文件 config.json 内容格式不正确") + else: + logger.error(f"读取配置文件 config.json 异常 {ex}") + raise DataSourceException("读取配置文件 config.json 异常") + + def get(self, attribute: str, _type: Type, default: Optional[Any] = None) -> Optional[Any]: + """ + 获取配置项的值 + + Args: + attribute: 配置项,多个配置项之间可使用.分隔 + _type: 预期类型,若类型不匹配会产生警告,并尝试进行自动转换 + default: 默认值,若要获取的配置项不存在或类型转换失败,返回默认值。默认:None + + Returns: + 配置项的值,若要获取的配置项不存在或类型转换失败,返回默认值 + """ + keys = attribute.split(".") + conf = self.__config + + for key in keys: + if not isinstance(conf, dict): + logger.warning(f"配置项 {attribute} 不存在, 已使用默认值: {default}, 请补全配置文件 config.json") + return default + + if key not in conf: + logger.warning(f"配置项 {attribute} 不存在, 已使用默认值: {default}, 请补全配置文件 config.json") + return default + + conf = conf[key] + + if isinstance(conf, _type): + return conf + + logger.warning( + f"配置项 {attribute} 数据类型不正确, 预期类型: {_type}, 实际类型: {type(conf)}, 请检查配置文件 config.json" + ) + + try: + auto_cast_value = _type(conf) + logger.success(f"配置项 {attribute} 数据类型自动转换成功") + return auto_cast_value + except Exception: + logger.warning(f"配置项 {attribute} 数据类型自动转换失败, 已使用默认值: {default}, 请检查配置文件 config.json") + return default + + +config: Config = Config() diff --git a/starbot_datasource/core/datasource.py b/starbot_datasource/core/datasource.py index 6137366..747076d 100644 --- a/starbot_datasource/core/datasource.py +++ b/starbot_datasource/core/datasource.py @@ -3,12 +3,13 @@ import json import os from json import JSONDecodeError -from typing import List, Dict, Union, Optional, NoReturn, Set +from typing import Dict, NoReturn, Set from loguru import logger from pydantic import ValidationError from starbot_executor import executor +from .. import config from ..core.event import EventType, DataSourceEvent from ..core.model import Up from ..exception.DataSourceException import DataSourceException @@ -88,48 +89,17 @@ def update(self, up: Up) -> NoReturn: executor.dispatch(up, EventType.DataSourceEvent, DataSourceEvent.DataSourceUpdated) -class DictDataSource(DataSource): - """ - 使用字典初始化的推送配置数据源 - """ - - def __init__(self, dict_config: Union[List[Dict], Dict]): - super().__init__() - self.__config = dict_config - - if isinstance(self.__config, dict): - self.__config = [self.__config] - - async def load(self) -> NoReturn: - """ - 从字典中初始化配置 - """ - if self.ups: - return - - logger.info("已选用 Dict 作为 Bot 数据源") - logger.info("开始从 Dict 中初始化 Bot 配置") - - for up in self.__config: - try: - self.add(Up(**up)) - except ValidationError as ex: - raise DataSourceException(f"提供的配置字典中缺少必须的 {ex.errors()[0].get('loc')[-1]} 参数") - - logger.success(f"成功从 Dict 中导入了 {len(self.ups)} 个 UP 主") - - class JsonDataSource(DataSource): """ 使用 JSON 初始化的推送配置数据源 """ - def __init__(self, json_file: str, auto_reload: Optional[bool] = True, auto_reload_interval: Optional[int] = 5): + def __init__(self): super().__init__() self.__config = None - self.__json_file = json_file - self.__auto_reload = auto_reload - self.__auto_reload_interval = auto_reload_interval + self.__json_file = config.get("datasource.json_datasource.file_path", str, "推送配置.json") + self.__auto_reload = config.get("datasource.json_datasource.auto_reload", bool, True) + self.__auto_reload_interval = config.get("datasource.json_datasource.auto_reload_interval", int, 5) async def load(self) -> NoReturn: """