ayaka.ayaka

  1'''ayaka核心'''
  2import asyncio
  3import datetime
  4import inspect
  5import json
  6from math import ceil
  7from pathlib import Path
  8from loguru import logger
  9from typing import List, Dict, Literal, Union
 10from .depend import AyakaCache
 11from .config import ayaka_root_config, ayaka_data_path
 12from .constant import _bot, _event, _group, _arg, _args, _message, _cmd, _enter_exit_during, app_list, group_list, bot_list, private_listener_dict
 13from .driver import on_message, get_driver, Message, MessageSegment, Bot, MessageEvent, GroupMessageEvent
 14from .state import AyakaState, AyakaTrigger, root_state
 15from .on import AyakaOn, AyakaTimer
 16
 17
 18class AyakaGroup:
 19    def __repr__(self) -> str:
 20        return f"{self.__class__.__name__}({self.bot_id}, {self.group_id}, {self.apps})"
 21
 22    def __init__(self, bot_id: int, group_id: int) -> None:
 23        self.bot_id = bot_id
 24        self.group_id = group_id
 25        self.state = root_state
 26
 27        # 添加app，并分配独立数据空间
 28        self.apps: List["AyakaApp"] = []
 29        self.cache_dict: Dict[str, Dict[str, AyakaCache]] = {}
 30        for app in app_list:
 31            # if app.name not in forbid_names:
 32            self.apps.append(app)
 33            self.cache_dict[app.name] = {}
 34
 35        group_list.append(self)
 36
 37        if ayaka_root_config.debug:
 38            print(self)
 39
 40    async def back(self):
 41        if self.state.parent:
 42            await self.state.exit()
 43            self.state = self.state.parent
 44            return self.state
 45
 46    async def goto(self, state: AyakaState):
 47        if _enter_exit_during.get() > 0:
 48            logger.warning("你正在AyakaState的enter/exit方法中进行状态转移，这可能会导致无法预料的错误")
 49
 50        keys = state.keys
 51
 52        # 找到第一个不同的结点
 53        n0 = len(keys)
 54        n1 = len(self.state.keys)
 55        n = min(n0, n1)
 56        for i in range(n):
 57            if keys[i] != self.state.keys[i]:
 58                break
 59        else:
 60            i += 1
 61
 62        # 回退
 63        for j in range(i, n1):
 64            await self.back()
 65        keys = keys[i:]
 66
 67        # 重新出发
 68        for key in keys:
 69            self.state = self.state[key]
 70            await self.state.enter()
 71        logger.opt(colors=True).debug(f"状态：<c>{self.state}</c>")
 72        return self.state
 73
 74    def get_app(self, name: str):
 75        '''根据app名获取该group所启用的app，不存在则返回None'''
 76        for app in self.apps:
 77            if app.name == name:
 78                return app
 79
 80
 81class AyakaApp:
 82    def __repr__(self) -> str:
 83        return f"{self.__class__.__name__}({self.name})"
 84
 85    def __init__(self, name: str) -> None:
 86        self.path = Path(inspect.stack()[1].filename)
 87
 88        for app in app_list:
 89            if app.name == name:
 90                raise Exception(
 91                    f"应用{app.name} 重复注册，已忽略注册时间更晚的应用！\n{app.path}(最早注册)\n{self.path}(被忽略)")
 92
 93        self.name = name
 94        self.ayaka_root_config = ayaka_root_config
 95        self.funcs = []
 96        self.on = AyakaOn(self)
 97        self.timers: List[AyakaTimer] = []
 98        self.root_state = root_state
 99        self._intro = "没有介绍"
100        self.state_helps: Dict[str, List[str]] = {}
101        self.idle_helps: List[str] = []
102
103        logger.opt(colors=True).success(f"应用加载成功 \"<c>{name}</c>\"")
104        app_list.append(self)
105        if ayaka_root_config.debug:
106            print(self)
107
108    @property
109    def intro(self):
110        help = self._intro
111        for h in self.idle_helps:
112            help += "\n" + h
113        return help
114
115    @property
116    def all_help(self):
117        help = self.intro
118        for s, hs in self.state_helps.items():
119            help += f"\n[{s}]"
120            for h in hs:
121                help += "\n" + h
122        return help
123
124    @property
125    def help(self):
126        '''获取当前状态下的帮助，没有找到则返回介绍'''
127        total_triggers = get_cascade_triggers(self.state)
128
129        helps = []
130        cmds = []
131        for ts in total_triggers:
132            flag = 1
133            for t in ts:
134                if t.app == self:
135                    for c in t.cmds:
136                        if c in cmds:
137                            break
138                    else:
139                        if flag:
140                            helps.append(f"[{t.state[1:]}]")
141                            flag = 0
142                        cmds.extend(t.cmds)
143                        helps.append(t.help)
144
145        if not helps:
146            return self.intro
147        return "\n".join(helps)
148
149    @help.setter
150    def help(self, help: str):
151        self._intro = help
152
153    @property
154    def user_name(self):
155        '''*timer触发时不可用*
156
157        当前消息的发送人的群名片或昵称
158        '''
159        s = self.event.sender
160        name = s.card or s.nickname
161        return name
162
163    @property
164    def user_id(self):
165        '''*timer触发时不可用*
166
167        当前消息的发送人的uid
168        '''
169        return self.event.user_id
170
171    @property
172    def bot(self):
173        '''*timer触发时不可用*
174
175        当前bot
176        '''
177        return _bot.get()
178
179    @property
180    def event(self):
181        '''*timer触发时不可用*
182
183        当前消息
184        '''
185        return _event.get()
186
187    @property
188    def cache(self):
189        '''*timer触发时不可用*
190
191        当前群组的缓存空间'''
192        return self.group.cache_dict[self.name]
193
194    @property
195    def group_id(self):
196        '''*timer触发时不可用*
197
198        当前群组的id
199
200        注：若群聊A正监听私聊B，当私聊B发送消息触发插件回调时，该属性仍可正确返回群聊A的id
201        '''
202        return self.group.group_id
203
204    @property
205    def bot_id(self):
206        '''*timer触发时不可用*
207
208        当前bot的id
209        '''
210        return self.group.bot_id
211
212    @property
213    def group(self):
214        '''*timer触发时不可用*
215
216        当前群组
217
218        注：若群聊A正监听私聊B，当私聊B发送消息触发插件回调时，该属性仍可正确返回群聊A
219        '''
220        return _group.get()
221
222    @property
223    def arg(self):
224        '''*timer触发时不可用*
225
226        当前消息在移除了命令后的剩余部分
227        '''
228        return _arg.get()
229
230    @property
231    def args(self):
232        '''*timer触发时不可用*
233
234        当前消息在移除了命令后，剩余部分按照空格分割后的数组
235
236        注：除了文字消息外，其他消息类型将自动分割，例如一串qq表情会被分割为多个元素
237        '''
238        return _args.get()
239
240    @property
241    def cmd(self):
242        '''*timer触发时不可用*
243
244        当前消息的命令头
245        '''
246        return _cmd.get()
247
248    @property
249    def message(self):
250        '''*timer触发时不可用*
251
252        当前消息
253        '''
254        return _message.get()
255
256    @property
257    def state(self):
258        return self.group.state
259
260    def get_state(self, *keys: str):
261        '''
262            假设当前状态为 `root.test.a` 即 `根.插件名.一级菜单项`
263
264            >>> get_state(key1, key2) -> [root.test].key1.key2
265
266            特别的，keys可以为空，例如：
267
268            >>> get_state() -> [root.test]
269        '''
270        keys = [self.name, *keys]
271        return root_state.join(*keys)
272
273    async def set_state(self, state_or_key: Union[AyakaState, str], *keys: str):
274        return await self.goto(state_or_key, *keys)
275
276    async def goto(self, state_or_key: Union[AyakaState, str], *keys: str):
277        '''当state_or_key为字符串时，调用get_state进行初始化(state_or_key.keys1.keys2)，keys可选填'''
278        if isinstance(state_or_key, AyakaState):
279            state = state_or_key
280        else:
281            state = self.get_state(*state_or_key.split("."), *keys)
282        return await self.group.goto(state)
283
284    async def back(self):
285        return await self.group.back()
286
287    def _add_func(self, func):
288        if func not in self.funcs:
289            self.funcs.append(func)
290
291    def on_deep_all(self, deep: Union[int, Literal["all"]] = "all"):
292        '''注册深度监听'''
293        def decorator(func):
294            func.deep = deep
295            self._add_func(func)
296            return func
297        return decorator
298
299    def on_no_block(self, block: bool = False):
300        '''注册非阻断'''
301        def decorator(func):
302            func.block = block
303            self._add_func(func)
304            return func
305        return decorator
306
307    def on_cmd(self, *cmds: str):
308        '''注册命令触发，不填写命令则视为文本消息'''
309        def decorator(func):
310            func.cmds = cmds
311            self._add_func(func)
312            return func
313        return decorator
314
315    def on_state(self, *states: Union[AyakaState, str, List[str]]):
316        '''注册有状态响应，不填写states则为root.插件名状态'''
317        _states = []
318        if not states:
319            states = [[]]
320        for s in states:
321            if isinstance(s, str):
322                s = self.get_state(s)
323            elif isinstance(s, list):
324                s = self.get_state(*s)
325            _states.append(s)
326
327        def decorator(func):
328            func.states = _states
329            self._add_func(func)
330            return func
331        return decorator
332
333    def set_start_cmds(self, *cmds: str):
334        '''设置应用启动命令，当然，你也可以通过on_cmd自定义启动方式'''
335        async def func():
336            '''打开应用'''
337            await self.start()
338        func.cmds = cmds
339        self.funcs.append(func)
340
341    async def start(self, state: str = None):
342        '''*timer触发时不可用*
343
344        启动应用，并发送提示
345
346        state参数为兼容旧API'''
347        if not state:
348            state = self.get_state()
349            await self.goto(state)
350        else:
351            states = state.split(".")
352            await self.goto(*states)
353        await self.send(f"已打开应用 [{self.name}]")
354
355    async def close(self):
356        '''*timer触发时不可用*
357
358        关闭应用，并发送提示'''
359        await self.goto(root_state)
360        await self.send(f"已关闭应用 [{self.name}]")
361
362    def add_listener(self, user_id: int):
363        '''为该群组添加对指定私聊的监听'''
364        private_listener_dict[user_id].append(self.group_id)
365
366    def remove_listener(self, user_id: int = 0):
367        '''默认移除该群组对其他私聊的所有监听'''
368        id = self.group_id
369
370        if user_id == 0:
371            for ids in private_listener_dict.values():
372                if id in ids:
373                    ids.remove(id)
374            return
375
376        if id in private_listener_dict[user_id]:
377            private_listener_dict[user_id].remove(self.group_id)
378
379    async def send(self, message):
380        '''发送消息，消息的类型可以是 Message | MessageSegment | str'''
381        # 这里不使用event，因为一些event可能来自其他设备的监听传递
382        await self.bot.send_group_msg(group_id=self.group_id, message=message)
383
384    def pack_messages(self, bot_id, messages):
385        '''转换为cqhttp node格式'''
386        data: List[MessageSegment] = []
387        for m in messages:
388            if isinstance(m, MessageSegment) and m.type == "node":
389                data.append(m)
390            else:
391                m = MessageSegment.node_custom(
392                    user_id=bot_id,
393                    nickname="Ayaka Bot",
394                    content=str(m)
395                )
396                data.append(m)
397        return data
398
399    async def send_many(self, messages):
400        '''发送合并转发消息，消息的类型可以是 List[Message | MessageSegment | str]'''
401        # 分割长消息组（不可超过100条
402        div_len = 100
403        div_cnt = ceil(len(messages) / div_len)
404        for i in range(div_cnt):
405            msgs = self.pack_messages(
406                self.bot_id,
407                messages[i*div_len: (i+1)*div_len]
408            )
409            await self.bot.call_api("send_group_forward_msg", group_id=self.group_id, messages=msgs)
410
411    def t_check(self, bot_id: int):
412        # 未连接
413        bot = get_bot(bot_id)
414        if not bot:
415            logger.warning(f"BOT({bot_id}) 未连接")
416            return
417
418        return bot
419
420    async def t_send(self, bot_id: int, group_id: int, message):
421        '''timer触发回调时，想要发送消息必须使用该方法，一些上下文亦无法使用'''
422        bot = self.t_check(bot_id)
423        if not bot:
424            return
425
426        await bot.send_group_msg(group_id=group_id, message=message)
427
428    async def t_send_many(self, bot_id: int, group_id: int, messages):
429        '''timer触发回调时，想要发送消息必须使用该方法，一些上下文亦无法使用'''
430        bot = self.t_check(bot_id)
431        if not bot:
432            return
433
434        # 分割长消息组（不可超过100条）谨慎起见，使用80作为单元长度
435        div_len = 80
436        div_cnt = ceil(len(messages) / div_len)
437        for i in range(div_cnt):
438            msgs = self.pack_messages(
439                bot_id,
440                messages[i*div_len: (i+1)*div_len]
441            )
442            await bot.call_api("send_group_forward_msg", group_id=group_id, messages=msgs)
443
444
445def get_bot(bot_id: int):
446    '''获取已连接的bot'''
447    bot_id = str(bot_id)
448    for bot in bot_list:
449        if bot.self_id == bot_id:
450            return bot
451
452
453def get_group(bot_id: int, group_id: int):
454    '''获取对应的AyakaGroup对象，自动增加'''
455    for group in group_list:
456        if group.bot_id == bot_id and group.group_id == group_id:
457            break
458    else:
459        group = AyakaGroup(bot_id, group_id)
460    return group
461
462
463def get_app(app_name: str):
464    for app in app_list:
465        if app.name == app_name:
466            return app
467
468
469async def deal_event(bot: Bot, event: MessageEvent):
470    '''处理收到的消息，将其分割为cmd和args，设置上下文相关变量的值，并将消息传递给对应的群组'''
471    if ayaka_root_config.exclude_old_msg:
472        time_i = int(datetime.datetime.now().timestamp())
473        if event.time < time_i - 60:
474            return
475
476    _bot.set(bot)
477    _event.set(event)
478
479    bot_id = int(bot.self_id)
480
481    if isinstance(event, GroupMessageEvent):
482        group_id = event.group_id
483        await deal_group(bot_id, group_id)
484
485    else:
486        id = event.user_id
487        group_ids = private_listener_dict.get(id, [])
488        ts = [asyncio.create_task(deal_group(bot_id, group_id))
489              for group_id in group_ids]
490        await asyncio.gather(*ts)
491
492
493async def deal_group(bot_id: int, group_id: int):
494    prefix = ayaka_root_config.prefix
495
496    # 群组
497    group = get_group(bot_id, group_id)
498    _group.set(group)
499
500    # 消息
501    message = _event.get().message
502    _message.set(message)
503
504    # 从子状态开始向上查找可用的触发
505    state = group.state
506    cascade_triggers = get_cascade_triggers(state, 0)
507
508    # 命令
509    # 消息前缀文本
510    first = get_first(message)
511    if first.startswith(prefix):
512        first = first[len(prefix):]
513        for ts in cascade_triggers:
514            if await deal_cmd_triggers(ts, message, first, state):
515                return
516
517    # 命令退化成消息
518    for ts in cascade_triggers:
519        if await deal_text_triggers(ts, message, state):
520            return
521
522
523async def deal_cmd_triggers(triggers: List[AyakaTrigger], message: Message, first: str, state: AyakaState):
524    sep = ayaka_root_config.separate
525
526    # 命令
527    temp = [t for t in triggers if t.cmds]
528    # 根据命令长度排序，长命令优先级更高
529    cmd_ts = [(c, t) for t in temp for c in t.cmds]
530    cmd_ts.sort(key=lambda x: len(x[0]), reverse=1)
531
532    for c, t in cmd_ts:
533        if first.startswith(c):
534            # 设置上下文
535            # 设置命令
536            _cmd.set(c)
537            # 设置参数
538            left = first[len(c):].lstrip(sep)
539            if left:
540                arg = Message([MessageSegment.text(left), *message[1:]])
541            else:
542                arg = Message(message[1:])
543            _arg.set(arg)
544            _args.set(divide_message(arg))
545
546            # 触发
547            log_trigger(c, t.app.name, state, t.func.__name__)
548            await t.run()
549
550            # 阻断后续
551            if t.block:
552                return True
553
554
555async def deal_text_triggers(triggers: List[AyakaTrigger], message: Message, state: AyakaState):
556    # 消息
557    text_ts = [t for t in triggers if not t.cmds]
558
559    # 设置上下文
560    # 设置命令
561    _cmd.set("")
562    # 设置参数
563    _arg.set(message)
564    _args.set(divide_message(message))
565
566    for t in text_ts:
567        # 触发
568        log_trigger("", t.app.name, state, t.func.__name__)
569        await t.run()
570
571        # 阻断后续
572        if t.block:
573            return True
574
575
576def get_cascade_triggers(state: AyakaState, deep: int = 0):
577    # 根据深度筛选funcs
578    ts = [
579        t for t in state.triggers
580        if t.deep == "all" or t.deep >= deep
581    ]
582    cascade_triggers = [ts]
583
584    # 获取父状态的方法
585    if state.parent:
586        cascade_triggers.extend(get_cascade_triggers(state.parent, deep+1))
587
588    # 排除空项
589    cascade_triggers = [ts for ts in cascade_triggers if ts]
590    return cascade_triggers
591
592
593def log_trigger(cmd, app_name, state, func_name):
594    '''日志记录'''
595    items = []
596    items.append(f"状态：<c>{state}</c>")
597    items.append(f"应用：<y>{app_name}</y>")
598    if cmd:
599        items.append(f"命令：<y>{cmd}</y>")
600    else:
601        items.append("命令：<g>无</g>")
602    items.append(f"回调：<c>{func_name}</c>")
603    info = " | ".join(items)
604    logger.opt(colors=True).debug(info)
605
606
607def get_first(message: Message):
608    first = ""
609    for m in message:
610        if m.type == "text":
611            first += str(m)
612        else:
613            break
614    return first
615
616
617def divide_message(message: Message) -> List[MessageSegment]:
618    args = []
619    sep = ayaka_root_config.separate
620
621    for m in message:
622        if m.is_text():
623            ss = str(m).split(sep)
624            args.extend(MessageSegment.text(s) for s in ss if s)
625        else:
626            args.append(m)
627
628    return args
629
630
631def regist_func(app: AyakaApp, func):
632    '''注册回调'''
633    # 默认是无状态应用，从root开始触发
634    states: List[AyakaState] = getattr(func, "states", [root_state])
635    # 默认是消息响应
636    cmds: List[str] = getattr(func, "cmds", [])
637    # 默认监听深度为0
638    deep: int = getattr(func, "deep", 0)
639    # 默认阻断
640    block: bool = getattr(func, "block", True)
641
642    # 注册
643    for s in states:
644        s.on_cmd(cmds, app, deep, block)(func)
645
646    return func
647
648
649driver = get_driver()
650
651
652@driver.on_startup
653async def startup():
654    # 注册所有回调
655    for app in app_list:
656        for func in app.funcs:
657            regist_func(app, func)
658
659    if ayaka_root_config.debug:
660        s = json.dumps(
661            root_state.dict(), ensure_ascii=0,
662            indent=4, default=repr
663        )
664        path = ayaka_data_path / "all_state.json"
665        with path.open("w+", encoding="utf8") as f:
666            f.write(s)
667
668on_message(priority=20, block=False, handlers=[deal_event])