AgentCoreService/message.py

#! /usr/bin/env python
# -*- coding: utf-8 -*-
# vim:fenc=utf-8


from enum import Enum, auto
from typing import Optional

from pydantic import BaseModel

class MessageType(Enum):
    TEXT = (1, "文本")
    AUDIO = (2, "音频")
    IMAGE = (3, "图片")
    VIDEO = (4, "视频")
    MINI_PROGRAM = (5, "小程序")
    LINK = (6, "链接")

    ACTIVE_TRIGGER = (101, "主动触发器")
    AGGREGATION_TRIGGER = (102, "消息聚合触发器")

    def __init__(self, code, description):
        self.code = code
        self.description = description

    def __repr__(self):
        return f"{self.__class__.__name__}.{self.name}"

class MessageChannel(Enum):
    CORP_WECHAT = (1, "企业微信")
    MINI_PROGRAM = (2, "小程序")

    SYSTEM = (101, "系统内部")

    def __init__(self, code, description):
        self.code = code
        self.description = description

    def __repr__(self):
        return f"{self.__class__.__name__}.{self.name}"

class Message(BaseModel):
     id: int
     type: MessageType
     channel: MessageChannel
     staff_id: Optional[str] = None
     user_id: str
     content: Optional[str] = None
     timestamp: int

     @staticmethod
     def build(type, channel, staff_id, user_id, content, timestamp):
         return Message(
             id=0,
             type=type,
             channel=channel,
             staff_id=staff_id,
             user_id=user_id,
             content=content,
             timestamp=timestamp
         )