# -*- 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()