spider_work/urbox_spider/urbox_spider.py

# -*- coding: utf-8 -*-
# Author : Charley
# Python : 3.10.8
# Date   : 2025/8/4 16:08
import random
import time
import requests
import inspect
import schedule
from loguru import logger
from parsel import Selector
from tenacity import retry, stop_after_attempt, wait_fixed
from mysql_pool import MySQLConnectionPool
from urbox_id_spider import urbox_id_main

category_dict = {
    "BASEBALL": 9,
    "SOCCER": 10,
    "BASKETBALL": 11,
    "FOOTBALL": 12
}

logger.remove()
logger.add("./logs/{time:YYYYMMDD}.log", encoding='utf-8', rotation="00:00",
           format="[{time:YYYY-MM-DD HH:mm:ss.SSS}] {level} {message}",
           level="DEBUG", retention="7 day")


def after_log(retry_state):
    """
    retry 回调
    :param retry_state: RetryCallState 对象
    """
    # 检查 args 是否存在且不为空
    if retry_state.args and len(retry_state.args) > 0:
        log = retry_state.args[0]  # 获取传入的 logger
    else:
        log = logger  # 使用全局 logger

    if retry_state.outcome.failed:
        log.warning(
            f"Function '{retry_state.fn.__name__}', Attempt {retry_state.attempt_number} Times")
    else:
        log.info(f"Function '{retry_state.fn.__name__}', Attempt {retry_state.attempt_number} succeeded")


@retry(stop=stop_after_attempt(5), wait=wait_fixed(1), after=after_log)
def get_proxys(log):
    """
    获取代理
    :return: 代理
    """
    tunnel = "x371.kdltps.com:15818"
    kdl_username = "t13753103189895"
    kdl_password = "o0yefv6z"
    try:
        proxies = {
            "http": "http://%(user)s:%(pwd)s@%(proxy)s/" % {"user": kdl_username, "pwd": kdl_password, "proxy": tunnel},
            "https": "http://%(user)s:%(pwd)s@%(proxy)s/" % {"user": kdl_username, "pwd": kdl_password, "proxy": tunnel}
        }
        return proxies
    except Exception as e:
        log.error(f"Error getting proxy: {e}")
        raise e


def get_product_list(log, sql_pool, category_name, category_status):
    page = 1
    while page <= 500:
        try:
            log.debug(
                f'--------------- {inspect.currentframe().f_code.co_name}, page {page}, category_name {category_name} start ---------------')
            len_items = get_product_single_page(log, sql_pool, page, category_name, category_status)
        except Exception as e:
            log.error(
                f"{inspect.currentframe().f_code.co_name} Request get_product_single_page for page:{page}, {e}")
            len_items = 0

        if len_items < 20:
            log.debug(f'--------------- page {page} has {len_items} items, break ---------------')
            break

        page += 1

        # 设置等待时间 避免查询太频繁
        time.sleep(random.uniform(0.5, 1))


@retry(stop=stop_after_attempt(5), wait=wait_fixed(1), after=after_log)
def get_product_single_page(log, sql_pool, page, category_name, category_status):
    headers = {
        "Accept": "application/json, text/plain, */*",
        "Accept-Language": "en,zh-CN;q=0.9,zh;q=0.8",
        # "Authorization": "Bearer 4001|rZwBadHCeDlJTRK52IFVpvcuay2hQjYFDLMO72xo",
        "Connection": "keep-alive",
        "Content-Type": "application/json",
        "Referer": "https://www.urboxwin.com/",
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/137.0.0.0 Safari/537.36",
    }
    url = "https://jp.urboxwin.com/webapi/breaking/index"
    data = {
        # "page": 1,
        "page": page,
        "per_page": 20,
        "status": category_status
        # "status": 9
    }
    response = requests.post(url, headers=headers, json=data)
    # print(response.text)
    response.raise_for_status()

    resp_json = response.json()
    json_list = resp_json.get("data", {}).get("data", [])
    if not json_list:
        log.debug("没有数据")
        return 0

    info_list = []
    for item in json_list:
        goods_id = item.get("goods_id")
        sale_status = item.get("sale_status")
        data_tuple = (goods_id, sale_status, category_name)
        # print(data_tuple)
        info_list.append(data_tuple)

    # 批量插入数据库 如果存在则更新
    sql = """
    INSERT INTO urbox_product_record (goods_id, sale_status, category_name) VALUES (%s, %s, %s)
    ON DUPLICATE KEY UPDATE sale_status = VALUES(sale_status)
    """
    # if info_list:
    #     sql_pool.insert_many(query=sql, args_list=info_list)

    # 在插入数据库之前进行去重
    seen = set()
    unique_info_list = []
    for item in info_list:
        goods_id = item[0]
        if goods_id not in seen:
            seen.add(goods_id)
            unique_info_list.append(item)

    if unique_info_list:
        sql_pool.insert_many(query=sql, args_list=unique_info_list)

    return len(json_list)


@retry(stop=stop_after_attempt(5), wait=wait_fixed(1), after=after_log)
def get_product_detail(log, product_id, sql_pool):
    log.debug(f'--------------- {inspect.currentframe().f_code.co_name}, product_id {product_id} start ---------------')
    headers = {
        "Accept": "application/json, text/plain, */*",
        "Accept-Language": "en,zh-CN;q=0.9,zh;q=0.8",
        # "Authorization": "Bearer 4001|rZwBadHCeDlJTRK52IFVpvcuay2hQjYFDLMO72xo",
        "Connection": "keep-alive",
        "Content-Type": "application/json",
        "Referer": "https://www.urboxwin.com/",
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/137.0.0.0 Safari/537.36",
    }
    url = "https://jp.urboxwin.com/webapi/breaking/detail"
    data = {
        # "id": "353"
        "id": f"{product_id}"
    }
    response = requests.post(url, headers=headers, json=data)
    # print(response.text)
    response.raise_for_status()
    resp_json = response.json()
    json_data = resp_json.get("data", {})
    card_number = json_data.get("number")
    tags = json_data.get("tags", [])
    tags_name_list = [tag.get("name") for tag in tags] if tags else []
    tags_name = " ".join(tags_name_list) if tags_name_list else ""

    title = json_data.get("title")
    score = json_data.get("score")
    available_at = json_data.get("available_at")
    end_at = json_data.get("end_at")
    buy_count = json_data.get("buy_count")  # str
    total_stock = json_data.get("total_stock")  # 总库存量
    available_stock = json_data.get("available_stock")  # 可用库存量

    # 使用 parsel 去除 HTML 标签
    try:
        detail = json_data.get("detail")
        if not detail:
            clean_detail = ""
        else:
            # 确保是字符串类型
            if isinstance(detail, bytes):
                detail = detail.decode('utf-8')

            selector = Selector(text=detail)
            text_nodes = selector.xpath('//text()').getall()
            clean_detail = ''.join(text_nodes).strip()
    except UnicodeDecodeError as e:
        log.error(f"商品 {product_id} 的detail字段编码错误: {e}")
        clean_detail = ""
    except Exception as e:
        log.error(f"商品 {product_id} 的detail字段解析失败: {e}")
        clean_detail = ""

    cover_image = json_data.get("cover_image", {}).get('thumb_o_file')  # 列表页图片
    top_images = json_data.get("top_images", [])
    top_images = [item.get('thumb_o_file') for item in top_images]
    top_images = '|'.join(top_images)  # 多图

    detail_images = json_data.get("detail_images", [])
    detail_images = [item.get('thumb_o_file') for item in detail_images]
    detail_images = '|'.join(detail_images)  # 详情信息图片

    current_at = json_data.get("current_at")
    live_time = json_data.get("live_time")
    stock_percentage = json_data.get("stock_percentage")  # 库存百分比
    live_link = json_data.get("live_link")
    buy_mode = json_data.get("buy_mode")  # 购买模式

    data_dict = {
        "goods_id": product_id,
        "card_number": card_number,
        "tags_name": tags_name,
        "title": title,
        "score": score,
        "available_at": available_at,
        "end_at": end_at,
        "buy_count": buy_count,
        "total_stock": total_stock,
        "available_stock": available_stock,
        "detail_str": clean_detail,
        "cover_image": cover_image,
        "top_images": top_images,
        "detail_images": detail_images,
        "current_at": current_at,
        "live_time": live_time,
        "stock_percentage": stock_percentage,
        "live_link": live_link,
        "buy_mode": buy_mode
    }
    # print(data_dict)

    # 根据商品id更新数据库
    sql_pool.update_one_or_dict(table="urbox_product_record", data=data_dict, condition={"goods_id": product_id})


def get_player_list(log, goods_id, sql_pool):
    page = 1
    while page <= 500:
        try:
            log.debug(
                f'--------------- {inspect.currentframe().f_code.co_name}, page {page}, goods_id {goods_id} start ---------------')
            len_items = get_player_single_page(log, goods_id, sql_pool, page)
        except Exception as e:
            log.error(
                f"{inspect.currentframe().f_code.co_name} Request get_player_single_page for page:{page}, {e}")
            len_items = 0

        if len_items < 20:
            log.debug(f'--------------- page {page} has {len_items} items, break ---------------')
            break

        page += 1

        # 设置等待时间 避免查询太频繁
        time.sleep(random.uniform(0.5, 1))

    # 更新商品状态
    sql_pool.update_one_or_dict(table="urbox_product_record", data={"state": 1}, condition={"goods_id": goods_id})


def get_player_single_page(log, goods_id, sql_pool, page):
    headers = {
        "Accept": "application/json, text/plain, */*",
        "Accept-Language": "en,zh-CN;q=0.9,zh;q=0.8",
        "Authorization": "Bearer 4001|rZwBadHCeDlJTRK52IFVpvcuay2hQjYFDLMO72xo",
        "Connection": "keep-alive",
        "Content-Type": "application/json",
        "Referer": "https://www.urboxwin.com/",
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/137.0.0.0 Safari/537.36",
    }
    url = "https://jp.urboxwin.com/webapi/breaking/report"
    data = {
        # "page": 1,
        "page": page,
        "per_page": 20,
        "id": f"{goods_id}"
        # "id": "353"
    }
    response = requests.post(url, headers=headers, json=data)
    # print(response.text)
    response.raise_for_status()
    resp_json = response.json()
    json_list = resp_json.get("data", {}).get("data", [])
    if not json_list:
        log.debug("没有数据")
        return 0

    info_list = []
    for item in json_list:
        data_id = item.get("id")
        user_name = item.get("user", {}).get("name")
        title = item.get("title")
        is_win_prize = item.get("is_win_prize")  # 是否中奖
        win_prize_text = item.get("win_prize_text")  # 中奖信息
        draw_finish_at = item.get("draw_finish_at")  # 开奖时间
        prize_image = item.get("prize_image", {})  # 奖品图片
        prize_image = prize_image.get('thumb_o_file') if prize_image else ""

        data_dict = {
            "goods_id": goods_id,
            "data_id": data_id,
            "user_name": user_name,
            "title": title,
            "is_win_prize": is_win_prize,
            "win_prize_text": win_prize_text,
            "draw_finish_at": draw_finish_at,
            "prize_image": prize_image
        }
        # print(data_dict)
        info_list.append(data_dict)

    if info_list:
        try:
            sql_pool.insert_many(table="urbox_player_record", data_list=info_list, ignore=True)
        except Exception as e:
            log.error(f"商品 {goods_id} 的player_record数据插入失败: {e}")

    return len(json_list)


@retry(stop=stop_after_attempt(100), wait=wait_fixed(3600), after=after_log)
def urbox_main(log):
    """
    主函数
    :param log: logger对象
    """
    log.info(
        f'开始运行 {inspect.currentframe().f_code.co_name} 爬虫任务.................................................')

    # 配置 MySQL 连接池
    sql_pool = MySQLConnectionPool(log=log)
    if not sql_pool.check_pool_health():
        log.error("数据库连接池异常")
        raise RuntimeError("数据库连接池异常")

    try:
        log.debug(f'Request start product list.........')
        for category_name, category_status in category_dict.items():
            log.debug(f'Request start for Category: {category_name}, Category Status: {category_status}.........')
            try:
                get_product_list(log, sql_pool, category_name, category_status)
            except Exception as e:
                log.error(
                    f"{inspect.currentframe().f_code.co_name} Request get_product_list for Category:{category_name} error: {e}")
        log.success(f'Request product list end.................................................')

        # 查询商品详情数据
        log.debug(f'Request start for product detail.........')
        sql_goods_id_list = sql_pool.select_all(
            query="SELECT goods_id FROM urbox_product_record WHERE card_number IS NULL")
        sql_goods_id_list = [item[0] for item in sql_goods_id_list]
        for goods_id in sql_goods_id_list:
            try:
                get_product_detail(log, goods_id, sql_pool)
            except Exception as e:
                log.error(
                    f"{inspect.currentframe().f_code.co_name} Request get_product_detail for goods_id:{goods_id} error: {e}")
            # 设置等待时间 避免查询太频繁
            time.sleep(random.uniform(0.5, 1))
        log.success(f'Request product detail end.................................................')

        # 查询 player 数据, 先查询 urbox_product_record 表中 状态为0的
        log.debug(f'Request start for player list.........')
        sql_goods_id_list_player = sql_pool.select_all(
            query="SELECT goods_id FROM urbox_product_record WHERE state = 0")
        sql_goods_id_list_player = [item[0] for item in sql_goods_id_list_player]
        for goods_id in sql_goods_id_list_player:
            try:
                get_player_list(log, goods_id, sql_pool)
            except Exception as e:
                log.error(
                    f"{inspect.currentframe().f_code.co_name} Request get_player_list for goods_id:{goods_id} error: {e}")

        # 20251127新增
        try:
            log.debug(f'============================ Request start for urbox_id list ============================')
            urbox_id_main(log=log)
        except Exception as e:
            log.error(
                f"{inspect.currentframe().f_code.co_name} Request urbox_id_main error: {e}")
        log.success(f'Request player list end.................................................')
    except Exception as e:
        log.error(f'{inspect.currentframe().f_code.co_name} error: {e}')
    finally:
        log.info(f'爬虫程序 {inspect.currentframe().f_code.co_name} 运行结束,等待下一轮的采集任务............')


def schedule_task():
    """
    爬虫模块 定时任务 的启动文件
    """
    # 立即运行一次任务
    # urbox_main(log=logger)

    # 设置定时任务
    schedule.every().day.at("00:01").do(urbox_main, log=logger)

    while True:
        schedule.run_pending()
        time.sleep(1)


if __name__ == '__main__':
    # get_product_list()
    # get_product_detail(logger,354, None)
    # get_player_list(logger, 350, None)
    schedule_task()