spider_work/clove_spider/clove_blind_box_spider.py

# -*- coding: utf-8 -*-
# Author : Charley
# Python : 3.10.8
# Date   : 2025/7/29 15:06
import json
import time
import requests
# from curl_cffi 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

# 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


@retry(stop=stop_after_attempt(5), wait=wait_fixed(1), after=after_log)
def get_blind_box_list(log):
    log.debug(f"{inspect.currentframe().f_code.co_name}---------------------- 开始获取盲盒列表 ----------------------")
    headers = {
        "accept": "application/graphql-response+json, application/graphql+json, application/json, text/event-stream, multipart/mixed",
        "content-type": "application/json",
        "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://api.prd.oripa.clove.jp/graphql"
    data = {
        "operationName": "orderedOripas",
        "query": "query orderedOripas($orderedOripasInput: OrderedOripasInput!) {\n  orderedOripas(orderedOripasInput: $orderedOripasInput) {\n    category\n    isR18\n    hasLastOne\n    id\n    isDaily\n    isAppraised\n    isUserLimited\n    requiredRankName\n    name\n    nameLogo\n    nameHidden\n    price\n    publishStatus\n    quantity\n    remaining\n    roundNumber\n    subImages\n    thumbnail\n    extraPrizeThreshold\n    video {\n      id\n      title\n      videoData\n      deleted\n      __typename\n    }\n    oripaSearchTargetPrizes {\n      searchTargetPrize {\n        titleJa\n        __typename\n      }\n      __typename\n    }\n    openAt\n    __typename\n  }\n}",
        "variables": {
            "orderedOripasInput": {
                "isR18": False,
                "sort": "LOW_REMAINING_RATE"  # 剩余数量
            }
        }
    }
    response = requests.post(url, headers=headers, json=data, timeout=60, proxies=get_proxys(log))
    # response = requests.post(url, headers=headers, json=data, timeout=60)
    # print(response.text)
    response.raise_for_status()
    return response.json()


def parse_list(log, resp_data, sql_pool):
    log.debug(f"{inspect.currentframe().f_code.co_name} start..........")
    items = resp_data["data"]["orderedOripas"]
    log.info("获取到%s个数据" % len(items))
    if not items:
        log.debug("没有数据")
        return

    info_list = []
    for item in items:
        # print(item)
        pid = item.get("id")
        category = item.get("category")
        title = item.get("name")
        price = item.get("price")
        publishStatus = item.get("publishStatus")
        quantity = item.get("quantity")  # 总数量
        remaining = item.get("remaining")  # 剩余数量

        # thumbnail有三种 {}, [], ''
        tag_image = item.get("thumbnail")
        if isinstance(tag_image, dict):
            # 处理字典类型的thumbnail，从中提取ja字段的值
            ja_images = tag_image.get("ja")
            if isinstance(ja_images, list) and len(ja_images) > 0:
                image = ja_images[0]  # 取第一个图片链接
            elif isinstance(ja_images, str):
                image = ja_images
            else:
                image = ''
        elif isinstance(tag_image, list):
            # 处理列表类型的thumbnail
            if len(tag_image) > 0:
                image = tag_image[0]
            else:
                image = ''
        elif isinstance(tag_image, str):
            # 处理字符串类型的thumbnail
            image = tag_image
        else:
            image = ''

        # subImages
        tag_subImages = item.get("subImages")
        if isinstance(tag_subImages, dict):
            # 处理字典类型的thumbnail，从中提取ja字段的值
            ja_subimages = tag_subImages.get("ja")
            if isinstance(ja_subimages, list) and len(ja_subimages) > 0:
                sub_image = ja_subimages[0]  # 取第一个图片链接
            elif isinstance(ja_subimages, str):
                sub_image = ja_subimages
            else:
                sub_image = ''
        elif isinstance(tag_subImages, list):
            # 处理列表类型的thumbnail
            if len(tag_subImages) > 0:
                sub_image = tag_subImages[0]
            else:
                sub_image = ''
        elif isinstance(tag_subImages, str):
            # 处理字符串类型的thumbnail
            sub_image = tag_subImages
        else:
            sub_image = ''

        open_at = item.get("openAt")

        data_dict = {
            "pid": pid,
            "category": category,
            "title": title,
            "price": price,
            "publish_status": publishStatus,
            "quantity": quantity,
            "remaining": remaining,
            "image": image,
            "sub_image": sub_image,
            "open_at": open_at
        }
        # print(data_dict)
        info_list.append(data_dict)

    # 插入数据库
    if info_list:
        try:
            sql_pool.insert_many(table="clove_blind_box_list_record", data_list=info_list, ignore=True)
            # sql = "INSERT IGNORE INTO clove_blind_box_list_record (pid, category, title, price, publish_status, quantity, remaining, image, sub_image, open_at) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)"
            # sql_pool.insert_all(sql, info_list)
        except Exception as e:
            log.warning(f"{inspect.currentframe().f_code.co_name}, {e[:500]}")


@retry(stop=stop_after_attempt(5), wait=wait_fixed(1), after=after_log)
def get_blind_box_detail(log, pid, sql_pool):
    headers = {
        "accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7",
        "referer": "https://oripa.clove.jp/oripa/All",
        "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://oripa.clove.jp/oripa/All/cmca1kd0i0001s601a4y07n2b"
    url = f"https://oripa.clove.jp/oripa/All/{pid}"
    # response = requests.get(url, headers=headers, cookies=cookies)
    response = requests.get(url, headers=headers, timeout=60, proxies=get_proxys(log))
    # print(response.text)

    selector = Selector(text=response.text)

    tag_shang = selector.xpath('//script[@id="__NEXT_DATA__"]/text()').get()

    if not tag_shang:
        log.debug(f"{inspect.currentframe().f_code.co_name} 没有 tag_shang 数据..............")
        # 更改 task 表状态为 2
        sql_pool.update_one_or_dict(table="clove_blind_box_task", data={"state": 2}, condition={"pid": pid})
        return

    json_data = json.loads(tag_shang)
    if json_data:
        displayedPrizes = json_data.get('props', {}).get('pageProps', {}).get('oripa', {}).get("displayedPrizes", [])
        log.debug(f"{inspect.currentframe().f_code.co_name} 获取到 {len(displayedPrizes)} 条数据..............")
        if not displayedPrizes:
            log.debug(f"{inspect.currentframe().f_code.co_name} 没有 displayedPrizes 数据..............")
            # 更改 task 表状态为 2
            sql_pool.update_one_or_dict(table="clove_blind_box_task", data={"state": 2}, condition={"pid": pid})
            return

        info_list = []
        for item in displayedPrizes:
            prize_id = item.get("id")
            prizeType = item.get("prizeType")
            quantity = item.get("quantity")
            mainDescription = item.get("mainDescription")
            mainDescriptionEn = item.get("mainDescriptionEn")
            subDescription = item.get("subDescription")
            kataban = item.get("kataban")
            imageUrl = item.get("imageUrl")
            prize_condition = item.get("condition")
            isReferencePriceTarget = item.get("isReferencePriceTarget")
            referencePrice = item.get("referencePrice")
            referencePriceUpdatedAt = item.get("referencePriceUpdatedAt")

            data_dict = {
                "pid": pid,
                "prize_id": prize_id,  # 奖品id
                "prize_type": prizeType,  # 奖品类型
                "quantity": quantity,  # 数量
                "main_description": mainDescription,  # 奖品描述
                "main_description_en": mainDescriptionEn,
                "sub_description": subDescription,  # 卡片等级
                "kataban": kataban,  # 卡标签
                "image_url": imageUrl,  # 图片地址
                "prize_condition": prize_condition,  # 奖品评分, PSA10
                "is_reference_price_target": isReferencePriceTarget,  # (疑似)是否开封状态 (1: 是, 0: 否)
                "reference_price": referencePrice,  # 参考价格
                "reference_price_updated_at": referencePriceUpdatedAt  # 参考价格更新时间
            }
            # print(data_dict)
            info_list.append(data_dict)

        if info_list:
            try:
                sql_pool.insert_many(table="clove_blind_box_detail_record", data_list=info_list, ignore=True)
                # 更新 task 表状态为 1
                sql_pool.update_one_or_dict(table="clove_blind_box_task", data={"state": 1}, condition={"pid": pid})
            except Exception as e:
                log.warning(f"{inspect.currentframe().f_code.co_name}, {e[:500]}")
    else:
        log.debug(f"{inspect.currentframe().f_code.co_name} 请求出错, 没有数据..............")


@retry(stop=stop_after_attempt(100), wait=wait_fixed(3600), after=after_log)
def blind_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:
        try:
            resp_data = get_blind_box_list(log)
            parse_list(log, resp_data, sql_pool)
        except Exception as e2:
            log.error(f"Request get_blind_box_list error: {e2}")

        # 先把新增的任务添加到任务表, 查询每日新增的 pid, 去重并插入 task 表
        sql_select_pids = sql_pool.select_all(
            # "SELECT DISTINCT pid FROM clove_blind_box_list_record WHERE DATE(gmt_create_time) = CURDATE() - INTERVAL 1 DAY")
            "SELECT DISTINCT pid FROM clove_blind_box_list_record WHERE DATE(gmt_create_time) = CURDATE()")
        sql_select_pids = [i[0] for i in sql_select_pids]
        sql_pool.insert_many(
            query="INSERT INTO clove_blind_box_task (pid) VALUES (%s) ON DUPLICATE KEY UPDATE pid=VALUES(pid)",
            args_list=sql_select_pids,
            ignore=True
        )

        time.sleep(5)

        # 获取详情
        sql_pids = sql_pool.select_all("SELECT pid FROM clove_blind_box_task WHERE state=0")
        sql_pids = [i[0] for i in sql_pids]
        for pid in sql_pids:
            try:
                log.debug(f"{inspect.currentframe().f_code.co_name} 获取 pid: {pid} 详情..............")
                get_blind_box_detail(log, pid, sql_pool)
            except Exception as e:
                log.error(f"get_blind_box_detail error: {e}")
                # 更改 task 表状态为 3
                sql_pool.update_one_or_dict(table="clove_blind_box_task", data={"state": 3}, condition={"pid": pid})

    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():
    """
    爬虫模块 定时任务 的启动文件
    """
    # 立即运行一次任务
    # blind_main(log=logger)

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

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


if __name__ == '__main__':
    # blind_main(logger)
    schedule_task()
    # get_blind_box_detail(logger, 'cmca1kd0i0001s601a4y07n2b')
    # test()