master

分支 (1)

管理

管理

master

wph_rush_to_buy
/
vip_assistant.py

import time
import math
import random
import os
import pickle
import json

import requests
from requests.utils import dict_from_cookiejar
from requests.cookies import RequestsCookieJar
from bs4 import BeautifulSoup
from exception import AsstException
from log import logger
from util import (
    DEFAULT_USER_AGENT,
    get_random_useragent,
    save_image,
    response_status,
    parse_json,
    open_image,
    check_login
)

class Assistant(object):
    def __init__(self):
        self.user_agent = DEFAULT_USER_AGENT
        self.headers = {
            'User-Agent': self.user_agent,
            'ContentType': 'text/html; charset=utf-8',
            'Accept-Encoding': 'gzip, deflate, sdch',
            'Accept-Language': 'zh-CN,zh;q=0.8',
            'Connection': 'keep-alive',
        }
        self.nick_name = 'init_nick_name'
        self.is_login = False
        self.cookies_dict = {}
        self.add_id = ''

        self.qrToken = ''
        self.sess = requests.Session()
        try:
            self._load_cookies()
        except Exception:
            logger.error('读取本地cookie报错')
            pass

    def marsCreat(self, leng=32):
        t = math.ceil(time.time()*1000)
        x = "0123456789abcdef"
        tmp = ''
        for i in range(leng):
            r = math.ceil(random.random()*1E8) % len(x)
            tmp += x[r:r+1]

        return str(t) + "_" + tmp

    def _load_cookies(self):
        cookies_file = ''
        for name in os.listdir('./cookies'):
            if name.endswith('.cookies'):
                cookies_file = './cookies/{0}'.format(name)
                break
        with open(cookies_file, 'rb') as f:
            local_cookies = pickle.load(f)
        self.sess.cookies.update(local_cookies)
        res_cookies_dic = requests.utils.dict_from_cookiejar(local_cookies)
        self.cookies_dict = res_cookies_dic
        logger.info(self.cookies_dict)
        self.is_login = self._validate_cookies()

    def _save_cookies(self):
        cookies_file = './cookies/{0}.cookies'.format(self.nick_name)
        cookies_file_path = './new_cookies/cookies.text'
        # res_cookies_dic = requests.utils.dict_from_cookiejar(self.sess.cookies)
        directory = os.path.dirname(cookies_file)
        if not os.path.exists(directory):
            os.makedirs(directory)
        with open(cookies_file, 'wb') as f:
            pickle.dump(self.sess.cookies, f)

    def _validate_cookies(self):
        """
        验证cookies是否有效(是否登录)
        通过访问用户订单列表页进行判断：若未登录，将会重定向到登录页面。
        :return: cookies是否有效 True/False
        """
        # url = 'https://order.vip.com/order/orderlist?orderStatus=all&type=all&pageNum=1'
        # url = 'https://myi.vip.com/api/common/get_verify_token'
        url = 'https://myi.vip.com/api/new_point/detail_list?cur_page=1&page_size=10&time_scope=THREE_MONTH&type=ALL'
        headers = {
            'User-Agent': self.user_agent,
            'Referer': 'https://myi.vip.com/',
        }
        try:
            resp = self.sess.get(url=url, headers=headers, allow_redirects=False)
            if resp.status_code == requests.codes.OK:
                logger.info('cookies验证有效')
                return True
        except Exception as e:
            logger.error(e)

        logger.info('验证cookies无效')
        self.sess = requests.session()
        return False

    def _get_login_page(self):
        """
        获取登录页面
        url: https://passport.vip.com/login
        :return:
        """
        url = "https://passport.vip.com/login"
        resp = self.sess.get(url, headers=self.headers)
        # for k, v in resp.cookies.items():
        #     self.cookies[k] = v
        self.sess.cookies['mars_cid'] = self.marsCreat()
        self.headers['Host'] = 'passport.vip.com'
        self.headers['Origin'] = 'https://passport.vip.com'
        self.headers['Referer'] = 'https://passport.vip.com/login'
        return resp

    def _get_qrToken(self):
        """
        获取qrToken参数
        :return:
        """
        url = "https://passport.vip.com/qrLogin/initQrLogin"
        headers = {
            'User-Agent': self.user_agent,
            'Referer': 'https://passport.vip.com/login',
        }
        resp = self.sess.post(url, headers=headers)
        resp_json = parse_json(resp.text)
        if resp_json['code'] != 200:
            logger.error('获取qrToken参数失败')
            return False
        else:
            logger.info('获取qrToken参数成功')
            self.qrToken = resp_json['qrToken']
            return True

    def _get_QRcode(self):
        """
        获取二维码图片
        :return:
        """
        url = 'https://passport.vip.com/qrLogin/getQrImage'
        payload = {
            'qrToken': self.qrToken
        }
        headers = {
            'User-Agent': self.user_agent,
            'Referer': 'https://passport.vip.com/login',
        }
        resp = self.sess.get(url, headers=headers, params=payload)
        if not response_status(resp):
            logger.info('获取二维码失败')
            return False
        QRCode_file = 'QRcode.png'
        save_image(resp, QRCode_file)
        logger.info('二维码获取成功，请打开唯品会APP扫描')
        open_image(QRCode_file)
        return True

    def _get_QRCode_ticket(self):
        """
        检查是否扫描二维码
        {"code":200,"msg":"执行成功","status":"INVALID","redirectUrl":null}
        status: NOT_SCANNED--未扫描   SCANNED---已扫描   INVALID--无效  CONFIRMED--确认
        :return:
        """
        url = 'https://passport.vip.com/qrLogin/checkStatus'
        payload = {
            'qrToken': self.qrToken
        }
        headers = {
            'User-Agent': self.user_agent,
            'Referer': 'https://passport.vip.com/login',
        }
        resp = self.sess.post(url=url, headers=headers, params=payload)
        if not response_status(resp):
            raise AsstException('获取二维码扫描结果异常')
            return False

        resp_json = parse_json(resp.text)
        if resp_json['code'] != 200:
            logger.info('Code: %s, Message: %s, status: %s', resp_json['code'], resp_json['msg'], resp_json['status'])
            return None
        else:
            logger.info('Code: %s, Message: %s, status: %s', resp_json['code'], resp_json['msg'], resp_json['status'])
            if resp_json['status'] == 'NOT_SCANNED':
                logger.info('未扫描')
                return None
            elif resp_json['status'] == 'INVALID':
                logger.info('二维码过期，请重新获取扫描')
                return None
            elif resp_json['status'] == 'SCANNED':
                logger.info('已扫描，未确认')
                return None
            else:
                logger.info('已完成手机客户端确认')
                return resp_json['status']

    def login_by_QRcode(self):
        """
        二维码登录
        :return:
        """
        if self.is_login:
            logger.info('登录成功')
            return

        self._get_login_page()

        self._get_qrToken()

        if not self._get_QRcode():
            logger.info('二维码下载失败')

        for _ in range(30):
            ticket = self._get_QRCode_ticket()
            if ticket:
                break
            time.sleep(4)
        else:
            raise AsstException('二维码过期，请重新获取扫描')

        logger.info('二维码登录成功')
        self.is_login = True
        self.nick_name = self.get_user_info()
        self._save_cookies()


    @check_login
    def get_user_info(self):
        """
        获取用户信息
        :return:
        """
        url = 'https://myi.vip.com/api/account/base_info'
        headers = {
            'User-Agent': self.user_agent,
            'Referer': 'https://myi.vip.com/basicinfo.html',
        }
        try:
            resp = self.sess.get(url=url, headers=headers)
            resp_json = parse_json(resp.text)
            logger.info(resp_json.get('data'))
            return resp_json.get('data')['nickname'] or 'vip'
        except Exception:
            return 'vip'

    def _get_item_detail_page(self, sku_code, sku_id):
        """
        访问商品详情页
        :param sku_code:
        :param sku_id:
        :return: 响应
        """
        url = 'https://detail.vip.com/detail-{0}-{1}.html'.format(sku_code, sku_id)
        headers = {
            'User-Agent': self.user_agent,
        }
        logger.info('请求商品详情页：https://detail.vip.com/detail-{0}-{1}.html'.format(sku_code, sku_id))
        page = requests.get(url=url, headers=headers)
        cookies = {}
        for k, v in page.cookies.items():
            cookies[k] = v
        logger.info(cookies)
        # res_cookies_dic = requests.utils.dict_from_cookiejar(self.sess.cookies)
        # logger.info(res_cookies_dic)
        return page

    @check_login
    def if_item_can_be_ordered(self, sku_code, sku_id, areaId):
        """
        判断商品是否能下单
        :param sku_id:
        :param areaId:
        :return: True/False
        """
        return self.get_single_item_stock(sku_code=sku_code, sku_id=sku_id, areaId=areaId)

    def get_single_item_stock(self, sku_code, sku_id, areaId):
        """
        获取单个商品库存状态
        944103111102
        :param sku_id:
        :param areaId:
        :return:
        """
        page = self._get_item_detail_page(sku_code, sku_id)
        url = 'https://stock.vip.com/detail'
        payload = {
            'callback': 'stock_detail',
            'merchandiseId': sku_id,
            'is_old': 1,
            'areaId': areaId,
            '_': round(time.time() * 1000),
        }
        headers = {
            'User-Agent': self.user_agent,
            'Referer': 'https://detail.vip.com/'
        }
        try:
            resp = self.sess.get(url=url, headers=headers, params=payload)
            resp_json = parse_json(resp.text)
            item = resp_json.get('items')[0]
            stock = item['stock']
            self.add_id = item['id']
            logger.info('当前库存: %s', stock)
            return stock > 0
        except requests.exceptions.Timeout:
            logger.error('查询 %s 库存信息超时(%ss)', sku_id, self.timeout)
            return False
        except requests.exceptions.RequestException as request_exception:
            logger.error('查询 %s 库存信息发生网络请求异常：%s', sku_id, request_exception)
            return False
        except Exception as e:
            logger.error('查询 %s 库存信息发生异常, resp: %s, exception: %s', sku_id, e)
            return False

    def add_item_to_cart(self, sku_id, areaId):
        """
        添加商品到购物车
        :param sku_id:
        :return:
        """
        url = 'https://mapi.vip.com/vips-mobile/rest/cart/pc/add_cart'
        headers = {
            'User-Agent': self.user_agent,
            'Referer': url
        }
        payload = {
            'callback': 'addCart',
            'app_name': 'shop_pc',
            'app_version': '4.0',
            'warehouse': 'VIP_NH',
            'fdc_area_id': areaId,
            'client': 'pc',
            'mobile_platform': '1',
            'province_id': '104104',
            'api_key': '70f71280d5d547b2a7bb370a529aeea1',
            'user_id': '290090873',
            'mars_cid': self.cookies_dict['mars_cid'],
            'wap_consumer': 'c',
            'size_id': self.add_id,
            'size_num': 1,
            'product_id': sku_id,
            'source_app': 'pc',
            'captcha_id': '',
            'ticket': '',
            'is_reserve': '0',
            'cart_ver': '4',
            'functions': 'canChecked',
            '_': round(time.time() * 1000),
        }
        logger.info('下单请求：https://mapi.vip.com/vips-mobile/rest/cart/pc/add_cart')
        resp = self.sess.get(url=url, headers=headers, params=payload)
        resp_json = parse_json(resp.text)
        logger.info(resp_json)


    def _add_or_change_cart_item(self, sku_id, count=1):
        """
        添加商品到购物车
        :param sku_id:
        :param count:
        :return:
        """
        logger.info('开始加入购物车，数量 %s', count)


    def buy_item_in_stock(self, sku_code, sku_id, areaId, wait_all=False, stock_interval=3, submit_retry=3, submit_interval=5):
        """
        :param sku_code:
        :param sku_id:
        :param areaId:
        :param wait_all: 是否等所有商品都有货才一起下单，可选参数，默认False
        :param stock_interval: 查询库存时间间隔，可选参数，默认3秒
        :param submit_retry: 提交订单失败后重试次数，可选参数，默认3次
        :param submit_interval: 提交订单失败后重试时间间隔，可选参数，默认5秒
        :return:
        """
        if not wait_all:
            logger.info('下单模式：%s 任一商品有货并且未下架均会尝试下单', sku_id)
            while True:
                if not self.if_item_can_be_ordered(sku_code=sku_code, sku_id=sku_id, areaId=areaId):
                    logger.info('%s 不满足下单条件，%ss后进行下一次查询', sku_id, stock_interval)
                else:
                    logger.info('%s 满足下单条件，开始执行', sku_id)
                    self.add_item_to_cart(sku_id=sku_id, areaId=areaId)


                time.sleep(stock_interval)

    def arr(self):
        """
        https://mapi.vip.com/vips-mobile/rest/cart/pc/get_shopping_cart/v2  购物车获取订单
        :return:
        """