spider_work/chaogujie_spider/cgj_daily_spider.py

# -*- coding: utf-8 -*-
# Author : Charley
# Python : 3.12.10
# Date   : 2026/07/06
"""潮谷街（吃谷商城）每日增量采集爬虫。

三级采集管道（结构对齐 kaji 的 kj_daily_spider）：
    1. 商户发现：翻页遍历「shop 列表页」group_buying_products（product_types/category_id 模式），
       从每个商品里提取 merchant_id / merchant_name，写入 cgj_shop_record。
    2. 商品采集：遍历库内每个商户的「历史成交」group_buying_products（merchant_id + status 模式），
       逐商品补 product/info 详情后写入 cgj_product_record。
    3. 玩家采集：遍历 cgj_product_record 中未采集过玩家的商品，抓 announcement_by_user
       的买家名单写入 cgj_player_record。

鉴权：潮谷街这几个接口 access-token 为空、无签名/加密，属公开（游客）接口，
只需带固定的 devicecode / version 等请求头即可，适合长期无人值守。

翻页约定（三个列表接口统一）：请求体/参数带 cursor（首页为空串），响应返回
    data.list（本页数据）、data.cursor（下一页游标）、data.done（true=到底）。
    循环条件：done 为 True 或 cursor 为空即停止。

【字段解析】save_shops / build_product / save_players 里的 `for item in items:` 循环中，
"""
import sys
import time
from datetime import datetime

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

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

# ==================== 基础配置 ====================
# 业务域名（来源：抓包，四个接口都走这个域）
BASE = "https://5fdc92.chaogujieapp.com"

PAGE_SIZE = 20  # 列表接口每页条数（抓包实测 20）
PLAYER_PAGE_SIZE = 20  # 玩家名单每页条数（抓包实测 20）
# MAX_PAGES = 500  # 单个列表翻页保护上限，防异常时无限翻页
# 增量早停阈值：某商户历史成交连续这么多页「全是已入库商品(0 新)」即停止翻页。
# 用「连续多页」而非「单页」，兼容历史成交列表里新商品跨页穿插的情况；调大更稳、调小更省。
STOP_AFTER_DUPE_PAGES = 2
MAX_PAGES = 5  # 单个列表翻页保护上限，防异常时无限翻页  每日任务时设置更小限制

# 「历史成交」过滤值（来源：抓包固定 status=1796）。九尾等仅有在售无历史成交的商户返回 0 条，符合预期。
HISTORY_STATUS = 1796

# 设备码 / App 版本（来源：抓包，公开接口无需登录 token）
DEVICE_CODE = "8420C51411ECF16C802600868F2D214A"
APP_VERSION = "1.0.15"

# 是否为每个商品补拉 product/info 详情（拿系列/规格/简介/视频/上下架时间等）。
# 关闭可大幅减少请求量，但详情维度字段会缺失。
FETCH_DETAIL = True
# 是否为每个商品补拉直播回放（live/user_live/detail，room_id 取自详情的 live_room_id）。
# 依赖 FETCH_DETAIL=True——要先有详情才能拿到 live_room_id。
FETCH_REPLAY = True
# 是否使用代理。潮谷街实测可直连；若遇 IP 风控再置 True 并配置 get_proxys。
USE_PROXY = False

# 固定 UA（来源：抓包，潮谷街为 Flutter 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")

# 基础请求头（来源：抓包）。POST 的 content-type 由 requests 的 json= 自动补，不放这里。
BASE_HEADERS = {
    "user-agent": UA,
    "accept-encoding": "gzip",
    "channel-version": APP_VERSION,  # 抓包 channel-version
    "access-token": "",  # 抓包为空：公开/游客接口，无需登录
    "source-id": "1",  # 抓包固定
    "language": "zh",  # 抓包固定
    "version": APP_VERSION,  # 抓包 App 版本
    "devicecode": DEVICE_CODE,  # 抓包设备码
    "platform-id": "1",  # 抓包固定（1=Android）
    "channel-name": "huawei",  # 抓包渠道
}


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 | str | None): 秒级时间戳；为 None / 0 / 空时返回 None。

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


@retry(stop=stop_after_attempt(5), wait=wait_fixed(1), after=after_log)
def cgj_request(log, path: str, method: str = "GET",
                params: dict = None, json_body: dict = None,
                extra_headers: dict = None) -> dict | None:
    """潮谷街通用请求函数（带重试）。

    潮谷街接口无签名/加密，仅需固定请求头，故本函数只负责拼 URL、发请求、判状态码。

    Args:
        log: 日志对象。
        path (str): 接口相对路径（以 / 开头，不含域名），如 "/api/home/group_buying_products"。
        method (str, optional): 请求方法，"GET" 或 "POST"。Defaults to "GET"。
        params (dict, optional): URL query 参数。Defaults to None。
        json_body (dict, optional): POST 的 JSON body。Defaults to None。
        extra_headers (dict, optional): 追加 / 覆盖的请求头。Defaults to None。

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

    Raises:
        RuntimeError: HTTP 状态码非 200 时抛出。
    """
    url = f"{BASE}{path}"
    req_headers = BASE_HEADERS.copy()
    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_group_buying_page(log, cursor: str = "", merchant_id=None) -> dict | None:
    """获取 group_buying_products 一页（发现模式 / 商户历史成交模式二合一）。

    该接口一站两用：
        - 不传 merchant_id：全局「shop 列表页」，用 product_types/category_id 过滤，用于商户发现。
        - 传 merchant_id：某商户「历史成交」，用 status=HISTORY_STATUS 过滤。
    两种模式响应结构一致（data.list / data.cursor / data.done）。

    Args:
        log: 日志对象。
        cursor (str, optional): 翻页游标，首页传空串。Defaults to ""。
        merchant_id (int | str | None, optional): 商户 id；None=商户发现模式。Defaults to None。

    Returns:
        dict | None: 响应 JSON；失败时由 cgj_request 抛异常触发重试。
    """
    if merchant_id is not None:
        body = {
            "cursor": cursor,
            "size": PAGE_SIZE,
            "image_size": "68",
            "status": HISTORY_STATUS,
            "merchant_id": merchant_id,
        }
    else:
        body = {
            "cursor": cursor,
            "size": PAGE_SIZE,
            "image_size": "178",
            "product_types": [1],
            "category_id": 1,
        }
    return cgj_request(log, "/api/home/group_buying_products", method="POST", json_body=body)


# ==================== 一、商户发现（shop 列表页） ====================
def save_shops(log, items: list, sql_pool, seen: set) -> int:
    """从 shop 列表页的商品项提取商户并写入 cgj_shop_record（存在则更新店名并复活）。

    ------------------------------------------------------------------
    商户维度目前只取驱动管道必需的 merchant_id / merchant_name，
    如需更多商户字段（如 merchant_image_url），在下面 for 循环里往 data_dict 加即可，
    并同步 schema.sql 的 cgj_shop_record 列与下方 upsert 的 SQL。
    ------------------------------------------------------------------

    Args:
        log: 日志对象。
        items (list[dict]): group_buying_products 返回的 data.list，每项含 merchant_id / merchant_name。
        sql_pool (MySQLConnectionPool): MySQL 连接池。
        seen (set): 跨页去重的 merchant_id 集合，避免重复写库。

    Returns:
        int: 本次实际写库的商户数（已去重）。
    """
    info_list = []
    for item in items:
        merchant_id = item.get("merchant_id")  # 商户 id（管道必需）
        merchant_name = item.get("merchant_name")  # 商户名（管道必需）

        data_dict = {"shop_id": merchant_id, "shop_name": merchant_name}

        if not merchant_id or merchant_id in seen:
            log.info(f"商户 {merchant_id} 已存在 seen，跳过")
            continue
        seen.add(merchant_id)
        info_list.append(data_dict)

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

    # 存在则更新店名并把 is_deleted 刷回 0：能出现在列表 = 商户在营业（曾注销的在此复活）
    sql = ("INSERT INTO cgj_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]
    # print(args_list)
    sql_pool.insert_many(query=sql, args_list=args_list)
    return len(info_list)


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

    翻页靠响应的 done / cursor 判断，merchant_id 唯一键去重兜底。

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

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

    while page < MAX_PAGES:
        try:
            resp = get_group_buying_page(log, cursor, merchant_id=None)
        except Exception as e:
            log.error(f"shop 列表页 cursor={cursor!r} 请求失败: {e}")
            break
        if not resp or resp.get("code") != 0:
            log.info(f"shop 列表页返回异常: {resp.get('msg') if resp else None}")
            break

        data = resp.get("data") or {}
        items = data.get("list") or []
        if not items:
            log.info(f"shop 列表页第 {page + 1} 页无数据，停止翻页")
            break

        save_shops(log, items, sql_pool, seen)
        log.info(f"shop 列表页第 {page + 1} 页完成，本页 {len(items)} 条，累计商户 {len(seen)}")

        if data.get("done") or not data.get("cursor"):
            break
        cursor = data["cursor"]
        page += 1

    return len(seen)


# ==================== 二、商品采集（商户历史成交 + 详情） ====================
def get_product_detail(log, product_id) -> dict:
    """获取商品详情 product/info，返回 data 节点（含 product_info / merchant_info 等）。

    Args:
        log: 日志对象。
        product_id (int | str): 商品 id。

    Returns:
        dict: 详情 data 字典；无数据时返回空字典。
    """
    resp = cgj_request(log, "/api/product/info", method="GET", params={"id": product_id})
    return (resp or {}).get("data") or {}


def get_live_detail(log, room_id: str) -> dict:
    """获取直播回放详情 live/user_live/detail，返回 data 节点。

    room_id 取自商品详情的 data.live_room_id。回放视频地址在 data.offline_address_info[].url
    （腾讯云 vod，name="默认"），另带 status / start_time / end_time / duration 等直播信息。

    Args:
        log: 日志对象。
        room_id (str): 直播间 id（商品详情的 live_room_id）。

    Returns:
        dict: 回放详情 data 字典；room_id 为空或无数据时返回空字典。
    """
    if not room_id:
        return {}
    resp = cgj_request(log, "/api/live/user_live/detail", method="POST", json_body={"room_id": room_id})
    return (resp or {}).get("data") or {}


def build_product(item: dict, detail: dict, shop_id, shop_name: str, live_detail: dict = None) -> dict:
    """把历史成交列表项 + 商品详情组装成 cgj_product_record 一行。

    ------------------------------------------------------------------
    列表项字段（item.*）来自 group_buying_products；详情字段（detail.*）来自 product/info。
    ------------------------------------------------------------------

    Args:
        item (dict): 历史成交列表里的商品项（group_buying_products data.list 项）。
        detail (dict): product/info 的 data 节点；FETCH_DETAIL 关闭或失败时为空字典。
        shop_id (int | str): 商户 id。
        shop_name (str): 商户名。
        live_detail (dict, optional): live/user_live/detail 的 data 节点（回放信息）；
            FETCH_REPLAY 关闭 / 无 live_room_id / 失败时为空字典。Defaults to None。

    Returns:
        dict: 与 cgj_product_record 列对应的数据字典。
    """
    # 详情里的嵌套节点（按需取用）
    product_info = detail.get("product_info") or {}
    basic_info = product_info.get("basic_info") or {}
    # print(product_info)
    # print('-------------------')
    # print(basic_info)

    # 直播回放（待确认）：回放地址在 offline_address_info[].url（腾讯云 vod），取第一个（name="默认"）
    live_detail = live_detail or {}
    offline_list = live_detail.get("offline_address_info") or []
    replay_url = offline_list[0].get("url") if offline_list else None

    row = {
        # ---- 管道必需（勿删）----
        "pid": item.get("id"),  # 商品 id
        "shop_id": shop_id,  # 商户 id
        "shop_name": shop_name,  # 商户名

        # ---- 列表项字段（item.*，待确认）----
        "title": item.get("name"),  # 商品名
        "price": item.get("price"),  # 价格
        "imgs": item.get("image_url"),  # 主图
        "stock_quantity": item.get("stock_quantity"),  # 库存
        "sold_quantity": item.get("sold_quantity"),  # 已售
        "remaining_quantity": item.get("remaining_quantity"),  # 剩余
        "card_edition_name": item.get("card_edition_name"),  # 卡版名

        # ---- 详情字段（detail.*，待确认；FETCH_DETAIL 关闭时为空）----
        "specs": basic_info.get("specs"),  # 规格
        "status": product_info.get("status"),  # 商品状态
        "listing_beg_time": product_info.get("listing_beg_time"),  # 上架时间(详情直接给文本)
        "listing_end_time": product_info.get("listing_end_time"),  # 下架时间(详情文本)

        # ---- 直播回放字段（来自 live/user_live/detail；FETCH_REPLAY 关闭时为空）----
        "live_room_id": detail.get("live_room_id"),  # 直播间 id（补采回放用，勿删）
        "replay_url": replay_url,  # 回放视频地址(offline_address_info[0].url，多个只取下标0)
        "live_start_time": live_detail.get("start_time"),  # 开播时间
        "live_end_time": live_detail.get("end_time"),  # 结束时间
    }
    return row


def filter_new_pids(shop_id, page_pids: list, sql_pool) -> set:
    """在「本页」的 pid 里查出哪些是新的（库中不存在），只查这一页、不整表拉取。

    为什么按页查而非一次性把该商户全部 pid 拉成集合：某些商户历史成交可能几十万条，
    整表拉取会 SELECT 出几十万行并在内存堆一个几十 MB 的集合，每轮每店都来一遍不划算。
    改为每页用 `pid IN (...)` 批量查已存在的（凭 pid 唯一索引很快），内存只占一页的量，
    商户体量再大也不受影响；增量本就只翻头几页（连续旧页即早停），每店只几次小查询。

    历史成交列表顺序不稳定、新商品会跨页穿插，所以判重用「集合」而非「单个停止线精确匹配」，
    再配合调用方的「连续 STOP_AFTER_DUPE_PAGES 页全旧」早停。

    Args:
        shop_id (int | str): 商户 id。
        page_pids (list): 本页商品 pid 列表（原始类型，可能为 int）。
        sql_pool (MySQLConnectionPool): MySQL 连接池。

    Returns:
        set[int]: 本页中「库里还没有」的 pid 集合；本页为空时返回空集合。
            pid 库内与 API 均为 int，直接按 int 比较（索引最优）。
    """
    pids = [int(p) for p in page_pids if p is not None]
    if not pids:
        return set()
    placeholders = ",".join(["%s"] * len(pids))
    rows = sql_pool.select_all(
        f"SELECT pid FROM cgj_product_record WHERE shop_id = %s AND pid IN ({placeholders})",
        (shop_id, *pids),
    )
    existing = {r[0] for r in rows} if rows else set()
    return {p for p in pids if p not in existing}


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

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

    Returns:
        int: 本商户写入的商品数。
    """
    cursor = ""
    page = 0
    saved = 0
    dupe_pages = 0  # 连续「全是已入库商品(0 新)」的页数，达到阈值即增量早停

    while page < MAX_PAGES:
        try:
            resp = get_group_buying_page(log, cursor, merchant_id=shop_id)
            # print(resp)
        except Exception as e:
            log.error(f"商户 {shop_id} 历史成交 cursor={cursor!r} 请求失败: {e}")
            break

        if not resp or resp.get("code") != 0:
            log.info(f"商户 {shop_id} 历史成交返回异常: {resp.get('msg') if resp else None}")
            break

        data = resp.get("data") or {}
        items = data.get("list") or []
        if not items:
            log.info(f"商户 {shop_id} 历史成交第 {page + 1} 页无数据，停止翻页")
            break

        # 增量判重：只查本页 pid 里哪些是新的（不整表拉取，适配商户体量增长）
        new_pids = filter_new_pids(shop_id, [it.get("id") for it in items], sql_pool) if incremental else None

        batch = []
        new_on_page = 0  # 本页「新商品」数（用于连续全旧页早停判断）
        for item in items:
            pid = item.get("id")
            # 已入库的商品直接跳过，连详情/回放都不再请求（省大量无用请求）
            if incremental and int(pid) not in new_pids:
                continue
            new_on_page += 1

            detail = {}
            if FETCH_DETAIL:
                try:
                    detail = get_product_detail(log, pid)
                except Exception as e:
                    log.error(f"商品 {pid} 详情请求失败: {e}")

            # 直播回放：room_id 取自详情的 live_room_id，再单独请求 live/user_live/detail
            live_detail = {}
            if FETCH_REPLAY and detail:
                room_id = detail.get("live_room_id")
                if room_id:
                    try:
                        live_detail = get_live_detail(log, room_id)
                    except Exception as e:
                        log.error(f"商品 {pid} 回放详情请求失败: {e}")

            row = build_product(item, detail, shop_id, shop_name, live_detail)
            if row:
                batch.append(row)

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

        log.info(f"商户 {shop_id} 历史成交第 {page + 1} 页完成，本页 {len(items)} 商品，新增 {new_on_page} 个")

        # 增量早停：连续 STOP_AFTER_DUPE_PAGES 页都没有新商品 → 已翻到旧数据区，停止翻页
        if incremental:
            if new_on_page == 0:
                dupe_pages += 1
                if dupe_pages >= STOP_AFTER_DUPE_PAGES:
                    log.info(f"商户 {shop_id} 连续 {dupe_pages} 页无新商品，增量早停")
                    break
            else:
                dupe_pages = 0

        if data.get("done") or not data.get("cursor"):
            break
        cursor = data["cursor"]
        page += 1

    return saved


# ==================== 三、玩家采集（买家名单） ====================
def get_player_page(log, product_id, cursor: str = "") -> dict | None:
    """获取某商品玩家（买家）名单的一页 announcement_by_user。

    Args:
        log: 日志对象。
        product_id (int | str): 商品 id。
        cursor (str, optional): 翻页游标，首页传空串。Defaults to ""。

    Returns:
        dict | None: 响应 JSON（含 data.list / cursor / done）；失败由 cgj_request 抛异常触发重试。
    """
    return cgj_request(
        log, "/api/product/announcement_by_user", method="GET",
        params={"id": product_id, "cursor": cursor, "size": PLAYER_PAGE_SIZE},
    )


def save_players(product_id, items: list, sql_pool) -> int:
    """解析玩家（买家）名单并写入 cgj_player_record。

    ------------------------------------------------------------------
    并同步 schema.sql 的 cgj_player_record 列。
    ------------------------------------------------------------------

    Args:
        product_id (int | str): 商品 id（写入 pid 列）。
        items (list[dict]): announcement_by_user 的 data.list，每项含 view_key/total/buyer_name/buyer_image_url。
        sql_pool (MySQLConnectionPool): MySQL 连接池。

    Returns:
        int: 本次写入的玩家记录数。
    """
    info_list = []
    for item in items:
        # print(item)
        info_list.append({
            "pid": product_id,  # 商品 id（管道必需，item 里没有）
            "view_key": item.get("view_key"),  # 唯一标识
            "total": item.get("total"),  # 份数
            "buyer_name": item.get("buyer_name")  # 买家昵称（脱敏）
        })
        # print(info_list)

    if info_list:
        sql_pool.insert_many(table="cgj_player_record", data_list=info_list, ignore=True)
    return len(info_list)


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

    Args:
        log: 日志对象。
        product_id (int | str): 商品 id。
        sql_pool (MySQLConnectionPool): MySQL 连接池。

    Returns:
        bool: True 表示抓到玩家数据，False 表示无数据。
    """
    cursor = ""
    page = 0
    has_data = False

    while page < MAX_PAGES:
        try:
            resp = get_player_page(log, product_id, cursor)
        except Exception as e:
            log.error(f"商品 {product_id} 玩家名单 cursor={cursor!r} 请求失败: {e}")
            break

        if not resp or resp.get("code") != 0:
            log.info(f"商品 {product_id} 暂无玩家: {resp.get('msg') if resp else None}")
            break

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

        has_data = True
        save_players(product_id, items, sql_pool)

        if data.get("done") or not data.get("cursor"):
            log.info(f"商品 {product_id} 玩家名单翻页完成")
            break
        cursor = data["cursor"]
        page += 1

    return has_data


# ==================== 四、回放补采 ====================
def refill_replays(log, sql_pool) -> int:
    """补采回放地址：对已采玩家(player_state=1)但 replay_url 仍空的商品，
    重取 live_room_id → live/user_live/detail → offline_address_info[0].url → 更新 replay_url。

    覆盖场景：首次入库时直播还没结束 / 回放 VOD 还没生成，offline_address_info 为空、
    replay_url 拿不到；直播结束、回放就绪后由本函数补回。库里已存 live_room_id 则直接用，
    没有（如首采详情失败）则回退重取商品详情拿 room_id。replay_url 仍空则留到下一轮再试。

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

    Returns:
        int: 本轮成功补采的回放数。
    """
    rows = sql_pool.select_all(
        "SELECT pid, live_room_id FROM cgj_product_record WHERE replay_url IS NULL AND player_state = 1"
    )
    rows = rows or []
    log.info(f"待补回放商品 {len(rows)} 个")
    filled = 0
    for pid, room_id in rows:
        try:
            # 库里没存 room_id（首采详情失败等）→ 回退重取详情
            if not room_id:
                detail = get_product_detail(log, pid)
                room_id = detail.get("live_room_id")
            if not room_id:
                continue

            live_detail = get_live_detail(log, room_id)
            offline_list = live_detail.get("offline_address_info") or []
            url = offline_list[0].get("url") if offline_list else None
            live_start_time = live_detail.get("start_time")  # 开播时间
            live_end_time = live_detail.get("end_time")  # 结束时间
            if url:
                sql_pool.update_one(
                    "UPDATE cgj_product_record SET replay_url = %s, live_start_time = %s, live_end_time = %s WHERE pid = %s",
                    (url, live_start_time, live_end_time, 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 cgj_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 cgj_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 cgj_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 表示开始采集
                    sql_pool.update_one("UPDATE cgj_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 cgj_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 cgj_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)但 replay_url 仍空的商品重取回放
        try:
            n = refill_replays(log, sql_pool)
            log.info(f"回放补采完成，本轮 {n} 条")
        except Exception as e:
            log.error(f"refill_replays 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 运行一次 cgj_main。"""
    # 立即运行一次（调试时取消注释）
    # cgj_main(log=logger)

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


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