# 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()