spider_work/deca_spider/stats/daily_report.py

# -*- coding: utf-8 -*-
# Author : Charley
# Python : 3.12.10
# Date   : 2026/08/05
"""得卡 DECA · 已售每日统计报告（Excel 多 Sheet，每一大项一个 sheet）。

统计范围：成交完成时间 completed_at 落在 [昨天 17:00:00, 今天 06:00:00]（含两端）的已售商品。
一个拼团商品 = 一个「组队」（组队售卖），故「每条明细」= 每个拼团商品。

报告结构（2026/08/14 由单 Sheet 分区改为多 Sheet，每 sheet 独立列宽、蓝条只覆盖本表宽度）：
    Sheet 平台总览 ：平台汇总 + 当日组齐环比(vs 昨日同窗口) + 商家 GMV 集中度(Top1/3/5/10) + 口径脚注
    Sheet 产品系列榜：当日各系列 GMV 榜(Top，含占比)
    Sheet 商家GMV榜 ：当日组齐 GMV 前 N 商家(含占比)
    Sheet 运营节奏 ：重点商家当日运营快照(新开团/已组齐/规格) + 平台组齐时段分布(近7日24h)
    Sheet 魔都明细 ：881226408 汇总 + 每条明细（含「参与人数(购买记录)」与售卖进度里程碑列；
                    汇总下附「购买记录覆盖检测」= 成交团 vs 已采购买记录，标注漏采多少 T(团)）
    Sheet 用户排行榜(魔都)：881226408 买家榜（deca_buy_record 按 user_id 聚合，参与金额倒序，
                    含 参与车数 / 参与金额 / 车均消费）(2026/08/17 新增)
    Sheet 卡皇明细 ：274584650 汇总 + 每条明细
    Sheet 其他商家 ：其余商家各一行汇总（中卡近似口径）
    注：原「魔都已售进度检测」独立 sheet 已于 2026/08/11 并入魔都明细（尾部到 25/50/75% 用时列）。

口径说明：
    - 销售额 = SUM(COALESCE(team_total_amount, sold_count * unit_price))
              随机团(选队随机/剩余随机)按 teams 逐队精算(team_total_amount，2026/08/11 起，
              见 docs/选队随机与剩余随机_总价口径与采集_20260811.md)；固定价团回落原公式。
    - 成团数 = 该时段成交的拼团商品数
    - 参与人数(魔都汇总 & 明细口径) = deca_buy_record 去重买家 user_id（真实参团人头；仅 881226408
                 采了购买记录）。魔都汇总「参与人数(真实买家)」= 跨其全部成交团去重（2026/08/14 起由
                 中卡近似切为真实买家）；明细「参与人数(本团)」= 各团单独去重，故明细逐团相加(人次) ≥ 汇总。
    - 中卡用户数(近似)(平台大盘/其他商家口径) = 拆卡报告 hit_user_nickname 去重（仅覆盖 report_state=1
                 有报告的商品；这些商家未采购买记录，只能用中卡用户近似，非真实参团人头，偏低）
    - 均拼单价 = 销售额 / 成团数
    - 人均消费 = 销售额 / 参与人数（魔都为真实买家；平台/其他商家为按中卡近似，偏高，仅供参考）
    - 卡密表 deca_kami_record 当前为空(FILL_KAMI 关)，故无「球队」维度，明细按商品维度出。

从项目根目录运行：python stats/daily_report.py（cwd=根目录，读根目录 application.yml）
"""
import os
import sys
import time
# 把项目根目录加入 import 路径：企微发送模块 auto_send_wx_msg.py 只在根目录留一份（WEBHOOK_URL 单点维护）
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from datetime import date, datetime

import schedule
from loguru import logger
from openpyxl import Workbook
from openpyxl.styles import Font, Alignment, PatternFill, Border, Side
from mysql_pool import MySQLConnectionPool

# 日志：按天切分文件，保留 7 天（常驻定时运行）
logger.remove()
logger.add("./logs/daily_report_{time:YYYYMMDD}.log", encoding="utf-8", rotation="00:00",
           format="[{time:YYYY-MM-DD HH:mm:ss.SSS}] {level} {message}", level="INFO", retention="7 day")

# 企微发送：报告生成后把 Excel 发到企业微信群机器人（只发表格，不发图；群由 auto_send_wx_msg.WEBHOOK_URL 决定）
SEND_WECHAT = True

# 两个要出「汇总 + 明细」的重点商家；其余商家统一进「其他商家汇总」
FOCUS_MERCHANTS = ["881226408", "274584650"]
# 魔都兄弟球星卡：其每条明细走扩展版——多「参与人数」列(deca_buy_record 去重买家)，
# 尾部并入售卖进度里程碑(从 progress 表算到 25/50/75% 各用了多久；首张快照已越过阈值则留空)。仅本商家如此。
MODDU_MID = "881226408"
OUT_PREFIX = "得卡已售每日报告"     # 输出文件名前缀，实际文件名后缀加运行当天日期

# 时间窗过滤（p 别名）：[昨天17:00, 今天06:00]（2026/08/15 由 03:00 延到 06:00，凌晨仍在播）
WIN_P = ("p.completed_at >= (CURDATE() - INTERVAL 1 DAY) + INTERVAL 17 HOUR "
         "AND p.completed_at <= CURDATE() + INTERVAL 6 HOUR")

# 明细表列规格：(表头, 明细dict取值键, 是否千分位金额格式)；取值键为「序号」时按行号自增填充。
# 开售时间/成交时间/售卖时长 三列相邻：售卖时长=成交-开售。
DETAIL_COLS = [
    ("序号", "序号", False), ("团名(商品标题)", "团名", False),
    ("系列", "系列", False), ("类型", "类型", False), ("单价", "单价", True),
    ("总份数", "总份数", False), ("进度%", "进度", False), ("总金额", "总金额", True),
    ("中卡人数", "中卡人数", False),   # 该团拆卡报告 hit_user_nickname 去重(中卡近似)，放开售时间前
    ("开售时间", "开售时间", False), ("成交时间", "成交时间", False), ("售卖时长", "售卖时长", False),
]
# 魔都(881226408)专属明细：在「中卡人数」前插「参与人数」(deca_buy_record 去重买家 user_id)，
# 并在尾部并入售卖进度里程碑（到 25/50/75% 用时，源 deca_onsale_product_progress_record）。
# 原「魔都已售进度检测」独立 sheet 于 2026/08/11 并入本明细，不再单独出 sheet。
MODDU_DETAIL_COLS = [
    ("序号", "序号", False), ("团名(商品标题)", "团名", False),
    ("系列", "系列", False), ("类型", "类型", False), ("单价", "单价", True),
    ("总份数", "总份数", False), ("进度%", "进度", False), ("总金额", "总金额", True),
    ("参与人数(本团)", "参与人数", False),
    ("中卡人数", "中卡人数", False),   # 该团拆卡报告 hit_user_nickname 去重(中卡近似)，放开售时间前
    ("开售时间", "开售时间", False), ("成交时间", "成交时间", False), ("售卖时长", "售卖时长", False),
    ("到25%用时", "到25%用时", False), ("到50%用时", "到50%用时", False),
    ("到75%用时", "到75%用时", False),
]
# 分区标题/大标题横向着色跨列数：取最宽的魔都明细，保证蓝条覆盖整表宽
SECTION_SPAN = len(MODDU_DETAIL_COLS)
# 汇总表指标键（商家/平台，dict 取值键，与显示标签解耦）
SUMMARY_HEADERS = ["销售额", "成团数", "参与人数", "均拼单价", "人均消费"]

# 平台大盘竖排汇总行：(显示标签, dict取值键)。参与人数为「魔都真实买家 + 其他商家中卡去重」
# 的混合口径（2026/08/14 起，见 fetch_platform_summary），故标签显式标注，避免误当纯真实人头。
PLATFORM_ROWS = [
    ("商家数", "商家数"), ("销售额", "销售额"), ("成团数", "成团数"),
    ("参与人数(魔都真实+其他中卡)", "参与人数"), ("均拼单价", "均拼单价"),
    ("人均消费", "人均消费"),
]

# 其他商家汇总表列：(显示表头, dict取值键, 是否金额格式)。这些商家未采购买记录，参与人数
# 只能用中卡用户近似，故表头标注「(近似)」，与魔都真实买家口径区分。
OTHER_COLS = [
    ("商家名", "商家名", False), ("销售额", "销售额", True), ("成团数", "成团数", False),
    ("中卡用户数(近似)", "参与人数", False), ("均拼单价", "均拼单价", True),
    ("人均消费(按中卡近似)", "人均消费", True),
]

# ---- 监测清单增强(2026/08/14)：环比 / 排行 / 集中度 / 时段分布 配置 ----
# 昨日同窗口(用于组齐环比)：[前天17:00, 昨天06:00]，与 WIN_P 整体平移一天、口径一致
WIN_P_YDAY = ("p.completed_at >= (CURDATE() - INTERVAL 2 DAY) + INTERVAL 17 HOUR "
              "AND p.completed_at <= (CURDATE() - INTERVAL 1 DAY) + INTERVAL 6 HOUR")
TOP_SERIES = 15              # 产品系列销售榜展示条数
TOP_MERCHANT = 10            # 商家 GMV 榜展示条数(监测清单要「GMV前十商家」)
TOP_USERS = None              # 魔都用户排行榜展示条数(按参与金额倒序取前 N；仅魔都采了购买记录)
CONC_TOPS = (1, 3, 5, 10)    # GMV 集中度统计的 TopN 档(Top1/3/5/10 占平台总 GMV)
HOUR_DIST_DAYS = 7           # 组齐时段分布回看天数(反映平台 24h 组齐节奏)

# ---- 样式 ----
FONT_TITLE = Font(name="Microsoft YaHei", bold=True, size=14, color="1F2A44")
FONT_SECTION = Font(name="Microsoft YaHei", bold=True, size=12, color="FFFFFF")
FONT_HEADER = Font(name="Microsoft YaHei", bold=True, color="1F3864")   # 表头：深蓝字
FONT_CELL = Font(name="Microsoft YaHei", size=10)                        # 正文
FILL_SECTION = PatternFill("solid", fgColor="4472C4")   # 分区标题：蓝底白字
FILL_HEADER = PatternFill("solid", fgColor="D9E1F2")     # 表头：浅蓝底
FILL_SUBTOTAL = PatternFill("solid", fgColor="FCE4D6")   # 汇总行：浅橙底
FILL_ZEBRA = PatternFill("solid", fgColor="F5F8FC")      # 明细隔行：淡蓝灰
THIN = Side(style="thin", color="D6DCE5")
BORDER = Border(left=THIN, right=THIN, top=THIN, bottom=THIN)
MONEY_FMT = "#,##0.00"
INT_FMT = "#,##0"                                        # 计数列千分位
ALIGN_LEFT = Alignment(horizontal="left", vertical="center")     # 文字：左对齐
ALIGN_RIGHT = Alignment(horizontal="right", vertical="center")   # 数字/金额：右对齐
ALIGN_CENTER = Alignment(horizontal="center", vertical="center", wrap_text=True)  # 表头：居中换行


FONT_UP = Font(name="Microsoft YaHei", bold=True, color="1E7D32")    # 环比涨：绿
FONT_DOWN = Font(name="Microsoft YaHei", bold=True, color="C62828")  # 环比跌：红


def _align_by_value(value) -> Alignment:
    """按单元格值类型选对齐方式：数字(含金额/占比)右对齐，其余(文字/时长/日期)左对齐。

    Args:
        value: 单元格值（int/float/str/None 等）。

    Returns:
        Alignment: 数字返回右对齐，其它返回左对齐（bool 视为非数字）。
    """
    if isinstance(value, bool):
        return ALIGN_LEFT
    return ALIGN_RIGHT if isinstance(value, (int, float)) else ALIGN_LEFT


def _trend_font(text: str) -> Font:
    """按环比文本的正负号返回涨跌配色字体（涨绿、跌红、其它常规）。

    Args:
        text (str): 环比字符串（如 "+12.3%"/"-8.0%"/"新增"/"—"）。

    Returns:
        Font: 以 "+" 开头或"新增"→绿；"-" 开头→红；否则常规正文字体。
    """
    if not text:
        return FONT_CELL
    if text.startswith("+") or text == "新增":
        return FONT_UP
    if text.startswith("-"):
        return FONT_DOWN
    return FONT_CELL


def get_window(pool) -> tuple[str, str]:
    """取当前时间窗的起止边界(用于报告标题展示)。

    Args:
        pool (MySQLConnectionPool): MySQL 连接池。

    Returns:
        tuple[str, str]: (起点datetime字符串, 终点datetime字符串)。
    """
    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 fetch_platform_summary(pool) -> dict:
    """统计平台大盘汇总(时间窗内全部已售商品)。

    参与人数为混合口径（2026/08/14 起）：魔都(881226408)采了真实购买记录，用 deca_buy_record
    去重真实买家；其余商家未采购买记录，仍用拆卡报告 hit_user_nickname 去重的中卡用户近似。
    两部分人群标识不同(魔都=user_id，其他=昵称)、无法跨口径去重，故直接相加，属近似上界。

    Args:
        pool (MySQLConnectionPool): MySQL 连接池。

    Returns:
        dict: 含 销售额/商家数/成团数/参与人数/均拼单价/人均消费 六项。
    """
    sql = f"""
    SELECT
        ROUND(SUM(COALESCE(p.team_total_amount, p.sold_count * p.unit_price)), 2) AS amount,
        COUNT(DISTINCT p.merchant_user_id)                               AS merchants,
        COUNT(*)                                                          AS grp,
        -- 其他商家(非魔都)中卡用户去重；魔都单独用真实买家，不计入此子查询
        (SELECT COUNT(DISTINCT r.hit_user_nickname)
           FROM deca_report_record r
           JOIN deca_product_record pp ON pp.product_code = r.product_code
          WHERE pp.merchant_user_id <> %s
            AND pp.completed_at >= (CURDATE() - INTERVAL 1 DAY) + INTERVAL 17 HOUR
            AND pp.completed_at <= CURDATE() + INTERVAL 6 HOUR
            AND r.hit_user_nickname IS NOT NULL AND r.hit_user_nickname <> '') AS others_people
    FROM deca_product_record p
    WHERE {WIN_P} AND p.unit_price IS NOT NULL AND p.sold_count IS NOT NULL
    """
    amount, merchants, groups, others_people = pool.select_all(sql, (MODDU_MID,))[0]
    # 平台参与人数 = 魔都真实买家(deca_buy_record 去重) + 其他商家中卡用户去重
    people = _fetch_real_buyers(pool, MODDU_MID) + (others_people or 0)
    return _pack_summary(amount, groups, people, extra={"商家数": merchants})


def fetch_merchant_summary(pool, mid: str) -> dict:
    """统计单个商家的汇总(时间窗内)。

    Args:
        pool (MySQLConnectionPool): MySQL 连接池。
        mid (str): 商家 merchant_user_id。

    Returns:
        dict: 含 商家名/销售额/成团数/参与人数/均拼单价/人均消费。
    """
    sql = f"""
    SELECT
        MAX(p.merchant_name)                                              AS mname,
        ROUND(SUM(COALESCE(p.team_total_amount, p.sold_count * p.unit_price)), 2) AS amount,
        COUNT(*)                                                          AS grp,
        (SELECT COUNT(DISTINCT r.hit_user_nickname)
           FROM deca_report_record r
           JOIN deca_product_record pp ON pp.product_code = r.product_code
          WHERE pp.merchant_user_id = %s
            AND pp.completed_at >= (CURDATE() - INTERVAL 1 DAY) + INTERVAL 17 HOUR
            AND pp.completed_at <= CURDATE() + INTERVAL 6 HOUR
            AND r.hit_user_nickname IS NOT NULL AND r.hit_user_nickname <> '') AS people
    FROM deca_product_record p
    WHERE p.merchant_user_id = %s AND {WIN_P}
      AND p.unit_price IS NOT NULL AND p.sold_count IS NOT NULL
    """
    row = pool.select_all(sql, (mid, mid))
    mname, amount, groups, people = row[0] if row else (None, None, 0, 0)
    # 魔都(881226408)采了真实购买记录：参与人数改用 deca_buy_record 去重真实买家，人均消费随之
    # 按真实人头计（覆盖上面 people 的中卡近似值）；其余重点商家无购买记录，仍沿用中卡近似。
    if mid == MODDU_MID:
        people = _fetch_real_buyers(pool, mid)
    d = _pack_summary(amount, groups, people)
    d["商家名"] = mname or mid
    d["商家ID"] = mid
    return d


def fetch_merchant_details(pool, mid: str) -> list[dict]:
    """取单个商家时间窗内每个拼团(组队)的明细，按总金额倒序。

    Args:
        pool (MySQLConnectionPool): MySQL 连接池。
        mid (str): 商家 merchant_user_id。

    Returns:
        list[dict]: 每元素为一条明细(团名/系列/类型/单价/份数/总份数/进度/总金额/中卡人数/开售时间/成交时间/售卖时长/回放)。
    """
    sql = f"""
    SELECT
        p.title, p.series_name, p.spec_name, p.unit_price, p.sold_count, p.card_count,
        ROUND(COALESCE(p.team_total_amount, p.sold_count * p.unit_price), 2) AS amount,
        p.completed_at, p.sale_start_at,
        TIMESTAMPDIFF(SECOND, p.sale_start_at, p.completed_at)            AS duration_secs,
        -- 中卡人数：该团拆卡报告去重命中用户(hit_user_nickname)，中卡近似口径
        (SELECT COUNT(DISTINCT r.hit_user_nickname) FROM deca_report_record r
          WHERE r.product_code = p.product_code
            AND r.hit_user_nickname IS NOT NULL AND r.hit_user_nickname <> '') AS hit_users,
        p.replay_url
    FROM deca_product_record p
    WHERE p.merchant_user_id = %s AND {WIN_P}
      AND p.unit_price IS NOT NULL AND p.sold_count IS NOT NULL
    ORDER BY amount DESC
    """
    rows = pool.select_all(sql, (mid,)) or []
    result = []
    for (title, series, spec, price, sold, card, amount,
         completed, sale_start, duration_secs, hit_users, replay) in rows:
        progress = round(sold / card * 100, 1) if card else None  # 售卖进度百分比
        result.append({
            "团名": title, "系列": series, "类型": spec, "单价": price,
            "份数": sold, "总份数": card, "进度": progress, "总金额": amount,
            "中卡人数": hit_users,                             # 该团拆卡报告去重命中用户(中卡近似)
            "开售时间": sale_start, "成交时间": completed,   # 开售=sale_start_at，成交=completed_at
            "售卖时长": _fmt_duration(duration_secs),         # 差值=成交-开售，反映多快卖完
            "回放": replay,
        })
    return result


def fetch_other_merchants(pool, exclude: list) -> list[dict]:
    """统计除重点商家外的其他商家汇总(时间窗内)，每商家一行，按销售额倒序。

    Args:
        pool (MySQLConnectionPool): MySQL 连接池。
        exclude (list[str]): 要排除的重点商家 ID 列表。

    Returns:
        list[dict]: 每元素为一个商家的汇总(商家名/销售额/成团数/参与人数/均拼单价/人均消费)。
    """
    placeholders = ",".join(["%s"] * len(exclude))
    sql = f"""
    SELECT
        p.merchant_user_id                                                AS mid,
        MAX(p.merchant_name)                                              AS mname,
        ROUND(SUM(COALESCE(p.team_total_amount, p.sold_count * p.unit_price)), 2) AS amount,
        COUNT(*)                                                          AS grp,
        (SELECT COUNT(DISTINCT r.hit_user_nickname)
           FROM deca_report_record r
           JOIN deca_product_record pp ON pp.product_code = r.product_code
          WHERE pp.merchant_user_id = p.merchant_user_id
            AND pp.completed_at >= (CURDATE() - INTERVAL 1 DAY) + INTERVAL 17 HOUR
            AND pp.completed_at <= CURDATE() + INTERVAL 6 HOUR
            AND r.hit_user_nickname IS NOT NULL AND r.hit_user_nickname <> '') AS people
    FROM deca_product_record p
    WHERE {WIN_P} AND p.merchant_user_id NOT IN ({placeholders})
      AND p.unit_price IS NOT NULL AND p.sold_count IS NOT NULL
    GROUP BY p.merchant_user_id
    ORDER BY amount DESC
    """
    rows = pool.select_all(sql, tuple(exclude)) or []
    result = []
    for mid, mname, amount, groups, people in rows:
        d = _pack_summary(amount, groups, people)
        d["商家名"] = mname or mid
        result.append(d)
    return result


def _pack_summary(amount, groups, people, extra: dict = None) -> dict:
    """把销售额/成团数/参与人数打包成统一汇总 dict，并算均拼单价、人均消费。

    Args:
        amount (Decimal | None): 销售额。
        groups (int | None): 成团数。
        people (int | None): 参与人数。
        extra (dict, optional): 额外要合并的字段(如商家数)。Defaults to None。

    Returns:
        dict: 含 销售额/成团数/参与人数/均拼单价/人均消费(+extra)。
    """
    amount = float(amount) if amount is not None else 0.0
    groups = groups or 0
    people = people or 0
    avg_group = round(amount / groups, 2) if groups else None  # 均拼单价=销售额/成团数
    avg_person = round(amount / people, 2) if people else None  # 人均消费=销售额/参与人数
    d = {"销售额": round(amount, 2), "成团数": groups, "参与人数": people,
         "均拼单价": avg_group, "人均消费": avg_person}
    if extra:
        d.update(extra)
    return d


def _fetch_real_buyers(pool, mid: str) -> int:
    """查某商家时间窗内 deca_buy_record 去重真实买家数(跨其全部成交团)。

    仅魔都(881226408)采了真实购买记录，故只有它能用此口径；其余商家该表无数据、返回 0。

    Args:
        pool (MySQLConnectionPool): MySQL 连接池。
        mid (str): 商家 merchant_user_id。

    Returns:
        int: 去重真实买家数；无数据时返回 0。
    """
    sql = """
    SELECT COUNT(DISTINCT b.user_id)
    FROM deca_buy_record b
    JOIN deca_product_record pp ON pp.product_code = b.product_code
    WHERE pp.merchant_user_id = %s
      AND pp.completed_at >= (CURDATE() - INTERVAL 1 DAY) + INTERVAL 17 HOUR
      AND pp.completed_at <= CURDATE() + INTERVAL 6 HOUR
    """
    row = pool.select_all(sql, (mid,))
    return int(row[0][0]) if row and row[0][0] is not None else 0


def _summary_rows(is_real: bool) -> list[tuple]:
    """按参与人数口径生成重点商家竖排汇总的(显示标签, 取值键)行规格。

    Args:
        is_real (bool): True=该商家参与人数为 deca_buy_record 真实买家(魔都)，标签用
            「参与人数(真实买家)」；False=中卡用户近似，标签用「中卡用户数(近似)」，人均消费
            标签相应标注「(按中卡近似)」。

    Returns:
        list[tuple[str, str]]: 每项 (显示标签, data 取值键)。
    """
    people = ("参与人数(真实买家)" if is_real else "中卡用户数(近似)", "参与人数")
    person_avg = ("人均消费" if is_real else "人均消费(按中卡近似)", "人均消费")
    return [("销售额", "销售额"), ("成团数", "成团数"), people,
            ("均拼单价", "均拼单价"), person_avg]


def fetch_moddu_details(pool, mid: str) -> list[dict]:
    """取「魔都」商家(mid)时间窗内每个拼团(组队)的扩展明细，按总金额倒序。

    在标准明细基础上多两类字段（原「魔都已售进度检测」独立 sheet 于 2026/08/11 并入此处）：
        - 参与人数：deca_buy_record 去重买家 user_id(真实参团人头；仅本商家采了购买记录)。
        - 到 25/50/75% 用时：用 deca_onsale_product_progress_record(分钟级进度序列)里
          `MIN(captured_at) WHERE progress_pct >= X` 减开售时间 sale_start_at 得到。
          仅当该商品在 tX 之前还有更早快照(证明从低于 X% 就开始观测)时才算真实穿越耗时；
          若首张快照 pct 已 ≥ X(坍缩)则该列留空，不输出「首次被监控时刻−开售」的假值。
          原「到100%用时」已于 2026/08/12 删除：满仓即下架、轮询抓不到 100% 快照，且与售卖时长重复。

    进度里程碑仅"progress 表上线(2026/08/11)后覆盖到售卖过程"的团才有值；更早售卖或首张
    快照已越过阈值的团查不到/不可信，相应列留空——商品行本身仍显示(总金额/成交时间等基础字段)。

    Args:
        pool (MySQLConnectionPool): MySQL 连接池。
        mid (str): 商家 merchant_user_id。

    Returns:
        list[dict]: 每条含 团名/系列/类型/单价/总份数/进度/总金额/参与人数(购买记录)/中卡人数/
                    开售时间/成交时间/售卖时长/到25%用时/到50%用时/到75%用时。
    """
    sql = f"""
    SELECT
        p.title, p.series_name, p.spec_name, p.unit_price, p.sold_count, p.card_count,
        ROUND(COALESCE(p.team_total_amount, p.sold_count * p.unit_price), 2) AS amount,
        p.sale_start_at, p.completed_at,
        TIMESTAMPDIFF(SECOND, p.sale_start_at, p.completed_at)            AS duration_secs,
        (SELECT COUNT(DISTINCT b.user_id) FROM deca_buy_record b
          WHERE b.product_code = p.product_code)                          AS buyers,
        -- 中卡人数：该团拆卡报告去重命中用户(hit_user_nickname)，中卡近似口径
        (SELECT COUNT(DISTINCT r.hit_user_nickname) FROM deca_report_record r
          WHERE r.product_code = p.product_code
            AND r.hit_user_nickname IS NOT NULL AND r.hit_user_nickname <> '') AS hit_users,
        -- 该商品最早一条进度快照时刻：用于判定里程碑是否「坍缩」(首张快照已越过阈值则该用时不可信)
        (SELECT MIN(pr.captured_at) FROM deca_onsale_product_progress_record pr
          WHERE pr.product_code = p.product_code)                          AS first_cap,
        (SELECT MIN(pr.captured_at) FROM deca_onsale_product_progress_record pr
          WHERE pr.product_code = p.product_code AND pr.progress_pct >= 25)  AS t25,
        (SELECT MIN(pr.captured_at) FROM deca_onsale_product_progress_record pr
          WHERE pr.product_code = p.product_code AND pr.progress_pct >= 50)  AS t50,
        (SELECT MIN(pr.captured_at) FROM deca_onsale_product_progress_record pr
          WHERE pr.product_code = p.product_code AND pr.progress_pct >= 75)  AS t75
    FROM deca_product_record p
    WHERE p.merchant_user_id = %s AND {WIN_P}
      AND p.unit_price IS NOT NULL AND p.sold_count IS NOT NULL
    ORDER BY amount DESC
    """
    rows = pool.select_all(sql, (mid,)) or []
    result = []
    for (title, series, spec, price, sold, card, amount, start, completed,
         duration_secs, buyers, hit_users, first_cap, t25, t50, t75) in rows:
        progress = round(sold / card * 100, 1) if card else None  # 售卖进度百分比
        result.append({
            "团名": title, "系列": series, "类型": spec, "单价": price,
            "总份数": card, "进度": progress, "总金额": amount,
            "参与人数": buyers,
            "中卡人数": hit_users,                       # 该团拆卡报告去重命中用户(中卡近似)
            "开售时间": start, "成交时间": completed,   # 开售=sale_start_at，成交=completed_at
            "售卖时长": _fmt_duration(duration_secs),   # 成交-开售，也即整团总时长
            # 到 X% 用时：仅当该商品在 tX 之前还有更早快照(未坍缩)时才输出真实穿越耗时，否则留空
            "到25%用时": _milestone_used(t25, first_cap, start),
            "到50%用时": _milestone_used(t50, first_cap, start),
            "到75%用时": _milestone_used(t75, first_cap, start),
        })
    return result


def fetch_moddu_user_ranking(pool, mid: str, top_n: int) -> tuple[list[dict], int]:
    """取「魔都」商家时间窗内的用户参与排行（按参与金额倒序，取前 top_n）。

    「一个拼团商品 = 一辆车（组队）」，以 deca_buy_record 购买记录按买家 user_id 聚合：
        - 参与车数 = COUNT(DISTINCT product_code)，该买家窗口内参与的不同团数。
        - 参与金额 = SUM(card_count × 该团 unit_price)。deca_buy_record 无金额字段，故用
          「购买份数 × 团单价」估算；固定价团精确，随机团(选队随机/剩余随机)每队价不同，
          此处按标称单价近似。同一买家在同一团的多条购买记录已由 SUM 累加。
        - 车均消费 = 参与金额 ÷ 参与车数。
    仅魔都(881226408)采了购买记录，故只有它能出此榜。

    Args:
        pool (MySQLConnectionPool): MySQL 连接池。
        mid (str): 商家 merchant_user_id。
        top_n (int): 榜单展示上限（取金额倒序前 N）。

    Returns:
        tuple[list[dict], int]: (rows, total_users)；rows 每项含 用户昵称/user_id/参与车数/
            参与金额/车均消费（已按金额倒序并截断到 top_n）；total_users 为窗口内全部参与
            买家数（供榜单标题展示、判断截断了多少）。
    """
    sql = f"""
    SELECT b.user_id,
           MAX(b.nickname)                            AS nickname,
           COUNT(DISTINCT b.product_code)             AS cars,
           ROUND(SUM(b.card_count * p.unit_price), 2) AS amount
    FROM deca_buy_record b
    JOIN deca_product_record p ON p.product_code = b.product_code
    WHERE p.merchant_user_id = %s AND {WIN_P}
      AND p.unit_price IS NOT NULL AND p.sold_count IS NOT NULL
    GROUP BY b.user_id
    ORDER BY amount DESC
    """
    rows = pool.select_all(sql, (mid,)) or []
    result = []
    for uid, nickname, cars, amount in rows[:top_n]:
        amt = float(amount) if amount is not None else 0.0
        cars = int(cars or 0)
        result.append({
            "用户昵称": nickname or "(匿名)",
            "user_id": uid,
            "参与车数": cars,
            "参与金额": round(amt, 2),
            "车均消费": round(amt / cars, 2) if cars else None,  # 车均=参与金额/参与车数
        })
    return result, len(rows)


def fetch_moddu_missing_teams(pool, mid: str) -> dict:
    """对比「魔都」成交明细与购买记录覆盖，算出漏采购买记录的 T(团)。

    魔都明细每条 = 一个成交拼团商品(T)，来自 deca_product_record；购买记录 deca_buy_record
    是另路采集的。个别团在采到购买记录前就满仓成交下架，会「漏采」——本函数以时间窗内成交
    团为基准，找出 deca_buy_record 里没有对应 product_code 的团，供魔都明细标注覆盖缺口。

    Args:
        pool (MySQLConnectionPool): MySQL 连接池。
        mid (str): 商家 merchant_user_id。

    Returns:
        dict: 含 成交团数/有记录团数/漏采团数/漏采团列表(list[(product_code, title)])。
    """
    total = pool.select_all(f"""
    SELECT COUNT(*) FROM deca_product_record p
    WHERE p.merchant_user_id = %s AND {WIN_P}
      AND p.unit_price IS NOT NULL AND p.sold_count IS NOT NULL
    """, (mid,))[0][0] or 0
    miss = pool.select_all(f"""
    SELECT p.product_code, p.title
    FROM deca_product_record p
    WHERE p.merchant_user_id = %s AND {WIN_P}
      AND p.unit_price IS NOT NULL AND p.sold_count IS NOT NULL
      AND NOT EXISTS(SELECT 1 FROM deca_buy_record b WHERE b.product_code = p.product_code)
    ORDER BY p.completed_at
    """, (mid,)) or []
    return {"成交团数": total, "有记录团数": total - len(miss),
            "漏采团数": len(miss), "漏采团列表": [(pc, t) for pc, t in miss]}


def _fmt_duration(secs) -> str | None:
    """把售卖时长(秒)格式化为可读中文字符串。

    差值口径 = 成交完成时间 completed_at − 开售时间 sale_start_at，反映该团多快卖完/售罄。

    Args:
        secs (int | None): 成交与开售之差(秒)；None 或负数(时间缺失/异常)视为无效。

    Returns:
        str | None: 形如 "2小时13分" / "45分07秒" / "38秒"；无效时返回 None(单元格留空)。
    """
    if secs is None or secs < 0:
        return None
    secs = int(secs)
    h, rem = divmod(secs, 3600)
    m, s = divmod(rem, 60)
    if h:
        return f"{h}小时{m}分"
    if m:
        return f"{m}分{s:02d}秒"
    return f"{s}秒"


def _diff_secs(later, earlier) -> int | None:
    """算两个时间点相隔的秒数，兼容 varchar 与 datetime 两种输入。

    Args:
        later (str | datetime | None): 较晚的时间点(如到达某进度的时刻)。
        earlier (str | datetime | None): 较早的时间点(如开售时间)。

    Returns:
        int | None: 相隔秒数；任一为空或解析失败时返回 None。
    """
    if not later or not earlier:
        return None
    try:
        if isinstance(earlier, str):
            earlier = datetime.strptime(earlier, "%Y-%m-%d %H:%M:%S")
        if isinstance(later, str):
            later = datetime.strptime(later, "%Y-%m-%d %H:%M:%S")
        return int((later - earlier).total_seconds())
    except Exception:
        return None


def _milestone_used(tx, first_cap, start) -> str | None:
    """算「到某进度用时」，首张快照已越过该阈值(坍缩)时诚实留空。

    到 X% 用时 = 首次 progress_pct ≥ X 的快照时刻 tx − 开售时间 start。但只有当该商品在
    tx 之前还存在更早的快照(即 tx 晚于该商品最早快照 first_cap)时，才证明我们是从低于 X%
    的进度观测到它穿越 X% 的，tx 才是真实穿越点；若首张快照 pct 就已 ≥ X(tx == first_cap)，
    tx 只是「首次被监控的时刻」，据此算出的是假耗时，一律留空，绝不输出坍缩值。

    Args:
        tx (datetime | str | None): 首次 progress_pct ≥ 阈值的快照时刻；None 表示从未达到该进度。
        first_cap (datetime | str | None): 该商品最早一条快照的时刻。
        start (datetime | str | None): 开售时间 sale_start_at。

    Returns:
        str | None: 可读用时字符串；未达到/坍缩/时间缺失时返回 None(单元格留空)。
    """
    if tx is None or first_cap is None:
        return None
    # tx 与首张快照同刻 → 第一次看到它就已越过该阈值，无法确定真实穿越时刻，坍缩留空
    if not _diff_secs(tx, first_cap):        # None(解析失败) 或 0(同刻) 都视为不可信
        return None
    return _fmt_duration(_diff_secs(tx, start))


def _style_row(ws, row_idx: int, ncol: int, font=None, fill=None, border=True):
    """给某一行的前 ncol 个单元格套字体/填充/边框。

    Args:
        ws: openpyxl worksheet。
        row_idx (int): 行号(1-based)。
        ncol (int): 要处理的列数(从第 1 列起)。
        font (Font, optional): 字体。Defaults to None。
        fill (PatternFill, optional): 填充。Defaults to None。
        border (bool, optional): 是否加边框。Defaults to True。
    """
    for c in range(1, ncol + 1):
        cell = ws.cell(row=row_idx, column=c)
        if font:
            cell.font = font
        if fill:
            cell.fill = fill
        if border:
            cell.border = BORDER


def _write_summary_block(ws, r: int, rows: list, data: dict, start_col: int = 1) -> int:
    """在第 r 行起竖排写汇总(每指标一行：标签列 + 数值列)，返回下一空闲行。

    竖排避免与明细表共用列宽导致的错位与金额列过窄(######)。显示标签与 data 取值键解耦，
    以便同一指标在不同口径下用不同标签(如「参与人数(真实买家)」vs「中卡用户数(近似)」)。
    start_col 可让整块右移(明细 sheet 里落到宽的 B/C 列，避开 A=序号 的窄列)。

    Args:
        ws: openpyxl worksheet。
        r (int): 起始行号。
        rows (list[tuple[str, str]]): 每项 (显示标签, data 取值键)。
        data (dict): 汇总数据 dict。
        start_col (int, optional): 标签列的列号(1-based)，数值列为其右一列。Defaults to 1。

    Returns:
        int: 下一个空闲行号。
    """
    for label, key in rows:
        lc = ws.cell(row=r, column=start_col, value=label)     # 标签(加粗深蓝、浅蓝底、左对齐)
        lc.font = FONT_HEADER
        lc.fill = FILL_HEADER
        lc.alignment = ALIGN_LEFT
        lc.border = BORDER
        value = data.get(key)
        val = ws.cell(row=r, column=start_col + 1, value=value)  # 数值(浅橙底、右对齐)
        if key in ("销售额", "均拼单价", "人均消费"):    # 金额格式按稳定的取值键判定
            val.number_format = MONEY_FMT
        elif isinstance(value, int) and not isinstance(value, bool):
            val.number_format = INT_FMT                 # 计数千分位(成团数/参与人数/商家数)
        val.font = FONT_CELL
        val.fill = FILL_SUBTOTAL
        val.alignment = _align_by_value(value)          # 数字右对齐、文字左对齐
        val.border = BORDER
        r += 1
    return r


def _write_section_title(ws, r: int, text: str, span: int = SECTION_SPAN) -> int:
    """在第 r 行写一行分区标题条(蓝底白字)，蓝条只覆盖 span 列(贴合本表宽度、右侧不留空)。

    Args:
        ws: openpyxl worksheet。
        r (int): 起始行号。
        text (str): 标题文字。
        span (int, optional): 蓝条覆盖列数。Defaults to SECTION_SPAN。

    Returns:
        int: 下一个空闲行号。
    """
    ws.cell(row=r, column=1, value=text)
    _style_row(ws, r, span, font=FONT_SECTION, fill=FILL_SECTION, border=False)
    return r + 1


def _write_details(ws, r: int, details: list, cols: list) -> int:
    """在第 r 行起写一个商家的明细表(表头 + 每条组队明细)。

    Args:
        ws: openpyxl worksheet。
        r (int): 起始行号。
        details (list[dict]): 明细数据，每元素为一条组队明细 dict(fetch_merchant_details
            或 fetch_moddu_details 的返回)。
        cols (list[tuple]): 列规格列表，每项 (表头, 取值键, 是否金额格式)；取值键为「序号」
            时按行号自增填充(DETAIL_COLS 或 MODDU_DETAIL_COLS)。

    Returns:
        int: 下一个空闲行号。
    """
    for i, (header, _key, _money) in enumerate(cols):
        hc = ws.cell(row=r, column=1 + i, value=header)
        hc.alignment = ALIGN_CENTER                    # 表头居中换行
    _style_row(ws, r, len(cols), font=FONT_HEADER, fill=FILL_HEADER)
    r += 1

    if not details:
        ws.cell(row=r, column=1, value="(该时段无成交)").font = FONT_CELL
        return r + 1

    for idx, d in enumerate(details, 1):
        zebra = FILL_ZEBRA if idx % 2 == 0 else None   # 偶数行淡蓝灰，便于横向读长行
        for i, (_header, key, is_money) in enumerate(cols):
            value = idx if key == "序号" else d.get(key)   # 「序号」列用行号，其余按键取值
            cell = ws.cell(row=r, column=1 + i, value=value)
            if is_money:
                cell.number_format = MONEY_FMT
            elif isinstance(value, int) and not isinstance(value, bool):
                cell.number_format = INT_FMT           # 计数千分位(序号/份数/总份数/参与人数)
            cell.font = FONT_CELL
            cell.alignment = _align_by_value(value)    # 数字右、文字左(时长/日期/团名等)
            cell.border = BORDER
            if zebra:
                cell.fill = zebra
        r += 1
    return r


def _window_metrics(pool, win: str) -> dict:
    """统计某成交时间窗内的组齐核心指标（GMV / 成团数 / 活跃商家数 / T均单价）。

    Args:
        pool (MySQLConnectionPool): MySQL 连接池。
        win (str): 完整的 completed_at 时间窗 WHERE 片段（如 WIN_P / WIN_P_YDAY）。

    Returns:
        dict: 含 组齐GMV/成团数/活跃商家数/T均单价（成团数为 0 时 T均单价为 None）。
    """
    sql = f"""
    SELECT ROUND(SUM(COALESCE(p.team_total_amount, p.sold_count * p.unit_price)), 2) AS gmv,
           COUNT(*)                           AS grp,
           COUNT(DISTINCT p.merchant_user_id) AS merchants
    FROM deca_product_record p
    WHERE {win} AND p.unit_price IS NOT NULL AND p.sold_count IS NOT NULL
    """
    gmv, groups, merchants = pool.select_all(sql)[0]
    gmv = float(gmv) if gmv is not None else 0.0
    groups = groups or 0
    return {"组齐GMV": round(gmv, 2), "成团数": groups, "活跃商家数": merchants or 0,
            "T均单价": round(gmv / groups, 2) if groups else None}  # T均单价=组齐GMV/成团数


def fetch_groupbuy_compare(pool) -> dict:
    """取当日窗口与昨日同窗口的组齐指标，供「当日组齐环比」区展示。

    Args:
        pool (MySQLConnectionPool): MySQL 连接池。

    Returns:
        dict: {"today": {...}, "yday": {...}}，各含 组齐GMV/成团数/活跃商家数/T均单价。
    """
    return {"today": _window_metrics(pool, WIN_P), "yday": _window_metrics(pool, WIN_P_YDAY)}


def fetch_series_ranking(pool, top_n: int) -> tuple[list, float]:
    """取当日窗口内各产品系列的销售榜（按 GMV 倒序）及全窗口总 GMV（算占比用）。

    Args:
        pool (MySQLConnectionPool): MySQL 连接池。
        top_n (int): 取前 N 个系列。

    Returns:
        tuple[list, float]: (rows, total_gmv)；rows 每项 (系列名, 成团数, GMV)。
    """
    sql = f"""
    SELECT COALESCE(NULLIF(p.series_name, ''), '(未标系列)')        AS series,
           COUNT(*)                                                 AS grp,
           ROUND(SUM(COALESCE(p.team_total_amount, p.sold_count * p.unit_price)), 2) AS gmv
    FROM deca_product_record p
    WHERE {WIN_P} AND p.unit_price IS NOT NULL AND p.sold_count IS NOT NULL
    GROUP BY series
    ORDER BY gmv DESC
    """
    rows = pool.select_all(sql) or []
    total = sum(float(r[2]) for r in rows if r[2] is not None)   # 全部系列合计(算占比分母)
    return rows[:top_n], total


def fetch_merchant_gmv_ranking(pool) -> tuple[list, float, dict]:
    """取当日窗口内全部商家的组齐 GMV 榜（倒序）及 Top1/3/5/10 集中度占比。

    集中度 = 前 K 名商家 GMV 之和 ÷ 平台组齐总 GMV，反映头部商家对大盘的贡献集中程度。

    Args:
        pool (MySQLConnectionPool): MySQL 连接池。

    Returns:
        tuple[list, float, dict]: (rows, total_gmv, concentration)；rows 每项
            (商家名, 成团数, GMV)（全量、倒序）；concentration 形如 {1: 0.32, 3: 0.55, ...}。
    """
    sql = f"""
    SELECT MAX(p.merchant_name)              AS mname,
           COUNT(*)                           AS grp,
           ROUND(SUM(COALESCE(p.team_total_amount, p.sold_count * p.unit_price)), 2) AS gmv
    FROM deca_product_record p
    WHERE {WIN_P} AND p.unit_price IS NOT NULL AND p.sold_count IS NOT NULL
    GROUP BY p.merchant_user_id
    ORDER BY gmv DESC
    """
    rows = pool.select_all(sql) or []
    gmvs = [float(r[2]) if r[2] is not None else 0.0 for r in rows]
    total = sum(gmvs)
    conc = {k: (sum(gmvs[:k]) / total if total else None) for k in CONC_TOPS}
    return rows, total, conc


def fetch_focus_ops_snapshot(pool, mid: str) -> dict:
    """取重点商家「当日新开团 / 已组齐 / 规格分布」运营快照（口径：publish_at 落在报告时间窗 [昨17:00, 今06:00]）。

    「已组齐」= 当日新开团里 product_code 已出现在成交表(deca_product_record 有 completed_at)的数量；
    「规格分布」按在售表 spec_name 聚合，形如 原箱×2 · 单盒×20。对应监测清单「魔都、卡皇当天销售细节」。

    Args:
        pool (MySQLConnectionPool): MySQL 连接池。
        mid (str): 商家 merchant_user_id。

    Returns:
        dict: 含 商家名/新开团数/已组齐数/规格分布(list[(规格, 数量)])。
    """
    # 口径对齐报告成交时间窗(2026/08/17)：新开团/已组齐/规格均按 publish_at 落在[昨17:00,今06:00]统计。
    # 原为自然日 DATE(publish_at)=CURDATE()，会漏掉前一晚上新的团(如卡皇昨晚 22:42 上新→自然日今天为 0)。
    base = ("FROM deca_onsale_product_record o "
            "WHERE o.merchant_user_id = %s "
            "AND o.publish_at >= (CURDATE() - INTERVAL 1 DAY) + INTERVAL 17 HOUR "
            "AND o.publish_at <= CURDATE() + INTERVAL 6 HOUR")
    # 商家名单独取(不加「当日」条件)：某商家当天没上新团时仍能显示名称，避免回退成纯 ID（2026/08/17）
    mname_row = pool.select_one(
        "SELECT MAX(merchant_name) FROM deca_onsale_product_record WHERE merchant_user_id = %s", (mid,))
    mname = mname_row[0] if mname_row else None
    total_new = pool.select_one(f"SELECT COUNT(*) {base}", (mid,))[0] or 0
    done = pool.select_one(
        f"SELECT COUNT(*) {base} "
        f"AND EXISTS(SELECT 1 FROM deca_product_record c "
        f"WHERE c.product_code = o.product_code AND c.completed_at IS NOT NULL)", (mid,))[0] or 0
    spec_rows = pool.select_all(
        f"SELECT COALESCE(NULLIF(o.spec_name, ''), '其它') sp, COUNT(*) c {base} "
        f"GROUP BY sp ORDER BY c DESC", (mid,)) or []
    return {"商家名": mname or mid, "新开团数": total_new, "已组齐数": done,
            "规格分布": [(r[0], int(r[1])) for r in spec_rows]}


def fetch_completion_hour_dist(pool, days: int) -> list[int]:
    """统计近 N 天全部成交(组齐)按小时的 24 桶分布，反映平台 24h 组齐节奏。

    Args:
        pool (MySQLConnectionPool): MySQL 连接池。
        days (int): 回看天数（含今天）。

    Returns:
        list[int]: 长度 24 的列表，索引 = 小时(0~23)，值 = 该小时成交团数。
    """
    sql = """
    SELECT HOUR(p.completed_at) AS h, COUNT(*) AS c
    FROM deca_product_record p
    WHERE p.completed_at >= CURDATE() - INTERVAL %s DAY AND p.completed_at IS NOT NULL
    GROUP BY h
    """
    dist = [0] * 24
    for h, c in pool.select_all(sql, (days,)) or []:
        if h is not None and 0 <= int(h) < 24:
            dist[int(h)] = int(c)
    return dist


def _bar(value: int, max_value: int, width: int = 20) -> str:
    """把数值渲染成等宽条形字符串（Excel 内迷你直方图）。

    Args:
        value (int): 当前值。
        max_value (int): 该组最大值（归一化条长用）。
        width (int, optional): 满值时的条长（字符数）。Defaults to 20。

    Returns:
        str: 由 █ 组成的条；max_value<=0 或 value<=0 时返回空串。
    """
    if max_value <= 0 or value <= 0:
        return ""
    return "█" * max(1, round(value / max_value * width))


def _fmt_pct_change(cur, prev) -> str:
    """把本期/上期数值算成带符号的环比字符串。

    Args:
        cur (float | int | None): 本期值。
        prev (float | int | None): 上期值。

    Returns:
        str: 形如 "+12.3%" / "-8.0%"；上期为 0 时本期>0 返回 "新增"、否则 "—"。
    """
    c = float(cur) if cur is not None else 0.0
    p = float(prev) if prev is not None else 0.0
    if p == 0:
        return "新增" if c > 0 else "—"
    return f"{(c - p) / p * 100:+.1f}%"


def _write_hgrid(ws, r: int, headers: list, rows: list, money_cols=(), pct_cols=(),
                 start_col: int = 1, empty_hint: str = "(无数据)") -> int:
    """在第 r 行起写一张普通横向表（表头 + 数据行），套统一样式，返回下一空闲行。

    可用 start_col 让表整体右移，把长文本列落在预设的宽列上（本 sheet B 列宽 50）。

    Args:
        ws: openpyxl worksheet。
        r (int): 起始行号。
        headers (list[str]): 列表头。
        rows (list[list]): 数据行，每行长度应 ≤ len(headers)。
        money_cols (tuple[int], optional): 需金额格式的列下标(0-based)。Defaults to ()。
        pct_cols (tuple[int], optional): 需百分比格式的列下标(0-based，存小数如 0.32)。Defaults to ()。
        start_col (int, optional): 起始列号(1-based)。Defaults to 1。
        empty_hint (str, optional): rows 为空时的占位文字。Defaults to "(无数据)"。

    Returns:
        int: 下一个空闲行号。
    """
    for i, h in enumerate(headers):
        cell = ws.cell(row=r, column=start_col + i, value=h)
        cell.font = FONT_HEADER
        cell.fill = FILL_HEADER
        cell.alignment = ALIGN_CENTER                   # 表头居中
        cell.border = BORDER
    r += 1
    if not rows:
        c = ws.cell(row=r, column=start_col, value=empty_hint)
        c.font = FONT_CELL
        c.border = BORDER
        return r + 1
    n = len(headers)
    for ri, row in enumerate(rows):
        zebra = FILL_ZEBRA if ri % 2 == 1 else None     # 隔行淡蓝灰
        for i in range(n):
            v = row[i] if i < len(row) else None
            cell = ws.cell(row=r, column=start_col + i, value=v)
            if i in money_cols:
                cell.number_format = MONEY_FMT
            elif i in pct_cols:
                cell.number_format = "0.0%"
            elif isinstance(v, int) and not isinstance(v, bool):
                cell.number_format = INT_FMT            # 计数千分位(成团数等)
            cell.font = FONT_CELL
            # 金额/占比/数字右对齐；条形图列(█ 文本)与名称等文字左对齐
            cell.alignment = ALIGN_RIGHT if (i in money_cols or i in pct_cols) else _align_by_value(v)
            cell.border = BORDER
            if zebra:
                cell.fill = zebra
        r += 1
    return r


def _set_widths(ws, widths: list):
    """按列表顺序设置 A、B、C… 列宽（仅前 26 列，够覆盖本报告各 sheet）。

    Args:
        ws: openpyxl worksheet。
        widths (list[float]): 各列宽度，索引 0→A、1→B…。
    """
    for i, w in enumerate(widths):
        ws.column_dimensions[chr(65 + i)].width = w


def _build_overview_sheet(ws, win: tuple, platform: dict, compare: dict,
                          mrank_conc: dict, notes: list):
    """构建「平台总览」sheet：平台汇总 + 当日组齐环比 + 商家 GMV 集中度 + 口径脚注。

    Args:
        ws: openpyxl worksheet。
        win (tuple[str, str]): (成交窗起, 成交窗止)。
        platform (dict): 平台汇总数据。
        compare (dict): 组齐环比数据(today/yday)。
        mrank_conc (dict): Top1/3/5/10 集中度占比。
        notes (list[str]): 口径脚注文本。
    """
    _set_widths(ws, [32, 18, 18, 12])
    r = 1
    ws.cell(row=r, column=1, value="得卡 DECA · 已售每日统计报告")
    _style_row(ws, r, 4, font=FONT_TITLE, border=False)
    r += 1
    ws.cell(row=r, column=1, value=f"成交时间窗 {win[0]} ~ {win[1]}")
    _style_row(ws, r, 4, font=FONT_CELL, border=False)
    r += 2

    # 平台汇总（竖排 标签/数值）
    r = _write_section_title(ws, r, "平台汇总", span=4)
    r = _write_summary_block(ws, r, PLATFORM_ROWS, platform, start_col=1)
    r += 1

    # 当日组齐环比（金额行今日/昨日单独套金额格式；环比列涨绿跌红）
    r = _write_section_title(ws, r, "当日组齐环比（vs 昨日同窗口）", span=4)
    for i, h in enumerate(["指标", "今日", "昨日", "环比"]):
        c = ws.cell(row=r, column=1 + i, value=h)
        c.font = FONT_HEADER; c.fill = FILL_HEADER
        c.alignment = ALIGN_CENTER; c.border = BORDER
    r += 1
    t_now, t_yday = compare["today"], compare["yday"]
    for ri, (key, is_money) in enumerate((("组齐GMV", True), ("成团数", False),
                                          ("活跃商家数", False), ("T均单价", True))):
        zebra = FILL_ZEBRA if ri % 2 == 1 else None
        lc = ws.cell(row=r, column=1, value=key)                      # 指标名：左对齐
        lc.font = FONT_CELL; lc.alignment = ALIGN_LEFT
        cur = ws.cell(row=r, column=2, value=t_now.get(key))
        prev = ws.cell(row=r, column=3, value=t_yday.get(key))
        for cell in (cur, prev):                                      # 今日/昨日：金额或计数、右对齐
            cell.number_format = MONEY_FMT if is_money else INT_FMT
            cell.font = FONT_CELL; cell.alignment = ALIGN_RIGHT
        chg = _fmt_pct_change(t_now.get(key), t_yday.get(key))
        cc = ws.cell(row=r, column=4, value=chg)                      # 环比：右对齐 + 涨绿跌红
        cc.font = _trend_font(chg); cc.alignment = ALIGN_RIGHT
        for col in range(1, 5):
            cell = ws.cell(row=r, column=col)
            cell.border = BORDER
            if zebra:
                cell.fill = zebra
        r += 1
    r += 1

    # 商家 GMV 集中度（详细榜单在「商家GMV榜」sheet）
    r = _write_section_title(ws, r, "商家 GMV 集中度（TopN 占平台组齐总 GMV）", span=4)
    conc_rows = [[f"Top{k} 集中度", mrank_conc.get(k)] for k in CONC_TOPS]
    r = _write_hgrid(ws, r, ["集中度档位", "占平台GMV"], conc_rows, pct_cols=(1,), start_col=1)
    r += 2

    # 口径脚注（长文本，向右溢出到空列显示）
    for note in notes:
        ws.cell(row=r, column=1, value=note).font = FONT_CELL
        r += 1


def _build_series_sheet(ws, series_rows: list, series_total: float):
    """构建「产品系列榜」sheet：当日各系列 GMV 榜 + 占比。

    Args:
        ws: openpyxl worksheet。
        series_rows (list): [(系列名, 成团数, GMV)]。
        series_total (float): 全窗口总 GMV(算占比分母)。
    """
    _set_widths(ws, [36, 10, 16, 10])
    r = _write_section_title(ws, 1, f"产品系列销售榜（当日 Top{TOP_SERIES}，按 GMV）", span=4)
    srows = [[name, int(g), float(gmv) if gmv is not None else 0,
              (float(gmv) / series_total if (series_total and gmv is not None) else None)]
             for name, g, gmv in series_rows]
    _write_hgrid(ws, r, ["系列", "成团数", "GMV", "占比"], srows,
                 money_cols=(2,), pct_cols=(3,), start_col=1)
    ws.freeze_panes = "A3"   # 冻结标题条 + 表头


def _build_mrank_sheet(ws, mrank_rows: list, mrank_total: float):
    """构建「商家GMV榜」sheet：当日组齐 GMV 前 N 商家 + 占比。

    Args:
        ws: openpyxl worksheet。
        mrank_rows (list): [(商家名, 成团数, GMV)]（全量倒序）。
        mrank_total (float): 平台组齐总 GMV(算占比分母)。
    """
    _set_widths(ws, [24, 10, 16, 10])
    r = _write_section_title(ws, 1, f"商家 GMV 榜（当日组齐口径，前 {TOP_MERCHANT}）", span=4)
    mrows = [[name, int(g), float(gmv) if gmv is not None else 0,
              (float(gmv) / mrank_total if (mrank_total and gmv is not None) else None)]
             for name, g, gmv in mrank_rows[:TOP_MERCHANT]]
    _write_hgrid(ws, r, ["商家", "成团数", "GMV", "占比"], mrows,
                 money_cols=(2,), pct_cols=(3,), start_col=1)
    ws.freeze_panes = "A3"


def _build_ops_sheet(ws, ops: list, hour_dist: list):
    """构建「运营节奏」sheet：重点商家当日运营快照 + 平台组齐时段分布。

    Args:
        ws: openpyxl worksheet。
        ops (list[dict]): 各重点商家运营快照。
        hour_dist (list[int]): 24 桶组齐时段分布。
    """
    _set_widths(ws, [16, 12, 22, 44])   # C 兼容 已组齐(数字) 与 时段分布条；D 放规格分布长文本
    r = _write_section_title(ws, 1, "重点商家当日运营快照（新开团 / 已组齐 / 规格）", span=4)
    ops_rows = [[o["商家名"], o["新开团数"], o["已组齐数"],
                 " · ".join(f"{sp}×{c}" for sp, c in o["规格分布"]) or "—"] for o in ops]
    r = _write_hgrid(ws, r, ["商家", "今日新开团", "已组齐", "规格分布"], ops_rows, start_col=1)
    r += 1
    r = _write_section_title(ws, r, f"平台组齐时段分布（近 {HOUR_DIST_DAYS} 日 24h 累计）", span=4)
    hmax = max(hour_dist) if hour_dist else 0
    hrows = [[f"{h:02d}时", hour_dist[h], _bar(hour_dist[h], hmax)] for h in range(24)]
    _write_hgrid(ws, r, ["时段", "成团数", "分布"], hrows, start_col=1)


def _build_detail_sheet(ws, title: str, summ: dict, details: list, cols: list,
                        is_real: bool, span: int, widths: list, miss_info: dict = None):
    """构建单个重点商家的明细 sheet：汇总(缩到 B/C 列) + 每条组队明细(从 A 列起)。

    Args:
        ws: openpyxl worksheet。
        title (str): sheet 顶部分区标题(含商家名与时间窗)。
        summ (dict): 该商家汇总数据。
        details (list[dict]): 每条组队明细。
        cols (list[tuple]): 明细列规格(DETAIL_COLS / MODDU_DETAIL_COLS)。
        is_real (bool): 参与人数是否真实买家口径(魔都 True，其余 False)。
        span (int): 标题条覆盖列数(= 明细列数)。
        widths (list[float]): 各列宽度。
        miss_info (dict, optional): 购买记录覆盖检测(fetch_moddu_missing_teams 返回)；非 None
            时在汇总块下方加「购买记录覆盖检测」小节，标注漏采多少 T 并列出漏团。仅魔都传入。
            Defaults to None。
    """
    _set_widths(ws, widths)
    r = _write_section_title(ws, 1, title, span=span)
    # 汇总缩到 B/C 列：标签落宽的 B(团名列)、数值落 C，避开 A=序号 的窄列
    r = _write_summary_block(ws, r, _summary_rows(is_real=is_real), summ, start_col=2)
    r += 1
    # 购买记录覆盖检测(仅魔都传入)：成交团 vs 已采购买记录，漏采的 T 逐个列出(文本向右溢出显示)
    if miss_info is not None:
        r = _write_section_title(ws, r, "购买记录覆盖检测（成交团 vs 已采购买记录）", span=span)
        cov = (f"成交 {miss_info['成交团数']} 团 · 采到购买记录 {miss_info['有记录团数']} 团 · "
               f"漏采 {miss_info['漏采团数']} 团（用户排行见「用户排行榜(魔都)」sheet）")
        # 漏采 >0 时用深蓝加粗字提醒；0 时常规字
        ws.cell(row=r, column=1, value=cov).font = FONT_HEADER if miss_info["漏采团数"] else FONT_CELL
        r += 1
        if miss_info["漏采团数"]:
            ws.cell(row=r, column=1,
                    value="漏采明细（下列团未采到购买记录，未计入用户排行）：").font = FONT_CELL
            r += 1
            for pc, t in miss_info["漏采团列表"]:
                ws.cell(row=r, column=1, value=f"　- {pc}　{t or ''}").font = FONT_CELL
                r += 1
        r += 1
    r = _write_section_title(ws, r, f"每条组队明细（共 {len(details)} 条，按总金额倒序）", span=span)
    hdr_row = r                                  # 明细表头行，用于冻结
    _write_details(ws, r, details, cols)
    ws.freeze_panes = f"A{hdr_row + 1}"          # 冻结到明细表头，滚动时表头常驻


def _build_user_ranking_sheet(ws, rows: list, total_users: int, top_n: int):
    """构建「用户排行榜(魔都)」sheet：按参与金额倒序的买家榜(参与车数/参与金额/车均消费)。

    Args:
        ws: openpyxl worksheet。
        rows (list[dict]): 用户排行数据(fetch_moddu_user_ranking 返回，已倒序截断)。
        total_users (int): 窗口内魔都全部参与买家数(用于标题展示)。
        top_n (int): 榜单展示上限(用于标题展示)。
    """
    _set_widths(ws, [8, 22, 16, 12, 16, 14])
    cap = "全部展示" if top_n is None else f"取前 {min(len(rows), top_n)}"
    r = _write_section_title(
        ws, 1, f"用户排行榜 · 魔都（共 {total_users} 人参与，{cap}，按参与金额倒序）", span=6)
    grid = [[i + 1, d["用户昵称"], d["user_id"], d["参与车数"], d["参与金额"], d["车均消费"]]
            for i, d in enumerate(rows)]
    r = _write_hgrid(ws, r, ["排名", "用户昵称", "user_id", "参与车数", "参与金额", "车均消费"],
                     grid, money_cols=(4, 5), start_col=1, empty_hint="(该时段无购买记录)")
    r += 1
    ws.cell(row=r, column=1,
            value="注：参与金额 = Σ(购买份数 × 团单价)（deca_buy_record 无金额字段，按份数×单价估算；"
                  "固定价团精确、随机团按标称单价近似）；参与车数=参与的不同团数；"
                  "车均消费 = 参与金额 ÷ 参与车数。").font = FONT_CELL
    ws.freeze_panes = "A3"   # 冻结标题条 + 表头


def _build_others_sheet(ws, others: list):
    """构建「其他商家」sheet：除重点商家外各商家一行汇总(中卡近似口径)。

    Args:
        ws: openpyxl worksheet。
        others (list[dict]): 每商家一行的汇总。
    """
    _set_widths(ws, [22, 14, 9, 15, 13, 16])
    r = _write_section_title(ws, 1, f"其他商家汇总（共 {len(others)} 家，按销售额倒序）", span=6)
    for i, (label, _key, _money) in enumerate(OTHER_COLS):
        hc = ws.cell(row=r, column=1 + i, value=label)
        hc.font = FONT_HEADER; hc.fill = FILL_HEADER
        hc.alignment = ALIGN_CENTER; hc.border = BORDER
    r += 1
    for oi, o in enumerate(others):
        zebra = FILL_ZEBRA if oi % 2 == 1 else None    # 隔行淡蓝灰
        for i, (_label, key, is_money) in enumerate(OTHER_COLS):
            value = o.get(key)
            cell = ws.cell(row=r, column=1 + i, value=value)
            if is_money:
                cell.number_format = MONEY_FMT
            elif isinstance(value, int) and not isinstance(value, bool):
                cell.number_format = INT_FMT           # 成团数/中卡用户数千分位
            cell.font = FONT_CELL
            cell.alignment = _align_by_value(value)    # 商家名左对齐、数字右对齐
            cell.border = BORDER
            if zebra:
                cell.fill = zebra
        r += 1
    ws.freeze_panes = "A2"


# 明细 sheet 列宽（各自独立设，不再与其它表共享）：
DETAIL_WIDTHS_MODDU = [8, 48, 16, 11, 13, 9, 8, 14, 14, 19, 19, 12, 11, 11, 11]  # 15 列(含里程碑)
DETAIL_WIDTHS_STD = [8, 48, 16, 11, 13, 9, 8, 14, 19, 19, 12]                    # 11 列(标准)
# 重点商家 ID → 明细 sheet 名（其余走商家名兜底）
DETAIL_SHEET_NAMES = {"881226408": "魔都明细", "274584650": "卡皇明细"}


def build_report(pool, out: str):
    """汇总各段数据并生成多 Sheet Excel 报告（每一大项一个 sheet，各自独立列宽）。

    Sheet 顺序：平台总览 / 产品系列榜 / 商家GMV榜 / 运营节奏 / 魔都明细 / 用户排行榜(魔都) /
        卡皇明细 / 其他商家。

    Args:
        pool (MySQLConnectionPool): MySQL 连接池。
        out (str): 导出的 xlsx 路径。
    """
    win = get_window(pool)
    platform = fetch_platform_summary(pool)
    compare = fetch_groupbuy_compare(pool)                       # 当日 vs 昨日组齐环比
    series_rows, series_total = fetch_series_ranking(pool, TOP_SERIES)   # 产品系列销售榜
    mrank_rows, mrank_total, mrank_conc = fetch_merchant_gmv_ranking(pool)  # 商家 GMV 榜 + 集中度
    ops = [fetch_focus_ops_snapshot(pool, mid) for mid in FOCUS_MERCHANTS]  # 重点商家运营快照
    hour_dist = fetch_completion_hour_dist(pool, HOUR_DIST_DAYS)  # 组齐时段 24h 分布
    # 魔都(MODDU_MID)明细走扩展版(带参与人数 + 进度里程碑)，其余重点商家走标准明细
    focus = []
    for mid in FOCUS_MERCHANTS:
        summ = fetch_merchant_summary(pool, mid)
        if mid == MODDU_MID:
            focus.append((mid, summ, fetch_moddu_details(pool, mid), MODDU_DETAIL_COLS))
        else:
            focus.append((mid, summ, fetch_merchant_details(pool, mid), DETAIL_COLS))
    others = fetch_other_merchants(pool, FOCUS_MERCHANTS)
    # 魔都用户排行(仅魔都采了购买记录) + 购买记录覆盖检测(成交团 vs 已采购买记录，看漏几个 T)
    moddu_user_rank, moddu_user_total = fetch_moddu_user_ranking(pool, MODDU_MID, TOP_USERS)
    moddu_miss = fetch_moddu_missing_teams(pool, MODDU_MID)

    # 口径脚注（放平台总览底部；解释两种「参与人数」口径的差别）
    notes = [
        "注：① 魔都兄弟球星卡「参与人数(真实买家)」= deca_buy_record 真实购买记录去重买家(跨其全部成交团)；"
        "各商家明细「参与人数(本团)」为各团单独去重买家，故明细逐团相加(人次) ≥ 汇总(跨团去重人头)。",
        "　　② 平台大盘参与人数 = 魔都真实买家 + 其他商家中卡去重(两口径人群标识不同、无法跨口径去重，直接相加，属近似上界)。",
        "　　③ 其他商家未采购买记录，「中卡用户数(近似)」= 拆卡报告 hit_user_nickname 去重(仅报告命中/中卡用户，"
        "非真实参团人头，偏低)，其「人均消费(按中卡近似)」据此计算、偏高，仅供参考。",
    ]

    wb = Workbook()
    ws = wb.active
    ws.title = "平台总览"
    _build_overview_sheet(ws, win, platform, compare, mrank_conc, notes)
    _build_series_sheet(wb.create_sheet("产品系列榜"), series_rows, series_total)
    _build_mrank_sheet(wb.create_sheet("商家GMV榜"), mrank_rows, mrank_total)
    _build_ops_sheet(wb.create_sheet("运营节奏"), ops, hour_dist)
    # 每个重点商家单独一个明细 sheet（各自独立列宽，互不迁就）
    for (mid, summ, details, cols) in focus:
        sheet_name = DETAIL_SHEET_NAMES.get(mid, f"{summ['商家名'][:8]}明细")
        ws_d = wb.create_sheet(sheet_name)
        is_real = (mid == MODDU_MID)
        widths = DETAIL_WIDTHS_MODDU if is_real else DETAIL_WIDTHS_STD
        # 魔都明细尾部附「购买记录覆盖检测」(漏采团数)；其余商家无购买记录、不检测
        miss = moddu_miss if mid == MODDU_MID else None
        _build_detail_sheet(ws_d, f"{summ['商家名']} · 汇总（成交时间窗 {win[0]} ~ {win[1]}）",
                            summ, details, cols, is_real, len(cols), widths, miss_info=miss)
        # 魔都明细后紧跟「用户排行榜(魔都)」sheet，让魔都相关表相邻
        if mid == MODDU_MID:
            _build_user_ranking_sheet(wb.create_sheet("用户排行榜(魔都)"),
                                      moddu_user_rank, moddu_user_total, TOP_USERS)
    _build_others_sheet(wb.create_sheet("其他商家"), others)

    wb.save(out)


def run_once(log) -> str:
    """连库生成已售每日报告并发送到微信群（只发 Excel）。

    Args:
        log: 日志对象。

    Returns:
        str: 生成的 xlsx 绝对路径；数据库连接池异常时返回空串。
    """
    pool = MySQLConnectionPool(log=log)
    if not pool.check_pool_health():
        log.error("数据库连接池异常")
        return ""
    # 绝对路径：不受运行目录影响、便于企微上传定位；文件名带当天日期避免覆盖
    out_file = os.path.abspath(f"{OUT_PREFIX}_{date.today():%Y%m%d}.xlsx")
    build_report(pool, out_file)
    log.info(f"报告已生成 -> {out_file}")

    # 发企微群（只发 Excel；失败仅告警，不影响报告产出）
    if SEND_WECHAT:
        try:
            from auto_send_wx_msg import send_wechat_group_file
            send_wechat_group_file(log=log, file_path=out_file)   # 只发 Excel，不发图
        except Exception as e:
            log.warning(f"企微发送跳过: {e}")
    return out_file


def main():
    """命令行一次性生成 + 发送（手动/调试用）。"""
    run_once(logger)


def schedule_task():
    """定时入口：每天 09:10 生成已售报告并发送到微信群。

    错开在售报告(on_sale 09:00 发)10 分钟发送，两份报告到群时间分开、便于查阅（企微为 webhook 直发，无 UI 冲突）。
    """
    # run_once(logger)  # 立即跑一次（调试时取消注释）
    schedule.every().day.at("09:10").do(run_once, logger)
    while True:
        schedule.run_pending()
        time.sleep(1)


if __name__ == "__main__":
    schedule_task()