spider_work/cxx_spider/super_vault_order_spider.py

# -*- coding: utf-8 -*-
# Author : Charley
# Python : 3.12.10
# Date   : 2026/08/21
"""超级仓库 SuperVault · 购买订单（真实买家）爬虫。

背景：
    已售统计报告 stats/super_vault_his_report.py 的「参与人数」原本用拆卡报告
    super_vault_report_record.user_name 去重近似——那只是「中卡用户」，偏低、也不是真实买家。
    order/simple/list 接口按「每笔购买订单」返回买家(username/avatar) + 下单时间 + 份数，
    用它可统计每个拼团商品的「真实购买人数」，补上报告缺的真实买家维度。

接口探测结论（2026/08/21 实测）：
    - URL：POST https://cxx.cardsvault.net/app/order/simple/list，body {"tid","pageNum","pageSize"}。
    - 免登录：不带 Authorization 也返回 200，故本脚本不带任何 token（登录态最小化）。
    - pageSize 被服务端锁死为 9：请求 100/200 均只回 9 条，无法加大。
    - 深翻页可用：可一路翻到末页（超尾页会被钳到最后一页），故能事后抓已售商品的全部订单。
    - tid == pid（与拆卡报告口径一致，即 super_vault_product_record.pid）。

去重/幂等策略（重要）：
    源接口无订单号，且 (user_name, order_time, count) 在源头就不唯一（热门团存在「多个匿名同秒各买
    1 份」等真实但同键的不同订单，实测大团约 0.1%）。故不建唯一键、不靠该组合去重——那会误删真实订单。
    改为按 pid「删旧 + 整体重灌」：每次抓完一个商品的全部订单后，先删该 pid 旧行再批量写入，
    幂等且不丢任何一笔。表上仅留 idx_pid 普通索引。

真实购买人数口径（留给报告层，本爬虫只如实存每一笔）：
    登录用户头像路径内嵌 uid（/avatar/{uid}/avatar_pic/...），可精确去重且识别复购；
    默认头像用户无 uid 但 user_name 为系统稳定 handle，也可按 user_name 去重；
    唯独「匿名」天生无法区分（多个匿名同名同默认头像）。参考统计 SQL：
        SELECT
          COUNT(*)                                         AS 订单笔数,
          SUM(`count`)                                     AS 总份数,
          COUNT(DISTINCT COALESCE(uid, user_name))         AS 去重身份数,
          SUM(is_anonymous)                                AS 匿名笔数,
          -- 真实购买人数 ≈ 非匿名去重身份 + 匿名笔数(每笔按1人,上界近似)
          COUNT(DISTINCT CASE WHEN is_anonymous=0 THEN COALESCE(uid, user_name) END)
            + SUM(is_anonymous)                            AS 真实购买人数近似
        FROM super_vault_order_record WHERE pid = %s;

运行方式：
    从项目根目录运行 python super_vault_order_spider.py（读根目录 application.yml）。
    首次为全量回填(984 个已售商品，数小时)，可断点续跑：每个商品抓完置 order_state=1，
    重跑只补 order_state 未完成(NULL/0/3)的商品；之后每天增量抓新增已售商品，成本很低。
"""
import os
import re
import sys
import time
import math
# 必须在 import mysql_pool 之前切到本文件所在目录（即项目根目录）：mysql_pool 在「导入期」
# 就按 cwd 相对路径读 application.yml（读不到会直接 UnboundLocalError）。
# 入口脚本已切过，这里是幂等的；单独跑本文件时则靠这行兜住任意启动目录。
ROOT_DIR = os.path.dirname(os.path.abspath(__file__))
os.chdir(ROOT_DIR)

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

# 日志统一写 logs/{日期}.log，sink 由入口脚本配置（见文件末尾 __main__）。
# 模块级绝不配置 sink：被 start_cxx_spider 等入口 import 时若在此 logger.remove()/add()，
# 会摘掉入口已装好的 sink（或反被后续 import 的模块摘掉），日志去向变成由 import 顺序决定。
#
# 日志目录优先取环境变量 CXX_LOG_DIR，缺省回落到项目根 logs/：
# 服务器的日志采集只读固定目录（如容器内 /app/logs），而项目代码未必就放在该目录的上一级，
# 故目录由部署环境指定、代码不写死；文件名固定 {time:YYYYMMDD}.log 以匹配采集端的格式要求。
LOG_DIR = os.environ.get("CXX_LOG_DIR") or os.path.join(ROOT_DIR, "logs")
LOG_PATH = os.path.join(LOG_DIR, "{time:YYYYMMDD}.log")

# 接口地址
ORDER_LIST_URL = "https://cxx.cardsvault.net/app/order/simple/list"

# 请求头：免登录，故不带 Authorization（登录态最小化）
HEADERS = {
    "User-Agent": "okhttp/4.9.0",
    "CXX-APP-API-VERSION": "V2",  # 必须添加
    "Content-Type": "application/json; charset=UTF-8",
}

PAGE_SIZE = 9          # 服务端锁死为 9，请求更大值无效，不要改
USE_PROXY = False      # 代理开关：默认直连，遇 IP 风控再置 True
PAGE_SLEEP = 0.2       # 每页请求间隔(秒)，限速避免触发风控
PRODUCT_SLEEP = 0.5    # 每个商品之间间隔(秒)
MAX_PRODUCTS_PER_RUN = 0  # 单轮最多处理的商品数，0=不限（断点续跑靠 order_state，全量可放心跑满）

# order_state 状态含义（写入 super_vault_product_record.order_state）
#   NULL       = 从未抓取
#   0(ACTIVE)  = 已抓但商品仍在售(未售罄)，订单还会增长，下轮继续重灌刷新
#   1(DONE)    = 已售罄/冻结，抓完不再重抓
#   2(EMPTY)   = 无购买记录
#   3(FAIL)    = 抓取失败/中断，下轮重试
STATE_ACTIVE = 0   # 在售未售罄，已抓但待刷新
STATE_DONE = 1     # 售罄冻结，已抓完
STATE_EMPTY = 2    # 无购买记录(total=0)
STATE_FAIL = 3     # 抓取失败/中断

# 从头像 url 提取用户 uid：形如 .../avatar/1444/avatar_pic/xxx.jpg → 1444；默认头像无此段
UID_PATTERN = re.compile(r"/avatar/(\d+)/")


def after_log(retry_state):
    """tenacity 重试回调，记录每次尝试的结果。

    Args:
        retry_state: tenacity 传入的 RetryCallState，其 args[0] 约定为日志对象。
    """
    log = retry_state.args[0] if retry_state.args else 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):
    """获取隧道代理配置。

    Args:
        log: 日志对象。

    Returns:
        dict: requests 可用的 proxies 字典。

    Raises:
        Exception: 组装代理配置异常时向上抛出以触发重试。
    """
    tunnel = "x371.kdltps.com:15818"
    kdl_username = "t13753103189895"
    kdl_password = "o0yefv6z"
    try:
        proxy = f"http://{kdl_username}:{kdl_password}@{tunnel}/"
        return {"http": proxy, "https": proxy}
    except Exception as e:
        log.error(f"Error getting proxy: {e}")
        raise e


@retry(stop=stop_after_attempt(5), wait=wait_fixed(1), after=after_log)
def get_order_single_page(log, pid, page_num):
    """获取某拼团商品单页的购买订单数据（带重试）。

    Args:
        log: 日志对象。
        pid (int): 拼团商品 id（接口里的 tid）。
        page_num (int): 页码，从 1 开始。

    Returns:
        dict | None: {"total": int, "current_page": int, "items": list}；业务码非 200 时返回 None。

    Raises:
        requests.HTTPError: HTTP 状态码非 2xx 时抛出以触发重试。
    """
    data = {"tid": str(pid), "pageNum": page_num, "pageSize": PAGE_SIZE}
    proxies = get_proxys(log) if USE_PROXY else None
    response = requests.post(ORDER_LIST_URL, headers=HEADERS, json=data, timeout=(5, 30), proxies=proxies)
    response.raise_for_status()

    result = response.json()
    if result.get("status") == 200:
        d = result.get("data", {}) or {}
        total = d.get("total", 0)
        current_page = d.get("pageNum", page_num)
        items = d.get("data", []) or []
        log.debug(f"pid={pid} 第 {current_page} 页 / 共 {total} 条，本页 {len(items)} 条")
        return {"total": total, "current_page": current_page, "items": items}

    log.error(f"pid={pid} 第 {page_num} 页 API 返回错误: {result.get('msg', '未知错误')}")
    return None


def parse_order_items(pid, items):
    """把接口返回的订单列表解析成待入库的 dict 列表。

    Args:
        pid (int): 拼团商品 id，写入每行。
        items (list[dict]): 接口 data.data，每元素含 userInfo/createTime/count。

    Returns:
        list[dict]: 每元素为 {pid, user_name, uid, is_anonymous, avatar, count, order_time}。
    """
    parsed = []
    for item in items:
        user_info = item.get("userInfo", {}) or {}
        user_name = user_info.get("username")
        avatar = user_info.get("avatar")

        # 从头像路径提取 uid（登录用户才有；默认头像/匿名为空）
        m = UID_PATTERN.search(avatar) if avatar else None
        uid = int(m.group(1)) if m else None

        parsed.append({
            "pid": pid,
            "user_name": user_name,
            "uid": uid,
            "is_anonymous": 1 if user_name == "匿名" else 0,  # 匿名天生无法区分，单独标记供报告层处理
            "avatar": avatar,
            "count": item.get("count"),          # 该笔购买份数
            "order_time": item.get("createTime"),  # 下单时间
        })
    return parsed


def crawl_product_orders(log, sql_pool, pid, frozen=True):
    """抓取单个拼团商品的全部购买订单并入库，返回落库条数。

    策略「攒齐全部页 → 删旧 → 整体重灌」：先逐页翻完 ceil(total/9) 页、把所有订单攒进内存，
    全部成功后再 DELETE 该 pid 旧行 + 批量插入。这样：
      - 天然幂等：重跑=整体替换，不会累加重复；
      - 不误删真实订单：不靠 (user,time,count) 唯一键去重，匿名同秒/同份数的不同订单都完整保留；
      - 失败安全：任一页多次重试仍失败即抛出，此时既未删也未插，旧数据/状态原样保留，交上层记 STATE_FAIL。
    源接口按 createTime 倒序，对已售罄(冻结)商品列表稳定，故一次翻页即为精确全量快照。

    终态标记：售罄(frozen=True)商品订单已冻结，标 STATE_DONE 不再重抓；仍在售(frozen=False)的商品
    订单会继续增长，标 STATE_ACTIVE 让下轮重灌刷新，直到售罄后被标 DONE。

    Args:
        log: 日志对象。
        sql_pool (MySQLConnectionPool): 数据库连接池。
        pid (int): 拼团商品 id。
        frozen (bool, optional): 该商品是否已售罄冻结。True 标 DONE，False 标 ACTIVE 待刷新。Defaults to True。

    Returns:
        int: 落库的订单条数；无购买记录返回 0。

    Raises:
        RuntimeError: 某页多次重试后仍返回空时抛出，触发上层记 STATE_FAIL。
    """
    # 第 1 页：拿 total 与本页数据
    first = get_order_single_page(log, pid, 1)
    if not first:
        raise RuntimeError(f"pid={pid} 第 1 页返回空")

    total = first["total"]
    if total == 0:
        # 售罄却无记录→标 EMPTY 不再抓；在售无记录→标 ACTIVE 待其产生订单
        empty_state = STATE_EMPTY if frozen else STATE_ACTIVE
        log.info(f"pid={pid} 无购买记录，置 order_state={empty_state}")
        sql_pool.update_one_or_dict(
            table="super_vault_product_record",
            data={"order_state": empty_state},
            condition={"pid": pid})
        return 0

    total_pages = math.ceil(total / PAGE_SIZE)
    log.info(f"pid={pid} 共 {total} 笔订单，{total_pages} 页，frozen={frozen}，开始抓取")

    # 攒齐所有页（先全部成功，再动数据库）
    all_items = parse_order_items(pid, first["items"])
    for page_num in range(2, total_pages + 1):
        time.sleep(PAGE_SLEEP)
        page = get_order_single_page(log, pid, page_num)
        if not page:
            raise RuntimeError(f"pid={pid} 第 {page_num} 页返回空")
        all_items.extend(parse_order_items(pid, page["items"]))

    # 全部页成功 → 删旧 + 整体重灌 + 标记状态（售罄=DONE 不再抓；在售=ACTIVE 待刷新）
    final_state = STATE_DONE if frozen else STATE_ACTIVE
    sql_pool.update_one_or_dict(
        query="DELETE FROM super_vault_order_record WHERE pid = %s",
        args=(pid,))
    if all_items:
        sql_pool.insert_many(table="super_vault_order_record", data_list=all_items)
    sql_pool.update_one_or_dict(
        table="super_vault_product_record",
        data={"order_state": final_state},
        condition={"pid": pid})
    log.info(f"pid={pid} 抓取完成，落库 {len(all_items)} 笔，order_state={final_state}")
    return len(all_items)


def get_pending_products(log, sql_pool, limit=0):
    """取待抓订单的已售商品列表（口径与已售报告一致），含「是否售罄冻结」标记。

    已售 = sold_count>0 且 status NOT IN(0,1,2,3)；只挑 order_state 未完成的
    （NULL/0=未抓或在售待刷新、3=失败重试），跳过 1=已抓完(售罄冻结)与 2=无记录；
    按成交时间倒序，新团优先补。frozen = sold_count>=count（售罄），决定抓完后标 DONE 还是 ACTIVE。

    Args:
        log: 日志对象。
        sql_pool (MySQLConnectionPool): 数据库连接池。
        limit (int, optional): 最多返回的商品数，0 表示不限。Defaults to 0。

    Returns:
        list[tuple[int, bool]]: 每元素为 (pid, frozen)，frozen 表示是否已售罄冻结。
    """
    sql = (
        "SELECT pid, (count > 0 AND sold_count >= count) AS frozen FROM super_vault_product_record "
        "WHERE sold_count > 0 AND status NOT IN (0, 1, 2, 3) "
        "AND (order_state IS NULL OR order_state NOT IN (1, 2)) "
        "ORDER BY COALESCE(completion_time, sold_time, sell_time) DESC")
    if limit and limit > 0:
        sql += f" LIMIT {int(limit)}"
    rows = sql_pool.select_all(sql) or []
    products = [(r[0], bool(r[1])) for r in rows]
    log.info(f"待抓订单的已售商品 {len(products)} 个")
    return products


@retry(stop=stop_after_attempt(100), wait=wait_fixed(3600), after=after_log)
def cxx_order_main(log, limit=0):
    """购买订单采集主函数（可断点续跑）。

    Args:
        log: 日志对象。
        limit (int, optional): 单轮最多处理的商品数，0=不限。Defaults to 0。

    Raises:
        RuntimeError: 数据库连接池异常时抛出以触发重试。
    """
    log.info(f"开始运行 {sys._getframe().f_code.co_name} 购买订单采集" + "." * 40)

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

    try:
        run_limit = limit or MAX_PRODUCTS_PER_RUN
        products = get_pending_products(log, sql_pool, run_limit)
        done, empty, fail = 0, 0, 0
        for idx, (pid, frozen) in enumerate(products, 1):
            log.info(f"[{idx}/{len(products)}] 处理 pid={pid} (frozen={frozen})")
            try:
                fetched = crawl_product_orders(log, sql_pool, pid, frozen)
                if fetched == 0:
                    empty += 1
                else:
                    done += 1
            except Exception as e:
                # 单个商品失败不拖垮整轮：记 order_state=3，下轮重试
                log.error(f"pid={pid} 抓取失败: {e}")
                fail += 1
                sql_pool.update_one_or_dict(
                    table="super_vault_product_record",
                    data={"order_state": STATE_FAIL},
                    condition={"pid": pid})
            time.sleep(PRODUCT_SLEEP)
        log.info(f"本轮结束：成功 {done}，无记录 {empty}，失败 {fail}，共处理 {len(products)}")
    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} 运行结束，等待下一轮" + "." * 20)


def schedule_task():
    """定时入口：启动即增量抓一次，并每天 02:00、08:00 各跑一次（08:00 为保险补跑）。

    三次触发均为增量：靠 order_state 只补未完成(NULL/0/3)的团，已抓完(1)/无记录(2)跳过；
    抓取按 pid「删旧重灌」天然幂等，schedule 单线程顺序执行不会并发重入，故重复跑安全、开销小。
    08:00 一轮用于补 02:00 失败(3)/漏抓的团、重刷在售(0)团，以及 02:00~08:00 间新售罄的团。
    """
    # cxx_order_main(logger)  # 启动即先增量跑一次，不必等到 02:00
    schedule.every().day.at("02:00").do(cxx_order_main, logger)
    schedule.every().day.at("08:00").do(cxx_order_main, logger)  # 保险：再补一轮
    while True:
        schedule.run_pending()
        time.sleep(1)


if __name__ == "__main__":
    # 独立运行才配置日志 sink：按天切分、保留 7 天，与入口脚本同格式同去向（根目录 logs/）
    logger.remove()
    # diagnose=False：关掉 traceback 的变量值标注，避免连库失败时把数据库密码写进日志
    logger.add(LOG_PATH, encoding='utf-8', rotation="00:00",
               format="[{time:YYYY-MM-DD HH:mm:ss.SSS}] {level} {message}",
               level="DEBUG", retention="7 day", backtrace=True, diagnose=False)
    # 全量回填（数小时，断点续跑）：直接 cxx_order_main(logger)
    # 小样本验证：cxx_order_main(logger, limit=5)
    # cxx_order_main(logger)
    schedule_task()