#!/usr/bin/env python # coding: utf-8 #=================================================== from utils import * from config import Constant from config import Log #--------------------------------------------------- import sys import os import cookielib import random import requests import time import xml.dom.minidom # for media upload import mimetypes from requests_toolbelt.multipart.encoder import MultipartEncoder #=================================================== class WXAPI(object): def __init__(self, host): self.wx_host = host self.wx_filehost = '' self.wx_conf = {} # jsLogin时这个appid只能使用: wx782c26e4c19acffb self.appid = Constant.API_APPID self.uuid = '' self.redirect_uri = '' self.skey = '' self.sid = '' self.uin = '' self.pass_ticket = '' self.base_request = {} self.synckey_dic = {} self.synckey = '' self.device_id = 'e' + repr(random.random())[2:17] # device_id: 登录手机设备 # web wechat 的格式为: e123456789012345 (e+15位随机数) # mobile wechat 的格式为: A1234567890abcde (A+15位随机数字或字母) self.user_agent = Constant.API_USER_AGENT self.cookie = None self.conf_factory() self.User = [] # 登陆账号信息 self.MemberList = [] # 好友+群聊+公众号+特殊账号 self.MemberCount = 0 self.ContactList = [] # 好友 self.GroupList = [] # 群 self.GroupMemeberList = {} # 群聊成员字典 # "group_id": [ # {member}, ... # ] self.PublicUsersList = [] # 公众号／服务号 self.SpecialUsersList = [] # 特殊账号 self.media_count = 0 def conf_factory(self): e = self.wx_host # wx.qq.com t, o, n = "login.weixin.qq.com", "file.wx.qq.com", "webpush.weixin.qq.com" if e.find("wx2.qq.com") > -1: t, o, n = "login.wx2.qq.com", "file.wx2.qq.com", "webpush.wx2.qq.com" elif e.find("wx8.qq.com") > -1: t, o, n = "login.wx8.qq.com", "file.wx8.qq.com", "webpush.wx8.qq.com" elif e.find("qq.com") > -1: t, o, n = "login.wx.qq.com", "file.wx.qq.com", "webpush.wx.qq.com" elif e.find("web2.wechat.com") > -1: t, o, n = "login.web2.wechat.com", "file.web2.wechat.com", "webpush.web2.wechat.com" elif e.find("wechat.com") > -1: t, o, n = "login.web.wechat.com", "file.web.wechat.com", "webpush.web.wechat.com" self.wx_filehost = o conf = { 'LANG': Constant.API_LANG, 'SpecialUsers': Constant.API_SPECIAL_USER, 'API_jsLogin': "https://" + t + "/jslogin", 'API_qrcode': "https://login.weixin.qq.com/l/", 'API_qrcode_img': "https://login.weixin.qq.com/qrcode/", 'API_login': "https://" + t + "/cgi-bin/mmwebwx-bin/login", 'API_synccheck': "https://" + n + "/cgi-bin/mmwebwx-bin/synccheck", 'API_webwxdownloadmedia': "https://" + o + "/cgi-bin/mmwebwx-bin/webwxgetmedia", 'API_webwxuploadmedia': "https://" + o + "/cgi-bin/mmwebwx-bin/webwxuploadmedia", 'API_webwxpreview': "https://" + e + "/cgi-bin/mmwebwx-bin/webwxpreview", 'API_webwxinit': "https://" + e + "/cgi-bin/mmwebwx-bin/webwxinit", 'API_webwxgetcontact': "https://" + e + "/cgi-bin/mmwebwx-bin/webwxgetcontact", 'API_webwxsync': "https://" + e + "/cgi-bin/mmwebwx-bin/webwxsync", 'API_webwxbatchgetcontact': "https://" + e + "/cgi-bin/mmwebwx-bin/webwxbatchgetcontact", 'API_webwxgeticon': "https://" + e + "/cgi-bin/mmwebwx-bin/webwxgeticon", 'API_webwxsendmsg': "https://" + e + "/cgi-bin/mmwebwx-bin/webwxsendmsg", 'API_webwxsendmsgimg': "https://" + e + "/cgi-bin/mmwebwx-bin/webwxsendmsgimg", 'API_webwxsendmsgvedio': "https://" + e + "/cgi-bin/mmwebwx-bin/webwxsendvideomsg", 'API_webwxsendemoticon': "https://" + e + "/cgi-bin/mmwebwx-bin/webwxsendemoticon", 'API_webwxsendappmsg': "https://" + e + "/cgi-bin/mmwebwx-bin/webwxsendappmsg", 'API_webwxgetheadimg': "https://" + e + "/cgi-bin/mmwebwx-bin/webwxgetheadimg", 'API_webwxgetmsgimg': "https://" + e + "/cgi-bin/mmwebwx-bin/webwxgetmsgimg", 'API_webwxgetmedia': "https://" + e + "/cgi-bin/mmwebwx-bin/webwxgetmedia", 'API_webwxgetvideo': "https://" + e + "/cgi-bin/mmwebwx-bin/webwxgetvideo", 'API_webwxlogout': "https://" + e + "/cgi-bin/mmwebwx-bin/webwxlogout", 'API_webwxgetvoice': "https://" + e + "/cgi-bin/mmwebwx-bin/webwxgetvoice", 'API_webwxupdatechatroom': "https://" + e + "/cgi-bin/mmwebwx-bin/webwxupdatechatroom", 'API_webwxcreatechatroom': "https://" + e + "/cgi-bin/mmwebwx-bin/webwxcreatechatroom", 'API_webwxstatusnotify': "https://" + e + "/cgi-bin/mmwebwx-bin/webwxstatusnotify", 'API_webwxcheckurl': "https://" + e + "/cgi-bin/mmwebwx-bin/webwxcheckurl", 'API_webwxverifyuser': "https://" + e + "/cgi-bin/mmwebwx-bin/webwxverifyuser", 'API_webwxfeedback': "https://" + e + "/cgi-bin/mmwebwx-bin/webwxsendfeedback", 'API_webwxreport': "https://" + e + "/cgi-bin/mmwebwx-bin/webwxstatreport", 'API_webwxsearch': "https://" + e + "/cgi-bin/mmwebwx-bin/webwxsearchcontact", 'API_webwxoplog': "https://" + e + "/cgi-bin/mmwebwx-bin/webwxoplog", 'API_checkupload': "https://" + e + "/cgi-bin/mmwebwx-bin/webwxcheckupload", 'API_webwxrevokemsg': "https://" + e + "/cgi-bin/mmwebwx-bin/webwxrevokemsg", 'API_webwxpushloginurl': "https://" + e + "/cgi-bin/mmwebwx-bin/webwxpushloginurl", 'CONTACTFLAG_CONTACT': 1, 'CONTACTFLAG_CHATCONTACT': 2, 'CONTACTFLAG_CHATROOMCONTACT': 4, 'CONTACTFLAG_BLACKLISTCONTACT': 8, 'CONTACTFLAG_DOMAINCONTACT': 16, 'CONTACTFLAG_HIDECONTACT': 32, 'CONTACTFLAG_FAVOURCONTACT': 64, 'CONTACTFLAG_3RDAPPCONTACT': 128, 'CONTACTFLAG_SNSBLACKLISTCONTACT': 256, 'CONTACTFLAG_NOTIFYCLOSECONTACT': 512, 'CONTACTFLAG_TOPCONTACT': 2048, 'MSGTYPE_TEXT': 1, 'MSGTYPE_IMAGE': 3, 'MSGTYPE_VOICE': 34, 'MSGTYPE_VIDEO': 43, 'MSGTYPE_MICROVIDEO': 62, 'MSGTYPE_EMOTICON': 47, 'MSGTYPE_APP': 49, 'MSGTYPE_VOIPMSG': 50, 'MSGTYPE_VOIPNOTIFY': 52, 'MSGTYPE_VOIPINVITE': 53, 'MSGTYPE_LOCATION': 48, 'MSGTYPE_STATUSNOTIFY': 51, 'MSGTYPE_SYSNOTICE': 9999, 'MSGTYPE_POSSIBLEFRIEND_MSG': 40, 'MSGTYPE_VERIFYMSG': 37, 'MSGTYPE_SHARECARD': 42, 'MSGTYPE_SYS': 10000, 'MSGTYPE_RECALLED': 10002, 'APPMSGTYPE_TEXT': 1, 'APPMSGTYPE_IMG': 2, 'APPMSGTYPE_AUDIO': 3, 'APPMSGTYPE_VIDEO': 4, 'APPMSGTYPE_URL': 5, 'APPMSGTYPE_ATTACH': 6, 'APPMSGTYPE_OPEN': 7, 'APPMSGTYPE_EMOJI': 8, 'APPMSGTYPE_VOICE_REMIND': 9, 'APPMSGTYPE_SCAN_GOOD': 10, 'APPMSGTYPE_GOOD': 13, 'APPMSGTYPE_EMOTION': 15, 'APPMSGTYPE_CARD_TICKET': 16, 'APPMSGTYPE_REALTIME_SHARE_LOCATION': 17, 'APPMSGTYPE_TRANSFERS': 2e3, 'APPMSGTYPE_RED_ENVELOPES': 2001, 'APPMSGTYPE_READER_TYPE': 100001, 'UPLOAD_MEDIA_TYPE_IMAGE': 1, 'UPLOAD_MEDIA_TYPE_VIDEO': 2, 'UPLOAD_MEDIA_TYPE_AUDIO': 3, 'UPLOAD_MEDIA_TYPE_ATTACHMENT': 4, } self.wx_conf = conf def getuuid(self): """ @brief Gets the uuid just used for login. @return Bool: whether operation succeed. """ url = self.wx_conf['API_jsLogin'] params = { 'appid': self.appid, 'fun': 'new', 'lang': self.wx_conf['LANG'], '_': int(time.time()), } data = post(url, params, False) regx = r'window.QRLogin.code = (\d+); window.QRLogin.uuid = "(\S+?)"' pm = re.search(regx, data) if pm: code = pm.group(1) self.uuid = pm.group(2) return code == '200' return False def genqrcode(self): """ @brief outprint the qrcode to stdout on macos/linux or open image on windows """ if sys.platform.startswith('win'): url = self.wx_conf['API_qrcode_img'] + self.uuid params = { 't': 'webwx', '_': int(time.time()) } data = post(url, params, False) if data == '': return qrcode_path = save_file('qrcode.jpg', data, './') os.startfile(qrcode_path) else: str2qr_terminal(self.wx_conf['API_qrcode'] + self.uuid) def waitforlogin(self, tip=1): """ @brief wait for scaning qrcode to login @param tip 1: wait for scan qrcode 0: wait for confirm @return Bool: whether operation succeed """ time.sleep(tip) url = self.wx_conf['API_login'] + '?tip=%s&uuid=%s&_=%s' % ( tip, self.uuid, int(time.time())) data = get(url) pm = re.search(r'window.code=(\d+);', data) code = pm.group(1) if code == '201': return True elif code == '200': pm = re.search(r'window.redirect_uri="(\S+?)";', data) r_uri = pm.group(1) + '&fun=new' self.redirect_uri = r_uri self.wx_host = r_uri.split('://')[1].split('/')[0] self.conf_factory() return True elif code == '408': echo(Constant.LOG_MSG_WAIT_LOGIN_ERR1) else: echo(Constant.LOG_MSG_WAIT_LOGIN_ERR2) return False def login(self): """ @brief login redirect_uri 有效时间是从扫码成功后算起， 大概是 300 秒，在此期间可以重新登录，但获取的联系人和群ID会改变 @return Bool: whether operation succeed """ data = get(self.redirect_uri) doc = xml.dom.minidom.parseString(data) root = doc.documentElement for node in root.childNodes: if node.nodeName == 'ret': if node.childNodes[0].data != "0": return False elif node.nodeName == 'skey': self.skey = node.childNodes[0].data elif node.nodeName == 'wxsid': self.sid = node.childNodes[0].data elif node.nodeName == 'wxuin': self.uin = node.childNodes[0].data elif node.nodeName == 'pass_ticket': self.pass_ticket = node.childNodes[0].data if '' in (self.skey, self.sid, self.uin, self.pass_ticket): return False self.base_request = { 'Uin': int(self.uin), 'Sid': self.sid, 'Skey': self.skey, 'DeviceID': self.device_id, } return True def webwxinit(self): """ @brief wechat initial 掉线后 300 秒可以重新使用此 api 登录 获取的联系人和群ID保持不变 @return Bool: whether operation succeed """ url = self.wx_conf['API_webwxinit'] + \ '?pass_ticket=%s&skey=%s&r=%s' % ( self.pass_ticket, self.skey, int(time.time()) ) params = { 'BaseRequest': self.base_request } dic = post(url, params) self.User = dic['User'] self.make_synckey(dic) return dic['BaseResponse']['Ret'] == 0 def webwxstatusnotify(self): """ @brief notify the mobile phone, this not necessary @return Bool: whether operation succeed """ url = self.wx_conf['API_webwxstatusnotify'] + \ '?lang=%s&pass_ticket=%s' % ( self.wx_conf['LANG'], self.pass_ticket ) params = { 'BaseRequest': self.base_request, "Code": 3, "FromUserName": self.User['UserName'], "ToUserName": self.User['UserName'], "ClientMsgId": int(time.time()) } dic = post(url, params) return dic['BaseResponse']['Ret'] == 0 def webwxgetcontact(self): """ @brief get all contacts: people, group, public user, special user @return Bool: whether operation succeed """ SpecialUsers = self.wx_conf['SpecialUsers'] url = self.wx_conf['API_webwxgetcontact'] + \ '?pass_ticket=%s&skey=%s&r=%s' % ( self.pass_ticket, self.skey, int(time.time()) ) dic = post(url, {}) self.MemberCount = dic['MemberCount'] self.MemberList = dic['MemberList'] ContactList = self.MemberList[:] GroupList = self.GroupList[:] PublicUsersList = self.PublicUsersList[:] SpecialUsersList = self.SpecialUsersList[:] for i in xrange(len(ContactList) - 1, -1, -1): Contact = ContactList[i] if Contact['VerifyFlag'] & 8 != 0: # 公众号/服务号 ContactList.remove(Contact) self.PublicUsersList.append(Contact) elif Contact['UserName'] in SpecialUsers: # 特殊账号 ContactList.remove(Contact) self.SpecialUsersList.append(Contact) elif Contact['UserName'].find('@@') != -1: # 群聊 ContactList.remove(Contact) self.GroupList.append(Contact) elif Contact['UserName'] == self.User['UserName']: # 自己 ContactList.remove(Contact) self.ContactList = ContactList return True def webwxbatchgetcontact(self, gid_list): """ @brief get group contacts @param gid_list, The list of group id @return List, list of group contacts """ url = self.wx_conf['API_webwxbatchgetcontact'] + \ '?type=ex&r=%s&pass_ticket=%s' % ( int(time.time()), self.pass_ticket ) params = { 'BaseRequest': self.base_request, "Count": len(gid_list), "List": [{"UserName": gid, "EncryChatRoomId": ""} for gid in gid_list] } dic = post(url, params) return dic['ContactList'] def synccheck(self): """ @brief check whether there's a message @return [retcode, selector] retcode: 0 successful 1100 logout 1101 login otherwhere selector: 0 nothing 2 message 6 unkonwn 7 webwxsync """ params = { 'r': int(time.time()), 'sid': self.sid, 'uin': self.uin, 'skey': self.skey, 'deviceid': self.device_id, 'synckey': self.synckey, '_': int(time.time()), } url = self.wx_conf['API_synccheck'] + '?' + urllib.urlencode(params) data = get(url) reg = r'window.synccheck={retcode:"(\d+)",selector:"(\d+)"}' pm = re.search(reg, data) retcode = pm.group(1) selector = pm.group(2) return [retcode, selector] def webwxsync(self): """ @brief sync the messages @return Dict{} """ url = self.wx_conf['API_webwxsync'] + \ '?sid=%s&skey=%s&pass_ticket=%s' % ( self.sid, self.skey, self.pass_ticket ) params = { 'BaseRequest': self.base_request, 'SyncKey': self.synckey_dic, 'rr': ~int(time.time()) } dic = post(url, params) if dic['BaseResponse']['Ret'] == 0: self.make_synckey(dic) return dic def webwxgetmsgimg(self, msgid): """ @brief get image in message @param msgid The id of message @return binary data of image """ url = self.wx_conf['API_webwxgetmsgimg'] + \ '?MsgID=%s&skey=%s' % (msgid, self.skey) data = get(url, api='webwxgetmsgimg') return data def webwxgetvoice(self, msgid): """ @brief get voice in message @param msgid The id of message @return binary data of voice """ url = self.wx_conf['API_webwxgetvoice'] + \ '?msgid=%s&skey=%s' % (msgid, self.skey) data = get(url, api='webwxgetvoice') return data def webwxgetvideo(self, msgid): """ @brief get video in message @param msgid The id of message @return binary data of video """ url = self.wx_conf['API_webwxgetvideo'] + \ '?msgid=%s&skey=%s' % (msgid, self.skey) data = get(url, api='webwxgetvideo') return data def webwxgeticon(self, id): """ @brief get user small icon @param id String @return binary data of icon """ url = self.wx_conf['API_webwxgeticon'] + \ '?username=%s&skey=%s' % (id, self.skey) data = get(url, api='webwxgeticon') return data def webwxgetheadimg(self, id): """ @brief get user head image @param id String @return binary data of image """ url = self.wx_conf['API_webwxgetheadimg'] + \ '?username=%s&skey=%s' % (id, self.skey) data = get(url, api='webwxgetheadimg') return data def webwxsendmsg(self, word, to='filehelper'): """ @brief send text message @param word String @param to User id @return dic Dict """ url = self.wx_conf['API_webwxsendmsg'] + \ '?pass_ticket=%s' % (self.pass_ticket) clientMsgId = str(int(time.time() * 1000)) + \ str(random.random())[:5].replace('.', '') params = { 'BaseRequest': self.base_request, 'Msg': { "Type": 1, "Content": trans_coding(word), "FromUserName": self.User['UserName'], "ToUserName": to, "LocalID": clientMsgId, "ClientMsgId": clientMsgId } } dic = post(url, params) return dic def webwxuploadmedia(self, file_path): """ @brief upload image @param file_path String @return Dict: json """ url = self.wx_conf['API_webwxuploadmedia'] + '?f=json' # 计数器 self.media_count = self.media_count + 1 fn = file_path # mime_type: # 'application/pdf' # 'image/jpeg' # 'image/png' # ... mime_type = mimetypes.guess_type(fn, strict=False)[0] if not mime_type: mime_type = 'text/plain' # 文档格式 # 微信服务器目前应该支持3种类型: # pic 直接显示，包含图片，表情 # video 不清楚 # doc 显示为文件，包含PDF等 media_type = 'pic' if mime_type.split('/')[0] == 'image' else 'doc' time_format = "%a %b %d %Y %T GMT%z (%Z)" last_modifie_date = time.strftime(time_format, time.localtime()) file_size = os.path.getsize(fn) pass_ticket = self.pass_ticket client_media_id = str(int(time.time() * 1000)) + \ str(random.random())[:5].replace('.', '') webwx_data_ticket = '' for item in self.cookie: if item.name == 'webwx_data_ticket': webwx_data_ticket = item.value break if (webwx_data_ticket == ''): Log.error("No Cookie\n") return None uploadmediarequest = json.dumps({ "BaseRequest": self.base_request, "ClientMediaId": client_media_id, "TotalLen": file_size, "StartPos": 0, "DataLen": file_size, "MediaType": 4 }, ensure_ascii=False).encode('utf8') multipart_encoder = MultipartEncoder( fields={ 'id': 'WU_FILE_' + str(self.media_count), 'name': fn, 'type': mime_type, 'lastModifieDate': last_modifie_date, 'size': str(file_size), 'mediatype': media_type, 'uploadmediarequest': uploadmediarequest, 'webwx_data_ticket': webwx_data_ticket, 'pass_ticket': pass_ticket, 'filename': ( fn, open(fn, 'rb'), mime_type.split('/')[1] ) }, boundary=( '-----------------------------' '1575017231431605357584454111' ) ) headers = { 'Host': self.wx_filehost, 'User-Agent': self.user_agent, 'Accept': ( 'text/html,application/xhtml+xml,' 'application/xml;q=0.9,*/*;q=0.8' ), 'Accept-Language': 'en-US,en;q=0.5', 'Accept-Encoding': 'gzip, deflate', 'Referer': 'https://' + self.wx_host, 'Content-Type': multipart_encoder.content_type, 'Origin': 'https://' + self.wx_host, 'Connection': 'keep-alive', 'Pragma': 'no-cache', 'Cache-Control': 'no-cache', } r = requests.post(url, data=multipart_encoder, headers=headers) if dic['BaseResponse']['Ret'] == 0: return dic return None def webwxsendmsgimg(self, user_id, media_id): """ @brief send image @param user_id String @param media_id String @return Bool: whether operation succeed """ url = self.wx_conf['API_webwxsendmsgimg'] + \ '?fun=async&f=json&pass_ticket=%s' % self.pass_ticket clientMsgId = str(int(time.time() * 1000)) + \ str(random.random())[:5].replace('.', '') data_json = { "BaseRequest": self.base_request, "Msg": { "Type": 3, "MediaId": media_id, "FromUserName": self.User['UserName'], "ToUserName": user_id, "LocalID": clientMsgId, "ClientMsgId": clientMsgId } } r = post(url, data_json) return dic['BaseResponse']['Ret'] == 0 def webwxsendemoticon(self, user_id, media_id): """ @brief send image @param user_id String @param media_id String @return Bool: whether operation succeed """ url = self.wx_conf['API_webwxsendemoticon'] + \ '?fun=sys&f=json&pass_ticket=%s' % self.pass_ticket clientMsgId = str(int(time.time() * 1000)) + \ str(random.random())[:5].replace('.', '') data_json = { "BaseRequest": self.base_request, "Msg": { "Type": 47, "EmojiFlag": 2, "MediaId": media_id, "FromUserName": self.User['UserName'], "ToUserName": user_id, "LocalID": clientMsgId, "ClientMsgId": clientMsgId } } r = post(url, data_json) return dic['BaseResponse']['Ret'] == 0 def webwxsendappmsg(self, user_id, data): """ @brief send app msg @param user_id String @param data Dict @return Bool: whether operation succeed """ url = self.wx_conf['API_webwxsendappmsg'] + \ '?fun=sys&f=json&pass_ticket=%s' % self.pass_ticket clientMsgId = str(int(time.time() * 1000)) + \ str(random.random())[:5].replace('.', '') content = ''.join([ "" % data['appid'], # 可使用其它AppID "%s" % data['title'], "", "", "%d" % data['type'], "", "", "", "", "%d" % data['totallen'], "%s" % data['attachid'], "%s" % data['fileext'], "", "", "", ]) data_json = { "BaseRequest": self.base_request, "Msg": { "Type": data['type'], "Content": content, "FromUserName": self.User['UserName'], "ToUserName": user_id, "LocalID": clientMsgId, "ClientMsgId": clientMsgId }, "Scene": 0 } r = post(url, data_json) return dic['BaseResponse']['Ret'] == 0 def webwxcreatechatroom(self, uid_arr): """ @brief create a chat group @param uid_arr [String] @return Bool: whether operation succeed """ url = self.wx_conf['API_webwxcreatechatroom'] + '?r=%s' % int(time.time()) params = { 'BaseRequest': self.base_request, 'Topic': '', 'MemberCount': len(uid_arr), 'MemberList': [{'UserName': uid} for uid in uid_arr], } dic = post(url, params) return dic['BaseResponse']['Ret'] == 0 def webwxupdatechatroom(self, add_arr, del_arr, invite_arr): """ @brief add/delete/invite member in chat group @param add_arr [uid: String] @param del_arr [uid: String] @param invite_arr [uid: String] @return Bool: whether operation succeed """ url = self.wx_conf['API_webwxupdatechatroom'] + '?r=%s' % int(time.time()) params = { 'BaseRequest': self.base_request, 'ChatRoomName': '', 'NewTopic': '', 'AddMemberList': add_arr, 'DelMemberList': del_arr, 'InviteMemberList': invite_arr, } dic = post(url, params) return dic['BaseResponse']['Ret'] == 0 def webwxrevokemsg(self, msgid, user_id, client_msgid): """ @brief revoke a message @param msgid String @param user_id String @param client_msgid String @return Bool: whether operation succeed """ url = self.wx_conf['API_webwxrevokemsg'] + '?r=%s' % int(time.time()) params = { 'BaseRequest': self.base_request, 'SvrMsgId': msgid, 'ToUserName': user_id, 'ClientMsgId': client_msgid } dic = post(url, params) return dic['BaseResponse']['Ret'] == 0 def webwxpushloginurl(self, uin): """ @brief push a login confirm alert to mobile device @param uin String @return dic Dict """ url = self.wx_conf['API_webwxpushloginurl'] + '?uin=%s' % uin dic = eval(get(url)) return dic def association_login(self): """ @brief login without scan qrcode @return Bool: whether operation succeed """ if self.uin != '': dic = self.webwxpushloginurl(self.uin) if dic['ret'] == '0': self.uuid = dic['uuid'] return True return False def send_text(self, user_id, text): """ @brief send text @param user_id String @param text String @return Bool: whether operation succeed """ dic = self.webwxsendmsg(text, user_id) return dic['BaseResponse']['Ret'] == 0 def send_img(self, user_id, file_path): """ @brief send image @param user_id String @param file_path String @return Bool: whether operation succeed """ response = self.webwxuploadmedia(file_path) media_id = "" if response is not None: media_id = response['MediaId'] return self.webwxsendmsgimg(user_id, media_id) def send_emot(self, user_id, file_path): """ @brief send emotion @param user_id String @param file_path String @return Bool: whether operation succeed """ response = self.webwxuploadmedia(file_path) media_id = "" if response is not None: media_id = response['MediaId'] return self.webwxsendemoticon(user_id, media_id) def send_file(self, user_id, file_path): """ @brief send file @param user_id String @param file_path String @return Bool: whether operation succeed """ title = file_path.split('/')[-1] data = { 'appid': Constant.API_WXAPPID, 'title': title, 'totallen': '', 'attachid': '', 'type': self.wx_conf['APPMSGTYPE_ATTACH'], 'fileext': title.split('.')[-1], } response = self.webwxuploadmedia(file_path) if response is not None: data['totallen'] = response['StartPos'] data['attachid'] = response['MediaId'] else: Log.error('File upload error') return self.webwxsendappmsg(user_id, data) def make_synckey(self, dic): self.synckey_dic = dic['SyncKey'] def foo(x): return str(x['Key']) + '_' + str(x['Val']) # synckey for synccheck self.synckey = '|'.join( [foo(keyVal) for keyVal in self.synckey_dic['List']]) def revoke_msg(self, msgid, user_id, client_msgid): """ @brief revoke a message @param msgid String @param user_id String @param client_msgid String @return Bool: whether operation succeed """ return self.webwxrevokemsg(msgid, user_id, client_msgid) # ----------------------------------------------------- # The following is getting user & group infomation apis def get_user_by_id(self, user_id): """ @brief get all type of name by user id @param user_id The id of user @return Dict: { 'UserName' # 微信动态ID 'RemarkName' # 备注 'NickName' # 微信昵称 'ShowName' # Log显示用的 } """ UnknownPeople = Constant.LOG_MSG_UNKNOWN_NAME + user_id name = { 'UserName': user_id, 'RemarkName': '', 'NickName': '', 'ShowName': '', } name['ShowName'] = UnknownPeople # yourself if user_id == self.User['UserName']: name['RemarkName'] = self.User['RemarkName'] name['NickName'] = self.User['NickName'] name['ShowName'] = name['NickName'] else: # 联系人 for member in self.MemberList: if member['UserName'] == user_id: r, n = member['RemarkName'], member['NickName'] name['RemarkName'] = r name['NickName'] = n name['ShowName'] = r if r else n break # 特殊帐号 for member in self.SpecialUsersList: if member['UserName'] == user_id: name['RemarkName'] = user_id name['NickName'] = user_id name['ShowName'] = user_id break return name def get_group_user_by_id(self, user_id, group_id): """ @brief get group user by user id @param user_id The id of user @param group_id The id of group @return Dict: { 'UserName' # 微信动态ID 'NickName' # 微信昵称 'DisplayName' # 群聊显示名称 'ShowName' # Log显示用的 'AttrStatus' # 群用户id } """ UnknownPeople = Constant.LOG_MSG_UNKNOWN_NAME + user_id name = { 'UserName': user_id, 'NickName': '', 'DisplayName': '', 'ShowName': '', 'AttrStatus': '', } name['ShowName'] = UnknownPeople # 群友 if group_id in self.GroupMemeberList: for member in self.GroupMemeberList[group_id]: if member['UserName'] == user_id: n, d = member['NickName'], member['DisplayName'] name['NickName'] = n name['DisplayName'] = d name['AttrStatus'] = member['AttrStatus'] name['ShowName'] = d if d else n break return name def get_group_by_id(self, group_id): """ @brief get basic info by group id @param group_id The id of group @return Dict: { 'UserName' # 微信动态ID 'NickName' # 微信昵称 'DisplayName' # 群聊显示名称 'ShowName' # Log显示用的 'OwnerUin' # 群主ID 'MemberCount' # 群人数 } """ UnknownGroup = Constant.LOG_MSG_UNKNOWN_GROUP_NAME + group_id group = { 'UserName': group_id, 'NickName': '', 'DisplayName': '', 'ShowName': '', 'OwnerUin': '', 'MemberCount': '', } for member in self.GroupList: if member['UserName'] == group_id: group['NickName'] = member['NickName'] group['DisplayName'] = member.get('DisplayName', '') group['ShowName'] = member.get('NickName', UnknownGroup) group['OwnerUin'] = member.get('OwnerUin', '') group['MemberCount'] = member['MemberCount'] break return group def get_user_id(self, name): """ @brief Gets the user id. @param name The user nickname or remarkname @return The user id. """ for member in self.MemberList: if name == member['RemarkName'] or name == member['NickName']: return member['UserName'] return None