spider_work/breakninja_spider/ninja_add_spider.py

# -*- coding: utf-8 -*-
# Author : Charley
# Python : 3.10.8
# Date   : 2025/9/8 11:56
import re
import requests
from parsel import Selector
import add_settings
import inspect
from loguru import logger
from tenacity import retry, stop_after_attempt, wait_fixed
from mysql_pool import MySQLConnectionPool

"""
第二个 页面  年份获取  /html/body/div[3]/span[position() >= 4 and position() <= 18]

"""

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")


def turn_page(log, category_name, year, series, series_url, sql_pool):
    """
    翻页
    :param log: 日志对象
    :param category_name: 类别名称
    :param year: 年份
    :param series: 系列名称
    :param series_url: 系列url
    :param sql_pool: 数据库连接池
    :return:
    """
    log.info(f"{inspect.currentframe().f_code.co_name} ->->-> series: {series}")
    turn_base_url = series_url.split('page_no=')[0] + 'page_no='
    page = 1
    max_page = 50
    while page < max_page + 1:
        # url = "https://www.breakninja.com/baseball/2025-Topps-All-Star-Game-Mega-Box-Baseball-Checklist.php?page_no=2"
        # 将参数 page_no 的值改为 page  构建新的URL
        current_url = turn_base_url + str(page)
        log.debug(f'{inspect.currentframe().f_code.co_name} ->->-> current_url: {current_url}')
        len_data, selector = turn_single_page(log, category_name, year, series, current_url, sql_pool)

        if len_data < 250:
            log.debug(f'{inspect.currentframe().f_code.co_name} ->->-> len_data: {len_data}, 少于250条,  break!!!')
            break

        max_page_text = selector.xpath('/html/body/div[3]/div[4]/strong[1]/text()').get()
        # if max_page_:
        if max_page_text:
            # 使用正则表达式提取 "of" 后面的数字
            match = re.search(r'of\s+(\d+)', max_page_text)
            if match:
                total_pages = int(match.group(1))
                max_page = (total_pages // 250) + 1  # 计算最大翻页次数
                log.debug(f'{inspect.currentframe().f_code.co_name} ->->-> max_page: {max_page}')

        page += 1


@retry(stop=stop_after_attempt(5), wait=wait_fixed(1), after=after_log)
def turn_single_page(log, category_name, year, series, current_url, sql_pool):
    """
    翻页  单页查询
    :param log:
    :param category_name:
    :param year:
    :param series:
    :param current_url:
    :param sql_pool:
    :return:
    """
    log.info(f"{inspect.currentframe().f_code.co_name} ->->-> series: {series}")
    response = requests.get(current_url, headers=add_settings.headers, timeout=60)
    # print(response.text)
    response.raise_for_status()

    selector = Selector(response.text)
    tag_table = selector.xpath('//table[@class="table table-striped table-bordered"]')  # 如果是这种  需要翻页
    log.debug(f'{inspect.currentframe().f_code.co_name} tag_table size: {len(tag_table)}')

    title = selector.xpath('//h1/text()').get()

    if tag_table:
        log.debug(f'{inspect.currentframe().f_code.co_name} ->->-> tag_table size 111: {len(tag_table)}')
        # len_data = parse_one_team(log, tag_table, category_name, series, sql_pool)
        len_data = parse_turn_single_team(log, tag_table, category_name, year, series, current_url, sql_pool, title)
    else:
        len_data = 0

    return len_data, selector


def parse_turn_single_team(log, tag_table, category_name, year, series, series_url, sql_pool, title):
    # 获取表头（第一行）
    head = tag_table.xpath('./thead/tr/th/text()').getall()
    # 统一表头字段名：将 "#" 替换为 "Number"
    head = ['Number' if h.strip() == '#' else h for h in head]
    head = ['Player' if h.strip() == 'Name' else h for h in head]
    head = ['Print Run' if h.strip() == 'Run' else h for h in head]

    # 获取所有行的数据
    rows = tag_table.xpath('./tbody/tr')

    data = []

    for row in rows:
        cells = row.xpath('./td/text()').getall()
        cells = [cell.strip() for cell in cells if cell.strip()]

        if not cells:
            cells = [''] * len(head)  # 如果没有数据，用空字符串填充

        # 将当前行的数据与表头组合成字典
        row_data = dict(zip(head, cells))

        # 检查是否所有字段都为空，如果都不为空才添加
        if any(value.strip() for value in row_data.values()):
            data.append(row_data)

    # 打印结果
    log.info(f'{inspect.currentframe().f_code.co_name} ->->-> len data size: {len(data)}')

    # 定义数据库中存在的字段
    database_fields = {
        'player', 'number', 'card_set', 'copies', 'team', 'print_run',
        'category_name', 'year', 'series', 'series_url', 'team_url', 'title'
    }

    info_list = []
    for item in data:
        item.update(
            {
                'category_name': category_name,
                'year': year,
                'series': series,
                'series_url': series_url,
                'title': title
            }
        )
        # print(item)

        # 处理字段名：转换为小写并替换特殊字符
        processed_item = {}
        for key, value in item.items():
            # 转换为小写，并将空格和特殊字符替换为下划线
            new_key = key.lower().replace(' ', '_').replace('-', '_').replace('/', '_').replace('(', '').replace(
                ')', '')
            processed_item[new_key] = value
        # print(processed_item)

        # 创建统一结构的字典，确保所有字段都存在
        unified_item = {}
        for field in database_fields:
            unified_item[field] = processed_item.get(field, None)

        info_list.append(unified_item)

    # print(info_list)

    try:
        if info_list:
            # sql_pool.insert_one_or_dict(table='breakninja_checklist_record_202511', data=processed_item)
            sql_pool.insert_many(table='breakninja_checklist_record_202511', data_list=info_list)
    except Exception as e:
        log.error(f"insert_many data: {e}")

    return len(data)


@retry(stop=stop_after_attempt(100), wait=wait_fixed(3600), after=after_log)
def ninja_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:
        for ca in add_settings.series_url_list:
            category_name = ca.get('category_name')
            years = ca.get('years')
            for ye in years:
                year = ye.get('year')
                series = ye.get('series')
                for ser in series:
                    series_name = ser.get('series_name')
                    series_url = ser.get('series_url')
                    log.debug(
                        f'category_name: {category_name}, year: {year}, series_name: {series_name}, series_url: {series_url}')

                    try:
                        if '?page_no=' in series_url:
                            log.debug(f'page_no in series_url, request turn_page.........')
                            turn_page(log, category_name, year, series_name, series_url, sql_pool)
                        else:
                            log.debug(f'page_no not in series_url, request turn_single_page.........')
                    except Exception as e:
                        log.error(f"Error processing series: {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} 运行结束,等待下一轮的采集任务............')


if __name__ == '__main__':
    ninja_main(logger)