🚧 update event models

This commit is contained in:
yanyongyu 2020-12-09 14:39:25 +08:00
parent 73bda494d9
commit 2bc05b2576
3 changed files with 398 additions and 306 deletions

View File

@ -9,6 +9,7 @@ import abc
from functools import reduce, partial
from dataclasses import dataclass, field
from typing import Any, Dict, Union, TypeVar, Optional, Callable, Iterable, Awaitable, Generic, TYPE_CHECKING
from typing_extensions import Literal
from pydantic import BaseModel
@ -137,169 +138,193 @@ class Bot(abc.ABC):
raise NotImplementedError
T = TypeVar("T", bound=BaseModel)
class Event(abc.ABC, BaseModel):
class Config:
extra = "allow"
class Event(abc.ABC, Generic[T]):
"""
Event 基类提供上报信息的关键信息其余信息可从原始上报消息获取
"""
def __init__(self, raw_event: Union[dict, T]):
"""
:参数:
* ``raw_event: Union[dict, T]``: 原始上报消息
"""
self._raw_event = raw_event
def __repr__(self) -> str:
return f"<Event {self.self_id}: {self.name} {self.time}>"
@property
def raw_event(self) -> Union[dict, T]:
"""原始上报消息"""
return self._raw_event
@property
@abc.abstractmethod
def id(self) -> int:
"""事件 ID"""
def get_type(self) -> Literal["message", "notice", "request", "meta_event"]:
raise NotImplementedError
@property
@abc.abstractmethod
def name(self) -> str:
"""事件名称"""
def get_event_name(self) -> str:
raise NotImplementedError
@property
@abc.abstractmethod
def self_id(self) -> str:
"""机器人 ID"""
def get_event_description(self) -> str:
raise NotImplementedError
@property
def get_log_string(self) -> str:
return f"[{self.get_event_name()}]: {self.get_event_description()}"
@abc.abstractmethod
def time(self) -> int:
"""事件发生时间"""
def get_session_id(self) -> str:
raise NotImplementedError
@property
@abc.abstractmethod
def type(self) -> str:
"""事件主类型"""
raise NotImplementedError
@type.setter
@abc.abstractmethod
def type(self, value) -> None:
raise NotImplementedError
# T = TypeVar("T", bound=BaseModel)
@property
@abc.abstractmethod
def detail_type(self) -> str:
"""事件详细类型"""
raise NotImplementedError
# class Event(abc.ABC, Generic[T]):
# """
# Event 基类。提供上报信息的关键信息,其余信息可从原始上报消息获取。
# """
@detail_type.setter
@abc.abstractmethod
def detail_type(self, value) -> None:
raise NotImplementedError
# def __init__(self, raw_event: Union[dict, T]):
# """
# :参数:
@property
@abc.abstractmethod
def sub_type(self) -> Optional[str]:
"""事件子类型"""
raise NotImplementedError
# * ``raw_event: Union[dict, T]``: 原始上报消息
# """
# self._raw_event = raw_event
@sub_type.setter
@abc.abstractmethod
def sub_type(self, value) -> None:
raise NotImplementedError
# def __repr__(self) -> str:
# return f"<Event {self.self_id}: {self.name} {self.time}>"
@property
@abc.abstractmethod
def user_id(self) -> Optional[int]:
"""触发事件的主体 ID"""
raise NotImplementedError
# @property
# def raw_event(self) -> Union[dict, T]:
# """原始上报消息"""
# return self._raw_event
@user_id.setter
@abc.abstractmethod
def user_id(self, value) -> None:
raise NotImplementedError
# @property
# @abc.abstractmethod
# def id(self) -> int:
# """事件 ID"""
# raise NotImplementedError
@property
@abc.abstractmethod
def group_id(self) -> Optional[int]:
"""触发事件的主体群 ID"""
raise NotImplementedError
# @property
# @abc.abstractmethod
# def name(self) -> str:
# """事件名称"""
# raise NotImplementedError
@group_id.setter
@abc.abstractmethod
def group_id(self, value) -> None:
raise NotImplementedError
# @property
# @abc.abstractmethod
# def self_id(self) -> str:
# """机器人 ID"""
# raise NotImplementedError
@property
@abc.abstractmethod
def to_me(self) -> Optional[bool]:
"""事件是否为发送给机器人的消息"""
raise NotImplementedError
# @property
# @abc.abstractmethod
# def time(self) -> int:
# """事件发生时间"""
# raise NotImplementedError
@to_me.setter
@abc.abstractmethod
def to_me(self, value) -> None:
raise NotImplementedError
# @property
# @abc.abstractmethod
# def type(self) -> str:
# """事件主类型"""
# raise NotImplementedError
@property
@abc.abstractmethod
def message(self) -> Optional["Message"]:
"""消息内容"""
raise NotImplementedError
# @type.setter
# @abc.abstractmethod
# def type(self, value) -> None:
# raise NotImplementedError
@message.setter
@abc.abstractmethod
def message(self, value) -> None:
raise NotImplementedError
# @property
# @abc.abstractmethod
# def detail_type(self) -> str:
# """事件详细类型"""
# raise NotImplementedError
@property
@abc.abstractmethod
def reply(self) -> Optional[dict]:
"""回复的消息"""
raise NotImplementedError
# @detail_type.setter
# @abc.abstractmethod
# def detail_type(self, value) -> None:
# raise NotImplementedError
@reply.setter
@abc.abstractmethod
def reply(self, value) -> None:
raise NotImplementedError
# @property
# @abc.abstractmethod
# def sub_type(self) -> Optional[str]:
# """事件子类型"""
# raise NotImplementedError
@property
@abc.abstractmethod
def raw_message(self) -> Optional[str]:
"""原始消息"""
raise NotImplementedError
# @sub_type.setter
# @abc.abstractmethod
# def sub_type(self, value) -> None:
# raise NotImplementedError
@raw_message.setter
@abc.abstractmethod
def raw_message(self, value) -> None:
raise NotImplementedError
# @property
# @abc.abstractmethod
# def user_id(self) -> Optional[int]:
# """触发事件的主体 ID"""
# raise NotImplementedError
@property
@abc.abstractmethod
def plain_text(self) -> Optional[str]:
"""纯文本消息"""
raise NotImplementedError
# @user_id.setter
# @abc.abstractmethod
# def user_id(self, value) -> None:
# raise NotImplementedError
@property
@abc.abstractmethod
def sender(self) -> Optional[dict]:
"""消息发送者信息"""
raise NotImplementedError
# @property
# @abc.abstractmethod
# def group_id(self) -> Optional[int]:
# """触发事件的主体群 ID"""
# raise NotImplementedError
@sender.setter
@abc.abstractmethod
def sender(self, value) -> None:
raise NotImplementedError
# @group_id.setter
# @abc.abstractmethod
# def group_id(self, value) -> None:
# raise NotImplementedError
# @property
# @abc.abstractmethod
# def to_me(self) -> Optional[bool]:
# """事件是否为发送给机器人的消息"""
# raise NotImplementedError
# @to_me.setter
# @abc.abstractmethod
# def to_me(self, value) -> None:
# raise NotImplementedError
# @property
# @abc.abstractmethod
# def message(self) -> Optional["Message"]:
# """消息内容"""
# raise NotImplementedError
# @message.setter
# @abc.abstractmethod
# def message(self, value) -> None:
# raise NotImplementedError
# @property
# @abc.abstractmethod
# def reply(self) -> Optional[dict]:
# """回复的消息"""
# raise NotImplementedError
# @reply.setter
# @abc.abstractmethod
# def reply(self, value) -> None:
# raise NotImplementedError
# @property
# @abc.abstractmethod
# def raw_message(self) -> Optional[str]:
# """原始消息"""
# raise NotImplementedError
# @raw_message.setter
# @abc.abstractmethod
# def raw_message(self, value) -> None:
# raise NotImplementedError
# @property
# @abc.abstractmethod
# def plain_text(self) -> Optional[str]:
# """纯文本消息"""
# raise NotImplementedError
# @property
# @abc.abstractmethod
# def sender(self) -> Optional[dict]:
# """消息发送者信息"""
# raise NotImplementedError
# @sender.setter
# @abc.abstractmethod
# def sender(self, value) -> None:
# raise NotImplementedError
@dataclass

View File

@ -2,220 +2,230 @@ from typing import Optional
from typing_extensions import Literal
from pydantic import BaseModel
from nonebot.adapters import Event
from nonebot.utils import escape_tag
from nonebot.typing import overrides
from nonebot.adapters import Event as BaseEvent
from nonebot.exception import NoLogException
from .message import Message
# class Event(BaseEvent):
# """
# CQHTTP 协议 Event 适配。继承属性参考 `BaseEvent <./#class-baseevent>`_ 。
# """
class Event(BaseEvent):
"""
CQHTTP 协议 Event 适配继承属性参考 `BaseEvent <./#class-baseevent>`_ 。
"""
# def __init__(self, raw_event: dict):
# if "message" in raw_event:
# raw_event["message"] = Message(raw_event["message"])
def __init__(self, raw_event: dict):
if "message" in raw_event:
raw_event["message"] = Message(raw_event["message"])
# super().__init__(raw_event)
super().__init__(raw_event)
# @property
# @overrides(BaseEvent)
# def id(self) -> Optional[int]:
# """
# - 类型: ``Optional[int]``
# - 说明: 事件/消息 ID
# """
# return self._raw_event.get("message_id") or self._raw_event.get("flag")
@property
@overrides(BaseEvent)
def id(self) -> Optional[int]:
"""
- 类型: ``Optional[int]``
- 说明: 事件/消息 ID
"""
return self._raw_event.get("message_id") or self._raw_event.get("flag")
# @property
# @overrides(BaseEvent)
# def name(self) -> str:
# """
# - 类型: ``str``
# - 说明: 事件名称,由类型与 ``.`` 组合而成
# """
# n = self.type + "." + self.detail_type
# if self.sub_type:
# n += "." + self.sub_type
# return n
@property
@overrides(BaseEvent)
def name(self) -> str:
"""
- 类型: ``str``
- 说明: 事件名称由类型与 ``.`` 组合而成
"""
n = self.type + "." + self.detail_type
if self.sub_type:
n += "." + self.sub_type
return n
# @property
# @overrides(BaseEvent)
# def self_id(self) -> str:
# """
# - 类型: ``str``
# - 说明: 机器人自身 ID
# """
# return str(self._raw_event["self_id"])
@property
@overrides(BaseEvent)
def self_id(self) -> str:
"""
- 类型: ``str``
- 说明: 机器人自身 ID
"""
return str(self._raw_event["self_id"])
# @property
# @overrides(BaseEvent)
# def time(self) -> int:
# """
# - 类型: ``int``
# - 说明: 事件发生时间
# """
# return self._raw_event["time"]
@property
@overrides(BaseEvent)
def time(self) -> int:
"""
- 类型: ``int``
- 说明: 事件发生时间
"""
return self._raw_event["time"]
# @property
# @overrides(BaseEvent)
# def type(self) -> str:
# """
# - 类型: ``str``
# - 说明: 事件类型
# """
# return self._raw_event["post_type"]
@property
@overrides(BaseEvent)
def type(self) -> str:
"""
- 类型: ``str``
- 说明: 事件类型
"""
return self._raw_event["post_type"]
# @type.setter
# @overrides(BaseEvent)
# def type(self, value) -> None:
# self._raw_event["post_type"] = value
@type.setter
@overrides(BaseEvent)
def type(self, value) -> None:
self._raw_event["post_type"] = value
# @property
# @overrides(BaseEvent)
# def detail_type(self) -> str:
# """
# - 类型: ``str``
# - 说明: 事件详细类型
# """
# return self._raw_event[f"{self.type}_type"]
@property
@overrides(BaseEvent)
def detail_type(self) -> str:
"""
- 类型: ``str``
- 说明: 事件详细类型
"""
return self._raw_event[f"{self.type}_type"]
# @detail_type.setter
# @overrides(BaseEvent)
# def detail_type(self, value) -> None:
# self._raw_event[f"{self.type}_type"] = value
@detail_type.setter
@overrides(BaseEvent)
def detail_type(self, value) -> None:
self._raw_event[f"{self.type}_type"] = value
# @property
# @overrides(BaseEvent)
# def sub_type(self) -> Optional[str]:
# """
# - 类型: ``Optional[str]``
# - 说明: 事件子类型
# """
# return self._raw_event.get("sub_type")
@property
@overrides(BaseEvent)
def sub_type(self) -> Optional[str]:
"""
- 类型: ``Optional[str]``
- 说明: 事件子类型
"""
return self._raw_event.get("sub_type")
# @sub_type.setter
# @overrides(BaseEvent)
# def sub_type(self, value) -> None:
# self._raw_event["sub_type"] = value
@sub_type.setter
@overrides(BaseEvent)
def sub_type(self, value) -> None:
self._raw_event["sub_type"] = value
# @property
# @overrides(BaseEvent)
# def user_id(self) -> Optional[int]:
# """
# - 类型: ``Optional[int]``
# - 说明: 事件主体 ID
# """
# return self._raw_event.get("user_id")
@property
@overrides(BaseEvent)
def user_id(self) -> Optional[int]:
"""
- 类型: ``Optional[int]``
- 说明: 事件主体 ID
"""
return self._raw_event.get("user_id")
# @user_id.setter
# @overrides(BaseEvent)
# def user_id(self, value) -> None:
# self._raw_event["user_id"] = value
@user_id.setter
@overrides(BaseEvent)
def user_id(self, value) -> None:
self._raw_event["user_id"] = value
# @property
# @overrides(BaseEvent)
# def group_id(self) -> Optional[int]:
# """
# - 类型: ``Optional[int]``
# - 说明: 事件主体群 ID
# """
# return self._raw_event.get("group_id")
@property
@overrides(BaseEvent)
def group_id(self) -> Optional[int]:
"""
- 类型: ``Optional[int]``
- 说明: 事件主体群 ID
"""
return self._raw_event.get("group_id")
# @group_id.setter
# @overrides(BaseEvent)
# def group_id(self, value) -> None:
# self._raw_event["group_id"] = value
@group_id.setter
@overrides(BaseEvent)
def group_id(self, value) -> None:
self._raw_event["group_id"] = value
# @property
# @overrides(BaseEvent)
# def to_me(self) -> Optional[bool]:
# """
# - 类型: ``Optional[bool]``
# - 说明: 消息是否与机器人相关
# """
# return self._raw_event.get("to_me")
@property
@overrides(BaseEvent)
def to_me(self) -> Optional[bool]:
"""
- 类型: ``Optional[bool]``
- 说明: 消息是否与机器人相关
"""
return self._raw_event.get("to_me")
# @to_me.setter
# @overrides(BaseEvent)
# def to_me(self, value) -> None:
# self._raw_event["to_me"] = value
@to_me.setter
@overrides(BaseEvent)
def to_me(self, value) -> None:
self._raw_event["to_me"] = value
# @property
# @overrides(BaseEvent)
# def message(self) -> Optional["Message"]:
# """
# - 类型: ``Optional[Message]``
# - 说明: 消息内容
# """
# return self._raw_event.get("message")
@property
@overrides(BaseEvent)
def message(self) -> Optional["Message"]:
"""
- 类型: ``Optional[Message]``
- 说明: 消息内容
"""
return self._raw_event.get("message")
# @message.setter
# @overrides(BaseEvent)
# def message(self, value) -> None:
# self._raw_event["message"] = value
@message.setter
@overrides(BaseEvent)
def message(self, value) -> None:
self._raw_event["message"] = value
# @property
# @overrides(BaseEvent)
# def reply(self) -> Optional[dict]:
# """
# - 类型: ``Optional[dict]``
# - 说明: 回复消息详情
# """
# return self._raw_event.get("reply")
@property
@overrides(BaseEvent)
def reply(self) -> Optional[dict]:
"""
- 类型: ``Optional[dict]``
- 说明: 回复消息详情
"""
return self._raw_event.get("reply")
# @reply.setter
# @overrides(BaseEvent)
# def reply(self, value) -> None:
# self._raw_event["reply"] = value
@reply.setter
@overrides(BaseEvent)
def reply(self, value) -> None:
self._raw_event["reply"] = value
# @property
# @overrides(BaseEvent)
# def raw_message(self) -> Optional[str]:
# """
# - 类型: ``Optional[str]``
# - 说明: 原始消息
# """
# return self._raw_event.get("raw_message")
@property
@overrides(BaseEvent)
def raw_message(self) -> Optional[str]:
"""
- 类型: ``Optional[str]``
- 说明: 原始消息
"""
return self._raw_event.get("raw_message")
# @raw_message.setter
# @overrides(BaseEvent)
# def raw_message(self, value) -> None:
# self._raw_event["raw_message"] = value
@raw_message.setter
@overrides(BaseEvent)
def raw_message(self, value) -> None:
self._raw_event["raw_message"] = value
# @property
# @overrides(BaseEvent)
# def plain_text(self) -> Optional[str]:
# """
# - 类型: ``Optional[str]``
# - 说明: 纯文本消息内容
# """
# return self.message and self.message.extract_plain_text()
@property
@overrides(BaseEvent)
def plain_text(self) -> Optional[str]:
"""
- 类型: ``Optional[str]``
- 说明: 纯文本消息内容
"""
return self.message and self.message.extract_plain_text()
# @property
# @overrides(BaseEvent)
# def sender(self) -> Optional[dict]:
# """
# - 类型: ``Optional[dict]``
# - 说明: 消息发送者信息
# """
# return self._raw_event.get("sender")
@property
@overrides(BaseEvent)
def sender(self) -> Optional[dict]:
"""
- 类型: ``Optional[dict]``
- 说明: 消息发送者信息
"""
return self._raw_event.get("sender")
@sender.setter
@overrides(BaseEvent)
def sender(self, value) -> None:
self._raw_event["sender"] = value
# @sender.setter
# @overrides(BaseEvent)
# def sender(self, value) -> None:
# self._raw_event["sender"] = value
class CQHTTPEvent(BaseModel):
type: str
class CQHTTPEvent(Event):
type: Literal["message", "notice", "request", "meta_event"]
time: int
self_id: int
post_type: str
class Config:
extra = "allow"
@overrides(Event)
def get_type(self) -> Literal["message", "notice", "request", "meta_event"]:
return self.type
@overrides(Event)
def get_event_name(self) -> str:
return self.post_type
@overrides(Event)
def get_event_description(self) -> str:
return str(self.dict())
# Models
@ -266,6 +276,12 @@ class MessageEvent(CQHTTPEvent):
post_type: Literal["message"]
message_type: str
@overrides(CQHTTPEvent)
def get_event_name(self) -> str:
sub_type = getattr(self, "sub_type", None)
return f"{self.post_type}.{self.message_type}" + (f".{sub_type}"
if sub_type else "")
class PrivateMessageEvent(MessageEvent):
message_type: Literal["private"]
@ -277,11 +293,20 @@ class PrivateMessageEvent(MessageEvent):
font: int
sender: Sender
@overrides(CQHTTPEvent)
def get_event_description(self) -> str:
return (f'Message {self.message_id} from {self.user_id} "' + "".join(
map(
lambda x: escape_tag(str(x))
if x.type == "text" else f"<le>{escape_tag(str(x))}</le>",
self.message)) + '"')
class GroupMessageEvent(MessageEvent):
message_type: Literal["group"]
sub_type: str
user_id: int
group_id: int
message_id: int
message: Message
raw_message: str
@ -289,12 +314,28 @@ class GroupMessageEvent(MessageEvent):
sender: Sender
anonymous: Anonymous
@overrides(CQHTTPEvent)
def get_event_description(self) -> str:
return (
f'Message {self.message_id} from {self.user_id}@[群:{self.group_id}] "'
+ "".join(
map(
lambda x: escape_tag(str(x))
if x.type == "text" else f"<le>{escape_tag(str(x))}</le>",
self.message)) + '"')
# Notice Events
class NoticeEvent(CQHTTPEvent):
post_type: Literal["notice"]
notice_type: str
@overrides(CQHTTPEvent)
def get_event_name(self) -> str:
sub_type = getattr(self, "sub_type", None)
return f"{self.post_type}.{self.notice_type}" + (f".{sub_type}"
if sub_type else "")
class GroupUploadNoticeEvent(NoticeEvent):
notice_type: Literal["group_upload"]
@ -378,6 +419,12 @@ class RequestEvent(CQHTTPEvent):
post_type: Literal["request"]
request_type: str
@overrides(CQHTTPEvent)
def get_event_name(self) -> str:
sub_type = getattr(self, "sub_type", None)
return f"{self.post_type}.{self.request_type}" + (f".{sub_type}"
if sub_type else "")
class FriendRequestEvent(RequestEvent):
request_type: Literal["friend"]
@ -395,10 +442,21 @@ class GroupRequestEvent(RequestEvent):
flag: str
# Meta Events
class MetaEvent(CQHTTPEvent):
post_type: Literal["meta_event"]
meta_event_type: str
@overrides(CQHTTPEvent)
def get_event_name(self) -> str:
sub_type = getattr(self, "sub_type", None)
return f"{self.post_type}.{self.meta_event_type}" + (f".{sub_type}" if
sub_type else "")
@overrides(CQHTTPEvent)
def get_log_string(self) -> str:
raise NoLogException
class LifecycleMetaEvent(MetaEvent):
meta_event_type: Literal["lifecycle"]

View File

@ -130,6 +130,15 @@ class AdapterException(NoneBotException):
self.adapter_name = adapter_name
class NoLogException(Exception):
"""
:说明:
指示 NoneBot 对当前 ``Event`` 进行处理但不显示 Log 信息可在 ``get_log_string`` 时抛出
"""
pass
class ApiNotAvailable(AdapterException):
"""
:说明: