diff options
| -rw-r--r-- | README.md | 21 | ||||
| -rw-r--r-- | src/.gitignore | 2 | ||||
| -rw-r--r-- | src/hydrorollcore/__init__.py | 6 | ||||
| -rw-r--r-- | src/hydrorollcore/cli.py | 31 | ||||
| -rw-r--r-- | src/hydrorollcore/config.py | 6 | ||||
| -rw-r--r-- | src/hydrorollcore/consts/templates.py | 6 | ||||
| -rw-r--r-- | src/hydrorollcore/core.py | 31 | ||||
| -rw-r--r-- | src/hydrorollcore/event.py | 41 | ||||
| -rw-r--r-- | src/hydrorollcore/logging.py (renamed from src/hydrorollcore/log.py) | 8 | ||||
| -rw-r--r-- | src/hydrorollcore/manager.py | 18 | ||||
| -rw-r--r-- | src/hydrorollcore/rule.py | 2 | ||||
| -rw-r--r-- | src/hydrorollcore/typing.py | 1 | ||||
| -rw-r--r-- | src/hydrorollcore/utils.py | 192 | ||||
| -rw-r--r-- | src/test.py | 4 |
14 files changed, 112 insertions, 257 deletions
@@ -41,13 +41,22 @@ 2. 创建规则包实例 + 创建`cli.py`并写入以下内容: + + ```python + import HydroRollCore + + client = HydroRollCore.Cli() + client.parse_args() + ``` + + 打开终端并执行: + ``` shell - mkdir myrules && cd myrules && mkdir rule1 - echo.> config.toml - echo.> __init__.py + python cli.py --new --path MyRule ``` - 在 `__init__.py` 创建一个 `rule` 实例并继承 `Rule` 基类, 通过编写合适的相关方法与类注册规则包实现规则的自定义。 + 你可以在生成的 `MyRule\rule.py` 创建一个或者多个 `rule` 实例并继承 `Rule` 基类, 通过编写合适的相关方法与类注册规则包实现规则的自定义。 ``` python from HydroRollCore import Rule, Result, Dice @@ -63,9 +72,11 @@ def check(self, dice: Dice) -> Result: """声明规则包检定方式""" - return Result("myevent.event1", True) + return Result("event1", True) ``` + `check`函数应当返回一个`Result`对象,它应当包含一个消息事件名(例如示例中的`event1`),该消息事件名应当在 `MyRule\event.py` 中被注册。消息事件的动态内容通过`{name}`的方式声明并通过`name="内容"`的方式实现。 + 3. 合理修改你的 `config.toml` 配置文件,完成注册! ### 🎍Sites diff --git a/src/.gitignore b/src/.gitignore new file mode 100644 index 00000000..3119f231 --- /dev/null +++ b/src/.gitignore @@ -0,0 +1,2 @@ +logs/ +test/
\ No newline at end of file diff --git a/src/hydrorollcore/__init__.py b/src/hydrorollcore/__init__.py index 7a2bba77..2aca9dd7 100644 --- a/src/hydrorollcore/__init__.py +++ b/src/hydrorollcore/__init__.py @@ -1,6 +1,6 @@ from HydroRollCore.cli import Cli -from HydroRollCore.config import Config from HydroRollCore.rule import Rule, Result, Dice -from HydroRollCore.core import Core +from HydroRollCore.typing import Config as ConfigTyping +from HydroRollCore.event import Event -__all__ = ["Core", "Rule", "Config", "Cli", "Result", "Dice"] +__all__ = ["Rule", "Cli", "Result", "Dice", "Event", "ConfigTyping"] diff --git a/src/hydrorollcore/cli.py b/src/hydrorollcore/cli.py index 5351d96d..e372b4b4 100644 --- a/src/hydrorollcore/cli.py +++ b/src/hydrorollcore/cli.py @@ -1,40 +1,45 @@ from pathlib import Path from .consts import templates - -# from tkinter import messagebox +from .logging import logger import argparse import os import sys +import importlib class Cli: - def parse_args(self): + def parse_args(self, argv: list[str] | None = None): parser = argparse.ArgumentParser(description="HydroRoll 命令行工具") parser.add_argument("--new", action="store_true", help="创建一个 HydroRoll 规则包模板") parser.add_argument("--run", action="store_true", help="运行 HydroRoll 规则包") - # parser.add_argument("--gui", action="store_true", help="显示弹窗") + parser.add_argument("--gui", action="store_true", help="显示弹窗") parser.add_argument("--path", help="指定路径") - args = parser.parse_args() + args = parser.parse_args(argv if argv else sys.argv[1:]) - # if args.gui: - # messagebox.showinfo("提示", "这是一个弹窗!") + if args.gui: + logger.critical("选项[--gui]尚未被支持!") + sys.exit(1) path = Path(args.path).resolve() if args.path else Path(os.getcwd()).resolve() if args.new and args.run: - print("无法确定的指令要求: 你同时指定了new与run指令。") + logger.error("无法确定的指令要求: 你同时指定了new与run指令。") sys.exit(1) if args.new: if path.exists(): - print("指定的文件夹已经存在!") + logger.error("指定的文件夹已经存在!") sys.exit(1) path.mkdir(parents=True, exist_ok=True) - (path / "rule.py").write_text(templates.RULE) - (path / "event.py").write_text(templates.EVENT) - (path / "dice.py").write_text(templates.DICE) + (path / "rule.py").write_text(templates.RULE, encoding="utf-8") + (path / "event.py").write_text(templates.EVENT, encoding="utf-8") + (path / "dice.py").write_text(templates.DICE, encoding="utf-8") + + logger.success("HydroRoll 规则包模板已创建!") - print("HydroRoll 规则包模板已创建!") + if args.run: + sys.path.append(str(path)) + importlib.import_module("event") 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/consts/templates.py b/src/hydrorollcore/consts/templates.py index 4cf72989..8ab60ae5 100644 --- a/src/hydrorollcore/consts/templates.py +++ b/src/hydrorollcore/consts/templates.py @@ -1,5 +1,6 @@ RULE = """from HydroRollCore import Rule, Result, Dice + class MyRule(Rule): \"\"\"自设规则包\"\"\" @@ -18,6 +19,7 @@ EVENT = """from HydroRollCore import Event __events__ = ["MyEvent"] + class MyEvent(Event): name = "event1" output = "检定成功!" @@ -25,6 +27,10 @@ class MyEvent(Event): DICE = """from HydroRollCore import Dice +import random +import re + + class BaseDice(Dice): \"\"\"多面骰\"\"\" 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/event.py b/src/hydrorollcore/event.py index e8edeb98..9ce34bd9 100644 --- a/src/hydrorollcore/event.py +++ b/src/hydrorollcore/event.py @@ -1,12 +1,43 @@ -from abc import ABCMeta +from .typing import Dict +from .logging import logger +import re -class Event(metaclass=ABCMeta): +__all__ = ["Event", "events"] + + +class Events: + """事件集合""" + + _events: Dict[str, str] = {} + + def regist(self, name: str, output: str) -> None: + self._events[name.lower()] = output + + def process(self, name: str, **kwargs) -> str: + if string := self._events.get(name.lower()): + return self._format(string, **kwargs) + logger.warning(f"事件[{name.lower()}]不存在,将返回空字符串!") + return "" + + def _format(self, string: str, **kwargs): + pattern = r"{(.*?)}" + values = re.findall(pattern, string) + for value in values: + kwarg = kwargs.get(value) + value = kwarg if kwarg else "" + string = re.sub(pattern, value, string) + return string + + +class Event: """事件基类""" name: str output: str - def __init__(self, name: str, output: str) -> None: - self.name = name - self.output = output + def __init_subclass__(cls) -> None: + events.regist(cls.name, cls.output) + + +events = Events() diff --git a/src/hydrorollcore/log.py b/src/hydrorollcore/logging.py index ed62d0ae..a559561f 100644 --- a/src/hydrorollcore/log.py +++ b/src/hydrorollcore/logging.py @@ -15,10 +15,14 @@ logger = multilogger( name="HydroRoll", payload="Core", level="DEBUG" if DEBUG else "INFO" ) current_path = Path(__file__).resolve().parent -LOG_PATH = current_path / "logs" / (datetime.now().strftime("%Y-%m-%d") + ".log") +LOG_PATH = current_path / "logs" if not LOG_PATH.exists(): LOG_PATH.mkdir(parents=True, exist_ok=True) -logger.add(sink=LOG_PATH, level="INFO", rotation="10 MB") # 每个日志文件最大为 10MB +logger.add( + sink=LOG_PATH / (datetime.now().strftime("%Y-%m-%d") + ".log"), + level="INFO", + rotation="10 MB", +) # 每个日志文件最大为 10MB def error_or_exception(message: str, exception: Exception, verbose: bool = True): diff --git a/src/hydrorollcore/manager.py b/src/hydrorollcore/manager.py new file mode 100644 index 00000000..42e79563 --- /dev/null +++ b/src/hydrorollcore/manager.py @@ -0,0 +1,18 @@ +from .event import Events, events +from .logging import logger +from .typing import Dict + + +class Manager: + """事件处理单元""" + + events: Events + + def __init__(self, _events: Events = None) -> None: + self.events = _events if _events else events + + def roll(self): + ... + + +manager = Manager() diff --git a/src/hydrorollcore/rule.py b/src/hydrorollcore/rule.py index 127ed805..a1f04151 100644 --- a/src/hydrorollcore/rule.py +++ b/src/hydrorollcore/rule.py @@ -1,6 +1,7 @@ from abc import ABCMeta, abstractmethod from enum import Enum from .exceptions import HydroError +from .typing import Dict __all__ = ["RuleLoadType", "Result", "Dice", "Rule"] @@ -62,6 +63,7 @@ class Rule(metaclass=ABCMeta): """规则基类""" name: str + dices: Dict[str, str] = {} priority: int = 0 @abstractmethod diff --git a/src/hydrorollcore/typing.py b/src/hydrorollcore/typing.py index d22e3e18..333a763b 100644 --- a/src/hydrorollcore/typing.py +++ b/src/hydrorollcore/typing.py @@ -1,5 +1,6 @@ from pydantic import BaseModel from typing import ( + Dict as Dict, TYPE_CHECKING as TYPE_CHECKING, TypeVar as TypeVar, Callable as Callable, 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 diff --git a/src/test.py b/src/test.py new file mode 100644 index 00000000..370adb41 --- /dev/null +++ b/src/test.py @@ -0,0 +1,4 @@ +import HydroRollCore + +client = HydroRollCore.Cli() +client.parse_args() |
