buff1.0/buff/other/xpjb-01.py

# BUFF选品01脚本

import re
from traceback import print_tb

import requests as req
import time,random,numpy as np
from datetime import datetime
from exceptiongroup import catch
import argparse
import mysql.connector
from numpy.ma.core import argsort
from selenium.webdriver.common.devtools.v85.system_info import ProcessInfo
import json




class xpjb():
    def __init__(self):
        # 数据库配置
        self.host = "121.40.50.77"
        self.user = "root"
        self.password = "Root@123456"
        self.database = "ry-vue"

        # 程序执行次数
        self.total_run_count = 0
        # 入参
        self.args = None

        # 选品账号的email
        self.xpzh_email = None
        # 选品账号的email_dh
        self.xpzh_email_dh = None
        # 选品账号的cookie
        self.xpzh_cookie = None
        # 选品有效品类信息
        self.xp_category_list=[]
        # 选品品质信息
        self.xp_rarity_list=[]
        # 选品类别信息
        self.xp_quality_list=[]
        # 选品外观信息
        self.xp_exterior_list=[]
        # 选品收藏品信息
        self.xp_itemset_list=[]

        # 已检索的BUFF商品数
        self.searchedGoodsNum =0;
        #BUFF搜寻商品入参
        self.search_good_param_init = {
            'game': 'csgo'
        }
        #BUFF搜寻商品入参数list,因为有些内容存在多个选择，检索条件采取组合模式，分批次请求，请求中加入暂停，以防锁号
        self.search_good_param_list=[]

        # buff选品接口配置
        self.goodsurl = 'https://buff.163.com/api/market/goods'
        self.my_headers = {
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/74.0.3729.131 Safari/537.36",
            "Host": "buff.163.com",
            "Referer": "https://buff.163.com/market/?game=csgo",
            "X-Requested-With": "XMLHttpRequest"
        }

    # 判断嵌套属性是否存在
    @staticmethod
    def has_nested_attribute(obj, attr_path):
        attrs = attr_path.split('.')
        try:
            for attr in attrs:
                obj = obj[attr] if isinstance(obj, dict) else getattr(obj, attr)
            return True
        except (KeyError, AttributeError):
            return False

    # 处理命令行参数
    def dealArgc(self):
        try:
            print('处理命令行参数--开始')
            # 创建解析器
            parser = argparse.ArgumentParser(description='BUFF选品脚本')
            # 添加一个位置参数，表示运行多久重新再运行 默认10秒
            parser.add_argument('--runtime', type=str,default=10, help='单次运行时间 默认10秒')
            # 添加一个位置参数，表示自动运行多少次程序，默认为10次
            parser.add_argument('--runcount', type=int,default=10, help='程序运行总次数,默认为10次')
            # 添加一个位置参数，表示每次调用BUFF接口程序运行间隔多少秒，默认为10秒
            parser.add_argument('--timeinterval', type=int,default=5, help='每次调用BUFF接口程序运行间隔多少秒，默认为10秒')
            # 添加一个位置参数，表示选品账号的手机号码
            parser.add_argument('--xp_phone', type=str ,default=17756576757, help='选品账号的手机号码，17756576757')
            # 添加一个位置参数，表示选品脚本停止条件标志，1 按照成功数量，2 按照页数和每页数量 默认为1
            parser.add_argument('--xptz_status', type=int ,default=1, help='选品脚本停止条件标志，1 按照成功数量，2 按照页数和每页数量 默认为1')
            # 添加一个位置参数，表示选品脚本停止条件标志的成功数量， 默认 200
            parser.add_argument('--xptz_success_num', type=int ,default=200, help='选品脚本停止条件标志的成功数量， 默认 200')
            # 添加一个位置参数，表示选品脚本停止条件标志的页数 默认 100
            parser.add_argument('--xptz_page_num', type=int ,default=100, help='选品脚本停止条件标志的页数 默认 100')
            # 添加一个位置参数，表示选品脚本停止条件标志的每页数量 默认 80
            parser.add_argument('--xptz_page_size', type=int ,default=80, help='选品脚本停止条件标志的每页数量 默认 80')
            # 添加一个位置参数，表示选品最低价(元)
            parser.add_argument('--xp_min_price', help='选品最低价')
            # 添加一个位置参数，表示选品最高价(元)
            parser.add_argument('--xp_max_price', help='选品最高价')
            # 添加一个位置参数，表示选品排序模式 price.asc 价格升序 price.desc 价格降序 sell_num.desc 在售数量
            parser.add_argument('--xp_sort',type=str, help='选品排序模式')

            # 解析命令行参数
            self.args = parser.parse_args()
            # 使用参数
            print(f'单次运行时间，默认10秒: {self.args.runtime}')
            print(f'自动运行多少次程序 默认10次: {self.args.runcount}')
            print(f'每次调用BUFF接口程序运行间隔多少秒 默认5秒: {self.args.timeinterval}')
            print(f'选品账号的手机号码: {self.args.xp_phone}')
            print(f'选品脚本停止条件标志: {self.args.xptz_status}')
            print(f'选品脚本停止条件标志的成功数量 默认 200: {self.args.xptz_success_num}')
            print(f'选品脚本停止条件标志的页数 默认 100: {self.args.xptz_page_num}')
            print(f'选品脚本停止条件标志的每页数量 默认 80: {self.args.xptz_page_size}')
            print(f'选品最低价: {self.args.xp_min_price}')
            print(f'选品最高价: {self.args.xp_max_price}')
            print(f'选品排序模式: {self.args.xp_sort}')

            print('处理命令行参数--成功')
        except Exception as e:
            print('处理命令行参数--失败',e)
            exit()

    # 获取选品品类信息
    def getXpCategory(self):
        print('获取选品品类信息')
        try:
            # 创建mysql连接
            conn = mysql.connector.connect(
                host=self.host,
                user=self.user,
                password=self.password,
                database=self.database
            )
            # 创建游标对象
            cursor = conn.cursor()
            # 执行SQL查询
            cursor.execute(f"select category_code from sys_category where del_flag = '0' and status = '0' ")
            # 获取所有记录列表
            results = cursor.fetchall()
            if len(results) == 0:
                print('选品品类信息不存在，请检查数据库配置')
                self.xp_category_list=[]

            for row in results:
                self.xp_category_list.append(row[0])

            # 关闭游标和连接
            cursor.close()
            conn.close()
        except Exception as e:
            print('获取选品品类信息--失败',e)
            exit()

    # 获取选品品质信息
    def getXpRarity(self):
        print('获取选品品质信息')
        try:
            # 创建mysql连接
            conn = mysql.connector.connect(
                host=self.host,
                user=self.user,
                password=self.password,
                database=self.database
            )
            # 创建游标对象
            cursor = conn.cursor()
            # 执行SQL查询
            cursor.execute(f"select quality_code from sys_quality where del_flag = '0' and quality_status = '0' ")
            # 获取所有记录列表
            results = cursor.fetchall()
            if len(results) == 0:
                print('选品品质信息不存在，请检查数据库配置')
                self.xp_rarity_list=[]

            for row in results:
                self.xp_rarity_list.append(row[0])

            # 关闭游标和连接
            cursor.close()
            conn.close()
        except Exception as e:
            print('获取选品品质信息--失败',e)
            exit()

    # 获取选品类别信息
    def getXpQuality(self):
        print('获取选品类别信息')
        try:
            # 创建mysql连接
            conn = mysql.connector.connect(
                host=self.host,
                user=self.user,
                password=self.password,
                database=self.database
            )
            # 创建游标对象
            cursor = conn.cursor()
            # 执行SQL查询
            cursor.execute(f"select type_code from sys_type where del_flag = '0' and typestatus = '0' ")
            # 获取所有记录列表
            results = cursor.fetchall()
            if len(results) == 0:
                print('选品类别信息不存在，请检查数据库配置')
                self.xp_quality_list=[]

            for row in results:
                self.xp_quality_list.append(row[0])

            # 关闭游标和连接
            cursor.close()
            conn.close()
        except Exception as e:
            print('获取选品类别信息--失败',e)
            exit()

    # 获取选品外观信息
    def getXpExterior(self):
        print('获取选品外观信息')
        try:
            # 创建mysql连接
            conn = mysql.connector.connect(
                host=self.host,
                user=self.user,
                password=self.password,
                database=self.database
            )
            # 创建游标对象
            cursor = conn.cursor()
            # 执行SQL查询
            cursor.execute(f"select appear_code from sys_appear where del_flag = '0' and appear_sta = '0' ")
            # 获取所有记录列表
            results = cursor.fetchall()
            if len(results) == 0:
                print('选品外观信息不存在，请检查数据库配置')
                self.xp_exterior_list=[]

            for row in results:
                self.xp_exterior_list.append(row[0])

            # 关闭游标和连接
            cursor.close()
            conn.close()
        except Exception as e:
            print('获取选品外观信息--失败',e)
            exit()

    # 获取选品收藏品信息
    def getXpItemset(self):
        print('获取选品收藏品信息')
        try:
            # 创建mysql连接
            conn = mysql.connector.connect(
                host=self.host,
                user=self.user,
                password=self.password,
                database=self.database
            )
            # 创建游标对象
            cursor = conn.cursor()
            # 执行SQL查询
            cursor.execute(f"select colect_code from sys_collect where del_flag = '0' and colect_sta = '0' ")
            # 获取所有记录列表
            results = cursor.fetchall()
            if len(results) == 0:
                print('选品收藏品信息不存在，请检查数据库配置')
                self.xp_itemset_list=[]

            for row in results:
                self.xp_itemset_list.append(row[0])

            # 关闭游标和连接
            cursor.close()
            conn.close()
        except Exception as e:
            print('获取选品收藏品信息--失败',e)
            exit()

    # 获取选品账号信息
    def getXpzh(self):
        print('获取选品账号信息')
        try:
            # 创建mysql连接
            conn = mysql.connector.connect(
                host=self.host,
                user=self.user,
                password=self.password,
                database=self.database
            )
            # 创建游标对象
            cursor = conn.cursor()
            # 执行SQL查询
            cursor.execute(f"SELECT id,phone,email,email_dh,cookie FROM sc_xpzh where phone = '{self.args.xp_phone}' and is_del='0' ")
            # 获取所有记录列表
            results = cursor.fetchall()
            if len(results) == 0:
                print('选品账号信息不存在，请检查手机号码是否正确')
                exit()

            for row in results:
                self.xpzh_email = row[2]
                self.xpzh_email_dh = row[3]
                self.xpzh_cookie = row[4]

            # 关闭游标和连接
            cursor.close()
            conn.close()
            print(f"选品账号手机号码： {self.args.xp_phone}")
            print(f"选品账号email：{self.xpzh_email}")
            print(f"选品账号cookie：{self.xpzh_cookie}")
            print(f"选品账号email：{self.xpzh_email}")
        except Exception as e:
            print('获取选品账号信息--失败',e)
            exit()

    # 获取检索BUFF物品的API接口入参
    def getSearchGoodParam(self):
        print('获取检索BUFF物品API接口入参')
        # 过滤掉空的列表
        non_empty_lists = {
            'category': self.xp_category_list,
            'rarity': self.xp_rarity_list,
            'quality': self.xp_quality_list,
            'exterior': self.xp_exterior_list,
            'itemset': self.xp_itemset_list
        }
        # 只保留非空的列表
        filtered_lists = {k: v for k, v in non_empty_lists.items() if v}

        # 如果所有列表都为空，返回空列表
        if not filtered_lists:
            return []
        # 生成所有可能的组合
        # 使用itertools.product来生成所有可能的组合
        from itertools import product

        # 获取所有非空列表的值
        values = list(filtered_lists.values())
        # 获取所有非空列表的键
        keys = list(filtered_lists.keys())

        for combination in product(*values):
            combination_dict = {keys[i]: combination[i] for i in range(len(keys))}
            self.search_good_param_list.append(combination_dict)

        for i in range(len(self.search_good_param_list)):
            self.search_good_param_list[i]['category_group'] = self.search_good_param_list[i]['category']

    # 生成检索BUFF物品的API接口最终入参
    def generate_combinations(self):
        #当self.search_good_param_list为空，加入一个空对象
        if len(self.search_good_param_list) == 0:
            self.search_good_param_list.append({})
        #遍历search_good_param_list数组，每个数组中增加search_good_param_init的值
        for i in range(len(self.search_good_param_list)):
            self.search_good_param_list[i].update(self.search_good_param_init)
        #xptz_status为2时候，将search_good_param_list数组中的每个数组的xptz_page_num和xptz_page_size
        if self.args.xptz_status == 2:
            for i in range(len(self.search_good_param_list)):
                self.search_good_param_list[i]['page_num'] = self.args.xptz_page_num
                self.search_good_param_list[i]['page_size'] = self.args.xptz_page_size
        elif self.args.xptz_status == 1:
            for i in range(len(self.search_good_param_list)):
                self.search_good_param_list[i]['page_num'] = self.args.xptz_page_num
                self.search_good_param_list[i]['page_size'] = self.args.xptz_page_size
        #选品最低价不为空，则将search_good_param_list数组中的每个数组的xptz_min_price的值设置为命令行参数--xptz_min_price的值
        if self.args.xp_min_price != None:
            for i in range(len(self.search_good_param_list)):
                self.search_good_param_list[i]['min_price'] = self.args.xp_min_price
        #选品最高价不为空，则将search_good_param_list数组中的每个数组的xptz_max_price的值设置为命令行参数--xptz_max_price的值
        if self.args.xp_max_price != None:
            for i in range(len(self.search_good_param_list)):
                self.search_good_param_list[i]['max_price'] = self.args.xp_max_price
        #选品排序值不为空，则将search_good_param_list数组中的每个数组的xptz_sort_type的值设置为命令行参数--xptz_sort_type的值
        if self.args.xp_sort != None:
            for i in range(len(self.search_good_param_list)):
                self.search_good_param_list[i]['sort_by'] = self.args.xp_sort

    # 选品
    def SearchGoods(self):
        print('选品开始')
        #simpleSearchGoods函数执行runcount次数，自动停止，两次之间暂停runtime秒
        for i in range(self.args.runcount):
            self.simpleSearchGoods(i)
            time.sleep(self.args.runtime)
        print('选品结束')

    # 单次选品
    def simpleSearchGoods(self,i):
        # 如果搜寻的BUFF商品已完成，则停止脚本
        if self.searchedGoodsNum >= self.args.xptz_success_num:
            print('搜寻的BUFF商品已完成，停止脚本')
            exit()
        # print(f'选品程序第{i}次执行开始')
        # print(self.search_good_param_list)
        # 遍历search_good_param_list
        for param in self.search_good_param_list:
            # print('选品入参：',param)
            # 获取BUFF商品信息
            gooditems = self.callMarketGoods(param)
            # print(gooditems)
            # 判断是否满足条件
            # 此处调整到单独的脚本xpjb-02.py处理
        # print(f'选品程序第{i}次执行结束')

    # 请求BUFF物品的API接口
    def callMarketGoods(self,param):
        # print('获取商品信息')
        #Cookie赋值
        self.my_headers.update({
            'Cookie': self.xpzh_cookie
        })
        datalist = []
        try:
            res = req.get(url=self.goodsurl, headers=self.my_headers, params=param)
            # if res.json()['code'] != 'OK' :
            print('选品接口请求返回报文:', res.text.strip())
                # return
            data = res.json()['data']
            datalist.extend(data['items'])
            total_page_num= data['total_page']
            # datalist 入数据库
            self.insertGoods(datalist)
            time.sleep(self.args.timeinterval)
            if total_page_num >= param['page_num']:
                total_page_num = param['page_num']
            # print('商品页数'+str(total_page_num))
            if int(total_page_num) > 1:
                for i in range(1, total_page_num):
                    time.sleep(self.args.timeinterval)
                    param.update({
                        'page_num': i
                    })
                    try:
                        datalist=[]
                        res = req.get(url=self.goodsurl, headers=self.my_headers, params=param)
                        # if res.json()['code'] != 'OK' :
                        print('选品接口请求返回报文:', res.text.strip())
                            # continue
                        data = res.json()['data']
                        datalist.extend(data['items'])
                        # datalist 入数据库
                        self.insertGoods(datalist)
                    except Exception as e:
                        print('选品接口请求--失败',e)
        except Exception as e:
            print('选品接口请求--失败',e)
        # print('数据获取成功')  # 打印数据列表
        return

    def insertGoods(self,datalist):
        # print('buff商品数据入库')
        try:
            # 创建mysql连接
            conn = mysql.connector.connect(
                host=self.host,
                user=self.user,
                password=self.password,
                database=self.database
            )
            cursor = conn.cursor()
            #遍历datalist
            for item in datalist:
                self.searchedGoodsNum += 1 # 搜索商品数量加1 为了跟入参xptz_success_num要求比对
                print(f"buff商品 id：{item['id']} 已入库")
                query = "INSERT INTO sys_buff_goods (goods_info, deal_flag,create_time) VALUES (%s, %s,now())"
                data = (json.dumps(item), '0')
                cursor.execute(query, data)
            conn.commit()
        except Exception as e:
            print('datalist入库--失败',e)
        finally:
            if conn.is_connected():
                cursor.close()
            conn.close()

if __name__ == '__main__':
    print('开启选品脚本')
    xp = xpjb()
    # 处理命令行参数
    xp.dealArgc()
    # 获取数据库中BUFF品类配置
    xp.getXpCategory()
    # 获取数据库中BUFF品质配置
    xp.getXpRarity()
    # 获取数据库中BUFF类别配置
    xp.getXpQuality()
    # 获取数据库中BUFF外观配置
    xp.getXpExterior()
    # 获取数据库中BUFF收藏品信息
    xp.getXpItemset()
    # 根据数据库中配置信息，组合商品选取API的入参列表
    xp.getSearchGoodParam()
    # 生成最终选品API的入参列表
    xp.generate_combinations()
    # 获取选品账号信息，根据脚本的入参xp_phone检索
    xp.getXpzh()
    # 选品
    xp.SearchGoods()