|
@@ -0,0 +1,639 @@
|
|
|
|
|
+# -*- 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()
|