nonebot2/nonebot/adapters/ding/event.py

146 lines
3.7 KiB
Python
Raw Normal View History

2020-12-30 00:36:29 +08:00
from enum import Enum
from typing import List, Optional
from typing_extensions import Literal
from pydantic import BaseModel, root_validator
2020-12-06 02:30:19 +08:00
2020-12-30 00:36:29 +08:00
from nonebot.typing import overrides
from nonebot.adapters import Event as BaseEvent
2020-12-03 00:59:32 +08:00
from .message import Message
class Event(BaseEvent):
"""
2020-12-31 17:58:09 +08:00
钉钉协议事件各事件字段参考 `钉钉文档`_
2020-12-30 00:36:29 +08:00
.. _钉钉文档:
https://ding-doc.dingtalk.com/document#/org-dev-guide/elzz1p
2020-12-03 00:59:32 +08:00
"""
2020-12-30 00:36:29 +08:00
chatbotUserId: str
@overrides(BaseEvent)
def get_type(self) -> Literal["message", "notice", "request", "meta_event"]:
raise ValueError("Event has no type!")
@overrides(BaseEvent)
def get_event_name(self) -> str:
raise ValueError("Event has no name!")
2020-12-03 00:59:32 +08:00
2020-12-30 00:36:29 +08:00
@overrides(BaseEvent)
def get_event_description(self) -> str:
raise ValueError("Event has no description!")
2020-12-30 00:36:29 +08:00
@overrides(BaseEvent)
def get_message(self) -> "Message":
raise ValueError("Event has no message!")
2020-12-30 00:36:29 +08:00
@overrides(BaseEvent)
def get_plaintext(self) -> str:
raise ValueError("Event has no plaintext!")
2020-12-03 00:59:32 +08:00
2020-12-30 00:36:29 +08:00
@overrides(BaseEvent)
def get_user_id(self) -> str:
raise ValueError("Event has no user_id!")
2020-12-03 00:59:32 +08:00
2020-12-30 00:36:29 +08:00
@overrides(BaseEvent)
def get_session_id(self) -> str:
raise ValueError("Event has no session_id!")
2020-12-03 00:59:32 +08:00
2020-12-30 00:36:29 +08:00
@overrides(BaseEvent)
def is_tome(self) -> bool:
return True
2020-12-03 00:59:32 +08:00
2020-12-30 00:36:29 +08:00
class TextMessage(BaseModel):
content: str
2020-12-03 00:59:32 +08:00
2020-12-30 00:36:29 +08:00
class AtUsersItem(BaseModel):
dingtalkId: str
staffId: Optional[str]
2020-12-03 00:59:32 +08:00
2020-12-30 00:36:29 +08:00
class ConversationType(str, Enum):
private = "1"
group = "2"
class MessageEvent(Event):
2020-12-31 17:58:09 +08:00
"""消息事件"""
2020-12-30 00:36:29 +08:00
msgtype: str
text: TextMessage
msgId: str
createAt: int # ms
conversationType: ConversationType
conversationId: str
senderId: str
senderNick: str
senderCorpId: str
sessionWebhook: str
sessionWebhookExpiredTime: int
isAdmin: bool
message: Message
@root_validator(pre=True)
def gen_message(cls, values: dict):
assert "msgtype" in values, "msgtype must be specified"
# 其实目前钉钉机器人只能接收到 text 类型的消息
assert values[
"msgtype"] in values, f"{values['msgtype']} must be specified"
content = values[values['msgtype']]['content']
# 如果是被 @,第一个字符将会为空格,移除特殊情况
if content[0] == ' ':
content = content[1:]
values["message"] = content
return values
2020-12-30 00:36:29 +08:00
@overrides(Event)
def get_type(self) -> Literal["message", "notice", "request", "meta_event"]:
return "message"
2020-12-30 20:08:22 +08:00
@overrides(Event)
2020-12-30 00:36:29 +08:00
def get_event_name(self) -> str:
return f"{self.get_type()}.{self.conversationType.name}"
2020-12-30 20:08:22 +08:00
@overrides(Event)
2020-12-30 00:36:29 +08:00
def get_event_description(self) -> str:
return f'Message[{self.msgtype}] {self.msgId} from {self.senderId} "{self.text.content}"'
2020-12-30 20:08:22 +08:00
@overrides(Event)
def get_message(self) -> Message:
return self.message
2020-12-30 20:08:22 +08:00
@overrides(Event)
2020-12-30 00:36:29 +08:00
def get_plaintext(self) -> str:
return self.text.content
2020-12-30 20:08:22 +08:00
@overrides(Event)
2020-12-30 00:36:29 +08:00
def get_user_id(self) -> str:
return self.senderId
2020-12-30 20:08:22 +08:00
@overrides(Event)
2020-12-30 00:36:29 +08:00
def get_session_id(self) -> str:
return self.senderId
class PrivateMessageEvent(MessageEvent):
2020-12-31 17:58:09 +08:00
"""私聊消息事件"""
2020-12-30 00:36:29 +08:00
chatbotCorpId: str
senderStaffId: Optional[str]
conversationType: ConversationType = ConversationType.private
2020-12-03 00:59:32 +08:00
2020-12-30 00:36:29 +08:00
class GroupMessageEvent(MessageEvent):
2020-12-31 17:58:09 +08:00
"""群消息事件"""
2020-12-30 00:36:29 +08:00
atUsers: List[AtUsersItem]
conversationType: ConversationType = ConversationType.group
conversationTitle: str
isInAtList: bool
2020-12-03 00:59:32 +08:00
2020-12-30 00:36:29 +08:00
@overrides(MessageEvent)
def is_tome(self) -> bool:
return self.isInAtList