spider_work/jiwu_spider/jw_onsale_report.py

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

  逻辑对齐参考项目 deca 的在售报表：只读库不触发抓取（抓取由 jw_onsale_spider 独立定时落库）。
  每天 09/15/20/01 四档各生成一份带小时的独立文件（每档=当刻实时在售快照，互不覆盖）。
  Sheet：概览 / 今日新增商家 / 商品明细（今日新上架行淡红高亮）/ 在售趋势（按每日快照差分，最近4天）。
  金额字段库中已是元(入库时已换算)，直接展示。
  """
import os
import sys
import time
from datetime import 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": "原盒"}  # 商品类型码→名
OUT_DIR = "./reports"                 # 报表输出目录（留档不删）
OUT_PREFIX = "集物星球在售日报"        # 文件名前缀
SEND_WECHAT = True                    # 是否发企微
REPORT_TIMES = ("09:00", "15:00", "20:00", "01:00")  # 四档生成时间(上午/下午/晚上/凌晨场)，对齐 deca 在售采集
TREND_DAYS = 4                        # 在售趋势展示天数(今日+前3日)
SHOP_LIMIT = 100                      # 「其他商家」sheet 存量商家展示上限
LISTING_HOUR_DAYS = 7                 # 「上架时段分布」统计近 N 天

logger.remove()
logger.add("./logs/onsale_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 build_overview(wb, pool) -> None:
    """写「概览」sheet：商家/在售/今日新增等 KPI 竖排。

    Args:
        wb: 工作簿。
        pool: 数据库连接池。
    """
    shop_total = pool.select_all("SELECT COUNT(*) FROM jw_shop_record")[0][0]
    onsale_total = pool.select_all("SELECT COUNT(*) FROM jw_onsale_product_record WHERE is_on_sale=1")[0][0]
    new_shops = pool.select_all("SELECT COUNT(*) FROM jw_shop_record WHERE DATE(gmt_create_time)=CURDATE()")[0][0]
    new_goods = pool.select_all(
        "SELECT COUNT(*) FROM jw_onsale_product_record WHERE is_on_sale=1 AND DATE(sold_time)=CURDATE()")[0][0]
    sold_total = pool.select_all(  # 在售商品已售份数合计(sold_count=stock-residue)，对齐 deca「今日累计已售(份)」
        "SELECT COALESCE(SUM(sold_count),0) FROM jw_onsale_product_record WHERE is_on_sale=1")[0][0]
    ws = xl.add_sheet(wb, "概览")
    xl.write_table(ws, 1, ["指标", "数值"], [
        ("商家总数", shop_total),
        ("在售商品总数", onsale_total),
        ("今日新增商家", new_shops),
        ("今日新增商品", new_goods),
        ("今日累计已售(份)", int(sold_total or 0)),
        ("生成时间", datetime.now().strftime("%Y-%m-%d %H:%M:%S")),
    ], ["text", "int"], freeze=False)


def build_new_shops(wb, pool) -> None:
    """写「今日新增商家」sheet：当天入库的商家一行一条。

    Args:
        wb: 工作簿。
        pool: 数据库连接池。
    """
    rows = pool.select_all(
        "SELECT s.corp_name, s.fans_amount, "
        "SUM(CASE WHEN p.is_on_sale=1 THEN 1 ELSE 0 END) AS onsale_cnt, "
        "SUM(CASE WHEN DATE(p.sold_time)=CURDATE() AND p.is_on_sale=1 THEN 1 ELSE 0 END) AS today_new "
        "FROM jw_shop_record s LEFT JOIN jw_onsale_product_record p ON p.corp_info_id = s.corp_info_id "
        "WHERE DATE(s.gmt_create_time)=CURDATE() "
        "GROUP BY s.corp_info_id, s.corp_name, s.fans_amount "
        "ORDER BY onsale_cnt DESC, s.fans_amount DESC") or []
    ws = xl.add_sheet(wb, "今日新增商家")
    xl.write_table(ws, 1, ["商家", "粉丝数", "在售商品数", "今日新上架"],
                   [(r[0], r[1], int(r[2] or 0), int(r[3] or 0)) for r in rows],
                   ["text", "int", "int", "int"])


def build_other_shops(wb, pool) -> None:
    """写「其他商家」sheet：非今日新增的存量商家（在售数倒序），一行一条。

    对齐 deca：商家表 LEFT JOIN 在售商品表，取 `gmt_create_time != CURDATE()` 的存量商家，
    与「今日新增商家」sheet 互补。

    Args:
        wb: 工作簿。
        pool: 数据库连接池。
    """
    rows = pool.select_all(
        "SELECT s.corp_name, s.fans_amount, "
        "SUM(CASE WHEN p.is_on_sale=1 THEN 1 ELSE 0 END) AS onsale_cnt, "
        "SUM(CASE WHEN DATE(p.sold_time)=CURDATE() AND p.is_on_sale=1 THEN 1 ELSE 0 END) AS today_new "
        "FROM jw_shop_record s LEFT JOIN jw_onsale_product_record p ON p.corp_info_id = s.corp_info_id "
        "WHERE DATE(s.gmt_create_time) != CURDATE() "
        "GROUP BY s.corp_info_id, s.corp_name, s.fans_amount "
        "ORDER BY onsale_cnt DESC, s.fans_amount DESC LIMIT %s", (SHOP_LIMIT,)) or []
    ws = xl.add_sheet(wb, "其他商家")
    xl.write_table(ws, 1, ["商家", "粉丝数", "在售商品数", "今日新上架"],
                   [(r[0], r[1], int(r[2] or 0), int(r[3] or 0)) for r in rows],
                   ["text", "int", "int", "int"])


def build_products(wb, pool) -> None:
    """写「商品明细」sheet：全部在售商品，今日新上架行淡红高亮。

    Args:
        wb: 工作簿。
        pool: 数据库连接池。
    """
    recs = pool.select_all(
        "SELECT corp_info_name, goods_name, gift_series, product_type, specification_name, sold_time, "
        "amount, stock_amount, residue_stock_amount, (DATE(sold_time)=CURDATE()) AS is_new "
        "FROM jw_onsale_product_record WHERE is_on_sale=1 ORDER BY corp_info_name, is_new DESC, goods_id")
    headers = ["商家", "商品名", "系列", "类型", "规格", "上架时间", "单价(元)",
               "已售数", "总份数", "进度", "已售总价(元)", "新品"]
    col_types = ["text", "text", "text", "text", "text", "text", "money",
                 "int", "int", "pct", "money", "text"]
    rows, new_flags = [], []
    for (corp, name, ip, ptype, spec, sold_time, amount, stock, residue, is_new) in recs:
        sold = (stock - residue) if isinstance(stock, int) and isinstance(residue, int) else None
        progress = (sold / stock) if isinstance(sold, int) and isinstance(stock, int) and stock > 0 else 0
        gmv = (sold * float(amount)) if isinstance(sold, int) and amount is not None else None  # amount 已是元(Decimal)
        rows.append((corp, name, ip, PRODUCT_TYPE_NAME.get(str(ptype), ptype), spec, fmt_dt(sold_time),
                     yuan(amount), sold, stock, progress, gmv, "🆕" if is_new else ""))
        new_flags.append(bool(is_new))
    ws = xl.add_sheet(wb, "商品明细")
    xl.write_table(ws, 1, headers, rows, col_types, new_flags=new_flags)


def fetch_onsale_trend(pool, days: int) -> list:
    """按每日快照差分出在售趋势（对齐 deca：集合差集算净新增）。

    查最近 days 天(WHERE 多覆盖一天做最早展示日的差分基线)的快照，按 snapshot_date 聚合成
    商家集合/商品集合；每天净新增 = 当日集合 - 前一日集合（集合差集，非数值相减）。

    Args:
        pool: 数据库连接池。
        days (int): 展示天数（今日 + 前 days-1 日）。

    Returns:
        list: 按日期倒序（最近在前）的 dict 列表，每项含
            {日期, 在售商家数, 在售拼团数, 新增商家数, 新增拼团数}；最早一天无基线时新增为 None。
    """
    rows = pool.select_all(
        "SELECT snapshot_date, corp_info_id, goods_id FROM jw_onsale_daily_record "
        "WHERE snapshot_date >= CURDATE() - INTERVAL %s DAY", (days,)) or []
    day_shops, day_goods = {}, {}
    for snap_date, cid, gid in rows:
        day_shops.setdefault(snap_date, set()).add(cid)
        day_goods.setdefault(snap_date, set()).add(gid)
    dates = sorted(day_shops.keys(), reverse=True)  # 新→旧
    result = []
    for snap_date in dates[:days]:
        prev = snap_date - timedelta(days=1)  # 前一日基线
        shops, goods = day_shops[snap_date], day_goods[snap_date]
        if prev in day_shops:
            new_shops = len(shops - day_shops[prev])  # 净新增商家=当日有、前日无
            new_goods = len(goods - day_goods[prev])  # 净新增拼团=当日有、前日无
        else:
            new_shops = new_goods = None  # 无基线，诚实留空
        result.append({"日期": str(snap_date), "在售商家数": len(shops), "在售拼团数": len(goods),
                       "新增商家数": new_shops, "新增拼团数": new_goods})
    return result


def build_trend(wb, trend: list) -> None:
    """写「在售趋势」sheet：最近 TREND_DAYS 天在售商家/拼团数 + 差分净新增（日期倒序）。

    Args:
        wb: 工作簿。
        trend (list): fetch_onsale_trend 结果。
    """
    headers = ["日期", "在售商家数", "在售拼团数", "新增商家数(差分)", "新增拼团数(差分)"]
    col_types = ["text", "int", "int", "int", "int"]
    data = [(t["日期"], t["在售商家数"], t["在售拼团数"], t["新增商家数"], t["新增拼团数"]) for t in trend]
    ws = xl.add_sheet(wb, "在售趋势")
    xl.write_table(ws, 1, headers, data, col_types)


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 build_listing_hour_dist(wb, pool) -> None:
    """写「上架时段分布」sheet：近 LISTING_HOUR_DAYS 天按上架小时(0~23)统计上架商品数。

    对齐 deca：按 `HOUR(sold_time)`(上架/开售时间) 分 24 桶，第三列为 █ 条形分布图。

    Args:
        wb: 工作簿。
        pool: 数据库连接池。
    """
    since = (datetime.now() - timedelta(days=LISTING_HOUR_DAYS)).strftime("%Y-%m-%d")
    dist = [0] * 24
    for h, c in pool.select_all(
            "SELECT HOUR(sold_time) h, COUNT(*) c FROM jw_onsale_product_record "
            "WHERE sold_time IS NOT NULL AND sold_time >= %s GROUP BY h", (since,)) or []:
        if h is not None and 0 <= int(h) < 24:
            dist[int(h)] = int(c)
    vmax = max(dist) if dist else 0
    data = [(f"{h:02d}时", dist[h], _bar(dist[h], vmax)) for h in range(24)]
    ws = xl.add_sheet(wb, "上架时段分布")
    xl.write_table(ws, 1, ["时段", "上架数", f"分布(近{LISTING_HOUR_DAYS}日累计)"], data,
                   ["text", "int", "text"])


def build_report(pool, out_file: str) -> None:
    """汇总各 sheet 生成在售日报 Excel（6 sheet，对齐 deca）。

    Args:
        pool: 数据库连接池。
        out_file (str): 输出文件路径。
    """
    wb = xl.new_workbook()
    build_overview(wb, pool)                                  # 1 概览
    build_new_shops(wb, pool)                                 # 2 今日新增商家
    build_other_shops(wb, pool)                               # 3 其他商家(存量)
    build_products(wb, pool)                                  # 4 商品明细
    build_trend(wb, fetch_onsale_trend(pool, TREND_DAYS))     # 5 在售趋势(快照差分)
    build_listing_hour_dist(wb, pool)                         # 6 上架时段分布
    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}_{datetime.now():%Y%m%d_%H时}.xlsx"))
    try:
        build_report(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():
    """定时入口：每天 09:00 / 15:00 / 20:00 / 01:00 各生成并发送一次在售日报（四档，各出一份带小时的文件）。"""
    # run_once(logger)  # 调试时取消注释立即跑一次
    for _hhmm in REPORT_TIMES:
        schedule.every().day.at(_hhmm).do(run_once, logger)
    while True:
        schedule.run_pending()
        time.sleep(1)


if __name__ == "__main__":
    schedule_task()