Python实现淘宝联盟二维码扫码自动登录与数据抓取

纸上星河

发布时间：2026-10-13| 4人浏览过

原创

本文介绍使用Python实现淘宝联盟数据抓取的自动登录方案。通过调用淘宝二维码接口实现扫码登录，解决验证码问题。利用Requests库的Session机制结合pickle序列化，实现Cookie的异步存储与读取，确保会话状态持久化。文章提供了核心代码逻辑，并探讨了Session过期后自动刷新票据的难点与思路。

这是一个关于自动登录淘宝联盟以抓取数据的小项目。此前在GitHub上见过类似的Python实现，因此决定采用Python开发。初次正式使用Python编写程序时，其简洁性令人印象深刻。尽管在2.7版本中曾遇到编码和环境迁移等困扰，但最终均得以解决。

核心在于解决登录问题。以往常受验证码困扰，如今支持二维码扫码登录则相对简便。以下是实现登录的Python代码，主要逻辑包括获取二维码并打印，随后持续轮询扫码状态。若二维码过期，则重新请求新的二维码（注：此处仅展示核心逻辑，因部分通用方法已封装，代码可能无法直接运行）。

def getQRCode(enableCmdQR):
  payload = {'_ksTS': str(time.time()), 'from': 'alimama'}
  qrCodeObj = utils.fetchAPI('https://qrlogin.taobao.com/qrcodelogin/generateQRCode4Login.do', payload,
                "json", None, True, True)
 
  print(qrCodeObj)
  utils.printQRCode('http:' + qrCodeObj['url'], enableCmdQR)
  lgToken = qrCodeObj['lgToken']
  return lgToken
 
 
def login(enableCmdQR=False):
  lgToken = getQRCode(enableCmdQR)
  code = 0
  successLoginURL = ""
  while code != 10006:
    payload = {'lgToken': lgToken,
          'defaulturl': 'http%3A%2F%2Flogin.taobao.com%2Fmember%2Ftaobaoke%2Flogin.htm%3Fis_login%3D1&_ksTS=' + str(
            time.time())}
 
    rObj = utils.fetchAPI('https://qrlogin.taobao.com/qrcodelogin/qrcodeLoginCheck.do', payload, "json", True,
               False)
    code = int(rObj['code'])
    if 10000 == code:
      # print("请扫描二维码登录")
      continue
    elif 10001 == code:
      print("已扫描二维码，请在确认登录")
    elif 10004 == code:
      print("已过期请重新扫描")
      login()
    elif 10006 == code:
      successLoginURL = rObj["url"]
      print("登录成功，正在跳转")
    else:
      print("未知错误，退出执行")
      sys.exit(0)
 
    time.sleep(5)
 
  print "登录成功跳转:" + successLoginURL
  r = utils.fetchAPI(successLoginURL, None, "raw", True, False, True)
  utils.fetchAPI(r.headers['Location'], None, "raw", True, True, False)

登录问题解决后，需处理状态保存。Python的Requests库功能强大，简单场景下可直接使用request.session进行会话管理。然而，由于项目涉及大量异步操作，需解决Cookie的存储与读取问题，此处采用pickle进行对象的序列化和反序列化。默认情况下，Cookie保存采用增量更新方式。

def save_cookies(cookies, overWrite=False):
  try:
    currentCookie = requests.utils.dict_from_cookiejar(cookies)
    if len(currentCookie) < 1:
      return
    oldCookie = requests.utils.dict_from_cookiejar(load_cookies())
    with open(config.COOKIE_FILE, 'w') as f:
      if not overWrite:
        cookieDict = dict(oldCookie, **currentCookie)
      else:
        cookieDict = requests.utils.dict_from_cookiejar(cookies)
      pickle.dump(cookieDict, f)
      print 'Saved cookie'
      print cookieDict
      f.close()
  except:
    print 'Save cookies failed', sys.exc_info()[0]
    sys.exit(99)


def load_cookies():
  try:
    with open(config.COOKIE_FILE, 'r') as f:
      cookies = requests.utils.cookiejar_from_dict(pickle.load(f))
      f.close()
  except:
    cookies = []
  return cookies

完成封装后，在发起requests.Session请求时，需加载已保存的Cookie，并在请求结束后保存新的Cookie。

s = requests.Session()
# 统一请求API
def fetchAPI(url, params=None, resultFormat="text", isNeedCookie=True, allowRedirects=True, saveCookie=False,
       method='GET'):
  try:
    cookies = load_cookies()
    if 'POST' == method:
      response = s.post(url, data=params, headers=config.Headers, cookies=cookies)
    else:
      response = s.get(url, params=params, headers=config.Headers, cookies=cookies,
               allow_redirects=allowRedirects)

    if "json" == resultFormat:
      result = response.json()
    elif "raw" == resultFormat:
      result = response
    else:
      result = response.text

    # if saveCookie:
    # print 'save cookie:' + str(response.cookies)
    save_cookies(response.cookies)

    return result

  except Exception, e:
    print e
    return False

完成上述两步后，后续请求可直接使用统一的API请求方法，运行效果良好。具体运行效果见截图：