spider_work/deca_spider/stats/weekly_report.py

# -*- coding: utf-8 -*-
# Author : Charley
# Python : 3.12.10
# Date   : 2026/08/24
"""得卡 DECA · 魔都兄弟球星卡「周报」统计（单 Sheet：魔都明细）。

在每日报告(daily_report.py)之外新增的每周任务：把「魔都明细」sheet 的口径由「单日单场」
放宽到「上一个完整自然周（周一~周日）」，统计该周内魔都(881226408)所有成交组队(拼团商品)。

与每日报告的关系：
    - 明细列、汇总块、样式、里程碑用时算法，全部直接复用 daily_report，本文件不重复实现
      渲染/样式，只重写 2 个「周口径」取数函数（把时间窗 WIN_P 换成 WIN_W）。
    - 不含每日报告魔都明细尾部的「购买记录覆盖检测」小节（主公要求周报去掉）。
    - daily_report.py 一个字不改；本文件为纯新增。

时间窗口（WIN_W）：上一个自然周 [上周一 00:00:00, 本周一 00:00:00)（左闭右开，含上周一~上周日
    整 7 天，按 completed_at 自然日历切分）。基准用 MySQL WEEKDAY()（0=周一..6=周日）从
    CURDATE() 回退到本周一，再减 7 天得上周一——故本任务定在每周一早上跑，正好汇总刚结束的完整周。
    注：魔都夜间场常成交到次日凌晨，按自然日历切分时，某周日夜场溢出到周一 00:xx 的团会计入
    「下一周」，此为主公选定的自然周(周一~周日)口径，非漏统计。

明细列（16 列，与 daily_report 魔都明细 sheet 完全一致，见 daily_report.MODDU_DETAIL_COLS）：
    序号/团名(商品标题)/系列/类型/单价/总份数/进度%/总金额/参与人数(本团)/中卡人数/
    开售时间/成交时间/售卖时长/到25%用时/到50%用时/到75%用时

口径说明（同 daily_report）：
    - 销售额 = SUM(COALESCE(team_total_amount, sold_count * unit_price))（随机团按 teams 精算）。
    - 成团数 = 该周成交的拼团商品数。
    - 参与人数(本团) = 各团 deca_buy_record 去重买家 user_id；汇总「参与人数(真实买家)」= 跨周内
      全部成交团去重（故明细逐团相加人次 ≥ 汇总去重人头）。
    - 中卡人数 = 该团拆卡报告 hit_user_nickname 去重（中卡近似）。
    - 到 25/50/75% 用时 = deca_onsale_product_progress_record 首次 pct≥X 的快照时刻 − 开售时间；
      首张快照已越阈值(坍缩)则留空。progress 表 2026/08/11 上线，更早的团相应列可能为空。

从 stats 目录运行：python weekly_report.py（cwd=stats，mysql_pool 读 stats/application.yml）。
"""
import os
import sys
import time
from datetime import timedelta

import schedule
from loguru import logger
from openpyxl import Workbook

# 把项目根目录加入 import 路径（与 daily_report 一致，便于潜在的根目录模块导入）
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, os.path.dirname(BASE_DIR))

from mysql_pool import MySQLConnectionPool
# 复用每日报告的渲染层/样式/列规格/无窗口依赖的纯算法（本文件不重复实现这些）
from daily_report import (
    MODDU_MID, MODDU_DETAIL_COLS, DETAIL_WIDTHS_MODDU,
    _pack_summary, _build_detail_sheet,
)

# 日志：按天切分文件，保留 7 天（常驻定时运行）。放本文件所在目录的 logs/，不依赖 cwd。
# 注：导入 daily_report 时其模块级已 logger.add 过每日报告 sink，这里 remove 后只保留周报 sink。
logger.remove()
logger.add(os.path.join(BASE_DIR, "logs", "weekly_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")

OUT_PREFIX = "得卡-魔都-已售周报告"     # 输出文件名前缀(只查魔都一家)，后缀加「上周一_上周日」两个日期
# 企微发送：报告生成后把 Excel 发到企业微信群机器人（群由 auto_send_wx_msg.WEBHOOK_URL 决定，与每日报告同群）
SEND_WECHAT = True

# ---- 周时间窗（WIN_W）：上一个自然周 [上周一 00:00:00, 本周一 00:00:00) 左闭右开 ----
# 本周一：WEEKDAY() 0=周一..6=周日，从今天回退到本周一 00:00:00（DATE，不含时分秒即 00:00:00）
_THIS_MONDAY = "(CURDATE() - INTERVAL WEEKDAY(CURDATE()) DAY)"
# 上周一 = 本周一 - 7 天
_LAST_MONDAY = f"({_THIS_MONDAY} - INTERVAL 7 DAY)"


def _win(alias: str) -> str:
    """生成某表别名在「上一个自然周」窗口内的 completed_at 过滤子句。

    Args:
        alias (str): SQL 中 deca_product_record 的表别名（如 "p" / "pp"）。

    Returns:
        str: 形如 "p.completed_at >= 上周一 AND p.completed_at < 本周一" 的过滤子句。
    """
    return (f"{alias}.completed_at >= {_LAST_MONDAY} "
            f"AND {alias}.completed_at < {_THIS_MONDAY}")


WIN_W = _win("p")   # 主表 p 的周窗口子句（供各取数 SQL 拼接）


def get_week_window(pool):
    """取上一个自然周的起止日期（供报告标题与文件名展示）。

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

    Returns:
        tuple[date, date]: (上周一 date, 上周日 date)。上周日 = 本周一 - 1 天。
    """
    wk_start, this_monday = pool.select_all(
        f"SELECT {_LAST_MONDAY}, {_THIS_MONDAY}")[0]
    last_sunday = this_monday - timedelta(days=1)   # 本周一(右开界) 前一天即上周日
    return wk_start, last_sunday


def fetch_moddu_summary_week(pool, mid: str) -> dict:
    """统计魔都商家「上一个自然周」的汇总（销售额/成团数/参与人数/均拼单价/人均消费）。

    参与人数用 deca_buy_record 去重真实买家（跨周内全部成交团），人均消费随之按真实人头计。

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

    Returns:
        dict: 含 商家名/商家ID/销售额/成团数/参与人数/均拼单价/人均消费。
    """
    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
    FROM deca_product_record p
    WHERE p.merchant_user_id = %s AND {WIN_W}
      AND p.unit_price IS NOT NULL AND p.sold_count IS NOT NULL
    """
    mname, amount, groups = pool.select_all(sql, (mid,))[0]
    # 参与人数(真实买家) = 周内该商家全部成交团的 deca_buy_record 去重 user_id
    people = pool.select_all(f"""
    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 {_win('pp')}
    """, (mid,))[0][0] or 0
    d = _pack_summary(amount, groups, people)
    d["商家名"] = mname or mid
    d["商家ID"] = mid
    return d


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

    列与算法同 daily_report.fetch_moddu_details，仅时间窗由单日单场换为上一个自然周(WIN_W)：
        - 参与人数：deca_buy_record 去重买家 user_id（本团真实参团人头）。
        - 到 25/50/75% 用时：deca_onsale_product_progress_record 首次 pct≥X 的快照时刻 − 开售时间；
          首张快照已越阈值(坍缩)则留空（判定见 daily_report._milestone_used）。

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

    Returns:
        list[dict]: 每条含 团名/系列/类型/单价/总份数/进度/总金额/参与人数/中卡人数/开售时间/
            成交时间/售卖时长/到25%用时/到50%用时/到75%用时。
    """
    # 里程碑/时长算法直接复用 daily_report（避免重复实现坍缩判定逻辑）
    from daily_report import _fmt_duration, _milestone_used
    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_W}
      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 build_week_report(pool, out: str, title: str):
    """生成单 Sheet（魔都明细）周报 Excel。

    仅一个 sheet「魔都明细」：汇总块 + 每条组队明细（16 列），结构/样式复用
    daily_report._build_detail_sheet。不含「购买记录覆盖检测」小节（miss_info=None）。

    Args:
        pool (MySQLConnectionPool): MySQL 连接池。
        out (str): 导出的 xlsx 路径。
        title (str): sheet 顶部分区标题（含商家名与周成交时间窗）。
    """
    summ = fetch_moddu_summary_week(pool, MODDU_MID)
    details = fetch_moddu_details_week(pool, MODDU_MID)

    wb = Workbook()
    ws = wb.active
    ws.title = "魔都明细"
    # is_real=True（魔都为真实买家口径）；span/widths 取魔都明细专属规格；
    # miss_info=None → 不输出「购买记录覆盖检测」小节（主公要求周报去掉此块）
    _build_detail_sheet(ws, title, summ, details, MODDU_DETAIL_COLS,
                        True, len(MODDU_DETAIL_COLS), DETAIL_WIDTHS_MODDU, miss_info=None)
    wb.save(out)


def run_once(log) -> str:
    """连库生成上一个自然周的魔都明细周报，落地到 stats 目录并发送到企业微信群。

    Args:
        log: 日志对象。

    Returns:
        str: 生成的 xlsx 绝对路径；数据库连接池异常时返回空串。
    """
    log.info("开始生成魔都周报" + "." * 30)
    pool = MySQLConnectionPool(log=log)
    if not pool.check_pool_health():
        log.error("数据库连接池异常")
        return ""
    wk_start, last_sunday = get_week_window(pool)
    title = (f"魔都兄弟球星卡 · 周汇总"
             f"（成交自然周 {wk_start} 00:00:00 ~ {last_sunday} 23:59:59）")
    # 输出锚定到本脚本所在目录(stats)，文件名带「上周一_上周日」两个日期，便于归档区分
    out_file = os.path.join(BASE_DIR, f"{OUT_PREFIX}_{wk_start:%Y%m%d}_{last_sunday:%Y%m%d}.xlsx")
    build_week_report(pool, out_file, title)
    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:20 生成上一个自然周的魔都明细周报。

    错开每日报告(09:10)10 分钟；周一早上跑正好汇总刚结束的完整自然周(上周一~周日)。
    """
    # run_once(logger)  # 立即跑一次（调试时取消注释）
    schedule.every().monday.at("09:20").do(run_once, logger)
    while True:
        schedule.run_pending()
        time.sleep(1)


if __name__ == "__main__":
    schedule_task()