wechatpy.messages 源代码

# -*- coding: utf-8 -*-
"""
    wechatpy.messages
    ~~~~~~~~~~~~~~~~~~

    This module defines all the messages you can get from WeChat server

    :copyright: (c) 2014 by messense.
    :license: MIT, see LICENSE for more details.
"""
import copy

from wechatpy.fields import BaseField, DateTimeField, FieldDescriptor, IntegerField, StringField

MESSAGE_TYPES = {}
COMPONENT_MESSAGE_TYPES = {}


def register_message(msg_type):
    def register(cls):
        MESSAGE_TYPES[msg_type] = cls
        return cls

    return register


def register_component_message(msg_type):
    def register(cls):
        COMPONENT_MESSAGE_TYPES[msg_type] = cls
        return cls

    return register


class MessageMetaClass(type):
    """Metaclass for all messages"""

    def __new__(mcs, name, bases, attrs):
        for b in bases:
            if not hasattr(b, "_fields"):
                continue

            for k, v in b.__dict__.items():
                if k in attrs:
                    continue
                if isinstance(v, FieldDescriptor):
                    attrs[k] = copy.deepcopy(v.field)

        mcs = super().__new__(mcs, name, bases, attrs)
        mcs._fields = {}

        for name, field in mcs.__dict__.items():
            if isinstance(field, BaseField):
                field.add_to_class(mcs, name)
        return mcs


class BaseMessage(metaclass=MessageMetaClass):
    """Base class for all messages and events"""

    type = "unknown"
    id = IntegerField("MsgId", 0)
    source = StringField("FromUserName")
    target = StringField("ToUserName")
    create_time = DateTimeField("CreateTime")
    time = IntegerField("CreateTime")

    def __init__(self, message):
        self._data = message

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


[文档]@register_message("text") class TextMessage(BaseMessage): """ 文本消息 详情请参阅 https://developers.weixin.qq.com/doc/offiaccount/Message_Management/Receiving_standard_messages.html """ type = "text" content = StringField("Content")
[文档]@register_message("image") class ImageMessage(BaseMessage): """ 图片消息 详情请参阅 https://developers.weixin.qq.com/doc/offiaccount/Message_Management/Receiving_standard_messages.html """ type = "image" media_id = StringField("MediaId") image = StringField("PicUrl")
[文档]@register_message("voice") class VoiceMessage(BaseMessage): """ 语音消息 详情请参阅 https://developers.weixin.qq.com/doc/offiaccount/Message_Management/Receiving_standard_messages.html """ type = "voice" media_id = StringField("MediaId") format = StringField("Format") recognition = StringField("Recognition")
[文档]@register_message("shortvideo") class ShortVideoMessage(BaseMessage): """ 短视频消息 详情请参阅 https://developers.weixin.qq.com/doc/offiaccount/Message_Management/Receiving_standard_messages.html """ type = "shortvideo" media_id = StringField("MediaId") thumb_media_id = StringField("ThumbMediaId")
[文档]@register_message("video") class VideoMessage(BaseMessage): """ 视频消息 详情请参阅 https://developers.weixin.qq.com/doc/offiaccount/Message_Management/Receiving_standard_messages.html """ type = "video" media_id = StringField("MediaId") thumb_media_id = StringField("ThumbMediaId")
[文档]@register_message("location") class LocationMessage(BaseMessage): """ 地理位置消息 详情请参阅 https://developers.weixin.qq.com/doc/offiaccount/Message_Management/Receiving_standard_messages.html """ type = "location" location_x = StringField("Location_X") location_y = StringField("Location_Y") scale = StringField("Scale") label = StringField("Label") @property def location(self): return self.location_x, self.location_y
[文档]@register_message("link") class LinkMessage(BaseMessage): """ 链接消息 详情请参阅 https://developers.weixin.qq.com/doc/offiaccount/Message_Management/Receiving_standard_messages.html """ type = "link" title = StringField("Title") description = StringField("Description") url = StringField("Url")
@register_message("miniprogrampage") class MiniProgramPageMessage(BaseMessage): """ 小程序卡片消息 详情请参阅 https://developers.weixin.qq.com/miniprogram/dev/framework/open-ability/customer-message/receive.html#小程序卡片消息 """ type = "miniprogrampage" app_id = StringField("AppId") title = StringField("Title") page_path = StringField("PagePath") thumb_url = StringField("ThumbUrl") thumb_media_id = StringField("ThumbMediaId") class UnknownMessage(BaseMessage): """未知消息类型""" pass class BaseComponentMessage(metaclass=MessageMetaClass): """Base class for all component messages and events""" type = "unknown" appid = StringField("AppId") create_time = DateTimeField("CreateTime") def __init__(self, message): self._data = message def __repr__(self): return f"{self.__class__.__name__}({repr(self._data)})" @register_component_message("component_verify_ticket") class ComponentVerifyTicketMessage(BaseComponentMessage): """ component_verify_ticket协议 """ type = "component_verify_ticket" verify_ticket = StringField("ComponentVerifyTicket") @register_component_message("unauthorized") class ComponentUnauthorizedMessage(BaseComponentMessage): """ 取消授权通知 """ type = "unauthorized" authorizer_appid = StringField("AuthorizerAppid") @register_component_message("authorized") class ComponentAuthorizedMessage(BaseComponentMessage): """ 新增授权通知 """ type = "authorized" authorizer_appid = StringField("AuthorizerAppid") authorization_code = StringField("AuthorizationCode") authorization_code_expired_time = StringField("AuthorizationCodeExpiredTime") pre_auth_code = StringField("PreAuthCode") @register_component_message("updateauthorized") class ComponentUpdateAuthorizedMessage(BaseComponentMessage): """ 更新授权通知 """ type = "updateauthorized" authorizer_appid = StringField("AuthorizerAppid") authorization_code = StringField("AuthorizationCode") authorization_code_expired_time = StringField("AuthorizationCodeExpiredTime") pre_auth_code = StringField("PreAuthCode") class ComponentUnknownMessage(BaseComponentMessage): """ 未知通知 """ type = "unknown"