diff options
Diffstat (limited to 'hydroroll/__init__.py')
| -rw-r--r-- | hydroroll/__init__.py | 223 |
1 files changed, 220 insertions, 3 deletions
diff --git a/hydroroll/__init__.py b/hydroroll/__init__.py index c72c991..b064bca 100644 --- a/hydroroll/__init__.py +++ b/hydroroll/__init__.py @@ -1,4 +1,221 @@ -name = "hydroroll" +from collections import defaultdict +from itertools import chain +from pathlib import Path +from typing import Dict, List, Optional, Type, Union +from iamai import ConfigModel, Plugin +from iamai.log import logger, error_or_exception +from iamai.utils import ModulePathFinder, get_classes_from_module_name, get_classes_from_dir, get_classes_from_module +from .config import GlobalConfig +from iamai.exceptions import LoadModuleError +from .utils import BasePlugin, RegexPluginBase, CommandPluginBase +import os +# from .core import Rule, RuleLoadType +from HydroRollCore import Rule, RuleLoadType -from hydroroll.bot import Bot -from hydroroll.config import GlobalConfig
\ No newline at end of file +class HydroRoll(Plugin): + + class Config(ConfigModel): + __config_name__ = "HydroRoll" + + priority = 0 + rules_priority_dict: Dict[int, List[Type[Rule]]] = defaultdict(list) + _module_path_finder = ModulePathFinder() + + + def _load_rule_class( + self, + rule_class: Type[Rule], + rule_load_type: RuleLoadType, + rule_file_path: Optional[str], + ): + logger.info(f'Loading rule from class "{rule_class!r}"') + """加载规则类。""" + priority = getattr(rule_class, "priority", None) + if type(priority) is int and priority >= 0: + for _rule in self.rules: + if _rule.__name__ == rule_class.__name__: + logger.warning( + f'Already have a same name rule "{_rule.__name__}"' + ) + rule_class.__rule_load_type__ = rule_load_type + rule_class.__rule_file_path__ = rule_file_path + self.rules_priority_dict[priority].append(rule_class) + logger.success( + f'Succeeded to load rule "{rule_class.__name__}" ' + f'from class "{rule_class!r}"' + ) + else: + error_or_exception( + f'Load rule from class "{rule_class!r}" failed:', + LoadModuleError( + f'Rule priority incorrect in the class "{rule_class!r}"' + ), + self.bot.config.bot.log.verbose_exception, + ) + + def _load_rules_from_module_name( + self, module_name: str, rule_load_type: RuleLoadType + ): + logger.info(f'Loading rules from module "{module_name}"') + """从模块名称中规则包模块。""" + try: + rule_classes = get_classes_from_module_name(module_name, Rule) + except ImportError as e: + error_or_exception( + f'Import module "{module_name}" failed:', + e, + self.bot.config.bot.log.verbose_exception, + ) + else: + for rule_class, module in rule_classes: + self._load_rule_class( + rule_class, + rule_load_type, + module.__file__, + ) + + def _load_rules( + self, + *rules: Union[Type[Plugin], str, Path], + rule_load_type: Optional[RuleLoadType] = None, + ): + """加载规则包。 + + Args: + *rules: 规则类、规则包模块名称或者规则包模块文件路径。类型可以是 `Type[Rule]`, `str` 或 `pathlib.Path`。 + 如果为 `Type[Rule]` 类型时，将作为规则类进行加载。 + 如果为 `str` 类型时，将作为规则包模块名称进行加载，格式和 Python `import` 语句相同。 + 例如：`path.of.rule`。 + 如果为 `pathlib.Path` 类型时，将作为规则包模块文件路径进行加载。 + 例如：`pathlib.Path("path/of/rule")`。 + rule_load_type: 规则加载类型，如果为 None 则自动判断，否则使用指定的类型。 + """ + logger.info("Loading rules...") + for rule_ in rules: + if isinstance(rule_, type): + if issubclass(rule_, Rule): + self._load_rule_class( + rule_, rule_load_type or RuleLoadType.CLASS, None + ) + else: + logger.error( + f'The rule class "{rule_!r}" must be a subclass of Rule' + ) + elif isinstance(rule_, str): + logger.warning(f'Loading rules from module "{rule_}"') + self._load_rules_from_module_name( + rule_, rule_load_type or RuleLoadType.NAME + ) + elif isinstance(rule_, Path): + logger.warning(f'Loading rules from path "{rule_}"') + if rule_.is_file(): + if rule_.suffix != ".py": + logger.error(f'The path "{rule_}" must endswith ".py"') + return + + rule_module_name = None + for path in self._module_path_finder.path: + try: + if rule_.stem == "__init__": + if rule_.resolve().parent.parent.samefile(Path(path)): + rule_module_name = rule_.resolve().parent.name + break + elif rule_.resolve().parent.samefile(Path(path)): + rule_module_name = rule_.stem + break + except OSError: + continue + if rule_module_name is None: + rel_path = rule_.resolve().relative_to(Path(".").resolve()) + if rel_path.stem == "__init__": + rule_module_name = ".".join(rel_path.parts[:-1]) + else: + rule_module_name = ".".join( + rel_path.parts[:-1] + (rel_path.stem,) + ) + + self._load_rules_from_module_name( + rule_module_name, rule_load_type or RuleLoadType.FILE + ) + else: + logger.error(f'The rule path "{rule_}" must be a file') + else: + logger.error(f"Type error: {rule_} can not be loaded as plugin") + + def load_rules(self, *rules: Union[Type[Rule], str, Path]): + """加载规则。 + + Args: + *rules: 规则类、规则包x模块名称或者规则包模块文件路径。类型可以是 `Type[Rule]`, `str` 或 `pathlib.Path`。 + 如果为 `Type[Rule]` 类型时，将作为规则类进行加载。 + 如果为 `str` 类型时，将作为规则包模块名称进行加载，格式和 Python `import` 语句相同。 + 例如：`path.of.rule`。 + 如果为 `pathlib.Path` 类型时，将作为规则包模块文件路径进行加载。 + 例如：`pathlib.Path("path/of/rule")`。 + """ + # self._extend_rules.extend(rules) + return self._load_rules(*rules) + + def _load_rules_from_dirs(self, *dirs: Path): + """从目录中加载规则包，以 `_` 开头的模块中的规则不会被导入。路径可以是相对路径或绝对路径。 + + Args: + *dirs: 储存包含规则的模块的模块路径。 + 例如：`pathlib.Path("path/of/rules/")` 。 + """ + logger.info("Loading rules from dirs...") + dirs = list(map(lambda x: str(x.resolve()), dirs)) # type: ignore maybe remove? + logger.warning(f'Loading rules from dirs "{", ".join(map(str, dirs))}"') + self._module_path_finder.path.extend(dirs) # type: ignore + # type: ignore + for rule_class, module in get_classes_from_dir(dirs, Rule): # type: ignore + self._load_rule_class(rule_class, RuleLoadType.DIR, module.__file__) + + def load_rules_from_dirs(self, *dirs: Path): + """从目录中加载规则，以 `_` 开头的模块中的规则不会被导入。路径可以是相对路径或绝对路径。 + + Args: + *dirs: 储存包含rule的模块的模块路径。 + 例如：`pathlib.Path("path/of/rules/")` 。 + """ + # self._extend_rule_dirs.extend(dirs) + self._load_rules_from_dirs(*dirs) + + @property + def rules(self) -> List[Type[Plugin]]: + """当前已经加载的规则包的列表。""" + return list(chain(*self.rules_priority_dict.values())) + + def __post_init__(self): + if not self.bot.global_state.get('init', False): + self.bot.global_state = dict() + self.bot.global_state['init'] = True + + self._load_rules_from_dirs(Path(os.path.join("\\".join(os.path.dirname(__file__).split('\\')[:-2]),"rules"))) #*self.config.rule['rule_dirs']) + # self._load_rules(*self.config.rule.rules) + + ... + + async def handle(self) -> None: + """ + @TODO: HydroRollCore should be able to handle all signals and tokens from Psi. + @BODY: HydroRollCore actives the rule-packages. + """ + + if self.event.message.get_plain_text() == '.core': + await self.event.reply("HydroRollCore is running.") + elif self.event.message.startswith('.show'): + try: + await self.event.reply(eval(self.event.message.get_plain_text()[6:])) + except Exception as e: + await self.event.reply(f"{e!r}") + + async def rule(self) -> bool: + """ + @TODO: Psi should be able to handle all message first. + @BODY: lexer module will return a list of tokens, parser module will parse the tokens into a tree, and executor module will execute the tokens with a stack with a bool return value. + """ + + if self.event.type != "message": + return False + return self.event.message.get_plain_text().startswith(".")
\ No newline at end of file |