spider_work/deca_spider/onsale_alert_spider.py

# -*- coding: utf-8 -*-
# Author : Charley
# Python : 3.12.10
# Date   : 2026/08/04
"""得卡 DECA · 指定商家在售商品提醒（新品上架 + 进度过半，常驻轮询）。

监控 MERCHANT_ID 商家的在售商品，两类提醒（各发一条独立消息）：
    1. 新商品上架：出现库里没有的新 product_code。
    2. 进度过半：售卖进度 sold_count/card_count 首次达到 HALF_THRESHOLD。
去重靠 deca_onsale_alert_record 两个标记位(new_notified/half_notified)，每类每商品只提醒一次。

发送渠道 SEND_CHANNEL 可切（2026/08/11 起默认企业微信）：
    - "qywx": 企业微信群机器人(markdown_v2)——发到 WEBHOOK_URL 配置的群，标题加粗、不带链接。
    - "pc"  : PC 版微信(wxauto4 文本)发给 WX_TARGET——纯文本文案，随时可切回。

消息不带商品链接(2026/08/11)：得卡商品分享落地页(share-detail 接口)只认「登录态(带 token)
生成的 shareCode」，免 token 详情返回的 shareCode 一律被判失效；为不引入登录态、降账号风控，
提醒消息直接不挂链接。三类提醒都发：新商品上架 / 拼团进度过半 / 一车结束战报。

运行时段：仅每天 RUN_START~次日 06:00 轮询（RUN_START/RUN_END 控制），其余时间休眠到下次开窗再跑。
RUN_START 默认 20:30，可用命令行传参覆盖（见下），窗口起点同时是「新上架」时间门槛：只提醒
publishAt 晚于该起点的商品。

从根目录运行：
    python onsale_alert_spider.py            # 默认 20:30 开始
    python onsale_alert_spider.py 17:00      # 改为 17:00 开始：17:00 后的新上架才提醒
    python onsale_alert_spider.py --start 17:00
（默认企微渠道，需 WEBHOOK_URL 配好）
"""
import sys
import time
import random
import argparse
from datetime import datetime, time as dtime, timedelta

from loguru import logger
from tenacity import retry, stop_after_attempt, wait_fixed
from mysql_pool import MySQLConnectionPool

import deca_sold_core as core
import deca_wechat                                   # PC 版微信发送(wxauto4)
from auto_send_wx_msg import send_wechat_group_msg   # 企微机器人发送(备用渠道)
# 复用 daily 的免 token 全站在售拉取（home/search，只拉不落库）；alert 独立自采、不依赖 buy_record 落库
from on_sale.deca_on_sale_daily_spider import fetch_all_onsale

logger.remove()
logger.add("./logs/onsale_alert_{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")

# ==================== 配置 ====================
MERCHANT_ID = "881226408"          # 监控的商家用户ID
HALF_THRESHOLD = 0.5               # 进度过半阈值(0.5=50%)
SEND_CHANNEL = "qywx"              # 发送渠道：qywx=企业微信群机器人(默认) / pc=PC版微信(wxauto4，可切回)
# WX_TARGET = "backup"               # PC 微信发送目标(好友备注名/群名)，改成实际接收人
WX_TARGET = "得卡-通知"               # PC 微信发送目标(好友备注名/群名)，改成实际接收人
# COLD_START_PUSH 已废弃(2026/08/08)：改为「只提醒本轮窗口起点(20:30)后新上架」，冷启动不再发全量快照
MIN_INTERVAL_SEC = 60              # 轮询间隔随机下限(秒)
MAX_INTERVAL_SEC = 90             # 轮询间隔随机上限(秒)，每轮在 [下限,上限] 取随机数，打散规律降风控
RUN_START = dtime(20, 30)          # 运行窗口开始默认 20:30；可由命令行传参覆盖(见 _parse_args)，同时作为「新上架」时间门槛
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 判新品)

# 本进程启动后「见过在售」的本商家 product_code 集合（模块级内存，进程重启即清空）：
# 结束战报只认这些 code——即只播报「程序运行之后」才结束的车；程序启动前就已结束的历史车不补发。
# 与持久化的 ended_notified 列配合：_seen_onsale 决定「管不管」，ended_notified 决定「发没发过」。
_seen_onsale: set = set()


def _get_detail_data(log, code: str) -> dict:
    """打商品详情接口（免 token）返回 data 层。

    Args:
        log: 日志对象。
        code (str): 商品编码(product_code)。

    Returns:
        dict: 详情 data 层；请求失败返回空 dict。
    """
    try:
        resp = core.do_request(log, DETAIL_PATH, {"code": code}, need_auth=False)  # 详情接口免登录，实测不带token也返回
        return (resp or {}).get("data") or {}
    except Exception as e:
        log.warning(f"取详情失败({code}): {e}")
        return {}


def get_publish_at(log, code: str) -> str | None:
    """调详情接口取该商品的上架时间(data.publishAt)，供新品判断用。

    Args:
        log: 日志对象。
        code (str): 商品编码(product_code)。

    Returns:
        str | None: publishAt 原文如 "2026-08-08 20:31:00"；取不到返回 None。
    """
    return _get_detail_data(log, code).get("publishAt")


def _window_start(now: datetime) -> datetime:
    """本轮运行窗口的起点（最近一个已过去的 RUN_START，默认 20:30），作为「新上架」时间门槛。

    RUN_START 可由命令行传参覆盖，故此处读全局值而非写死时刻。

    Args:
        now (datetime): 当前时间。

    Returns:
        datetime: 当天时刻 >= RUN_START → 今天 RUN_START；否则 → 昨天 RUN_START（跨午夜窗口取窗口起点那天）。
    """
    if now.time() >= RUN_START:
        return datetime.combine(now.date(), RUN_START)
    return datetime.combine(now.date() - timedelta(days=1), RUN_START)


def _is_new_arrival(publish_at_text: str, window_start: datetime) -> bool:
    """判断商品上架时间(publishAt)是否晚于窗口起点，即本轮才新上架。

    Args:
        publish_at_text (str): 详情 publishAt，格式 "%Y-%m-%d %H:%M:%S"，可能为空。
        window_start (datetime): 本轮窗口起点(20:30)。

    Returns:
        bool: True=上架时间 >= 窗口起点（本轮新上架）；空值/解析失败按 False（保守，不误报老货）。
    """
    if not publish_at_text:
        return False
    try:
        pub = datetime.strptime(publish_at_text.strip(), "%Y-%m-%d %H:%M:%S")
    except (ValueError, TypeError):
        return False
    return pub >= window_start


def _fmt_money(v) -> str:
    """把价格数值格式化为紧凑字符串：整数去掉小数(80.00→80)，非整去尾零(1.50→1.5)。

    Args:
        v: 价格(Decimal/float/int/str)。

    Returns:
        str: 紧凑价格字符串。
    """
    f = float(v)
    if f == int(f):
        return str(int(f))
    return f"{f:g}"


def _fmt_price(unit, mn, mx) -> str:
    """构造价格文案：区间价(min~max)优先，其次单价，与 App 详情页口径一致。

    得卡「单箱选队」等商品 unitPrice 常为 0，真实价落在 minUnitPrice~maxUnitPrice 区间
    (如详情页「¥1.50 ~ 80」)，故区间存在且非单点时优先展示区间，避免误显 ¥0.00。

    Args:
        unit: 单价 unitPrice(可能为 0/None)。
        mn: 最低单价 minUnitPrice(可能为 None)。
        mx: 最高单价 maxUnitPrice(可能为 None)。

    Returns:
        str: 价格文案，如 "¥1.5~80" / "¥400" / "¥-"。
    """
    if mn is not None and mx is not None and float(mn) != float(mx):
        return f"¥{_fmt_money(mn)}~{_fmt_money(mx)}"      # 区间价（单箱选队等）
    if unit is not None and float(unit) > 0:
        return f"¥{_fmt_money(unit)}"                     # 正常单价
    if mn is not None and float(mn) > 0:
        return f"¥{_fmt_money(mn)}"                       # unit 为 0/None 时退回 min
    return "¥-"


def _fmt_item(r: dict, kind: str, pct: float = None, plain: bool = False) -> str:
    """构造一条通知条目：标题 + 价格 + 份数/进度 + 余·共（不带链接）。

    2026/08/11 起消息不挂商品链接（share-detail 只认带 token 生成的 shareCode，免 token 一律失效，
    为不引入登录态直接去链接）；markdown 渠道把标题加粗，纯文本渠道直出标题。

    Args:
        r (dict): parse_product 产出的商品字典。
        kind (str): "new"=新品(显示份数) / "half"=过半(显示进度%) / "onsale"=当前在售快照(显示进度%)。
        pct (float, optional): 售卖进度百分比(0~100)，kind 为 "half"/"onsale" 时用。Defaults to None。
        plain (bool, optional): True=纯文本(PC微信)/False=markdown(企微)。Defaults to False。

    Returns:
        str: 一条通知文案（两行）。
    """
    # 标题行：markdown 渠道加粗突出，纯文本渠道直出
    title_line = r["title"] if plain else f"**{r['title']}**"
    # 价格：区间价优先(单箱选队等 unitPrice=0 的商品用 min~max)，避免误显 ¥0.00
    price_text = _fmt_price(r.get("unit_price"), r.get("min_unit_price"), r.get("max_unit_price"))
    # 信息行：新品显示份数，过半/当前在售显示进度百分比
    if kind in ("half", "onsale"):
        info = (f"💰 {price_text} ｜ 📈 进度{pct:.0f}% ｜ "
                f"🎯 余{r['available_stock']}/{r['card_count']}")
    else:
        info = (f"💰 {price_text} ｜ 📦 {r['card_count']}份 ｜ "
                f"🎯 余{r['available_stock']}/{r['card_count']}")
    return f"{title_line}\n{info}"


def _dispatch(log, items: list, title: str) -> bool:
    """按 SEND_CHANNEL 把一批条目发出去（pc=PC微信整段文本 / qywx=企微 markdown）。

    Args:
        log: 日志对象。
        items (list[str]): 已构造好的通知条目列表。
        title (str): 消息标题。

    Returns:
        bool: 发送成功返回 True；无条目或发送失败返回 False（调用方据此决定是否置位提醒标记）。
    """
    if not items:
        return False
    if SEND_CHANNEL == "pc":
        # PC 微信：标题 + 编号清单，拼成一整段纯文本，一条消息发给 WX_TARGET
        body = [f"{i}. {it}" for i, it in enumerate(items, 1)]
        text = title + "\n\n" + "\n----------------------------------\n".join(body)
        return bool(deca_wechat.send_text(text, who=WX_TARGET))
    # 企微机器人：成功返回 dict、失败返回 None
    return bool(send_wechat_group_msg(log=log, items=items, title=title))


def _pc_section(title: str, items: list) -> str:
    """把一组条目拼成 PC 微信纯文本的一个分区（小标题 + 编号清单）。

    Args:
        title (str): 分区小标题（如「新商品上架」）。
        items (list[str]): 已构造好的通知条目列表。

    Returns:
        str: 该分区的纯文本（含小标题与编号清单）；items 为空返回空串。
    """
    if not items:
        return ""
    body = [f"{i}. {it}" for i, it in enumerate(items, 1)]
    return f"{title}（{len(items)}款）\n" + "\n----------------------------------\n".join(body)


def _dispatch_pc_combined(log, mname: str, new_items: list, half_items: list) -> bool:
    """PC 微信：把新品与过半两类合并为一条纯文本消息发出（减少操作，一次发完）。

    Args:
        log: 日志对象。
        mname (str): 商家名称，用于消息大标题。
        new_items (list[str]): 新品上架通知条目。
        half_items (list[str]): 进度过半通知条目。

    Returns:
        bool: 发送成功返回 True；无内容或发送失败返回 False（调用方据此决定是否置位新品标记）。
    """
    sections = []
    new_sec = _pc_section("【新商品上架】", new_items)
    half_sec = _pc_section("【拼团进度过半】", half_items)
    if new_sec:
        sections.append(new_sec)
    if half_sec:
        sections.append(half_sec)
    if not sections:
        return False
    header = f"得卡 · {mname} 在售提醒"
    text = header + "\n\n" + "\n\n==================================\n\n".join(sections)
    return bool(deca_wechat.send_text(text, who=WX_TARGET))


def _progress(sold, card) -> float:
    """计算售卖进度(0~1)，卡片份数缺失或为 0 时返回 0。

    Args:
        sold (int | None): 已售份数。
        card (int | None): 总份数。

    Returns:
        float: 进度比例 sold/card；无法计算时返回 0.0。
    """
    if not card or card <= 0 or sold is None:
        return 0.0
    return sold / card


def fetch_onsale(log) -> tuple[list, bool]:
    """免 token 抓全站在售、筛出 MERCHANT_ID 的在售商品（alert 独立自采，不依赖 buy_record 落库）。

    复用 daily 的 fetch_all_onsale（home/search 免 token、只拉不落库）拿全站在售，
    再按 merchant_user_id 筛出本商家、按 product_code 去重。全链路免 token。

    Args:
        log: 日志对象。

    Returns:
        tuple[list, bool]: (本商家在售商品字典列表[parse_product 结果，含 title/unit_price/card_count/
            sold_count/available_stock/merchant_name 等], 全站是否正常翻到底)。
            bool=False 表示全站抓取中途异常/未取全，调用方据此放弃本轮下架/结束对账，避免误判。
    """
    all_products, ok = fetch_all_onsale(log)
    rows = {p["product_code"]: p for p in all_products
            if p.get("merchant_user_id") == MERCHANT_ID}
    log.info(f"商家 {MERCHANT_ID} 当前在售商品 {len(rows)} 个（全站 {len(all_products)} 筛出，免 token，ok={ok}）")
    return list(rows.values()), ok


def _insert_alert(pool, r: dict, progress_pct: float, new_notified: int,
                  half_notified: int):
    """新增一条在售监控记录（share_code 列 2026/08/11 起不再写入，消息已去链接）。

    Args:
        pool (MySQLConnectionPool): MySQL 连接池。
        r (dict): parse_product 产出的商品字典。
        progress_pct (float): 售卖进度百分比(0~100)。
        new_notified (int): 新品提醒标记位 0/1。
        half_notified (int): 过半提醒标记位 0/1。
    """
    sql = (f"INSERT IGNORE INTO {T_ALERT} "
           "(product_code, merchant_user_id, merchant_name, title, unit_price, card_count, "
           "sold_count, progress, available_stock, groupbuy_status_name, "
           "new_notified, half_notified) "
           "VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)")
    args = (r["product_code"], r["merchant_user_id"], r["merchant_name"], r["title"],
            r["unit_price"], r["card_count"], r["sold_count"], round(progress_pct, 2),
            r["available_stock"], r["groupbuy_status_name"],
            new_notified, half_notified)
    pool.insert_many(query=sql, args_list=[args])


def _update_progress(pool, r: dict, progress_pct: float, half_notified: int):
    """更新已有记录的售卖进度等动态字段(及可选的过半标记位)。

    Args:
        pool (MySQLConnectionPool): MySQL 连接池。
        r (dict): parse_product 产出的商品字典。
        progress_pct (float): 售卖进度百分比(0~100)。
        half_notified (int): 过半提醒标记位 0/1(命中过半时置 1)。
    """
    sql = (f"UPDATE {T_ALERT} SET sold_count=%s, progress=%s, available_stock=%s, "
           "groupbuy_status_name=%s, half_notified=%s WHERE product_code=%s")
    pool.update_one(sql, (r["sold_count"], round(progress_pct, 2), r["available_stock"],
                          r["groupbuy_status_name"], half_notified, r["product_code"]))


def _mark_new_notified(pool, codes: list):
    """把一批商品的 new_notified 置 1（仅在新品提醒发送成功后调用）。

    Args:
        pool (MySQLConnectionPool): MySQL 连接池。
        codes (list[str]): 待置位的 product_code 列表；为空直接返回。
    """
    if not codes:
        return
    placeholders = ",".join(["%s"] * len(codes))
    pool.update_one(
        f"UPDATE {T_ALERT} SET new_notified=1 WHERE product_code IN ({placeholders})",
        tuple(codes))


def _is_sale_ended(data: dict, now_ts: int) -> tuple[bool, str]:
    """据商品详情 data 判断该车是否售卖结束（售罄 或 已过销售结束时间）。

    与 buy_record_spider.is_sale_ended 同逻辑，内联于此避免 import 采集脚本引入其模块级副作用。
    详情请求失败时 data 为空 dict，两条件均不命中返回未结束——保守，防抓取抖动误判下架。

    预售/未开卖的车：详情接口 availableStock 也返回 0（库存未分配）、soldCount=0、saleStartAt 在未来，
    仅凭 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。
        now_ts (int): 当前时间戳(秒)。

    Returns:
        tuple[bool, str]: (是否已结束, 原因文本)；未结束时原因为空串。
    """
    # ① 开卖时间未到 → 预售态，绝不算结束（预售车 availableStock=0 是「未分配」而非「卖光」）
    start_text = data.get("saleStartAt")
    if start_text:
        try:
            start_ts = time.mktime(time.strptime(start_text, "%Y-%m-%d %H:%M:%S"))
            if now_ts < start_ts:
                return False, ""
        except (ValueError, OverflowError):
            pass
    # ② 售罄：改用「已售 >= 总份数」判定，不再用 availableStock。
    #    直播实时放量卖的车，availableStock 是「当前放出、还没被抢的量」而非总剩余，卖到一半也会瞬时抖成 0，
    #    旧逻辑凭 availableStock<=0 会把仍在售的车误判售罄发假战报（2026/09/04 GB26090464871 卖 2/31 时 availableStock=0）。
    #    本函数只在「车已离开在售列表」后作二次确认，从严只认卖满，与下方「结束即全部售出」口径一致。
    sold = data.get("soldCount")
    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:
            end_ts = time.mktime(time.strptime(end_text, "%Y-%m-%d %H:%M:%S"))
            if now_ts >= end_ts:
                return True, f"已过结束时间(saleEndAt={end_text})"
        except (ValueError, OverflowError):
            pass
    return False, ""


def _fmt_ended(title: str, card, price_text: str, buyers: int, plain: bool = False) -> str:
    """构造一条「一车结束」战报正文：标题 + 价格 + 售出总件数 + 参与拆卡人数。

    车结束(拼团成交/售罄)即全部份数售出，故「售出件数」直接取总份数 card_count；
    参与拆卡人数为该车 deca_buy_record 按 user_id 去重的买家数。

    Args:
        title (str): 商品标题。
        card (int | None): 总份数 cardCount，即售出总件数。
        price_text (str): 已格式化的价格文案(_fmt_price 产出，区间价优先)。
        buyers (int): 去重购买人数(参与拆卡人数)。
        plain (bool, optional): True=纯文本(PC微信)/False=markdown(企微，标题加粗)。Defaults to False。

    Returns:
        str: 战报正文（三行）。
    """
    card_txt = card if card is not None else "?"
    title_line = title if plain else f"**{title}**"
    return (f"{title_line}\n"
            f"💰 {price_text} ｜ 🎯 售出 {card_txt} 件\n"
            f"👥 {buyers} 人参与拆卡")


def _send_ended(log, mname: str, item_text: str) -> bool:
    """按 SEND_CHANNEL 发送一条「一车结束」战报（每辆车一条独立消息）。

    Args:
        log: 日志对象。
        mname (str): 商家名称，用于消息标题。
        item_text (str): _fmt_ended 产出的战报正文。

    Returns:
        bool: 发送成功返回 True；失败返回 False（调用方据此决定是否置 ended_notified）。
    """
    header = f"🏁 得卡 · {mname} 一车结束"
    if SEND_CHANNEL == "pc":
        return bool(deca_wechat.send_text(header + "\n\n" + item_text, who=WX_TARGET))
    return bool(send_wechat_group_msg(log=log, items=[item_text], title=header))


def _mark_ended_notified(pool, code: str):
    """把一辆车的 ended_notified 置 1（仅在结束战报发送成功后调用）。

    Args:
        pool (MySQLConnectionPool): MySQL 连接池。
        code (str): 商品编码(product_code)。
    """
    pool.update_one(f"UPDATE {T_ALERT} SET ended_notified=1 WHERE product_code=%s", (code,))


def _count_buyers(pool, code: str) -> int:
    """查 deca_buy_record 中该车按 user_id 去重的购买人数(即参与拆卡人数)。

    数据由 buy_record_spider(独立进程)持续采集；若该车未被采到则返回 0。

    Args:
        pool (MySQLConnectionPool): MySQL 连接池。
        code (str): 商品编码(product_code)。

    Returns:
        int: 去重购买人数；无记录或查询异常返回 0。
    """
    rows = pool.select_all(
        "SELECT COUNT(DISTINCT user_id) FROM deca_buy_record WHERE product_code=%s", code)
    if rows and rows[0] and rows[0][0] is not None:
        return int(rows[0][0])
    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。

    候选条件三取交：本进程见过在售(_seen_onsale) ∩ 本轮已不在在售(onsale_codes) ∩ 未播报过结束
    (ended_notified=0)。命中的候选再打一次详情，仅 _is_sale_ended 确认售罄/过结束时间才播报——
    防「全站抓取抖动导致某车临时缺席」被误判为结束。发送成功才置位；失败保持 0，下轮重发不丢。

    Args:
        log: 日志对象。
        pool (MySQLConnectionPool): MySQL 连接池。
        existing (dict): 库内该商家记录 {code: {ended/title/mname/unit_price/card/sold/...}}。
        onsale_codes (set): 本轮在售的 product_code 集合。
    """
    now_ts = int(time.time())
    for code, info in existing.items():
        if info["ended"] == 1:
            continue                      # 已播报过结束，跳过
        if code not in _seen_onsale:
            continue                      # 本进程运行后没见它在售 → 属「之前的车」，不补发
        if code in onsale_codes:
            continue                      # 本轮仍在售，未结束
        # 疑似消失：二次打详情确认（失败/仍在售都不判结束，下轮再看，防抓取抖动误报）
        data = _get_detail_data(log, code)
        ended, reason = _is_sale_ended(data, now_ts)
        if not ended:
            log.info(f"[疑似消失未确认结束] {code} 详情仍在售或取详情失败，跳过(可能抓取抖动) | {info['title']}")
            continue
        # 【2026/08/31】soldCount==0 的车不发结束战报：这类是「上架后无人问津、被商家撤下或到 saleEndAt
        # 到期」的流车，从没成交（既不在在售 is_on_sale=0、也不在已售成交列表），购买记录必空。旧逻辑
        # 「结束即全部售出」会把它虚报成「售出<总份数>件 / 0 人参与拆卡」，属误发（主公反馈老是误发）。
        # 故 soldCount==0 时静默不发，并置位 ended_notified 停止下轮重复对账/重打详情。
        # 只在详情明确返回 soldCount==0 时静默；soldCount 缺失(None)不静默，保守走正常播报，防抓取抖动误吞真售罄车。
        sold_now = data.get("soldCount")
        if sold_now == 0:
            _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:
            card = info["card"]
        # 价格：区间价(minUnitPrice~maxUnitPrice)优先，详情取不到再回退库内 unit_price
        price_text = _fmt_price(data.get("unitPrice"), data.get("minUnitPrice"),
                                data.get("maxUnitPrice"))
        if price_text == "¥-" and info["unit_price"] is not None:
            price_text = _fmt_price(info["unit_price"], None, None)
        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']}")
        else:
            log.warning(f"[结束战报发送失败] {code} 保持 ended_notified=0，下轮重发 | {info['title']}")


def run_once(log, pool):
    """跑一轮监控：拉在售 → 结束对账(发一车结束战报) → 比对库内状态 → 发新品/过半提醒 → 落库。

    Args:
        log: 日志对象。
        pool (MySQLConnectionPool): MySQL 连接池。
    """
    products, ok = fetch_onsale(log)
    if not ok:
        log.warning("全站在售抓取未取全(请求异常)，本轮跳过：不做新品/过半/结束判断，避免误判车下架")
        return

    # 库内该商家已监控商品：含结束标记与结束战报所需的静态字段(标题/商家名/单价/份数)
    existing_rows = pool.select_all(
        "SELECT product_code, new_notified, half_notified, ended_notified, "
        "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, mid_, mname_, uprice, ccount, scount in existing_rows:
        existing[code] = {"new": new, "half": half, "ended": ended,
                          "title": title, "mid": mid_, "mname": mname_, "unit_price": uprice,
                          "card": ccount, "sold": scount}
    is_cold = len(existing) == 0  # 冷启动：库内该商家零记录（仅作日志提示，逻辑与常规轮一致）
    if is_cold:
        log.info(f"首次运行：库内该商家零记录，只提醒本轮窗口起点({RUN_START:%H:%M})后新上架的商品，老货静默建档")

    onsale_codes = {r["product_code"] for r in products}  # 本轮在售 code 集合
    # 结束对账：只播报本进程运行后见过在售、之后确认结束的车（历史已结束车不补发；置位防重发）
    _detect_and_report_ended(log, pool, existing, onsale_codes)
    _seen_onsale.update(onsale_codes)  # 本轮在售并入「见过在售」集合，供下轮结束对账（实现「只管运行后」）

    if not products:
        log.info("本商家本轮无在售商品，已完成结束对账，跳过新品/过半")
        return

    plain = SEND_CHANNEL == "pc"  # PC 微信用纯文本，企微用 markdown
    new_items = []     # 新品上架提醒文案
    half_items = []    # 进度过半提醒文案
    pending_new = []   # 待发新品的 product_code：仅在提醒发送成功后才置 new_notified=1

    window_start = _window_start(datetime.now())  # 「新上架」时间门槛：本轮窗口起点(最近的 RUN_START，默认 20:30)

    for r in products:
        code = r["product_code"]
        ratio = _progress(r["sold_count"], r["card_count"])
        pct = ratio * 100
        over_half = ratio >= HALF_THRESHOLD

        if code not in existing:
            # 库里没有的候选：打一次详情拿上架时间，只有 publishAt 晚于窗口起点(RUN_START，默认 20:30)才算「本轮新上架」
            publish_at = get_publish_at(log, code)
            if _is_new_arrival(publish_at, window_start):
                # 真·本轮新上架：new_notified 先记 0，发送成功再置 1（发失败下轮自动重发，不丢）
                _insert_alert(pool, r, pct, new_notified=0,
                              half_notified=1 if over_half else 0)
                new_items.append(_fmt_item(r, "new", plain=plain))
                pending_new.append(code)
                if over_half:  # 新品上架即已过半，一并提示（half 逻辑保持原状）
                    half_items.append(_fmt_item(r, "half", pct=pct, plain=plain))
            else:
                # 上架早于窗口起点(或拿不到上架时间)的老货：静默建档、不提醒（new_notified 直接置 1，避免下轮反复判断）
                _insert_alert(pool, r, pct, new_notified=1,
                              half_notified=1 if over_half else 0)
                log.info(f"[静默建档] {code} 上架 {publish_at or '未知'} 早于窗口起点 {window_start:%m-%d %H:%M}，不提醒 | {r['title']}")
        else:
            # 已在表：new_notified=0 视为「待发新品」(上次发失败残留 或 人工改回 0)，重新纳入新品提醒
            if existing[code]["new"] == 0:
                new_items.append(_fmt_item(r, "new", plain=plain))
                pending_new.append(code)
            # 过半逻辑保持原状（half_notified 策略不改动）
            already_half = existing[code]["half"] == 1
            if over_half and not already_half:
                _update_progress(pool, r, pct, half_notified=1)
                half_items.append(_fmt_item(r, "half", pct=pct, plain=plain))
            else:
                _update_progress(pool, r, pct, half_notified=1 if already_half else 0)

    mname = products[0].get("merchant_name") or MERCHANT_ID

    # 常规轮次推送；sent_ok 记新品提醒是否发送成功，决定要不要置位 new_notified
    sent_ok = False
    if SEND_CHANNEL == "pc":
        # PC 微信：新品 + 过半合并成一条消息发出（减少操作，一次发完）
        if new_items or half_items:
            log.info(f"合并推送(pc)：新品{len(new_items)}个、过半{len(half_items)}个")
            sent_ok = _dispatch_pc_combined(log, mname, new_items, half_items)
    else:
        # 企微渠道：仍按两类各发一条 markdown（新品这条成功与否决定 sent_ok）
        if new_items:
            log.info(f"新商品上架 {len(new_items)} 个，推送(qywx)")
            sent_ok = _dispatch(log, new_items, f"🆕 得卡[{mname}] 新商品上架 {len(new_items)} 个")
        if half_items:
            log.info(f"进度过半 {len(half_items)} 个，推送(qywx)")
            _dispatch(log, half_items, f"🔥 得卡[{mname}] 拼团进度过半 {len(half_items)} 个")

    # 新品提醒发送成功后才置位 new_notified=1；失败则保持 0，下一轮继续重发（不丢）
    if pending_new:
        if sent_ok:
            _mark_new_notified(pool, pending_new)
            log.info(f"新品提醒发送成功，new_notified 置 1：{len(pending_new)} 个")
        else:
            log.warning(f"新品提醒发送失败，new_notified 保持 0，下轮重发：{len(pending_new)} 个")

    if not new_items and not half_items:
        log.info("本轮无新品、无新达标过半商品")


@retry(stop=stop_after_attempt(100), wait=wait_fixed(600), after=core.after_log)
def main_task(log):
    """在售监控主函数：建连接池 → 跑一轮监控(挂了每 10 分钟重试)。

    Args:
        log: 日志对象。

    Raises:
        RuntimeError: 数据库连接池异常时抛出以触发重试。
    """
    log.info(f"开始运行 {sys._getframe().f_code.co_name} 在售提醒监控" + "." * 40)
    pool = MySQLConnectionPool(log=log)
    if not pool.check_pool_health():
        log.error("数据库连接池异常")
        raise RuntimeError("数据库连接池异常")
    try:
        run_once(log, pool)
    except Exception as e:
        log.error(f"{sys._getframe().f_code.co_name} error: {e}")
    finally:
        log.info(f"在售提醒监控 {sys._getframe().f_code.co_name} 运行结束，等待下一轮" + "." * 20)


def _in_run_window(now: datetime) -> bool:
    """判断当前时刻是否在运行窗口 [RUN_START, 次日 RUN_END) 内。

    Args:
        now (datetime): 当前时间。

    Returns:
        bool: 在窗口内返回 True。窗口跨午夜，故「晚于开始 或 早于结束」即算命中。
    """
    t = now.time()
    return t >= RUN_START or t < RUN_END


def _seconds_to_window(now: datetime) -> int:
    """计算从 now 到下一次窗口开始(当天 RUN_START)的休眠秒数（仅窗口外调用）。

    Args:
        now (datetime): 当前时间。

    Returns:
        int: 需休眠的秒数；若当天 RUN_START 已过则顺延到次日。
    """
    start = now.replace(hour=RUN_START.hour, minute=RUN_START.minute, second=0, microsecond=0)
    if start <= now:                # 当天 RUN_START 已过 → 顺延到次日同一时刻
        start += timedelta(days=1)
    return int((start - now).total_seconds())


def schedule_task():
    """常驻循环：仅在每天 RUN_START~次日 RUN_END 运行；窗口内每轮随机间隔轮询，窗口外休眠到下次开窗。"""
    while True:
        now = datetime.now()
        if not _in_run_window(now):
            wait = _seconds_to_window(now)
            logger.info(f"当前不在运行窗口({RUN_START:%H:%M}~次日{RUN_END:%H:%M})，休眠 {wait}s 到 {RUN_START:%H:%M} 再跑")
            time.sleep(wait)
            continue
        main_task(log=logger)
        wait = random.randint(MIN_INTERVAL_SEC, MAX_INTERVAL_SEC)  # 每轮随机间隔(秒)
        logger.info(f"下一轮 {wait}s 后运行")
        time.sleep(wait)


def _parse_start_time(text: str) -> dtime:
    """把命令行传入的开始时间文本解析为 datetime.time。

    Args:
        text (str): 开始时间文本，格式 "HH:MM" 或 "HH:MM:SS"，如 "20:30" / "17:00"。

    Returns:
        dtime: 解析出的 time 对象。

    Raises:
        argparse.ArgumentTypeError: 格式非法(非 HH:MM[:SS] 或时分秒越界)时抛出，供 argparse 提示用户。
    """
    text = text.strip()
    for fmt in ("%H:%M:%S", "%H:%M"):
        try:
            return datetime.strptime(text, fmt).time()
        except ValueError:
            continue
    raise argparse.ArgumentTypeError(f"开始时间格式非法：{text!r}，应为 HH:MM 或 HH:MM:SS，如 20:30")


def _parse_args() -> argparse.Namespace:
    """解析命令行参数，取运行窗口开始时间(默认 20:30)。

    支持位置参数与 --start 两种写法，二者等价，方便直接 `python xxx.py 17:00`。

    Returns:
        argparse.Namespace: 含 start(datetime.time) 属性；未传时为默认 RUN_START。
    """
    parser = argparse.ArgumentParser(
        description="得卡 DECA 在售提醒：可指定运行窗口开始时间(该时间后的新上架才提醒)")
    parser.add_argument(
        "start", nargs="?", type=_parse_start_time, default=None,
        help="运行窗口开始时间 HH:MM[:SS]，默认 20:30；位置参数写法，如 17:00")
    parser.add_argument(
        "--start", dest="start_opt", type=_parse_start_time, default=None,
        help="运行窗口开始时间 HH:MM[:SS]，与位置参数等价，如 --start 17:00")
    return parser.parse_args()


if __name__ == "__main__":
    # logger.add(sys.stderr, level="INFO")  # 控制台同步输出，便于观察
    _args = _parse_args()
    # 位置参数优先，其次 --start，都未传则保持默认 RUN_START(20:30)
    _start = _args.start or _args.start_opt
    if _start is not None:
        RUN_START = _start  # 覆盖模块级默认，窗口判定与「新上架」门槛均随之改变
        logger.info(f"运行窗口开始时间由命令行指定为 {RUN_START:%H:%M}")
    schedule_task()