spider_work/kaji_spider/kj_daily_spider.py

# -*- coding: utf-8 -*-
# Author : Charley
# Python : 3.12.10
# Date   : 2026/07/02
"""卡集（card.kaji.com）每日采集爬虫。

三级采集管道（结构对齐 zc_new_daily_spider）：
    1. 商户发现：遍历「在售主列表」forsale/main，从每个在售商品里提取商户，写入 kj_shop_record。
    2. 商品采集：遍历库内每个商户的「已售/已完成列表」merchant?tp=2，逐商品补 detail 后写入 kj_product_record。
    3. 玩家采集：遍历 kj_product_record 中未采集过玩家的商品，抓 result/merge 中奖/参与名单写入 kj_player_record。

鉴权：卡集接口的 Authorization 采用 Open-Auth-Sig 请求签名（见 kj_auth.py），
每请求用当前时间实时生成，无登录 token、无过期问题，适合长期无人值守。
其中商户已售列表 merchant?tp=2 为公开接口，不需要签名。
"""
import re
import sys
import time
from datetime import datetime
from urllib.parse import unquote
import requests
import schedule
from loguru import logger
from tenacity import retry, stop_after_attempt, wait_fixed
from mysql_pool import MySQLConnectionPool

import kj_auth

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

# ==================== 基础配置 ====================
# 两个业务域名（来源：抓包）：主列表 / videoPlay 走 server 域，商户列表 / 详情 / 玩家走 page 域
SERVER_BASE = "https://server.ssl1.kaji6.com"
PAGE_BASE = "https://page.ssl1.kaji6.com"

PAGE_SIZE = 10  # 列表接口每页条数
PLAYER_PAGE_SIZE = 30  # 玩家名单每页条数（抓包实测为 30）
MAX_PAGES = 100  # 单个列表翻页保护上限，防异常时无限翻页

# 是否为每个商品补拉 detail（拿开始/结束时间、商户销量/粉丝、规格）。
# 关闭可大幅减少请求量，但 start_date/end_date/规格/商户 sold_number/fans 会缺失。
FETCH_DETAIL = True
# 是否使用代理。卡集实测可直连；若遇 IP 风控再置 True 并配置 get_proxys。
USE_PROXY = False

# 固定 UA（来源：抓包，卡集为 uni-app WebView）
UA = ("Mozilla/5.0 (Linux; Android 11; Pixel 5 Build/RQ3A.211001.001; wv) "
      "AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/148.0.7778.120 "
      "Mobile Safari/537.36 uni-app Html5Plus/1.0 (Immersed/52.727272)")

# 基础请求头（Authorization 每请求单独生成，不放这里）
BASE_HEADERS = {
    "deviceType": "phone",
    "Accept": "application/json, text/plain, */*",
    "plat": "android",
    "version": "2.5.39",  # 来源：抓包 App 版本
    "appVersionCode": "10003",  # 来源：抓包 App versionCode
    "user-agent": UA,
}


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

    Args:
        retry_state: tenacity 传入的 RetryCallState 对象，含调用参数与结果。
    """
    # 约定业务函数首个位置参数为 log；取不到时回退全局 logger
    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):
    """获取隧道代理配置（默认不启用，见 USE_PROXY）。

    Args:
        log: 日志对象。

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

    Raises:
        Exception: 组装代理配置异常时向上抛出以触发重试。
    """
    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 ts_to_dt(ts) -> str | None:
    """将秒级 Unix 时间戳转成 'YYYY-MM-DD HH:MM:SS' 字符串。

    Args:
        ts (int | None): 秒级时间戳；为 None / 0 时返回 None。

    Returns:
        str | None: 格式化时间字符串；无有效时间戳时返回 None。
    """
    if not ts:
        return None
    return datetime.fromtimestamp(ts).strftime("%Y-%m-%d %H:%M:%S")


def parse_desc(desc: str) -> tuple[str | None, int | None]:
    """从 detail.good.desc（URL 编码）中提取拼团系列与拼团天数。

    desc 解码后各字段以回车(\\r)分隔，形如：
        拼团系列：27-28
        拼团规格：1张/包，1包/盒，1盒/箱，共3包 3张
        拼团份数：1205份
        拼团时间：15天

    Args:
        desc (str): detail.good.desc 原始 URL 编码串；为空时返回 (None, None)。

    Returns:
        tuple[str | None, int | None]: (拼团系列, 拼团天数)。系列为文本(如 "27-28")，
            天数为整数(如 15)；对应项提取失败为 None。
    """
    if not desc:
        return None, None
    text = unquote(desc)  # URL 解码：%EF%BC%9A→：  %0D→\r
    m_series = re.search(r"拼团系列[：:]\s*(.*?)\s*(?:[\r\n]|$)", text)
    m_days = re.search(r"拼团时间[：:]\s*(\d+)", text)  # 只取数字，"15天"→15
    series = m_series.group(1).strip() if m_series else None
    days = int(m_days.group(1)) if m_days else None
    return series, days


@retry(stop=stop_after_attempt(5), wait=wait_fixed(1), after=after_log)
def kj_request(log, path: str, method: str = "GET", base: str = PAGE_BASE,
               params: dict = None, json_body: dict = None,
               need_auth: bool = True, extra_headers: dict = None):
    """卡集通用请求函数（带重试）。

    根据 need_auth 决定是否为该接口路径实时生成 Open-Auth-Sig 签名并注入 Authorization。

    Args:
        log: 日志对象。
        path (str): 接口相对路径，不含域名 / '/api/v4/' 前缀 / query，如 "goodlist/forsale/main"。
        method (str, optional): 请求方法，"GET" 或 "POST"。Defaults to "GET"。
        base (str, optional): 域名前缀，SERVER_BASE 或 PAGE_BASE。Defaults to PAGE_BASE。
        params (dict, optional): URL query 参数。Defaults to None。
        json_body (dict, optional): POST 的 JSON body。Defaults to None。
        need_auth (bool, optional): 是否注入 Authorization 签名。Defaults to True。
        extra_headers (dict, optional): 追加 / 覆盖的请求头。Defaults to None。

    Returns:
        dict | None: 响应 JSON；非 200 时抛异常触发重试。

    Raises:
        RuntimeError: HTTP 状态码非 200 时抛出。
    """
    url = f"{base}/api/v4/{path}"
    req_headers = BASE_HEADERS.copy()
    if need_auth:
        # 签名基于不含 query 的 path，kj_auth 内部会补 '/api/v4/' 前缀并裁掉 query
        req_headers["Authorization"] = kj_auth.gen_authorization(path)
    if extra_headers:
        req_headers.update(extra_headers)

    proxies = get_proxys(log) if USE_PROXY else None
    if method.upper() == "POST":
        resp = requests.post(url, headers=req_headers, json=json_body, params=params, timeout=(5, 30), proxies=proxies)
    else:
        resp = requests.get(url, headers=req_headers, params=params, timeout=(5, 30), proxies=proxies)

    if resp.status_code != 200:
        log.error(f"请求失败 {resp.status_code}: {url}")
        raise RuntimeError(f"HTTP {resp.status_code}")
    return resp.json()


# ==================== 一、商户发现（在售主列表） ====================
def get_forsale_page(log, fetch_from: int, fetch_size: int = PAGE_SIZE):
    """获取一页在售主列表 goodlist/forsale/main。

    Args:
        log: 日志对象。
        fetch_from (int): 游标起始位置（首页为 1，逐页按 fetch_size 递增）。
        fetch_size (int, optional): 每页条数。Defaults to PAGE_SIZE。

    Returns:
        dict | None: 响应 JSON（含 goodList / isFetchEnd）；失败返回 None。
    """
    return kj_request(
        log, "goodlist/forsale/main", base=SERVER_BASE,
        params={"fetchFrom": str(fetch_from), "fetchSize": str(fetch_size)},
        need_auth=True,
    )


def save_shops(log, good_list: list, sql_pool, seen: set) -> int:
    """从在售商品列表提取商户并写入 kj_shop_record（存在则更新店名）。

    在售商品项只带 merchantAlias / merchantName，故本步只写 shop_id + shop_name；
    sold_number / fans 由商品采集阶段的 detail.publisher 回填。

    Args:
        log: 日志对象。
        good_list (list[dict]): forsale 返回的 goodList，每项含 merchantAlias / merchantName。
        sql_pool (MySQLConnectionPool): MySQL 连接池。
        seen (set): 跨页去重的 shop_id 集合，避免重复写库。

    Returns:
        int: 本次实际写库的商户数（已去重）。
    """
    info_list = []
    # print(good_list)
    for item in good_list:
        shop_id = item.get("merchantAlias")
        shop_name = item.get("merchantName")
        data_dict = {"shop_id": shop_id, "shop_name": shop_name}

        if not shop_id or shop_id in seen: continue
        seen.add(shop_id)
        info_list.append(data_dict)
        # print(data_dict)

    if not info_list:
        log.info("无新商户，不写库")
        return 0

    # 存在则更新店名并把 is_deleted 刷回 0：能出现在在售列表 = 商户在营业
    # （曾被标记注销的商户在此复活，重新纳入采集）
    sql = ("INSERT INTO kj_shop_record (shop_id, shop_name) VALUES (%s, %s) "
           "ON DUPLICATE KEY UPDATE shop_name = VALUES(shop_name), is_deleted = 0")
    args_list = [(d["shop_id"], d["shop_name"]) for d in info_list]
    sql_pool.insert_many(query=sql, args_list=args_list)
    return len(info_list)


def get_shop_list(log, sql_pool) -> int:
    """遍历在售主列表翻页，发现并写入全部在售商户。

    翻页靠响应的 isFetchEnd 与空页判断，shop_id 唯一键去重兜底翻页游标的不确定性。

    Args:
        log: 日志对象。
        sql_pool (MySQLConnectionPool): MySQL 连接池。

    Returns:
        int: 去重后发现的商户总数。
    """
    seen = set()
    fetch_from = 1

    while fetch_from <= MAX_PAGES * PAGE_SIZE:
        try:
            data = get_forsale_page(log, fetch_from, PAGE_SIZE)
        except Exception as e:
            log.error(f"在售主列表 fetch_from={fetch_from} 请求失败: {e}")
            break
        if not data:
            break

        good_list = data.get("goodList", [])
        if not good_list:
            log.info(f"在售主列表 fetch_from={fetch_from} 无数据，停止翻页")
            break

        save_shops(log, good_list, sql_pool, seen)
        log.info(f"在售主列表 fetch_from={fetch_from} 完成，本页 {len(good_list)} 条，累计商户 {len(seen)}")

        if data.get("isFetchEnd") or len(good_list) < PAGE_SIZE:
            break
        fetch_from += PAGE_SIZE

    return len(seen)


# ==================== 二、商品采集（商户已售列表） ====================
def get_merchant_sold_page(log, merchant_code: str, page_index: int, page_size: int = PAGE_SIZE):
    """获取某商户「已售/已完成」列表的一页（merchant?tp=2，公开接口免签名）。

    Args:
        log: 日志对象。
        merchant_code (str): 商户编码（kj_shop_record.shop_id，如 MCT7550827）。
        page_index (int): 页码，从 1 开始。
        page_size (int, optional): 每页条数。Defaults to PAGE_SIZE。

    Returns:
        dict | None: 响应 JSON（含 list / totalPage）；失败返回 None。
    """
    return kj_request(
        log, f"merchant/1/goodlist/{merchant_code}", base=PAGE_BASE,
        params={"pageIndex": str(page_index), "pageSize": str(page_size), "tp": "2"},
        need_auth=False,
    )


def get_good_detail(log, good_code: str) -> dict:
    """获取商品详情，返回 detail.good（含时间 / 商户 publisher / 规格）。

    Args:
        log: 日志对象。
        good_code (str): 商品编码，如 ZC7653941。

    Returns:
        dict: detail 中的 good 字典；无数据时返回空字典。
    """
    j = kj_request(
        log, f"good/{good_code}/1/detail", base=PAGE_BASE,
        params={"referer": "MerchantList"}, need_auth=True,
    )
    return (j or {}).get("good", {}) or {}


def get_video(log, good_code: str, play_code: str) -> str | None:
    """获取商品「拆卡回放」视频地址。

    play_code 取自 detail.good.broadcast.playCode（为空表示回放未就绪）；body 的 sign 由
    kj_auth.video_sign 生成，videoPlay 接口本身不需要 Authorization。响应的 media_url 即视频地址。

    Args:
        log: 日志对象。
        good_code (str): 商品编码。
        play_code (str): 回放播放码，来自 detail.good.broadcast.playCode。

    Returns:
        str | None: 回放视频 URL；play_code 为空或接口无视频时返回 None。
    """
    if not play_code:
        return None
    ts = int(time.time())
    body = {"playCode": play_code, "sign": kj_auth.video_sign(ts, good_code, play_code), "ts": ts}
    j = kj_request(log, f"good/videoPlay/{good_code}", method="POST", base=SERVER_BASE,
                   json_body=body, need_auth=False, extra_headers={"Content-Type": "application/json"})
    print(j)
    return (j or {}).get("media_url") if j and j.get("code") == 0 else None


def build_product(good_item: dict, detail_good: dict, merchant_code: str, merchant_name: str, video_url: str) -> dict:
    """把商户列表项 + 商品详情组装成 kj_product_record 一行。

    卡集为拼团模型，zc 表中的直播 / 存储 / 多价字段无对应，统一置 None。

    Args:
        good_item (dict): merchant?tp=2 列表里的商品项（goodCode/title/pic/price/totalNum/currentNum/status）。
        detail_good (dict): detail.good，补 startAt/overAt/state/规格；detail 失败时为空字典。
        merchant_code (str): 商户编码。
        merchant_name (str): 商户名。
        video_url (str): 回放视频 URL。

    Returns:
        dict: 与 kj_product_record 列对应的数据字典。
    """
    # 时间戳转文本：ts_to_dt(detail_good.get("startAt"))
    imgs = detail_good.get("pic", {}).get("carousel", [])
    imgs = ','.join(imgs) if imgs else None

    # 拼团系列 / 拼团时间(天)：从 detail.good.desc 解码提取
    group_series, group_days = parse_desc(detail_good.get("desc", ""))
    # print(good_item)
    # print('------------------------------')
    # print(detail_good)

    row = {
        "shop_id": merchant_code,  # 商户编码
        "shop_name": merchant_name,  # 商户名
        "pid": good_item.get("goodCode"),  #
        "title": good_item.get("title"),  # 标题
        "price": good_item.get("price"),  # 价格
        "total_num": good_item.get("totalNum"),  # 总数量
        "current_num": good_item.get("currentNum"),  # 当前数量
        "status": good_item.get("status"),  # 状态
        "imgs": imgs,  # 图片链接, ','分割
        "start_at": ts_to_dt(detail_good.get("startAt")),  # 开始时间
        "over_at": ts_to_dt(detail_good.get("overAt")),  # 结束时间
        "spec_name": detail_good.get("spec", {}).get("name"),  # 规格配置
        "spec_content": detail_good.get("spec", {}).get("content"),  # 产品规格
        "group_series": group_series,  # 拼团系列，如 "27-28"
        "group_days": group_days,  # 拼团时间(天)，如 15
        # "publisher_sale": detail_good.get("publisher", {}).get("sale"),  # 在售
        # "publisher_fans": detail_good.get("publisher", {}).get("fans"),  # 出售者粉丝
        "live_start_at": ts_to_dt((detail_good.get("broadcast") or {}).get("startAt")),
        "video_url": video_url,
    }
    return row


def get_shop_stop_pid(shop_id: str, sql_pool) -> str | None:
    """取该商户已入库商品里 start_at 最新那条的 pid，作为增量翻页的停止线。

    merchant?tp=2 结果按 start_at 倒序（最新在前），daily 翻页碰到这个 pid 即可早停：
    它及其之后的商品都是上一轮已采过的。依赖 (shop_id, start_at) 复合索引，单值查询很快。

    Args:
        shop_id (str): 商户编码。
        sql_pool (MySQLConnectionPool): MySQL 连接池。

    Returns:
        str | None: 该商户最新商品的 pid(goodCode)；该商户暂无记录时返回 None（触发全量采集）。
    """
    row = sql_pool.select_one(
        "SELECT pid FROM kj_product_record WHERE shop_id = %s ORDER BY start_at DESC LIMIT 1",
        (shop_id,),
    )
    return row[0] if row else None


def get_sold_list(log, shop_id: str, shop_name: str, sql_pool, incremental: bool = True) -> int:
    """遍历某商户已售列表翻页，逐商品补详情后写入 kj_product_record。

    Args:
        log: 日志对象。
        shop_id (str): 商户编码（shop_id）。
        shop_name (str): 商户名。
        sql_pool (MySQLConnectionPool): MySQL 连接池。
        incremental (bool, optional): True(daily) 按 start_at 最新 pid 早停，只采新商品；
            False(history) 全量深翻所有页。Defaults to True。

    Returns:
        int: 本商户写入的商品数。
    """
    page_index = 1
    saved = 0
    stopped = False
    # 增量停止线：该商户上次采到的最新商品 pid（接口按 start_at 新在前，翻页碰到它即早停）
    stop_pid = get_shop_stop_pid(shop_id, sql_pool) if incremental else None

    while page_index <= MAX_PAGES:
        try:
            data = get_merchant_sold_page(log, shop_id, page_index)
        except Exception as e:
            log.error(f"商户 {shop_id} 已售列表第 {page_index} 页请求失败: {e}")
            break

        if not data:
            log.info(f"商户 {shop_id} 已售列表第 {page_index} 页无数据，停止翻页")
            break

        # 商户无效（已注销 / 不存在）：merchant?tp=2 返回 code=1、msg='无效商家'。
        # 首页即无效 → 标记 is_deleted=1；之后 kj_main 的「WHERE is_deleted = 0」会自动跳过它，不再浪费请求。
        if data.get("code") != 0:
            if page_index == 1:
                log.info(f"商户 {shop_id} 无效（{data.get('msg')}），标记 is_deleted=1")
                sql_pool.update_one("UPDATE kj_shop_record SET is_deleted = 1 WHERE shop_id = %s", (shop_id,))
            break

        good_list = data.get("list", [])
        if not good_list:
            log.info(f"商户 {shop_id} 已售列表第 {page_index} 页无数据，停止翻页")
            break
        total_page = data.get("totalPage", 1)

        batch = []
        for gi in good_list:
            code = gi.get("goodCode")
            # 增量早停：碰到上次采到的最新商品，它及之后都是已采过的旧数据
            if stop_pid and code == stop_pid:
                log.info(f"商户 {shop_id} 碰到停止线 pid={stop_pid}，增量早停")
                stopped = True
                break
            detail_good = {}
            if FETCH_DETAIL:
                try:
                    detail_good = get_good_detail(log, code)
                except Exception as e:
                    log.error(f"商品 {code} detail 请求失败: {e}")

            # 如需回放视频地址（media_url）：playCode 在 detail_good["broadcast"]["playCode"]（空=未就绪）
            # 首次拿不到不影响主数据 —— refill_videos 会在拼团完成后兜底补采
            try:
                video_url = get_video(log, code, (detail_good.get("broadcast") or {}).get("playCode"))
            except Exception as e:
                log.error(f"商品 {code} 视频获取失败: {e}")
                video_url = None

            row = build_product(gi, detail_good, shop_id, shop_name, video_url)
            # print(row)
            if row:
                batch.append(row)

        if batch:
            # 已存在（pid 唯一）则跳过：已完成商品为终态，无需覆盖
            sql_pool.insert_many(table="kj_product_record", data_list=batch, ignore=True)
            saved += len(batch)

        if stopped:
            log.info(f"商户 {shop_id} 增量早停，停止翻页")
            break

        log.info(f"商户 {shop_id} 已售列表第 {page_index}/{total_page} 页完成，本页 {len(good_list)} 商品")
        page_index += 1
        if page_index > total_page:
            log.info(f"商户 {shop_id} 已售列表翻页完成")
            break

    return saved


# ==================== 三、玩家采集（中奖/参与名单） ====================
def get_player_page(log, good_code: str, fetch_from: int, fetch_size: int = PLAYER_PAGE_SIZE):
    """获取某商品玩家名单的一页 good/{code}/result/merge。

    Args:
        log: 日志对象。
        good_code (str): 商品编码。
        fetch_from (int): 游标起始位置（首页为 1，逐页按 fetch_size 递增）。
        fetch_size (int, optional): 每页条数。Defaults to PLAYER_PAGE_SIZE。

    Returns:
        dict | None: 响应 JSON（含 code / list / isFetchEnd）；失败返回 None。
    """
    return kj_request(
        log, f"good/{good_code}/result/merge", base=PAGE_BASE,
        params={"fetchFrom": str(fetch_from), "fetchSize": str(fetch_size), "q": ""},
        need_auth=True,
    )


def save_players(log, good_code: str, player_list: list, sql_pool) -> int:
    """解析玩家名单并写入 kj_player_record。

    匿名玩家（userId=0、userName 为空）用 anonymousCode 兜底作为标识。

    Args:
        log: 日志对象。
        good_code (str): 商品编码（写入 pid 列）。
        player_list (list[dict]): result/merge 的 list，每项含 userId/userName/total/anonymousCode。
        sql_pool (MySQLConnectionPool): MySQL 连接池。

    Returns:
        int: 本次写入的玩家记录数。
    """
    log.info(f"开始保存商品 {good_code} 的玩家名单")
    info_list = []
    for item in player_list:
        anon = item.get("anonymousCode")
        user_id = item.get("userId")
        user_name = item.get("userName")
        if not user_name:  # 匿名玩家 userName 为空
            user_name = f"匿名_{anon}" if anon else None
        if not user_id:  # 匿名玩家 userId=0，用匿名码兜底
            user_id = anon
        info_list.append({
            "pid": good_code,  # 商品编码，来自函数参数（item 里没有商品标识）
            "give_number": item.get("total"),  # 该玩家份数：卡集字段是 total
            "user_id": str(user_id) if user_id is not None else None,
            "user_name": user_name,
        })

    if info_list:
        sql_pool.insert_many(table="kj_player_record", data_list=info_list)
    return len(info_list)


def get_player_list(log, good_code: str, sql_pool) -> bool:
    """遍历某商品玩家名单翻页并写库。

    Args:
        log: 日志对象。
        good_code (str): 商品编码。
        sql_pool (MySQLConnectionPool): MySQL 连接池。

    Returns:
        bool: True 表示抓到玩家数据，False 表示无数据（如拼团未完成）。
    """
    fetch_from = 1
    has_data = False

    while fetch_from <= MAX_PAGES * PLAYER_PAGE_SIZE:
        try:
            data = get_player_page(log, good_code, fetch_from, PLAYER_PAGE_SIZE)
        except Exception as e:
            log.error(f"商品 {good_code} 玩家名单 fetch_from={fetch_from} 请求失败: {e}")
            break
        if not data:
            log.info(f"商品 {good_code} 玩家名单 fetch_from={fetch_from} 无数据，停止翻页")
            break

        if data.get("code") != 0:
            # code=1 常见于"拼团未完成的商品"，视为暂无玩家
            log.info(f"商品 {good_code} 暂无玩家: {data.get('msg')}")
            break

        plist = data.get("list", [])
        if not plist:
            log.info(f"商品 {good_code} 玩家名单翻页完成")
            break

        has_data = True
        save_players(log, good_code, plist, sql_pool)

        if data.get("isFetchEnd") or len(plist) < PLAYER_PAGE_SIZE:
            log.info(f"商品 {good_code} 玩家名单翻页完成")
            break
        fetch_from += PLAYER_PAGE_SIZE

    return has_data


# ==================== 四、视频补采 ====================
def refill_videos(log, sql_pool) -> int:
    """补采视频地址：对拼团已完成（player_state=1）但 video_url 仍空的商品，
    重新拉 detail 取 broadcast.playCode → get_video → 更新 video_url。

    覆盖场景：首次入库时该商品还处于「即将拆卡 / 正在拆卡」等中间状态，
    broadcast.playCode 为空、视频拿不到；拼团完成、回放就绪后由本函数补回。
    playCode 仍空则跳过，留到下一轮再试，直到 video_url 有值。

    Args:
        log: 日志对象。
        sql_pool (MySQLConnectionPool): MySQL 连接池。

    Returns:
        int: 本轮成功补采的视频数。
    """
    rows = sql_pool.select_all(
        "SELECT pid FROM kj_product_record WHERE video_url IS NULL AND player_state = 1"
    )
    pids = [r[0] for r in rows] if rows else []
    log.info(f"待补视频商品 {len(pids)} 个")
    filled = 0
    for pid in pids:
        try:
            detail = get_good_detail(log, pid)
            play_code = (detail.get("broadcast") or {}).get("playCode")
            if not play_code:
                # 回放仍未就绪（正在拆卡 / 即将拆卡等），留到下一轮
                continue
            url = get_video(log, pid, play_code)
            if url:
                sql_pool.update_one(
                    "UPDATE kj_product_record SET video_url = %s WHERE pid = %s",
                    (url, pid),
                )
                filled += 1
                log.info(f"商品 {pid} 视频已补采 → {url}")
        except Exception as e:
            log.error(f"商品 {pid} 视频补采失败: {e}")
    return filled


# ==================== 主流程 ====================
@retry(stop=stop_after_attempt(100), wait=wait_fixed(3600), after=after_log)
def kj_main(log):
    """卡集每日采集主函数：商户发现 → 商品采集 → 玩家采集。

    Args:
        log: 日志对象。

    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:
        # 1) 商户发现
        try:
            n = get_shop_list(log, sql_pool)
            log.info(f"商户发现完成，去重商户 {n} 个")
        except Exception as e:
            log.error(f"get_shop_list error: {e}")

        time.sleep(5)

        # 2) 商品采集：遍历库内所有商户的已售列表
        try:
            shop_rows = sql_pool.select_all("SELECT shop_id, shop_name FROM kj_shop_record WHERE is_deleted = 0")
            log.info(f"待采集商户 {len(shop_rows)} 个")
            for shop_id, shop_name in shop_rows:
                try:
                    cnt = get_sold_list(log, shop_id, shop_name, sql_pool)
                    log.info(f"商户 {shop_id} {shop_name} 商品采集完成，写入 {cnt} 个")
                except Exception as e:
                    log.error(f"get_sold_list error（商户 {shop_id}）: {e}")
        except Exception as e:
            log.error(f"iterate_shop_list error: {e}")

        time.sleep(5)

        # 3) 玩家采集：遍历尚未成功采集玩家的商品
        try:
            prod_rows = sql_pool.select_all("SELECT pid FROM kj_product_record WHERE player_state != 1")
            pids = [row[0] for row in prod_rows] if prod_rows else []
            log.info(f"待采集玩家的商品 {len(pids)} 个")
            for pid in pids:
                try:
                    # 先置 1 表示开始采集（对齐 zc 断点标记）
                    sql_pool.update_one("UPDATE kj_product_record SET player_state = 1 WHERE pid = %s", (pid,))
                    has_data = get_player_list(log, pid, sql_pool)
                    if not has_data:
                        # 无玩家（如拼团未完成）置 2，下轮仍会重试
                        sql_pool.update_one("UPDATE kj_product_record SET player_state = 2 WHERE pid = %s", (pid,))
                except Exception as pid_error:
                    log.error(f"商品 {pid} 玩家采集失败: {pid_error}")
                    try:
                        sql_pool.update_one("UPDATE kj_product_record SET player_state = 3 WHERE pid = %s", (pid,))
                    except Exception as update_error:
                        log.error(f"更新商品 {pid} 状态失败: {update_error}")
        except Exception as e:
            log.error(f"iterate_player_list error: {e}")

        # 4) 视频补采：玩家采集之后，对拼团已完成(player_state=1)但 video_url 仍空的商品重取视频
        try:
            n = refill_videos(log, sql_pool)
            log.info(f"视频补采完成，本轮 {n} 条")
        except Exception as e:
            log.error(f"refill_videos error: {e}")
    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():
    """定时任务入口：每天 00:01 运行一次 kj_main。"""
    # 立即运行一次（调试时取消注释）
    # kj_main(log=logger)

    schedule.every().day.at("00:01").do(kj_main, log=logger)
    while True:
        schedule.run_pending()
        time.sleep(1)


if __name__ == "__main__":
    # kj_main(logger)
    schedule_task()