spider_work/deca_spider/on_sale/deca_sold_spider.py

# -*- coding: utf-8 -*-
# Author : Charley
# Python : 3.12.10
# Date   : 2026/08/03
"""得卡 DECA 商家历史成交采集爬虫（独立任务）。

抓取每个商家的「历史成交」groupbuy/merchant/sold-list，INSERT IGNORE 写入 deca_sold_record。
历史成交为终态、只增不改，用 product_code 唯一键去重增量采集。
deca_sold_record 预留 report_state / replay_state，供后续通过 product_code / live_id
采集「拆卡报告」与「视频回放」。

鉴权：本接口需 Authorization Bearer（需 token），复用 deca_daily_spider 的登录/续期逻辑
（优先 refreshToken 续期，避开阿里云验证码）。签名规则同样复用。
"""
import sys
import time

import schedule
from loguru import logger
from tenacity import retry, stop_after_attempt, wait_fixed
from mysql_pool import MySQLConnectionPool
import deca_daily_spider as deca  # 复用 do_request / ensure_token / make_signature / after_log

# 覆盖 deca 导入时装的 logger handler，本任务日志独立到 sold_*.log
logger.remove()
logger.add("./logs/sold_{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")

PAGE_SIZE = 20     # 历史成交每页条数（抓包实测 20）
MAX_PAGES = 200    # 单商家历史成交翻页保护上限


def parse_sold(item: dict) -> dict | None:
    """把历史成交列表项解析成 deca_sold_record 一行。

    Args:
        item (dict): sold-list 的 data.list 项（结构同在售商品列表）。

    Returns:
        dict | None: 与 deca_sold_record 列对应的数据字典；无 code 时返回 None。
    """
    code = item.get("code")
    if not code:
        return None
    m = item.get("merchant") or {}
    return {
        "product_code": code,
        "merchant_user_id": str(m.get("merchantUserID")) if m.get("merchantUserID") else None,
        "merchant_name": m.get("merchantName"),
        "title": item.get("title"),
        "card_product_title": item.get("cardProductTitle"),
        "cover_image_url": item.get("coverImageUrl"),
        "unit_price": item.get("unitPrice"),
        "min_unit_price": item.get("minUnitPrice"),
        "max_unit_price": item.get("maxUnitPrice"),
        "card_count": item.get("cardCount"),
        "sold_count": item.get("soldCount"),
        "available_stock": item.get("availableStock"),
        "groupbuy_status": item.get("groupbuyStatus"),
        "groupbuy_status_name": item.get("groupbuyStatusName"),
        "play_type": item.get("playType"),
        "live_id": item.get("liveId"),
        "completed_at": item.get("completedAt") or None,   # 成交完成时间
        "publicity_at": item.get("publicityAt") or None,
    }


def get_sold_list(log, merchant_user_id: str, pool) -> int:
    """翻页拉取某商家历史成交，INSERT IGNORE 写入 deca_sold_record。

    历史成交为终态，用 product_code 唯一键去重，只增不改。

    Args:
        log: 日志对象。
        merchant_user_id (str): 商家用户 ID。
        pool (MySQLConnectionPool): MySQL 连接池。

    Returns:
        int: 本商家写入的历史成交商品数（已去重）。
    """
    page = 1
    saved = 0
    total = None
    while page <= MAX_PAGES:
        body = {"merchantUserId": merchant_user_id, "page": page, "pageSize": PAGE_SIZE}
        try:
            resp = deca.do_request(log, "/api/v1/app/groupbuy/merchant/sold-list", body, need_auth=True)
        except Exception as e:
            log.error(f"商家 {merchant_user_id} 历史成交第 {page} 页请求失败: {e}")
            break
        if not resp or resp.get("code") != 0:
            log.info(f"商家 {merchant_user_id} 历史成交返回异常: {resp.get('msg') if resp else None}")
            break

        data = resp.get("data") or {}
        if total is None:
            total = data.get("total")
        items = data.get("list") or []
        if not items:
            break

        rows = [r for r in (parse_sold(it) for it in items) if r]
        if rows:
            pool.insert_many(table="deca_sold_record", data_list=rows, ignore=True)
            saved += len(rows)

        # 翻页终止：已覆盖 total，或本页不足一页
        if total is not None and page * PAGE_SIZE >= total:
            break
        if len(items) < PAGE_SIZE:
            break
        page += 1
        time.sleep(0.2)

    if saved:
        log.info(f"商家 {merchant_user_id} 历史成交入库 {saved} 个(total={total})")
    return saved


def fill_sold_details(log, pool) -> int:
    """给未补详情(publish_at 为空)的历史成交商品补上架/开售/结束时间。

    调 groupbuy/detail 取 publishAt/saleStartAt/saleEndAt。历史成交为终态，
    每个商品补一次即可（publish_at 非空后不再重复拉）。

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

    Returns:
        int: 本轮成功补详情的商品数。
    """
    rows = pool.select_all(
        "SELECT product_code FROM deca_sold_record WHERE publish_at IS NULL") or []
    log.info(f"待补详情历史成交商品 {len(rows)} 个")
    filled = 0
    for (code,) in rows:
        try:
            d = deca.get_product_detail(log, code)
            if not d:
                continue
            pool.update_one(
                "UPDATE deca_sold_record SET publish_at=%s, sale_start_at=%s, sale_end_at=%s "
                "WHERE product_code=%s",
                (d.get("publishAt") or None, d.get("saleStartAt") or None,
                 d.get("saleEndAt") or None, code))
            filled += 1
        except Exception as e:
            log.error(f"历史成交商品 {code} 补详情失败: {e}")
        time.sleep(0.3)  # 轻微限速
    return filled


@retry(stop=stop_after_attempt(100), wait=wait_fixed(3600), after=deca.after_log)
def main_task(log):
    """遍历库内所有商家，采集历史成交写入 deca_sold_record。

    商家名单取自 deca_shop_record（由 deca_daily_spider 维护）。

    Args:
        log: 日志对象。

    Raises:
        RuntimeError: 数据库连接池异常时抛出以触发重试。
    """
    log.info(f"开始运行 {sys._getframe().f_code.co_name} 商家历史成交采集" + "." * 40)
    pool = MySQLConnectionPool(log=log)
    if not pool.check_pool_health():
        log.error("数据库连接池异常")
        raise RuntimeError("数据库连接池异常")

    try:
        deca.ensure_token(log)  # 预取 token（后续请求自动续期）
        rows = pool.select_all("SELECT merchant_user_id FROM deca_shop_record")
        uids = [r[0] for r in rows] if rows else []
        log.info(f"待采集商家 {len(uids)} 个")
        for uid in uids:
            try:
                get_sold_list(log, uid, pool)
            except Exception as e:
                log.error(f"get_sold_list error（商家 {uid}）: {e}")
            time.sleep(0.3)  # 轻微限速

        # 补详情：给未补(publish_at 为空)的历史成交商品拉 groupbuy/detail 的上架/开售/结束时间
        try:
            n = fill_sold_details(log, pool)
            log.info(f"历史成交补详情完成，本轮 {n} 个")
        except Exception as e:
            log.error(f"fill_sold_details 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():
    """定时任务入口：每天 03:00 采集一次商家历史成交。"""
    main_task(log=logger)  # 立即跑一次（调试时取消注释）

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


if __name__ == "__main__":
    schedule_task()