mirror of
https://github.com/nonebot/nonebot2.git
synced 2024-11-25 01:25:04 +08:00
113 lines
3.0 KiB
Python
113 lines
3.0 KiB
Python
import json
|
|
from enum import Enum
|
|
from typing import Any, Dict, Optional, Type
|
|
|
|
from pydantic import BaseModel, Field, ValidationError
|
|
from typing_extensions import Literal
|
|
|
|
from nonebot.adapters import Event as BaseEvent
|
|
from nonebot.adapters import Message as BaseMessage
|
|
from nonebot.log import logger
|
|
from nonebot.typing import overrides
|
|
|
|
|
|
class SenderPermission(str, Enum):
|
|
OWNER = 'OWNER'
|
|
ADMINISTRATOR = 'ADMINISTRATOR'
|
|
MEMBER = 'MEMBER'
|
|
|
|
|
|
class SenderGroup(BaseModel):
|
|
id: int
|
|
name: str
|
|
permission: SenderPermission
|
|
|
|
|
|
class SenderInfo(BaseModel):
|
|
id: int
|
|
name: str = Field(alias='memberName')
|
|
permission: SenderPermission
|
|
group: SenderGroup
|
|
|
|
|
|
class PrivateSenderInfo(BaseModel):
|
|
id: int
|
|
nickname: str
|
|
remark: str
|
|
|
|
|
|
class Event(BaseEvent):
|
|
self_id: int
|
|
type: str
|
|
|
|
@classmethod
|
|
def new(cls, data: Dict[str, Any]) -> "Event":
|
|
type = data['type']
|
|
|
|
def all_subclasses(cls: Type[Event]):
|
|
return set(cls.__subclasses__()).union(
|
|
[s for c in cls.__subclasses__() for s in all_subclasses(c)])
|
|
|
|
event_class: Optional[Type[Event]] = None
|
|
for subclass in all_subclasses(cls):
|
|
if subclass.__name__ != type:
|
|
continue
|
|
event_class = subclass
|
|
|
|
if event_class is None:
|
|
return Event.parse_obj(data)
|
|
|
|
while issubclass(event_class, Event):
|
|
try:
|
|
return event_class.parse_obj(data)
|
|
except ValidationError as e:
|
|
logger.info(
|
|
f'Failed to parse {data} to class {event_class.__name__}: {e}. '
|
|
'Fallback to parent class.')
|
|
event_class = event_class.__base__
|
|
|
|
raise ValueError(f'Failed to serialize {data}.')
|
|
|
|
@overrides(BaseEvent)
|
|
def get_type(self) -> Literal["message", "notice", "request", "meta_event"]:
|
|
from . import message, notice, request
|
|
if isinstance(self, message.MessageEvent):
|
|
return 'message'
|
|
elif isinstance(self, notice.NoticeEvent):
|
|
return 'notice'
|
|
elif isinstance(self, request.RequestEvent):
|
|
return 'request'
|
|
else:
|
|
return 'meta_event'
|
|
|
|
@overrides(BaseEvent)
|
|
def get_event_name(self) -> str:
|
|
return self.type
|
|
|
|
@overrides(BaseEvent)
|
|
def get_event_description(self) -> str:
|
|
return str(self.dict())
|
|
|
|
@overrides(BaseEvent)
|
|
def get_message(self) -> BaseMessage:
|
|
raise ValueError("Event has no message!")
|
|
|
|
@overrides(BaseEvent)
|
|
def get_plaintext(self) -> str:
|
|
raise ValueError("Event has no message!")
|
|
|
|
@overrides(BaseEvent)
|
|
def get_user_id(self) -> str:
|
|
raise ValueError("Event has no message!")
|
|
|
|
@overrides(BaseEvent)
|
|
def get_session_id(self) -> str:
|
|
raise ValueError("Event has no message!")
|
|
|
|
@overrides(BaseEvent)
|
|
def is_tome(self) -> bool:
|
|
return False
|
|
|
|
def normalize_dict(self, **kwargs) -> Dict[str, Any]:
|
|
return json.loads(self.json(**kwargs))
|