ayaka.on

  1'''注册回调[旧API]'''
  2import asyncio
  3import datetime
  4from typing import TYPE_CHECKING
  5from loguru import logger
  6from .config import ayaka_root_config
  7from .state import root_state
  8
  9if TYPE_CHECKING:
 10    from .ayaka import AyakaApp
 11
 12
 13class AyakaOn:
 14    def __init__(self, app: "AyakaApp") -> None:
 15        self.app = app
 16
 17    def state(self, *states: str):
 18        '''注册有状态回调'''
 19        if "*" in str(states):
 20            def decorator(func):
 21                func = self.app.on_deep_all("all")(func)
 22                func = self.app.on_state()(func)
 23                return func
 24            return decorator
 25        states = [s.split(".") for s in states]
 26        return self.app.on_state(*states)
 27
 28    def idle(self, super=False):
 29        '''注册无状态回调'''
 30        if super:
 31            def decorator(func):
 32                func = self.app.on_state(root_state)(func)
 33                func = self.app.on_deep_all("all")(func)
 34                return func
 35            return decorator
 36        return self.app.on_state(root_state)
 37
 38    def command(self, *cmds: str):
 39        return self.app.on_cmd(*cmds)
 40
 41    def text(self):
 42        return self.app.on_no_block()
 43
 44    def everyday(self, h: int, m: int, s: int):
 45        '''每日定时触发'''
 46        return self.interval(86400, h, m, s)
 47
 48    def interval(self, gap: int, h=-1, m=-1, s=-1, show=True):
 49        '''在指定的时间点后循环触发'''
 50        return self.on_timer(gap, h, m, s, show)
 51
 52    def on_timer(self, gap: int, h: int, m: int, s: int, show=True):
 53        '''在指定的时间点后循环触发'''
 54        def decorator(func):
 55            t = AyakaTimer(self.app.name, gap, h, m, s, func, show)
 56            self.app.timers.append(t)
 57            return func
 58        return decorator
 59
 60
 61class AyakaTimer:
 62    def __repr__(self) -> str:
 63        return f"AyakaTimer({self.name}, {self.gap}, {self.func.__name__})"
 64
 65    def __init__(self, name: str, gap: int, h: int, m: int, s: int, func, show=True) -> None:
 66        self.name = name
 67        self.h = h
 68        self.m = m
 69        self.s = s
 70        self.gap = gap
 71        self.func = func
 72        self.show = show
 73        if ayaka_root_config.debug:
 74            print(self)
 75
 76    def start(self):
 77        asyncio.create_task(self.run_forever())
 78
 79    async def run_forever(self):
 80        # 有启动时间点要求的
 81        time_i = int(datetime.datetime.now().timestamp())
 82        if self.h >= 0:
 83            _time_i = self.h*3600+self.m*60+self.s
 84            # 移除时区偏差
 85            time_i -= 57600
 86            gap = 86400 - (time_i - _time_i) % 86400
 87            await asyncio.sleep(gap)
 88        elif self.m >= 0:
 89            _time_i = self.m*60+self.s
 90            gap = 3600 - (time_i-_time_i) % 3600
 91            await asyncio.sleep(gap)
 92        elif self.s >= 0:
 93            _time_i = self.s
 94            gap = 60 - (time_i-_time_i) % 60
 95            await asyncio.sleep(gap)
 96
 97        while True:
 98            if self.show:
 99                logger.opt(colors=True).debug(
100                    f"定时任务 | 插件：<y>{self.name}</y> | 回调：<c>{self.func.__name__}</c>")
101            asyncio.create_task(self.func())
102            await asyncio.sleep(self.gap)