spider_work/cxx_spider/super_vault_daily_spider.py

# -*- coding: utf-8 -*-
# Author : Charley
# Python : 3.10.8
# Date   : 2026/1/28 11:12
import inspect
import time
import requests
from loguru import logger
from mysql_pool import MySQLConnectionPool
from tenacity import retry, stop_after_attempt, wait_fixed

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


HEADERS = {
    "User-Agent": "okhttp/4.9.0",
    # "Connection": "Keep-Alive",
    # "Accept-Encoding": "gzip",
    "Authorization": "",
    "CXX-APP-API-VERSION": "V2",  # 必须添加
    # "deviceType": "2",
    # "udid": "20f902c10f6163a19bf137d801731d9f",
    # "time": str(int(time.time() * 1000)),
    "Content-Type": "application/json; charset=UTF-8"
}


def after_log(retry_state):
    """
    retry 回调
    :param retry_state: RetryCallState 对象
    """
    # 检查 args 是否存在且不为空
    if retry_state.args and len(retry_state.args) > 0:
        log = retry_state.args[0]  # 获取传入的 logger
    else:
        log = logger  # 使用全局 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):
    """
    获取代理
    :return: 代理
    """
    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


@retry(stop=stop_after_attempt(5), wait=wait_fixed(1), after=after_log)
def get_video_url(log, pid):
    """
    获取视频地址
    :param log: logger对象
    :param pid: 视频地址
    :return: 视频地址
    """
    log.debug(f"正在获取视频地址: {pid}")
    url = "https://cxx.cardsvault.net/app/teamup/detail"
    params = {
        # "id": "1730"
        "id": str(pid)
    }
    response = requests.get(url, headers=HEADERS, params=params, timeout=22)
    response.raise_for_status()
    result = response.json()
    liveInfo = result.get("data", {}).get("liveInfo", {})
    live_id = liveInfo.get("id") if liveInfo else None
    vodUrl = liveInfo.get("vod_info", {}).get("vodUrl")
    return live_id, vodUrl


@retry(stop=stop_after_attempt(5), wait=wait_fixed(1), after=after_log)
def get_vod_single_page(log, page_num=1, token=""):
    """
    获取单页数据
    :param log: logger对象
    :param page_num: 页码
    :param token: token
    :return: 数据
    """
    url = "https://cxx.cardsvault.net/app/teamup/vod/list"
    data = {
        "pageSize": 20,
        "pageNum": page_num
    }
    HEADERS["Authorization"] = token
    response = requests.post(url, headers=HEADERS, json=data, timeout=22)
    response.raise_for_status()

    result = response.json()
    print(result)
    if result.get("status") == 200:
        data = result.get("data", {})
        total = data.get("total", 0)
        current_page = data.get("pageNum", 1)
        items = data.get("data", [])

        log.info(f"当前查询的是 ->->-> 第 {current_page} 页，共 {total} 条记录")
        log.debug(f"当前页数据数量: {len(items)}")

        return {
            "total": total,
            "current_page": current_page,
            "items": items,
        }
    else:
        log.error(f"API 返回错误: {result.get('msg', '未知错误')}")
        return None


def parse_list_items(log, items):
    """
    解析列表项
    :param log: logger对象
    :param items: 列表项
    :return: 解析后的列表项
    """
    parsed_items = []
    log.debug(f"正在解析列表项.................")
    for item in items:
        pid = item.get("id")
        serial = item.get("serial")  # 编号
        title = item.get("title")
        type_name = item.get("typeName")  # 随机卡种
        isPre = item.get("isPre")
        count = item.get("count")

        totalPrice = item.get("totalPrice")
        totalPrice = totalPrice / 100 if totalPrice else 0
        signPrice = item.get("signPrice")
        signPrice = signPrice / 100 if signPrice else 0

        sellTime = item.get("sellTime")
        sellDays = item.get("sellDays")
        status = item.get("status")  # 9:完成 8:待发货
        groupNum = item.get("groupNum")
        description = item.get("description")
        createTime = item.get("createTime")
        completionTime = item.get("completionTime")  # 完成时间
        cover_url = item.get("cover", {}).get("url")  # 封面图

        anchor_id = item.get("anchor", {}).get("id")
        anchor_userName = item.get("anchor", {}).get("userName")
        soldCount = item.get("soldCount")
        detailUrl = item.get("detailUrl")
        goodsUrl = item.get("goodsUrl")
        standardName = item.get("standardName")  # 规格
        liveTaskTime = item.get("liveTaskTime")  # 直播时间

        try:
            live_id, vodUrl = get_video_url(log, pid)
        except Exception as e:
            log.error(f"Error getting video URL: {e}")
            live_id, vodUrl = None, None

        parsed_item = {
            "pid": pid,
            "title": title,
            "serial": serial,
            "type_name": type_name,
            "is_pre": isPre,
            "count": count,
            "total_price": totalPrice,
            "sign_price": signPrice,
            "sell_time": sellTime,
            "sell_days": sellDays,
            "status": status,
            "group_num": groupNum,
            "description": description,
            "create_time": createTime,
            "completion_time": completionTime,
            "cover_url": cover_url,
            "anchor_id": anchor_id,
            "anchor_username": anchor_userName,
            "sold_count": soldCount,
            "detail_url": detailUrl,
            "goods_url": goodsUrl,
            "standard_name": standardName,
            "live_task_time": liveTaskTime,
            "live_id": live_id,
            "vod_url": vodUrl
        }
        # print(parsed_item)
        parsed_items.append(parsed_item)
    return parsed_items


def get_vod_list(log, sql_pool, token):
    """
    获取列表数据
    :param log: logger对象
    :param sql_pool: 数据库连接池
    :param token: token
    """
    page_num = 1
    max_pages = 2

    while page_num <= max_pages:
        log.debug(f"正在获取第 {page_num} 页的数据.................")
        page_result = get_vod_single_page(log, page_num, token)

        if not page_result:
            log.error(f"获取第 {page_num} 页失败 !!!")
            break

        # 每页获取后立即解析
        items = parse_list_items(log, page_result["items"])
        sql_pool.insert_many(table="super_vault_product_record", data_list=items, ignore=True)

        page_num += 1


# ----------------------------------------------------------------------------------------------------------------------

def get_report_single_page(log, page_num, detail_id, token):
    """
    获取单页数据
    :param log: logger对象
    :param page_num: 页码
    :param detail_id: 商品id
    :param token: token
    :return: 数据
    """
    log.debug(f"正在获取第 {page_num} 页的 <拆卡报告> 数据.................")
    # token = "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJDSEFPWElOWElORyNBUFAiLCJhdWQiOiJDSEFPWElOWElORyIsIm5iZiI6MTc2OTU4MTI5NiwiZGF0YSI6Ijk1MjMiLCJpc3MiOiI3ViNweHlQZSIsImV4cCI6MTc3MDc4MTI5NiwiaWF0IjoxNzY5NTgxMjk2LCJqdGkiOiIyYjkwNzZhMS0wYjU1LTQ0ZjItOGZlZC0yMWZiZmI0ZjUyYWIifQ.iDzTZLDslCP0y2nc2Jp4TGEsNbQiCRKcUeRsIyG3iOg"

    url = "https://cxx.cardsvault.net/app/teamup/report/list"
    data = {
        "pageSize": 20,
        "my": 0,
        "pageNum": page_num,
        # "tid": 1780
        "tid": detail_id
    }
    HEADERS["Authorization"] = token
    response = requests.post(url, headers=HEADERS, json=data, timeout=22)
    # print(response.text)
    response.raise_for_status()

    result = response.json()
    if result.get("status") == 200:
        data = result.get("data", {})
        total = data.get("total", 0)
        current_page = data.get("pageNum", 1)
        items = data.get("data", [])

        log.info(f"当前查询的是 ->->-> 第 {current_page} 页，共 {total} 条记录")
        log.debug(f"当前页数据数量: {len(items)}")

        return {
            "total": total,
            "current_page": current_page,
            "items": items
        }
    else:
        log.error(f"API 返回错误: {result.get('msg', '未知错误')}")
        return None


def parse_report_items(log, detail_id, items):
    """
    解析列表项
    :param log: logger对象
    :param detail_id: 商品id
    :param items: 列表项
    :return: 解析后的列表项
    """
    parsed_items = []
    log.debug(f"正在解析 <拆卡报告> 列表项.................")
    for item in items:
        userName = item.get("userName")
        level = item.get("level")
        teamNameCn = item.get("teamNameCn")
        teamNameEn = item.get("teamNameEn")
        count = item.get("count")

        picture_url = item.get("picture", {}).get("url")
        alias = item.get("alias")  # 别名
        createTime = item.get("createTime")

        data_dict = {
            "pid": detail_id,
            "user_name": userName,
            "level": level,
            "team_name_cn": teamNameCn,
            "team_name_en": teamNameEn,
            "count": count,
            "picture_url": picture_url,
            "alias": alias,
            "create_time": createTime
        }
        parsed_items.append(data_dict)

    return parsed_items


def get_report_list(log, detail_id, token, sql_pool):
    """
    获取列表数据
    :param log: logger对象
    :param detail_id: 商品id
    :param token: token
    :param sql_pool: 数据库连接池
    """
    page_num = 1
    total_pages = 99
    items_per_page = 20  # pageSize

    while page_num <= total_pages:
        log.debug(f"正在获取第 {page_num} 页的数据.................")
        page_result = get_report_single_page(log, page_num, detail_id, token)

        if not page_result:
            log.error(f"获取第 {page_num} 页失败 !!!")
            break

        # 第一次请求时更新真实的总页数
        if page_num == 1:
            total_count = page_result["total"]
            if total_count == 0:
                log.info("No new records found.")
                # 更改状态为2
                sql_pool.update_one_or_dict(
                    table="super_vault_product_record",
                    data={"report_state": 2},
                    condition={"pid": detail_id}
                )
                break

            total_pages = (total_count + items_per_page - 1) // items_per_page
            log.info(f"总共 {total_pages} 页")

        items = parse_report_items(log, detail_id, page_result["items"])
        sql_pool.insert_many(table="super_vault_report_record", data_list=items, ignore=True)
        sql_pool.update_one_or_dict(
            table="super_vault_product_record",
            data={"report_state": 1},
            condition={"pid": detail_id}
        )

        page_num += 1


# ----------------------------------------------------------------------------------------------------------------------
# 补充/更新任务：修正 INSERT IGNORE「只增不改」导致的字段固化
#   列表接口靠 pid 唯一键 INSERT IGNORE 去重，商品首次入库常在预售阶段（sign_price/sold_count/
#   status 尚为 0/初始值，且未开播故 live_id/vod_url/completion_time 为空），此后即便平台更新，
#   库里也不再回写——这正是「大量团缺 completion_time / 直播回放」的根因（实测 status!=9 且已售出
#   的团里约 92% 其实已完成、detail 里已有 completionTime）。本段对 status!=9 或缺直播信息的团
#   逐个查 detail 接口刷新会变字段，补齐 status/completion_time/live_task_time/live_id/vod_url 等。

@retry(stop=stop_after_attempt(5), wait=wait_fixed(1), after=after_log)
def get_teamup_detail(log, pid, token):
    """请求单个拼团的详情，返回其最新完整字段。

    Args:
        log: 日志对象。
        pid (int): 拼团商品 id。
        token (str): 鉴权 token。

    Returns:
        dict | None: 详情 data 字典；接口业务码非 200 或无数据时返回 None。

    Raises:
        requests.HTTPError: HTTP 状态码非 2xx 时抛出以触发重试。
    """
    HEADERS["Authorization"] = token
    response = requests.get("https://cxx.cardsvault.net/app/teamup/detail",
                            headers=HEADERS, params={"id": str(pid)}, timeout=22)
    response.raise_for_status()
    result = response.json()
    if result.get("status") != 200:
        log.warning(f"detail 接口业务码异常 pid={pid}: {result.get('msg', '未知错误')}")
        return None
    return result.get("data") or None


def build_update_fields(data):
    """从详情 data 里挑出「会随售卖进程变化」的字段，组装成 UPDATE 用的 dict。

    signPrice/totalPrice 接口以「分」返回，统一 /100 转「元」入库（与列表解析口径一致）。
    直播相关字段（live_task_time/live_id/vod_url）为「直播后才产生」，首次预售入库常为空，一并补。

    Args:
        data (dict): get_teamup_detail 返回的详情字典。

    Returns:
        dict: 待更新字段 {列名: 值}，含 status/sold_count/sign_price/total_price/count/
              completion_time/sold_time/sell_time/sell_days/live_task_time/live_id/vod_url。
    """
    total_price = data.get("totalPrice")
    sign_price = data.get("signPrice")
    live_info = data.get("liveInfo") or {}               # 未开播可能为 None
    vod_info = live_info.get("vod_info") or {}           # 无回放可能为 None
    return {
        "status": data.get("status"),                          # 状态(9:完成 8:待发货 ...)
        "sold_count": data.get("soldCount"),                   # 已售份数
        "sign_price": sign_price / 100 if sign_price else 0,   # 单价(分→元)
        "total_price": total_price / 100 if total_price else 0,  # 团总价(分→元)
        "count": data.get("count"),                            # 总份数
        "completion_time": data.get("completionTime"),         # 完成时间(发货完成态才有)
        "sold_time": data.get("soldTime"),                     # 售罄成交时刻(补 completion_time 缺口,近期团几乎都有)
        "sell_time": data.get("sellTime"),                     # 开售时间
        "sell_days": data.get("sellDays"),                     # 售卖天数
        "live_task_time": data.get("liveTaskTime"),            # 直播时间
        "live_id": live_info.get("id"),                        # 直播 id
        "vod_url": vod_info.get("vodUrl"),                     # 直播回放地址
    }


def update_stale_products(log, sql_pool, token):
    """刷新陈旧拼团的会变字段，回写被 INSERT IGNORE 固化的旧值。

    候选 = status!=9（字段仍会变、常有已完成却未回写的团）或缺直播信息（live_task_time/
    live_id/vod_url 之一为空，直播后才产生、首次入库时常缺）的商品；逐个查详情并 UPDATE，
    使 sign_price/sold_count/status/total_price/completion_time/直播回放等保持最新。
    已完成(status=9)且直播信息齐全的字段已定型，不在候选内、省请求。

    Args:
        log: 日志对象。
        sql_pool (MySQLConnectionPool): 数据库连接池。
        token (str): 鉴权 token。
    """
    rows = sql_pool.select_all(
        "SELECT pid FROM super_vault_product_record "
        "WHERE status <> 9 OR status IS NULL "
        "OR vod_url IS NULL OR live_id IS NULL OR live_task_time IS NULL "
        "OR (sold_time IS NULL AND sold_count > 0)")
    pids = [r[0] for r in rows] if rows else []
    log.info(f"待刷新未完成拼团 {len(pids)} 个")
    updated = 0
    for pid in pids:
        try:
            data = get_teamup_detail(log, pid, token)
            if not data:
                continue
            sql_pool.update_one_or_dict(
                table="super_vault_product_record",
                data=build_update_fields(data),
                condition={"pid": pid})
            updated += 1
        except Exception as e:
            log.error(f"刷新 pid={pid} 失败: {e}")
        time.sleep(0.3)  # 限速，避免请求过密触发风控
    log.info(f"未完成拼团刷新完成，成功更新 {updated}/{len(pids)} 个")


def backfill_sold_time(log, sql_pool, token):
    """精准补齐已售罄团缺失的 sold_time（售罄成交时刻）。

    只挑「已进入售后阶段(status 3/8)、已有人购买(sold_count>0)、但 sold_time 仍为空」的团：
    这类团常在列表首次入库(INSERT IGNORE)后才售罄，soldTime 由平台事后生成、旧值不再回写，故需
    逐个查 detail 补。刻意不扫 status=9 的历史团——早期平台无 soldTime 字段、源头即为空，查了也
    补不到、纯浪费请求；未售罄的(status 0/10 等)源头同样无值，一并排除。仅当 detail 真返回
    soldTime 时才 UPDATE，且只写 sold_time 单字段，避免误改其它数据。

    Args:
        log: 日志对象。
        sql_pool (MySQLConnectionPool): 数据库连接池。
        token (str): 鉴权 token。
    """
    rows = sql_pool.select_all(
        "SELECT pid FROM super_vault_product_record "
        "WHERE sold_time IS NULL AND sold_count > 0 AND status IN (3, 8)")
    pids = [r[0] for r in rows] if rows else []
    log.info(f"待补 sold_time 的已售罄团 {len(pids)} 个")
    filled = 0
    for pid in pids:
        try:
            data = get_teamup_detail(log, pid, token)
            if not data:
                continue
            sold_time = data.get("soldTime")
            if not sold_time:  # 源头也无(团未真正售罄)，跳过不误写
                continue
            sql_pool.update_one_or_dict(
                table="super_vault_product_record",
                data={"sold_time": sold_time},
                condition={"pid": pid})
            filled += 1
        except Exception as e:
            log.error(f"补 sold_time pid={pid} 失败: {e}")
        time.sleep(0.3)  # 限速，避免请求过密触发风控
    log.info(f"sold_time 补齐完成，成功回写 {filled}/{len(pids)} 个")


@retry(stop=stop_after_attempt(100), wait=wait_fixed(3600), after=after_log)
def cxx_daily_main(log):
    """
    主函数
    :param log: logger对象
    """
    log.info(
        f'开始运行 {inspect.currentframe().f_code.co_name} 爬虫任务....................................................')

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

    try:
        token = sql_pool.select_one("SELECT token FROM super_vault_token")

        # 获取所有 pid
        try:
            get_vod_list(log, sql_pool, token[0])
        except Exception as e:
            log.error(f"Error fetching last_product_id: {e}")

        # 精准补齐已售罄团(status 3/8)的 sold_time：轻量、只改单字段，最新入库的团先补上
        try:
            backfill_sold_time(log, sql_pool, token[0])
        except Exception as e:
            log.error(f"Error backfilling sold_time: {e}")

        # 刷新未完成拼团(status!=9)的会变字段，修正 INSERT IGNORE 固化的旧 sign_price/sold_count/status 等
        try:
            update_stale_products(log, sql_pool, token[0])
        except Exception as e:
            log.error(f"Error updating stale products: {e}")

        time.sleep(5)

        # 获取所有 report_state = 0 的 pid
        sql_detail_id_list = sql_pool.select_all("SELECT pid FROM super_vault_product_record WHERE report_state != 1")
        if sql_detail_id_list:
            sql_detail_id_list = [item[0] for item in sql_detail_id_list]
            for detail_id in sql_detail_id_list:
                try:
                    get_report_list(log, detail_id, token[0], sql_pool)
                except Exception as e:
                    log.error(f"Error fetching last_product_id: {e}")
                    # 更改状态为3
                    sql_pool.update_one_or_dict(
                        table="super_vault_product_record",
                        data={"report_state": 3},
                        condition={"pid": detail_id}
                    )
        else:
            log.info("No new records found.")

    except Exception as e:
        log.error(f'{inspect.currentframe().f_code.co_name} error: {e}')
    finally:
        log.info(f'爬虫程序 {inspect.currentframe().f_code.co_name} 运行结束,等待下一轮的采集任务............')



if __name__ == '__main__':
    # get_vod_list(logger, None, '')
    # get_vod_single_page(logger, 1,'eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJDSEFPWElOWElORyNBUFAiLCJhdWQiOiJDSEFPWElOWElORyIsIm5iZiI6MTc4NjEwMjA5MywiZGF0YSI6Ijk1MjMiLCJpc3MiOiI3ViNweHlQZSIsImV4cCI6MTc4NzMwMjA5MywiaWF0IjoxNzg2MTAyMDkzLCJqdGkiOiI4MWE0YTIxNi1mNTc0LTQ2ZDUtYmIxOS05ZjMzNWMxYTczM2UifQ.ieTyqCqFGsDBt1WkO1aHI58hngXP9jcViNYzGRdkOYA')
    # get_report_single_page(logger,1, '','')

    cxx_daily_main(logger)
    # schedule_task()