ayaka.driver.ayakabot_driver.event

  1import inspect
  2from typing_extensions import Literal
  3from typing import TYPE_CHECKING, List, Type, Optional
  4from pydantic import BaseModel
  5from pygtrie import StringTrie
  6from pydantic import BaseModel
  7
  8from .result import ResultStore
  9from .model import DataclassEncoder
 10from .message import Message
 11
 12from loguru import logger
 13
 14if TYPE_CHECKING:
 15    from .bot import Bot
 16
 17
 18class Event(BaseModel):
 19    """
 20    OneBot v11 协议事件，字段与 OneBot 一致。各事件字段参考 `OneBot 文档`_
 21
 22    .. OneBot 文档:
 23        https://github.com/botuniverse/onebot-11/blob/master/README.md
 24    """
 25
 26    class Config:
 27        extra = "allow"
 28        json_encoders = {Message: DataclassEncoder}
 29
 30    def __str__(self) -> str:
 31        return f"[{self.get_event_name()}]: {self.get_event_description()}"
 32
 33    def get_log_string(self) -> str:
 34        """获取事件日志信息的方法。
 35
 36        通常你不需要修改这个方法，只有当希望 ayaka 隐藏该事件日志时，可以抛出 `NoLogException` 异常。
 37
 38        异常:
 39            NoLogException
 40        """
 41        return f"[{self.get_event_name()}]: {self.get_event_description()}"
 42
 43    def get_plaintext(self) -> str:
 44        """获取消息纯文本的方法。
 45
 46        通常不需要修改，默认通过 `get_message().extract_plain_text` 获取。
 47        """
 48        return self.get_message().extract_plain_text()
 49
 50    __event__ = ""
 51    time: int
 52    self_id: int
 53    post_type: str
 54
 55    def get_type(self) -> str:
 56        return self.post_type
 57
 58    def get_event_name(self) -> str:
 59        return self.post_type
 60
 61    def get_event_description(self) -> str:
 62        return str(self.dict())
 63
 64    def get_message(self) -> Message:
 65        raise ValueError("Event has no message!")
 66
 67    def get_plaintext(self) -> str:
 68        raise ValueError("Event has no message!")
 69
 70    def get_user_id(self) -> str:
 71        raise ValueError("Event has no context!")
 72
 73    def get_session_id(self) -> str:
 74        raise ValueError("Event has no context!")
 75
 76    def is_tome(self) -> bool:
 77        return False
 78
 79
 80# Models
 81class Sender(BaseModel):
 82    user_id: Optional[int] = None
 83    nickname: Optional[str] = None
 84    sex: Optional[str] = None
 85    age: Optional[int] = None
 86    card: Optional[str] = None
 87    area: Optional[str] = None
 88    level: Optional[str] = None
 89    role: Optional[str] = None
 90    title: Optional[str] = None
 91
 92    class Config:
 93        extra = "allow"
 94
 95
 96class Reply(BaseModel):
 97    time: int
 98    message_type: str
 99    message_id: int
100    real_id: int
101    sender: Sender
102    message: Message
103
104    class Config:
105        extra = "allow"
106
107
108class Anonymous(BaseModel):
109    id: int
110    name: str
111    flag: str
112
113    class Config:
114        extra = "allow"
115
116
117class File(BaseModel):
118    id: str
119    name: str
120    size: int
121    busid: int
122
123    class Config:
124        extra = "allow"
125
126
127class Status(BaseModel):
128    online: bool
129    good: bool
130
131    class Config:
132        extra = "allow"
133
134
135# Message Events
136class MessageEvent(Event):
137    """消息事件"""
138
139    __event__ = "message"
140    post_type: Literal["message"]
141    sub_type: str
142    user_id: int
143    message_type: str
144    message_id: int
145    message: Message
146    raw_message: str
147    font: int
148    sender: Sender
149    reply: Optional[Reply] = None
150    """
151    :说明: 消息中提取的回复消息，内容为 ``get_msg`` API 返回结果
152
153    :类型: ``Optional[Reply]``
154    """
155
156    def get_event_name(self) -> str:
157        sub_type = getattr(self, "sub_type", None)
158        return f"{self.post_type}.{self.message_type}" + (
159            f".{sub_type}" if sub_type else ""
160        )
161
162    def get_message(self) -> Message:
163        return self.message
164
165    def get_plaintext(self) -> str:
166        return self.message.extract_plain_text()
167
168    def get_user_id(self) -> str:
169        return str(self.user_id)
170
171    def get_session_id(self) -> str:
172        return str(self.user_id)
173
174
175class PrivateMessageEvent(MessageEvent):
176    """私聊消息"""
177
178    __event__ = "message.private"
179    message_type: Literal["private"]
180
181    def get_event_description(self) -> str:
182        return (
183            f'Message {self.message_id} from {self.user_id} \n'
184            + "".join(
185                map(
186                    lambda x: str(x) if x.is_text() else f"<c>{x}</c>",
187                    self.message,
188                )
189            )
190        )
191
192
193class GroupMessageEvent(MessageEvent):
194    """群消息"""
195
196    __event__ = "message.group"
197    message_type: Literal["group"]
198    group_id: int
199    anonymous: Optional[Anonymous] = None
200
201    def get_event_description(self) -> str:
202        return (
203            f'Message {self.message_id} from {self.user_id}@[群:{self.group_id}] \n'
204            + "".join(
205                map(
206                    lambda x: str(x) if x.is_text() else f"<c>{x}</c>",
207                    self.message,
208                )
209            )
210        )
211
212    def get_session_id(self) -> str:
213        return f"group_{self.group_id}_{self.user_id}"
214
215
216# Notice Events
217class NoticeEvent(Event):
218    """通知事件"""
219
220    __event__ = "notice"
221    post_type: Literal["notice"]
222    notice_type: str
223
224    def get_event_name(self) -> str:
225        sub_type = getattr(self, "sub_type", None)
226        sub_type = f".{sub_type}" if sub_type else ""
227        return f"{self.post_type}.{self.notice_type}{sub_type}"
228
229
230class GroupUploadNoticeEvent(NoticeEvent):
231    """群文件上传事件"""
232
233    __event__ = "notice.group_upload"
234    notice_type: Literal["group_upload"]
235    user_id: int
236    group_id: int
237    file: File
238
239    def get_user_id(self) -> str:
240        return str(self.user_id)
241
242    def get_session_id(self) -> str:
243        return f"group_{self.group_id}_{self.user_id}"
244
245
246class GroupAdminNoticeEvent(NoticeEvent):
247    """群管理员变动"""
248
249    __event__ = "notice.group_admin"
250    notice_type: Literal["group_admin"]
251    sub_type: str
252    user_id: int
253    group_id: int
254
255    def is_tome(self) -> bool:
256        return self.user_id == self.self_id
257
258    def get_user_id(self) -> str:
259        return str(self.user_id)
260
261    def get_session_id(self) -> str:
262        return f"group_{self.group_id}_{self.user_id}"
263
264
265class GroupDecreaseNoticeEvent(NoticeEvent):
266    """群成员减少事件"""
267
268    __event__ = "notice.group_decrease"
269    notice_type: Literal["group_decrease"]
270    sub_type: str
271    user_id: int
272    group_id: int
273    operator_id: int
274
275    def is_tome(self) -> bool:
276        return self.user_id == self.self_id
277
278    def get_user_id(self) -> str:
279        return str(self.user_id)
280
281    def get_session_id(self) -> str:
282        return f"group_{self.group_id}_{self.user_id}"
283
284
285class GroupIncreaseNoticeEvent(NoticeEvent):
286    """群成员增加事件"""
287
288    __event__ = "notice.group_increase"
289    notice_type: Literal["group_increase"]
290    sub_type: str
291    user_id: int
292    group_id: int
293    operator_id: int
294
295    def is_tome(self) -> bool:
296        return self.user_id == self.self_id
297
298    def get_user_id(self) -> str:
299        return str(self.user_id)
300
301    def get_session_id(self) -> str:
302        return f"group_{self.group_id}_{self.user_id}"
303
304
305class GroupBanNoticeEvent(NoticeEvent):
306    """群禁言事件"""
307
308    __event__ = "notice.group_ban"
309    notice_type: Literal["group_ban"]
310    sub_type: str
311    user_id: int
312    group_id: int
313    operator_id: int
314    duration: int
315
316    def is_tome(self) -> bool:
317        return self.user_id == self.self_id
318
319    def get_user_id(self) -> str:
320        return str(self.user_id)
321
322    def get_session_id(self) -> str:
323        return f"group_{self.group_id}_{self.user_id}"
324
325
326class FriendAddNoticeEvent(NoticeEvent):
327    """好友添加事件"""
328
329    __event__ = "notice.friend_add"
330    notice_type: Literal["friend_add"]
331    user_id: int
332
333    def get_user_id(self) -> str:
334        return str(self.user_id)
335
336    def get_session_id(self) -> str:
337        return str(self.user_id)
338
339
340class GroupRecallNoticeEvent(NoticeEvent):
341    """群消息撤回事件"""
342
343    __event__ = "notice.group_recall"
344    notice_type: Literal["group_recall"]
345    user_id: int
346    group_id: int
347    operator_id: int
348    message_id: int
349
350    def is_tome(self) -> bool:
351        return self.user_id == self.self_id
352
353    def get_user_id(self) -> str:
354        return str(self.user_id)
355
356    def get_session_id(self) -> str:
357        return f"group_{self.group_id}_{self.user_id}"
358
359
360class FriendRecallNoticeEvent(NoticeEvent):
361    """好友消息撤回事件"""
362
363    __event__ = "notice.friend_recall"
364    notice_type: Literal["friend_recall"]
365    user_id: int
366    message_id: int
367
368    def get_user_id(self) -> str:
369        return str(self.user_id)
370
371    def get_session_id(self) -> str:
372        return str(self.user_id)
373
374
375class NotifyEvent(NoticeEvent):
376    """提醒事件"""
377
378    __event__ = "notice.notify"
379    notice_type: Literal["notify"]
380    sub_type: str
381    user_id: int
382    group_id: int
383
384    def get_user_id(self) -> str:
385        return str(self.user_id)
386
387    def get_session_id(self) -> str:
388        return f"group_{self.group_id}_{self.user_id}"
389
390
391class PokeNotifyEvent(NotifyEvent):
392    """戳一戳提醒事件"""
393
394    __event__ = "notice.notify.poke"
395    sub_type: Literal["poke"]
396    target_id: int
397    group_id: Optional[int] = None
398
399    def is_tome(self) -> bool:
400        return self.target_id == self.self_id
401
402    def get_session_id(self) -> str:
403        if not self.group_id:
404            return str(self.user_id)
405        return super().get_session_id()
406
407
408class LuckyKingNotifyEvent(NotifyEvent):
409    """群红包运气王提醒事件"""
410
411    __event__ = "notice.notify.lucky_king"
412    sub_type: Literal["lucky_king"]
413    target_id: int
414
415    def is_tome(self) -> bool:
416        return self.target_id == self.self_id
417
418    def get_user_id(self) -> str:
419        return str(self.target_id)
420
421    def get_session_id(self) -> str:
422        return f"group_{self.group_id}_{self.target_id}"
423
424
425class HonorNotifyEvent(NotifyEvent):
426    """群荣誉变更提醒事件"""
427
428    __event__ = "notice.notify.honor"
429    sub_type: Literal["honor"]
430    honor_type: str
431
432    def is_tome(self) -> bool:
433        return self.user_id == self.self_id
434
435
436# Request Events
437class RequestEvent(Event):
438    """请求事件"""
439
440    __event__ = "request"
441    post_type: Literal["request"]
442    request_type: str
443
444    def get_event_name(self) -> str:
445        sub_type = getattr(self, "sub_type", None)
446        sub_type = f".{sub_type}" if sub_type else ""
447        return f"{self.post_type}.{self.request_type}{sub_type}"
448
449
450class FriendRequestEvent(RequestEvent):
451    """加好友请求事件"""
452
453    __event__ = "request.friend"
454    request_type: Literal["friend"]
455    user_id: int
456    comment: str
457    flag: str
458
459    def get_user_id(self) -> str:
460        return str(self.user_id)
461
462    def get_session_id(self) -> str:
463        return str(self.user_id)
464
465    async def approve(self, bot: "Bot", remark: str = ""):
466        return await bot.set_friend_add_request(
467            flag=self.flag, approve=True, remark=remark
468        )
469
470    async def reject(self, bot: "Bot"):
471        return await bot.set_friend_add_request(flag=self.flag, approve=False)
472
473
474class GroupRequestEvent(RequestEvent):
475    """加群请求/邀请事件"""
476
477    __event__ = "request.group"
478    request_type: Literal["group"]
479    sub_type: str
480    group_id: int
481    user_id: int
482    comment: str
483    flag: str
484
485    def get_user_id(self) -> str:
486        return str(self.user_id)
487
488    def get_session_id(self) -> str:
489        return f"group_{self.group_id}_{self.user_id}"
490
491    async def approve(self, bot: "Bot"):
492        return await bot.set_group_add_request(
493            flag=self.flag, sub_type=self.sub_type, approve=True
494        )
495
496    async def reject(self, bot: "Bot", reason: str = ""):
497        return await bot.set_group_add_request(
498            flag=self.flag, sub_type=self.sub_type, approve=False, reason=reason
499        )
500
501
502# Meta Events
503class MetaEvent(Event):
504    """元事件"""
505
506    __event__ = "meta_event"
507    post_type: Literal["meta_event"]
508    meta_event_type: str
509
510    def get_event_name(self) -> str:
511        sub_type = getattr(self, "sub_type", None)
512        return f"{self.post_type}.{self.meta_event_type}" + (
513            f".{sub_type}" if sub_type else ""
514        )
515
516    def get_log_string(self) -> str:
517        return ""
518
519
520class LifecycleMetaEvent(MetaEvent):
521    """生命周期元事件"""
522
523    __event__ = "meta_event.lifecycle"
524    meta_event_type: Literal["lifecycle"]
525    sub_type: str
526
527
528class HeartbeatMetaEvent(MetaEvent):
529    """心跳元事件"""
530
531    __event__ = "meta_event.heartbeat"
532    meta_event_type: Literal["heartbeat"]
533    status: Status
534    interval: int
535
536
537# 属实奢侈了一把，家人们
538_t = StringTrie(separator=".")
539
540# define `model` first to avoid globals changing while `for`
541model = None
542for model in globals().values():
543    if not inspect.isclass(model) or not issubclass(model, Event):
544        continue
545    _t["." + model.__event__] = model
546
547
548def get_event_model(event_name) -> List[Type[Event]]:
549    """
550    :说明:
551
552      根据事件名获取对应 ``Event Model`` 及 ``FallBack Event Model`` 列表
553
554    :返回:
555
556      - ``List[Type[Event]]``
557    """
558    return [model.value for model in _t.prefixes("." + event_name)][::-1]
559
560
561def json_to_event(json_data) -> Optional[Event]:
562    if not isinstance(json_data, dict):
563        return None
564
565    if "post_type" not in json_data:
566        ResultStore.add_result(json_data)
567        return
568
569    try:
570        post_type = json_data["post_type"]
571        detail_type = json_data.get(f"{post_type}_type")
572        detail_type = f".{detail_type}" if detail_type else ""
573        sub_type = json_data.get("sub_type")
574        sub_type = f".{sub_type}" if sub_type else ""
575        models = get_event_model(post_type + detail_type + sub_type)
576        for model in models:
577            try:
578                event = model.parse_obj(json_data)
579                break
580            except:
581                logger.debug(f"Event Parser Error {model.__name__}")
582        else:
583            event = Event.parse_obj(json_data)
584
585        return event
586
587    except:
588        logger.exception()
589        logger.opt(colors=True).error(f"<r>Failed to parse event.\nRaw: {json_data}</r>")
590
591
592__all__ = [
593    "Event",
594    "MessageEvent",
595    "PrivateMessageEvent",
596    "GroupMessageEvent",
597    "NoticeEvent",
598    "GroupUploadNoticeEvent",
599    "GroupAdminNoticeEvent",
600    "GroupDecreaseNoticeEvent",
601    "GroupIncreaseNoticeEvent",
602    "GroupBanNoticeEvent",
603    "FriendAddNoticeEvent",
604    "GroupRecallNoticeEvent",
605    "FriendRecallNoticeEvent",
606    "NotifyEvent",
607    "PokeNotifyEvent",
608    "LuckyKingNotifyEvent",
609    "HonorNotifyEvent",
610    "RequestEvent",
611    "FriendRequestEvent",
612    "GroupRequestEvent",
613    "MetaEvent",
614    "LifecycleMetaEvent",
615    "HeartbeatMetaEvent",
616    "get_event_model",
617]