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

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