spider_work/psa_spider/psa_pop_detail_spider.py

# -*- coding: utf-8 -*-
# Author : Charley
# Python : 3.10.8
# Date   : 2025/9/15 14:53
import inspect
import random
import time
import schedule
import user_agent
from curl_cffi import requests
from loguru import logger
from parsel import Selector
from urllib.parse import urljoin
from mysql_pool import MySQLConnectionPool
from DrissionPage import ChromiumPage, ChromiumOptions
from tenacity import retry, stop_after_attempt, wait_fixed

logger.remove()
logger.add("logs/pop_player_{time:YYYYMMDD}.log", encoding='utf-8', rotation="00:00",
           format="[{time:YYYY-MM-DD HH:mm:ss.SSS}] {level} {message}",
           level="DEBUG", retention="3 day")

client_identifier_list = [
    "edge99", "edge101", "chrome99", "chrome100", "chrome101", "chrome104", "chrome107",
    "chrome110", "chrome116", "chrome119", "chrome120", "chrome123", "chrome124",
    "chrome99_android", "safari15_3", "safari15_5", "safari17_0", "safari17_2_ios"
]

BASE_URL = 'https://www.psacard.com'

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",
    'User-Agent': user_agent.generate_user_agent()
}

category_link_list = {
    'Baseball Cards': 'https://www.psacard.com/pop/baseball-cards/20003',
    'Baseball Coins': 'https://www.psacard.com/pop/baseball-coins/82797',
    'Basketball Cards': 'https://www.psacard.com/pop/basketball-cards/20019',
    'Basketball Coins': 'https://www.psacard.com/pop/basketball-coins/83007',
    'Boxing / Wrestling Cards / MMA': 'https://www.psacard.com/pop/boxing-wrestling-cards-mma/20021',
    'Football Cards': 'https://www.psacard.com/pop/football-cards/20014',
    'Football Coins': 'https://www.psacard.com/pop/football-coins/83011',
    'Golf Cards': 'https://www.psacard.com/pop/golf-cards/20023',
    'Hockey Cards': 'https://www.psacard.com/pop/hockey-cards/20020',
    'Hockey Coins': 'https://www.psacard.com/pop/hockey-coins/83012',
    'Minor League Cards': 'https://www.psacard.com/pop/minor-league-cards/20031',
    'Misc Cards': 'https://www.psacard.com/pop/misc-cards/20033',
    'Multi-Sport Cards': 'https://www.psacard.com/pop/multi-sport-cards/20006',
    'Multi-Sport Coins': 'https://www.psacard.com/pop/multi-sport-coins/102825',
    'Non-Sport Cards': 'https://www.psacard.com/pop/non-sport-cards/20032',
    'Non-Sport Coins': 'https://www.psacard.com/pop/non-sport-coins/82981',
    'Packs': 'https://www.psacard.com/pop/packs/20017',
    'Pins': 'https://www.psacard.com/pop/pins/20013',
    'Soccer Cards': 'https://www.psacard.com/pop/soccer-cards/20004',
    'TCG Cards': 'https://www.psacard.com/pop/tcg-cards/156940',
    'Tickets': 'https://www.psacard.com/pop/tickets/20022',
    # 'Game-Used Bats': 'https://www.psacard.com/pop/bats'
}


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(2), after=after_log)
def get_proxys(log):
    # 已购买账户  北美
    # http_proxy = "http://u1952150085001297:sJMHl4qc4bM0@proxy.123proxy.cn:36927"
    # https_proxy = "http://u1952150085001297:sJMHl4qc4bM0@proxy.123proxy.cn:36927"
    http_proxy = "http://u1952150085001297:sJMHl4qc4bM0@proxy.123proxy.cn:36931"
    https_proxy = "http://u1952150085001297:sJMHl4qc4bM0@proxy.123proxy.cn:36931"

    try:
        proxySettings = {
            "http": http_proxy,
            "https": https_proxy,
        }
        return proxySettings
    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_sets_data(log, category, category_id, category_link, tag_year, tag_year_link, sql_pool):
    # 隧道域名:端口号
    # tunnel = "x371.kdltps.com:15818"
    tunnel = "proxy.123proxy.cn:36927"
    options = ChromiumOptions()
    # options.set_paths(local_port=9137, user_data_path=r'D:\Drissionpage_temp\topps1_port_9137')
    # options = ChromiumOptions()
    options.set_proxy("http://" + tunnel)
    options.auto_port(True)
    # options.headless(True)
    options.set_argument("--disable-gpu")
    options.set_argument("-accept-lang=en-US")
    options.set_argument('--start-maximized')
    tab = ChromiumPage(options)
    try:
        # tab_url = f"https://www.psacard.com/pop/baseball-cards/2024/260045"
        tab_url = tag_year_link
        tab.get(tab_url)
        log.debug(f'{inspect.currentframe().f_code.co_name} -> 页面加载成功, url: {tab_url}')

        # time.sleep(11111)

        # 循环翻页直到最后一页
        page = 1
        while True:
            log.debug(f'{inspect.currentframe().f_code.co_name} -> 当前页码: {page}')
            html = tab.html
            if not html:
                log.error(f'{inspect.currentframe().f_code.co_name} -> 页面加载失败...........')
                raise Exception('页面加载失败, 重新加载........')

            selector = Selector(text=html)
            tag_tr_list = selector.xpath('//table[@id="tableSets"]/tbody/tr[position() > 1]')

            info_list = []
            for tag_tr in tag_tr_list:
                set_name = tag_tr.xpath('./td[@class="text-left"]/a[1]/text()').get()
                set_name_url = tag_tr.xpath('./td[@class="text-left"]/a[1]/@href').get()
                set_name_url = urljoin(BASE_URL, set_name_url) if set_name_url else None
                set_id = set_name_url.split('/')[-1] if set_name_url else None
                # print(set_name, set_name_url)
                data_dict = {
                    'category': category,
                    'category_id': category_id,
                    'category_link': category_link,
                    'year': tag_year,
                    'year_link': tag_year_link,
                    'set_name': set_name,
                    'set_link': set_name_url,
                    'set_id': set_id
                }
                info_list.append(data_dict)
                # try:
                #     get_player_list(log, data_dict, sql_pool)
                # except Exception as e:
                #     log.error(f'get_player_list error: {e}')

            # 保存数据
            if info_list:
                sql_pool.insert_many(table='psa_pop_player_sets', data_list=info_list, ignore=True)

            # 检查是否还有下一页
            next_button = tab.ele('#tableSets_next')
            if next_button and 'disabled' not in next_button.attr('class'):
                # 点击下一页按钮
                next_button.click()
                log.debug(f'{inspect.currentframe().f_code.co_name} -> 点击下一页按钮')
                # tab.wait.load_start() # 等待页面加载
                # time.sleep(2)  # 等待页面加载
            else:
                # 没有下一页了，退出循环
                log.debug(f'{inspect.currentframe().f_code.co_name} -> 没有下一页了, 退出循环, 最后页码: {page}')
                break

            page += 1

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


@retry(stop=stop_after_attempt(5), wait=wait_fixed(1), after=after_log)
def get_years_data(log, category, category_id, category_link, sql_pool):
    """
    获取详情数据
    :param log:
    :param category:
    :param category_id:
    :param category_link:
    :param sql_pool:
    """
    try:
        with requests.Session() as session:
            resp = session.get(category_link, impersonate=random.choice(client_identifier_list), headers=headers,
                               proxies=get_proxys(log), timeout=22, allow_redirects=False)
            # log.debug(resp.text)

        if 'Just a moment' in resp.text:
            log.debug('Just a moment , retrying.....')
            raise Exception('Just a moment')

        resp_selector = Selector(text=resp.text)

        tag_tr_list = resp_selector.xpath('//table[@id="tableCategory"]/tbody/tr')
        for tag_tr in tag_tr_list:
            tag_year = tag_tr.xpath('./td[1]/a/text()').get()
            tag_year_link = tag_tr.xpath('./td[1]/a/@href').get()
            tag_year_link = BASE_URL + tag_year_link if tag_year_link else tag_year_link

            try:
                get_sets_data(log, category, category_id, category_link, tag_year, tag_year_link, sql_pool)
            except Exception as e1:
                log.error(f"Error getting sets data: {e1}")

    except Exception as e:
        log.error(f"Error getting detail data: {e}")
        raise e


@retry(stop=stop_after_attempt(5), wait=wait_fixed(3), after=after_log)
def get_player_single_page(log, category_id, set_id, start, length, draw):
    """
    获取单页球员数据
    :param log: 日志对象
    :param category_id: 分类ID
    :param set_id: 集合ID
    :param start: 起始位置
    :param length: 数据长度
    :param draw: 请求序号
    :return: 响应数据字典或None
    """
    player_headers = {
        "accept": "application/json, text/javascript, */*; q=0.01",
        "content-type": "application/x-www-form-urlencoded; charset=UTF-8",
        "referer": f"https://www.psacard.com/pop/baseball-cards/2024/bowman/{set_id}",
        "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://www.psacard.com/Pop/GetSetItems"
    data = {
        "draw": str(draw),
        "start": str(start),
        "length": str(length),
        "search": "",
        "headingID": str(set_id),
        "categoryID": str(category_id),
        "isPSADNA": "false"
    }

    try:
        response = requests.post(
            url,
            impersonate=random.choice(client_identifier_list),
            headers=player_headers,
            data=data,
            timeout=22,
            proxies=get_proxys(log)
        )

        if response.status_code == 200:
            # print(response.json())
            return response.json()
        elif response.status_code == 403:
            log.error(f"请求被拒绝，请检查IP地址是否被封禁, set_id: {set_id}")
            # return None
            raise Exception('请求被拒绝')
        else:
            log.error(f"请求失败，状态码: {response.status_code}, set_id: {set_id}")
            # return None
            raise Exception('请求失败')

    except Exception as e:
        log.error(f"获取单页球员数据出错, set_id: {set_id}, start: {start}, error: {e}")
        # return None
        raise Exception('获取单页球员数据出错')


def get_player_list(log, category_id, set_id, sql_pool):
    """
    获取球员列表数据（带翻页功能）
    :param log: 日志对象
    :param category_id: category_id
    :param set_id: set_id
    :param sql_pool: 数据库连接池
    """
    start = 0  # 起始位置
    length = 300  # 每页数量
    draw = 1  # 请求序列号

    while True:
        log.debug(f"正在获取球员数据, category_id:{category_id}, set_id: {set_id}, page: {draw}, start: {start}")

        # 获取单页数据
        try:
            response_data = get_player_single_page(log, category_id, set_id, start, length, draw)
        except Exception as e:
            log.error(f"获取单页球员数据出错, category_id:{category_id}, set_id: {set_id}, start: {start}, error: {e}")
            response_data = None

        if response_data is None:
            log.error(f"获取球员数据失败, category_id:{category_id}, set_id: {set_id}, page: {draw}, break !!!")
            # sql_pool.update_one_or_dict(
            #     table='psa_pop_player_sets',
            #     data={'player_state': 2},
            #     condition={'set_id': set_id}
            # )
            break

        # 检查是否有数据返回
        player_data_list = response_data.get('data')

        # 如果是第一页  删除第一条
        if draw == 1:
            player_data_list.pop(0)

        if len(player_data_list) > 0:
            log.debug(
                f"获取到 {len(player_data_list)} 条球员数据, category_id:{category_id}, set_id: {set_id}, start: {start}")

            info_list = []
            for pl_data in player_data_list:
                spec_id = pl_data.get('SpecID')
                card_number = pl_data.get('CardNumber')
                subject_name = pl_data.get('SubjectName')
                card_set = pl_data.get('Variety')
                grade_total = pl_data.get('GradeTotal')

                data_dict = {
                    'category_id': category_id,
                    'set_id': set_id,
                    'spec_id': spec_id,
                    'card_number': card_number,
                    'subject_name': subject_name,
                    'card_set': card_set,
                    'grade_total': grade_total
                }
                # print(f'data_dict:{data_dict}')
                info_list.append(data_dict)

            # 保存数据到数据库
            if info_list:
                sql_pool.insert_many(table='psa_pop_player_record', data_list=info_list, ignore=True)

            # 如果返回数据少于请求长度，说明已是最后一页
            if len(player_data_list) < length:
                log.debug(f"已到达最后一页, set_id: {set_id}")
                break

            # 更新参数准备下一页
            start += length
            draw += 1

            # 添加延迟避免请求过于频繁
            time.sleep(1)
        else:
            log.debug(f"没有更多数据, set_id: {set_id}")
            break


@retry(stop=stop_after_attempt(100), wait=wait_fixed(3600), after=after_log)
def pop_main(log):
    """
    主函数
    """
    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:
        log.debug(".......... 开始获取数据报告 ..........")
        for category, category_link in category_link_list.items():
            log.debug(f"{category}第一次查询, 开始获取数据.......")
            try:
                category_id = category_link.split('/')[-1]
                get_years_data(log, category, category_id, category_link, sql_pool)
            except Exception as e1:
                log.error(f"Error getting detail data: {e1}")
    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} 运行结束,等待下一轮的采集任务............')


@retry(stop=stop_after_attempt(100), wait=wait_fixed(3600), after=after_log)
def player_main(log):
    """
    主函数
    """
    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:
        log.debug(".......... 开始获取数据报告 ..........")
        sql_sets_list = sql_pool.select_all(
            # "select category_id, set_id from psa_pop_player_sets where player_state = 0"
            "select category_id, set_id from psa_pop_player_sets")
        for category_set in sql_sets_list:
            category_id, set_id = category_set
            try:
                log.debug(f"category_id:{category_id}第一次查询, 开始获取数据.......")
                get_player_list(log, category_id, set_id, sql_pool)
                # sql_pool.update_one_or_dict(
                #     table='psa_pop_player_sets',
                #     data={'player_state': 1},
                #     condition={'set_id': set_id}
                # )

            except Exception as e1:
                log.error(f"Error getting detail data: {e1}")
                # sql_pool.update_one_or_dict(
                #     table='psa_pop_player_sets',
                #     data={'player_state': 2},
                #     condition={'set_id': set_id}
                # )
    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():
    """
    两个爬虫模块的启动文件
    bidding_main
    weika_change_card_by_id_spider
    change_card_main
    """
    # 立即运行一次任务
    # pop_main(log=logger)

    # player_main(log=logger)

    # 设置定时任务
    schedule.every().saturday.at("08:00").do(pop_main, log=logger)

    schedule.every().wednesday.at("08:00").do(player_main, log=logger)

    # schedule.every().day.at("00:30").do(player_main, log=logger)

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


if __name__ == '__main__':
    # get_detail_data(logger, '','https://www.psacard.com/pop/bats',None)
    # get_sets_data(logger)
    # get_player_single_page(logger, '21172', '279481', 0, 300, 1)

    # aa_dict = {
    #     'category': 'baseball',
    #     'category_id': '20003',
    #     'category_link': 'https://www.psacard.com/pop/bats',
    #     'tag_year': '2004',
    #     'tag_year_link': '',
    #     'set_name': '',
    #     'set_name_url': '',
    #     'set_id': '279664'
    # }
    # get_player_list(logger, aa_dict, None)

    # pop_main(log=logger)
    player_main(log=logger)

    # schedule_task()