spider_work/scp_spider/scp_spider.py

# -*- coding: utf-8 -*-
# Author : Charley
# Python : 3.12.10
# Date   : 2026/07/15
"""
SCP Auctions (catalogs.scpauctions.com) 每周增量爬虫（周调度）
逻辑（两阶段）：
  1. GET /auctions/past 只解析首页（最新的拍卖会排在首页顶部）
  2. 查库 select distinct event_id from scp_auction，得到已爬过的场次
  3. 差集 = 新增拍卖会
  4. 没有新增 → 本轮无数据可抓，仍补跑一次阶段二兜底后结束
  5. 阶段一 列表：对每个新增拍卖会入库 scp_auction，再翻页抓 lot 列表入库 scp_lot（state 默认 0）
  6. 阶段二 详情：扫库 scp_lot 中 state != 1 的记录 → 逐条进详情页抓字段写回

说明：SCP 的拍卖会一旦结束即定型，新场次会出现在 /auctions/past 首页顶部；
      因此增量只需盯首页，用 event_id 差集识别新增场次即可（老场次早已在库）。
"""
import sys
import time
import random

import schedule
from curl_cffi import requests
from loguru import logger
from tenacity import retry, stop_after_attempt, wait_fixed
from mysql_pool import MySQLConnectionPool

from scp_core import (
    client_identifier_list,
    crawl_one_auction,
    get_auction_list,
    save_auction,
    update_details_for_pending,
    after_log,
)

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 get_existing_event_ids(log, sql_pool):
    """查库返回已爬过的 event_id 集合。

    Args:
        log: logger 对象。
        sql_pool: MySQL 连接池；为 None 时返回空集合（视作库内无任何场次）。

    Returns:
        set[str]: 已存在的 event_id 字符串集合。
    """
    if sql_pool is None:
        log.warning("sql_pool 为 None，视为库内无任何场次（将全量重抓首页）")
        return set()

    rows = sql_pool.select_all("select distinct event_id from scp_auction_record")
    keys = {str(r[0]) for r in rows} if rows else set()
    log.info(f"库中已存在 {len(keys)} 个 event_id")
    return keys


def diff_new_auctions(log, recent_auctions, existing_keys):
    """从首页拍卖会中筛出库里没有的新增场次。

    Args:
        log: logger 对象。
        recent_auctions (list[dict]): get_auction_list(only_first_page=True) 的返回。
        existing_keys (set[str]): 已存在的 event_id 集合。

    Returns:
        list[dict]: 待抓取的新增拍卖会列表。
    """
    new_list = [a for a in recent_auctions if a["event_id"] not in existing_keys]
    log.info(f"新增待抓取拍卖会数: {len(new_list)} -> {[a['event_id'] for a in new_list]}")
    return new_list


def run_incremental(log, sql_pool):
    """增量抓取主流程（阶段一）。

    Args:
        log: logger 对象。
        sql_pool: MySQL 连接池；为 None 时不入库，仅在内存收集并 print 样本。

    Returns:
        None: 无返回值。
    """
    impersonate = random.choice(client_identifier_list)
    with requests.Session() as session:
        try:
            recent = get_auction_list(log, session, impersonate, only_first_page=True)
        except Exception as e:
            log.error(f"获取首页拍卖会列表失败: {e}")
            return

        existing_keys = get_existing_event_ids(log, sql_pool)
        new_auctions = diff_new_auctions(log, recent, existing_keys)

        if not new_auctions:
            log.info("本轮无新增拍卖会，跳过 list 抓取")
            return

        for idx, auc in enumerate(new_auctions, 1):
            log.info(f"========== [{idx}/{len(new_auctions)}] 新增拍卖会 {auc['event_id']} ({auc['auction_name']}) ==========")
            try:
                save_auction(log, sql_pool, auc)
                crawl_one_auction(log, sql_pool, session, impersonate, auc)
            except Exception as e:
                log.error(f"拍卖会 {auc['event_id']} 抓取异常: {e}")
                continue


@retry(stop=stop_after_attempt(100), wait=wait_fixed(3600), after=after_log)
def scp_main(log):
    """周调度主函数：增量 list + 补详情。

    Args:
        log: logger 对象。

    Returns:
        None: 无返回值。
    """
    log.info(f"开始运行 {sys._getframe().f_code.co_name} 增量爬虫任务 ...")

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

    try:
        # 阶段一：抓新增拍卖会的 lot 列表入库
        try:
            run_incremental(log, sql_pool)
        except Exception as e:
            log.error(f"增量抓取失败: {e}")

        # 阶段二：扫库 state != 1 的 lot 补抓详情（含上一轮失败/中断的兜底）
        try:
            update_details_for_pending(log, sql_pool)
        except Exception as e:
            log.error(f"详情补抓失败: {e}")

    except Exception as e:
        log.error(f"{sys._getframe().f_code.co_name} error: {e}")
    finally:
        log.info(f"爬虫程序 {sys._getframe().f_code.co_name} 运行结束，等待下一轮采集 ...")


def schedule_task():
    """启动调度器：先即时跑一次，之后每周一 05:00 跑一次增量。

    Returns:
        None: 常驻循环，无返回值。
    """
    scp_main(log=logger)

    schedule.every().monday.at("05:00").do(scp_main, log=logger)
    while True:
        schedule.run_pending()
        time.sleep(1)


if __name__ == "__main__":
    # 测试时直接跑一次
    # scp_main(log=logger)
    # 上生产再切回 schedule
    schedule_task()