spider_work/cxx_spider/_backfill_gap.py

# -*- coding: utf-8 -*-
# Author : Charley
# 历史回补脚本：按 pid 直取 /detail 与 /report/list，补齐 vod/list 滚动窗口已滚出的历史数据。
# 可续跑：产品走 INSERT IGNORE，报告靠 report_state 门控（1/2=已完成，0/3=待处理）。
#
# 用法：
#   python _backfill_gap.py                  # 用默认区间 [1638, 2812]
#   python _backfill_gap.py 2800 3200        # 指定回补 pid 区间 [2800, 3200]
# 说明：区间内已存在的 pid 会被自动跳过，无效/不存在的 pid 会被识别并计入"无效"，不写库。
import os
import sys
import time
import requests

os.chdir(os.path.dirname(os.path.abspath(__file__)))
from mysql_pool import MySQLConnectionPool
from loguru import logger

logger.remove()
logger.add("./_backfill_gap.log", encoding="utf-8",
           format="[{time:YYYY-MM-DD HH:mm:ss}] {message}", level="INFO")
logger.add(sys.stdout, format="[{time:HH:mm:ss}] {message}", level="INFO")

# 回补区间默认值（覆盖 2026 年 3-8 月缺口）；可用命令行参数覆盖
DEFAULT_PID_MIN, DEFAULT_PID_MAX = 1638, 2812
DETAIL = "https://cxx.cardsvault.net/app/teamup/detail"
REPORT = "https://cxx.cardsvault.net/app/teamup/report/list"


def build_headers(token):
    """构造请求头（与线上爬虫一致的精简头）。

    Args:
        token (str): 鉴权 token，来自 super_vault_token 表。

    Returns:
        dict: requests 可用的 headers。
    """
    return {
        "User-Agent": "okhttp/4.9.0",
        "Authorization": token,
        "CXX-APP-API-VERSION": "V2",
        "Content-Type": "application/json; charset=UTF-8",
    }


def req(method, url, headers, retries=4, **kw):
    """带重试的请求，返回解析后的 JSON。

    Args:
        method (str): "GET" 或 "POST"。
        url (str): 请求地址。
        headers (dict): 请求头。
        retries (int, optional): 重试次数。Defaults to 4。

    Returns:
        dict | None: 响应 JSON；多次失败返回 None。
    """
    for i in range(retries):
        try:
            r = requests.request(method, url, headers=headers, timeout=22, **kw)
            r.raise_for_status()
            return r.json()
        except Exception as e:
            if i == retries - 1:
                logger.warning(f"请求失败 {url} {kw.get('params') or kw.get('json')}: {e}")
                return None
            time.sleep(1)
    return None


def map_detail_to_row(d):
    """把 /detail 的 data 映射为 super_vault_product_record 的一行。

    Args:
        d (dict): detail 接口返回的 data 字段。

    Returns:
        dict: 产品行；从 liveInfo 直接取 live_id / vod_url，report_state 置 0。
    """
    li = d.get("liveInfo") or {}
    total_price = d.get("totalPrice")
    sign_price = d.get("signPrice")
    return {
        "pid": d.get("id"),
        "title": d.get("title"),
        "serial": d.get("serial"),
        "type_name": d.get("typeName"),
        "is_pre": d.get("isPre"),
        "count": d.get("count"),
        "total_price": total_price / 100 if total_price else 0,
        "sign_price": sign_price / 100 if sign_price else 0,
        "sell_time": d.get("sellTime"),
        "sell_days": d.get("sellDays"),
        "status": d.get("status"),
        "group_num": d.get("groupNum"),
        "description": d.get("description"),
        "create_time": d.get("createTime"),
        "completion_time": d.get("completionTime"),
        "cover_url": (d.get("cover") or {}).get("url"),
        "anchor_id": (d.get("anchor") or {}).get("id"),
        "anchor_username": (d.get("anchor") or {}).get("userName"),
        "sold_count": d.get("soldCount"),
        "detail_url": d.get("detailUrl"),
        "goods_url": d.get("goodsUrl"),
        "standard_name": d.get("standardName"),
        "live_task_time": d.get("liveTaskTime"),
        "live_id": li.get("id"),
        "vod_url": (li.get("vod_info") or {}).get("vodUrl"),
        "report_state": 0,
    }


def backfill_reports(pool, headers, pid):
    """抓取单个 pid 的全部拆卡报告并入库，同时更新 report_state。

    Args:
        pool (MySQLConnectionPool): 连接池。
        headers (dict): 请求头。
        pid (int): 商品 id。

    Returns:
        int: 本次入库尝试的报告条数（0 表示该商品无报告）。
    """
    page_num, total_pages, page_size = 1, 1, 20
    inserted = 0
    while page_num <= total_pages:
        j = req("POST", REPORT, headers,
                json={"pageSize": page_size, "my": 0, "pageNum": page_num, "tid": pid})
        if not j or j.get("status") != 200:
            raise RuntimeError(f"report/list 返回异常 pid={pid} page={page_num}")
        data = j.get("data") or {}
        total = data.get("total", 0)
        if page_num == 1:
            if total == 0:
                pool.update_one_or_dict(table="super_vault_product_record",
                                        data={"report_state": 2}, condition={"pid": pid})
                return 0
            total_pages = (total + page_size - 1) // page_size
        items = data.get("data", []) or []
        rows = [{
            "pid": pid,
            "user_name": it.get("userName"),
            "level": it.get("level"),
            "team_name_cn": it.get("teamNameCn"),
            "team_name_en": it.get("teamNameEn"),
            "count": it.get("count"),
            "picture_url": (it.get("picture") or {}).get("url"),
            "alias": it.get("alias"),
            "create_time": it.get("createTime"),
        } for it in items]
        if rows:
            pool.insert_many(table="super_vault_report_record", data_list=rows, ignore=True)
            inserted += len(rows)
        page_num += 1
    pool.update_one_or_dict(table="super_vault_product_record",
                            data={"report_state": 1}, condition={"pid": pid})
    return inserted


def main(pid_min, pid_max):
    """回补主流程：先补产品详情，再补拆卡报告，全程可续跑。

    Args:
        pid_min (int): 回补区间起始 pid（含）。
        pid_max (int): 回补区间结束 pid（含）。
    """
    pool = MySQLConnectionPool(log=logger)
    if not pool.check_pool_health():
        logger.error("数据库连接池异常")
        return
    token = pool.select_one("SELECT token FROM super_vault_token")[0]
    headers = build_headers(token)

    # ---------- Phase A: 补产品详情 ----------
    have = set(x[0] for x in pool.select_all("SELECT pid FROM super_vault_product_record"))
    missing = [p for p in range(pid_min, pid_max + 1) if p not in have]
    logger.info(f"Phase A 开始：区间[{pid_min},{pid_max}] 缺失 {len(missing)} 个 pid 待探测")
    inserted_p = invalid = 0
    for i, pid in enumerate(missing, 1):
        j = req("GET", DETAIL, headers, params={"id": str(pid)})
        d = (j or {}).get("data") or {}
        if j and j.get("status") == 200 and d.get("id"):
            pool.insert_many(table="super_vault_product_record",
                             data_list=[map_detail_to_row(d)], ignore=True)
            inserted_p += 1
        else:
            invalid += 1
        if i % 50 == 0 or i == len(missing):
            logger.info(f"  Phase A 进度 {i}/{len(missing)}  新增产品{inserted_p} 无效{invalid}")
        time.sleep(0.1)
    logger.info(f"Phase A 完成：新增产品 {inserted_p} 条，无效 pid {invalid} 个")

    # ---------- Phase B: 补拆卡报告（含之前超时遗留 + 本次新增） ----------
    pending = [x[0] for x in pool.select_all(
        "SELECT pid FROM super_vault_product_record WHERE report_state NOT IN (1,2) ORDER BY pid")]
    logger.info(f"Phase B 开始：待抓报告产品 {len(pending)} 个")
    ok = err = total_reports = 0
    for i, pid in enumerate(pending, 1):
        try:
            total_reports += backfill_reports(pool, headers, pid)
            ok += 1
        except Exception as e:
            err += 1
            pool.update_one_or_dict(table="super_vault_product_record",
                                    data={"report_state": 3}, condition={"pid": pid})
            logger.warning(f"  pid={pid} 报告回补失败: {e}")
        if i % 25 == 0 or i == len(pending):
            logger.info(f"  Phase B 进度 {i}/{len(pending)}  成功{ok} 失败{err} 累计报告{total_reports}")
        time.sleep(0.05)
    logger.info(f"Phase B 完成：成功{ok} 失败{err}，新增拆卡报告约 {total_reports} 条")

    p_cnt = pool.select_one("SELECT COUNT(*) FROM super_vault_product_record")[0]
    r_cnt = pool.select_one("SELECT COUNT(*) FROM super_vault_report_record")[0]
    logger.info(f"===== 回补结束：产品表共 {p_cnt} 条，报告表共 {r_cnt} 条 =====")


def parse_range_args(argv):
    """从命令行参数解析回补 pid 区间。

    Args:
        argv (list[str]): sys.argv[1:]，可为空或 [起始pid, 结束pid]。

    Returns:
        tuple[int, int]: (pid_min, pid_max)；未传参时返回默认区间。

    Raises:
        ValueError: 参数无法转为整数或 pid_min > pid_max 时抛出。
    """
    if len(argv) >= 2:
        pid_min, pid_max = int(argv[0]), int(argv[1])
        if pid_min > pid_max:
            raise ValueError(f"起始pid({pid_min}) 不能大于 结束pid({pid_max})")
        return pid_min, pid_max
    return DEFAULT_PID_MIN, DEFAULT_PID_MAX


if __name__ == "__main__":
    _pid_min, _pid_max = parse_range_args(sys.argv[1:])
    main(_pid_min, _pid_max)