diff options
| author | 2023-11-03 03:28:35 +0800 | |
|---|---|---|
| committer | 2023-11-03 03:28:35 +0800 | |
| commit | 5a2033860a328c4116f0ede2874915315e7487b0 (patch) | |
| tree | f7b6d88d50ce33bd5fb52fbcfbca906738f412d6 /hydroroll/__init__.py | |
| parent | 4bf6db5200affc2f623aa02301020092c0789d19 (diff) | |
| download | HydroRoll-5a2033860a328c4116f0ede2874915315e7487b0.tar.gz HydroRoll-5a2033860a328c4116f0ede2874915315e7487b0.zip | |
Co-authored-by: HadalFauna <HadalFauna@users.noreply.github.com>
Diffstat (limited to 'hydroroll/__init__.py')
| -rw-r--r-- | hydroroll/__init__.py | 221 |
1 files changed, 0 insertions, 221 deletions
diff --git a/hydroroll/__init__.py b/hydroroll/__init__.py deleted file mode 100644 index b064bca..0000000 --- a/hydroroll/__init__.py +++ /dev/null @@ -1,221 +0,0 @@ -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 - -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 |
