spider_work/snkrdunk_jp_spider/snk_jp_spider.py

# -*- coding: utf-8 -*-
# Author : Charley
# Python : 3.12.10
# Date   : 2026/07/22
"""SNKRDUNK 日本站 爬虫（任务1 列表页 + 任务2 详情页/交易记录）。

任务1（列表页）：抓取搜索列表页中「带 SOLD 字样」的二手出品，存入表1 snkrdunk_jp_used_item。
任务2（详情页）：抓取单个详情页完整字段存入表1，并抓其交易记录（买卖历史）存入表2 snkrdunk_jp_trading_history。

接口说明（均为同源 GET、无需登录 / 无加密）：
  - 列表页： GET /search?...（Next.js SSR 页面，商品数据内嵌在 HTML 的 self.__next_f flight 里，非独立 JSON 接口）
  - 详情：   GET /v1/apparels/{apparel_id}/used/{used_item_id}
  - 交易记录：GET /v3/products/{product_id}/trading-history?range=all
             返回 trades[]，每条 {price, soldAt, title(品相), label(枚数)}。

关于「给交易记录配图」（需求3）：
  交易记录接口是平台聚合/匿名化数据，每条只有「价格/时间/品相/枚数」，本身不含图片和出品ID——
  这是设计使然。若日后需要配图，可复用现有管线 match_data 的思路：用 (product_id + 品相 + 价格)
  关联表1，取表1 的 primary_image_url 回填到表2（届时给表2 加一个 image_url 字段即可）。
"""
import re
import sys
import json
import time
import random
import requests
import schedule
import user_agent
from loguru import logger
from mysql_pool import MySQLConnectionPool
from datetime import datetime, timezone, timedelta
from tenacity import retry, stop_after_attempt, wait_fixed

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")

# 北京时区：UTC+8（与现有管线 trading_at 保持一致）
BEIJING_TZ = timezone(timedelta(hours=8))

# 表名
TABLE_USED_ITEM = "snkrdunk_jp_used_item"            # 表1：二手出品/商品信息
TABLE_TRADING_HISTORY = "snkrdunk_jp_trading_history"  # 表2：交易记录

# 列表页 URL 模板：{page} 处替换页码，其余筛选条件与给定的搜索 URL 完全一致
LIST_URL_TEMPLATE = (
    "https://snkrdunk.com/search?"
    "keywords=Pokemon+Card+Game+%E3%83%88%E3%83%AC%E3%82%AB+%28%E3%82%B7%E3%83%B3%E3%82%B0%E3%83%AB%E3%82%AB%E3%83%BC%E3%83%89%29"
    "&searchCategoryIds=6%2F33"
    "&brandIds=pokemon"
    "&sort=launch"
    "&itemConditions=like_new,minor_scratches,moderate_scratches,significant_damage,"
    "psa_10,psa_9,psa_8_below,bgs_10_black,bgs_10_gold,bgs_9_5,bgs_9_below,"
    "ars_10_plus,ars_10,ars_9,ars_8_below"
    "&page={page}"
)

# 翻页安全上限，防止解析异常导致死循环
MAX_PAGE = 500

headers = {
    "accept-language": "ja",
    "user-agent": user_agent.generate_user_agent(),
}


# ============================== 公共工具 ==============================
def after_log(retry_state):
    """tenacity retry 回调，统一打印重试日志。

    :param 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(1), after=after_log)
def get_proxys(log):
    """获取代理配置（快代理隧道）。

    :param log: 日志对象。
    :return: 代理字典 {"http": ..., "https": ...}。
    """
    tunnel = "x371.kdltps.com:15818"
    kdl_username = "t13753103189895"
    kdl_password = "o0yefv6z"
    try:
        proxies = {
            "http": "http://%(user)s:%(pwd)s@%(proxy)s/" % {"user": kdl_username, "pwd": kdl_password, "proxy": tunnel},
            "https": "http://%(user)s:%(pwd)s@%(proxy)s/" % {"user": kdl_username, "pwd": kdl_password, "proxy": tunnel}
        }
        return proxies
    except Exception as e:
        log.error(f"Error getting proxy: {e}")
        raise e


def utc_to_beijing(utc_str):
    """将带 Z 的 UTC 时间字符串转换成北京时间字符串。

    接口返回的 soldAt / releasedAt 形如 "2026-07-21T23:16:09Z"，结尾 Z 表示 UTC，
    北京时间为 UTC+8。

    :param utc_str: UTC 时间字符串，格式 "%Y-%m-%dT%H:%M:%SZ"。
    :return: 北京时间字符串，格式 "%Y-%m-%d %H:%M:%S"；入参为空或解析失败返回 None。
    """
    if not utc_str:
        return None
    try:
        dt_utc = datetime.strptime(utc_str, "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=timezone.utc)
        return dt_utc.astimezone(BEIJING_TZ).strftime("%Y-%m-%d %H:%M:%S")
    except Exception:
        return None


def upsert_used_items(log, sql_pool, data_list):
    """批量 upsert 到表1（按 apparel_id + used_item_id 唯一键去重更新）。

    使用 INSERT ... ON DUPLICATE KEY UPDATE，且更新时用 COALESCE(VALUES(col), col)，
    保证列表页带来的空字段不会覆盖掉详情页已写入的更完整数据（反之亦然）。
    同一次调用的所有 dict 必须拥有相同的 key。

    :param log: 日志对象。
    :param sql_pool: 数据库连接池 MySQLConnectionPool。
    :param data_list: list[dict]，表1 入库字典列表。
    :return: int，影响行数；无数据返回 0。
    """
    if not data_list:
        return 0

    cols = list(data_list[0].keys())
    key_cols = {"apparel_id", "used_item_id"}  # 唯一键列不参与 UPDATE
    col_sql = ", ".join(f"`{c}`" for c in cols)
    placeholder = "(" + ", ".join(["%s"] * len(cols)) + ")"  # 单行占位，executemany 自动合并多行
    update_sql = ", ".join(
        f"`{c}`=COALESCE(VALUES(`{c}`), `{c}`)" for c in cols if c not in key_cols
    )
    sql = (
        f"INSERT INTO `{TABLE_USED_ITEM}` ({col_sql}) VALUES {placeholder} "
        f"ON DUPLICATE KEY UPDATE {update_sql}"
    )
    # 用封装的 insert_many（query + args_list 模式，内部 executemany，
    # pymysql 会识别并保留 ON DUPLICATE KEY UPDATE 尾巴、自动合并多行）
    args_list = [tuple(d.get(c) for c in cols) for d in data_list]
    total = sql_pool.insert_many(query=sql, args_list=args_list)
    log.info(f"upsert 表1 完成，影响 {total} 行")
    return total


# ============================== 任务1：列表页 ==============================
def _extract_flight_cards(html):
    """从列表页 HTML 的 flight 数据中抽取所有商品卡片对象。

    Next.js 把服务端渲染的数据分片写在若干 self.__next_f.push([1,"..."]) 里，
    这里先把这些分片的字符串拼接还原成完整 flight 文本，再用「花括号配平」的方式
    逐个提取以 {"displayCardPattern" 开头的卡片 JSON 对象。

    :param html: 列表页 HTML 源码。
    :return: list[dict]，每个元素为一张卡片对象；解析失败返回空列表。
    """
    # 1) 抽取并拼接所有 flight 分片字符串
    parts = re.findall(r'self\.__next_f\.push\(\[1,"((?:[^"\\]|\\.)*)"\]\)', html)
    flight = ""
    for p in parts:
        try:
            # 分片本身是 JS 字符串字面量，用 JSON 解码还原转义
            flight += json.loads('"' + p + '"')
        except Exception:
            continue

    # 2) 花括号配平，逐个抠出卡片对象
    cards = []
    marker = '{"displayCardPattern"'
    start = flight.find(marker)
    while start != -1:
        depth = 0
        in_str = False
        escape = False
        end = -1
        for i in range(start, len(flight)):
            ch = flight[i]
            if escape:
                escape = False
                continue
            if ch == '\\':
                escape = True
                continue
            if ch == '"':
                in_str = not in_str
                continue
            if in_str:
                continue
            if ch == '{':
                depth += 1
            elif ch == '}':
                depth -= 1
                if depth == 0:
                    end = i
                    break
        if end == -1:
            break
        obj_str = flight[start:end + 1]
        try:
            cards.append(json.loads(obj_str))
        except Exception:
            pass
        start = flight.find(marker, end + 1)

    return cards


def _parse_link_ids(link):
    """从卡片 link 中解析 apparel_id 与 used_item_id。

    :param link: 详情页链接，如 https://snkrdunk.com/apparels/835482/used/47705323。
    :return: tuple(apparel_id:int|None, used_item_id:int|None)；解析失败返回 (None, None)。
    """
    m = re.search(r'/apparels/(\d+)/used/(\d+)', link or "")
    if not m:
        return None, None
    return int(m.group(1)), int(m.group(2))


@retry(stop=stop_after_attempt(5), wait=wait_fixed(1), after=after_log)
def get_single_page(log, page):
    """获取并解析单页列表数据，仅返回「已售」卡片。

    :param log: 日志对象。
    :param page: 页码。
    :return: tuple(sold_list:list[dict], total_cards:int)。
             sold_list 为已售卡片解析后的入库字典列表；total_cards 为该页卡片总数
             （用于判断是否已翻到底）。
    """
    log.info(f"获取列表第 {page} 页数据....................................................")
    url = LIST_URL_TEMPLATE.format(page=page)
    response = requests.get(url, headers={**headers, "accept": "text/html"},
                            proxies=get_proxys(log), timeout=22)
    response.raise_for_status()

    cards = _extract_flight_cards(response.text)

    sold_list = []
    for card in cards:
        pattern = card.get("displayCardPattern", "")
        # 只保留带 SOLD 字样（displayCardPattern 含 SoldOut）的已售卡片
        if "SoldOut" not in pattern:
            continue

        link = card.get("link", "")
        apparel_id, used_item_id = _parse_link_ids(link)
        if not apparel_id or not used_item_id:
            log.warning(f"卡片 link 无法解析 id，跳过：{link}")
            continue

        # 去掉图片尺寸参数，保留原图地址，如 ...jpeg?size=m -> ...jpeg
        image_url = (card.get("imageUrl") or "").split("?")[0] or None

        sold_list.append({
            "apparel_id": apparel_id,
            "used_item_id": used_item_id,
            "name_ja": card.get("title"),
            "condition_grade": card.get("condition"),
            "price": card.get("salePrice"),
            "is_sold": 1,
            "primary_image_url": image_url,
            "detail_url": link,
            "data_source": "list",
        })

    log.info(f"第 {page} 页共 {len(cards)} 张卡片，其中已售 {len(sold_list)} 张")
    return sold_list, len(cards)


def get_list_data(log, sql_pool):
    """翻页抓取整个列表，将所有已售卡片写入表1（自动翻到底）。

    :param log: 日志对象。
    :param sql_pool: 数据库连接池 MySQLConnectionPool。
    """
    page = 1
    total_sold = 0
    while page <= MAX_PAGE:
        try:
            sold_list, total_cards = get_single_page(log, page)
        except Exception as e:
            log.error(f"获取列表第 {page} 页出错：{e}")
            break

        # 该页无任何卡片，说明已翻到底
        if total_cards == 0:
            log.info(f"列表第 {page} 页无数据，翻页结束")
            break

        try:
            upsert_used_items(log, sql_pool, sold_list)
            total_sold += len(sold_list)
        except Exception as e:
            log.error(f"第 {page} 页数据入库出错：{e}")

        page += 1
        time.sleep(random.uniform(0.3, 1.2))

    log.info(f"列表抓取结束，累计入库已售卡片 {total_sold} 张")


@retry(stop=stop_after_attempt(100), wait=wait_fixed(3600), after=after_log)
def list_main(log):
    """列表页爬虫主函数（任务1 入口）。

    :param log: 日志对象。
    """
    log.info(f'开始运行 {sys._getframe().f_code.co_name} 爬虫任务....................................................')

    sql_pool = MySQLConnectionPool(log=log)
    if not sql_pool.check_pool_health():
        log.error("数据库连接池异常")
        raise RuntimeError("数据库连接池异常")

    try:
        get_list_data(log, sql_pool)
    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} 运行结束,等待下一轮的采集任务............')


# ============================== 任务2：详情页 + 交易记录 ==============================
@retry(stop=stop_after_attempt(5), wait=wait_fixed(1), after=after_log)
def get_detail(log, apparel_id, used_item_id):
    """抓取单个二手出品的详情 JSON。

    :param log: 日志对象。
    :param apparel_id: 商品(款式)ID，URL 中 /apparels/{apparel_id}。
    :param used_item_id: 二手出品ID，URL 中 /used/{used_item_id}。
    :return: dict，详情接口返回的完整 JSON；请求失败抛出异常。
    """
    log.info(f"获取详情数据 -> apparel={apparel_id}, used={used_item_id} .............................")
    url = f"https://snkrdunk.com/v1/apparels/{apparel_id}/used/{used_item_id}"
    response = requests.get(url, headers={**headers, "accept": "application/json"},
                            proxies=get_proxys(log), timeout=22)
    response.raise_for_status()
    return response.json()


def parse_detail(log, detail_json, apparel_id, used_item_id):
    """把详情 JSON 解析成表1 入库字典。

    :param log: 日志对象。
    :param detail_json: get_detail 返回的完整 JSON。
    :param apparel_id: 商品(款式)ID。
    :param used_item_id: 二手出品ID。
    :return: tuple(data_dict:dict, product_id:int|None)。
             data_dict 为表1 入库字典；product_id 供后续抓交易记录使用。
    """
    item = detail_json.get("apparelUsedItem", {}) or {}
    apparel = item.get("apparel", {}) or {}

    # 品牌（取第一个）
    brands = apparel.get("brands") or []
    brand_id = brands[0].get("id") if brands else None
    brand_name = brands[0].get("localizedName") if brands else None

    # 类目（多个时用 / 连接，取日文名）
    categories = apparel.get("categories") or []
    category_name = " / ".join(c.get("localizedName", "") for c in categories) if categories else None

    # 主图去掉尺寸参数
    primary_image_url = ((item.get("primaryPhoto") or {}).get("imageUrl") or "").split("?")[0] or None

    # 全部图片 URL，多个用英文逗号拼接
    image_urls = item.get("imageUrls") or []
    image_urls_str = ",".join(image_urls) if image_urls else None

    # 是否已售：优先用 isDisplaySold，其次用状态码 4=取引完了
    is_sold = 1 if item.get("isDisplaySold") or item.get("status") == 4 else 0

    product_id = apparel.get("productId") or None

    data_dict = {
        "apparel_id": apparel_id,
        "used_item_id": used_item_id,
        "product_id": product_id,
        "product_number": apparel.get("productNumber") or None,
        "name_ja": apparel.get("localizedName") or None,
        "name_en": apparel.get("name") or None,
        "brand_id": brand_id,
        "brand_name": brand_name,
        "category_name": category_name,
        "quantity_text": (item.get("size") or {}).get("localizedName") or None,
        "condition_grade": item.get("displayShortConditionTitle") or None,
        "condition_desc": item.get("displayWearCount") or None,
        "price": item.get("price"),
        "sale_status": item.get("status"),
        "sale_status_text": item.get("statusText") or None,
        "is_sold": is_sold,
        "primary_image_url": primary_image_url,
        "image_urls": image_urls_str,
        "detail_url": f"https://snkrdunk.com/apparels/{apparel_id}/used/{used_item_id}",
        "released_at": utc_to_beijing(apparel.get("releasedAt")),
        "data_source": "detail",
    }
    return data_dict, product_id


@retry(stop=stop_after_attempt(5), wait=wait_fixed(1), after=after_log)
def get_trading_history(log, product_id):
    """抓取某商品的完整交易记录（买卖历史）。

    :param log: 日志对象。
    :param product_id: 商品目录ID(catalog)，来自详情 apparel.productId。
    :return: list[dict]，交易记录原始 trades 列表；请求失败抛出异常。
    """
    log.info(f"获取交易记录 -> product_id={product_id} .............................")
    url = f"https://snkrdunk.com/v3/products/{product_id}/trading-history"
    params = {"range": "all"}
    response = requests.get(url, headers={**headers, "accept": "application/json"},
                            params=params, proxies=get_proxys(log), timeout=22)
    response.raise_for_status()
    return response.json().get("trades", []) or []


def save_trading_history(log, sql_pool, product_id, apparel_id, product_number, trades):
    """把交易记录写入表2，采用「联合唯一索引 + INSERT IGNORE」累积去重。

    背景：
      - 接口硬性截断到最近 20 条，不能删旧再插（会丢历史），必须**跨天累积**。
      - 表2 有联合唯一键 (product_id, sold_at, sold_price, condition_grade, quantity_text,
        trade_index_in_second)，配合 INSERT IGNORE 保证跨天抓到的重复记录自动跳过、
        新记录正常入库。

    trade_index_in_second 计算：
      同一个 (sold_at + 价格 + 品相 + 枚数) 签名可能对应多笔真实成交（如 6 笔同秒
      ¥1000 的 B 品 1枚），按接口返回顺序编号 0/1/2/... 作为位序，才能把它们全部保留。
      前提是接口对 trades 的排序稳定（同 soldAt 内相对顺序不变），实测方案。

    :param log: 日志对象。
    :param sql_pool: 数据库连接池 MySQLConnectionPool。
    :param product_id: 商品目录ID。
    :param apparel_id: 关联的商品(款式)ID，冗余存储便于关联表1。
    :param product_number: 品番，冗余存储。
    :param trades: get_trading_history 返回的 trades 列表（按接口原始顺序传入）。
    :return: int，本次实际新增行数（已存在的靠 INSERT IGNORE 跳过、不计入）。
    """
    if not trades:
        log.warning(f"product_id={product_id} 无交易记录")
        return 0

    sig_counter = {}  # 记录每个签名出现过几次，用作 trade_index_in_second
    data_list = []
    for tr in trades:
        sold_at = utc_to_beijing(tr.get("soldAt"))
        sold_price = tr.get("price")
        condition_grade = tr.get("title")   # 接口 title 即品相 A/B/C...
        quantity_text = tr.get("label")     # 接口 label 即枚数 1枚/2枚...
        sig = (sold_at, sold_price, condition_grade, quantity_text)
        idx = sig_counter.get(sig, 0)
        sig_counter[sig] = idx + 1
        data_list.append({
            "product_id": product_id,
            "apparel_id": apparel_id,
            "product_number": product_number,
            "sold_price": sold_price,
            "condition_grade": condition_grade,
            "quantity_text": quantity_text,
            "sold_at": sold_at,
            "trade_index_in_second": idx,
        })

    new_rows = sql_pool.insert_many(table=TABLE_TRADING_HISTORY, data_list=data_list, ignore=True)
    log.info(f"交易记录入库 -> product_id={product_id}，接口 {len(trades)} 条，本次新增 {new_rows} 条（重复的已跳过）")
    return new_rows


def crawl_detail_and_history(log, sql_pool, apparel_id, used_item_id, seen_product_ids=None):
    """抓取单个详情页：详情入表1 + 交易记录入表2（任务2 单条流程）。

    详情(表1) 每次都要抓（每个出品各自独立）；但交易记录(表2) 是按 product_id 聚合的，
    同一 product_id 每次 run 只需抓一次——传入 seen_product_ids 集合可实现批量模式下
    的自动去重（避免同款卡的多个出品重复调交易记录接口）。

    :param log: 日志对象。
    :param sql_pool: 数据库连接池 MySQLConnectionPool。
    :param apparel_id: 商品(款式)ID。
    :param used_item_id: 二手出品ID。
    :param seen_product_ids: 可选，本次 run 已抓过交易记录的 product_id 集合；命中则跳过。
                             为 None 时不做去重（单条模式用）。
    """
    # 1) 详情 -> 表1
    detail_json = get_detail(log, apparel_id, used_item_id)
    data_dict, product_id = parse_detail(log, detail_json, apparel_id, used_item_id)
    upsert_used_items(log, sql_pool, [data_dict])

    # 2) 交易记录 -> 表2
    if not product_id:
        log.warning(f"apparel={apparel_id} 未取到 product_id，跳过交易记录")
        return

    # 同一 product_id 每次 run 只抓一次交易记录（同款卡的多个出品共用同一份买卖历史）
    if seen_product_ids is not None:
        if product_id in seen_product_ids:
            log.info(f"product_id={product_id} 本次已抓过交易记录，跳过")
            return
        seen_product_ids.add(product_id)

    try:
        trades = get_trading_history(log, product_id)
        save_trading_history(log, sql_pool, product_id, apparel_id, data_dict.get("product_number"), trades)
    except Exception as e:
        log.error(f"抓取交易记录出错 product_id={product_id}: {e}")


def detail_main(log, apparel_id=None, used_item_id=None):
    """详情+交易记录爬虫主函数（任务2 入口）。

    传入 apparel_id / used_item_id 时只抓该条；不传时从表1 里筛出列表页抓来的、
    尚未补全详情的已售项（data_source='list'）逐个补全详情与交易记录。

    :param log: 日志对象。
    :param apparel_id: 可选，指定要抓的商品(款式)ID。
    :param used_item_id: 可选，指定要抓的二手出品ID。
    """
    log.info(f'开始运行 {sys._getframe().f_code.co_name} 爬虫任务....................................................')

    sql_pool = MySQLConnectionPool(log=log)
    if not sql_pool.check_pool_health():
        log.error("数据库连接池异常")
        raise RuntimeError("数据库连接池异常")

    # 模式一：指定单条
    if apparel_id and used_item_id:
        try:
            crawl_detail_and_history(log, sql_pool, apparel_id, used_item_id)
        except Exception as e:
            log.error(f'抓取详情 apparel={apparel_id} used={used_item_id} 异常: {e}')
        return

    # 模式二：批量补全表1 中列表页抓来的记录
    rows = sql_pool.select_all(
        f"SELECT apparel_id, used_item_id FROM `{TABLE_USED_ITEM}` "
        f"WHERE data_source = 'list' ORDER BY id"
    )
    log.info(f"待补全详情的记录数：{len(rows)}")
    seen_product_ids = set()  # 本次 run 已抓过交易记录的 product_id，避免重复调接口
    for _apparel_id, _used_item_id in rows:
        try:
            crawl_detail_and_history(log, sql_pool, _apparel_id, _used_item_id, seen_product_ids)
        except Exception as e:
            log.error(f'抓取详情 apparel={_apparel_id} used={_used_item_id} 异常: {e}')
        time.sleep(random.uniform(0.3, 1.0))
    log.info(f"批量模式结束，共抓取 {len(seen_product_ids)} 个不同 product_id 的交易记录")

    log.info(f'爬虫程序 {sys._getframe().f_code.co_name} 运行结束............')


def schedule_task():
    """定时任务启动入口（每日 00:01 先跑列表，再补全详情+交易记录）。"""
    def daily_job():
        list_main(log=logger)      # 任务1：列表 -> 表1
        detail_main(logger)        # 任务2：批量补全详情 -> 表1 + 交易记录 -> 表2

    # daily_job()

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


if __name__ == '__main__':
    # 任务2 单条示例：主公给的详情页 https://snkrdunk.com/apparels/835482/used/47574548
    # detail_main(logger, apparel_id=835482, used_item_id=47574548)

    # 任务1：抓列表页所有已售 -> 表1
    # list_main(log=logger)

    # 任务2：批量补全表1 中列表页记录的详情与交易记录（先跑完 list_main 再放开）
    # detail_main(logger)

    # 定时任务（列表 + 详情/交易记录 每日跑一次）
    schedule_task()