spider_work/hugginsandscott_spider/hug_spider.py

# -*- coding: utf-8 -*-
# Author : Charley
# Python : 3.12.10
# Date   : 2026/5/19 16:56
import time
import inspect
import requests
import schedule
import user_agent
from loguru import logger
from parsel import Selector
from mysql_pool import MySQLConnectionPool
from tenacity import retry, stop_after_attempt, wait_fixed

"""
目标网站: https://hugginsandscott.com/auction/2026/winter/
"""

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": "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()
}


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: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(2), after=after_log)
def get_details(log, url, sql_pool, sql_id):
    """
    获取详情数据
    :param log: logger对象
    :param url: 详情页URL
    :param sql_pool: MySQL连接池
    :param sql_id: 数据ID
    :return: 标题和描述
    """
    log.info(f">>>>>>>>>>>>>> 正在爬取详情数据URL: {url} <<<<<<<<<<<<<<")
    response = requests.get(url, headers=headers, timeout=10, proxies=get_proxys(log))
    response.raise_for_status()
    selector = Selector(response.text)
    year = selector.xpath('//section//div/ul/li[2]/text()').get()
    year = year.replace('Year: ', '') if year else None

    auction = selector.xpath('//section//div/ul/li[3]/text()').get()
    auction = auction.replace('Auction: ', '') if auction else None

    imgs = selector.xpath('//section//button/span/img/@src').getall()
    imgs = ','.join(imgs) if imgs else None
    # print(year, auction, imgs)

    # 更新数据和状态
    sql_pool.update_one_or_dict(
        table="hugginsandscott_record",
        data={"year": year, "auction": auction, "imgs": imgs, "state": 1},
        condition={"id": sql_id}
    )


@retry(stop=stop_after_attempt(5), wait=wait_fixed(2), after=after_log)
def get_single_page(log, page, auc_url, sql_pool):
    """
    获取单页数据
    :param log: logger对象
    :param page: 页码
    :param auc_url: 拍卖会URL
    :param sql_pool: MySQL连接池
    :return: 该页数据条数
    """
    log.info(f"Requesting page {page} for auction:{auc_url}........................... started")
    # url = "https://hugginsandscott.com/auction/2026/winter/"
    params = {
        # "page": "3"
        "page": page
    }
    response = requests.get(auc_url, headers=headers, params=params, timeout=10, proxies=get_proxys(log))
    # print(response.text)
    response.raise_for_status()

    selector = Selector(response.text)
    tag_tr_list = selector.xpath("//table/tbody/tr")

    info_list = []
    for tag_tr in tag_tr_list:
        lot_no = tag_tr.xpath("./td[1]/text()").get()
        title = tag_tr.xpath("./td[2]/a/text()").get()
        title = title.strip() if title else ""

        detail_url = tag_tr.xpath("./td[2]/a/@href").get()
        detail_url = 'https://hugginsandscott.com' + detail_url if detail_url else ""

        category = tag_tr.xpath("./td[3]/text()").get()
        sold_for = tag_tr.xpath("./td[4]/text()").get()
        sold_for = sold_for.replace(",", "").replace("$", "") if sold_for else "0"

        data_dict = {
            "auction_url": auc_url,
            "lot_no": lot_no,
            "title": title,
            "detail_url": detail_url,
            "category": category,
            "sold_for": sold_for
        }
        # print(data_dict)
        info_list.append(data_dict)

    # save
    if info_list:
        sql_pool.insert_many(table="hugginsandscott_record", data_list=info_list, ignore=True)

    return len(info_list)


def get_sold_list(log, auc_url, sql_pool):
    """
    获取已售列表
    :param log: logger对象
    :param auc_url: 拍卖会URL
    :param sql_pool: MySQL连接池
    :return: 无
    """
    page = 1
    max_page = 10

    while page <= max_page:
        try:
            len_list = get_single_page(log, page, auc_url, sql_pool)
        except Exception as e:
            log.error(f"Error getting page {page}: {e}")
            continue

        if len_list < 10:
            log.warning(f"No data on page {page}, stopping further requests")
            break

        page += 1


@retry(stop=stop_after_attempt(5), wait=wait_fixed(2), after=after_log)
def get_all_auctions(log, sql_pool):
    """
    获取所有拍卖会
    :param log: logger对象
    :param sql_pool: MySQL连接池
    :return: 无
    """
    log.info(">>>>>>>>>>>>>> 正在获取所有拍卖会 <<<<<<<<<<<<")
    # //div[@class="max-w-screen-xl px-4 py-8 mx-auto"]//a/@href
    url = 'https://hugginsandscott.com/search'
    response = requests.get(url, headers=headers, timeout=10, proxies=get_proxys(log))
    response.raise_for_status()
    selector = Selector(response.text)
    tag_a_list = selector.xpath("//div[@class='max-w-screen-xl px-4 py-8 mx-auto']//a/@href").getall()
    log.info(f"Total auctions: {len(tag_a_list)}")
    # 补全 https://hugginsandscott.com/
    tag_a_list = [f"https://hugginsandscott.com{tag_a}" for tag_a in tag_a_list]

    # 批量查询库中已存在的 URL，避免 N 次单条查询
    if tag_a_list:
        placeholders = ','.join(['%s'] * len(tag_a_list))
        sql = f"SELECT auction_url FROM hugginsandscott_record WHERE auction_url IN ({placeholders})"
        existed_rows = sql_pool.select_all(sql, tuple(tag_a_list))
        # 库中已存在的 URL 集合，便于 O(1) 查找
        existed_urls = {row[0] for row in existed_rows}
        # 用集合差集筛选出不在库中的新拍卖会 URL
        new_tag_a_list = [tag_a for tag_a in tag_a_list if tag_a not in existed_urls]
    else:
        new_tag_a_list = []

    log.info(f"Total auctions after filter: {len(new_tag_a_list)}")

    # 打印不在库中的新 URL 列表
    for auc_url in new_tag_a_list:
        try:
            log.info(f"Getting sold list for:{auc_url}...........................")
            get_sold_list(log, auc_url, sql_pool)
        except Exception as e:
            log.error(f"{inspect.currentframe().f_code.co_name} -> getting sold list for:{auc_url}: {e}")


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

    # 配置 MySQL 连接池
    sql_pool = MySQLConnectionPool(log=log)
    if not sql_pool:
        log.error("MySQL数据库连接失败")
        raise Exception("MySQL数据库连接失败")

    try:
        try:
            get_all_auctions(log, sql_pool)
        except Exception as e:
            log.error(f'Error getting all auctions list: {e}')

        # 更新详情页
        log.debug('Updating detail pages........................... started')
        # sql_result = sql_pool.select_all('select id, detail_url from hugginsandscott_record where state = 0')
        sql_result = sql_pool.select_all(
            'select id, detail_url from hugginsandscott_record where state != 1 order by id')
        for row in sql_result:
            sql_id = row[0]
            detail_url = row[1]
            try:
                get_details(log, detail_url, sql_pool, sql_id)
            except Exception as e:
                log.error(f'Error getting details for {detail_url}: {e}')
                # 更新数据和状态
                sql_pool.update_one_or_dict(
                    table="hugginsandscott_record",
                    data={"state": 2},
                    condition={"id": sql_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():
    """每半个月 跑一次增量"""
    hug_main(log=logger)

    def run_semimonthly():
        # 每月 1 号和 15 号执行（半月一次）
        from datetime import date
        if date.today().day in (1, 15):
            hug_main(log=logger)

    schedule.every().day.at("05:00").do(run_semimonthly)
    while True:
        schedule.run_pending()
        time.sleep(1)


if __name__ == '__main__':
    # get_single_page(logger, 1, None)d
    # get_details(logger,
    #             "https://hugginsandscott.com/auction/2026/Winter/1/1952-topps-311-mickey-mantle-high-number-psa-vg-3-mba-silver-diamond-centered",
    #             None, 1)

    # hug_main(log=logger)
    schedule_task()
    # sql_pool_ = MySQLConnectionPool(log=logger)
    # get_all_auctions(logger, sql_pool_)