fix(buy_record): 优化售罄判断逻辑及结束战报内容

- 改用已售份数(soldCount)与总份数(totalCardCount)比较判断售罄，避免依赖库存(availableStock)误判预售和直播售卖中断
- 修正直播卖车售罄判定，解决库存瞬时为0导致漏采和假战报问题
- 新增购买时间、组齐耗时等结束战报信息，提升战报丰富度
- 统计同系列及同商家上一辆车的重复买家人数作为战报附加内容
- 按业务日窗口统计当日重复购买人数，修正重复购买计算口径
- 添加辅助函数处理购买时间获取、时间格式化、数据库购买记录统计等功能
- 优化监控数据库已有商品信息，支持新字段merchant_user_id读取与使用

deca_spider/buy_record_analysis/buy_record_spider.py

@@ -231,9 +231,10 @@ def is_sale_ended(data: dict, now_ts: int) -> tuple[bool, str]:
     """根据详情返回判断商品售卖是否已结束（售罄或到结束时间）。
 
     预售/未开卖的车：详情接口 availableStock 也返回 0（库存未分配）、soldCount=0、saleStartAt 在未来，
-    仅凭 availableStock<=0 会把它误判为「售罄」→ 被加入 _ended_codes 拉黑，导致真正开卖后整车漏采
-    （2026/08/24 修复 #342 预售车：publishAt 21:26 发布预告、saleStartAt 22:45 才开卖，其间详情
-    availableStock=0）。故加两道护栏：① saleStartAt 未到直接判未结束；② 售罄须 soldCount>0（确实卖过）。
+    仅凭 availableStock<=0 会把它误判为「售罄」→ 被加入 _ended_codes 拉黑，导致真正开卖后整车漏采。
+    直播实时放量卖的车同理：卖到一半 availableStock 会瞬时抖成 0（"这一刻没货可抢"≠"卖光"）。
+    故 ① saleStartAt 未到直接判未结束；② 售罄改判 soldCount>=totalCardCount（卖满才算），彻底不看 availableStock
+    （2026/09/04：GB26090464871 卖 2/31 时 availableStock=0 被旧逻辑误判售罄、拉黑，漏采后续爆发，0 人参与）。
 
     Args:
         data (dict): 详情接口 data 层。
@@ -251,11 +252,15 @@ def is_sale_ended(data: dict, now_ts: int) -> tuple[bool, str]:
                 return False, ""
         except (ValueError, OverflowError):
             pass
-    # ② 售罄：库存<=0 且确实卖出过(soldCount>0)，避免预售/未开卖车(soldCount=0)被误判售罄
-    stock = data.get("availableStock")
+    # ② 售罄：改用「已售 >= 总份数」判定，不再用 availableStock。
+    #    直播实时放量卖的车，availableStock 反映的是「当前这一刻放出、还没被抢的量」而非总剩余，
+    #    卖到一半也会瞬时抖成 0（"这一秒没货可抢"≠"卖光"；实测 GB26090464871 卖 2/31 时 availableStock=0）。
+    #    旧逻辑凭 availableStock<=0 会把它误判售罄→拉黑→漏采后续真实抢车爆发（2026/09/04，0 人参与拆卡）。
+    #    detail 的 totalCardCount 实测可靠(=总份数)；真售罄另有 is_on_sale=0 那道兜底移出监控，故从严只认卖满。
     sold = data.get("soldCount")
-    if stock is not None and stock <= 0 and sold is not None and sold > 0:
-        return True, f"售罄(availableStock={stock}, soldCount={sold})"
+    total = data.get("totalCardCount")
+    if sold is not None and total is not None and total > 0 and sold >= total:
+        return True, f"售罄(soldCount={sold}/{total})"
     end_text = data.get("saleEndAt")
     if end_text:
         try:

deca_spider/onsale_alert_spider.py

@@ -61,6 +61,7 @@ RUN_START = dtime(20, 30)          # 运行窗口开始默认 20:30；可由命
 RUN_END = dtime(6, 0)              # 运行窗口结束：次日 06:00（窗口跨午夜；2026/08/15 由 03:00 延到 06:00，昨天 5 点还在播）
 MAX_PROD_PAGES = 100               # 在售商品翻页保护上限
 T_ALERT = "deca_onsale_alert_record"
+T_PROD = core.T_PROD
 ON_SALE_PATH = "/api/v1/app/groupbuy/merchant/on-sale-list"       # 商家在售商品列表(need_auth)
 DETAIL_PATH = "/api/v1/app/groupbuy/detail"                       # 商品详情(取 publishAt 判新品)
 
@@ -363,9 +364,9 @@ def _is_sale_ended(data: dict, now_ts: int) -> tuple[bool, str]:
     详情请求失败时 data 为空 dict，两条件均不命中返回未结束——保守，防抓取抖动误判下架。
 
     预售/未开卖的车：详情接口 availableStock 也返回 0（库存未分配）、soldCount=0、saleStartAt 在未来，
-    仅凭 availableStock<=0 会把它误判为「售罄」→ 发出假的「一车结束」战报（2026/08/24 修复 #342
-    预售车：publishAt 21:26 发布预告、saleStartAt 22:45 才开卖，其间短暂闪现在售列表又撤下被误判结束）。
-    故加两道护栏：① saleStartAt 未到直接判未结束；② 售罄须 soldCount>0（确实卖过）。
+    仅凭 availableStock<=0 会把它误判为「售罄」→ 发出假的「一车结束」战报。直播实时放量卖的车同理：
+    卖到一半 availableStock 也会瞬时抖成 0。故 ① saleStartAt 未到直接判未结束；② 售罄改判 soldCount>=totalCardCount
+    （卖满才算），彻底不看 availableStock（2026/09/04：GB26090464871 卖 2/31 时 availableStock=0 曾致 buy_record 误杀）。
 
     Args:
         data (dict): 商品详情接口(groupbuy/detail) data 层，可能为空 dict。
@@ -383,11 +384,14 @@ def _is_sale_ended(data: dict, now_ts: int) -> tuple[bool, str]:
                 return False, ""
         except (ValueError, OverflowError):
             pass
-    # ② 售罄：库存<=0 且确实卖出过(soldCount>0)，避免预售/未开卖车(soldCount=0)被误判售罄
-    stock = data.get("availableStock")
+    # ② 售罄：改用「已售 >= 总份数」判定，不再用 availableStock。
+    #    直播实时放量卖的车，availableStock 是「当前放出、还没被抢的量」而非总剩余，卖到一半也会瞬时抖成 0，
+    #    旧逻辑凭 availableStock<=0 会把仍在售的车误判售罄发假战报（2026/09/04 GB26090464871 卖 2/31 时 availableStock=0）。
+    #    本函数只在「车已离开在售列表」后作二次确认，从严只认卖满，与下方「结束即全部售出」口径一致。
     sold = data.get("soldCount")
-    if stock is not None and stock <= 0 and sold is not None and sold > 0:
-        return True, f"售罄(availableStock={stock}, soldCount={sold})"
+    total = data.get("totalCardCount")
+    if sold is not None and total is not None and total > 0 and sold >= total:
+        return True, f"售罄(soldCount={sold}/{total})"
     end_text = data.get("saleEndAt")
     if end_text:
         try:
@@ -468,6 +472,120 @@ def _count_buyers(pool, code: str) -> int:
     return 0
 
 
+def _get_last_buy_time(pool, code: str):
+    """取该车购买记录里最新一笔的 purchased_at。"""
+    rows = pool.select_all(
+        "SELECT MAX(purchased_at) FROM deca_buy_record WHERE product_code=%s", code)
+    if rows and rows[0] and rows[0][0] is not None:
+        return rows[0][0]
+    return None
+
+
+def _parse_dt(value) -> datetime | None:
+    """把接口/数据库时间值转成 datetime。"""
+    if isinstance(value, datetime):
+        return value
+    if not value:
+        return None
+    try:
+        return datetime.strptime(str(value), "%Y-%m-%d %H:%M:%S")
+    except ValueError:
+        return None
+
+
+def _count_overlap_buyers(pool, code_a: str, code_b: str) -> int:
+    """统计两辆车去重买家交集人数。"""
+    rows = pool.select_all(
+        "SELECT COUNT(DISTINCT a.user_id) "
+        "FROM deca_buy_record a "
+        "WHERE a.product_code=%s "
+        "AND EXISTS (SELECT 1 FROM deca_buy_record b "
+        "            WHERE b.product_code=%s AND b.user_id=a.user_id)",
+        (code_a, code_b))
+    if rows and rows[0] and rows[0][0] is not None:
+        return int(rows[0][0])
+    return 0
+
+
+def _count_repeat_buyers_today(pool, merchant_user_id: str, code: str,
+                               window_start: str, window_end: str) -> int:
+    """统计当前车买家里，当日(运行窗口内)重复买过本商家其他车的人数。"""
+    rows = pool.select_all(
+        "SELECT COUNT(DISTINCT b.user_id) "
+        "FROM deca_buy_record b "
+        "WHERE b.product_code=%s "
+        "AND EXISTS (SELECT 1 FROM deca_buy_record x "
+        "            WHERE x.merchant_user_id=%s "
+        "              AND x.user_id=b.user_id "
+        "              AND x.product_code<>%s "
+        "              AND x.purchased_at>=%s AND x.purchased_at<=%s)",
+        (code, merchant_user_id, code, window_start, window_end))
+    if rows and rows[0] and rows[0][0] is not None:
+        return int(rows[0][0])
+    return 0
+
+
+def _fetch_previous_realtime(pool, merchant_user_id: str, code: str, before_text: str,
+                             series_name: str | None = None) -> dict | None:
+    """按购买记录实时找某商家在当前车之前结束的上一辆车；可选限制同 series_name。"""
+    if not before_text:
+        return None
+    sql = (
+        "SELECT b.product_code, MAX(b.title) AS title, MAX(b.purchased_at) AS ended_at, "
+        "       COALESCE(NULLIF(MAX(p.series_name), ''), NULLIF(MAX(o.series_name), '')) AS series_name "
+        "FROM deca_buy_record b "
+        f"LEFT JOIN {T_PROD} p ON p.product_code=b.product_code "
+        "LEFT JOIN deca_onsale_product_record o ON o.product_code=b.product_code "
+        "WHERE b.merchant_user_id=%s AND b.product_code<>%s AND b.purchased_at<%s")
+    args = [merchant_user_id, code, before_text]
+    if series_name:
+        sql += (
+            " AND COALESCE(NULLIF(p.series_name, ''), NULLIF(o.series_name, ''), '')=%s")
+        args.append(series_name)
+    sql += " GROUP BY b.product_code ORDER BY ended_at DESC LIMIT 1"
+    rows = pool.select_all(sql, tuple(args))
+    if not rows:
+        return None
+    prev_code, title, ended_at, series = rows[0]
+    return {
+        "product_code": prev_code,
+        "title": title,
+        "ended_at": ended_at,
+        "series_name": series,
+    }
+
+
+def _biz_window(dt: datetime) -> tuple[str, str]:
+    """把某时刻归入其成交业务日窗口 [D-1 13:00, D 06:00]（口径与 daily_report 一致，2026/09/01 起点由 17:00 提前到 13:00）。
+
+    夜市成交(13:00~次日06:00)算一个业务日：时刻 ≥13:00 归当日窗口起点，<06:00 归昨日窗口。战报只在
+    20:30~06:00 发、组齐时刻(取 buy_record 最后一单近似)必落该区间，两分支都正确。
+
+    Args:
+        dt (datetime): 参照时刻（车的组齐时刻）。
+
+    Returns:
+        tuple[str, str]: (窗口起, 窗口止)，均 'YYYY-MM-DD HH:MM:SS'。
+    """
+    if dt.time() >= dtime(13, 0):
+        start = datetime.combine(dt.date(), dtime(13, 0))
+    else:
+        start = datetime.combine(dt.date() - timedelta(days=1), dtime(13, 0))
+    end = datetime.combine(start.date() + timedelta(days=1), dtime(6, 0))
+    return start.strftime("%Y-%m-%d %H:%M:%S"), end.strftime("%Y-%m-%d %H:%M:%S")
+
+
+def _fmt_duration(seconds: int | None) -> str:
+    """把秒数格式化成“X小时X分钟”或“X分钟”。"""
+    if seconds is None or seconds < 0:
+        return "-"
+    total_minutes = max(0, int(seconds) // 60)
+    hours, minutes = divmod(total_minutes, 60)
+    if hours:
+        return f"{hours}小时{minutes:02d}分钟"
+    return f"{minutes}分钟"
+
+
 def _detect_and_report_ended(log, pool, existing: dict, onsale_codes: set):
     """检测本进程运行后消失的车、二次确认结束后逐辆发战报并置位 ended_notified。
 
@@ -505,6 +623,10 @@ def _detect_and_report_ended(log, pool, existing: dict, onsale_codes: set):
             _mark_ended_notified(pool, code)
             log.info(f"[结束静默] {code} soldCount=0 流车(无人成交/撤下)，不发战报并置位防重扫 | {info['title']}")
             continue
+        end_dt = _parse_dt(_get_last_buy_time(pool, code))
+        end_text = end_dt.strftime("%Y-%m-%d %H:%M:%S") if end_dt else ""
+        sale_start_text = data.get("saleStartAt") or ""
+        sale_start_dt = _parse_dt(sale_start_text)
         # 售出件数=总份数(结束即全部售出)：优先详情最新 cardCount，缺失回退库内旧值
         card = data.get("cardCount")
         if card is None:
@@ -517,6 +639,50 @@ def _detect_and_report_ended(log, pool, existing: dict, onsale_codes: set):
         buyers = _count_buyers(pool, code)  # 该车去重购买人数(参与拆卡人数)
         item_text = _fmt_ended(info["title"], card, price_text, buyers,
                                plain=(SEND_CHANNEL == "pc"))
+        extra_lines = []
+        if end_dt:
+            if sale_start_dt:
+                duration_text = _fmt_duration(int((end_dt - sale_start_dt).total_seconds()))
+                extra_lines.append(f"⏱ 组齐时间 {end_dt:%m-%d %H:%M} ｜ 共花费 {duration_text}")
+            else:
+                extra_lines.append(f"⏱ 组齐时间 {end_dt:%m-%d %H:%M} ｜ 共花费 -")
+        else:
+            extra_lines.append("⏱ 组齐时间 未知 ｜ 共花费 -")
+
+        current_series = (data.get("giftInfo") or {}).get("items") or [{}]
+        current_series_name = ""
+        if current_series:
+            first = current_series[0] or {}
+            current_series_name = (first.get("seriesName") or "").strip()
+
+        mid = info["mid"] or MERCHANT_ID
+        same_series_prev = _fetch_previous_realtime(
+            pool, mid, code, end_text, current_series_name or None)
+        if same_series_prev:
+            same_series_overlap = _count_overlap_buyers(
+                pool, code, same_series_prev["product_code"])
+            extra_lines.append(
+                f"🔁 同系列上一辆 {same_series_prev['title']}（重复 {same_series_overlap} 人）")
+        else:
+            extra_lines.append("🔁 同系列上一辆 无（重复 0 人）")
+
+        merchant_prev = _fetch_previous_realtime(pool, mid, code, end_text)
+        if merchant_prev:
+            merchant_overlap = _count_overlap_buyers(pool, code, merchant_prev["product_code"])
+            extra_lines.append(
+                f"🔁 商家上一辆 {merchant_prev['title']}（重复 {merchant_overlap} 人）")
+        else:
+            extra_lines.append("🔁 商家上一辆 无（重复 0 人）")
+
+        # 📅 当日重复购买（改进1，2026/09/01）：按「组齐时刻」所在业务日窗口 [D-1 13:00, D 06:00]、仅本商家，
+        # 口径与 daily_report 一致；修掉原先用 _window_start(datetime.now()) 造成的窗口错位(补发/非当晚结束车会误算 0)。
+        if end_dt:
+            win_s, win_e = _biz_window(end_dt)
+            repeat_today = _count_repeat_buyers_today(pool, mid, code, win_s, win_e)
+            extra_lines.append(f"📅 当日重复购买 {repeat_today} 人")
+        else:
+            extra_lines.append("📅 当日重复购买 - 人")
+        item_text = item_text + "\n" + "\n".join(extra_lines)
         if _send_ended(log, info["mname"] or MERCHANT_ID, item_text):
             _mark_ended_notified(pool, code)
             log.info(f"[结束战报已发] {code} {reason} 售出{card}件 {buyers}人 | {info['title']}")
@@ -539,12 +705,12 @@ def run_once(log, pool):
     # 库内该商家已监控商品：含结束标记与结束战报所需的静态字段(标题/商家名/单价/份数)
     existing_rows = pool.select_all(
         "SELECT product_code, new_notified, half_notified, ended_notified, "
-        "title, merchant_name, unit_price, card_count, sold_count "
+        "title, merchant_user_id, merchant_name, unit_price, card_count, sold_count "
         f"FROM {T_ALERT} WHERE merchant_user_id=%s", MERCHANT_ID) or []
     existing = {}
-    for code, new, half, ended, title, mname_, uprice, ccount, scount in existing_rows:
+    for code, new, half, ended, title, mid_, mname_, uprice, ccount, scount in existing_rows:
         existing[code] = {"new": new, "half": half, "ended": ended,
-                          "title": title, "mname": mname_, "unit_price": uprice,
+                          "title": title, "mid": mid_, "mname": mname_, "unit_price": uprice,
                           "card": ccount, "sold": scount}
     is_cold = len(existing) == 0  # 冷启动：库内该商家零记录（仅作日志提示，逻辑与常规轮一致）
     if is_cold: