ayaka.on

  1'''注册回调'''
  2import asyncio
  3import datetime
  4from typing import Callable, Coroutine, TYPE_CHECKING, Union, List
  5from loguru import logger
  6from .config import INIT_STATE, ayaka_root_config
  7
  8if TYPE_CHECKING:
  9    from .ayaka import AyakaApp
 10
 11
 12class AyakaOn:
 13    def __init__(self, app: "AyakaApp") -> None:
 14        self.app = app
 15
 16    def everyday(self, h: int, m: int, s: int):
 17        '''每日定时触发'''
 18        return self.interval(86400, h, m, s)
 19
 20    def interval(self, gap: int, h=-1, m=-1, s=-1):
 21        '''在指定的时间点后循环触发'''
 22        return self.on_timer(gap, h, m, s)
 23
 24    def state(self, *states: str):
 25        '''注册有状态回调'''
 26        if not states:
 27            states = INIT_STATE
 28
 29        def decorator(func):
 30            # 取出之前存的参数
 31            return self.on_handle(func.cmds, states, False)(func)
 32        return decorator
 33
 34    def idle(self, super=False):
 35        '''注册无状态回调'''
 36        def decorator(func):
 37            # 取出之前存的参数
 38            return self.on_handle(func.cmds, None, super)(func)
 39        return decorator
 40
 41    def command(self, *cmds: str):
 42        def decorator(func):
 43            func.cmds = cmds
 44            return func
 45        return decorator
 46
 47    def text(self):
 48        def decorator(func):
 49            func.cmds = ""
 50            return func
 51        return decorator
 52
 53    def on_handle(self, cmds: Union[List[str], str], states: Union[List[str], str], super: bool):
 54        '''注册'''
 55        cmds = ensure_list(cmds)
 56        states = ensure_list(states)
 57
 58        def decorator(func: Callable[[], Coroutine]):
 59            for state in states:
 60                for cmd in cmds:
 61                    t = AyakaTrigger(self.app.name, cmd, state, super, func)
 62                    self.app.triggers.append(t)
 63
 64                # 如果有帮助，自动添加到_help中
 65                doc = func.__doc__
 66                if doc:
 67                    if state is None:
 68                        state = INIT_STATE
 69                    if state not in self.app._help:
 70                        self.app._help[state] = []
 71                    cmd_str = '/'.join(cmds)
 72                    if not cmd_str:
 73                        cmd_str = "*"
 74                    self.app._help[state].append(f"- {cmd_str} {doc}")
 75
 76            return func
 77        return decorator
 78
 79    def on_timer(self, gap: int, h: int, m: int, s: int):
 80        '''在指定的时间点后循环触发'''
 81        def decorator(func):
 82            t = AyakaTimer(self.app.name, gap, h, m, s, func)
 83            self.app.timers.append(t)
 84            return func
 85        return decorator
 86
 87
 88class AyakaTrigger:
 89    def __repr__(self) -> str:
 90        return f"AyakaTrigger({self.app_name}, {self.cmd}, {self.state}, {self.super}, {self.func.__name__})"
 91
 92    def __init__(self, app_name, cmd, state, super, func) -> None:
 93        self.app_name = app_name
 94        self.cmd = cmd
 95        self.state = state
 96        self.super = super
 97        self.func = func
 98        if ayaka_root_config.debug:
 99            print(self)
100
101    async def run(self):
102        # 日志记录
103        items = []
104
105        if self.cmd:
106            items.append("<y>命令</y>")
107        else:
108            items.append("<g>消息</g>")
109
110        app_name = f"<y>{self.app_name}</y>"
111        if self.state is not None:
112            app_name += f" <g>{self.state}</g>"
113        items.append(app_name)
114
115        if self.cmd:
116            items.append(f"<y>{self.cmd}</y>")
117
118        items.append(f"执行回调 <c>{self.func.__name__}</c>")
119        info = " | ".join(items)
120        logger.opt(colors=True).debug(info)
121
122        # 运行回调
123        await self.func()
124
125
126class AyakaTimer:
127    def __repr__(self) -> str:
128        return f"AyakaTimer({self.name}, {self.gap}, {self.func.__name__})"
129
130    def __init__(self, name: str, gap: int, h: int, m: int, s: int, func) -> None:
131        self.name = name
132        self.h = h
133        self.m = m
134        self.s = s
135        self.gap = gap
136        self.func = func
137        if ayaka_root_config.debug:
138            print(self)
139
140    def start(self):
141        asyncio.create_task(self.run_forever())
142
143    async def run_forever(self):
144        # 有启动时间点要求的
145        time_i = int(datetime.datetime.now().timestamp())
146        if self.h >= 0:
147            _time_i = self.h*3600+self.m*60+self.s
148            # 移除时区偏差
149            time_i -= 57600
150            gap = 86400 - (time_i - _time_i) % 86400
151            await asyncio.sleep(gap)
152        elif self.m >= 0:
153            _time_i = self.m*60+self.s
154            gap = 3600 - (time_i-_time_i) % 3600
155            await asyncio.sleep(gap)
156        elif self.s >= 0:
157            _time_i = self.s
158            gap = 60 - (time_i-_time_i) % 60
159            await asyncio.sleep(gap)
160
161        while True:
162            logger.opt(colors=True).debug(f"触发定时任务 <y>{self.name}</y>")
163            asyncio.create_task(self.func())
164            await asyncio.sleep(self.gap)
165
166
167def ensure_list(items):
168    if isinstance(items, tuple):
169        return [item for item in items]
170    if not isinstance(items, list):
171        return [items]
172    return items