spider_work/bgs_spider/bgs_new_daily_spider.py

# -*- coding: utf-8 -*-
# Author  : Charley
# Python  : 3.8.10
# Date: 2024-11-12 16:40
import time
import requests
import schedule
import user_agent
from loguru import logger
import concurrent.futures
from tenacity import stop_after_attempt, wait_fixed, retry
from mysq_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="1 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):
    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 save_data(mysql_pool, info):
    """
    :param mysql_pool:
    :param info:
    :return:
    """
    sql = "INSERT INTO beckett_bgs_record(set_name, player_name, date_graded, centering_grade, corner_grade, edges_grade, surfaces_grade, auto_grade, final_grade, total_grade, cards_grade, number) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)"
    mysql_pool.insert_one(sql, info)


@retry(stop=stop_after_attempt(5), wait=wait_fixed(1), after=after_log)
def get_data(log, bgs_id, mysql_pool):
    """
    :param log:
    :param bgs_id:
    :param mysql_pool:
    :return:
    """
    headers = {
        "accept": "application/json, text/plain, */*",
        "user-agent": user_agent.generate_user_agent()
    }
    url = "https://www.beckett.com/api/grading/lookup"
    params = {
        "category": "BGS",
        "serialNumber": str(bgs_id)
    }
    response = requests.get(url, headers=headers, params=params, proxies=get_proxys(log), timeout=5)
    if response.status_code == 404:
        # 没有数据 No Record Found 将状态改为3
        log.warning(f"No Record Found for {bgs_id}")
        mysql_pool.update_one("UPDATE bgs_task SET state=3 WHERE auth_code=%s", (bgs_id,))
        return

    if response.status_code != 200:
        # 查询失败  将状态改为2
        log.warning(f"Error getting data for {bgs_id}, {response.status_code}")
        mysql_pool.update_one("UPDATE bgs_task SET state=2 WHERE auth_code=%s", (bgs_id,))
        return

    # print(response.json())
    result_dict = response.json()
    if result_dict:
        """
            "label": "silver",
            "non_bccg_card_total": 0,
            "item_id": "17932864",
            "set_name": "2024 Magic the Gathering Secret Lair Dungeons & Dragons 50th Anniversary Bonus Card Foil",
            "sport_name": "Magic",
            "card_key": "0879",
            "player_name": "Minsc & Boo, Timeless Heroes M",
            "date_graded": "Thursday, April 17, 2025",
            "center_grade": "9.5",
            "corners_grade": "8.5",
            "edges_grade": "8.5",
            "surfaces_grade": "10.0",
            "autograph_grade": "0.0",
            "final_grade": "8.5",
        """
        set_name = result_dict.get('set_name')
        player_name = result_dict.get('player_name')
        date_graded = result_dict.get('date_graded')
        centering_grade = result_dict.get('center_grade')
        corner_grade = result_dict.get('corners_grade')
        edges_grade = result_dict.get('edges_grade')
        surfaces_grade = result_dict.get('surface_grade')
        auto_grade = result_dict.get('autograph_grade')
        final_grade = result_dict.get('final_grade')
        total_grade = result_dict.get('pop_report')
        cards_grade = result_dict.get('pop_higher')
        info = (set_name, player_name, date_graded, centering_grade, corner_grade, edges_grade, surfaces_grade,
                auto_grade, final_grade, total_grade, cards_grade, int(bgs_id))

        # 检查所有值是否都为 None或空字符串, 不包含bgs_id
        all_none_or_empty = all(x is None or x == '' for x in info[:-1])
        if all_none_or_empty:
            log.debug("All values are empty")
        else:
            # print(info)
            save_data(mysql_pool, info)
            # 查询成功  将状态改为1
            mysql_pool.update_one("UPDATE bgs_task SET state=1 WHERE auth_code=%s", (bgs_id,))


def process_urls(log, ids, mysql_pool, batch_size=1000, max_workers=5):
    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
        for i in range(0, len(ids), batch_size):
            # print(i)
            batch = ids[i:i + batch_size]
            # print(batch)
            try:
                futures_to_urls = {executor.submit(get_data, log, url, mysql_pool): url for url in batch}
                for future in concurrent.futures.as_completed(futures_to_urls):
                    url = futures_to_urls[future]
                    try:
                        future.result()
                        log.debug(f"处理 {url} 成功")
                    except Exception as exc:
                        log.debug(f"处理 {url} 出错: {exc}")
            except Exception as e:
                log.error(f"提交任务失败: {e}")


@retry(stop=stop_after_attempt(100), wait=wait_fixed(3600), after=after_log)
def bgs_main(log):
    try:
        log.info(
            "开始运行 bgs_main 爬虫任务............................................................")
        sql_pool = MySQLConnectionPool(log=log)
        if not sql_pool:
            log.error("数据库连接失败")
            raise Exception("数据库连接失败")

        max_bgs_id = sql_pool.select_one("SELECT MAX(number) AS max_number FROM beckett_bgs_record")
        # print(max_bgs_id_list)
        max_bgs_id = max_bgs_id[0]
        log.info(f"max_bgs_id 从 {max_bgs_id} 开始爬取.........................")
        # bgs_id_list = [i for i in range(max_bgs_id, max_bgs_id + 3001)]


        # 20251029 往前2000  往后3000
        # while True:
        # 倒序查 20000个 20251205修改
        """
        倒序查 20000个 20251205修改
        """
        sql_bgs_id_list = sql_pool.select_all(
            "SELECT auth_code FROM bgs_task WHERE state!=1 AND auth_code BETWEEN %s AND %s LIMIT 20000",
            (max_bgs_id - 2000, max_bgs_id + 18000))

        # sql_bgs_id_list = sql_pool.select_all(
        #     "SELECT auth_code FROM bgs_task WHERE state!=1 AND auth_code > 18239485 LIMIT 5000")
            # "SELECT auth_code FROM bgs_task WHERE state!=1 AND auth_code > 17990000 LIMIT 5000"
        sql_bgs_id_list = [bid[0] for bid in sql_bgs_id_list]
        # for bid in sql_bgs_id_list:
        try:
            process_urls(log, sql_bgs_id_list, sql_pool, batch_size=1000,
                         max_workers=3)  # 根据需要调整batch_size和max_workers
            # get_data(bid, mysql_pool)
        except Exception as e:
            log.error('process urls: ', e)

        # time.sleep(5)

    except Exception as e:
        log.error(e)
    finally:
        log.info("爬虫程序运行结束,等待下一轮的采集任务.....................")


def schedule_task():
    """
    设置定时任务
    """
    # 立即运行一次任务
    bgs_main(logger)

    # 设置定时任务
    schedule.every().day.at("03:01").do(bgs_main, logger)
    while True:
        schedule.run_pending()
        time.sleep(1)


if __name__ == '__main__':
    schedule_task()
    # get_data('1000743')