fix(comparison): 派生 best 排除缺菜店, 避免虚低价当"最低价"

pricebot comparison_results[].rank 是纯价格排序(含缺菜店), server 派生 best
若照单全收, 会把缺菜(漏菜)店的虚低总价当 best_price → 记录页戴"最低"红框 +
算出虚假省额。

- _derive / _derive_from_results 派生 best 时按 platform_results[pid].
  skipped_dish_count 排除缺菜店; 源平台永远全菜, 全目标缺菜时回落到源
  (is_source_best、saved=0), 不虚报省额。
- platform_results 内层结构宽松(老客户端透传可伪造), _is_short 用
  isinstance 兜底, 值非 dict 时按"不缺菜"处理, 不打 500。
- harvest_done 传入 done_params.platform_results; 不传→纯 rank/price 老行为不变。
- 新增纯函数测试: 排除缺菜 / 全缺菜回落源 / 不传保持老行为 / 内层非 dict 不崩,
  覆盖 _derive 与 _derive_from_results 两条路径。

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

alembic/versions/coupon_claim_event.py

@@ -69,7 +69,11 @@ def upgrade() -> None:
         unique=False,
     )
 
-    # 旧表只能回填当前仍保留的 trace；历史上已被每日去重覆盖的关联无法恢复。
+    # 旧表按 (device, coupon, 自然日) 去重，trace_id 可空且不在唯一键里：同一
+    # (trace_id, coupon_id) 可能散落在多行（如一次会话的 /step 帧跨零点，把同一张券
+    # 写进相邻两天）。新表按 (trace_id, coupon_id) 唯一，整表 1:1 复制会撞
+    # uq_coupon_claim_event_trace_coupon。回填时按 (trace_id, coupon_id) 只取 id 最大
+    # （最近写入）的一行。历史上已被每日去重覆盖的关联仍无法恢复。
     op.execute(
         """
         INSERT INTO coupon_claim_event (
@@ -81,6 +85,12 @@ def upgrade() -> None:
             vendor, coupon_name, claimed_count, reason, extra, created_at, updated_at
         FROM coupon_claim_record
         WHERE trace_id IS NOT NULL
+          AND id IN (
+              SELECT MAX(id)
+              FROM coupon_claim_record
+              WHERE trace_id IS NOT NULL
+              GROUP BY trace_id, coupon_id
+          )
         """
     )
 

app/admin/routers/withdraw.py

@@ -129,13 +129,11 @@ def withdraw_health_check(db: AdminDb) -> WxpayHealthCheckOut:
         issues.append("免确认授权回调地址未配置")
 
     # 实际是否自动对账 = env 部署总闸(worker 起没起)AND 运营后台 DB 开关(本轮跑不跑)。
+    # 自动查单属于非阻断运维能力:状态继续返回给调用方,但关闭时不计入微信提现配置 issues,
+    # 避免把“没有自动扫单”误报成“无法打款”。
     worker_running = settings.WITHDRAW_AUTO_RECONCILE_ENABLED
     daily_on = bool(app_config.get_value(db, "withdraw_auto_reconcile_enabled"))
     auto_reconcile_enabled = worker_running and daily_on
-    if not worker_running:
-        issues.append("自动对账 worker 未启动(部署侧 env WITHDRAW_AUTO_RECONCILE_ENABLED=false)")
-    elif not daily_on:
-        issues.append("自动对账运营开关已关闭(系统配置页可开)")
 
     return WxpayHealthCheckOut(
         ok=not issues,

app/repositories/comparison.py

@@ -55,13 +55,22 @@ def _product_names_from_items(items: list | None) -> str | None:
 def _derive(payload: ComparisonRecordIn) -> dict:
     """从上报 payload 派生结构化列(best/saved/is_source_best/status)。"""
     results = payload.comparison_results
+    _pr = payload.platform_results or {}
 
-    # 最优 = rank 最小的一条;协议已升序,但不信顺序,显式按 rank/price 兜底取最小价。
-    best = None
+    def _is_short(r) -> bool:
+        # 缺菜(漏菜)店: 少买了菜总价虚低, 不参与最优评选。逐平台 skipped 在 platform_results, 行里没有。
+        # platform_results 内层结构宽松(pricebot/老客户端透传, 可伪造), 值非 dict 时按"不缺菜"处理, 不崩。
+        info = _pr.get(r.platform_id) if r.platform_id else None
+        return isinstance(info, dict) and (info.get("skipped_dish_count") or 0) > 0
+
+    # 最优 = 非缺菜里 rank 最小(=最便宜)的一条;协议已升序,但不信顺序,显式按 rank/price 取。
+    # 源平台永远全菜, 故全目标缺菜时回落到源(is_source_best、saved=0), 不把虚低价当最低。
     priced = [r for r in results if r.price is not None]
-    if priced:
+    clean = [r for r in priced if not _is_short(r)]
+    best = None
+    if clean:
         best = min(
-            priced,
+            clean,
             key=lambda r: (r.rank if r.rank is not None else 10**9, r.price),
         )
 
@@ -196,14 +205,29 @@ def upsert_record(
 # ============================================================
 
 
-def _derive_from_results(results: list[dict]) -> dict:
+def _derive_from_results(
+    results: list[dict], platform_results: dict | None = None
+) -> dict:
     """从 done 帧 comparison_results(pricebot 原始 dict 列表)派生结构化列。
-    等价 _derive,但吃原始字段(is_source/price/rank/platform_id/store_name...)而非 pydantic 对象。"""
+    等价 _derive,但吃原始字段(is_source/price/rank/platform_id/store_name...)而非 pydantic 对象。
+
+    platform_results(done.params.platform_results): 逐平台 skipped_dish_count 在这里(行里没有)。
+    传入则派生 best 时排除缺菜(漏菜)店 —— 少买了菜总价虚低, 不能当记录级"最低价"/算虚假省额;
+    源平台永远全菜, 故全目标缺菜时 best 回落到源(is_source_best、不虚报省)。不传→纯 rank/price, 行为不变。"""
+    _pr = platform_results or {}
+
+    def _is_short(r: dict) -> bool:
+        # platform_results 内层结构宽松(pricebot/客户端透传), 值非 dict 时按"不缺菜"处理, 不崩。
+        pid = r.get("platform_id")
+        info = _pr.get(pid) if pid else None
+        return isinstance(info, dict) and (info.get("skipped_dish_count") or 0) > 0
+
     priced = [r for r in results if r.get("price") is not None]
+    clean = [r for r in priced if not _is_short(r)]  # 缺菜店排除出最优评选
     best = None
-    if priced:
+    if clean:
         best = min(
-            priced,
+            clean,
             key=lambda r: (r.get("rank") if r.get("rank") is not None else 10**9, r["price"]),
         )
     src_row = next((r for r in results if r.get("is_source")), None)
@@ -389,7 +413,7 @@ def harvest_done(
     返回 (记录, 是否本次**新**落成 success)——供调用方据此幂等发一次邀请奖。
     行不存在(理论上帧0已建;防御)则新建。"""
     results = done_params.get("comparison_results") or []
-    derived = _derive_from_results(results)
+    derived = _derive_from_results(results, done_params.get("platform_results"))
     # 菜品:pricebot 已把源单菜品塞进 comparison_results[源行].items
     items = next((r.get("items") or [] for r in results if r.get("is_source")), [])
     fields = dict(

tests/test_derive_best_excludes_short.py

@@ -0,0 +1,102 @@
+"""_derive_from_results / _derive: 缺菜(漏菜)店总价虚低, 不当记录级"最低价"。
+
+回归: pricebot comparison_results[].rank 是纯价格排序(含缺菜), server 派生 best 若照单全收,
+会把缺菜店的虚低价当 best_price → 记录页戴"最低"红框 + 虚假省额。
+修复: 派生 best 时按 platform_results[pid].skipped_dish_count 排除缺菜店(源平台永远全菜, 仍可当 best)。
+纯函数, 不碰 DB。
+"""
+from app.repositories.comparison import _derive, _derive_from_results
+from app.schemas.compare_record import ComparisonRecordIn, ComparisonResultIn
+
+
+def test_derive_from_results_excludes_short_ordered_from_best():
+    # jd 缺 2 道菜 → 虚低 ¥25(rank=1); tb 全有 ¥38.5; 源美团 ¥42。best 应是 tb(干净最便宜), 不是 jd。
+    results = [
+        {"platform_id": "meituan", "platform_name": "美团", "price": 42.0, "is_source": True, "rank": 3},
+        {"platform_id": "jd_waimai", "platform_name": "京东外卖", "price": 25.0, "is_source": False, "rank": 1},
+        {"platform_id": "taobao_flash", "platform_name": "淘宝闪购", "price": 38.5, "is_source": False, "rank": 2},
+    ]
+    platform_results = {
+        "jd_waimai": {"skipped_dish_count": 2},
+        "taobao_flash": {"skipped_dish_count": 0},
+        "meituan": {"is_source": True},
+    }
+    d = _derive_from_results(results, platform_results)
+    assert d["best_platform_id"] == "taobao_flash"
+    assert d["best_price_cents"] == 3850
+    assert d["saved_amount_cents"] == 4200 - 3850   # 350, 用干净店算省额, 不是缺菜虚低价 42-25=17元
+    assert d["is_source_best"] is False
+
+
+def test_derive_from_results_all_targets_short_falls_back_to_source():
+    # 唯一比源便宜的都是缺菜 → 不crown缺菜店; 源全菜 → best=源, is_source_best, 不虚报省额。
+    results = [
+        {"platform_id": "meituan", "price": 42.0, "is_source": True, "rank": 2},
+        {"platform_id": "jd_waimai", "price": 25.0, "is_source": False, "rank": 1},
+    ]
+    platform_results = {"jd_waimai": {"skipped_dish_count": 3}}
+    d = _derive_from_results(results, platform_results)
+    assert d["best_platform_id"] == "meituan"
+    assert d["is_source_best"] is True
+    assert d["saved_amount_cents"] == 0
+
+
+def test_derive_from_results_no_platform_results_keeps_old_behavior():
+    # 不传 platform_results(老 harvest / 无缺菜信息)→ 行为不变: 纯 rank/price 选 best。
+    results = [
+        {"platform_id": "meituan", "price": 42.0, "is_source": True, "rank": 2},
+        {"platform_id": "jd_waimai", "price": 25.0, "is_source": False, "rank": 1},
+    ]
+    d = _derive_from_results(results)
+    assert d["best_platform_id"] == "jd_waimai"
+    assert d["best_price_cents"] == 2500
+
+
+def test_derive_from_results_malformed_platform_results_no_crash():
+    # 内层值非 dict(异常/伪造上报)→ 不抛 AttributeError, 按"不缺菜"处理, 照常选最便宜。
+    results = [
+        {"platform_id": "meituan", "price": 42.0, "is_source": True, "rank": 2},
+        {"platform_id": "jd_waimai", "price": 25.0, "is_source": False, "rank": 1},
+    ]
+    d = _derive_from_results(results, {"jd_waimai": "oops"})
+    assert d["best_platform_id"] == "jd_waimai"
+    assert d["best_price_cents"] == 2500
+
+
+def test_derive_pydantic_excludes_short():
+    # _derive(老客户端 POST 路径)同样排除缺菜店: jd 缺菜虚低 ¥25 不当 best, 取干净的淘宝 ¥38.5。
+    payload = ComparisonRecordIn(
+        trace_id="t-short-pyd",
+        source_price=42.0,
+        source_platform_id="meituan",
+        comparison_results=[
+            ComparisonResultIn(platform_id="meituan", platform_name="美团", price=42.0, is_source=True, rank=3),
+            ComparisonResultIn(platform_id="jd_waimai", platform_name="京东外卖", price=25.0, is_source=False, rank=1),
+            ComparisonResultIn(platform_id="taobao_flash", platform_name="淘宝闪购", price=38.5, is_source=False, rank=2),
+        ],
+        platform_results={
+            "jd_waimai": {"skipped_dish_count": 2},
+            "taobao_flash": {"skipped_dish_count": 0},
+        },
+    )
+    d = _derive(payload)
+    assert d["best_platform_id"] == "taobao_flash"
+    assert d["best_price_cents"] == 3850
+    assert d["saved_amount_cents"] == 4200 - 3850
+    assert d["is_source_best"] is False
+
+
+def test_derive_pydantic_malformed_platform_results_no_crash():
+    # _derive 的 platform_results 来自老客户端透传(可伪造): 内层非 dict 不应打 500。
+    payload = ComparisonRecordIn(
+        trace_id="t-malformed-pyd",
+        source_price=42.0,
+        comparison_results=[
+            ComparisonResultIn(platform_id="meituan", price=42.0, is_source=True, rank=2),
+            ComparisonResultIn(platform_id="jd_waimai", price=25.0, is_source=False, rank=1),
+        ],
+        platform_results={"jd_waimai": "oops"},
+    )
+    d = _derive(payload)  # 不抛 AttributeError
+    assert d["best_platform_id"] == "jd_waimai"
+    assert d["best_price_cents"] == 2500