淘宝直播数据爬取 + 淘宝模拟登陆

目录

• 直播数据爬取
• 模拟登陆

直播数据爬取

可以在 js 数据中找到 sign 的加密方式

分析得知 sign 加密方式为 (d.token + "&" + 时间戳 + "&" + appkey + "&" + data)

d.token的值，发现这个值在cookie当中出现了，与时间戳结合在一起，这个token值会过期，大概是2小时

请求数据接口的时候请求头必须带cookie，且包含_m_h5_tk与_m_h5_tk_enc这两个字段，而且发现不需要登录就可以获取这个cookie，响应返回的信息当中也包括了这两个字段，这样就可以获取到token了

import requests
from urllib import parse

def _m_h5():

    data = parse.unquote(\'%7B%22param%......innerId%22%3A%22%22%7D\')
    params = {
            \'appKey\': 12574478,
            \'data\': data
    }
    # 请求空获取cookies
    url = \'https://acs.m.taobao.com/h5/mtop.taobao.social.feed.aggregate/1.0/\'
    resp = requests.get(url, params=params)
    cookiejar = requests.utils.dict_from_cookiejar(resp.cookies)
    m_ht_tk = cookiejar[\'_m_h5_tk\']
    m_h5_tk_enc = cookiejar[\'_m_h5_tk_enc\']
    print(m_ht_tk, m_h5_tk_enc)
    return m_ht_tk, m_h5_tk_enc


_m_h5()

import requests
import time
import json
import hashlib

from urllib import parse
from openpyxl import load_workbook
from apscheduler.schedulers.blocking import BlockingScheduler


class TaobaoLiveSpider:
    def __init__(self):
        self.start_time = parse.quote(\'2020-10-15 11:00:11\')
        self.end_time = parse.quote(\'2020-11-14 11:00:11\')
        # 观看次数 固定格式
        self.data = parse.unquote(f\'%7B%22param%2_abstract_indicator%5C%22%2C%5C%22queryDetail%5C%22%3Afalse%2C%5C%22startTime%5C%22%3A%5C%22{self.start_time}%5C%22%2C%5C%22endTime%5C%22%3A%5C%22{self.end_time}%5C%22%2C%5C%22timeType%5C%22%3A2%2C%5C%22sign%5C%2AfaC%5C%22%7D%5D%5C%22%2C%5C%22extra%5C%22%3Anull%7D%22%2C%22innerId%22%3A%22%22%7D\')  # 每次请求需携带的 data，错误会报 非法请求参数
        # 引导交易 固定格式
        self.guid_deal_data = parse.unquote(f\'%7B%22paracbot_slr_lime_rpt_ov_deal%5C%22%2C%5C%22queryDetail%5C%22%3Afalse%2C%5C%22startTime%5C%22%3A%5C%22{self.start_time}%5C%22%2C%5C%22endTime%5C%22%3A%5C%22{self.end_time}%5C%22%2C%5C%22timeType%5C%22%3A2%2C%5C%22sign%5C%22%3Anull%2C%5C%22limit%5C%22%3A1%2C%5C%22row%5C%22%3A%5C%22%5B%5D%5C%22%2C%5C%22measure%5C%22%3A%5C%22%5B%...22%22%7D\')
        # 粉丝平均在线 固定格式
        self.fans_average_on_line_time_data = parse.unquote(f\'%7B%22pabstract_indicator%5C%22%2C%5C%22queryDetail%5C%22%3Afalse%2C%5C%22startTime%5C%22%3A%5C%22{self.start_time}%5C%22%2C%5C%22endTime%5C%22%3A%5C%22{self.end_time}%5C%22%2C%5C%22timeType%5C%22%3A2%2C%5C%22sign%5C%25C%5C%...2%7D\')
        # print(self.data)
        # print(self.guid_deal_data)
        # print(self.fans_average_on_line_time_data)

        self.now_time = str(round(time.time(), 3)).replace(\'.\', \'\')  # 13 当前位时间戳
        self.url = \'https://h5api.m.taobao.com/h5/mtop.alibaba.iic.xinsightshop.olap.query/1.0/\'
        self.m_ht_tk, self.m_h5_tk_enc = \'\', \'\'
        self.local_now_time = time.strftime(\'%Y-%m-%d %H:%M:%S\')
        self.line = 2

    def get_taobao_cookie(self):
        """
        cookie 字段
        :return: m_ht_tk, m_h5_tk_enc
        """
        # data = parse.unquote(data)
        params = {
            \'appKey\': 27522***,
            \'data\': self.data
        }
        url = \'https://acs.m.taobao.com/h5/mtop.taobao.social.feed.aggregate/1.0/\'
        resp = requests.get(url, params=params)
        cookiejar = requests.utils.dict_from_cookiejar(resp.cookies)
        self.m_ht_tk = cookiejar[\'_m_h5_tk\']
        self.m_h5_tk_enc = cookiejar[\'_m_h5_tk_enc\']

    def headers(self):
        self.get_taobao_cookie()
        m_ht_tk, m_h5_tk_enc = self.m_ht_tk, self.m_h5_tk_enc
        # print(\'m_ht_tk, m_h5_tk_enc\', m_ht_tk, m_h5_tk_enc)
        return {
            \'cookie\': f\'_samesite_flag_=true; enc=zUS4Q0thPUweeJoV2t7U6F7Boh...cookie14=Uoe0b0ORnwCV7g%3D%3D; _m_h5_tk={m_ht_tk}; _m_h5_tk_enc={m_h5_tk_enc}; l=eBQZK...M-z4WFdhUU3iV4LClmbXGZtE1Ab_kVQUg5TgQGwt5iPeN5-BXtyeA6i1..\',
            \'origin\': \'https://databot.taobao.com\',
            \'referer\': \'https://databot.taobao.com/\',
            \'user-agent\': \'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/83.0.4103.116 Safari/537.36\'
        }

    def params(self, data):
        return {
            \'jsv\': \'2.6.0\',
            \'appKey\': \'27522***\',
            \'t\': self.now_time,
            \'sign\': self.sign(data),
            \'api\': \'mtop.alibaba.iic.xinsightshop.olap.query\',
            \'v\': \'1.0\',
            \'type\': \'originaljson\',
            \'dataType\': \'json\',
            \'timeout\': 20000,
            \'H5Request\': \'true\',
            \'data\': data
                }

    def sign(self, data):
        """
        sign 加密方式
        :return: sign 字符串
        """
        # m_ht_tk, _ = self.get_taobao_cookie(self.data)
        m_ht_tk = self.m_ht_tk.split(\'_\')[0]
        sign = m_ht_tk + \'&\' + self.now_time + \'&\' + \'27522***\' + \'&\' + data
        sign = hashlib.md5(sign.encode())
        sign = sign.hexdigest()
        return sign

    def spider(self):
        resp = requests.get(self.url, headers=self.headers(), params=self.params(self.data)).json()
        print(resp)
        visit = resp[\'data\'][\'data\'][\'cellset\'][1]
        watching_count = visit[5][\'value\']
        click_probability = visit[7][\'value\']
        on_line_count = visit[6][\'value\']
        average_on_line_time = visit[2][\'value\']
        live_room_watching_count = visit[0][\'value\']
        fans_make_up = visit[-2][\'value\']
        extra_add_flow = visit[1][\'value\']
        add_fans = visit[4][\'value\']
        goods_click = visit[3][\'value\']
        goods_click_fans_make_up = visit[-1][\'value\']

        resp = requests.get(self.url, headers=self.headers(), params=self.params(self.guid_deal_data)).json()
        visit = resp[\'data\'][\'data\'][\'cellset\'][1]
        guid_deal_count = visit[0][\'value\']
        guid_deal_percent = visit[1][\'value\']
        guid_deal_money = visit[2][\'value\']
        guid_deal_money_percent = visit[3][\'value\']

        resp = requests.get(self.url, headers=self.headers(), params=self.params(self.fans_average_on_line_time_data)).json()
        visit = resp[\'data\'][\'data\'][\'cellset\'][1]
        fans_average_on_line_time = visit[0][\'value\']

        print(watching_count, click_probability, on_line_count, average_on_line_time, fans_average_on_line_time,
              live_room_watching_count, fans_make_up, extra_add_flow,
              add_fans, goods_click, goods_click_fans_make_up, guid_deal_count, guid_deal_percent,
              guid_deal_money, guid_deal_money_percent, self.local_now_time)
        return watching_count, click_probability, on_line_count, average_on_line_time, fans_average_on_line_time,\
              live_room_watching_count, fans_make_up, extra_add_flow,\
              add_fans, goods_click, goods_click_fans_make_up, guid_deal_count, guid_deal_percent,\
              guid_deal_money, guid_deal_money_percent, self.local_now_time

    def write_excel(self):
        workbook = load_workbook(\'data.xlsx\')
        wb = workbook[\'Sheet1\']
        line = wb.max_row
        watching_count, click_probability, on_line_count, average_on_line_time, fans_average_on_line_time, \
        live_room_watching_count, fans_make_up, extra_add_flow, \
        add_fans, goods_click, goods_click_fans_make_up, guid_deal_count, guid_deal_percent, \
        guid_deal_money, guid_deal_money_percent, self.local_now_time = self.spider()
        line += 1
        wb[f\'A{line}\'] = self.local_now_time
        wb[f\'B{line}\'] = watching_count
        wb[f\'C{line}\'] = click_probability
        wb[f\'D{line}\'] = on_line_count
        wb[f\'E{line}\'] = average_on_line_time
        wb[f\'F{line}\'] = fans_average_on_line_time
        wb[f\'G{line}\'] = live_room_watching_count
        wb[f\'H{line}\'] = fans_make_up
        wb[f\'I{line}\'] = extra_add_flow
        wb[f\'J{line}\'] = add_fans
        wb[f\'K{line}\'] = goods_click
        wb[f\'L{line}\'] = goods_click_fans_make_up
        wb[f\'M{line}\'] = guid_deal_count
        wb[f\'N{line}\'] = guid_deal_percent
        wb[f\'O{line}\'] = guid_deal_money
        wb[f\'P{line}\'] = guid_deal_money_percent
        workbook.save(\'data.xlsx\')

def main():
    run = TaobaoLiveSpider()
    run.write_excel()


scheduler = BlockingScheduler()
scheduler.add_job(main, \'interval\', seconds=1800, id=\'main\')
scheduler.start()

模拟登陆

import re
import os
import json

import requests


s = requests.Session()
# cookies序列化文件
COOKIES_FILE_PATH = \'taobao_login_cookies.txt\'


class UsernameLogin:

    def __init__(self, loginId, umidToken, ua, password2):
        """
        账号登录对象
        :param loginId: 用户名
        :param umidToken: 新版登录新增参数
        :param ua: 淘宝的ua参数
        :param password2: 加密后的密码
        """
        # 检测是否需要验证码的URL
        self.user_check_url = \'https://login.taobao.com/newlogin/account/check.do?appName=taobao&fromSite=0\'
        # 验证淘宝用户名密码URL
        self.verify_uaername_password_url = "https://login.taobao.com/newlogin/login.do?appName=taobao&fromSite=0"
        # 访问st码URL
        self.vst_url = \'https://login.taobao.com/member/vst.htm?st={}\'
        # 淘宝个人 主页
        # self.my_taobao_url = \'http://i.taobao.com/my_taobao.htm\'
        self.my_taobao_url = \'https://zhaoshang.tmall.com/channel/index.htm?\'

        # 淘宝用户名
        self.loginId = loginId
        # 淘宝用户名
        self.umidToken = umidToken
        # 淘宝关键参数，包含用户浏览器等一些信息，很多地方会使用，从浏览器或抓包工具中复制，可重复使用
        self.ua = ua
        # 加密后的密码，从浏览器或抓包工具中复制，可重复使用
        self.password2 = password2

        # 请求超时时间
        self.timeout = 3

    def _user_check(self):
        """
        检测账号是否需要验证码
        :return:
        """
        data = {
            \'loginId\': self.loginId,
            \'ua\': self.ua,
        }
        try:
            response = s.post(self.user_check_url, data=data, timeout=self.timeout)
            response.raise_for_status()
        except Exception as e:
            print(f\'检测是否需要验证码请求失败，原因：{e}\')
            raise e
        check_resp_data = response.json()[\'content\'][\'data\']
        needcode = False
        # 判断是否需要滑块验证，一般短时间密码错误多次可能出现
        if \'isCheckCodeShowed\' in check_resp_data:
            needcode = True
        print(\'是否需要滑块验证：{}\'.format(needcode))
        return needcode

    def _get_umidToken(self):
        """
        获取umidToken参数
        :return:
        """
        response = s.get(\'https://login.taobao.com/member/login.jhtml\')
        st_match = re.search(r\'"umidToken":"(.*?)"\', response.text)
        print(st_match.group(1))
        return st_match.group(1)

    @property
    def verify_login_password(self):
        """
        验证用户名密码，并获取st码申请URL
        :return: 验证成功返回st码申请地址
        """
        headers = {
            \'Referer\': \'https://login.taobao.com/member/login.jhtml\',
            \'User-Agent\': \'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/80.0.3987.163 Safari/537.36\',
            \'Content-Type\': \'application/x-www-form-urlencoded\',
            \'origin\': \'https://login.taobao.com\'
        }
        # 登录toabao.com提交的数据，如果登录失败，可以从浏览器复制你的form data
        data = {
            \'loginId\': self.loginId,
            \'password2\': self.password2,
            \'keepLogin\': \'false\',
            \'ua\': self.ua,
            \'umidGetStatusVal\': \'255\',
            \'screenPixel\': \'1920x1080\',
            \'navlanguage\': \'zh-CN\',
            \'navUserAgent\': \'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/85.0.4183.121 Safari/537.36\',
            \'navPlatform\': \'Win32\',
            \'appName\': \'taobao\',
            \'appEntrance\': \'taobao_pc\',
            \'_csrf_token\': \'kmnF0VwnseLLEq0kw7qLI\',
            \'umidToken\': \'899a8d2fdf9d6a93a5d7472bcb8db95227a9e034\',
            \'hsiz\': \'167696b8b2d8a066b7b0e4e0e140d905\',
            \'bizParams\': \'\',
            \'style\': \'default\',
            \'appkey\': \'00000000\',
            \'from\': \'tbTop\',
            \'isMobile\': \'false\',
            \'lang\': \'zh_CN\',
            \'returnUrl\': \'https://www.taobao.com/\',
            \'fromSite\': \'0\'
        }

        try:
            response = s.post(self.verify_uaername_password_url, headers=headers, data=data,
                              timeout=self.timeout)
            response.raise_for_status()
            # 从返回的页面中提取申请st码地址
        except Exception as e:
            print(\'验证用户名和密码请求失败，原因：\')
            raise e
        # 提取申请st码url
        print(response.json())
        apply_st_url_match = response.json()[\'content\'][\'data\'][\'asyncUrls\'][0]
        # 存在则返回
        if apply_st_url_match:
            print(\'验证用户名密码成功，st码申请地址：{}\'.format(apply_st_url_match))
            return apply_st_url_match
        else:
            raise RuntimeError(\'用户名密码验证失败！response：{}\'.format(response.text))

    def _apply_st(self):
        """
        申请st码
        :return: st码
        """
        apply_st_url = self.verify_login_password
        try:
            response = s.get(apply_st_url)
            response.raise_for_status()
        except Exception as e:
            print(\'申请st码请求失败，原因：\')
            raise e
        st_match = re.search(r\'"data":{"st":"(.*?)"}\', response.text)
        if st_match:
            print(\'获取st码成功，st码：{}\'.format(st_match.group(1)))
            return st_match.group(1)
        else:
            raise RuntimeError(\'获取st码失败！response：{}\'.format(response.text))

    def login(self):
        """
        使用st码登录
        :return:
        """
        # 加载cookies文件
        if self._load_cookies():
            return True
        # 判断是否需要滑块验证
        self._user_check()
        st = self._apply_st()
        headers = {
            \'Host\': \'login.taobao.com\',
            \'Connection\': \'Keep-Alive\',
            \'User-Agent\': \'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.36\'
        }
        try:
            response = s.get(self.vst_url.format(st), headers=headers)
            response.raise_for_status()
        except Exception as e:
            print(\'st码登录请求，原因：\')
            raise e
        # 登录成功，提取跳转淘宝用户主页url
        my_taobao_match = re.search(r\'top.location.href = "(.*?)"\', response.text)
        if my_taobao_match:
            print(\'登录淘宝成功，跳转链接：{}\'.format(my_taobao_match.group(1)))
            self.my_taobao_url = my_taobao_match.group(1)
            self._serialization_cookies()
            return True
        else:
            raise RuntimeError(\'登录失败！response：{}\'.format(response.text))

    def _load_cookies(self):
        # 1、判断cookies序列化文件是否存在
        if not os.path.exists(COOKIES_FILE_PATH):
            return False
        # 2、加载cookies
        s.cookies = self._deserialization_cookies()
        # 3、判断cookies是否过期
        try:
            self.get_taobao_nick_name()
        except Exception as e:
            os.remove(COOKIES_FILE_PATH)
            print(\'cookies过期，删除cookies文件！\')
            return False
        print(\'加载淘宝cookies登录成功!!!\')
        return True

    def _serialization_cookies(self):
        """
        序列化cookies
        :return:
        """
        cookies_dict = requests.utils.dict_from_cookiejar(s.cookies)
        print(cookies_dict)
        with open(COOKIES_FILE_PATH, \'w+\', encoding=\'utf-8\') as file:
            json.dump(cookies_dict, file)
            print(\'保存cookies文件成功！\')

    def _deserialization_cookies(self):
        """
        反序列化cookies
        :return:
        """
        with open(COOKIES_FILE_PATH, \'r+\', encoding=\'utf-8\') as file:
            cookies_dict = json.load(file)
            cookies = requests.utils.cookiejar_from_dict(cookies_dict)
            return cookies

    def get_taobao_nick_name(self):
        """
        获取淘宝昵称
        :return: 淘宝昵称
        """
        headers = {
            \'User-Agent\': \'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.36\'
        }
        params = {
            \'spm\': \'687.8433302/new.sidebar.1.296f226aVUQtum\',
            \'spm\': \'a217wi.openworkbeanchtmall_web\'
        }
        try:
            response = s.get(self.my_taobao_url, headers=headers)
            # print(response.text)
            response.raise_for_status()
        except Exception as e:
            print(\'获取淘宝主页请求失败！原因：\')
            raise e
        # 提取淘宝昵称
        # nick_name_match = re.search(r\'<input id="mtb-nickname" type="hidden" value="(.*?)"/>\', response.text)
        nick_name_match = re.findall("erNick = \'(.*?)\'", response.text)[0]
        if nick_name_match:
            print(f\'登录淘宝成功，你的用户名是：{nick_name_match}\')
            return nick_name_match
        else:
            raise RuntimeError(\'获取淘宝昵称失败！response：{}\'.format(response.text))

    def get_live(self):
        headers = {
            \'User-Agent\': \'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.36\',
            \'referer\': \'https://liveplatform.taobao.com/live/liveList.htm?\'
        }
        url = \'https://liveplatform.taobao.com/live/action.do?currentPage=1&pagesize=20&api=get_live_list\'
        response = s.get(url, headers=headers).text
        print(response)


if __name__ == \'__main__\':
    # 说明：loginId、umidToken、ua、password2这4个参数都是从浏览器登录页面复制过来的。
    # 如何复制4个参数：
    # # 1、浏览器打开：https://login.taobao.com/member/login.jhtml
    # # 2、F12打开调试窗口，左边有个Preserve log，勾选上，这样页面跳转请求记录不会丢失
    # # 3、输入用户名密码登录，然后找到请求：newlogin/login.do 这个是登录请求
    # # 4、复制上面的4个参数到下面，基本就可以运行了
    # # 5、如果运行报错可以微信私聊猪哥，没加猪哥微信的可以关注猪哥微信公众号[裸睡的猪]，回复：加群

    # 淘宝用户名：手机 用户名 都可以
    loginId = \'username\'
    # 改版后增加的参数，后面考虑解密这个参数
    umidToken = \'ae2d51d\',
    # 淘宝重要参数，从浏览器或抓包工具中复制，可重复使用
    ua = \'137#Qtc9hE9o9IpDz/4p38vMDW2hkgwdkrNMhdtCDn7DaPtK8zuCzjGfX9gy1gOwsfS1\'
    # 加密后的密码，从浏览器或抓包工具中复制，可重复使用
    password2 = \'c3107629fbd23e5b0b6c4696e146d7207299d6d80fb8b00154a143736cb\'

    ul = UsernameLogin(loginId, umidToken, ua, password2)
    ul.login()
    ul.get_taobao_nick_name()
    ul.get_live()