werobot里的一些python代码

随便点开看的微信公众号平台python客户端代码werobot, 另一个是wechatpy. 很久没写python了, 看开源代码总能发现一些有意思的写法.

https://github.com/offu/WeRoBot

python metaclass

看起来使用metaclass, 可以对所有后续的实例初始化过程进行统一管理, 修改属性添加函数或是收集起来都没问题.

这个metaclass, 介入了子类的初始化流程. 看起来是在类初始化的时候, 根据类参数里的type名收集到同一个dict里, 实例本身作为dict value中的一个元素. 估计可以实现这种场景需求: 目标需要一个type类型来处理, 用metaclass可以遍历所有实例, 逐个处理一遍.

metaclass的定义, 看起来需要继承type类. 这里的type.__new__和type.__init__可以修改为super().__, 感觉就不会那么突兀.

class WeRoBotMetaClass(type):
    TYPES = {}

    def __new__(mcs, name, bases, attrs):
        return type.__new__(mcs, name, bases, attrs)

    def __init__(cls, name, bases, attrs):
        if '__type__' in attrs:
            if isinstance(attrs['__type__'], list):
                for _type in attrs['__type__']:
                    cls.TYPES[_type] = cls
            else:
                cls.TYPES[attrs['__type__']] = cls
        type.__init__(cls, name, bases, attrs)

子类的定义:

from werobot.messages.entries import StringEntry, IntEntry, FloatEntry
from werobot.messages.base import WeRoBotMetaClass


class MessageMetaClass(WeRoBotMetaClass):
    pass


class WeChatMessage(object, metaclass=MessageMetaClass):
    message_id = IntEntry('MsgId', 0)
    target = StringEntry('ToUserName')
    source = StringEntry('FromUserName')
    time = IntEntry('CreateTime', 0)

    def __init__(self, message):
        self.__dict__.update(message)


class TextMessage(WeChatMessage):
    __type__ = 'text'
    content = StringEntry('Content')


class ImageMessage(WeChatMessage):
    __type__ = 'image'
    img = StringEntry('PicUrl')

from werobot.messages.entries import StringEntry, IntEntry, FloatEntry
from werobot.messages.base import WeRoBotMetaClass


class EventMetaClass(WeRoBotMetaClass):
    pass


class WeChatEvent(object, metaclass=EventMetaClass):
    target = StringEntry('ToUserName')
    source = StringEntry('FromUserName')
    time = IntEntry('CreateTime')
    message_id = IntEntry('MsgID', 0)

    def __init__(self, message):
        self.__dict__.update(message)


class SimpleEvent(WeChatEvent):
    key = StringEntry('EventKey')

用法, 看起来是设置为decorator, 套用在目标函数上, 以后遇到这种类型的消息就会被调用处理.

class BaseRoBot(object):
    """
    BaseRoBot 是整个应用的核心对象，负责提供 handler 的维护，消息和事件的处理等核心功能。
    """
    message_types = [
        'subscribe_event',
        'unsubscribe_event',
        'click_event',
        'view_event',,
        'submit_membercard_user_info_event',  # event
        'text',
        'image',
        'link',
        'location',
        'voice',
        'unknown',
        'video',
        'shortvideo'
        ...
    ]

        def __init__(
        self,
        token=None,
        logger=None,
        enable_session=None,
        session_storage=None,
        app_id=None,
        app_secret=None,
        encoding_aes_key=None,
        config=None,
        **kwargs
    ):

        self._handlers = {k: [] for k in self.message_types}
        self._handlers['all'] = []
        self.make_error_page = make_error_page


    def add_handler(self, func, type='all'):
        """
        为 BaseRoBot 实例添加一个 handler。

        :param func: 要作为 handler 的方法。
        :param type: handler 的种类。
        :return: None
        """
        if not callable(func):
            raise ValueError("{} is not callable".format(func))

        self._handlers[type].append(
            (func, len(signature(func).parameters.keys()))
        )

    def get_handlers(self, type):
        return self._handlers.get(type, []) + self._handlers['all']

        def handler(self, f):
        """
        为每一条消息或事件添加一个 handler 方法的装饰器。
        """
        self.add_handler(f, type='all')
        return f

    def text(self, f):
        """
        为文本 ``(text)`` 消息添加一个 handler 方法的装饰器。
        """
        self.add_handler(f, type='text')
        return f

    def image(self, f):
        """
        为图像 ``(image)`` 消息添加一个 handler 方法的装饰器。
        """
        self.add_handler(f, type='image')

获取消息返回的时候, 从handler里获取已经定义的注册方法, 然后一个个调用.

    def get_reply(self, message):
        """
        根据 message 的内容获取 Reply 对象。

        :param message: 要处理的 message
        :return: 获取的 Reply 对象
        """
        session_storage = self.session_storage

        id = None
        session = None
        if session_storage and hasattr(message, "source"):
            id = to_binary(message.source)
            session = session_storage[id]

        handlers = self.get_handlers(message.type)
        try:
            for handler, args_count in handlers:
                args = [message, session][:args_count]
                reply = handler(*args)
                if session_storage and id:
                    session_storage[id] = session
                if reply:
                    return process_function_reply(reply, message=message)
        except:
            self.logger.exception("Catch an exception")

最简单的robot应用示例, 使用@robot.text定义text类型需要调用的处理方法

from django.conf.urls import url
from django.contrib import admin
from werobot import WeRoBot
from werobot.contrib.django import make_view
from werobot.utils import generate_token

robot = WeRoBot(
    SESSION_STORAGE=False,
    token="TestDjango",
    app_id="9998877",
    encoding_aes_key=generate_token(32)
)


@robot.text
def text_handler():
    return 'hello'


@robot.error_page
def make_error_page(url):
    return '喵'


urlpatterns = [
    url(r'^admin/', admin.site.urls),
    url(r'^robot/', make_view(robot))
]

@cache_property的使用

这个写法也挺有意思, 将类的某个方法的运行结果缓存起来, 实现一个方法只需要一次性调用. 缓存的思路挺有意思, 把方法名当作实例本身的参数, 方法的计算结果作为参数值.

一般我们都没有这种需求, 因为这种需要一次性初始化的场景, 直接在__init__函数里调用就可以了, 除非初始化耗时实在太长了.

这里面还用到propert进行包裹, 这样获取函数计算结果的时候, 就不用额外调用 object.method(), 可以直接当作参数来获取了.

def cached_property(method):
    prop_name = '_{}'.format(method.__name__)

    @wraps(method)
    def wrapped_func(self, *args, **kwargs):
        if not hasattr(self, prop_name):
            setattr(self, prop_name, method(self, *args, **kwargs))
        return getattr(self, prop_name)

    return property(wrapped_func)

使用示例

class BaseRoBot(object):


    def __init__(
        self,
        token=None,
        logger=None,
        enable_session=None,
        session_storage=None,
        app_id=None,
        app_secret=None,
        encoding_aes_key=None,
        config=None,
        **kwargs
    ):
    ...


    @cached_property
    def crypto(self):
        app_id = self.config.get("APP_ID", None)
        if not app_id:
            raise ConfigError(
                "You need to provide app_id to encrypt/decrypt messages"
            )

        encoding_aes_key = self.config.get("ENCODING_AES_KEY", None)
        if not encoding_aes_key:
            raise ConfigError(
                "You need to provide encoding_aes_key "
                "to encrypt/decrypt messages"
            )
        self.use_encryption = True

        from .crypto import MessageCrypt
        return MessageCrypt(
            token=self.config["TOKEN"],
            encoding_aes_key=encoding_aes_key,
            app_id=app_id
        )

    @cached_property
    def client(self):
        return Client(self.config)

class WeRoBot(BaseRoBot):
    """
    WeRoBot 是一个继承自 BaseRoBot 的对象，在 BaseRoBot 的基础上使用了 bottle 框架，
    提供接收微信服务器发来的请求的功能。
    """
    @cached_property
    def wsgi(self):
        if not self._handlers:
            raise RuntimeError('No Handler.')
        from bottle import Bottle
        from werobot.contrib.bottle import make_view

        app = Bottle()
        app.route('<t:path>', ['GET', 'POST'], make_view(self))
        return app

    def run(
        self, server=None, host=None, port=None, enable_pretty_logging=True
    ):
        """
        运行 WeRoBot。

        :param server: 传递给 Bottle 框架 run 方法的参数，详情见\
        `bottle 文档 <https://bottlepy.org/docs/dev/deployment.html#switching-the-server-backend>`_
        :param host: 运行时绑定的主机地址
        :param port: 运行时绑定的主机端口
        :param enable_pretty_logging: 是否开启 log 的输出格式优化
        """
        if enable_pretty_logging:
            from werobot.logger import enable_pretty_logging
            enable_pretty_logging(self.logger)
        if server is None:
            server = self.config["SERVER"]
        if host is None:
            host = self.config["HOST"]
        if port is None:
            port = self.config["PORT"]
        try:
            self.wsgi.run(server=server, host=host, port=port)
        except KeyboardInterrupt:
            exit(0)