spider_work/courtyard_spider/replenish_price.py

# -*- coding: utf-8 -*-
# Author : Charley
# Python : 3.10.8
# Date   : 2025/12/11 13:26
import requests
import user_agent
from loguru import logger
from mysql_pool import MySQLConnectionPool
from tenacity import retry, stop_after_attempt, wait_fixed

headers = {
    "accept": "application/json",
    "referer": "https://courtyard.io/",
    "user-agent": user_agent.generate_user_agent()
}


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 replenish_price(log, bag_id, sql_pool=None):
    """
    补全价格
    :param log:
    :param bag_id:
    :param sql_pool:
    :return:
    """
    log.info(f"========================== 获取商品详情, bag_id: {bag_id} ==========================")
    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:
        # print(item)
        bag_id = item.get("id")
        price = item.get("saleDetails", {}).get("salePriceUsd")
        log.debug(f"bag_id: {bag_id}, price: {price}")

        # 根据bag_id更新price  如果在数据库中存在则更新，否则插入
        sql_pool.update_one("UPDATE courtyard_list_record SET price = %s WHERE bag_id = %s", (price, bag_id))


if __name__ == '__main__':
    sql_pool = MySQLConnectionPool(log=logger)
    replenish_price(logger, "pkmn-basic-pack",sql_pool)