feat: Global config support

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
+from .core.datasource import DataSource, JsonDataSource

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()

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:
         """