spider_work/jiwu_spider/jw_sold_report.py

# -*- coding: utf-8 -*-
# Author : Charley
# Python : 3.12.10
# Date   : 2026/08/19
"""集物星球已售日报：只读 jw_sold_product_record / jw_player_record，生成 Excel 发企微群。

  排版与口径对齐参考项目 deca 的《已售每日统计报告》（8→9 sheet）：只读库不触发抓取。
  业务日窗口：成交完成时间落在 [昨天17:00:00, 今天06:00:00]（含两端）算「昨天」的已售（对齐 deca 的
  completed_at 口径，凌晨仍开播故终点延到 06:00）。成交完成时间取 finish_time(结束)，为空退回 soldout_time(售罄)。
  报告 09:10 跑，已售采集脚本 08:00 先落库（窗口 06:00 关闭后再采）。

  9 个 sheet：平台总览 / 产品系列榜 / 商家GMV榜 / 运营节奏 / Jake明细 / Jake用户排行榜 /
    九叔明细 / 九叔用户排行榜 / 其他商家。
  与 deca 差异：集物两重点商家(Jake/九叔)都有真实购买记录，故「参与人数」全用真实去重买家(deca 对非重点
    商家用中卡近似)、两商家各出一个用户排行榜；集物未采进度轨迹，故明细无「到25/50/75%用时」里程碑列。
  金额字段库中已是元(入库时已换算)，直接展示。
  """
import os
import sys
import time
from datetime import date, datetime, timedelta

import schedule
from loguru import logger
from mysql_pool import MySQLConnectionPool

import auto_send_wx_msg as wx
import jw_report_excel as xl

PRODUCT_TYPE_NAME = {"1": "福袋", "2": "变风盒", "3": "错版卡", "4": "原盒"}
# 重点关注商家：(corpInfoId, 展示名, sheet名前缀)，各出「{前缀}明细」+「{前缀}用户排行榜」
FOCUS_CORPS = [(100716, "Jake球星卡", "Jake"), (100715, "九叔的喷火龙", "九叔")]

# 已售业务日窗口：成交完成时间落在 [昨17:00, 今06:00]（含两端）；成交完成时间取 finish_time(结束)，
# 为空退回 soldout_time(售罄)。对齐参考项目 deca 的 completed_at 口径。
WIN_SOLD = ("COALESCE(finish_time, soldout_time) >= (CURDATE() - INTERVAL 1 DAY) + INTERVAL 17 HOUR "
            "AND COALESCE(finish_time, soldout_time) <= CURDATE() + INTERVAL 6 HOUR")
# 昨日同窗口(用于组齐环比)：[前天17:00, 昨天06:00]，与 WIN_SOLD 整体平移一天、口径一致
WIN_SOLD_YDAY = ("COALESCE(finish_time, soldout_time) >= (CURDATE() - INTERVAL 2 DAY) + INTERVAL 17 HOUR "
                 "AND COALESCE(finish_time, soldout_time) <= (CURDATE() - INTERVAL 1 DAY) + INTERVAL 6 HOUR")

TOP_SERIES = 15                       # 产品系列榜展示上限
TOP_MERCHANT = 10                     # 商家GMV榜展示上限
CONC_TOPS = (1, 3, 5, 10)            # GMV集中度档位 TopK
HOUR_DIST_DAYS = 7                    # 成交时段分布统计近 N 天

OUT_DIR = "./reports"
OUT_PREFIX = "集物星球已售日报"
SEND_WECHAT = True
REPORT_TIME = "09:10"  # 晚于在售日报 10 分钟，避开并发

logger.remove()
logger.add("./logs/sold_report_{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")


def yuan(raw) -> float | None:
    """把库中金额(已是元, DECIMAL)转 float 供 Excel。

    Args:
        raw: 库中金额值（Decimal/数值/None）。

    Returns:
        float | None: 元；空值返回 None。
    """
    return float(raw) if raw is not None else None


def fmt_dt(v) -> str:
    """把时间字段格式化为 "YYYY-MM-DD HH:MM"。

    Args:
        v (datetime | str | None): 时间值。

    Returns:
        str: 格式化字符串；空值返回空串。
    """
    if not v:
        return ""
    if isinstance(v, datetime):
        return v.strftime("%Y-%m-%d %H:%M")
    return str(v)


def _duration(start, end) -> str:
    """算售卖时长文案（成交时间 - 开售时间），格式 "X小时Y分"（对齐 deca）。

    Args:
        start (datetime | None): 开售时间。
        end (datetime | None): 成交时间。

    Returns:
        str: 形如 "72小时13分"；缺失/异常返回空串。
    """
    if not (isinstance(start, datetime) and isinstance(end, datetime)):
        return ""
    secs = (end - start).total_seconds()
    if secs < 0:
        return ""
    return f"{int(secs // 3600)}小时{int((secs % 3600) // 60)}分"


def fmt_pct_change(today: float, yday: float) -> str:
    """算今日相对昨日同窗口的环比涨跌百分比文案。

    Args:
        today (float): 今日值。
        yday (float): 昨日同窗口值。

    Returns:
        str: 形如 "+12.3%"/"-5.0%"；昨日为 0 返回 "—"。
    """
    if not yday:
        return "—"
    return f"{(today - yday) / yday * 100:+.1f}%"


def _bar(value: int, vmax: int, width: int = 20) -> str:
    """按占最大值比例生成 █ 条形迷你图字符串。

    Args:
        value (int): 当前值。
        vmax (int): 最大值。
        width (int, optional): 满格字符数。Defaults to 20。

    Returns:
        str: █ 组成的条形；value/vmax 为空返回空串。
    """
    if not vmax or not value:
        return ""
    return "█" * max(1, round(value / vmax * width))


def _line(ws, row: int, text: str) -> int:
    """在 A 列写一行纯文本（供覆盖检测/漏采明细/脚注这类非表格文案）。

    Args:
        ws: 工作表。
        row (int): 行号。
        text (str): 文本。

    Returns:
        int: 下一行号。
    """
    ws.cell(row=row, column=1, value=text)
    return row + 1


def get_window(pool) -> tuple:
    """算出已售业务日窗口的起止时刻（供报告展示）。

    Args:
        pool: 数据库连接池。

    Returns:
        tuple: (start, end) 两个 "YYYY-MM-DD HH:MM:SS" 字符串，即 [昨17:00, 今06:00]。
    """
    row = pool.select_all(
        "SELECT (CURDATE() - INTERVAL 1 DAY) + INTERVAL 17 HOUR, CURDATE() + INTERVAL 6 HOUR")[0]
    return str(row[0]), str(row[1])


def _sold_gmv(rec: dict) -> tuple:
    """由一条已售(成交/组齐)记录算出 (售出份数, GMV元)。

    已售历史 corp/history 里的团都是**成交组齐团**，总份数全部售出，故售出份数 = `stock_amount`
    （实测全表 `SUM(购买记录 buy_count) == stock_amount`；该接口 `residueStockAmount` 对成交团恒
    等于总份数、不可用 stock−residue，否则算出售出=0、GMV=0）。GMV = 售出份数 × 单价元。

    Args:
        rec (dict): fetch_sold_rows 的一条。

    Returns:
        tuple: (sold_count, gmv_yuan)；字段缺失时对应项为 None。
    """
    stock = rec.get("stock_amount")
    amount = rec.get("amount")
    sold = stock if isinstance(stock, int) else None  # 成交组齐团=全部售出，售出份数=总份数
    gmv = (sold * float(amount)) if isinstance(sold, int) and amount is not None else None  # amount 已是元
    return sold, gmv


def _agg(rows: list) -> dict:
    """聚合一批已售记录的核心指标。

    Args:
        rows (list): fetch_sold_rows 结果的子集。

    Returns:
        dict: {teams 成团数, merchants 商家数, gmv 总GMV元, avg_unit 均团单价元}。
    """
    teams = len(rows)
    merchants = len({r["corp_info_id"] for r in rows})
    gmv = sum(g for _, g in map(_sold_gmv, rows) if g) or 0.0
    units = [float(r["amount"]) for r in rows if r.get("amount") is not None]
    avg_unit = (sum(units) / len(units)) if units else 0.0
    return {"teams": teams, "merchants": merchants, "gmv": gmv, "avg_unit": avg_unit}


def fetch_sold_rows(pool, win: str) -> list:
    """取指定业务日窗口内的已售全量（含各字段）。

    Args:
        pool: 数据库连接池。
        win (str): WHERE 时间窗条件（WIN_SOLD / WIN_SOLD_YDAY）。

    Returns:
        list: 每元素为 dict，含商家/商品/价格/库存/时间/buy_fetched 等字段（金额为元）。
    """
    recs = pool.select_all(
        "SELECT corp_info_id, corp_info_name, goods_id, goods_name, goods_ip_name, gift_series, product_type, "
        "specification_name, amount, highest_price, lowest_price, stock_amount, residue_stock_amount, "
        "sold_time, soldout_time, finish_time, buy_fetched FROM jw_sold_product_record "
        f"WHERE {win} ORDER BY corp_info_id, gmt_create_time DESC")
    keys = ["corp_info_id", "corp_info_name", "goods_id", "goods_name", "goods_ip_name", "gift_series",
            "product_type", "specification_name", "amount", "highest_price", "lowest_price", "stock_amount",
            "residue_stock_amount", "sold_time", "soldout_time", "finish_time", "buy_fetched"]
    return [dict(zip(keys, r)) for r in recs]


def fetch_buyers_map(log, pool, goods_ids: list) -> dict:
    """批量统计指定商品的购买人数（购买记录去重买家数）。

    数据源 jw_player_record（玩家维度，每商品每买家一行，含真实 userId）。

    Args:
        log: 日志对象。
        pool: 数据库连接池。
        goods_ids (list): 商品ID列表。

    Returns:
        dict: {goods_id: 购买人数}；无购买记录的商品不在字典内（取用时默认 0）。
    """
    if not goods_ids:
        return {}
    ph = ",".join(["%s"] * len(goods_ids))
    rows = pool.select_all(
        f"SELECT goods_id, COUNT(DISTINCT user_id) FROM jw_player_record WHERE goods_id IN ({ph}) GROUP BY goods_id",
        tuple(goods_ids))
    return {r[0]: int(r[1]) for r in rows}


def fetch_corp_distinct_buyers(pool, win: str) -> dict:
    """按商家统计业务日窗口内的去重买家数（跨该商家全部成交团去重人头，非人次）。

    Args:
        pool: 数据库连接池。
        win (str): 时间窗条件（WIN_SOLD）。

    Returns:
        dict: {corp_info_id: 去重买家数}。
    """
    rows = pool.select_all(
        "SELECT s.corp_info_id, COUNT(DISTINCT p.user_id) "
        "FROM jw_sold_product_record s JOIN jw_player_record p ON p.goods_id = s.goods_id "
        f"WHERE {win} GROUP BY s.corp_info_id")
    return {r[0]: int(r[1]) for r in rows}


def fetch_platform_distinct_buyers(pool, win: str) -> int:
    """统计业务日窗口内全平台去重买家数（跨所有成交团去重人头）。

    Args:
        pool: 数据库连接池。
        win (str): 时间窗条件（WIN_SOLD）。

    Returns:
        int: 全平台去重买家数。
    """
    row = pool.select_all(
        "SELECT COUNT(DISTINCT p.user_id) "
        "FROM jw_sold_product_record s JOIN jw_player_record p ON p.goods_id = s.goods_id "
        f"WHERE {win}")
    return int(row[0][0]) if row and row[0][0] is not None else 0


def fetch_completion_hour_dist(pool, days: int) -> list:
    """按成交完成时间的小时(0~23)统计近 days 天成交团数分布。

    Args:
        pool: 数据库连接池。
        days (int): 统计近 N 天。

    Returns:
        list: 长度 24 的列表，索引=小时，值=该小时成交团数。
    """
    since = (datetime.now() - timedelta(days=days)).strftime("%Y-%m-%d")
    dist = [0] * 24
    for h, c in pool.select_all(
            "SELECT HOUR(COALESCE(finish_time, soldout_time)) h, COUNT(*) c FROM jw_sold_product_record "
            "WHERE COALESCE(finish_time, soldout_time) >= %s GROUP BY h", (since,)) or []:
        if h is not None and 0 <= int(h) < 24:
            dist[int(h)] = int(c)
    return dist


def fetch_user_ranking(pool, corp_id: int) -> list:
    """取某重点商家在业务日窗口内的用户消费排行（购买记录 join 已售表算金额）。

    Args:
        pool: 数据库连接池。
        corp_id (int): 商家 corpInfoId。

    Returns:
        list: 每元素 (user_id, user_nick, 参与车数, 参与金额元)，按金额倒序。
    """
    return pool.select_all(
        "SELECT b.user_id, MAX(b.user_nick) nick, COUNT(DISTINCT b.goods_id) cars, "
        "SUM(b.buy_count * s.amount) spent "
        "FROM jw_player_record b JOIN jw_sold_product_record s ON s.goods_id = b.goods_id "
        f"WHERE s.corp_info_id = %s AND {WIN_SOLD} "
        "GROUP BY b.user_id ORDER BY spent DESC", (corp_id,)) or []


def build_overview(wb, all_rows: list, yday_rows: list, platform_buyers: int, window: tuple) -> None:
    """写「平台总览」sheet：平台汇总 + 当日组齐环比 + 商家GMV集中度（对齐 deca 排版）。

    Args:
        wb: 工作簿。
        all_rows (list): 今日业务日窗口内已售全量。
        yday_rows (list): 昨日同窗口已售全量（算环比）。
        platform_buyers (int): 全平台跨团去重买家数（人头，非人次）。
        window (tuple): 业务日窗口 (start, end)。
    """
    t, y = _agg(all_rows), _agg(yday_rows)
    buyers_total = platform_buyers
    per_cap = (t["gmv"] / buyers_total) if buyers_total else 0.0
    ws = xl.add_sheet(wb, "平台总览")

    row = xl.write_title(ws, 1, "集物星球 · 已售每日统计报告")
    row = _line(ws, row, f"成交时间窗 {window[0]} ~ {window[1]}")
    row += 1
    row = xl.write_section_title(ws, row, 4, "平台汇总")
    row = xl.write_kv(ws, row, [
        ("商家数", t["merchants"], "int"),
        ("销售额", round(t["gmv"], 2), "money"),
        ("成团数", t["teams"], "int"),
        ("参与人数", buyers_total, "int"),
        ("均拼单价", round(t["avg_unit"], 2), "money"),
        ("人均消费", round(per_cap, 2), "money"),
    ])
    row += 1

    # 当日组齐环比（数值预格式化成字符串，避免整数被套用金额格式；同列今日/昨日既有金额又有计数）
    def _n(v, money=False):
        return f"{v:,.2f}" if money else f"{int(v)}"
    row = xl.write_section_title(ws, row, 4, "当日组齐环比（vs 昨日同窗口）")
    comp = [
        ("组齐GMV", _n(t["gmv"], True), _n(y["gmv"], True), fmt_pct_change(t["gmv"], y["gmv"])),
        ("成团数", _n(t["teams"]), _n(y["teams"]), fmt_pct_change(t["teams"], y["teams"])),
        ("活跃商家数", _n(t["merchants"]), _n(y["merchants"]), fmt_pct_change(t["merchants"], y["merchants"])),
        ("T均单价", _n(t["avg_unit"], True), _n(y["avg_unit"], True), fmt_pct_change(t["avg_unit"], y["avg_unit"])),
    ]
    row = xl.write_table(ws, row, ["指标", "今日", "昨日", "环比"], comp, ["text", "text", "text", "text"])
    row += 1

    # 商家 GMV 集中度 Top1/3/5/10
    row = xl.write_section_title(ws, row, 4, "商家 GMV 集中度（TopN 占平台组齐总 GMV）")
    merch_gmv = {}
    for r in all_rows:
        _, g = _sold_gmv(r)
        merch_gmv[r["corp_info_id"]] = merch_gmv.get(r["corp_info_id"], 0.0) + (g or 0)
    gmvs = sorted(merch_gmv.values(), reverse=True)
    total = sum(gmvs) or 0
    conc = [(f"Top{k} 集中度", (sum(gmvs[:k]) / total) if total else 0) for k in CONC_TOPS]
    xl.write_table(ws, row, ["集中度档位", "占平台GMV"], conc, ["text", "pct"])


def build_series(wb, all_rows: list) -> None:
    """写「产品系列榜」sheet：按 IP 系列(goods_ip_name)汇总 GMV 倒序取 TopN。

    Args:
        wb: 工作簿。
        all_rows (list): 今日业务日窗口内已售全量。
    """
    agg = {}  # 系列 -> [成团数, GMV]
    for r in all_rows:
        _, g = _sold_gmv(r)
        key = r.get("gift_series") or "(未分类)"
        a = agg.setdefault(key, [0, 0.0])
        a[0] += 1
        a[1] += (g or 0)
    total = sum(v[1] for v in agg.values()) or 0
    ranked = sorted(agg.items(), key=lambda kv: kv[1][1], reverse=True)[:TOP_SERIES]
    data = [(name, cnt, round(g, 2), (g / total if total else 0)) for name, (cnt, g) in ranked]
    ws = xl.add_sheet(wb, "产品系列榜")
    ws.column_dimensions["A"].width = 24  # 系列名较长；同时保证 4 列分区标题条罩住整段标题
    row = xl.write_section_title(ws, 1, 4, f"产品系列销售榜（当日 Top{TOP_SERIES}，按 GMV）")
    xl.write_table(ws, row, ["系列", "成团数", "GMV", "占比"], data, ["text", "int", "money", "pct"])


def build_merchant_rank(wb, all_rows: list) -> None:
    """写「商家GMV榜」sheet：按商家汇总 GMV 倒序取 TopN。

    Args:
        wb: 工作簿。
        all_rows (list): 今日业务日窗口内已售全量。
    """
    agg = {}  # corp_info_id -> [商家名, 成团数, GMV]
    for r in all_rows:
        _, g = _sold_gmv(r)
        a = agg.setdefault(r["corp_info_id"], [r.get("corp_info_name"), 0, 0.0])
        a[1] += 1
        a[2] += (g or 0)
    total = sum(v[2] for v in agg.values()) or 0
    ranked = sorted(agg.values(), key=lambda v: v[2], reverse=True)[:TOP_MERCHANT]
    data = [(name, cnt, round(g, 2), (g / total if total else 0)) for name, cnt, g in ranked]
    ws = xl.add_sheet(wb, "商家GMV榜")
    ws.column_dimensions["A"].width = 22  # 商家名较长；同时保证 4 列分区标题条罩住整段标题
    row = xl.write_section_title(ws, 1, 4, f"商家 GMV 榜（当日组齐口径，前 {TOP_MERCHANT}）")
    xl.write_table(ws, row, ["商家", "成团数", "GMV", "占比"], data, ["text", "int", "money", "pct"])


def build_ops(wb, pool, all_rows: list, window: tuple) -> None:
    """写「运营节奏」sheet：重点商家当日运营快照 + 平台成交时段分布（对齐 deca）。

    Args:
        wb: 工作簿。
        pool: 数据库连接池。
        all_rows (list): 今日业务日窗口内已售全量。
        window (tuple): 业务日窗口 (start, end)，用于判「今日新开团」。
    """
    ws = xl.add_sheet(wb, "运营节奏")
    try:
        w_start = datetime.strptime(window[0], "%Y-%m-%d %H:%M:%S")
        w_end = datetime.strptime(window[1], "%Y-%m-%d %H:%M:%S")
    except (ValueError, TypeError):
        w_start = w_end = None

    row = xl.write_section_title(ws, 1, 4, "重点商家当日运营快照（新开团 / 已组齐 / 规格）")
    snap = []
    for cid, name, _ in FOCUS_CORPS:
        sub = [r for r in all_rows if r["corp_info_id"] == cid]
        new_open = sum(1 for r in sub if w_start and isinstance(r.get("sold_time"), datetime)
                       and w_start <= r["sold_time"] <= w_end)
        specs = {}
        for r in sub:
            s = r.get("specification_name")
            if s:
                specs[s] = specs.get(s, 0) + 1
        spec_txt = " · ".join(f"{k}×{v}" for k, v in sorted(specs.items(), key=lambda x: -x[1])) or "—"
        snap.append((name, new_open, len(sub), spec_txt))
    row = xl.write_table(ws, row, ["商家", "今日新开团", "已组齐", "规格分布"], snap,
                         ["text", "int", "int", "text"])
    row += 1

    row = xl.write_section_title(ws, row, 4, f"平台成交时段分布（近 {HOUR_DIST_DAYS} 日 24h 累计）")
    dist = fetch_completion_hour_dist(pool, HOUR_DIST_DAYS)
    vmax = max(dist) if dist else 0
    data = [(f"{h:02d}时", dist[h], _bar(dist[h], vmax)) for h in range(24)]
    xl.write_table(ws, row, ["时段", "成团数", "分布"], data, ["text", "int", "text"])


def build_focus_detail(wb, name: str, sheet_name: str, rank_sheet: str, rows: list,
                       buyers_map: dict, win_text: str, corp_buyers_n: int) -> None:
    """写某重点商家「明细」sheet：汇总 + 购买记录覆盖检测 + 每条组队明细（对齐 deca）。

    Args:
        wb: 工作簿。
        name (str): 商家展示名。
        sheet_name (str): 本 sheet 名。
        rank_sheet (str): 对应的用户排行榜 sheet 名（覆盖检测文案里指引）。
        rows (list): 该商家今日已售记录。
        buyers_map (dict): 各商品的参与人数映射（供明细列「参与人数(本团)」逐团展示）。
        win_text (str): 成交时间窗文案 "start ~ end"（写进汇总标题）。
        corp_buyers_n (int): 该商家跨团去重买家数（人头，供汇总块「参与人数(真实买家)」）。
    """
    ws = xl.add_sheet(wb, sheet_name)
    total_gmv = sum(g for _, g in map(_sold_gmv, rows) if g) or 0.0
    total_buyers = corp_buyers_n  # 跨团去重人头（汇总口径），明细列的「参与人数(本团)」另用 buyers_map 逐团
    units = [float(r["amount"]) for r in rows if r.get("amount") is not None]
    avg_unit = (sum(units) / len(units)) if units else 0.0
    per_cap = (total_gmv / total_buyers) if total_buyers else 0.0

    row = xl.write_section_title(ws, 1, 12, f"{name} · 汇总（成交时间窗 {win_text}）")
    row = xl.write_kv(ws, row, [
        ("销售额", round(total_gmv, 2), "money"),
        ("成团数", len(rows), "int"),
        ("参与人数(真实买家)", total_buyers, "int"),
        ("均拼单价", round(avg_unit, 2), "money"),
        ("人均消费", round(per_cap, 2), "money"),
    ])
    row += 1

    # 购买记录覆盖检测：buy_fetched=1 表示该团购买记录已采全
    fetched = sum(1 for r in rows if r.get("buy_fetched") == 1)
    missing = [r for r in rows if r.get("buy_fetched") != 1]
    row = xl.write_section_title(ws, row, 12, "购买记录覆盖检测（成交团 vs 已采购买记录）")
    row = _line(ws, row, f"成交 {len(rows)} 团 · 采到购买记录 {fetched} 团 · 漏采 {len(missing)} 团"
                         f"（用户排行见「{rank_sheet}」sheet）")
    if missing:
        row = _line(ws, row, "漏采明细（下列团未采到购买记录，未计入用户排行）：")
        for m in missing:
            row = _line(ws, row, f"　- {m.get('goods_id')}　{m.get('goods_name') or ''}")
    row += 1

    row = xl.write_section_title(ws, row, 12, f"每条组队明细（共 {len(rows)} 条，按总金额倒序）")
    headers = ["序号", "团名(商品标题)", "系列", "类型", "单价", "总份数", "进度%", "总金额",
               "参与人数(本团)", "开售时间", "成交时间", "售卖时长"]
    col_types = ["int", "text", "text", "text", "money", "int", "pct", "money",
                 "int", "text", "text", "text"]
    detailed = sorted(rows, key=lambda r: (_sold_gmv(r)[1] or 0), reverse=True)
    data = []
    for i, r in enumerate(detailed, 1):
        sold, gmv = _sold_gmv(r)
        stock = r.get("stock_amount")
        prog = (sold / stock) if isinstance(sold, int) and isinstance(stock, int) and stock > 0 else 0
        ct = r.get("finish_time") or r.get("soldout_time")  # 成交时间
        data.append((i, r.get("goods_name"), r.get("gift_series"),
                     PRODUCT_TYPE_NAME.get(str(r.get("product_type")), r.get("product_type")),
                     yuan(r.get("amount")), stock, prog, gmv,
                     buyers_map.get(r["goods_id"], 0),
                     fmt_dt(r.get("sold_time")), fmt_dt(ct), _duration(r.get("sold_time"), ct)))
    xl.write_table(ws, row, headers, data, col_types)


def build_user_ranking(wb, name: str, sheet_name: str, corp_id: int, pool) -> None:
    """写某重点商家「用户排行榜」sheet：买家按消费金额倒序（对齐 deca）。

    Args:
        wb: 工作簿。
        name (str): 商家展示名。
        sheet_name (str): 工作表名。
        corp_id (int): 商家 corpInfoId。
        pool: 数据库连接池。
    """
    rows = fetch_user_ranking(pool, corp_id)
    data = []
    for i, (uid, nick, cars, spent) in enumerate(rows, 1):
        spent_f = float(spent) if spent is not None else 0.0
        avg = (spent_f / cars) if cars else 0.0
        data.append((i, nick or "-", uid, cars, round(spent_f, 2), round(avg, 2)))
    ws = xl.add_sheet(wb, sheet_name)
    row = xl.write_section_title(
        ws, 1, 6, f"用户排行榜 · {name}（共 {len(rows)} 人参与，全部展示，按参与金额倒序）")
    row = xl.write_table(ws, row, ["排名", "用户昵称", "user_id", "参与车数", "参与金额", "车均消费"],
                         data, ["int", "text", "text", "int", "money", "money"])
    _line(ws, row, "注：参与金额 = Σ(购买份数 × 团单价)；参与车数 = 参与的不同团数；车均消费 = 参与金额 ÷ 参与车数。")


def build_others(wb, rows: list, corp_buyers: dict) -> None:
    """写「其他商家」sheet：非重点商家按商家汇总（销售额倒序，对齐 deca；人数用真实去重买家）。

    Args:
        wb: 工作簿。
        rows (list): 非重点商家的今日已售记录。
        corp_buyers (dict): {corp_info_id: 跨团去重买家数}（人头）。
    """
    agg = {}  # corp_info_id -> [商家名, 成团数, GMV, [单价...]]
    for r in rows:
        _, g = _sold_gmv(r)
        a = agg.setdefault(r["corp_info_id"], [r.get("corp_info_name"), 0, 0.0, []])
        a[1] += 1
        a[2] += (g or 0)
        a[3].append(r.get("amount"))
    data = []
    for cid, (name, cnt, g, amts) in agg.items():
        buyers = corp_buyers.get(cid, 0)  # 跨团去重人头
        units = [float(a) for a in amts if a is not None]
        avg = (sum(units) / len(units)) if units else 0.0
        per = (g / buyers) if buyers else 0.0
        data.append((name, round(g, 2), cnt, buyers, round(avg, 2), round(per, 2)))
    data.sort(key=lambda x: x[1], reverse=True)
    ws = xl.add_sheet(wb, "其他商家")
    row = xl.write_section_title(ws, 1, 6, f"其他商家汇总（共 {len(data)} 家，按销售额倒序）")
    xl.write_table(ws, row, ["商家名", "销售额", "成团数", "参与人数", "均拼单价", "人均消费"],
                   data, ["text", "money", "int", "int", "money", "money"])


def build_report(log, pool, out_file: str) -> None:
    """汇总 9 个 sheet 生成已售日报 Excel（排版对齐 deca）。

    Args:
        log: 日志对象。
        pool: 数据库连接池。
        out_file (str): 输出文件路径。
    """
    all_rows = fetch_sold_rows(pool, WIN_SOLD)
    yday_rows = fetch_sold_rows(pool, WIN_SOLD_YDAY)
    window = get_window(pool)
    win_text = f"{window[0]} ~ {window[1]}"  # 明细 sheet 标题复用
    buyers_map = fetch_buyers_map(log, pool, [r["goods_id"] for r in all_rows])  # 各团去重买家(明细列用)
    corp_buyers = fetch_corp_distinct_buyers(pool, WIN_SOLD)                     # 各商家跨团去重人头
    platform_buyers = fetch_platform_distinct_buyers(pool, WIN_SOLD)            # 全平台跨团去重人头
    focus_ids = {cid for cid, _, _ in FOCUS_CORPS}

    wb = xl.new_workbook()
    build_overview(wb, all_rows, yday_rows, platform_buyers, window)     # 1 平台总览
    build_series(wb, all_rows)                                           # 2 产品系列榜
    build_merchant_rank(wb, all_rows)                                    # 3 商家GMV榜
    build_ops(wb, pool, all_rows, window)                                # 4 运营节奏
    for cid, name, prefix in FOCUS_CORPS:                                # 5-8 重点商家明细+用户榜
        sub = [r for r in all_rows if r["corp_info_id"] == cid]
        build_focus_detail(wb, name, f"{prefix}明细", f"{prefix}用户排行榜", sub,
                           buyers_map, win_text, corp_buyers.get(cid, 0))
        build_user_ranking(wb, name, f"{prefix}用户排行榜", cid, pool)
    others = [r for r in all_rows if r["corp_info_id"] not in focus_ids]
    build_others(wb, others, corp_buyers)                               # 9 其他商家
    xl.save(wb, out_file)


def run_once(log) -> str:
    """生成一次已售日报并发企微（只读库；DB 异常则跳过本轮）。

    Args:
        log: 日志对象。

    Returns:
        str: 生成的 Excel 路径；失败返回空串。
    """
    log.info("开始生成已售日报" + "." * 30)
    pool = MySQLConnectionPool(log=log)
    if not pool.check_pool_health():
        log.error("数据库连接池异常，跳过本轮")
        return ""
    os.makedirs(OUT_DIR, exist_ok=True)
    out_file = os.path.abspath(os.path.join(OUT_DIR, f"{OUT_PREFIX}_{date.today():%Y%m%d}.xlsx"))
    try:
        build_report(log, pool, out_file)
        log.success(f"已售日报已生成: {out_file}")
    except Exception as e:
        log.error(f"生成已售日报失败: {e}")
        return ""
    if SEND_WECHAT:
        wx.send_wechat_group_file(log=log, file_path=out_file)
    return out_file


def schedule_task():
    """定时入口：每天 REPORT_TIME 生成并发送已售日报。"""
    # run_once(logger)  # 调试时取消注释立即跑一次
    schedule.every().day.at(REPORT_TIME).do(run_once, logger)
    while True:
        schedule.run_pending()
        time.sleep(1)


if __name__ == "__main__":
    schedule_task()