spider_work/cxx_spider/super_vault_on_sale_spider.py

# -*- coding: utf-8 -*-
# Author : Charley
# Python : 3.10.8
# Date   : 2026/1/28 11:12
import inspect
import requests
from loguru import logger
from datetime import datetime
from mysql_pool import MySQLConnectionPool
from tenacity import retry, stop_after_attempt, wait_fixed

"""
SuperVault
"""
# 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")


HEADERS = {
    "User-Agent": "okhttp/4.9.0",
    # "Connection": "Keep-Alive",
    # "Accept-Encoding": "gzip",
    "Authorization": "",
    "CXX-APP-API-VERSION": "V2",  # 必须添加
    # "deviceType": "2",
    # "udid": "20f902c10f6163a19bf137d801731d9f",
    # "time": str(int(time.time() * 1000)),
    "Content-Type": "application/json; charset=UTF-8"
}


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_vod_single_page(log, page_num=1):
    """
    获取单页数据
    :param log: logger对象
    :param page_num: 页码
    :return: 数据
    """
    url = "https://cxx.cardsvault.net/app/teamup/list"
    data = {
        "pageSize": 20,
        "pageNum": page_num
    }
    response = requests.post(url, headers=HEADERS, json=data, timeout=22)
    response.raise_for_status()

    result = response.json()
    # print(result)
    if result.get("status") == 200:
        data = result.get("data", {})
        total = data.get("total", 0)
        current_page = data.get("pageNum", 1)
        items = data.get("data", [])

        log.info(f"当前查询的是 ->->-> 第 {current_page} 页，共 {total} 条记录")
        log.debug(f"当前页数据数量: {len(items)}")

        return {
            "total": total,
            "current_page": current_page,
            "items": items,
        }
    else:
        log.error(f"API 返回错误: {result.get('msg', '未知错误')}")
        return None


def parse_list_items(log, items):
    """
    解析列表项
    :param log: logger对象
    :param items: 列表项
    :return: 解析后的列表项
    """
    parsed_items = []
    log.debug(f"正在解析列表项.................")
    for item in items:
        pid = item.get("id")
        serial = item.get("serial")  # 编号
        title = item.get("title")
        type_name = item.get("typeName")  # 随机卡种
        isPre = item.get("isPre")
        count = item.get("count")

        totalPrice = item.get("totalPrice")
        totalPrice = totalPrice / 100 if totalPrice else 0
        signPrice = item.get("signPrice")
        signPrice = signPrice / 100 if signPrice else 0

        sellTime = item.get("sellTime")
        sellDays = item.get("sellDays")
        status = item.get("status")  # 9:完成 8:待发货
        statusName = item.get("statusName")
        description = item.get("description")
        createTime = item.get("createTime")
        cover_url = item.get("cover", {}).get("url")  # 封面图

        anchor_id = item.get("anchor", {}).get("id")
        anchor_userName = item.get("anchor", {}).get("userName")
        soldCount = item.get("soldCount")
        detailUrl = item.get("detailUrl")
        goodsUrl = item.get("goodsUrl")
        standardName = item.get("standardName")  # 规格

        crawl_date = datetime.now().strftime("%Y-%m-%d")

        parsed_item = {
            "pid": pid,
            "title": title,
            "serial": serial,
            "type_name": type_name,
            "is_pre": isPre,
            "count": count,
            "total_price": totalPrice,
            "sign_price": signPrice,
            "sell_time": sellTime,
            "sell_days": sellDays,
            "status": status,
            "status_name": statusName,
            "description": description,
            "create_time": createTime,
            "cover_url": cover_url,
            "anchor_id": anchor_id,
            "anchor_username": anchor_userName,
            "sold_count": soldCount,
            "detail_url": detailUrl,
            "goods_url": goodsUrl,
            "standard_name": standardName,
            "crawl_date": crawl_date
        }
        # print(parsed_item)
        parsed_items.append(parsed_item)
    return parsed_items


def get_vod_list(log, sql_pool):
    """
    获取列表数据
    :param log: logger对象
    :param sql_pool: 数据库连接池
    """
    page_num = 1
    total_pages = 9
    items_per_page = 20  # pageSize

    while page_num <= total_pages:
        log.debug(f"正在获取第 {page_num} 页的数据.................")
        page_result = get_vod_single_page(log, page_num)

        if not page_result:
            log.error(f"获取第 {page_num} 页失败 !!!")
            break

        # 第一次请求时更新真实的总页数
        if page_num == 1:
            total_count = page_result["total"]
            total_pages = (total_count + items_per_page - 1) // items_per_page
            log.info(f"总共 {total_pages} 页")

        # 每页获取后立即解析
        items = parse_list_items(log, page_result["items"])
        sql_pool.insert_many(table="super_vault_on_sale_record", data_list=items, ignore=True)

        page_num += 1


@retry(stop=stop_after_attempt(100), wait=wait_fixed(3600), after=after_log)
def cxx_sale_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:
        # 获取所有 pid
        try:
            get_vod_list(log, sql_pool)
        except Exception as e:
            log.error(f"Error fetching last_product_id: {e}")

    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} 运行结束,等待下一轮的采集任务............')


if __name__ == '__main__':
    # get_vod_list(logger, None)
    # get_vod_single_page(logger, 1)
    cxx_sale_main(logger)
    # schedule_task()