spider_work/auctions90s_spider/auctions90s_core.py

# -*- coding: utf-8 -*-
# Author : Charley
# Python : 3.12.10
# Date   : 2026/5/28
"""
90sAuctions 公用模块：HTTP 配置、ASP.NET postback 切换 auction、单页解析、详情解析。
被 auctions90s_history.py / auctions90s_spider.py / auctions90s_retry.py 复用。
"""
import random
import re
from curl_cffi import requests
import user_agent
from loguru import logger
from parsel import Selector
from curl_cffi.requests import BrowserType
from tenacity import retry, stop_after_attempt, wait_fixed

# 站点常量
SITE_ORIGIN = "https://90sauctions.com"
GALLERY_URL = f"{SITE_ORIGIN}/Lots/Gallery"

# 数据库表名（结构复制自 lelands_record）
TABLE_NAME = "auctions90s_record"

# 直接用库内置的所有浏览器指纹
client_identifier_list = [b.value for b in BrowserType]

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

# 列表页价格前缀：进行中为 "CURRENT BID $..."，已结束为 "SOLD FOR $..."
PRICE_PREFIX_RE = re.compile(r'^\s*(?:SOLD\s+FOR|CURRENT\s+BID)\s*\$', re.IGNORECASE)


def after_log(retry_state):
    """tenacity 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):
    """
    获取代理。

    :param log: (loguru.Logger) 日志对象，用于记录代理获取异常
    :return: (dict) 形如 {"http": "...", "https": "..."} 的代理字典
    :raises Exception: 当代理服务不可达时由 tenacity 触发重试，最终抛出
    """
    http_proxy = "http://u1952150085001297:sJMHl4qc4bM0@proxy.123proxy.cn:36931"
    https_proxy = "http://u1952150085001297:sJMHl4qc4bM0@proxy.123proxy.cn:36931"
    try:
        return {"http": http_proxy, "https": https_proxy}
    except Exception as e:
        log.error(f"Error getting proxy: {e}")
        raise e


def extract_auction(description, log=logger):
    """
    从详情页 lblOldAuction 的文本列表中提取拍卖会名称（双引号内的字符串）。

    :param description: (list[str]) selector.getall() 返回的字符串列表，
                        典型形如 ['Item was in Auction "Inaugural Auction",...']
    :param log: (loguru.Logger) 日志对象
    :return: (str | None) 提取到的 auction 名，失败或空时返回 None
    """
    try:
        if not description or not isinstance(description, list):
            return None
        for item in description:
            if not item or not isinstance(item, str):
                continue
            text = item.strip()
            if not text:
                continue
            match = re.search(r'"(.+?)"', text)
            if match:
                auction = match.group(1).strip()
                return auction if auction else None
        return None
    except Exception as e:
        log.error(f"extract_auction error: {e}")
        return None


def _pick_hidden(selector, field_id):
    """
    从 ASP.NET 页面中提取隐藏字段值（__VIEWSTATE / __EVENTVALIDATION 等）。

    :param selector: (parsel.Selector) 已解析的页面 Selector
    :param field_id: (str) 隐藏字段的 id，如 "__VIEWSTATE"
    :return: (str) 字段值，未找到时返回空字符串
    """
    return selector.xpath(f'//input[@id="{field_id}"]/@value').get() or ''


def parse_auction_list(selector):
    """
    解析 Gallery 页面侧边栏 Auction 下拉框中的全部选项。

    :param selector: (parsel.Selector) Gallery 首页 Selector
    :return: (list[dict]) [{"id": "-1", "name": "All Auctions"}, {"id": "12", "name": "June 2026"}, ...]
    """
    options = selector.xpath('//select[@id="Auction"]/option')
    result = []
    for opt in options:
        aid = opt.xpath('./@value').get()
        name = opt.xpath('./text()').get()
        if aid is None:
            continue
        result.append({"id": aid.strip(), "name": (name or '').strip()})
    return result


@retry(stop=stop_after_attempt(3), wait=wait_fixed(2), after=after_log)
def get_auction_list(log, session, impersonate):
    """
    GET Gallery 首页并解析出全部具体拍卖会（排除 -1 All Auctions）。

    :param log: (loguru.Logger) 日志对象
    :param session: (curl_cffi.requests.Session) 复用的会话对象
    :param impersonate: (str) curl_cffi 浏览器指纹标识
    :return: (list[dict]) [{"id": "12", "name": "June 2026"}, ...]
    :raises requests.HTTPError: 首页请求非 2xx 时抛出
    """
    log.info("获取全部拍卖会列表")
    resp = session.get(GALLERY_URL, headers=headers, impersonate=impersonate,
                       proxies=get_proxys(log), timeout=15)
    resp.raise_for_status()
    sel = Selector(resp.text)
    all_opts = parse_auction_list(sel)
    # 过滤掉 All Auctions(-1)，只保留具体拍卖会
    real = [o for o in all_opts if o["id"] != "-1"]
    log.info(f"共解析到 {len(real)} 个拍卖会：{[(o['id'], o['name']) for o in real]}")
    return real


@retry(stop=stop_after_attempt(5), wait=wait_fixed(2), after=after_log)
def setup_auction_session(log, session, impersonate, auction_id):
    """
    通过 ASP.NET __doPostBack 把 Gallery 的 Auction 筛选切换到指定 auction_id。
    切换后服务端 session 记住该选择，后续 GET /Lots/Gallery?page=N 都返回该 auction 数据。

    :param log: (loguru.Logger) 日志对象
    :param session: (curl_cffi.requests.Session) 复用的会话对象
    :param impersonate: (str) curl_cffi 浏览器指纹标识
    :param auction_id: (str) "-1"(All Auctions) 或具体 id 如 "12"
    :return: None
    :raises RuntimeError: 切换后页面中 selected option 与 auction_id 不一致时抛出
    """
    log.info(f"切换 Auction -> {auction_id}")
    proxies = get_proxys(log)

    # 1) 首次 GET 拿 ViewState
    resp = session.get(GALLERY_URL, headers=headers, impersonate=impersonate,
                       proxies=proxies, timeout=15)
    resp.raise_for_status()
    sel = Selector(resp.text)

    # 2) 构造 postback 表单。控件名前缀均为 ctl00$
    form_data = {
        '__EVENTTARGET': 'ctl00$Auction',
        '__EVENTARGUMENT': '',
        '__LASTFOCUS': '',
        '__VIEWSTATE': _pick_hidden(sel, '__VIEWSTATE'),
        '__VIEWSTATEGENERATOR': _pick_hidden(sel, '__VIEWSTATEGENERATOR'),
        '__EVENTVALIDATION': _pick_hidden(sel, '__EVENTVALIDATION'),
        'ctl00$SearchIn': 'title',
        'ctl00$SearchText': '',
        'ctl00$BrowseBy': 'gallery',
        'ctl00$Auction': str(auction_id),
    }

    post_headers = {
        **headers,
        'Content-Type': 'application/x-www-form-urlencoded',
        'Referer': GALLERY_URL,
        'Origin': SITE_ORIGIN,
    }

    resp = session.post(GALLERY_URL, headers=post_headers, data=form_data,
                        impersonate=impersonate, proxies=proxies, timeout=20)
    resp.raise_for_status()

    # 验证切换是否成功
    sel2 = Selector(resp.text)
    selected_val = sel2.xpath('//select[@id="Auction"]/option[@selected]/@value').get()
    log.info(f"切换后 Auction 选中值: {selected_val}")
    if selected_val != str(auction_id):
        raise RuntimeError(f"切换 Auction 失败，预期 {auction_id} 实际 {selected_val}")


def _clean_price(raw):
    """
    清洗列表页的成交/当前价文本，去掉前缀和千分位逗号。

    :param raw: (str | None) 形如 "SOLD FOR $1,850" / "CURRENT BID $325"
    :return: (str | None) 纯数字字符串（如 "1850" / "325"），输入为空时返回 None
    """
    if not raw:
        return None
    price = PRICE_PREFIX_RE.sub('', raw)
    return price.replace(',', '').strip() or None


@retry(stop=stop_after_attempt(5), wait=wait_fixed(2), after=after_log)
def get_single_page(log, page, sql_pool, session, impersonate,
                    auction_id=None, auction_name=None):
    """
    抓取并落库 Gallery 的单页数据。

    :param log: (loguru.Logger) 日志对象
    :param page: (int) 页码（从 1 开始）
    :param sql_pool: (MySQLConnectionPool) 数据库连接池，None 时不落库（调试用）
    :param session: (curl_cffi.requests.Session) 复用的会话对象（需已 setup_auction_session）
    :param impersonate: (str) curl_cffi 浏览器指纹标识
    :param auction_id: (str | None) 当前 session 切换到的 auction id，写入 auctions90s_record.auction_id
    :param auction_name: (str | None) 同上，写入 auctions90s_record.auction_name
    :return: (int) 本页解析到并落库的条数；无数据时返回 0
    """
    log.info(f">>>>>>>>>>>>>> 正在爬取 auction={auction_id}({auction_name}) 第 {page} 页 <<<<<<<<<<<<<<")

    response = session.get(GALLERY_URL, impersonate=impersonate, headers=headers,
                           params={"page": f"{page}"},
                           proxies=get_proxys(log), timeout=10, allow_redirects=False)
    response.raise_for_status()

    selector = Selector(response.text)
    tag_div_list = selector.xpath(
        '//div[@class="items"]/div/div[@class="row"]//div[@class="col-lg-3 col-md-4 col-sm-6"]')

    if not tag_div_list or len(tag_div_list) == 0:
        log.warning(f"--------------- 第 {page} 页无数据 ---------------")
        return 0

    info_list = []
    for tag_div in tag_div_list:
        # 商品标题与详情页绝对地址（列表里 a 标签已是完整 url，无需拼接）
        title = tag_div.xpath('.//p/a/text()').get()
        detail_url = tag_div.xpath('.//p/a/@href').get()

        # Bids / Opening Bid / Status 三个 strong 顺序固定
        tag_div_p = tag_div.xpath('.//div/p[2]/strong/text()').getall()
        bids = tag_div_p[0] if tag_div_p else None
        opening_bid = tag_div_p[1] if len(tag_div_p) > 1 else None
        opening_bid = opening_bid.replace('$', '').replace(',', '').strip() if opening_bid else None
        status = tag_div_p[2] if len(tag_div_p) > 2 else None

        # 价格：列表卡片底部 a 文本，进行中为 "CURRENT BID $..."，结束后为 "SOLD FOR $..."
        price = tag_div.xpath('.//div[@class="item-price"]/a/text()').get()
        price = _clean_price(price)

        data_dict = {
            "title": title,
            "detail_url": detail_url,
            "bids": bids,
            "opening_bid": opening_bid,
            "status": status,
            "price": price,
            "auction_id": int(auction_id) if auction_id is not None else None,
            "auction_name": auction_name,
        }
        info_list.append(data_dict)

    if info_list and sql_pool is not None:
        sql_pool.insert_many(table=TABLE_NAME, data_list=info_list, ignore=True)
    return len(info_list)


def crawl_one_auction(log, sql_pool, session, impersonate,
                      auction_id, auction_name, max_page=460):
    """
    抓取单个拍卖会的全部页（switch 到该 auction → 翻页直到无数据）。

    :param log: (loguru.Logger) 日志对象
    :param sql_pool: (MySQLConnectionPool) 数据库连接池
    :param session: (curl_cffi.requests.Session) 复用的会话对象
    :param impersonate: (str) curl_cffi 浏览器指纹标识
    :param auction_id: (str) 当前 session 切换到的 auction id
    :param auction_name: (str) 拍卖会名，写入数据库
    :param max_page: (int) 最大页码上限，作为兜底保护
    :return: (int) 该 auction 抓到的总条数
    """
    setup_auction_session(log, session, impersonate, auction_id)

    page = 1
    total = 0
    while page <= max_page:
        try:
            n = get_single_page(log, page, sql_pool, session, impersonate,
                                auction_id=auction_id, auction_name=auction_name)
        except Exception as e:
            log.error(f"auction={auction_id} page={page} 抓取失败: {e}")
            break
        if n == 0:
            log.info(f"auction={auction_id} 翻到第 {page} 页无数据，结束")
            break
        total += n
        page += 1
    log.info(f"auction={auction_id}({auction_name}) 共抓取 {total} 条")
    return total


@retry(stop=stop_after_attempt(5), wait=wait_fixed(2), after=after_log)
def get_details(log, url, sql_pool, sql_id):
    """
    获取详情页：分类、图片列表，写回数据库并把 state 置 1。

    :param log: (loguru.Logger) 日志对象
    :param url: (str) 详情页 URL（来自列表页 detail_url 字段）
    :param sql_pool: (MySQLConnectionPool) 数据库连接池
    :param sql_id: (int) 数据库记录主键 id
    :return: None
    :raises requests.HTTPError: 详情页非 2xx 时由 tenacity 重试，超限后抛出
    """
    log.info(f">>>>>>>>>>>>>> 正在爬取详情数据URL: {url} <<<<<<<<<<<<<<")
    response = requests.get(url, headers=headers,
                            impersonate=random.choice(client_identifier_list),
                            timeout=10, proxies=get_proxys(log))
    response.raise_for_status()
    selector = Selector(response.text)
    category = selector.xpath('//a[@id="MainContent_hCategory"]/text()').get()
    # 右侧主图 + 缩略图区。缩略图列表里包含主图链接，因此需要去重保序
    imgs = selector.xpath('//div[@class="col-md-5 col-sm-5"]//a[not(@id="Zoomer")]/@href').getall()
    imgs = list(dict.fromkeys(imgs)) if imgs else []
    imgs_str = ','.join(imgs) if imgs else None

    sql_pool.update_one_or_dict(
        table=TABLE_NAME,
        data={"category": category, "imgs": imgs_str, "state": 1},
        condition={"id": sql_id}
    )


def update_details_for_pending(log, sql_pool):
    """
    扫描库里 state != 1 的记录，逐条抓详情；详情失败置 state=2。

    :param log: (loguru.Logger) 日志对象
    :param sql_pool: (MySQLConnectionPool) 数据库连接池
    :return: None
    """
    log.debug('Updating detail pages ...........................')
    sql_result = sql_pool.select_all(f'select id, detail_url from {TABLE_NAME} where state != 1')
    for row in sql_result:
        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=TABLE_NAME,
                data={"state": 2},
                condition={"id": sql_id}
            )