【Python高级语法】11、面向对象设计
- Python
- 2024-01-23
- 19
继承 - 是基于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
发表评论