Python 面向对象设计 - 腾讯即时通信以及微信示例

仅供学习,转载请注明出处

面向对象设计

  • 继承 - 是基于Python中的属性查找(如X.name)
  • 多态 - 在X.method方法中,method的意义取决于X的类型
  • 封装 - 方法和运算符实现行为,数据隐藏默认是一种惯例

参考实例

腾讯即时通信模块,初级封装

#! /usr/bin/env python
# coding: utf-8

import random
import time


class Message(object):

    def __init__(self, msgarr=[], toacc=''):
        self.msgbody = msgarr # 此处为MsgDict对象实例的列表或者空列表
        self.toacc = toacc # toacc为字符串(单发)或者列表(批量发)
        self.msgrandom = random.randint(1, 1000000000)
        self.msgrequest = {
            'To_Account': toacc, # 消息接收方账号
            'MsgRandom': self.msgrandom, # 消息随机数,由随机函数产生
            'MsgBody': [t.msg for t in msgarr]
        }

    def del_option(self, option):
        if option in (set(self.msgrequest)-set(['To_Account', 'MsgRandom', 'MsgBody'])):
            self.__dict__.pop(option)
            self.msgrequest.pop(option)

    def append_msg(self, msg):
        self.msgbody.append(msg)
        self.msgrequest['MsgBody'].append(msg.msg)

    def insert_msg(self, index, msg):
        self.msgbody.insert(index, msg)
        self.msgrequest['MsgBody'].insert(msg.msg)

    def del_msg(self, index):
        if index in range(len(self.msgbody)):
            del self.msgbody[index]
            del sel.msgrequest['MsgBody'][index]

    def set_from(self, fromacc):
        # 指定消息的发送方,默认为服务器发送
        self.fromacc = fromacc
        self.msgrequest['From_Account'] = fromacc

    def set_to(self, toacc):
        # 指定消息的接收方,可以为String(单发),可以为List(批量发送)
        self.toacc = toacc
        self.msgrequest['To_Account'] = toacc

    def refresh_random(self):
        self.msgrandom = random.randint(1, 1000000000)
        self.msgrequest['MsgRandom'] = self.msgrandom, # 消息随机数,由随机函数产生

    def set_sync(self, sync):
        # 同步选项:1, 把消息同步到From_Account在线终端和漫游上
        #           2, 消息不同步至From_Account
        #           若不填写,默认情况下会将消息同步
        #           仅在单发单聊消息中可调用
        self.sync = sync
        self.msgrequest['SyncOtherMachine'] = sync

    def set_timestamp(self):
        # 设置消息时间戳,unix时间, 仅在单发单聊消息中可以调用
        self.timestamp = int(time.time())
        self.msgrequest['MsgTimeStamp'] = self.timestamp

    def set_offlinepush(self, pushflag=0, desc='', ext='', sound=''):
        # 仅适用于APNa推送,不适用于安卓厂商推送
        self.msgrequest['OfflinePushInfo'] = {
            'PushFlag': pushflag,
            'Desc': desc,
            'Ext': ext,
            'Sound': sound
        }


class MsgDict(object):

    def __init__(self, msgtype='', msgcontent={}):
        self.msgtype = msgtype
        self.msgcontent = msgcontent

    @property
    def msg(self):
        return {
            'MsgType': self.msgtype,
            'MsgContent': self.msgcontent
        } 

    def set_content(self, content):
        self.msgcontent = content


class TextMsg(MsgDict):

    def __init__(self, text='', msgtype='TIMTextElem'): 
        self.text = text
        content = {'Text': text}
        super(TextMsg, self).__init__(msgtype=msgtype, msgcontent=content)

    def set_text(self, text):
        self.text = text
        self.msgcontent['Text'] = text


class LocationMsg(MsgDict):

    def __init__(self, desc='', latitude=0, longitude=0, msgtype='TIMLocationElem'): 
        self.desc = desc
        self.latitude = latitude
        self.longitude = longitude
        content = {
            'Desc': desc,  # 地理位置描述信息, String
            'Latitude': latitude, # 纬度, Number
            'Longitude': longitude # 经度, Number
        }
        super(LocationMsg, self).__init__(msgtype=msgtype, msgcontent=content)

    def set_desc(self, desc):
        self.desc = desc
        self.msgcontent['Desc'] = desc

    def set_location(self, latitude, longitude):
        self.latitude = latitude
        self.longitude = longitude
        self.msgcontent['Latitude'] = latitude
        self.msgcontent['Longitude'] = longitude

    def set_latitude(self, latitude):
        self.latitude = latitude
        self.msgcontent['Latitude'] = latitude

    def set_longitude(self, longitude):
        self.longitude = longitude
        self.msgcontent['Longitude'] = longitude


class FaceMsg(MsgDict):

    def __init__(self, index=1, data='', msgtype='TIMFaceElem'): 
        self.index = index 
        self.data = data
        content = {
            'Index': index, # 表情索引,用户自定义, Number
            'Data': data # 额外数据, String
        }
        super(TextMsg, self).__init__(msgtype=msgtype, msgcontent=content)

    def set_index(self, index):
        self.index = index
        self.msgcontent['Index'] = index

    def set_data(self, data):
        self.data = data
        self.msgcontent['Data'] = data


class CustomMsg(MsgDict):

    def __init__(self, data='', desc='', ext='', sound='', msgtype='TIMCustomElem'): 
        self.data = data
        self.desc = desc
        self.ext = ext
        self.sound = sound
        content = {
            'Data': data, # 自定义消息数据。不作为APNS的payload中字段下发,故从payload中无法获取Data字段, String
            'Desc': desc, # 自定义消息描述,当接收方为iphone后台在线时,做ios离线Push时文本展示
            'Ext': ext, # 扩展字段,当接收方为ios系统且应用处在后台时,此字段作为APNS请求包Payloads中的ext键值下发,Ext的协议格式由业务方确定,APNS只做透传
            'Sound': sound # 自定义APNS推送铃声
        }
        super(CustomMsg, self).__init__(msgtype=msgtype, msgcontent=content)

    def set_data(self, data):
        self.data = data
        self.msgcontent['Data'] = data

    def set_desc(self, desc):
        self.desc = desc
        self.msgcontent['Desc'] = desc

    def set_ext(self, ext):
        self.ext = ext
        self.msgcontent['Ext'] = ext

    def set_sound(self, sound):
        self.sound = sound
        self.msgcontent['Sound'] = sound


class SoundMsg(MsgDict):

    def __init__(self, uuid='', size=0, second=0, msgtype='TIMSoundElem'): 
        self.uuid = uuid 
        self.size = size 
        self.second = second
        content = {
            'UUID': uuid, # 语音序列号,后台用于索引语音的键值,String
            'Size': size, # 语音数据大小, Number 
            'Second': second # 语音时长,单位秒 Number 
        }
        super(SoundMsg, self).__init__(msgtype=msgtype, msgcontent=content)

    def set_uuid(self, uuid):
        self.uuid = uuid
        self.msgcontent['UUID'] = uuid

    def set_size(self, size):
        self.size = size
        self.msgcontent['Size'] = size

    def set_second(self, second):
        self.second = second
        self.msgcontent['Second'] = second


class ImageMsg(MsgDict):

    def __init__(self, uuid='', imgformat=0, imginfo=[], msgtype='TIMImageElem'): 
        self.uuid = uuid 
        self.imgformat = imgformat 
        self.imginfo = imginfo 
        content = {
            'UUID': uuid, # 图片序列号,后台用于索引语音的键值,String
            'ImageFormat': imgformat, # 图片格式, BMP=1, JPG=2, GIF=3, 其他=0, Number 
            'ImageInfoArray': [t.info for t in imginfo] # 原图,缩略图或者大图下载信息, Array
        }
        super(ImageMsg, self).__init__(msgtype=msgtype, msgcontent=content)

    def set_uuid(self, uuid):
        self.uuid = uuid
        self.msgcontent['UUID'] = uuid

    def set_format(self, imgformat):
        self.imgformat = imgformat
        self.msgcontent['ImageFormat'] = imgformat

    def append_info(self, info):
        # info 为ImageInfo对象实例
        self.imginfo.append(info)
        self.msgcontnet['ImageInfoArray'].append(info.info)

    def insert_info(self, index, info):
        self.imginfo.insert(index, info)
        self.msgcontent['ImageInfoArray'].insert(index, info.info)

    def del_info(self, index):
        del self.imginfo[index]
        del self.msgcontent['ImageInfoArray'][index]


class FileMsg(MsgDict):

    def __init__(self, uuid='', size=0, name='', msgtype='TIMFileElem'): 
        self.uuid = uuid 
        self.size = size 
        self.name = name
        content = {
            'UUID': uuid, # 文件序列号,后台用于索引语音的键值,String
            'FileSize': size, # 文件数据大小, Number 
            'FileName': name # 文件名称/路径, String
        }
        super(FileMsg, self).__init__(msgtype=msgtype, msgcontent=content)

    def set_uuid(self, uuid):
        self.uuid = uuid
        self.msgcontent['UUID'] = UUID

    def set_size(self, size):
        self.size = size
        self.msgcontent['FileSize'] = size

    def set_name(self, name):
        self.name = name
        self.msgcontent['FileName'] = name


class ImageInfo(object):

    def __init__(self, itype=1, size=0, width=0, height=0, url=''):
        #图片类型, 1-原图, 2-大图, 3-缩略图, 0-其他
        self.itype = itype 
        # 图片数据大小,Number
        self.size = size
        # 图片宽度,Number
        self.width = width
        # 图片高度, Number
        self.height = height
        # 图片下载地址,String
        self.url = url

    @property
    def info(self):
        return {
            'Type': self.itype, 
            'Size': self.size,  
            'Width': self.width,  
            'Height': self.height, 
            'URL': self.url 
        }

    def set_type(self, itype):
        self.itype = itype

    def set_size(self, size):
        self.size = size

    def set_width(self, width):
        self.width = width

    def set_height(self, height):
        self.height = height

    def set_url(self, url):
        self.url = url

微信开发包,python实现, wechat_sdk开发

http://wechat-python-sdk.com/

截取部分代码,学习类的设计

from __future__ import unicode_literals

import time

from wechat_sdk.lib.crypto import BasicCrypto
from wechat_sdk.lib.request import WechatRequest
from wechat_sdk.exceptions import NeedParamError
from wechat_sdk.utils import disable_urllib3_warning


class WechatConf(object):
    """ WechatConf 配置类

    该类将会存储所有和微信开发相关的配置信息, 同时也会维护配置信息的有效性.
    """

    def __init__(self, **kwargs):
        """
        :param kwargs: 配置信息字典, 可用字典 key 值及对应解释如下:
                       'token': 微信 Token

                       'appid': App ID
                       'appsecret': App Secret

                       'encrypt_mode': 加解密模式 ('normal': 明文模式, 'compatible': 兼容模式, 'safe': 安全模式(默认))
                       'encoding_aes_key': EncodingAESKey 值 (传入此值必须保证同时传入 token, appid, 否则抛出异常)

                       'access_token_getfunc': access token 获取函数 (用于单机及分布式环境下, 具体格式参见文档)
                       'access_token_setfunc': access token 写入函数 (用于单机及分布式环境下, 具体格式参见文档)
                       'access_token_refreshfunc': access token 刷新函数 (用于单机及分布式环境下, 具体格式参见文档)
                       'access_token': 直接导入的 access token 值, 该值需要在上一次该类实例化之后手动进行缓存并在此处传入, 如果不
                                       传入, 将会在需要时自动重新获取 (传入 access_token_getfunc 和 access_token_setfunc 函数
                                       后将会自动忽略此处的传入值)
                       'access_token_expires_at': 直接导入的 access token 的过期日期, 该值需要在上一次该类实例化之后手动进行缓存
                                                  并在此处传入, 如果不传入, 将会在需要时自动重新获取 (传入 access_token_getfunc
                                                  和 access_token_setfunc 函数后将会自动忽略此处的传入值)

                       'jsapi_ticket_getfunc': jsapi ticket 获取函数 (用于单机及分布式环境下, 具体格式参见文档)
                       'jsapi_ticket_setfunc': jsapi ticket 写入函数 (用于单机及分布式环境下, 具体格式参见文档)
                       'jsapi_ticket_refreshfunc': jsapi ticket 刷新函数 (用于单机及分布式环境下, 具体格式参见文档)
                       'jsapi_ticket': 直接导入的 jsapi ticket 值, 该值需要在上一次该类实例化之后手动进行缓存并在此处传入, 如果不
                                       传入, 将会在需要时自动重新获取 (传入 jsapi_ticket_getfunc 和 jsapi_ticket_setfunc 函数
                                       后将会自动忽略此处的传入值)
                       'jsapi_ticket_expires_at': 直接导入的 jsapi ticket 的过期日期, 该值需要在上一次该类实例化之后手动进行缓存
                                                  并在此处传入, 如果不传入, 将会在需要时自动重新获取 (传入 jsapi_ticket_getfunc
                                                  和 jsapi_ticket_setfunc 函数后将会自动忽略此处的传入值)

                       'partnerid': 财付通商户身份标识, 支付权限专用
                       'partnerkey': 财付通商户权限密钥 Key, 支付权限专用
                       'paysignkey': 商户签名密钥 Key, 支付权限专用

                       'checkssl': 是否检查 SSL, 默认不检查 (False), 可避免 urllib3 的 InsecurePlatformWarning 警告
        :return:
        """

        self.__request = WechatRequest()

        if kwargs.get('checkssl') is not True:
            disable_urllib3_warning()  # 可解决 InsecurePlatformWarning 警告

        self.__token = kwargs.get('token')

        self.__appid = kwargs.get('appid')
        self.__appsecret = kwargs.get('appsecret')

        self.__encrypt_mode = kwargs.get('encrypt_mode', 'safe')
        self.__encoding_aes_key = kwargs.get('encoding_aes_key')
        self.__crypto = None
        self._update_crypto()

        self.__access_token_getfunc = kwargs.get('access_token_getfunc')
        self.__access_token_setfunc = kwargs.get('access_token_setfunc')
        self.__access_token_refreshfunc = kwargs.get('access_token_refreshfunc')
        self.__access_token = kwargs.get('access_token')
        self.__access_token_expires_at = kwargs.get('access_token_expires_at')

        self.__jsapi_ticket_getfunc = kwargs.get('jsapi_ticket_getfunc')
        self.__jsapi_ticket_setfunc = kwargs.get('jsapi_ticket_setfunc')
        self.__jsapi_ticket_refreshfunc = kwargs.get('jsapi_ticket_refreshfunc')
        self.__jsapi_ticket = kwargs.get('jsapi_ticket')
        self.__jsapi_ticket_expires_at = kwargs.get('jsapi_ticket_expires_at')

        self.__partnerid = kwargs.get('partnerid')
        self.__partnerkey = kwargs.get('partnerkey')
        self.__paysignkey = kwargs.get('paysignkey')

    @property
    def token(self):
        """ 获取当前 Token """
        self._check_token()
        return self.__token

    @token.setter
    def token(self, token):
        """ 设置当前 Token """
        self.__token = token
        self._update_crypto()  # 改动 Token 需要重新更新 Crypto

    @property
    def appid(self):
        """ 获取当前 App ID """
        return self.__appid

    @property
    def appsecret(self):
        """ 获取当前 App Secret """
        return self.__appsecret

    def set_appid_appsecret(self, appid, appsecret):
        """ 设置当前 App ID 及 App Secret"""
        self.__appid = appid
        self.__appsecret = appsecret
        self._update_crypto()  # 改动 App ID 后需要重新更新 Crypto

    @property
    def encoding_aes_key(self):
        """ 获取当前 EncodingAESKey """
        return self.__encoding_aes_key

    @encoding_aes_key.setter
    def encoding_aes_key(self, encoding_aes_key):
        """ 设置当前 EncodingAESKey """
        self.__encoding_aes_key = encoding_aes_key
        self._update_crypto()  # 改动 EncodingAESKey 需要重新更新 Crypto

    @property
    def encrypt_mode(self):
        return self.__encrypt_mode

    @encrypt_mode.setter
    def encrypt_mode(self, encrypt_mode):
        """ 设置当前加密模式 """
        self.__encrypt_mode = encrypt_mode
        self._update_crypto()

    @property
    def crypto(self):
        """ 获取当前 Crypto 实例 """
        return self.__crypto

    @property
    def access_token(self):
        """ 获取当前 access token 值, 本方法会自行维护 access token 有效性 """
        self._check_appid_appsecret()

        if callable(self.__access_token_getfunc):
            self.__access_token, self.__access_token_expires_at = self.__access_token_getfunc()

        if self.__access_token:
            now = time.time()
            if self.__access_token_expires_at - now > 60:
                return self.__access_token

        self.grant_access_token()  # 从腾讯服务器获取 access token 并更新
        return self.__access_token

    @property
    def jsapi_ticket(self):
        """ 获取当前 jsapi ticket 值, 本方法会自行维护 jsapi ticket 有效性 """
        self._check_appid_appsecret()

        if callable(self.__jsapi_ticket_getfunc):
            self.__jsapi_ticket, self.__jsapi_ticket_expires_at = self.__jsapi_ticket_getfunc()

        if self.__jsapi_ticket:
            now = time.time()
            if self.__jsapi_ticket_expires_at - now > 60:
                return self.__jsapi_ticket

        self.grant_jsapi_ticket()  # 从腾讯服务器获取 jsapi ticket 并更新
        return self.__jsapi_ticket

    @property
    def partnerid(self):
        """ 获取当前财付通商户身份标识 """
        return self.__partnerid

    @property
    def partnerkey(self):
        """ 获取当前财付通商户权限密钥 Key """
        return self.__partnerkey

    @property
    def paysignkey(self):
        """ 获取商户签名密钥 Key """
        return self.__paysignkey

    def grant_access_token(self):
        """
        获取 access token 并更新当前配置
        :return: 返回的 JSON 数据包 (传入 access_token_refreshfunc 参数后返回 None)
        """
        self._check_appid_appsecret()

        if callable(self.__access_token_refreshfunc):
            self.__access_token, self.__access_token_expires_at = self.__access_token_refreshfunc()
            return

        response_json = self.__request.get(
            url="https://api.weixin.qq.com/cgi-bin/token",
            params={
                "grant_type": "client_credential",
                "appid": self.__appid,
                "secret": self.__appsecret,
            },
            access_token=self.__access_token
        )
        self.__access_token = response_json['access_token']
        self.__access_token_expires_at = int(time.time()) + response_json['expires_in']

        if callable(self.__access_token_setfunc):
            self.__access_token_setfunc(self.__access_token, self.__access_token_expires_at)

        return response_json

    def grant_jsapi_ticket(self):
        """
        获取 jsapi ticket 并更新当前配置
        :return: 返回的 JSON 数据包 (传入 jsapi_ticket_refreshfunc 参数后返回 None)
        """
        self._check_appid_appsecret()

        if callable(self.__jsapi_ticket_refreshfunc):
            self.__jsapi_ticket, self.__jsapi_ticket_expires_at = self.__jsapi_ticket_refreshfunc()
            return

        response_json = self.__request.get(
            url="https://api.weixin.qq.com/cgi-bin/ticket/getticket",
            params={
                "type": "jsapi",
            },
            access_token=self.access_token,
        )
        self.__jsapi_ticket = response_json['ticket']
        self.__jsapi_ticket_expires_at = int(time.time()) + response_json['expires_in']

        if callable(self.__jsapi_ticket_setfunc):
            self.__jsapi_ticket_setfunc(self.__jsapi_ticket, self.__jsapi_ticket_expires_at)

        return response_json

    def get_access_token(self):
        """
        获取 Access Token 及 Access Token 过期日期, 仅供缓存使用, 如果希望得到原生的 Access Token 请求数据请使用 :func:`grant_token`
        **仅为兼容 v0.6.0 以前版本使用, 自行维护 access_token 请使用 access_token_setfunc 和 access_token_getfunc 进行操作**
        :return: dict 对象, key 包括 `access_token` 及 `access_token_expires_at`
        """
        self._check_appid_appsecret()

        return {
            'access_token': self.access_token,
            'access_token_expires_at': self.__access_token_expires_at,
        }

    def get_jsapi_ticket(self):
        """
        获取 Jsapi Ticket 及 Jsapi Ticket 过期日期, 仅供缓存使用, 如果希望得到原生的 Jsapi Ticket 请求数据请使用 :func:`grant_jsapi_ticket`
        **仅为兼容 v0.6.0 以前版本使用, 自行维护 jsapi_ticket 请使用 jsapi_ticket_setfunc 和 jsapi_ticket_getfunc 进行操作**
        :return: dict 对象, key 包括 `jsapi_ticket` 及 `jsapi_ticket_expires_at`
        """
        self._check_appid_appsecret()

        return {
            'jsapi_ticket': self.jsapi_ticket,
            'jsapi_ticket_expires_at': self.__jsapi_ticket_expires_at,
        }

    def _check_token(self):
        """
        检查 Token 是否存在
        :raises NeedParamError: Token 参数没有在初始化的时候提供
        """
        if not self.__token:
            raise NeedParamError('Please provide Token parameter in the construction of class.')

    def _check_appid_appsecret(self):
        """
        检查 AppID 和 AppSecret 是否存在
        :raises NeedParamError: AppID 或 AppSecret 参数没有在初始化的时候完整提供
        """
        if not self.__appid or not self.__appsecret:
            raise NeedParamError('Please provide app_id and app_secret parameters in the construction of class.')

    def _update_crypto(self):
        """
        根据当前配置内容更新 Crypto 类
        """
        if self.__encrypt_mode in ['compatible', 'safe'] and self.__encoding_aes_key is not None:
            if self.__token is None or self.__appid is None:
                raise NeedParamError('Please provide token and appid parameters in the construction of class.')
            self.__crypto = BasicCrypto(self.__token, self.__encoding_aes_key, self.__appid)
        else:
            self.__crypto = None

思维锻炼

设计讲师和学生类,讲师有上课,备课等方法,学生有听课,做练习等方法,均有姓名、性别、年龄等基本属性

设计聊天Message类

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

扫码关注云+社区

领取腾讯云代金券