修复：收敛领券计划数迁移链

记录每轮计划券数，将跳过券计入逐场分母，并在中途退出或无逐券结果时返回稳定的成功率口径。

alembic/versions/comparison_record_fail_reason.py

@@ -0,0 +1,49 @@
+"""comparison_record.fail_reason (失败卡展示原因)
+
+Revision ID: comparison_record_fail_reason
+Revises: user_manual_risk_fields
+Create Date: 2026-07-28 12:00:00.000000
+
+失败记录的展示原因:information 具体则=它;笼统则由写路径从 platform_results 捞出的
+业务原因;纯系统失败为 None(端侧品牌兜底)。见 repositories.comparison._derive_fail_display。
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = 'comparison_record_fail_reason'
+down_revision: Union[str, Sequence[str], None] = 'user_manual_risk_fields'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    with op.batch_alter_table('comparison_record', schema=None) as batch_op:
+        batch_op.add_column(sa.Column('fail_reason', sa.String(length=256), nullable=True))
+    # 回填老失败记录:information 具体的直接搬过来(笼统/系统失败留 None → 端侧品牌兜底)。
+    # 新记录由写路径 _derive_fail_display 落库(含 platform_results 救援/补判),不走这条。
+    # platform_results 只在 raw_payload 里,SQL 里不易解析,故老记录不做救援/补判(可接受:
+    # 老 mixed/打烊记录回退品牌兜底);具体 information 的老记录本次即可显示真实原因。
+    op.execute(
+        """
+        UPDATE comparison_record
+        SET fail_reason = information
+        WHERE status = 'failed'
+          AND information IS NOT NULL
+          AND information <> ''
+          AND information NOT IN (
+              '比价过程出错，请稍后重试',
+              '比价出错',
+              '比价未完成',
+              'done 参数缺少可验证的目标平台结果'
+          )
+        """
+    )
+
+
+def downgrade() -> None:
+    with op.batch_alter_table('comparison_record', schema=None) as batch_op:
+        batch_op.drop_column('fail_reason')

alembic/versions/coupon_session_planned_count.py

@@ -0,0 +1,28 @@
+"""add planned coupon count to coupon session
+
+Revision ID: coupon_session_planned_count
+Revises: comparison_record_fail_reason
+"""
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+
+from alembic import op
+
+revision: str = "coupon_session_planned_count"
+down_revision: str | None = "comparison_record_fail_reason"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "coupon_session",
+        sa.Column("planned_coupon_count", sa.Integer(), nullable=True),
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("coupon_session", "planned_coupon_count")

app/admin/repositories/coupon_data.py

@@ -23,6 +23,7 @@ from app.repositories.coupon_state import DEFAULT_PLATFORMS, coupon_id_to_platfo
 
 _SLOT_OK = ("success", "already_claimed")
 _SLOT_TRIED = ("success", "already_claimed", "failed")
+_SESSION_POINT_TOTAL = (*_SLOT_TRIED, "skipped")
 
 
 def _cn_hour(dt: datetime) -> int:
@@ -166,6 +167,10 @@ def _session_to_row(
     point_stats: dict | None = None,
 ) -> dict:
     """CouponSession ORM → 明细行 dict(主表「领券数据」与「用户全部领券」抽屉共用)。"""
+    succeeded = point_stats["succeeded"] if point_stats else 0
+    event_total = point_stats["tried"] if point_stats else 0
+    planned_total = r.planned_coupon_count or 0
+    point_total = max(event_total, planned_total)
     return {
         "id": r.id,
         "trace_id": r.trace_id,
@@ -182,8 +187,8 @@ def _session_to_row(
         "app_env": r.app_env,
         "started_at": r.started_at,
         "claimed_count": r.claimed_count,
-        "point_success_count": point_stats["succeeded"] if point_stats else None,
-        "point_total_count": point_stats["tried"] if point_stats else None,
+        "point_success_count": succeeded if point_total > 0 else None,
+        "point_total_count": point_total if point_total > 0 else None,
         "trace_url": r.trace_url,
         "ad_revenue_yuan": ad_revenue_yuan,
     }
@@ -202,7 +207,7 @@ def _point_scores_by_trace(db: Session, trace_ids: list[str]) -> dict[str, dict[
         )
         .where(
             CouponClaimEvent.trace_id.in_(trace_ids),
-            CouponClaimEvent.status.in_(_SLOT_TRIED),
+            CouponClaimEvent.status.in_(_SESSION_POINT_TOTAL),
         )
         .group_by(CouponClaimEvent.trace_id)
     ).all()

app/api/v1/coupon.py

@@ -93,6 +93,13 @@ def _record_claims_blocking(
             )
 
 
+def _merge_planned_count_blocking(
+    trace_id: str | None, planned_count: int | None
+) -> None:
+    with SessionLocal() as db:
+        coupon_repo.merge_session_planned_count(db, trace_id, planned_count)
+
+
 def _mark_completed_blocking(
     device_id: str, user_id: int | None, trace_id: str | None
 ) -> None:
@@ -176,6 +183,18 @@ async def coupon_step(
 
     resp_json = resp.json()
 
+    # pricebot 每帧 status.progress.total 都带本轮计划券数。独立回写 session 后，
+    # 即使用户在第一张出结果前退出，admin 也能显示 0/N，而不是空值。
+    progress = (resp_json.get("status") or {}).get("progress") or {}
+    planned_count = progress.get("total")
+    if trace_id and isinstance(planned_count, int) and planned_count > 0:
+        try:
+            await run_in_threadpool(
+                _merge_planned_count_blocking, trace_id, planned_count
+            )
+        except Exception as e:  # noqa: BLE001
+            logger.warning("coupon planned count write failed: %s", e)
+
     # 领券结果沉淀:每日资产 + 逐次事件；中间帧和 done 全量帧均幂等写库。
     # 当前只记录、不参与"要不要领"判断(MVP 先不去重)。写库失败不影响返回。
     if device_id:

app/models/comparison.py

@@ -102,6 +102,11 @@ class ComparisonRecord(Base):
     # done 帧 information 文案。成功:"在美团找到同店,到手价 ¥X…";
     # 失败:具体原因(如"美团、京东外卖均未找到该商品")。前端在比价失败时当原因展示。
     information: Mapped[str | None] = mapped_column(String(256), nullable=True)
+    # 失败卡「原因」行的展示文案(仅 status=failed 时非空):information 具体则=它;笼统则从
+    # platform_results 捞出的业务原因(打烊/未起送/找不到店或菜/单点不配送);纯系统失败为 None
+    # → 端侧显示品牌兜底「网络开小差…」。写路径(harvest_done / upsert_record)落库时派生。
+    # 见 repositories.comparison._derive_fail_display。
+    fail_reason: Mapped[str | None] = mapped_column(String(256), nullable=True)
 
     # ===== 明细(JSON,越详细越好)=====
     # 下单菜品 [{name, qty, specs?}]

app/models/coupon_state.py

@@ -277,6 +277,9 @@ class CouponSession(Base):
     platform_elapsed: Mapped[dict | None] = mapped_column(_JSON, nullable=True)
     # 领到总张数(收尾帧带)。
     claimed_count: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    # pricebot 本轮计划处理的券数。由 /coupon/step 每帧 status.progress.total 回写；
+    # 即使用户中途退出、尚无任何单券终态，admin 也能以计划数作为成功率分母。
+    planned_coupon_count: Mapped[int | None] = mapped_column(Integer, nullable=True)
     # 本次 session 至少领到一张(status∈{success,already_claimed})的平台 id 列表,如 ["meituan-waimai","jd-waimai"]。
     # admin「领券数据」据此算整单成功率(②)/点位成功率(③);服务端 /step 逐帧按 trace_id 并集写入
     # (见 coupon_state.merge_session_platform_success)。旧行=NULL → 视作空集。

app/repositories/comparison.py

@@ -52,6 +52,81 @@ def _product_names_from_items(items: list | None) -> str | None:
     return joined[:500] or None
 
 
+# ---- 失败记录的展示文案(记录页失败卡「原因」行)------------------------------
+# information 具体就直出;笼统(_GENERIC_INFO)则从 platform_results 捞一条用户可读的业务
+# 原因;捞不到 → None(端侧显示品牌兜底「网络开小差…」)。pricebot 把 store_closed /
+# no_delivery 漏成了 status=failed,这里按 reason 关键字补判;打烊类 reason 常带一坨脏店名
+# (店名+月售+起送+配送…),统一成简短模板。自动化黑话(搜索失败/读价失败/购物车残留/裸
+# FAILED…)不给用户看 → 归入品牌兜底。
+
+# pricebot 组不出具体原因时的笼统 information(线上统计的大头),一律走品牌兜底。
+_GENERIC_INFO = {
+    "比价过程出错，请稍后重试",
+    "比价出错",
+    "比价未完成",
+    "done 参数缺少可验证的目标平台结果",
+}
+# 干净业务结局 status(直接可信),按展示优先级(越靠前越先选)。
+_BIZ_STATUS_PRIORITY = (
+    "below_minimum",
+    "no_delivery",
+    "store_closed",
+    "items_not_found",
+    "store_not_found",
+)
+
+
+def _store_closed_text(reason: str | None) -> str:
+    """打烊/暂停营业/休息类 reason 常带脏店名元数据 → 只留结论,套简短模板。"""
+    r = reason or ""
+    if "暂停营业" in r:
+        state = "暂停营业"
+    elif "休息" in r:
+        state = "休息中"
+    else:
+        state = "已打烊"
+    return f"门店{state}，无法比价"
+
+
+def _target_display_reason(platform_results: dict | None) -> str | None:
+    """从逐平台结果里挑一条"可展示给用户"的失败原因;挑不到返回 None。
+    ① status 命中干净业务结局集 → 直接采信(打烊套模板,其余用 reason);
+    ② 补判 pricebot 漏成 status=failed 的两类:打烊(套模板)、单点不配送(reason 本身干净);
+    自动化黑话(搜索失败/读价失败/购物车残留/裸 FAILED…)一律不展示 → None。"""
+    pr = platform_results or {}
+    targets = [
+        v for v in pr.values() if isinstance(v, dict) and not v.get("is_source")
+    ]
+    for want in _BIZ_STATUS_PRIORITY:  # ① 干净 status 优先
+        for v in targets:
+            if v.get("status") == want:
+                if want == "store_closed":
+                    return _store_closed_text(v.get("reason"))
+                if v.get("reason"):
+                    return v["reason"]
+    for v in targets:  # ② 漏成 failed 的业务结局补判
+        if v.get("status") != "failed":
+            continue
+        reason = (v.get("reason") or "").strip()
+        if any(k in reason for k in ("打烊", "暂停营业", "休息")):
+            return _store_closed_text(reason)
+        if "单点不配送" in reason:
+            return reason
+    return None
+
+
+def _derive_fail_display(
+    information: str | None, platform_results: dict | None
+) -> str | None:
+    """失败记录展示文案:information 具体则直出;笼统则从 platform_results 捞/补判;
+    都拿不到 → None(端侧品牌兜底)。仅在 status=failed 时调用。"""
+    info = (information or "").strip()
+    text = info if (info and info not in _GENERIC_INFO) else _target_display_reason(
+        platform_results
+    )
+    return text[:256] if text else None
+
+
 def _derive(payload: ComparisonRecordIn) -> dict:
     """从上报 payload 派生结构化列(best/saved/is_source_best/status)。"""
     results = payload.comparison_results
@@ -105,6 +180,11 @@ def _derive(payload: ComparisonRecordIn) -> dict:
         "saved_amount_cents": saved_amount_cents,
         "is_source_best": is_source_best,
         "status": status,
+        "fail_reason": (
+            _derive_fail_display(payload.information, _pr)
+            if status == "failed"
+            else None
+        ),
     }
 
 
@@ -414,11 +494,19 @@ def harvest_done(
     行不存在(理论上帧0已建;防御)则新建。"""
     results = done_params.get("comparison_results") or []
     derived = _derive_from_results(results, done_params.get("platform_results"))
+    fail_reason = (
+        _derive_fail_display(
+            done_params.get("information"), done_params.get("platform_results")
+        )
+        if derived["status"] == "failed"
+        else None
+    )
     # 菜品:pricebot 已把源单菜品塞进 comparison_results[源行].items
     items = next((r.get("items") or [] for r in results if r.get("is_source")), [])
     fields = dict(
         business_type=business_type or "food",
         information=done_params.get("information") or None,
+        fail_reason=fail_reason,
         # best_deeplink 来自客户端剪贴板采集,harvest 拿不到 → 留空(灰度期 fromComparison 会补;
         # 纯 harvest 行「再次比价」退化为按 package 拉起 App。要精确深链需客户端另传,后续)。
         trace_url=trace_url or done_params.get("trace_url"),

app/repositories/coupon_state.py

@@ -157,6 +157,22 @@ def session_app_env(db: Session, trace_id: str | None) -> str | None:
     ).scalar_one_or_none()
 
 
+def merge_session_planned_count(
+    db: Session, trace_id: str | None, planned_count: int | None
+) -> None:
+    """把 pricebot 队列总数回写 session，供中途退出/全跳过场次计算逐券分母。"""
+    if not trace_id or planned_count is None or planned_count <= 0:
+        return
+    row = db.execute(
+        select(CouponSession).where(CouponSession.trace_id == trace_id)
+    ).scalar_one_or_none()
+    if row is None:
+        return
+    # 队列在一个 trace 内固定；取大值可防乱序旧帧覆盖，也兼容调度层补入预跳券。
+    row.planned_coupon_count = max(row.planned_coupon_count or 0, planned_count)
+    db.commit()
+
+
 def record_claims(
     db: Session,
     device_id: str,

app/schemas/compare_record.py

@@ -173,6 +173,8 @@ class ComparisonRecordOut(BaseModel):
     skipped_dish_count: int | None = None
     status: str
     information: str | None = None
+    # 失败卡「原因」文案:具体失败给具体原因,纯系统失败为 None(端侧品牌兜底)。见模型 fail_reason。
+    fail_reason: str | None = None
     items: list = []
     comparison_results: list = []
     skipped_dish_names: list = []

tests/test_compare_harvest.py

@@ -102,6 +102,7 @@ def test_harvest_done_derives_and_newly_success_once(client) -> None:
         assert rec.is_source_best is False
         assert rec.store_name == "测试店"
         assert rec.information == "美团更便宜"
+        assert rec.fail_reason is None               # 成功记录不派生失败原因
         assert rec.items == [{"name": "肥牛饭", "qty": 1}]
         assert rec.trace_url.endswith("/done/")
         # 再来一次(重试 done)→ 已 success,newly_success=False(发奖不重复触发)
@@ -110,6 +111,35 @@ def test_harvest_done_derives_and_newly_success_once(client) -> None:
         assert newly2 is False
 
 
+def test_harvest_done_failed_derives_fail_reason(client) -> None:
+    """failed 记录:记录级 information 笼统,但 fail_reason 从 platform_results 救出具体原因
+    (id 3030 型:美团系统失败 + 京东 items_not_found → 展示京东那条)。"""
+    tid = _tid()
+    done_failed = {
+        "comparison_results": [
+            {"platform_id": "taobao_flash", "platform_name": "淘宝闪购",
+             "package": "com.taobao.taobao", "price": 23.04, "is_source": True, "rank": 1,
+             "items": [{"name": "肥牛饭", "qty": 1}]},
+        ],
+        "platform_results": {
+            "taobao_flash": {"is_source": True, "status": "source", "price": 23.04},
+            "meituan_waimai": {"is_source": False, "status": "failed",
+                               "reason": "搜索店铺失败, 无法跳转到搜索页"},
+            "jd_waimai_standalone": {"is_source": False, "status": "items_not_found",
+                                     "reason": "京东外卖此店内未找到这些菜品"},
+        },
+        "information": "比价过程出错，请稍后重试",
+    }
+    with SessionLocal() as db:
+        crud.harvest_running(db, trace_id=tid, user_id=None)
+        rec, newly = crud.harvest_done(db, trace_id=tid, user_id=None,
+                                       done_params=done_failed)
+        assert newly is False                        # 没落成 success
+        assert rec.status == "failed"
+        assert rec.fail_reason == "京东外卖此店内未找到这些菜品"
+        assert rec.information == "比价过程出错，请稍后重试"  # 原文案仍留存
+
+
 def test_harvest_abort_cancels_running(client) -> None:
     tid = _tid()
     with SessionLocal() as db:

tests/test_coupon_point_score.py

@@ -14,10 +14,11 @@ from app.admin.repositories.coupon_data import (
 from app.admin.security import create_admin_token
 from app.db.session import SessionLocal
 from app.models.coupon_state import CouponClaimEvent, CouponSession
+from app.repositories.coupon_state import merge_session_planned_count
 
 
 def test_point_scores_by_trace() -> None:
-    """已领算成功、失败算尝试、跳过不进分母。"""
+    """已领算成功；失败和跳过均属于本轮计划券，进入分母。"""
     db = SessionLocal()
     trace = "point-score-trace"
     try:
@@ -37,7 +38,7 @@ def test_point_scores_by_trace() -> None:
 
         stats = _point_scores_by_trace(db, [trace])[trace]
         assert stats["succeeded"] == 2
-        assert stats["tried"] == 3
+        assert stats["tried"] == 4
         details = coupon_point_details(db, trace_id=trace)
         assert [item["status"] for item in details] == [
             "success", "already_claimed", "failed", "skipped"
@@ -48,8 +49,8 @@ def test_point_scores_by_trace() -> None:
         db.close()
 
 
-def test_skipped_detail_does_not_create_a_score() -> None:
-    """仅有 skipped 时按需明细仍可查到，但列表没有虚假的 0/0 分数。"""
+def test_skipped_detail_creates_zero_score() -> None:
+    """仅有 skipped 时也应显示 0/1，而不是把计划券静默成空值。"""
     db = SessionLocal()
     trace = "point-score-skipped"
     try:
@@ -63,7 +64,7 @@ def test_skipped_detail_does_not_create_a_score() -> None:
         db.flush()
 
         scores = _point_scores_by_trace(db, [trace, "missing-trace"])
-        assert trace not in scores
+        assert scores[trace] == {"succeeded": 0, "tried": 1}
         assert "missing-trace" not in scores
         assert coupon_point_details(db, trace_id=trace)[0]["status"] == "skipped"
     finally:
@@ -114,6 +115,88 @@ def test_coupon_data_report_returns_scores_without_embedding_details() -> None:
         db.close()
 
 
+def test_report_uses_planned_count_when_events_are_partial_or_missing() -> None:
+    """退出前只有部分/没有逐券终态时，计划数仍是稳定分母。"""
+    db = SessionLocal()
+    report_date = date(2020, 1, 6)
+    partial_trace = "point-score-planned-partial"
+    empty_trace = "point-score-planned-empty"
+    try:
+        db.add_all([
+            CouponSession(
+                trace_id=partial_trace,
+                device_id="planned-partial-device",
+                status="abandoned",
+                app_env="prod",
+                platforms=[],
+                planned_coupon_count=8,
+                started_at=datetime(2020, 1, 6, tzinfo=UTC),
+                started_date=report_date,
+            ),
+            CouponSession(
+                trace_id=empty_trace,
+                device_id="planned-empty-device",
+                status="abandoned",
+                app_env="prod",
+                platforms=[],
+                planned_coupon_count=8,
+                started_at=datetime(2020, 1, 6, tzinfo=UTC),
+                started_date=report_date,
+            ),
+            CouponClaimEvent(
+                trace_id=partial_trace,
+                device_id="planned-partial-device",
+                coupon_id="mt-planned-success",
+                claim_date=report_date,
+                status="success",
+            ),
+        ])
+        db.flush()
+
+        report = coupon_data_report(
+            db,
+            date_from=report_date.isoformat(),
+            date_to=report_date.isoformat(),
+            app_env="prod",
+        )
+        rows = {item["trace_id"]: item for item in report["items"]}
+        assert rows[partial_trace]["point_success_count"] == 1
+        assert rows[partial_trace]["point_total_count"] == 8
+        assert rows[empty_trace]["point_success_count"] == 0
+        assert rows[empty_trace]["point_total_count"] == 8
+    finally:
+        db.rollback()
+        db.close()
+
+
+def test_planned_count_only_grows_for_a_trace() -> None:
+    """乱序帧不得用较小的队列总数覆盖已经观测到的计划数。"""
+    db = SessionLocal()
+    trace = "point-score-planned-upsert"
+    try:
+        row = CouponSession(
+            trace_id=trace,
+            device_id="planned-upsert-device",
+            status="started",
+            app_env="prod",
+            platforms=[],
+            started_at=datetime(2020, 1, 7, tzinfo=UTC),
+            started_date=date(2020, 1, 7),
+        )
+        db.add(row)
+        db.commit()
+
+        merge_session_planned_count(db, trace, 8)
+        merge_session_planned_count(db, trace, 3)
+        db.refresh(row)
+        assert row.planned_coupon_count == 8
+    finally:
+        db.rollback()
+        db.execute(delete(CouponSession).where(CouponSession.trace_id == trace))
+        db.commit()
+        db.close()
+
+
 def test_coupon_point_details_endpoint() -> None:
     """前端点击使用的接口按约定返回 trace_id 和逐券 items。"""
     db = SessionLocal()

tests/test_fail_reason.py

@@ -0,0 +1,122 @@
+"""失败卡展示原因派生(repositories.comparison._derive_fail_display)单元测试。
+
+用例取自线上真实 failed 记录(platform_results 形态),覆盖:
+  - information 具体 → 直出
+  - information 笼统 + platform_results 有干净业务结局 → 救援出该原因(id 3030/2964 型)
+  - information 笼统 + 仅系统失败(搜索失败等黑话)→ None(端侧品牌兜底,id 3027 型)
+  - store_closed / no_delivery 被 pricebot 漏成 status=failed → 按 reason 关键字补判
+  - 打烊类脏店名 blob → 统一简短模板
+  - platform_results 为空 / 非对象 → None
+"""
+from __future__ import annotations
+
+from app.repositories import comparison as crud
+
+
+def test_specific_information_passthrough() -> None:
+    # information 本身具体(未达起送/找不到菜等)→ 直出,不看 platform_results
+    assert (
+        crud._derive_fail_display("淘宝闪购未达起送门槛，可加菜凑单后下单", {})
+        == "淘宝闪购未达起送门槛，可加菜凑单后下单"
+    )
+    assert crud._derive_fail_display("未识别到商品", {}) == "未识别到商品"
+
+
+def test_generic_info_rescued_from_items_not_found() -> None:
+    # id 3030 型:美团系统失败 + 京东 items_not_found,记录级 information 笼统 → 救出京东那条
+    pr = {
+        "eleme": {"is_source": True, "status": "source", "price": 23.04},
+        "meituan_waimai": {
+            "is_source": False, "status": "failed",
+            "reason": "搜索店铺失败, 无法跳转到搜索页",
+        },
+        "jd_waimai_standalone": {
+            "is_source": False, "status": "items_not_found",
+            "reason": "京东外卖此店内未找到这些菜品",
+        },
+    }
+    assert (
+        crud._derive_fail_display("比价过程出错，请稍后重试", pr)
+        == "京东外卖此店内未找到这些菜品"
+    )
+
+
+def test_generic_info_rescued_from_store_not_found() -> None:
+    # id 2964 型:美团系统失败 + 京东 store_not_found → 救出京东相似店铺文案
+    pr = {
+        "taobao_flash": {"is_source": True, "status": "source", "price": 127.98},
+        "meituan": {
+            "is_source": False, "status": "failed",
+            "reason": "比价过程出错，请稍后重试",
+        },
+        "jd_waimai": {
+            "is_source": False, "status": "store_not_found",
+            "reason": "未在京东找到「黔珍味·贵州牛肉蘸水健康菜 (望京店)」相似店铺",
+        },
+    }
+    assert (
+        crud._derive_fail_display("比价过程出错，请稍后重试", pr)
+        == "未在京东找到「黔珍味·贵州牛肉蘸水健康菜 (望京店)」相似店铺"
+    )
+
+
+def test_generic_info_pure_system_failure_returns_none() -> None:
+    # id 3027 型:唯一目标平台是自动化黑话失败 → 不给用户看 → None(端侧品牌兜底)
+    pr = {
+        "eleme": {"is_source": True, "status": "source", "price": 18.83},
+        "meituan_waimai": {
+            "is_source": False, "status": "failed",
+            "reason": "搜索店铺失败, 无法跳转到搜索页",
+        },
+    }
+    assert crud._derive_fail_display("比价过程出错，请稍后重试", pr) is None
+
+
+def test_store_closed_leaked_to_failed_is_rescued_and_cleaned() -> None:
+    # 打烊被漏成 status=failed;reason 常带脏店名 blob → 统一简短模板
+    pr = {
+        "jd_waimai": {"is_source": True, "status": "source", "price": 25},
+        "taobao_flash": {
+            "is_source": False, "status": "failed",
+            "reason": "淘宝闪购「沙胆彪炭炉牛杂煲(...)，蜂鸟准时达，月售300+，起送¥20」本店已休息,无法比价",
+        },
+    }
+    assert crud._derive_fail_display("比价过程出错，请稍后重试", pr) == "门店休息中，无法比价"
+
+    pr2 = {
+        "taobao_flash": {"is_source": True, "status": "source", "price": 31.83},
+        "meituan": {
+            "is_source": False, "status": "failed",
+            "reason": "美团「奈雪的茶（北京王府井奥莱·香江」门店已打烊,无法比价",
+        },
+    }
+    assert crud._derive_fail_display("比价出错", pr2) == "门店已打烊，无法比价"
+
+
+def test_no_delivery_leaked_to_failed_is_rescued() -> None:
+    # 单点不配送被漏成 status=failed;reason 本身干净 → 直接用
+    reason = "京东外卖该商家所选商品单点不配送，无法进入结算比价"
+    pr = {
+        "taobao_flash": {"is_source": True, "status": "source", "price": 20.1},
+        "jd_waimai_standalone": {
+            "is_source": False, "status": "failed", "reason": reason,
+        },
+    }
+    assert crud._derive_fail_display("比价过程出错，请稍后重试", pr) == reason
+
+
+def test_empty_or_missing_platform_results_returns_none() -> None:
+    # 「比价出错」+ 空 {} / None / 非对象:引擎早夭,无可展示原因 → None
+    assert crud._derive_fail_display("比价出错", {}) is None
+    assert crud._derive_fail_display("比价过程出错，请稍后重试", None) is None
+    assert crud._derive_fail_display("比价出错", []) is None  # 老 array 形态,防御
+
+
+def test_clean_status_wins_over_priority_order() -> None:
+    # 多个业务结局同现时按 _BIZ_STATUS_PRIORITY 选(below_minimum 优先于 store_not_found)
+    pr = {
+        "src": {"is_source": True, "status": "source", "price": 30},
+        "a": {"is_source": False, "status": "store_not_found", "reason": "未找到店铺A"},
+        "b": {"is_source": False, "status": "below_minimum", "reason": "B未达起送门槛"},
+    }
+    assert crud._derive_fail_display("比价过程出错，请稍后重试", pr) == "B未达起送门槛"