diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/hydrorollcore/config.py | 6 | ||||
| -rw-r--r-- | src/hydrorollcore/core.py | 31 | ||||
| -rw-r--r-- | src/hydrorollcore/utils.py | 192 |
3 files changed, 0 insertions, 229 deletions
diff --git a/src/hydrorollcore/config.py b/src/hydrorollcore/config.py deleted file mode 100644 index b76f4b2c..00000000 --- a/src/hydrorollcore/config.py +++ /dev/null @@ -1,6 +0,0 @@ -from pydantic import BaseModel - - -class Config(BaseModel): - rule_dir: list = [] - rules: list = [] diff --git a/src/hydrorollcore/core.py b/src/hydrorollcore/core.py deleted file mode 100644 index 6f6bbcc1..00000000 --- a/src/hydrorollcore/core.py +++ /dev/null @@ -1,31 +0,0 @@ -import importlib -from typing import List -from .exceptions import RuleLoadError -from .rule import Rule -from .config import Config - - -class Core: - def __init__(self, config: Config): - self.rule_dir = config.rule_dir - self.rules = config.rules - - def load_rules(self) -> List[Rule]: - loaded_rules = [] - for rule in self.rules: - try: - module = importlib.import_module(rule) - except ImportError as e: - raise RuleLoadError(f"Failed to load rule {rule}: {e}") from e - try: - rule_cls = getattr(module, rule.split(".")[-1]) - if not issubclass(rule_cls, Rule): - raise RuleLoadError( - f"Class '{rule_cls.__name__}' is not a subclass of 'Rule'" - ) - except AttributeError as e: - raise RuleLoadError( - f"Failed to get rule class from module '{rule}': {e}" - ) from e - loaded_rules.append(rule_cls()) - return loaded_rules diff --git a/src/hydrorollcore/utils.py b/src/hydrorollcore/utils.py deleted file mode 100644 index 1333815a..00000000 --- a/src/hydrorollcore/utils.py +++ /dev/null @@ -1,192 +0,0 @@ -import os -import json -import asyncio -import inspect -import os.path -import pkgutil -import importlib -import traceback -import dataclasses -from abc import ABC -from types import ModuleType -from functools import partial -from typing_extensions import ParamSpec -from importlib.abc import MetaPathFinder -from importlib.machinery import PathFinder -from typing import Any, List, Type, Tuple, TypeVar, Callable, Iterable, Coroutine - -from HydroRollCore.config import Config - -__all__ = [ - "ModulePathFinder", - "is_config_class", - "get_classes_from_module", - "get_classes_from_module_name", - "get_classes_from_dir", - "DataclassEncoder", - "samefile", - "sync_func_wrapper", -] - -_T = TypeVar("_T") -_P = ParamSpec("_P") -_R = TypeVar("_R") - - -class ModulePathFinder(MetaPathFinder): - """用于查找 HydroRollCore 组件的元路径查找器。""" - - path: List[str] = [] - - def find_spec(self, fullname, path=None, target=None): - if path is None: - path = [] - return PathFinder.find_spec(fullname, self.path + list(path), target) - - -def is_config_class(config_class: Any) -> bool: - """判断一个对象是否是配置类。 - - Args: - config_class: 待判断的对象。 - - Returns: - 返回是否是配置类。 - """ - return ( - inspect.isclass(config_class) - and issubclass(config_class, Config) - and isinstance(getattr(config_class, "__config_name__", None), str) - and ABC not in config_class.__bases__ - and not inspect.isabstract(config_class) - ) - - -def get_classes_from_module( - module: ModuleType, super_class: Type[_T] -) -> List[Type[_T]]: - """从模块中查找指定类型的类。 - - Args: - module: Python 模块。 - super_class: 要查找的类的超类。 - - Returns: - 返回符合条件的类的列表。 - """ - classes: List[Type[_T]] = [] - for _, module_attr in inspect.getmembers(module, inspect.isclass): - module_attr: type - if ( - (inspect.getmodule(module_attr) or module) is module - and issubclass(module_attr, super_class) - and module_attr != super_class - and ABC not in module_attr.__bases__ - and not inspect.isabstract(module_attr) - ): - classes.append(module_attr) - return classes - - -def get_classes_from_module_name( - name: str, super_class: Type[_T] -) -> List[Tuple[Type[_T], ModuleType]]: - """从指定名称的模块中查找指定类型的类。 - - Args: - name: 模块名称，格式和 Python `import` 语句相同。 - super_class: 要查找的类的超类。 - - Returns: - 返回由符合条件的类和模块组成的元组的列表。 - - Raises: - ImportError: 当导入模块过程中出现错误。 - """ - try: - importlib.invalidate_caches() - module = importlib.import_module(name) - importlib.reload(module) - return list( - map(lambda x: (x, module), get_classes_from_module(module, super_class)) - ) - except BaseException as e: - # 不捕获 KeyboardInterrupt - # 捕获 KeyboardInterrupt 会阻止用户关闭 Python 当正在导入的模块陷入死循环时 - if isinstance(e, KeyboardInterrupt): - raise e - raise ImportError(e, traceback.format_exc()) from e - - -def get_classes_from_dir( - dirs: Iterable[str], super_class: Type[_T] -) -> List[Tuple[Type[_T], ModuleType]]: - """从指定路径列表中的所有模块中查找指定类型的类，以 `_` 开头的插件不会被导入。路径可以是相对路径或绝对路径。 - - Args: - dirs: 储存模块的路径的列表。 - super_class: 要查找的类的超类。 - - Returns: - 返回由符合条件的类和模块组成的元组的列表。 - """ - classes: List[Tuple[Type[_T], ModuleType]] = [] - for module_info in pkgutil.iter_modules(dirs): - if not module_info.name.startswith("_"): - try: - classes.extend( - get_classes_from_module_name(module_info.name, super_class) - ) - except ImportError: - continue - return classes - - -class DataclassEncoder(json.JSONEncoder): - """用于解析 MessageSegment 的 JSONEncoder 类。""" - - def default(self, o): - return o.as_dict() if dataclasses.is_dataclass(o) else super().default(o) - - -def samefile(path1: str, path2: str) -> bool: - """一个 `os.path.samefile` 的简单包装。 - - Args: - path1: 路径1。 - path2: 路径2。 - - Returns: - 如果两个路径是否指向相同的文件或目录。 - """ - try: - return path1 == path2 or os.path.samefile(path1, path2) - except OSError: - return False - - -def sync_func_wrapper( - func: Callable[_P, _R], to_thread: bool = False -) -> Callable[_P, Coroutine[None, None, _R]]: - """包装一个同步函数为异步函数 - - Args: - func: 待包装的同步函数。 - to_thread: 在独立的线程中运行同步函数。 - - Returns: - 异步函数。 - """ - if to_thread: - - async def _wrapper(*args: _P.args, **kwargs: _P.kwargs): - loop = asyncio.get_running_loop() - func_call = partial(func, *args, **kwargs) - return await loop.run_in_executor(None, func_call) - - else: - - async def _wrapper(*args: _P.args, **kwargs: _P.kwargs): - return func(*args, **kwargs) - - return _wrapper |