spider_work/courtyard_spider/courtyard_spider.py

# -*- coding: utf-8 -*-
# Author : Charley
# Python : 3.10.8
# Date   : 2025/12/2 14:08
import threading
import time
import inspect
import requests
import schedule
import user_agent
from loguru import logger
from parsel import Selector
from datetime import datetime
from mysql_pool import MySQLConnectionPool
from DrissionPage import ChromiumPage, ChromiumOptions
from tenacity import retry, stop_after_attempt, wait_fixed

"""
扣驾的
"""

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 = {
    "accept": "application/json",
    "referer": "https://courtyard.io/",
    "user-agent": user_agent.generate_user_agent()
}



# 全局变量标识首次运行是否完成
detail_first_run_completed = False

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_goods_list(log, sql_pool):
    """
    获取商品列表
    :param log: logger对象
    :param sql_pool: MySQL连接池对象
    :return:
    """
    log.info(f"========================== 开始获取商品列表 ==========================")
    url = "https://api.courtyard.io/vending-machines"
    response = requests.get(url, headers=headers, timeout=22)
    # print(response.text)
    response.raise_for_status()

    vendingMachines = response.json().get("vendingMachines", [])

    for item in vendingMachines:
        bag_id = item.get("id")
        bag_title = item.get("title")
        # sealed_pack_animation = item.get("sealedPackAnimation")
        # sealed_pack_image = item.get("sealedPackImage")
        category_title = item.get("category", {}).get("title")
        price = item.get("saleDetails", {}).get("salePriceUsd")

        data_dict = {
            "bag_id": bag_id,
            "bag_title": bag_title,
            "category": category_title,
            "price":price
        }
        # log.info(f'get_goods_list: {data_dict}')

        try:
            get_goods_detail(log, data_dict, sql_pool)
        except Exception as e:
            log.error(f"Error processing item: {e}")

    # 保存数据
    # if info_list:
    #     log.info(f"获取商品列表成功, 共 {len(info_list)} 条数据")
    #     sql_pool.insert_many(table="courtyard_vending_machines_record", data_list=info_list, ignore= True)


@retry(stop=stop_after_attempt(5), wait=wait_fixed(1), after=after_log)
def get_goods_detail(log, query_dict: dict, sql_pool=None):
    """
    获取商品详情
    :param log: logger对象
    :param query_dict: query_dict
    :param sql_pool: MySQL连接池对象
    :return:
    """
    log.info(f"========================== 获取商品详情 ==========================")
    url = "https://api.courtyard.io/index/query/recent-pulls"
    params = {
        "limit": "250",
        # "vendingMachineIds": "pkmn-basic-pack"
        "vendingMachineIds": query_dict["bag_id"]
    }
    response = requests.get(url, headers=headers, params=params, timeout=22)
    # print(response.text)
    response.raise_for_status()

    pulls = response.json().get("assets", [])
    info_list = []
    for item in pulls:
        detail_title = item.get("title")
        detail_id = item.get("proof_of_integrity")

        if not detail_id:
            log.error(f"信息异常, detail_id: {detail_id}")
            continue

        asset_pictures = item.get("asset_pictures", [])
        img_front = asset_pictures[0] if len(asset_pictures) > 0 else None
        img_back = asset_pictures[1] if len(asset_pictures) > 1 else None

        # crawl_date = time.strftime("%Y-%m-%d", time.localtime())

        data_dict = {
            "bag_id": query_dict["bag_id"],
            "bag_title": query_dict["bag_title"],
            "category": query_dict["category"],
            "price": query_dict["price"],
            "detail_id": detail_id,
            "detail_title": detail_title,
            "img_front": img_front,
            "img_back": img_back,
            # "crawler_date": crawl_date
        }
        # log.info(f'data_dict:{data_dict}')
        info_list.append(data_dict)

    # 保存数据
    if info_list:
        log.info(f"获取商品详情成功, 共 {len(info_list)} 条数据")
        sql_pool.insert_many(table="courtyard_list_record", data_list=info_list, ignore=True)


def convert_time_format(time_str):
    """
    将时间字符串转换为标准格式
    :param time_str: 原始时间字符串，如 "December 3, 2025 at 4:29 PM"
    :return: 标准时间格式字符串，如 "2025-12-03 16:29:00"
    """
    if not time_str:
        return None

    try:
        dt_obj = datetime.strptime(time_str, "%B %d, %Y at %I:%M %p")
        return dt_obj.strftime("%Y-%m-%d %H:%M:%S")
    except ValueError as e:
        logger.warning(f"时间转换失败: {time_str}, 错误: {e}")
        return None


@retry(stop=stop_after_attempt(5), wait=wait_fixed(1), after=after_log)
def get_sale_detail_single_page(log, page, sql_id, detail_id, sql_pool=None):
    """
    获取商品详情
    :param log: logger对象
    :param page: page对象
    :param sql_id: 数据库id
    :param detail_id: 商品详情id
    :param sql_pool: MySQL连接池对象
    :return:
    """
    log.info(f"========================== 获取商品 <sale> 详情, sql_id: {sql_id} ==========================")
    # page_url = "https://courtyard.io/asset/a4f0bbebd858370567f1779fddf0f55630810116d80965e33940fc8ff5ac94b4"
    page_url = f"https://courtyard.io/asset/{detail_id}"
    page.get(page_url)
    page.wait.load_start()
    log.debug(f'{inspect.currentframe().f_code.co_name} -> 页面加载成功, url: {page_url}')

    html = page.html
    if not html:
        log.error(f'{inspect.currentframe().f_code.co_name} -> 页面加载失败...........')
        raise Exception('页面加载失败, 重新加载........')  # 抛出异常以便重试

    selector = Selector(text=html)

    # 方法一：通过文本内容匹配（优先）
    correlation_spans = selector.xpath('//span[contains(text(), ":")]/text()')
    correlation_id = None

    for text_selector in correlation_spans:
        correlation_id = text_selector.get()  # ✅ 获取字符串

        # match = re.search(r'[\w\s]+:\s*(\d+)', text)
        # if match:
        #     correlation_id = match.group(1)
        #     break  # 获取第一个有效 ID
    # correlation_spans = selector.xpath('//span[contains(text(), ":")]')
    # correlation_text = None
    #
    # for span in correlation_spans:
    #     text = span.get()
    #     if text and ":" in text:
    #         correlation_text = text
    #         break

    # 如果方法一失败，使用方法二：通过结构定位（备用）
    if not correlation_id:
        correlation_span = selector.xpath('//a[contains(@href, "cgccards.com")]/preceding-sibling::span[1]/text()')
        correlation_id = correlation_span.get()

    # 初始化所有可能的字段为None
    data_dict = {"detail_id": detail_id, "correlation_id": correlation_id, "burn_from": None,
                 "burn_to": None, "burn_time": None, "sale_price": None, "sale_from": None, "sale_to": None,
                 "sale_time": None, "mint_price": None, "mint_from": None, "mint_to": None, "mint_time": None}

    # 提取各个标签的数据
    # tag_div_list = selector.xpath('//div[@class="MuiBox-root css-aylq9e"]/div')# class="MuiBox-root css-1b09fes"
    # 获取 "Activity history" 后面的 div
    activity_div = selector.xpath('//h6[text()="Activity history"]/following-sibling::div[1]/div')
    for tag_div in activity_div:
        tag_name = tag_div.xpath('./div[1]/div/span/text()').get()
        if not tag_name:
            continue

        if tag_name == "Burn":
            data_dict["burn_from"] = tag_div.xpath('./div[2]/div[1]//h6/text()').get()
            data_dict["burn_to"] = tag_div.xpath('./div[2]/div[2]//h6/text()').get()
            data_dict["burn_time"] = tag_div.xpath('./div[2]/div[3]/span/@aria-label').get()
            # December 3, 2025 at 4:29 PM 转换时间格式
            data_dict["burn_time"] = convert_time_format(data_dict["burn_time"])
        elif tag_name == "Sale":
            sale_price = tag_div.xpath('./div[2]/span/text()').get()
            if sale_price:
                sale_price = sale_price.replace("$", "").replace(",", "")
            data_dict["sale_price"] = sale_price

            data_dict["sale_from"] = tag_div.xpath('./div[3]/div[1]//h6/text()').get()
            data_dict["sale_to"] = tag_div.xpath('./div[3]/div[2]//h6/text()').get()
            data_dict["sale_time"] = tag_div.xpath('./div[3]/div[3]/span/@aria-label').get()
            # December 3, 2025 at 4:29 PM 转换时间格式
            data_dict["sale_time"] = convert_time_format(data_dict["sale_time"])
        elif tag_name == "Mint":
            mint_price = tag_div.xpath('./div[2]/span/text()').get()
            if mint_price:
                mint_price = mint_price.replace("$", "").replace(",", "")
            data_dict["mint_price"] = mint_price

            data_dict["mint_from"] = tag_div.xpath('./div[3]/div[1]//h6/text()').get()
            data_dict["mint_to"] = tag_div.xpath('./div[3]/div[2]//h6/text()').get()
            data_dict["mint_time"] = tag_div.xpath('./div[3]/div[3]/span/@aria-label').get()
            # December 3, 2025 at 4:29 PM 转换时间格式
            data_dict["mint_time"] = convert_time_format(data_dict["mint_time"])

    # log.info(f'Sale detail data: {data_dict}')

    # 保存数据
    sql_pool.insert_one_or_dict(table="courtyard_detail_record", data=data_dict, ignore=True)


@retry(stop=stop_after_attempt(5), wait=wait_fixed(1), after=after_log)
def get_sale_detail_list(log, sql_pool=None):
    """
    获取商品详情
    :param log: logger对象
    # :param detail_id_list: 详情id列表
    :param sql_pool: MySQL连接池对象
    :return:
    """
    log.info(f"========================== 获取商品 <sale> 详情  LIST ==========================")
    options = ChromiumOptions()
    options.set_paths(local_port=9138, user_data_path=r'D:\Drissionpage_temp\courtyard_port_9138')
    # options.set_proxy("http://" + tunnel)
    # options.auto_port(True)

    options.no_imgs(True)
    # 最大化
    options.set_argument("--start-maximized")
    options.set_argument("--disable-gpu")
    options.set_argument("-accept-lang=en-US")
    page = ChromiumPage(options)
    try:
        sql_detail_id_list = sql_pool.select_all("SELECT id, detail_id FROM courtyard_list_record WHERE state = 0")

        for sql_detail_id in sql_detail_id_list:
            sql_id = sql_detail_id[0]
            detail_id = sql_detail_id[1]
            try:
                get_sale_detail_single_page(log, page, sql_id, detail_id, sql_pool)
                sql_pool.update_one("UPDATE courtyard_list_record SET state = 1 WHERE id = %s", (sql_id,))
            except Exception as e:
                log.error(f'get_sale_detail_single_page error: {e}')
                sql_pool.update_one("UPDATE courtyard_list_record SET state = 2 WHERE id = %s", (sql_id,))

    except Exception as e:
        log.error(f'get_response error: {e}')
        raise 'get_response error'
    finally:
        page.quit()


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

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

    try:
        try:
            log.debug('------------------- 开始获取商品列表 -------------------')
            get_goods_list(log, sql_pool)
        except Exception as e:
            log.error(f'get_goods_list error: {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} 运行结束,等待下一轮的采集任务............')

    end = time.time()
    elapsed_time = end - start
    log.info(f'============================== 本次爬虫运行时间:{elapsed_time:.2f} 秒 ===============================')

    return elapsed_time


@retry(stop=stop_after_attempt(100), wait=wait_fixed(3600), after=after_log)
def detail_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("数据库连接池异常")

    global detail_first_run_completed

    try:
        # 获取详情页信息
        try:
            log.debug('------------------- 获取商品 detail 数据 -------------------')
            get_sale_detail_list(log, sql_pool)
        except Exception as e:
            log.error(f'get_sale_detail_list error: {e}')

    except Exception as e:
        log.error(f'{inspect.currentframe().f_code.co_name} error: {e}')
    finally:
        detail_first_run_completed = True
        log.info(f'爬虫程序 {inspect.currentframe().f_code.co_name} 运行结束,等待下一轮的采集任务............')


def control_list_mask(log):
    """
    控制列表爬虫任务  每10分钟运行
    :param log: logger对象
    """
    while True:
        log.info(
            f'--------------------- 开始运行 {inspect.currentframe().f_code.co_name} 新一轮的爬虫任务 ---------------------')
        elapsed_time = list_main(log)
        # 计算剩余时间
        wait_time = max(0, 300 - int(elapsed_time))
        if wait_time > 0:
            log.info(f"程序运行时间{elapsed_time:.2f}秒, 小于 5 分钟，等待 {wait_time:.2f} 秒后再开始下一轮任务")
            time.sleep(wait_time)
        else:
            log.info("程序运行时间大于等于5分钟，直接开始下一轮任务")


def scheduled_detail_main(log):
    """定时任务调用的包装函数"""
    global detail_first_run_completed
    if detail_first_run_completed:
        detail_main(log)
    else:
        log.info("Skipping scheduled task as first run is not completed yet")


def run_threaded(job_func, *args, **kwargs):
    """
    在新线程中运行给定的函数，并传递参数。

    :param job_func: 要运行的目标函数
    :param args: 位置参数
    :param kwargs: 关键字参数
    """
    job_thread = threading.Thread(target=job_func, args=args, kwargs=kwargs)
    job_thread.start()



def schedule_task():
    """
    设置定时任务
    """
    # 启动 control_list_mask 任务线程
    list_thread = threading.Thread(target=control_list_mask, args=(logger,))
    list_thread.daemon = True  # 设置为守护线程，主程序退出时自动结束
    list_thread.start()

    # 启动 detail_main 任务线程（首次运行）
    detail_thread = threading.Thread(target=detail_main, args=(logger,))
    detail_thread.daemon = True
    detail_thread.start()

    # 设置定时任务  每天
    # schedule.every().day.at("00:01").do(run_threaded, detail_main, logger)
    schedule.every().day.at("00:01").do(run_threaded, scheduled_detail_main, logger)

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


if __name__ == '__main__':
    schedule_task()
    # detail_main(log=logger)
    # get_sale_detail_list(log, ((1, 'a4f0bbebd858370567f1779fddf0f55630810116d80965e33940fc8ff5ac94b4'),
    #                            (2, 'a4f0bbebd858370567f1779fddf0f55630810116d80965e33940fc8ff5ac94b4')), sql_pool)