# BUFF选品02脚本 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 # BUFF验证通过的商品数量 self.dealedGoodsNum = 0 #BUFF搜寻商品入参 self.search_good_param_init = { 'game': 'csgo' } #待处理的buff商品信息list self.undeal_goods_list=[] # buff选品接口配置 self.goodsurl = 'https://buff.163.com/api/market/goods' self.goodinfourl='https://buff.163.com/api/market/goods/info?goods_id=' 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') # 添加一个位置参数，表示选品脚本停止条件标志的成功数量， 默认 200 parser.add_argument('--xptz_success_num', type=int ,default=200, help='选品脚本停止条件标志的成功数量， 默认 200') # 解析命令行参数 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'选品脚本停止条件标志的成功数量 默认 200: {self.args.xptz_success_num}') print('处理命令行参数--成功') 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() # 处理选品 def dealGoods(self): print('开始处理选品') for i in range(self.args.runcount): self.simpledealGoods() time.sleep(self.args.runtime) print('结束处理选品') # 单次选品处理 def simpledealGoods(self): if self.dealedGoodsNum >= self.args.xptz_success_num: print('搜寻的BUFF商品已完成，停止脚本') exit() # 请求数据库获取待处理的BUFF商品 self.getUndealBuffGoods() for i in range(len(self.undeal_goods_list)): self.dealsingleGoods(self.undeal_goods_list[i]) # 处理单个BUFF商品，做判断是否可以抢购 def dealsingleGoods(self,item): goodItem = json.loads(item) print('开始处理选品id',goodItem['id']) mslist=[] mslist=self.getGoodsMSinfo(goodItem['id']) goodsinfo={ } time.sleep(self.args.timeinterval) #获取buff商品的磨损信息 def getGoodsMSinfo(self,goods_id): try: mslist=[] res = req.get(url=self.goodinfourl+str(goods_id), headers=self.my_headers) # if res.json()['code'] != 'OK' : # print('BUFF商品信息请求返回报文:', res.text.strip()) # continue data = res.json()['data'] mslist.extend(data['paintwear_choices']) print('BUFF商品的磨损信息:', mslist) except Exception as e: print('BUFF商品信息请求--失败',e) finally: return mslist # 从数据库获取待处理的BUFF商品 def getUndealBuffGoods(self): print('从数据库获取待处理的BUFF商品') try: # 创建mysql连接 conn = mysql.connector.connect( host = self.host, user = self.user, password = self.password, database = self.database ) goods_id_list=[] self.undeal_goods_list=[] # 创建游标对象 cursor = conn.cursor() # 执行SQL查询 cursor.execute("SELECT goods_id,goods_info from sys_buff_goods where deal_flag ='0' limit 0 , 10 for update") # 获取所有记录列表 results = cursor.fetchall() if len(results) == 0: print('数据库中信息不存在') exit() for row in results: goods_id_list.append(row[0]) self.undeal_goods_list.append(row[1]) # 使用join()将列表转换为逗号分隔的字符串 placeholders = ', '.join(['%s'] * len(goods_id_list)) # 编写UPDATE语句 update_query = f"UPDATE sys_buff_goods SET deal_flag ='1' WHERE goods_id in ({placeholders})" # 执行UPDATE语句 cursor.execute(update_query, goods_id_list) conn.commit() # 关闭游标和连接 cursor.close() conn.close() except Exception as e: print('获取待处理的BUFF商品信息--失败',e) exit() if __name__ == '__main__': print('开启选品脚本') xp = xpjb() # 处理命令行参数 xp.dealArgc() # 获取选品账号信息，根据脚本的入参xp_phone检索 xp.getXpzh() # 选品 xp.dealGoods()