ayaka.state

  1import inspect
  2import re
  3from typing import TYPE_CHECKING, List, Literal, Union, Awaitable, Callable
  4from typing_extensions import Self
  5import asyncio
  6import datetime
  7from loguru import logger
  8
  9from .config import ayaka_root_config
 10from .constant import _enter_exit_during
 11from .depend import AyakaDepend, AyakaInput
 12
 13
 14if TYPE_CHECKING:
 15    from .ayaka import AyakaApp
 16
 17
 18def add_flag():
 19    _enter_exit_during.set(_enter_exit_during.get()+1)
 20
 21
 22def sub_flag():
 23    _enter_exit_during.set(_enter_exit_during.get()-1)
 24
 25
 26def ensure_regex(data: Union[str, re.Pattern], escape=True):
 27    if isinstance(data, str):
 28        if escape:
 29            data = re.escape(data)
 30        data = re.compile(data)
 31    return data
 32
 33
 34def ensure_regex_list(cmds: List[Union[str, re.Pattern]], escape=True):
 35    _cmds: List[re.Pattern] = []
 36    for cmd in cmds:
 37        _cmds.append(ensure_regex(cmd, escape))
 38    return _cmds
 39
 40
 41class AyakaState:
 42    def __init__(self, key="root", parent: Self = None):
 43        self.key = key
 44        self.parent = parent
 45        if not parent:
 46            self.keys = [key]
 47        else:
 48            self.keys = [*parent.keys, key]
 49        self.children: List[Self] = []
 50
 51        self.enter_funcs = []
 52        self.exit_funcs = []
 53        self.triggers: List[AyakaTrigger] = []
 54
 55    def __getitem__(self, k):
 56        if isinstance(k, slice):
 57            s = AyakaState("")
 58            s.keys = self.keys[k]
 59            if s.keys:
 60                s.key = s.keys[-1]
 61            return s
 62
 63        if isinstance(k, int):
 64            s = AyakaState(self.keys[k])
 65            return s
 66
 67        for node in self.children:
 68            if node.key == k:
 69                return node
 70        node = self.__class__(k, self)
 71        self.children.append(node)
 72        return node
 73
 74    def __getattr__(self, k: str):
 75        return self[k]
 76
 77    def __iter__(self):
 78        return iter(self.children)
 79
 80    def __str__(self) -> str:
 81        return ayaka_root_config.state_separate.join(self.keys)
 82
 83    def __repr__(self) -> str:
 84        return f"{self.__class__.__name__}({self})"
 85
 86    def join(self, *keys: str) -> Self:
 87        node = self
 88        for key in keys:
 89            node = node[key]
 90        return node
 91
 92    def belong(self, node: Self):
 93        if len(self.keys) < len(node.keys):
 94            return False
 95
 96        for i in range(len(node.keys)):
 97            if node.keys[i] != self.keys[i]:
 98                return False
 99
100        return True
101
102    def __ge__(self, node: Self):
103        return self.belong(node)
104
105    def __le__(self, node: Self):
106        return node.belong(self)
107
108    def __gt__(self, node: Self):
109        return self >= node and len(self.keys) > len(node.keys)
110
111    def __lt__(self, node: Self):
112        return self <= node and len(self.keys) < len(node.keys)
113
114    def __eq__(self, node: Self):
115        return self <= node and len(self.keys) == len(node.keys)
116
117    def dict(self):
118        data = [child.dict() for child in self.children]
119        return {
120            "name": self.key,
121            "triggers": self.triggers,
122            "children": data,
123        }
124
125    async def enter(self):
126        if ayaka_root_config.debug:
127            print(">>>", self.key)
128        add_flag()
129        for func in self.enter_funcs:
130            await func()
131        sub_flag()
132
133    async def exit(self):
134        if ayaka_root_config.debug:
135            print("<<<", self.key)
136        add_flag()
137        for func in self.exit_funcs:
138            await func()
139        sub_flag()
140
141    def on_enter(self):
142        def decorator(func):
143            self.enter_funcs.append(func)
144            return func
145        return decorator
146
147    def on_exit(self):
148        def decorator(func):
149            self.exit_funcs.append(func)
150            return func
151        return decorator
152
153    def on_cmd(self, cmds: List[Union[str, re.Pattern]], app: "AyakaApp", deep: Union[int, Literal["all"]] = 0, block=True):
154        _cmds = ensure_regex_list(cmds)
155
156        def decorator(func):
157            t = AyakaTrigger(func, _cmds, deep, app, block, self)
158            self.triggers.append(t)
159            return func
160        return decorator
161
162    def on_text(self, app: "AyakaApp", deep: Union[int, Literal["all"]] = 0, block=True):
163        return self.on_cmd([], app, deep, block)
164
165
166class AyakaTrigger:
167    def __init__(self, func: Callable[..., Awaitable], cmds: List[re.Pattern], deep: Union[int, Literal["all"]], app: "AyakaApp", block: bool, state: AyakaState):
168        raw_cmds: List[str] = [cmd.pattern for cmd in cmds]
169        self.func = func
170        self.cmds = cmds
171        self.raw_cmds = raw_cmds
172        self.deep = deep
173        self.app = app
174        self.block = block
175        self.state = state
176
177        # 默认没有解析模型
178        models: List[AyakaInput] = []
179        sig = inspect.signature(func)
180        for k, v in sig.parameters.items():
181            cls = v.annotation
182            if issubclass(cls, AyakaInput):
183                models.append(cls)
184
185        # 生成帮助
186        doc = "" if not func.__doc__ else f"| {func.__doc__}"
187        cmd_str = '/'.join(raw_cmds) if raw_cmds else "<任意文字>"
188
189        if not models:
190            help = f"- {cmd_str} {doc}"
191        else:
192            helps = []
193            for model in models:
194                data = model.help()
195                keys_str = " ".join(f"<{k}>" for k in data.keys())
196                data_str = "\n".join(f"    <{k}> {v}" for k, v in data.items())
197                helps.append(f"- {cmd_str} {keys_str} {doc}\n{data_str}")
198            help = "\n".join(helps)
199
200        self.help = help
201        if len(state.keys) > 1:
202            s = str(state[1:])
203            if s not in self.app.state_helps:
204                self.app.state_helps[s] = []
205            self.app.state_helps[s].append(help)
206        else:
207            self.app.idle_helps.append(help)
208
209        if ayaka_root_config.debug:
210            print(repr(self))
211
212    async def run(self):
213        params = {}
214        sig = inspect.signature(self.func)
215
216        for k, v in sig.parameters.items():
217            cls = v.annotation
218            if issubclass(cls, AyakaDepend):
219                d = await cls.create_by_app(self.app)
220                if not d:
221                    return False
222                params[k] = d
223
224        await self.func(**params)
225        return True
226
227    def __repr__(self) -> str:
228        return f"{self.__class__.__name__}({self})"
229
230    def __str__(self) -> str:
231        data = {
232            "func": self.func.__name__,
233            "app_name": self.app.name,
234            "cmds": self.raw_cmds,
235            "deep": self.deep,
236            "block": self.block
237        }
238        return " ".join(f"{k}={v}" for k, v in data.items())
239
240
241class AyakaTimer:
242    def __repr__(self) -> str:
243        return f"AyakaTimer({self.app.name}, {self.gap}, {self.func.__name__})"
244
245    def __init__(self,  app: "AyakaApp", gap: int, h: int, m: int, s: int, func, show=True) -> None:
246        self.app = app
247        self.h = h
248        self.m = m
249        self.s = s
250        self.gap = gap
251        self.func = func
252        self.show = show
253        if ayaka_root_config.debug:
254            print(self)
255
256    def start(self):
257        asyncio.create_task(self.run_forever())
258
259    async def run_forever(self):
260        # 有启动时间点要求的
261        time_i = int(datetime.datetime.now().timestamp())
262        if self.h >= 0:
263            _time_i = self.h*3600+self.m*60+self.s
264            # 移除时区偏差
265            time_i -= 57600
266            gap = 86400 - (time_i - _time_i) % 86400
267            await asyncio.sleep(gap)
268        elif self.m >= 0:
269            _time_i = self.m*60+self.s
270            gap = 3600 - (time_i-_time_i) % 3600
271            await asyncio.sleep(gap)
272        elif self.s >= 0:
273            _time_i = self.s
274            gap = 60 - (time_i-_time_i) % 60
275            await asyncio.sleep(gap)
276
277        while True:
278            if self.show:
279                logger.opt(colors=True).debug(
280                    f"定时任务 | 插件：<y>{self.app.name}</y> | 回调：<c>{self.func.__name__}</c>")
281            asyncio.create_task(self.func())
282            await asyncio.sleep(self.gap)
283
284
285root_state = AyakaState()