spider_work/scp_spider/scp_core.py

# -*- coding: utf-8 -*-
# Author : Charley
# Python : 3.12.10
# Date   : 2026/07/15
"""
SCP Auctions (catalogs.scpauctions.com) 公用模块：HTTP 配置、拍卖会列表解析、
场次内 lot 列表分页抓取、详情页解析。被 scp_history.py / scp_spider.py 复用。

目标网站: https://catalogs.scpauctions.com/auctions/past
逻辑要点:
    1. 站点为 Bidsquare 托管平台，页面服务端渲染（SSR），无独立数据 API，
       全部数据直接在 HTML 里，用 parsel 解析即可（本地开了 VPN，请求统一走代理）。
    2. 拍卖会列表页 /auctions/past 分页：?page=N（每页 24 场），翻到空页为止。
       每张拍卖会卡片是 <div class="gtm-visible_event" data-event_id=...>：
         event_id（去重键）、标题、类型、起止时间、主页/图录链接、封面图。
    3. 图录（lot 列表）页 catalog_url?page=N&limit=120（limit 上限 120）：
       lot 卡片是 <div id="stl-{item_id}" data-item_id=...>，含 Lot 号 / 标题 /
       详情链接 / 缩略图 / 出价数 / 成交状态与成交价。翻到空页为止。
    4. lot 详情页 /online-auctions/{house}/{slug}-{item_id}：
         标题取 og:title；成交价在 .bidding-price（流拍则为空）；
         属性（Collection/Sport/Type/Athlete/Team 等）在 .item-attributes li，
           label/span 成对，字段数量不固定（2~5 个，可能缺 Athlete/Team），
           故按 label 建字典再按需取值，缺失置空；
         多图为 s1.img.bidsquare.com/item/{l|s}/... 归一化到 large 尺寸去重。
"""
import re
import random
from datetime import datetime

from loguru import logger
from parsel import Selector
from curl_cffi import requests
from curl_cffi.requests import BrowserType
from urllib.parse import urljoin
from tenacity import retry, stop_after_attempt, wait_fixed

# —— 站点常量 ——
BASE_URL = "https://catalogs.scpauctions.com"  # 站点根，用于拼接相对链接
PAST_URL = "https://catalogs.scpauctions.com/auctions/past"  # 拍卖会（历史）列表页
CATALOG_PAGE_SIZE = 120  # 图录页每页 lot 数（limit 参数，站点上限 120）
IMG_ITEM_RE = r'https://s1\.img\.bidsquare\.com/item/[ls]/\d+/\d+\.jpe?g(?:\?t=[^"\'\s\\]+)?'  # 拍品图 URL

# 直接用库内置的所有浏览器指纹（伪装真实浏览器 TLS/UA，规避潜在的指纹风控）
client_identifier_list = [b.value for b in BrowserType]

# 不写死 user-agent：curl_cffi 会按 impersonate 指纹自动带匹配的 UA，
# 写死会造成 TLS 指纹与 UA 头矛盾，反而更易被识别。只保留通用、不冲突的头。
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",
    "accept-language": "en-US,en;q=0.9",
}


def after_log(retry_state):
    """tenacity retry 回调，统一打印重试日志。

    Args:
        retry_state: tenacity.RetryCallState，retry 框架自动传入。

    Returns:
        None: 仅打印日志，无返回值。
    """
    if retry_state.args and len(retry_state.args) > 0:
        log = retry_state.args[0]
    else:
        log = 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):
    """获取代理字典，全部请求方法的 proxies 参数都走这里。

    本地开了 VPN，从大陆直连站点不通，请求必须走代理。默认用住宅代理（北美出口，
    与 SCP 面向的美区一致）。若要改用本地 VPN 客户端的 HTTP 代理端口（如 Clash 的
    127.0.0.1:7890），把 http_proxy / https_proxy 换成对应地址即可。

    Args:
        log: logger 对象。

    Returns:
        dict: requests 风格代理字典 {"http": ..., "https": ...}。

    Raises:
        Exception: 透传内部异常以便 tenacity 触发重试。
    """
    # 住宅 ip 池 北美（与参考项目 rea_spider 一致）；如需切本地 VPN 端口在此改
    http_proxy = "http://u1952150085001297:sJMHl4qc4bM0@proxy.123proxy.cn:36927"
    https_proxy = "http://u1952150085001297:sJMHl4qc4bM0@proxy.123proxy.cn:36927"
    # 本地 VPN（Clash）示例，二选一：
    # http_proxy = https_proxy = "http://127.0.0.1:7890"

    try:
        return {
            "http": http_proxy,
            "https": https_proxy,
        }
    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_selector(log, session, impersonate, url):
    """GET 一个页面并返回 parsel Selector（带重试，走代理）。

    Args:
        log: logger 对象。
        session (requests.Session): curl_cffi 会话对象。
        impersonate (str): 浏览器指纹标识。
        url (str): 目标绝对 URL。

    Returns:
        Selector: 响应 HTML 的 parsel 解析对象。

    Raises:
        Exception: HTTP 状态非 2xx 时透传异常以触发重试。
    """
    resp = session.get(url, headers=headers, impersonate=impersonate,
                       proxies=get_proxys(log), timeout=30)
    resp.raise_for_status()
    return Selector(resp.text)


def _parse_datetime(raw):
    """把站点时间原文解析成 datetime（丢弃时区名，保留东部墙钟时间）。

    Args:
        raw (str): 时间原文，如 "May 20, 2026 01:00PM EDT" 或带 "Start:" 前缀。

    Returns:
        datetime | None: 解析成功返回 datetime；失败或空返回 None。
    """
    if not raw:
        return None
    # 去掉 Start:/End: 前缀与末尾时区名（EDT/EST/…），只留 "May 20, 2026 01:00PM"
    text = re.sub(r'^\s*(Start|End)\s*:\s*', '', raw.strip(), flags=re.I)
    m = re.search(r'([A-Za-z]{3,9}\s+\d{1,2},\s*\d{4}\s+\d{1,2}:\d{2}\s*[AP]M)', text, re.I)
    if not m:
        return None
    try:
        return datetime.strptime(m.group(1).replace("  ", " ").upper(), "%b %d, %Y %I:%M%p")
    except ValueError:
        return None


def parse_auction_list(selector):
    """解析 /auctions/past 单页的拍卖会卡片列表。

    Args:
        selector (Selector): 拍卖会列表页某一页 GET 响应的 parsel 解析对象。

    Returns:
        list[dict]: 每场一个 dict，字段：event_id, auction_name, auction_type,
            event_status, start_time_raw, end_time_raw, start_time, end_time,
            auction_url, catalog_url, cover_img；无卡片返回空列表。
    """
    auctions = []
    for card in selector.css('div.gtm-visible_event[data-event_id]'):
        event_id = card.attrib.get("data-event_id", "").strip()
        if not event_id:
            continue

        # 标题 + 拍卖会主页链接
        title_a = card.css('.list-top h1 a')
        auction_name = (title_a.xpath('normalize-space(.)').get() or
                        card.attrib.get("data-event_name", "")).strip()
        auction_url = title_a.attrib.get("href", "").strip()

        # 类型（Timed Auction 等）
        auction_type = (card.css('.list-top label::text').get() or "").strip()

        # 起止时间：.list-top 下多个 <span>，按前缀区分
        start_raw = end_raw = ""
        for sp in card.css('.list-top span'):
            t = (sp.xpath('normalize-space(.)').get() or "").strip()
            if t.lower().startswith("start"):
                start_raw = t
            elif t.lower().startswith("end"):
                end_raw = t

        # 图录链接（去掉锚点 #catalog）
        catalog_url = (card.css('a.view-catalog-btn::attr(href)').get() or "").strip()
        catalog_url = catalog_url.split("#")[0]

        cover_img = (card.css('.slider-for img::attr(src)').get() or "").strip()

        start_dt = _parse_datetime(start_raw)
        end_dt = _parse_datetime(end_raw)
        auctions.append({
            "event_id": event_id,
            "auction_name": auction_name,
            "auction_type": auction_type,
            "event_status": card.attrib.get("data-event_status", "").strip(),
            "start_time_raw": re.sub(r'^\s*Start\s*:\s*', '', start_raw, flags=re.I).strip(),
            "end_time_raw": re.sub(r'^\s*End\s*:\s*', '', end_raw, flags=re.I).strip(),
            "start_time": start_dt.strftime("%Y-%m-%d %H:%M:%S") if start_dt else None,
            "end_time": end_dt.strftime("%Y-%m-%d %H:%M:%S") if end_dt else None,
            "auction_url": urljoin(BASE_URL, auction_url) if auction_url else None,
            "catalog_url": urljoin(BASE_URL, catalog_url) if catalog_url else None,
            "cover_img": cover_img or None,
        })
    return auctions


@retry(stop=stop_after_attempt(3), wait=wait_fixed(2), after=after_log)
def get_auction_list(log, session, impersonate, only_first_page=False):
    """翻页抓取 /auctions/past 全部（或仅首页）拍卖会。

    Args:
        log: logger 对象。
        session (requests.Session): curl_cffi 会话对象。
        impersonate (str): 浏览器指纹标识。
        only_first_page (bool, optional): True 仅取首页（增量用，最近场次在首页顶部）；
            False 翻完全部页（全量用）。Defaults to False。

    Returns:
        list[dict]: 拍卖会 dict 列表，已按 event_id 去重。
    """
    scope = "首页" if only_first_page else "全部"
    log.info(f"获取{scope}拍卖会列表")
    result, seen = [], set()
    page = 1
    while True:
        url = f"{PAST_URL}?page={page}"
        sel = _get_selector(log, session, impersonate, url)
        page_auctions = parse_auction_list(sel)
        if not page_auctions:
            break  # 空页 = 已翻过末页
        new = [a for a in page_auctions if a["event_id"] not in seen]
        for a in new:
            seen.add(a["event_id"])
            result.append(a)
        if not new:
            break  # 本页全重复（越界回卷兜底），停止
        log.info(f"  第 {page} 页解析到 {len(page_auctions)} 场（新增 {len(new)}，累计 {len(result)}）")
        if only_first_page:
            break
        page += 1
    log.info(f"共解析到 {len(result)} 场拍卖会")
    return result


def _clean_price(text):
    """把成交价文本清洗成纯数字字符串。

    Args:
        text (str): 原始价格文本，如 "$90,000"。

    Returns:
        str: 去掉 $ 和千分位逗号后的数字串，如 "90000"；无价格返回空串。
    """
    if not text:
        return ""
    m = re.search(r'\$?\s*([\d,]+)', text)
    return m.group(1).replace(",", "") if m else ""


def parse_lot_cards(selector, auction):
    """解析图录页单页的 lot 卡片。

    Args:
        selector (Selector): 图录页某一页 GET 响应的 parsel 解析对象。
        auction (dict): 当前场次 dict（含 event_id / auction_name），回填到每条 lot。

    Returns:
        list[dict]: 每条 lot 一个 dict，字段：item_id, event_id, auction_name,
            lot_number, title, detail_url, list_img, sold_status, sold_for, bids；
            无卡片返回空列表。
    """
    rows = []
    for card in selector.css('div[id^="stl-"][data-item_id]'):
        item_id = card.attrib.get("data-item_id", "").strip()
        if not item_id:
            continue

        title_a = card.css('.lot_title a')
        title = (title_a.xpath('normalize-space(.)').get() or "").strip()
        detail_url = (title_a.attrib.get("href") or "").strip()

        # Lot 号："Lot 1" → "1"
        lot_raw = (card.css('.lot_Num::text').get() or "").strip()
        lot_number = (re.search(r'(\d+)', lot_raw) or [None, ""])[1] if lot_raw else ""

        list_img = (card.css('.catalog_img img::attr(src)').get() or "").strip()

        # 底部区："9 BidsSold for$90,000" / "24 BidsUnsold"
        bottom = card.css('.item-list-bottom')
        bottom_text = (bottom.xpath('normalize-space(.)').get() or "") if bottom else ""
        bids_m = re.search(r'(\d+)\s*Bids?', bottom_text, re.I)
        bids = bids_m.group(1) if bids_m else ""
        if re.search(r'Unsold', bottom_text, re.I):
            sold_status, sold_for = "unsold", ""
        elif re.search(r'Sold\s*for', bottom_text, re.I):
            sold_status = "sold"
            sold_for = _clean_price(re.sub(r'.*Sold\s*for', '', bottom_text, flags=re.I))
        else:
            sold_status, sold_for = "", ""

        rows.append({
            "item_id": item_id,
            "event_id": auction["event_id"],
            "auction_name": auction.get("auction_name"),
            "lot_number": lot_number,
            "title": title,
            "detail_url": urljoin(BASE_URL, detail_url) if detail_url else None,
            "list_img": list_img or None,
            "sold_status": sold_status or None,
            "sold_for": sold_for or None,
            "bids": bids or None,
        })
    return rows


@retry(stop=stop_after_attempt(5), wait=wait_fixed(2), after=after_log)
def _fetch_lot_page(log, session, impersonate, catalog_url, page):
    """抓取图录页的某一页 lot 卡片。

    Args:
        log: logger 对象。
        session (requests.Session): curl_cffi 会话对象。
        impersonate (str): 浏览器指纹标识。
        catalog_url (str): 图录页绝对 URL（不含 query）。
        page (int): 页码，从 1 开始。

    Returns:
        Selector: 该页响应的 parsel 解析对象。
    """
    url = f"{catalog_url}?page={page}&limit={CATALOG_PAGE_SIZE}"
    return _get_selector(log, session, impersonate, url)


def fetch_auction_lots(log, session, impersonate, auction):
    """抓取一个场次下的全部 lot 列表（翻页，去重 item_id）。

    Args:
        log: logger 对象。
        session (requests.Session): curl_cffi 会话对象。
        impersonate (str): 浏览器指纹标识。
        auction (dict): 场次 dict（含 event_id / auction_name / catalog_url）。

    Returns:
        list[dict]: 该场次全部 lot 列表（阶段一字段，不含详情）。
    """
    catalog_url = auction.get("catalog_url")
    if not catalog_url:
        log.warning(f"场次 {auction['event_id']} 无 catalog_url，跳过")
        return []

    log.info(f"开始抓取场次 {auction['event_id']}({auction.get('auction_name')}) 的 lot 列表")
    all_lots, seen = [], set()
    page = 1
    while True:
        sel = _fetch_lot_page(log, session, impersonate, catalog_url, page)
        lots = parse_lot_cards(sel, auction)
        if not lots:
            break  # 空页 = 已翻过末页
        new = [x for x in lots if x["item_id"] not in seen]
        for x in new:
            seen.add(x["item_id"])
        all_lots.extend(new)
        log.info(f"  第 {page} 页 {len(lots)} 条（新增 {len(new)}，累计 {len(all_lots)}）")
        if not new:
            break  # 本页全重复，停止
        page += 1
    log.info(f"场次 {auction['event_id']} 共抓 {len(all_lots)} 条 lot")
    return all_lots


def parse_lot_detail(selector):
    """解析 lot 详情页，抽取标题、成交价、属性、多图。

    详情页属性区（.item-attributes）字段数量不固定：不同拍品可能是 2~5 个，
    Athlete/Team 等常缺失。故先把所有 li 的 label→span 建成字典，再按需取字段，
    缺失字段返回 None，保证解析健壮。

    Args:
        selector (Selector): 详情页 GET 响应的 parsel 解析对象。

    Returns:
        dict: {
            "title": 标题, "sold_for": 成交价(纯数字串,流拍为空),
            "sold_status": "sold"/"unsold", "bids": 出价数,
            "collection", "sport", "item_type", "athlete", "team": 属性(缺失为 None),
            "imgs": 多图 URL 逗号拼接
        }。
    """
    # 标题：og:title 最干净（页面首个 h1 是拍卖会名，非拍品标题）
    title = (selector.css('meta[property="og:title"]::attr(content)').get() or "").strip()
    if not title:
        # 兜底：取非拍卖会名的那个 h1
        h1s = [t.strip() for t in selector.css('h1 ::text, h1::text').getall() if t.strip()]
        title = h1s[1] if len(h1s) > 1 else (h1s[0] if h1s else "")

    # 成交价：.bidding-price 内首个 $金额；流拍则为空
    price_text = selector.css('.bidding-price').xpath('normalize-space(.)').get() or ""
    sold_for = _clean_price(price_text)
    sold_status = "sold" if sold_for else "unsold"

    # 出价数：页面 "[9 Bids]"
    body_text = selector.xpath('normalize-space(//body)').get() or ""
    bids_m = re.search(r'\[?\s*(\d+)\s*Bids?\s*\]?', body_text, re.I)
    bids = bids_m.group(1) if bids_m else ""

    # 属性：.item-attributes li -> {label: value}，字段数量不固定，按需取
    attrs = {}
    for li in selector.css('.item-attributes li'):
        label = (li.css('label::text').get() or "").strip().rstrip(":")
        value = (li.css('span').xpath('normalize-space(.)').get() or "").strip()
        if label:
            attrs[label.lower()] = value

    # 多图：/item/{l|s}/... 统一归一化到 large 尺寸并去重（保持出现顺序）
    imgs, seen = [], set()
    for u in selector.re(IMG_ITEM_RE):
        large = u.replace("/item/s/", "/item/l/")
        if large not in seen:
            seen.add(large)
            imgs.append(large)

    return {
        "title": title,
        "sold_for": sold_for or None,
        "sold_status": sold_status,
        "bids": bids or None,
        "collection": attrs.get("collection") or None,
        "sport": attrs.get("sport") or None,
        "item_type": attrs.get("type") or None,
        "athlete": attrs.get("athlete") or None,
        "team": attrs.get("team") or None,
        "imgs": ",".join(imgs) if imgs else None,
    }


@retry(stop=stop_after_attempt(3), wait=wait_fixed(2), after=after_log)
def fetch_lot_detail(log, session, impersonate, detail_url):
    """GET lot 详情页并解析。

    Args:
        log: logger 对象。
        session (requests.Session): curl_cffi 会话对象。
        impersonate (str): 浏览器指纹标识。
        detail_url (str): 详情页绝对 URL。

    Returns:
        dict: parse_lot_detail 的返回结果。
    """
    log.debug(f"获取详情 {detail_url}")
    sel = _get_selector(log, session, impersonate, detail_url)
    return parse_lot_detail(sel)


def save_auction(log, sql_pool, auction):
    """把一场拍卖会写入 scp_auction_record（幂等，靠 event_id 唯一索引去重）。

    Args:
        log: logger 对象。
        sql_pool: MySQL 连接池；传 None 时不入库。
        auction (dict): parse_auction_list 产出的场次 dict。

    Returns:
        None: 仅入库，无返回值。
    """
    if sql_pool is None:
        return
    row = {k: auction.get(k) for k in (
        "event_id", "auction_name", "auction_type", "event_status",
        "start_time_raw", "end_time_raw", "start_time", "end_time",
        "auction_url", "catalog_url", "cover_img",
    )}
    sql_pool.insert_many(table="scp_auction_record", data_list=[row], ignore=True)


def crawl_one_auction(log, sql_pool, session, impersonate, auction):
    """抓取单个场次的全部 lot 列表并入库（阶段一：只抓列表，不进详情页）。

    两阶段设计：本函数只负责 lot 列表入库（state 默认 0），详情字段由后续
    update_details_for_pending 扫库 state != 1 的记录单独补抓。抓完把
    scp_auction_record.lots_state 置 1，标记该场列表已抓。

    Args:
        log: logger 对象。
        sql_pool: MySQL 连接池；传 None 时只返回数据，不入库。
        session (requests.Session): curl_cffi 会话对象。
        impersonate (str): 浏览器指纹标识。
        auction (dict): 场次 dict（含 event_id / auction_name / catalog_url）。

    Returns:
        list[dict]: 该场次全部 lot 列表数据（阶段一字段，不含详情）。
    """
    lots = fetch_auction_lots(log, session, impersonate, auction)

    if sql_pool is not None:
        if lots:
            sql_pool.insert_many(table="scp_lot_record", data_list=lots, ignore=True)
        # 标记该场 lot 列表已抓
        sql_pool.update_one_or_dict(
            table="scp_auction_record",
            data={"lots_state": 1},
            condition={"event_id": auction["event_id"]},
        )

    log.info(f"场次 {auction['event_id']}({auction.get('auction_name')}) 共抓 {len(lots)} 条 lot")
    return lots


def get_details(log, detail_url, sql_pool, sql_id):
    """对单条已入库 lot 补抓详情（阶段二），写回 scp_lot_record。

    Args:
        log: logger 对象。
        detail_url (str): 详情页 URL。
        sql_pool: MySQL 连接池。
        sql_id: 数据库记录 id。

    Returns:
        None: 仅入库，无返回值。
    """
    log.info(f">>> 补抓详情 {detail_url}")
    impersonate = random.choice(client_identifier_list)
    with requests.Session() as session:
        detail = fetch_lot_detail(log, session, impersonate, detail_url)

    # 详情字段 + 标记已抓；成交价/状态/出价以详情页为准（覆盖列表阶段的值）
    data = {**detail, "state": 1}
    sql_pool.update_one_or_dict(
        table="scp_lot_record",
        data=data,
        condition={"id": sql_id},
    )


def update_details_for_pending(log, sql_pool):
    """扫库里 state != 1 的 lot，逐条补抓详情。

    Args:
        log: logger 对象。
        sql_pool: MySQL 连接池。

    Returns:
        None: 仅入库，无返回值。
    """
    log.debug("Updating detail pages ...")
    rows = sql_pool.select_all(
        "select id, detail_url from scp_lot_record where state != 1"
    )
    for row in rows:
        sql_id, detail_url = row[0], 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="scp_lot_record",
                data={"state": 2},
                condition={"id": sql_id},
            )