nonebot2/nonebot/adapters/ding/event.py

from enum import Enum
from typing import List, Optional
from typing_extensions import Literal

from pydantic import BaseModel

from nonebot.utils import escape_tag
from nonebot.typing import overrides
from nonebot.adapters import Event as BaseEvent

from .message import Message


class Event(BaseEvent):
    """
    钉钉 协议 Event 适配。各事件字段参考 `钉钉文档`_

    .. _钉钉文档:
        https://ding-doc.dingtalk.com/document#/org-dev-guide/elzz1p
    """

    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 type!")

    @overrides(BaseEvent)
    def get_event_description(self) -> str:
        raise ValueError("Event has no type!")

    @overrides(BaseEvent)
    def get_message(self) -> "Message":
        raise ValueError("Event has no type!")

    @overrides(BaseEvent)
    def get_plaintext(self) -> str:
        raise ValueError("Event has no type!")

    @overrides(BaseEvent)
    def get_user_id(self) -> str:
        raise ValueError("Event has no type!")

    @overrides(BaseEvent)
    def get_session_id(self) -> str:
        raise ValueError("Event has no type!")

    @overrides(BaseEvent)
    def is_tome(self) -> bool:
        return True


class TextMessage(BaseModel):
    content: str


class AtUsersItem(BaseModel):
    dingtalkId: str
    staffId: Optional[str]


class ConversationType(str, Enum):
    private = "1"
    group = "2"


class MessageEvent(Event):
    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

    @overrides(Event)
    def get_type(self) -> Literal["message", "notice", "request", "meta_event"]:
        return "message"

    @overrides(BaseEvent)
    def get_event_name(self) -> str:
        return f"{self.get_type()}.{self.conversationType.name}"

    @overrides(BaseEvent)
    def get_event_description(self) -> str:
        return f'Message[{self.msgtype}] {self.msgId} from {self.senderId} "{self.text.content}"'

    @overrides(BaseEvent)
    def get_plaintext(self) -> str:
        return self.text.content

    @overrides(BaseEvent)
    def get_user_id(self) -> str:
        return self.senderId

    @overrides(BaseEvent)
    def get_session_id(self) -> str:
        return self.senderId


class PrivateMessageEvent(MessageEvent):
    chatbotCorpId: str
    senderStaffId: Optional[str]
    conversationType: ConversationType = ConversationType.private


class GroupMessageEvent(MessageEvent):
    atUsers: List[AtUsersItem]
    conversationType: ConversationType = ConversationType.group
    conversationTitle: str
    isInAtList: bool

    @overrides(MessageEvent)
    def is_tome(self) -> bool:
        return self.isInAtList