spider_work/rea_spider/rea_core.py

# -*- coding: utf-8 -*-
# Author : Charley
# Python : 3.12.10
# Date   : 2026/07/10
"""
REA (collectrea.com) 公用模块：HTTP 配置、场次列表解析、场次内 lot 分页抓取、详情页解析。
被 rea_history.py / rea_spider.py 复用。

目标网站: https://collectrea.com/search
逻辑要点:
    1. 站点为 Laravel + Livewire + Alpine.js，页面服务端渲染（SSR），无独立数据 API。
       curl_cffi 拿到的是「Alpine 未执行」的原始 HTML，解析选择器均按原始 HTML 校准。
    2. /search 页三个 tab 面板按文档顺序排列（role="tabpanel"）：
         面板1 = RECENT AUCTIONS（最近 16 场，Swiper 封面卡）
         面板2 = AUCTION ARCHIVE（历史归档 68 场，按年份分组文字链）
         面板3 = PAST CATALOG（往期图录，<select>，不属于本爬虫范围）
       两个 archives 面板的链接形如 /archives/{year}/{month}/，即一个「场次」。
    3. 场次页 /archives/{year}/{month}/ 是 Livewire 搜索组件，预置过滤该场次。
       翻页直接用查询参数：?page=N&pageSize=M（Livewire WithPagination 同步到 URL），
       超出末页返回空列表 → 作为停止条件。列表卡片只有 lot_id/lot_number/slug/封面图。
    4. lot 详情页 /archives/{year}/{Season}/{lotNumber}/{slug}：
         标题在 <h1>（排除 "... - Item detail" 面包屑那条）；
         Sold For / Year / Auction / Lot # / Category 在 <dl> 的 dt/dd；
         多图为 rea-image-archive CDN 上的 -N.jpg。
"""
import re
import random
from loguru import logger
from parsel import Selector
from curl_cffi import requests
from curl_cffi.requests import BrowserType
from urllib.parse import urljoin, quote
from tenacity import retry, stop_after_attempt, wait_fixed

# —— 站点常量 ——
BASE_URL = "https://collectrea.com"  # 站点根，用于拼接相对链接
SEARCH_URL = "https://collectrea.com/search"  # 场次列表页（含三个 tab）
PAGE_SIZE = 200  # 场次内 lot 每页条数（Livewire pageSize 参数，实测最高支持 1000）；逐分类抓取时用较大值减少请求数
IMG_CDN_PREFIX = "https://rea-image-archive.nyc3.cdn.digitaloceanspaces.com"  # 拍品图 CDN 前缀

# 直接用库内置的所有浏览器指纹（应对 Cloudflare 的 TLS 指纹校验）
client_identifier_list = [b.value for b in BrowserType]

# 不写死 user-agent：curl_cffi 会按 impersonate 指纹自动带「与 JA3 匹配的 UA」，
# 若在此写死 UA 会覆盖它，造成 TLS 指纹与 UA 头矛盾，反而更易被 Cloudflare 识别。
# 这里只保留通用、不与指纹冲突的头。
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 框架自动传入。
    """
    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 参数都走这里。

    默认直连（返回 None）。站点走 Cloudflare，实测 curl_cffi + 浏览器指纹直连可用；
    若后续被限流/封禁，在此按 {"http": ..., "https": ...} 填入代理即可，函数已带重试。

    Args:
        log: logger 对象。

    Returns:
        dict | None: requests 风格代理字典；直连时返回 None。

    Raises:
        Exception: 透传内部异常以便 tenacity 触发重试。
    """
    # 住宅ip池 北美
    http_proxy = "http://u1952150085001297:sJMHl4qc4bM0@proxy.123proxy.cn:36927"
    https_proxy = "http://u1952150085001297:sJMHl4qc4bM0@proxy.123proxy.cn:36927"

    # url = "https://ifconfig.me"
    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_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,
                       # timeout=20)
                       proxies=get_proxys(log), timeout=20)
    resp.raise_for_status()
    return Selector(resp.text)


def _auction_from_href(href):
    """把场次相对链接解析成 {auction_key, auction_name, url}。

    Args:
        href (str): 形如 /archives/2025/spring/ 的相对链接。

    Returns:
        dict | None: {"auction_key": "2025/spring", "auction_name": "Spring 2025",
            "url": "https://collectrea.com/archives/2025/spring/"}；非法链接返回 None。
    """
    parts = [p for p in href.strip("/").split("/") if p]  # ['archives','2025','spring']
    if len(parts) < 3 or parts[0] != "archives":
        return None
    year, month = parts[1], parts[2]
    return {
        "auction_key": f"{year}/{month}",  # 唯一标识（等价 wheatland 的 auction_id）
        "auction_name": f"{month.title()} {year}",  # 展示名，如 "Spring 2025"
        "url": urljoin(BASE_URL, href if href.endswith("/") else href + "/"),
    }


def parse_auction_list(selector, only_recent=False):
    """解析 /search 页的场次列表。

    页面三个 role="tabpanel" 面板按文档顺序排列，含 archives 链接的两个面板依次为
    「RECENT（最近）」「AUCTION ARCHIVE（历史）」。only_recent=True 时只取最近面板。

    Args:
        selector (Selector): /search 页 GET 响应的 parsel 解析对象。
        only_recent (bool, optional): 是否只取「最近」面板。Defaults to False（取全部）。

    Returns:
        list[dict]: 每个元素为 _auction_from_href 返回的场次 dict，已按 auction_key 去重。

    Raises:
        ValueError: 页面找不到任何含 archives 链接的 tab 面板时抛出（说明响应异常）。
    """
    # 只取「叶子」tabpanel：原始 HTML 最外层有个包裹用的 <section role="tabpanel">，
    # 它嵌套了全部内层面板；用 not(.//*[@role="tabpanel"]) 排除外壳，得到真实的 3 个面板。
    panels = selector.xpath('//*[@role="tabpanel" and not(.//*[@role="tabpanel"])]')
    # 收集「含 archives 链接」的面板，保持文档顺序：[0]=最近，[1]=历史
    link_panels = []
    for p in panels:
        hrefs = [h for h in p.css('a::attr(href)').getall() if "/archives/" in h]
        if hrefs:
            link_panels.append(hrefs)

    if not link_panels:
        raise ValueError("找不到任何含 archives 链接的 tab 面板，页面结构可能已变更")

    target_panels = link_panels[:1] if only_recent else link_panels

    result, seen = [], set()
    for hrefs in target_panels:
        for href in hrefs:
            auc = _auction_from_href(href)
            if auc and auc["auction_key"] not in seen:
                seen.add(auc["auction_key"])
                result.append(auc)
    return result


@retry(stop=stop_after_attempt(3), wait=wait_fixed(2), after=after_log)
def get_auction_list(log, session, impersonate, only_recent=False):
    """GET /search 首页，解析出场次列表。

    Args:
        log: logger 对象。
        session (requests.Session): curl_cffi 会话对象。
        impersonate (str): 浏览器指纹标识。
        only_recent (bool, optional): True 仅取「最近」面板（增量用）；False 取全部（全量用）。
            Defaults to False。

    Returns:
        list[dict]: [{"auction_key": ..., "auction_name": ..., "url": ...}, ...]。
    """
    scope = "最近" if only_recent else "全部"
    log.info(f"获取{scope}场次列表")
    sel = _get_selector(log, session, impersonate, SEARCH_URL)
    auctions = parse_auction_list(sel, only_recent=only_recent)
    log.info(f"共解析到 {len(auctions)} 个场次：{[a['auction_key'] for a in auctions[:5]]}...")
    return auctions


def parse_lot_cards(selector, auction):
    """解析场次页单页的 lot 结果卡片。

    列表卡片信息极简：只有内部 lot_id（数字 wire:key）、详情链接（含 lot_number 与 slug）、
    封面图。标题/价格/明细均在详情页，由第二阶段补抓。

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

    Returns:
        list[dict]: 每条 lot 一个 dict，字段见 crawl_one_auction 说明；无卡片返回空列表。
    """
    # 只有 lot 卡片是「纯数字」wire:key；筛选器控件是 wire:key="item-..." / "Category" 等非数字，天然区分
    lot_ids = selector.re(r'wire:key="(\d+)"')

    # 详情链接：/archives/{year}/{Season}/{lotNumber}/{slug}
    cards = []
    for a in selector.css('a[href*="/archives/"]'):
        href = a.attrib.get("href", "")
        parts = [p for p in href.strip("/").split("/") if p]
        # ['archives', year, Season, lotNumber, slug...]
        if len(parts) < 5 or parts[0] != "archives" or not parts[3].isdigit():
            continue
        cards.append({
            "lot_number": parts[3],  # lot 编号
            "slug": "/".join(parts[4:]),  # slug（可能含斜杠，保险起见 join）
            "detail_url": urljoin(BASE_URL, href),  # 详情绝对 URL
        })

    # 数字 wire:key 与详情卡片按文档顺序 1:1 对应，逐条配对回填 lot_id
    rows = []
    for i, card in enumerate(cards):
        rows.append({
            "auction_key": auction["auction_key"],  # 场次唯一标识
            "auction_name": auction["auction_name"],  # 场次展示名
            "lot_id": lot_ids[i] if i < len(lot_ids) else "",  # 站内 lot 唯一 id（wire:key）
            **card,
        })
    return rows


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

    Args:
        log: logger 对象。
        session (requests.Session): curl_cffi 会话对象。
        impersonate (str): 浏览器指纹标识。
        auction_url (str): 场次页绝对 URL，如 https://collectrea.com/archives/2025/spring/。
        page (int): 页码，从 1 开始。
        extra_query (str, optional): 追加的查询串（不带前导 &），
            如 "Category[]=Football...&sortBy=Price:asc"。Defaults to ""。

    Returns:
        Selector: 该页响应的 parsel 解析对象。
    """
    url = f"{auction_url}?page={page}&pageSize={PAGE_SIZE}"
    if extra_query:
        url += f"&{extra_query}"
    return _get_selector(log, session, impersonate, url)


def parse_category_facet(selector):
    """从场次页左侧筛选栏解析 Category 分类列表及各分类的数量。

    用于绕过搜索结果 1000 条硬上限：逐个分类过滤抓取。

    Args:
        selector (Selector): 场次页任意一页的 parsel 解析对象（筛选栏恒渲染）。

    Returns:
        list[dict]: 每个元素为 {"value": 分类原值, "count": 该分类数量(int或None)}；
            无分类栏时返回空列表。
    """
    cats = []
    for inp in selector.css('input[name="Category[]"]'):
        value = (inp.attrib.get("value") or "").strip()  # parsel 已把 &amp; 还原为 &
        if not value:
            continue
        # 紧邻的 <label> 文本形如 "Football Cards and Memorabilia (152)"
        label = inp.xpath('./following-sibling::label[1]')
        text = label.xpath('normalize-space(.)').get() if label else ""
        m = re.search(r'\((\d+)\)\s*$', text or "")
        cats.append({"value": value, "count": int(m.group(1)) if m else None})
    return cats


def _collect_pages(log, session, impersonate, auction, extra_query, seen, all_lots):
    """按给定过滤条件从 page=1 翻到空页，去重后追加到 all_lots。

    Args:
        log: logger 对象。
        session (requests.Session): curl_cffi 会话对象。
        impersonate (str): 浏览器指纹标识。
        auction (dict): 场次 dict。
        extra_query (str): 追加查询串（不带前导 &），空串表示整场不过滤。
        seen (set): 跨分类共享的 detail_url 去重集合（原地更新）。
        all_lots (list): 累积结果列表（原地追加）。

    Returns:
        int: 本次新增（去重后）的 lot 条数。
    """
    page = 1
    added = 0
    while True:
        sel = _fetch_lot_page(log, session, impersonate, auction["url"], page, extra_query)
        lots = parse_lot_cards(sel, auction)
        if not lots:
            break  # 空页 = 已翻过末页（含 1000 上限触顶）
        new = [x for x in lots if x["detail_url"] not in seen]
        for x in new:
            seen.add(x["detail_url"])
        all_lots.extend(new)
        added += len(new)
        # 本页全是重复（分页异常/越界回卷兜底），停止
        if not new:
            break
        page += 1
    return added


def fetch_auction_lots(log, session, impersonate, auction):
    """抓取一个场次下的全部 lot 列表（绕过 1000 条搜索上限）。

    站点搜索结果有 1000 条硬上限（任何排序下深翻页最多取到前 1000）。为取全，
    改为逐个 Category 分类过滤抓取（各分类天然只属该场次）；某分类 >1000 时再分别
    用价格升序 + 降序各取 ≤1000 并去重，可覆盖到 2000。全程以 detail_url 去重。

    若某分类 >2000，升+降序仍无法全覆盖，会打 warning 提示漏采条数（当前站点单分类
    最大约 1100，暂无此情况）。若页面无分类栏，则退回整场翻页（受 1000 上限）。

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

    Returns:
        list[dict]: 该场次全部 lot 列表（阶段一字段，不含详情）。
    """
    log.info(f"开始抓取场次 {auction['auction_key']} 的 lot 列表")
    all_lots, seen = [], set()

    # 先取首页拿分类 facet（同时这一页也含 lot，但按分类重抓更全，故此处只用来取 facet）
    first_sel = _fetch_lot_page(log, session, impersonate, auction["url"], 1)
    cats = parse_category_facet(first_sel)

    if not cats:
        # 兜底：无分类栏，整场直接翻页（最多 1000）
        log.warning(f"{auction['auction_key']} 未解析到 Category 分类栏，退回整场翻页（受 1000 上限）")
        _collect_pages(log, session, impersonate, auction, "", seen, all_lots)
        log.info(f"场次 {auction['auction_key']} 共抓 {len(all_lots)} 条 lot")
        return all_lots

    log.info(f"{auction['auction_key']} 共 {len(cats)} 个分类，逐类抓取以绕过 1000 上限")
    for c in cats:
        value, count = c["value"], c["count"]
        if count == 0:
            continue
        cat_q = f"Category[]={quote(value, safe='')}"
        if count is not None and count > 1000:
            if count > 2000:
                log.warning(f"  分类[{value}] {count} 条 >2000，升+降序仍会漏约 {count - 2000} 条")
            for sort in ("Price:asc", "Price:desc"):
                n = _collect_pages(log, session, impersonate, auction,
                                   f"{cat_q}&sortBy={quote(sort, safe='')}", seen, all_lots)
                log.info(f"  分类[{value}] {sort} 新增 {n}（累计 {len(all_lots)}）")
        else:
            n = _collect_pages(log, session, impersonate, auction, cat_q, seen, all_lots)
            log.info(f"  分类[{value}](facet={count}) 新增 {n}（累计 {len(all_lots)}）")

    log.info(f"场次 {auction['auction_key']} 共抓 {len(all_lots)} 条 lot")
    return all_lots


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

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

    Returns:
        str: 去掉 $ 和千分位逗号后的数字串，如 "336000"；空值返回空串。
    """
    if not text:
        return ""
    return text.replace("$", "").replace(",", "").strip()


def parse_lot_detail(selector):
    """解析 lot 详情页，抽取标题、5 个明细字段、多图。

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

    Returns:
        dict: {
            "title": 标题, "sold_for": 成交价(纯数字串), "year": 年份,
            "auction": 场次名(如 "2025 Spring"), "lot_no": Lot 编号,
            "category": 分类, "imgs": 多图逗号拼接串
        }。
    """
    # 标题：页面有两个 h1，其一为 "xxx - Item detail" 面包屑，取另一条真实标题
    h1s = [t.strip() for t in selector.css('h1::text').getall() if t.strip()]
    real = [h for h in h1s if "Item detail" not in h]
    title = real[-1] if real else (h1s[-1] if h1s else "")

    # dt/dd 明细字段
    fields = {}
    for dt in selector.css('dt'):
        label = (dt.xpath('normalize-space(.)').get() or "").rstrip(":").strip()
        dd = dt.xpath('following-sibling::dd[1]')
        value = dd.xpath('normalize-space(.)').get() if dd else ""
        if label:
            fields[label] = value or ""

    # 多图：CDN 上的拍品图，去重并保持出现顺序
    imgs, seen = [], set()
    for u in selector.re(rf'{IMG_CDN_PREFIX}/[^"\'\s]+'):
        if u not in seen:
            seen.add(u)
            imgs.append(u)

    return {
        "title": title,
        "sold_for": _clean_price(fields.get("Sold For", "")),
        "year": fields.get("Year", ""),
        "auction": fields.get("Auction", ""),
        "lot_no": fields.get("Lot #", ""),
        "category": fields.get("Category", ""),
        "imgs": ",".join(imgs),
    }


@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 crawl_one_auction(log, sql_pool, session, impersonate, auction):
    """抓取单个场次的全部 lot 列表（阶段一：只抓列表，不进详情页）。

    与 wheatland 一致的两阶段设计：本函数只负责列表入库（state 默认 0），
    详情字段由后续 update_details_for_pending 扫库 state != 1 的记录单独补抓。

    入库字段（表 rea_record）：
        auction_key, auction_name, lot_id, lot_number, slug, detail_url,
        （以下由阶段二补写）title, sold_for, year, auction, lot_no, category, imgs, state。

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

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

    # 入库（state 默认 0 待补详情；detail_url 建唯一索引，ignore 去重）
    if sql_pool is not None and lots:
        sql_pool.insert_many(table="rea_record", data_list=lots, ignore=True)

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


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

    Args:
        log: logger 对象。
        detail_url (str): 详情页 URL。
        sql_pool: MySQL 连接池。
        sql_id: 数据库记录 id。
    """
    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}  # 详情 7 字段 + 标记已抓
    sql_pool.update_one_or_dict(
        table="rea_record",
        data=data,
        condition={"id": sql_id},
    )


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

    Args:
        log: logger 对象。
        sql_pool: MySQL 连接池。
    """
    log.debug("Updating detail pages ...")
    rows = sql_pool.select_all(
        "select id, detail_url from rea_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="rea_record",
                data={"state": 2},
                condition={"id": sql_id},
            )