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: Union[AyakaState, str, List[str]], *keys: str):
274        '''变更当前群组的状态，state可以是AyakaState、字符串或字符串列表，若字符串内包含.符号，还会自动对其进行分割'''
275        return await self.goto(state, *keys)
276
277    async def goto(self, state: Union[AyakaState, str, List[str]], *keys: str):
278        # keys为兼容旧API（0.5.2及以前
279        '''变更当前群组的状态，state可以是AyakaState、字符串或字符串列表，若字符串内包含.符号，还会自动对其进行分割'''
280        if isinstance(state, str):
281            if "." in state:
282                state = self.get_state(*state.split("."))
283            else:
284                state = self.get_state(state, *keys)
285        elif isinstance(state, list):
286            state = self.get_state(*state)
287        return await self.group.goto(state)
288
289    async def back(self):
290        '''回退当前群组的状态'''
291        return await self.group.back()
292
293    def _add_func(self, func):
294        '''如果不存在就加入self.funcs'''
295        if func not in self.funcs:
296            self.funcs.append(func)
297
298    def on_deep_all(self, deep: Union[int, Literal["all"]] = "all"):
299        '''注册深度监听'''
300        def decorator(func):
301            func.deep = deep
302            self._add_func(func)
303            return func
304        return decorator
305
306    def on_no_block(self, block: bool = False):
307        '''注册非阻断'''
308        def decorator(func):
309            func.block = block
310            self._add_func(func)
311            return func
312        return decorator
313
314    def on_cmd(self, *cmds: str):
315        '''注册命令触发，不填写命令则视为文本消息'''
316        def decorator(func):
317            func.cmds = cmds
318            self._add_func(func)
319            return func
320        return decorator
321
322    def on_text(self):
323        '''注册消息触发'''
324        def decorator(func):
325            self._add_func(func)
326            return func
327        return decorator
328
329    def on_state(self, *states: Union[AyakaState, str, List[str]]):
330        '''注册有状态响应，不填写states则为root.插件名状态'''
331        _states = []
332        if not states:
333            states = [[]]
334        for s in states:
335            if isinstance(s, str):
336                s = self.get_state(s)
337            elif isinstance(s, list):
338                s = self.get_state(*s)
339            _states.append(s)
340
341        def decorator(func):
342            func.states = _states
343            self._add_func(func)
344            return func
345        return decorator
346
347    def on_idle(self):
348        '''注册根结点回调'''
349        return self.on_state(self.root_state)
350
351    def set_start_cmds(self, *cmds: str):
352        '''设置应用启动命令，当然，你也可以通过app.on_cmd自定义启动方式'''
353        @self.on_idle()
354        @self.on_cmd(*cmds)
355        async def start():
356            '''打开应用'''
357            await self.start()
358
359    def set_close_cmds(self, *cmds: str):
360        '''设置应用关闭命令，当然，你也可以通过app.on_cmd自定义关闭方式'''
361        @self.on_state()
362        @self.on_deep_all()
363        @self.on_cmd(*cmds)
364        async def close():
365            '''关闭应用'''
366            await self.close()
367
368    async def start(self, state: str = None):
369        '''*timer触发时不可用*
370
371        启动应用，并发送提示
372
373        state参数为兼容旧API'''
374        if not state:
375            state = self.get_state()
376            await self.goto(state)
377        else:
378            states = state.split(".")
379            await self.goto(*states)
380        await self.send(f"已打开应用 [{self.name}]")
381
382    async def close(self):
383        '''*timer触发时不可用*
384
385        关闭应用，并发送提示'''
386        await self.goto(root_state)
387        await self.send(f"已关闭应用 [{self.name}]")
388
389    def add_listener(self, user_id: int):
390        '''为该群组添加对指定私聊的监听'''
391        private_listener_dict[user_id].append(self.group_id)
392
393    def remove_listener(self, user_id: int = 0):
394        '''默认移除该群组对其他私聊的所有监听'''
395        id = self.group_id
396
397        if user_id == 0:
398            for ids in private_listener_dict.values():
399                if id in ids:
400                    ids.remove(id)
401            return
402
403        if id in private_listener_dict[user_id]:
404            private_listener_dict[user_id].remove(self.group_id)
405
406    async def send(self, message):
407        '''发送消息，消息的类型可以是 Message | MessageSegment | str'''
408        # 这里不使用event，因为一些event可能来自其他设备的监听传递
409        await self.bot.send_group_msg(group_id=self.group_id, message=message)
410
411    def pack_messages(self, bot_id, messages):
412        '''转换为cqhttp node格式'''
413        data: List[MessageSegment] = []
414        for m in messages:
415            if isinstance(m, MessageSegment) and m.type == "node":
416                data.append(m)
417            else:
418                m = MessageSegment.node_custom(
419                    user_id=bot_id,
420                    nickname="Ayaka Bot",
421                    content=str(m)
422                )
423                data.append(m)
424        return data
425
426    async def send_many(self, messages):
427        '''发送合并转发消息，消息的类型可以是 List[Message | MessageSegment | str]'''
428        # 分割长消息组（不可超过100条
429        div_len = 100
430        div_cnt = ceil(len(messages) / div_len)
431        for i in range(div_cnt):
432            msgs = self.pack_messages(
433                self.bot_id,
434                messages[i*div_len: (i+1)*div_len]
435            )
436            await self.bot.call_api("send_group_forward_msg", group_id=self.group_id, messages=msgs)
437
438    def t_check(self, bot_id: int):
439        # 未连接
440        bot = get_bot(bot_id)
441        if not bot:
442            logger.warning(f"BOT({bot_id}) 未连接")
443            return
444
445        return bot
446
447    async def t_send(self, bot_id: int, group_id: int, message):
448        '''timer触发回调时，想要发送消息必须使用该方法，一些上下文亦无法使用'''
449        bot = self.t_check(bot_id)
450        if not bot:
451            return
452
453        await bot.send_group_msg(group_id=group_id, message=message)
454
455    async def t_send_many(self, bot_id: int, group_id: int, messages):
456        '''timer触发回调时，想要发送消息必须使用该方法，一些上下文亦无法使用'''
457        bot = self.t_check(bot_id)
458        if not bot:
459            return
460
461        # 分割长消息组（不可超过100条）谨慎起见，使用80作为单元长度
462        div_len = 80
463        div_cnt = ceil(len(messages) / div_len)
464        for i in range(div_cnt):
465            msgs = self.pack_messages(
466                bot_id,
467                messages[i*div_len: (i+1)*div_len]
468            )
469            await bot.call_api("send_group_forward_msg", group_id=group_id, messages=msgs)
470
471
472def get_bot(bot_id: int):
473    '''获取已连接的bot'''
474    bot_id = str(bot_id)
475    for bot in bot_list:
476        if bot.self_id == bot_id:
477            return bot
478
479
480def get_group(bot_id: int, group_id: int):
481    '''获取对应的AyakaGroup对象，自动增加'''
482    for group in group_list:
483        if group.bot_id == bot_id and group.group_id == group_id:
484            break
485    else:
486        group = AyakaGroup(bot_id, group_id)
487    return group
488
489
490def get_app(app_name: str):
491    for app in app_list:
492        if app.name == app_name:
493            return app
494
495
496async def deal_event(bot: Bot, event: MessageEvent):
497    '''处理收到的消息，将其分割为cmd和args，设置上下文相关变量的值，并将消息传递给对应的群组'''
498    if ayaka_root_config.exclude_old_msg:
499        time_i = int(datetime.datetime.now().timestamp())
500        if event.time < time_i - 60:
501            return
502
503    _bot.set(bot)
504    _event.set(event)
505
506    bot_id = int(bot.self_id)
507
508    if isinstance(event, GroupMessageEvent):
509        group_id = event.group_id
510        await deal_group(bot_id, group_id)
511
512    else:
513        id = event.user_id
514        group_ids = private_listener_dict.get(id, [])
515        ts = [asyncio.create_task(deal_group(bot_id, group_id))
516              for group_id in group_ids]
517        await asyncio.gather(*ts)
518
519
520async def deal_group(bot_id: int, group_id: int):
521    prefix = ayaka_root_config.prefix
522
523    # 群组
524    group = get_group(bot_id, group_id)
525    _group.set(group)
526
527    # 消息
528    message = _event.get().message
529    _message.set(message)
530
531    # 从子状态开始向上查找可用的触发
532    state = group.state
533    cascade_triggers = get_cascade_triggers(state, 0)
534
535    # 命令
536    # 消息前缀文本
537    first = get_first(message)
538    if first.startswith(prefix):
539        first = first[len(prefix):]
540        for ts in cascade_triggers:
541            if await deal_cmd_triggers(ts, message, first, state):
542                return
543
544    # 命令退化成消息
545    for ts in cascade_triggers:
546        if await deal_text_triggers(ts, message, state):
547            return
548
549
550async def deal_cmd_triggers(triggers: List[AyakaTrigger], message: Message, first: str, state: AyakaState):
551    sep = ayaka_root_config.separate
552
553    # 命令
554    temp = [t for t in triggers if t.cmds]
555    # 根据命令长度排序，长命令优先级更高
556    cmd_ts = [(c, t) for t in temp for c in t.cmds]
557    cmd_ts.sort(key=lambda x: len(x[0]), reverse=1)
558
559    for c, t in cmd_ts:
560        if first.startswith(c):
561            # 设置上下文
562            # 设置命令
563            _cmd.set(c)
564            # 设置参数
565            left = first[len(c):].lstrip(sep)
566            if left:
567                arg = Message([MessageSegment.text(left), *message[1:]])
568            else:
569                arg = Message(message[1:])
570            _arg.set(arg)
571            _args.set(divide_message(arg))
572
573            # 触发
574            log_trigger(c, t.app.name, state, t.func.__name__)
575            await t.run()
576
577            # 阻断后续
578            if t.block:
579                return True
580
581
582async def deal_text_triggers(triggers: List[AyakaTrigger], message: Message, state: AyakaState):
583    # 消息
584    text_ts = [t for t in triggers if not t.cmds]
585
586    # 设置上下文
587    # 设置命令
588    _cmd.set("")
589    # 设置参数
590    _arg.set(message)
591    _args.set(divide_message(message))
592
593    for t in text_ts:
594        # 触发
595        log_trigger("", t.app.name, state, t.func.__name__)
596        await t.run()
597
598        # 阻断后续
599        if t.block:
600            return True
601
602
603def get_cascade_triggers(state: AyakaState, deep: int = 0):
604    # 根据深度筛选funcs
605    ts = [
606        t for t in state.triggers
607        if t.deep == "all" or t.deep >= deep
608    ]
609    cascade_triggers = [ts]
610
611    # 获取父状态的方法
612    if state.parent:
613        cascade_triggers.extend(get_cascade_triggers(state.parent, deep+1))
614
615    # 排除空项
616    cascade_triggers = [ts for ts in cascade_triggers if ts]
617    return cascade_triggers
618
619
620def log_trigger(cmd, app_name, state, func_name):
621    '''日志记录'''
622    items = []
623    items.append(f"状态：<c>{state}</c>")
624    items.append(f"应用：<y>{app_name}</y>")
625    if cmd:
626        items.append(f"命令：<y>{cmd}</y>")
627    else:
628        items.append("命令：<g>无</g>")
629    items.append(f"回调：<c>{func_name}</c>")
630    info = " | ".join(items)
631    logger.opt(colors=True).debug(info)
632
633
634def get_first(message: Message):
635    first = ""
636    for m in message:
637        if m.type == "text":
638            first += str(m)
639        else:
640            break
641    return first
642
643
644def divide_message(message: Message) -> List[MessageSegment]:
645    args = []
646    sep = ayaka_root_config.separate
647
648    for m in message:
649        if m.is_text():
650            ss = str(m).split(sep)
651            args.extend(MessageSegment.text(s) for s in ss if s)
652        else:
653            args.append(m)
654
655    return args
656
657
658def regist_func(app: AyakaApp, func):
659    '''注册回调'''
660    # 默认是无状态应用，从root开始触发
661    states: List[AyakaState] = getattr(func, "states", [root_state])
662    # 默认是消息响应
663    cmds: List[str] = getattr(func, "cmds", [])
664    # 默认监听深度为0
665    deep: int = getattr(func, "deep", 0)
666    # 默认阻断
667    block: bool = getattr(func, "block", True)
668
669    # 注册
670    for s in states:
671        s.on_cmd(cmds, app, deep, block)(func)
672
673    return func
674
675
676driver = get_driver()
677
678
679@driver.on_startup
680async def startup():
681    # 注册所有回调
682    for app in app_list:
683        for func in app.funcs:
684            regist_func(app, func)
685
686    if ayaka_root_config.debug:
687        s = json.dumps(
688            root_state.dict(), ensure_ascii=0,
689            indent=4, default=repr
690        )
691        path = ayaka_data_path / "all_state.json"
692        with path.open("w+", encoding="utf8") as f:
693            f.write(s)
694
695on_message(priority=20, block=False, handlers=[deal_event])