itchat 库学习（登录篇）

本篇讲解itchat 的登录流程

第一步入口

import itchat
itchat.auto_login()

第二步找到auto_login最终调用处

跟进代码，调用到itchat__init__.py

def new_instance():newInstance = Core()instanceList.append(newInstance)return newInstanceoriginInstance = new_instance()
#...
auto_login                  = originInstance.auto_login

这里我们知道，auto_login代码最终调用到Core这个类中
回到Cory.py代码：

 from .components import load_components#...def auto_login(self, hotReload=False, statusStorageDir='itchat.pkl',enableCmdQR=False, picDir=None, qrCallback=None,loginCallback=None, exitCallback=None):''' 这是文档注释的关键点，此方法的定义位置it is defined in components/register.py'''raise NotImplementedError()#...load_components(Core)

点击跟进方法load_components，来到components/__init__.py

def load_components(core):load_contact(core)load_hotreload(core)load_login(core)load_messages(core)load_register(core)

上文代码提示，auto_login方法定义在components/register.py

def auto_login(self, hotReload=False, statusStorageDir='itchat.pkl',enableCmdQR=False, picDir=None, qrCallback=None,loginCallback=None, exitCallback=None):if not test_connect():logger.info("You can't get access to internet or wechat domain, so exit.")sys.exit()self.useHotReload = hotReloadself.hotReloadDir = statusStorageDirif hotReload:if self.load_login_status(statusStorageDir,loginCallback=loginCallback, exitCallback=exitCallback):returnself.login(enableCmdQR=enableCmdQR, picDir=picDir, qrCallback=qrCallback,loginCallback=loginCallback, exitCallback=exitCallback)self.dump_login_status(statusStorageDir)else:self.login(enableCmdQR=enableCmdQR, picDir=picDir, qrCallback=qrCallback,loginCallback=loginCallback, exitCallback=exitCallback)

跟进到这里，我们可以找到最终的调用关系：

itchat.auto_login → Core.py的auto_login → components/register.py的auto_login

第三步 auto_login之后的方法调用

components/register.py的auto_login方法内容很多，我们这里只关心最后的这个方法
self.login(enableCmdQR=enableCmdQR, picDir=picDir, qrCallback=qrCallback, loginCallback=loginCallback, exitCallback=exitCallback)
代码又回到Cory.py这里

    def login(self, enableCmdQR=False, picDir=None, qrCallback=None,loginCallback=None, exitCallback=None):''' log in like web wechat does#...it is defined in components/login.py#...'''raise NotImplementedError()

方法追踪同上，这里不再赘述
快速来到components/login.py的方法login

def login(self, enableCmdQR=False, picDir=None, qrCallback=None,loginCallback=None, exitCallback=None):if self.alive or self.isLogging:logger.warning('itchat has already logged in.')returnself.isLogging = Truewhile self.isLogging:uuid = push_login(self)if uuid:qrStorage = io.BytesIO()else:logger.info('Getting uuid of QR code.')# 获取UUIDwhile not self.get_QRuuid():time.sleep(1)logger.info('Downloading QR code.')# 根据UUID获取二维码图片qrStorage = self.get_QR(enableCmdQR=enableCmdQR,picDir=picDir, qrCallback=qrCallback)logger.info('Please scan the QR code to log in.')isLoggedIn = Falsewhile not isLoggedIn:#向微信后台查询二维码的扫描状态status = self.check_login()logger.info('server result status:'+status)if hasattr(qrCallback, '__call__'):qrCallback(uuid=self.uuid, status=status, qrcode=qrStorage.getvalue())if status == '200':isLoggedIn = Trueelif status == '201':if isLoggedIn is not None:logger.info('Please press confirm on your phone.')isLoggedIn = Noneelif status != '408':breakif isLoggedIn:breakelif self.isLogging:logger.info('Log in time out, reloading QR code.')else:return # log in process is stopped by userlogger.info('Loading the contact, this may take a little while.')self.web_init()self.show_mobile_login()self.get_contact(True)if hasattr(loginCallback, '__call__'):r = loginCallback()else:utils.clear_screen()if os.path.exists(picDir or config.DEFAULT_QR):os.remove(picDir or config.DEFAULT_QR)logger.info('Login successfully as %s' % self.storageClass.nickName)self.start_receiving(exitCallback)self.isLogging = False

解析上边的函数，大致做了以下四件事

1. 获取UUID
2. 根据UUID获取二维码
3. 查询二维码的扫描状态
4. 完成登录操作

1. 获取UUID

 while not self.get_QRuuid():time.sleep(1)

调用到components/login.py的方法get_QRuuid
成功返回uuid，失败返回None

def get_QRuuid(self):url = '%s/jslogin' % config.BASE_URLparams = {'appid' : 'wx782c26e4c19acffb','fun'   : 'new', }headers = { 'User-Agent' : config.USER_AGENT }r = self.s.get(url, params=params, headers=headers)# r.text 是微信服务器返回的数据# window.QRLogin.code = 200; window.QRLogin.uuid = "Aci1lDJMMQ==";# 通过正则表达式过滤出UUIDregx = r'window.QRLogin.code = (\d+); window.QRLogin.uuid = "(\S+?)";'data = re.search(regx, r.text)if data and data.group(1) == '200':self.uuid = data.group(2)return self.uuid

2. 根据UUID获取二维码

qrStorage = self.get_QR(enableCmdQR=enableCmdQR,picDir=picDir, qrCallback=qrCallback)

调用到components/login.py的方法get_QR

def get_QR(self, uuid=None, enableCmdQR=False, picDir=None, qrCallback=None):uuid = uuid or self.uuidpicDir = picDir or config.DEFAULT_QRqrStorage = io.BytesIO()# 根据UUID拼接成字符串# QRCode库制作二维码图片# 这里的二维码是根据特定内容，本地拼接的qrCode = QRCode('https://login.weixin.qq.com/l/' + uuid)qrCode.png(qrStorage, scale=10)if hasattr(qrCallback, '__call__'):qrCallback(uuid=uuid, status='0', qrcode=qrStorage.getvalue())else:with open(picDir, 'wb') as f:f.write(qrStorage.getvalue())if enableCmdQR:utils.print_cmd_qr(qrCode.text(1), enableCmdQR=enableCmdQR)else:#调用系统图库，展示二维码，供用户扫描utils.print_qr(picDir)return qrStorage

3. 查询二维码的扫描状态

 status = self.check_login()

调用到components/login.py的方法check_login

def check_login(self, uuid=None):uuid = uuid or self.uuidurl = '%s/cgi-bin/mmwebwx-bin/login' % config.BASE_URLlocalTime = int(time.time())params = 'loginicon=true&uuid=%s&tip=1&r=%s&_=%s' % (uuid, int(-localTime / 1579), localTime)headers = { 'User-Agent' : config.USER_AGENT }#根据UUID拼接参数，查询后台二维码的扫描点击状态r = self.s.get(url, params=params, headers=headers)# 返回扫描状态和扫描用户头像的base64字符串regx = r'window.code=(\d+)'data = re.search(regx, r.text)# 200:登陆成功 201:扫描成功 408:图片过期if data and data.group(1) == '200':if process_login_info(self, r.text):return '200'else:return '400'elif data:return data.group(1)else:return '400'

4. 登录成功后的操作

def process_login_info(core, loginContent):''' when finish login (scanning qrcode)* syncUrl and fileUploadingUrl will be fetched* deviceid and msgid will be generated* skey, wxsid, wxuin, pass_ticket will be fetched测试数据* loginContent = window.redirect_uri="https://wx.qq.com/cgi-bin/mmwebwx-bin/webwxnewloginpage?ticket=ATOqLtUX8b2sjUm3jfAI8zXd@qrticket_0&uuid=wb23AIsF4g==&lang=zh_CN&scan=1560760557";* *  '''regx = r'window.redirect_uri="(\S+)";'core.loginInfo['url'] = re.search(regx, loginContent).group(1)headers = { 'User-Agent' : config.USER_AGENT }r = core.s.get(core.loginInfo['url'], headers=headers, allow_redirects=False)core.loginInfo['url'] = core.loginInfo['url'][:core.loginInfo['url'].rfind('/')]for indexUrl, detailedUrl in (("wx2.qq.com"      , ("file.wx2.qq.com", "webpush.wx2.qq.com")),("wx8.qq.com"      , ("file.wx8.qq.com", "webpush.wx8.qq.com")),("qq.com"          , ("file.wx.qq.com", "webpush.wx.qq.com")),("web2.wechat.com" , ("file.web2.wechat.com", "webpush.web2.wechat.com")),("wechat.com"      , ("file.web.wechat.com", "webpush.web.wechat.com"))):fileUrl, syncUrl = ['https://%s/cgi-bin/mmwebwx-bin' % url for url in detailedUrl]if indexUrl in core.loginInfo['url']:core.loginInfo['fileUrl'], core.loginInfo['syncUrl'] = \fileUrl, syncUrlbreakelse:core.loginInfo['fileUrl'] = core.loginInfo['syncUrl'] = core.loginInfo['url']core.loginInfo['deviceid'] = 'e' + repr(random.random())[2:17]core.loginInfo['logintime'] = int(time.time() * 1e3)core.loginInfo['BaseRequest'] = {}# r.text = <error><ret>0</ret><message></message><skey>@crypt_174d98fa_a748e5c5e0923b23d558299174f0a806</skey><wxsid>yDRV0mMef8blZGj9</wxsid><wxuin>801548400</wxuin><pass_ticket>UgmksX5oEF4ChZKY2dc4zCt1phhVv8AH5fRlsN95HJlcAimQixFWjmjsFmXHuMoK</pass_ticket><isgrayscale>1</isgrayscale></error>for node in xml.dom.minidom.parseString(r.text).documentElement.childNodes:if node.nodeName == 'skey':core.loginInfo['skey'] = core.loginInfo['BaseRequest']['Skey'] = node.childNodes[0].dataelif node.nodeName == 'wxsid':core.loginInfo['wxsid'] = core.loginInfo['BaseRequest']['Sid'] = node.childNodes[0].dataelif node.nodeName == 'wxuin':core.loginInfo['wxuin'] = core.loginInfo['BaseRequest']['Uin'] = node.childNodes[0].dataelif node.nodeName == 'pass_ticket':core.loginInfo['pass_ticket'] = core.loginInfo['BaseRequest']['DeviceID'] = node.childNodes[0].dataif not all([key in core.loginInfo for key in ('skey', 'wxsid', 'wxuin', 'pass_ticket')]):logger.error('Your wechat account may be LIMITED to log in WEB wechat, error info:\n%s' % r.text)core.isLogging = Falsereturn Falsereturn True

内容很多，这里主要做的事情是core.loginInfo字典的赋值操作，
字典的KEY为：'skey', 'wxsid', 'wxuin', 'pass_ticket','url','fileUrl','syncUrl'

1. 获取微信用户信息

components/login.py的方法web_init，登录成功用户的数据获取，数据很多，以后用到时候再做研究

def web_init(self):url = '%s/webwxinit' % self.loginInfo['url']params = {'r': int(-time.time() / 1579),'pass_ticket': self.loginInfo['pass_ticket'], }data = { 'BaseRequest': self.loginInfo['BaseRequest'], }headers = {'ContentType': 'application/json; charset=UTF-8','User-Agent' : config.USER_AGENT, }r = self.s.post(url, params=params, data=json.dumps(data), headers=headers)dic = json.loads(r.content.decode('utf-8', 'replace'))# deal with login infoutils.emoji_formatter(dic['User'], 'NickName')self.loginInfo['InviteStartCount'] = int(dic['InviteStartCount'])self.loginInfo['User'] = wrap_user_dict(utils.struct_friend_info(dic['User']))self.memberList.append(self.loginInfo['User'])self.loginInfo['SyncKey'] = dic['SyncKey']self.loginInfo['synckey'] = '|'.join(['%s_%s' % (item['Key'], item['Val'])for item in dic['SyncKey']['List']])self.storageClass.userName = dic['User']['UserName']self.storageClass.nickName = dic['User']['NickName']# deal with contact list returned when initcontactList = dic.get('ContactList', [])chatroomList, otherList = [], []for m in contactList:if m['Sex'] != 0:otherList.append(m)elif '@@' in m['UserName']:m['MemberList'] = [] # don't let dirty info pollute the listchatroomList.append(m)elif '@' in m['UserName']:# mp will be dealt in update_local_friends as wellotherList.append(m)if chatroomList:update_local_chatrooms(self, chatroomList)if otherList:update_local_friends(self, otherList)return dic

2. 同步手机上登录状态

def show_mobile_login(self):url = '%s/webwxstatusnotify?lang=zh_CN&pass_ticket=%s' % (self.loginInfo['url'], self.loginInfo['pass_ticket'])data = {'BaseRequest': self.loginInfo['BaseRequest'],'Code': 3,'FromUserName': self.storageClass.userName,'ToUserName': self.storageClass.userName,'ClientMsgId': int(time.time()), }headers = {'ContentType': 'application/json; charset=UTF-8','User-Agent': config.USER_AGENT, }r = self.s.post(url, data=json.dumps(data), headers=headers)return ReturnValue(rawResponse=r)

3. 获取登录用户通讯录信息

def get_contact(self, update=False):if not update:return utils.contact_deep_copy(self, self.chatroomList)def _get_contact(seq=0):url = '%s/webwxgetcontact?r=%s&seq=%s&skey=%s' % (self.loginInfo['url'],int(time.time()), seq, self.loginInfo['skey'])headers = {'ContentType': 'application/json; charset=UTF-8','User-Agent' : config.USER_AGENT, }try:r = self.s.get(url, headers=headers)except:logger.info('Failed to fetch contact, that may because of the amount of your chatrooms')for chatroom in self.get_chatrooms():self.update_chatroom(chatroom['UserName'], detailedMember=True)return 0, []j = json.loads(r.content.decode('utf-8', 'replace'))return j.get('Seq', 0), j.get('MemberList')seq, memberList = 0, []while 1:seq, batchMemberList = _get_contact(seq)memberList.extend(batchMemberList)if seq == 0:breakchatroomList, otherList = [], []for m in memberList:if m['Sex'] != 0:otherList.append(m)elif '@@' in m['UserName']:chatroomList.append(m)elif '@' in m['UserName']:# mp will be dealt in update_local_friends as wellotherList.append(m)if chatroomList:update_local_chatrooms(self, chatroomList)if otherList:update_local_friends(self, otherList)return utils.contact_deep_copy(self, chatroomList)

4. 开启心跳，扫描信息

def start_receiving(self, exitCallback=None, getReceivingFnOnly=False):self.alive = Truedef maintain_loop():retryCount = 0while self.alive:try:i = sync_check(self)if i is None:self.alive = Falseelif i == '0':passelse:msgList, contactList = self.get_msg()if msgList:msgList = produce_msg(self, msgList)for msg in msgList:self.msgList.put(msg)if contactList:chatroomList, otherList = [], []for contact in contactList:if '@@' in contact['UserName']:chatroomList.append(contact)else:otherList.append(contact)chatroomMsg = update_local_chatrooms(self, chatroomList)chatroomMsg['User'] = self.loginInfo['User']self.msgList.put(chatroomMsg)update_local_friends(self, otherList)retryCount = 0except requests.exceptions.ReadTimeout:passexcept:retryCount += 1logger.error(traceback.format_exc())if self.receivingRetryCount < retryCount:self.alive = Falseelse:time.sleep(1)self.logout()if hasattr(exitCallback, '__call__'):exitCallback()else:logger.info('LOG OUT!')if getReceivingFnOnly:return maintain_loopelse:maintainThread = threading.Thread(target=maintain_loop)maintainThread.setDaemon(True)maintainThread.start()