比价记录失败卡展示具体原因(新增 fail_reason)

失败记录不再一律「网络开小差」:新增记录级 fail_reason 派生列——information
具体则直出,笼统则从 platform_results 救出业务原因(找不到店/菜、未起送、打烊、
单点不配送等),纯系统失败为 None → 端侧品牌兜底。store_closed/no_delivery 被
pricebot 漏成 status=failed 的按 reason 补判,打烊脏店名统一简短模板。接入
harvest_done 与灰度期 upsert_record 两条写路径。

- models: comparison_record.fail_reason 列
- repositories: _derive_fail_display + 补判/清洗 helper,两条写路径接入
- schemas: ComparisonRecordOut 暴露 fail_reason
- alembic: 加列 + 回填老 specific 失败记录
- tests: _derive_fail_display 单测(8 例)+ harvest 失败落库集成测试

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

.env.example

@@ -175,9 +175,3 @@ PANGLE_REPORT_SITE_ID_TEST=5832303
 # APPLOG_MAX_BATCH=500                      # 单批最大条数(超 → 422;导入期常量,改需重启)
 # APPLOG_MAX_BODY_BYTES=2097152            # 请求体上限 2MB(超 → 413;运行期可调)
 # APPLOG_MAX_MSG_BYTES=8192                # 单条 msg 超此字节数截断
-
-# ===== 华为 Push =====
-HUAWEI_PUSH_APP_ID=
-HUAWEI_PUSH_APP_SECRET=
-# 0=正式消息(默认)；1=测试消息(仅开发联调，生产必须保持 0)
-HUAWEI_PUSH_TARGET_USER_TYPE=0

alembic/versions/6d2309208549_merge_comparison_platforms_fail_reason_.py

@@ -1,26 +0,0 @@
-"""merge comparison platforms + fail_reason heads
-
-Revision ID: 6d2309208549
-Revises: comparison_platforms_col, comparison_record_fail_reason
-Create Date: 2026-07-29 01:48:41.868083
-
-"""
-from typing import Sequence, Union
-
-from alembic import op
-import sqlalchemy as sa
-
-
-# revision identifiers, used by Alembic.
-revision: str = '6d2309208549'
-down_revision: Union[str, Sequence[str], None] = ('comparison_platforms_col', 'comparison_record_fail_reason')
-branch_labels: Union[str, Sequence[str], None] = None
-depends_on: Union[str, Sequence[str], None] = None
-
-
-def upgrade() -> None:
-    pass
-
-
-def downgrade() -> None:
-    pass

alembic/versions/comparison_platforms_col.py

@@ -1,47 +0,0 @@
-"""add platforms unified array column to comparison_record
-
-展示模型统一数组(pricebot done.params.platforms 原样存): 每平台一行、自带
-status/is_best/display, 记录页据此直接渲染, 不再靠 comparison_results + 客户端合并 + 前端派生。
-纯新增列, 老记录为空 → 前端回退老 comparison_results。
-
-Revision ID: comparison_platforms_col
-Revises: user_manual_risk_fields
-"""
-from __future__ import annotations
-
-from collections.abc import Sequence
-
-import sqlalchemy as sa
-from sqlalchemy.dialects import postgresql
-
-from alembic import op
-
-revision: str = "comparison_platforms_col"
-down_revision: str | None = "user_manual_risk_fields"
-branch_labels: str | Sequence[str] | None = None
-depends_on: str | Sequence[str] | None = None
-
-_JSON = sa.JSON().with_variant(postgresql.JSONB(), "postgresql")
-
-
-def upgrade() -> None:
-    # 幂等: 线上为了提前给历史数据补 platforms(2026-07-29), 已手动
-    # `ALTER TABLE comparison_record ADD COLUMN IF NOT EXISTS platforms jsonb
-    #  NOT NULL DEFAULT '[]'::jsonb`(与本 migration 定义一致)。列已存在时跳过,
-    # 否则上线 alembic upgrade head 会撞 DuplicateColumn 直接部署失败。
-    bind = op.get_bind()
-    cols = {c["name"] for c in sa.inspect(bind).get_columns("comparison_record")}
-    if "platforms" in cols:
-        return
-    with op.batch_alter_table("comparison_record") as batch_op:
-        batch_op.add_column(
-            sa.Column(
-                "platforms", _JSON, nullable=False,
-                server_default=sa.text("'[]'"),
-            )
-        )
-
-
-def downgrade() -> None:
-    with op.batch_alter_table("comparison_record") as batch_op:
-        batch_op.drop_column("platforms")

alembic/versions/guide_video_scene_unique.py

@@ -1,31 +0,0 @@
-"""guide video play count is independent for coupon and comparison
-
-Revision ID: guide_video_scene_unique
-Revises: 6d2309208549
-"""
-from alembic import op
-
-revision = "guide_video_scene_unique"
-down_revision = "6d2309208549"
-branch_labels = None
-depends_on = None
-
-
-def upgrade() -> None:
-    op.drop_index("uq_guide_video_play_user_seq", table_name="guide_video_play")
-    op.create_index(
-        "uq_guide_video_play_user_scene_seq",
-        "guide_video_play",
-        ["user_id", "scene", "seq"],
-        unique=True,
-    )
-
-
-def downgrade() -> None:
-    op.drop_index("uq_guide_video_play_user_scene_seq", table_name="guide_video_play")
-    op.create_index(
-        "uq_guide_video_play_user_seq",
-        "guide_video_play",
-        ["user_id", "seq"],
-        unique=True,
-    )

app/admin/repositories/ad_revenue.py

@@ -43,7 +43,7 @@ _KNOWN_PROD_BUSINESS_CODE_IDS = frozenset({"104098712", "104099389"})
 _TEST_BUSINESS_CODE_IDS = frozenset({"104127529", "104127626", "104137445"})
 
 
-def business_code_ids(db: Session, app_env: str | None) -> set[str]:
+def _business_code_ids(db: Session, app_env: str | None) -> set[str]:
     """返回指定应用环境下可用于业务收益对账的 GroMore 聚合代码位。"""
     prod_config = app_config.get_ad_config(db)
     prod_ids = set(_KNOWN_PROD_BUSINESS_CODE_IDS) | {
@@ -320,10 +320,10 @@ def ad_revenue_report(
 
     # 业务口径仅保留正式配置/测试业务链路实际使用的代码位。穿山甲“全量”还包含广告测试
     # demo、插屏等没有客户端收益上报的曝光，两边直接比较会天然产生假差额。
-    business_ids: set[str] | None = None
+    business_code_ids: set[str] | None = None
     if revenue_scope == "business":
-        business_ids = business_code_ids(db, app_env)
-        events = [e for e in events if e.get("our_code_id") in business_ids]
+        business_code_ids = _business_code_ids(db, app_env)
+        events = [e for e in events if e.get("our_code_id") in business_code_ids]
 
     # 排序:time=按时间倒序(新→旧);ecpm=按 eCPM 数值倒序(eCPM 原值是字符串「分」,转数值排;
     # 纯发奖行用其发奖采用的 eCPM,缺失/非法计 0 排末尾)。
@@ -381,7 +381,7 @@ def ad_revenue_report(
             date_from=date_from,
             date_to=date_to,
             app_env=app_env,
-            our_code_ids=business_ids,
+            our_code_ids=business_code_ids,
         )
         if pangle_aggs:
             by_date = {a["date"]: a for a in pangle_aggs}

app/admin/repositories/queries.py

@@ -14,7 +14,6 @@ from sqlalchemy.orm import Session
 
 from app.core import rewards
 from app.core.config import settings
-from app.models.ad_ecpm import AdEcpmRecord
 from app.models.ad_feed_reward import AdFeedRewardRecord
 from app.models.ad_reward import AdRewardRecord
 from app.models.admin import AdminAuditLog
@@ -1258,15 +1257,11 @@ def user_reward_stats(
     date_from: datetime | None = None,
     date_to: datetime | None = None,
     withdraw_source: str | None = None,
-    app_env: str | None = None,
-    revenue_scope: str = "all",
-    feed_scene: str | None = None,
 ) -> dict:
     """提现详情「用户统计区」10 项。窗口作用于除「现金余额」外的所有项(余额是当前快照)。
 
-    口径:激励视频/信息流奖励数量只统计 granted;数量——视频按条数、信息流按份数(unit_count 累加)。
-    平均 Draw eCPM 与广告收益报表一致:基于 ad_ecpm_record 的全部 draw/feed 展示记录计算，
-    不以是否发奖为筛选条件。各「提现」= 该来源累计金币折现。
+    口径:激励视频/信息流只统计 granted;数量——视频按条数、信息流按份数(unit_count 累加);
+    平均 eCPM 用原始分值(分/千次)按记录取算术平均;各「提现」= 该来源累计金币折现。
     传统任务 = 窗口内正向金币中,排除广告(reward_video/feed_ad_reward)与人工调整后的折现。
     """
     withdraw_source_conds = (
@@ -1301,68 +1296,30 @@ def user_reward_stats(
 
     # 只投影本统计实际使用的列。避免滚动发布或旧本地库尚未补齐无关新列时，
     # SQLAlchemy 因 select(ORM) 自动展开整表字段而让提现详情整体 500。
-    business_ids: set[str] | None = None
-    if revenue_scope == "business":
-        # 与广告收益报表共用正式/测试业务代码位集合，避免两个页面随配置切换后再次漂移。
-        from app.admin.repositories.ad_revenue import business_code_ids
-
-        business_ids = business_code_ids(db, app_env)
-
-    rv_conds = [
-        AdRewardRecord.user_id == user_id,
-        AdRewardRecord.reward_scene == "reward_video",
-        AdRewardRecord.status == "granted",
-        *_window_conds(AdRewardRecord.created_at, date_from, date_to),
-    ]
-    if app_env is not None:
-        rv_conds.append(AdRewardRecord.app_env == app_env)
-    if business_ids is not None:
-        rv_conds.append(AdRewardRecord.our_code_id.in_(business_ids))
     rv = db.execute(
         select(AdRewardRecord.ecpm_raw, AdRewardRecord.coin).where(
-            *rv_conds,
+            AdRewardRecord.user_id == user_id,
+            AdRewardRecord.reward_scene == "reward_video",
+            AdRewardRecord.status == "granted",
+            *_window_conds(AdRewardRecord.created_at, date_from, date_to),
         )
     ).all()
     rv_ecpms = [rewards.parse_ecpm_fen(r.ecpm_raw) for r in rv if r.ecpm_raw]
     rv_coins = sum(r.coin for r in rv)
 
-    feed_reward_conds = [
-        AdFeedRewardRecord.user_id == user_id,
-        AdFeedRewardRecord.status == "granted",
-        *_window_conds(AdFeedRewardRecord.created_at, date_from, date_to),
-    ]
-    if app_env is not None:
-        feed_reward_conds.append(AdFeedRewardRecord.app_env == app_env)
-    if feed_scene is not None:
-        feed_reward_conds.append(AdFeedRewardRecord.feed_scene == feed_scene)
-    if business_ids is not None:
-        feed_reward_conds.append(AdFeedRewardRecord.our_code_id.in_(business_ids))
-    feed_rewards = db.execute(
+    feed = db.execute(
         select(
             AdFeedRewardRecord.unit_count,
+            AdFeedRewardRecord.ecpm_raw,
             AdFeedRewardRecord.coin,
         ).where(
-            *feed_reward_conds,
+            AdFeedRewardRecord.user_id == user_id,
+            AdFeedRewardRecord.status == "granted",
+            *_window_conds(AdFeedRewardRecord.created_at, date_from, date_to),
         )
     ).all()
-    feed_coins = sum(f.coin for f in feed_rewards)
-
-    feed_impression_conds = [
-        AdEcpmRecord.user_id == user_id,
-        AdEcpmRecord.ad_type.in_(("draw", "feed")),
-        *_window_conds(AdEcpmRecord.created_at, date_from, date_to),
-    ]
-    if app_env is not None:
-        feed_impression_conds.append(AdEcpmRecord.app_env == app_env)
-    if feed_scene is not None:
-        feed_impression_conds.append(AdEcpmRecord.feed_scene == feed_scene)
-    if business_ids is not None:
-        feed_impression_conds.append(AdEcpmRecord.our_code_id.in_(business_ids))
-    feed_impressions = db.execute(
-        select(AdEcpmRecord.ecpm_raw).where(*feed_impression_conds)
-    ).all()
-    # 与 ad_revenue.category_stats 相同:每次展示权重相同，非法原值按 parse_ecpm_fen 记 0。
-    feed_ecpms = [rewards.parse_ecpm_fen(row.ecpm_raw) for row in feed_impressions]
+    feed_ecpms = [rewards.parse_ecpm_fen(f.ecpm_raw) for f in feed if f.ecpm_raw]
+    feed_coins = sum(f.coin for f in feed)
 
     trad_coins = db.execute(
         select(func.coalesce(func.sum(CoinTransaction.amount), 0)).where(
@@ -1381,7 +1338,7 @@ def user_reward_stats(
         "reward_video_count": len(rv),
         "reward_video_avg_ecpm": round(sum(rv_ecpms) / len(rv_ecpms), 2) if rv_ecpms else 0.0,
         "reward_video_cash_cents": _coins_to_cents(rv_coins),
-        "feed_count": int(sum(f.unit_count for f in feed_rewards)),
+        "feed_count": int(sum(f.unit_count for f in feed)),
         "feed_avg_ecpm": round(sum(feed_ecpms) / len(feed_ecpms), 2) if feed_ecpms else 0.0,
         "feed_cash_cents": _coins_to_cents(feed_coins),
     }

app/admin/routers/guide_video.py

@@ -9,7 +9,7 @@ client_max_body_size,见 shaguabijia-admin-web/deploy/nginx/admin.shaguabijia.co
 """
 from __future__ import annotations
 
-from typing import Annotated, Literal
+from typing import Annotated
 
 from fastapi import APIRouter, Depends, File, HTTPException, Request, UploadFile
 
@@ -27,21 +27,14 @@ router = APIRouter(
 )
 
 
-GuideScene = Literal["coupon", "comparison"]
-
-
-def _out(db: AdminDb, scene: GuideScene) -> GuideVideoConfigOut:
+def _out(db: AdminDb) -> GuideVideoConfigOut:
     """配置 + 播放统计合成响应(四个写接口都以最新状态返回,前端一次同步到位)。"""
-    return GuideVideoConfigOut(
-        scene=scene,
-        **guide_video.get_config(db, scene),
-        **guide_video.play_stats(db, scene),
-    )
+    return GuideVideoConfigOut(**guide_video.get_config(db), **guide_video.play_stats(db))
 
 
 @router.get("", response_model=GuideVideoConfigOut, summary="新手引导视频配置(领券浮层)")
-def get_config(db: AdminDb, scene: GuideScene = "coupon") -> GuideVideoConfigOut:
-    return _out(db, scene)
+def get_config(db: AdminDb) -> GuideVideoConfigOut:
+    return _out(db)
 
 
 @router.patch("", response_model=GuideVideoConfigOut, summary="改开关/次数/金币(带审计)")
@@ -50,23 +43,21 @@ def update_config(
     request: Request,
     admin: Annotated[AdminUser, Depends(require_role("operator"))],
     db: AdminDb,
-    scene: GuideScene = "coupon",
 ) -> GuideVideoConfigOut:
     before, after = guide_video.update_config(
         db,
         enabled=body.enabled,
         max_plays=body.max_plays,
         reward_coin=body.reward_coin,
-        scene=scene,
         admin_id=admin.id,
         commit=False,
     )
     write_audit(
         db, admin, action="guide_video.update", target_type="guide_video", target_id=None,
-        detail={"scene": scene, "before": before, "after": after}, ip=get_client_ip(request), commit=False,
+        detail={"before": before, "after": after}, ip=get_client_ip(request), commit=False,
     )
     db.commit()
-    return _out(db, scene)
+    return _out(db)
 
 
 @router.post("/video", response_model=GuideVideoConfigOut, summary="上传新手引导视频(MP4,带审计)")
@@ -74,26 +65,23 @@ async def upload_video(
     request: Request,
     admin: Annotated[AdminUser, Depends(require_role("operator"))],
     db: AdminDb,
-    file: Annotated[UploadFile, File()],
-    scene: GuideScene = "coupon",
+    file: UploadFile = File(...),
 ) -> GuideVideoConfigOut:
     data = await file.read()
     try:
         url = media.save_guide_video(data)
     except media.MediaError as e:
         raise HTTPException(status_code=400, detail=str(e)) from e
-    before, after = guide_video.set_video(
-        db, url, scene=scene, admin_id=admin.id, commit=False
-    )
+    before, after = guide_video.set_video(db, url, admin_id=admin.id, commit=False)
     write_audit(
         db, admin, action="guide_video.set_video", target_type="guide_video", target_id=None,
-        detail={"scene": scene, "before": before.get("video_url"), "after": url, "bytes": len(data)},
+        detail={"before": before.get("video_url"), "after": url, "bytes": len(data)},
         ip=get_client_ip(request), commit=False,
     )
     db.commit()
     # 提交成功后再删旧片,避免新片没落库就把旧片丢了
     media.delete_guide_video(before.get("video_url"))
-    return _out(db, scene)
+    return _out(db)
 
 
 @router.delete("/video", response_model=GuideVideoConfigOut, summary="移除新手引导视频(带审计)")
@@ -101,16 +89,13 @@ def delete_video(
     request: Request,
     admin: Annotated[AdminUser, Depends(require_role("operator"))],
     db: AdminDb,
-    scene: GuideScene = "coupon",
 ) -> GuideVideoConfigOut:
     """移除后 /guide-video/start 一律返回 should_play=false,领券浮层回到「只放广告」。"""
-    before, after = guide_video.set_video(
-        db, None, scene=scene, admin_id=admin.id, commit=False
-    )
+    before, after = guide_video.set_video(db, None, admin_id=admin.id, commit=False)
     write_audit(
         db, admin, action="guide_video.delete_video", target_type="guide_video", target_id=None,
-        detail={"scene": scene, "before": before.get("video_url")}, ip=get_client_ip(request), commit=False,
+        detail={"before": before.get("video_url")}, ip=get_client_ip(request), commit=False,
     )
     db.commit()
     media.delete_guide_video(before.get("video_url"))
-    return _out(db, scene)
+    return _out(db)

app/admin/routers/users.py

@@ -88,11 +88,6 @@ def get_user_reward_stats(
     withdraw_source: Annotated[
         str | None, Query(pattern="^(coin_cash|invite_cash)$")
     ] = None,
-    app_env: Annotated[str | None, Query(pattern="^(prod|test)$")] = None,
-    revenue_scope: Annotated[str, Query(pattern="^(business|all)$")] = "all",
-    feed_scene: Annotated[
-        str | None, Query(pattern="^(comparison|coupon|welfare)$")
-    ] = None,
 ) -> UserRewardStats:
     """提现详情抽屉「用户统计区」。date_from/date_to 都不传 = 注册至今(全量)。"""
     if not user_repo.user_exists(db, user_id):
@@ -104,9 +99,6 @@ def get_user_reward_stats(
             date_from=date_from,
             date_to=date_to,
             withdraw_source=withdraw_source,
-            app_env=app_env,
-            revenue_scope=revenue_scope,
-            feed_scene=feed_scene,
         )
     )
 

app/admin/schemas/guide_video.py

@@ -7,7 +7,6 @@ from app.repositories.guide_video import MAX_PLAYS_LIMIT, REWARD_COIN_LIMIT
 
 
 class GuideVideoConfigOut(BaseModel):
-    scene: str
     enabled: bool
     video_url: str | None = None   # 相对地址 /media/guide_video/xxx.mp4;未配片 = None
     max_plays: int

app/admin/schemas/user.py

@@ -60,7 +60,7 @@ class UserRewardStats(BaseModel):
     reward_video_avg_ecpm: float  # 平均激励视频 eCPM(分/千次)
     reward_video_cash_cents: int  # 激励视频提现(金币折现)
     feed_count: int  # 累计信息流广告数(granted 份数,unit_count 累加)
-    feed_avg_ecpm: float  # 全部 Draw/feed 实际展示的平均 eCPM(分/千次,含未发奖展示)
+    feed_avg_ecpm: float  # 平均信息流广告 eCPM(分/千次)
     feed_cash_cents: int  # 信息流广告提现(金币折现)
 
 

app/api/v1/compare.py

@@ -108,7 +108,7 @@ def _harvest_done_blocking(
 
 def _harvest_abort_blocking(
     trace_id: str, status_hint: str, reason: str | None, trace_url: str | None,
-) -> int | None:
+) -> None:
     with SessionLocal() as db:
         rec = crud_compare.harvest_abort(
             db, trace_id=trace_id, status=status_hint, reason=reason, trace_url=trace_url,
@@ -118,7 +118,6 @@ def _harvest_abort_blocking(
         extra={"phase": "harvest_abort",
                "status": (rec.status if rec else None), "reason": reason},
     )
-    return rec.id if rec is not None else None
 
 
 async def _forward(
@@ -292,10 +291,7 @@ async def trace_epilogue(
 
 @router.post("/trace/finalize", summary="比价 trace 收尾上云 (透传 + 夭折落库)")
 async def trace_finalize(
-    request: Request,
-    background_tasks: BackgroundTasks,
-    user: OptionalUser,
-    db: DbSession,
+    request: Request, user: OptionalUser, db: DbSession
 ) -> dict[str, Any]:
     _ensure_compare_allowed(user, db)
     # 用户终止 / Phase1 未识别没到 done 帧: pricebot 打包半截上云返回 {trace_url};
@@ -306,16 +302,12 @@ async def trace_finalize(
         request, "/api/trace/finalize", user, harvest_first_frame=False,
     )
     try:
-        record_id = await run_in_threadpool(
+        await run_in_threadpool(
             _harvest_abort_blocking, trace_id,
             (meta.get("status") or "cancelled"),
             (meta.get("reason") or meta.get("information")),
             (resp.get("trace_url") if isinstance(resp, dict) else None),
         )
-        if record_id is not None:
-            background_tasks.add_task(
-                backfill_comparison_llm_cost, record_id, trace_id
-            )
     except Exception as e:  # noqa: BLE001
         logger.warning("harvest_abort failed trace=%s: %s", trace_id, e)
     return resp

app/core/config.py

@@ -88,8 +88,6 @@ class Settings(BaseSettings):
     # (OAuth 换 token 时 client_id 即 AppId)。发送走 v1 messages:send,成功码 80000000。
     HUAWEI_PUSH_APP_ID: str = ""
     HUAWEI_PUSH_APP_SECRET: str = ""
-    # 0=正式消息(默认，受正式消息频控)；1=测试消息(仅开发联调，勿用于生产)。
-    HUAWEI_PUSH_TARGET_USER_TYPE: int = Field(default=0, ge=0, le=1)
     HUAWEI_PUSH_TOKEN_ENDPOINT: str = "https://oauth-login.cloud.huawei.com/oauth2/v3/token"
     HUAWEI_PUSH_SEND_ENDPOINT_TEMPLATE: str = (
         "https://push-api.cloud.huawei.com/v1/{app_id}/messages:send"

app/integrations/vendor_push.py

@@ -31,7 +31,6 @@ from app.core.config import settings
 logger = logging.getLogger("shagua.vendor_push")
 
 TYPE_ACCESSIBILITY_DISABLED = "accessibility_disabled"
-DATA_EVENT_NOTIFICATION_CREATED = "notification_created"
 SUPPORTED_VENDORS = frozenset({"honor", "huawei", "vivo", "xiaomi", "oppo"})
 
 # vendor key → 中文名(测试/配置状态接口展示用)
@@ -154,49 +153,6 @@ def send_accessibility_disabled(
     )
 
 
-def send_data_event(
-    push_vendor: str,
-    push_token: str,
-    *,
-    event: str,
-    notification_id: str,
-    mock: bool = False,
-) -> dict[str, Any]:
-    """发送无界面的轻量事件；当前只允许站内消息创建事件。"""
-    vendor = normalize_vendor(push_vendor)
-    token = push_token.strip() if push_token else ""
-    if not vendor or vendor not in SUPPORTED_VENDORS:
-        raise VendorPushError(f"unsupported push vendor: {push_vendor}")
-    if not token:
-        raise VendorPushError("push token is empty")
-    if event != DATA_EVENT_NOTIFICATION_CREATED:
-        raise VendorPushError(f"unsupported data event: {event}")
-    if not notification_id:
-        raise VendorPushError("notification_id is empty")
-
-    payload = {"event": event, "notificationId": str(notification_id)}
-    if mock:
-        return {"mock": True, "vendor": vendor, "payload": payload}
-
-    # vivo 的通知已设置 foregroundShow=false：App 在前台时必走
-    # onForegroundMessageArrived，正好就是本事件需要的刷新信号；不重复占用一次推送配额。
-    if vendor == "vivo":
-        return {"skipped": True, "reason": "foreground notification callback"}
-
-    # OPush 当前只支持通知栏消息，没有服务端透传单推接口。
-    # 客户端在 OPPO 首页可见时轮询未读数；这里必须安全跳过，不能请求不存在的
-    # /message/transparent/unicast（该地址会稳定返回 HTTP 404）。
-    if vendor == "oppo":
-        return {"skipped": True, "reason": "oppo does not support data messages"}
-
-    dispatch: dict[str, Callable[[str, dict[str, str]], dict[str, Any]]] = {
-        "honor": _send_honor_data,
-        "huawei": _send_huawei_data,
-        "xiaomi": _send_xiaomi_data,
-    }
-    return dispatch[vendor](token, payload)
-
-
 def _require(value: str, name: str) -> str:
     if not value:
         raise VendorPushError(f"{name} not configured")
@@ -294,15 +250,9 @@ def _honor_access_token() -> str:
 def _send_honor(token: str, title: str, body: str, extras: dict[str, str]) -> dict[str, Any]:
     app_id = _require(settings.HONOR_PUSH_APP_ID, "HONOR_PUSH_APP_ID")
     access_token = _honor_access_token()
-    click_action: dict[str, Any]
-    if extras.get("notificationId"):
-        # type=3 只负责打开首页，不保证 data 会变成目标 Activity 的 extras。消息中心通知必须
-        # 用自定义页面(type=1)+ intent URI，把 notificationId/type/feedbackId 等直接带给
-        # MainActivity；否则华为/荣耀系统能展示通知，但用户点击后客户端收不到任何导航参数。
-        click_action = {"type": 1, "intent": _click_intent_uri(extras)}
-    else:
-        click_action = {"type": 3}
     payload = {
+        # clickAction type=3(打开应用首页)时,荣耀点击会把 data JSON 的键值对注入启动 intent 的
+        # extras(与 HMS 同机制)→ MainActivity.consumeNavTarget 读 notif_id/notif_type 直达落地。
         "data": json.dumps(_click_extras(extras), ensure_ascii=False),
         "notification": {"title": title, "body": body},
         "android": {
@@ -311,7 +261,7 @@ def _send_honor(token: str, title: str, body: str, extras: dict[str, str]) -> di
             "notification": {
                 "title": title,
                 "body": body,
-                "clickAction": click_action,
+                "clickAction": {"type": 3},
                 "importance": "NORMAL",
             },
         },
@@ -333,25 +283,6 @@ def _send_honor(token: str, title: str, body: str, extras: dict[str, str]) -> di
     return data
 
 
-def _send_honor_data(token: str, payload: dict[str, str]) -> dict[str, Any]:
-    app_id = _require(settings.HONOR_PUSH_APP_ID, "HONOR_PUSH_APP_ID")
-    access_token = _honor_access_token()
-    data = _request_json(
-        "POST",
-        settings.HONOR_PUSH_SEND_ENDPOINT_TEMPLATE.format(app_id=app_id),
-        json={"data": json.dumps(payload, ensure_ascii=False), "token": [token]},
-        headers={
-            "Content-Type": "application/json; charset=UTF-8",
-            "Authorization": f"Bearer {access_token}",
-            "timestamp": str(int(time.time() * 1000)),
-        },
-    )
-    code = data.get("code")
-    if code is not None and int(code) != 200:
-        raise VendorPushError(f"honor data push failed: {data}")
-    return data
-
-
 def _huawei_access_token() -> str:
     """华为 OAuth2 client_credentials 换 access_token(client_id 即 AGC 应用的 AppId)。"""
     cache_key = "huawei"
@@ -381,24 +312,18 @@ def _send_huawei(token: str, title: str, body: str, extras: dict[str, str]) -> d
     '80100000' 为部分成功(单 token 场景仍视为失败,错误里带原始响应便于排障)。"""
     app_id = _require(settings.HUAWEI_PUSH_APP_ID, "HUAWEI_PUSH_APP_ID")
     access_token = _huawei_access_token()
-    click_action: dict[str, Any]
-    if extras.get("notificationId"):
-        # type=3 只打开首页，Mate 20/EMUI 不会把 message.data 自动拆成启动 Intent extras。
-        # type=1 的自定义 intent 才能稳定携带反馈记录 id，且冷启动/onNewIntent 都走同一路由。
-        click_action = {"type": 1, "intent": _click_intent_uri(extras)}
-    else:
-        click_action = {"type": 3}
     payload = {
         "validate_only": False,
         "message": {
+            # click_action type=3(打开应用首页)时,HMS 点击会把 data JSON 的键值对注入启动 intent
+            # 的 extras → MainActivity.consumeNavTarget 读 notif_id/notif_type 直达落地。
             "data": json.dumps(_click_extras(extras), ensure_ascii=False),
             "android": {
-                "target_user_type": settings.HUAWEI_PUSH_TARGET_USER_TYPE,
                 "ttl": f"{settings.PUSH_TIME_TO_LIVE_SEC}s",
                 "notification": {
                     "title": title,
                     "body": body,
-                    "click_action": click_action,
+                    "click_action": {"type": 3},
                     "importance": "NORMAL",
                 },
             },
@@ -419,29 +344,6 @@ def _send_huawei(token: str, title: str, body: str, extras: dict[str, str]) -> d
     return data
 
 
-def _send_huawei_data(token: str, payload: dict[str, str]) -> dict[str, Any]:
-    app_id = _require(settings.HUAWEI_PUSH_APP_ID, "HUAWEI_PUSH_APP_ID")
-    access_token = _huawei_access_token()
-    data = _request_json(
-        "POST",
-        settings.HUAWEI_PUSH_SEND_ENDPOINT_TEMPLATE.format(app_id=app_id),
-        json={
-            "validate_only": False,
-            "message": {
-                "data": json.dumps(payload, ensure_ascii=False),
-                "token": [token],
-            },
-        },
-        headers={
-            "Content-Type": "application/json; charset=UTF-8",
-            "Authorization": f"Bearer {access_token}",
-        },
-    )
-    if str(data.get("code", "")) != "80000000":
-        raise VendorPushError(f"huawei data push failed: {data}")
-    return data
-
-
 def _vivo_auth_token() -> str:
     cache_key = "vivo"
     cached = _cache_get(cache_key)
@@ -483,22 +385,15 @@ def _send_vivo(token: str, title: str, body: str, extras: dict[str, str]) -> dic
         "timeToLive": settings.PUSH_TIME_TO_LIVE_SEC,
         "requestId": uuid.uuid4().hex,
         "pushMode": settings.VIVO_PUSH_MODE,
-        # vivo SDK 4.1.5 只有在前台展示关闭时才调用
-        # OpenClientPushMessageReceiver.onForegroundMessageArrived。
-        # App 在该回调中刷新服务端未读数并补发一条本地通知；后台/锁屏时仍由
-        # vivo 系统展示，因而任何场景都只会出现一条通知。
-        "foregroundShow": False,
-        # vivo 官方 VPush 角标字段：离线收到通知时先由桌面自动 +1；
-        # App 启动/同步后再由统一未读数通过系统 API 精确校准。
-        "addBadge": True,
         "clientCustomMap": extras,
     }
-    # vivo Push SDK 480+ 已不再回调 skipType=3；必须使用 skipType=4 的完整 Intent URI。
-    # URI 同时包含 data/scheme、显式 component 和 S. extras，OriginOS 才会把业务参数
-    # 原样交给 MainActivity（仅靠隐式 deeplink 会打开 App，但可能剥掉 extras）。
+    # 点击落地:消息中心推送(带 notificationId)→ skipType=4 + skipContent=intent uri,由 vivo
+    # 系统直启 MainActivity 并携带 S. extras(与小米 notify_effect=2 同机制)。不依赖客户端
+    # VivoPushReceiver.onNotificationMessageClicked 里的后台 startActivity——Android 10+ BAL
+    # 会静默拦掉,receiver 路径仅作兜底。无 notificationId 的召回类保持 skipType=1 仅打开首页。
     if extras.get("notificationId"):
         payload["skipType"] = 4
-        payload["skipContent"] = _vivo_click_intent_uri(extras)
+        payload["skipContent"] = _click_intent_uri(extras)
     else:
         payload["skipType"] = 1
     if settings.VIVO_PUSH_CATEGORY:
@@ -512,25 +407,6 @@ def _send_vivo(token: str, title: str, body: str, extras: dict[str, str]) -> dic
             "authToken": auth_token,
         },
     )
-    # 10089 = 当前 vivo 应用尚未开通 VPush「离线自动角标」能力。
-    # 角标只是通知中心未读数的镜像，不能因此让整条通知发送失败：去掉 addBadge
-    # 原样重试，通知到达/应用同步后仍由客户端系统 API 按统一未读数精确设置。
-    if int(data.get("result", -1)) == 10089:
-        logger.warning("vivo VPush badge not enabled (10089); retrying without addBadge")
-        fallback_payload = dict(payload)
-        fallback_payload.pop("addBadge", None)
-        fallback_payload["requestId"] = uuid.uuid4().hex
-        data = _request_json(
-            "POST",
-            settings.VIVO_PUSH_SEND_ENDPOINT,
-            json=fallback_payload,
-            headers={
-                "Content-Type": "application/json",
-                "authToken": auth_token,
-            },
-        )
-        if int(data.get("result", -1)) == 0:
-            data = {**data, "badgeFallback": True}
     if int(data.get("result", -1)) != 0:
         raise VendorPushError(f"vivo push failed: {data}")
     return data
@@ -581,28 +457,6 @@ def _send_xiaomi(token: str, title: str, body: str, extras: dict[str, str]) -> d
     return data
 
 
-def _send_xiaomi_data(token: str, payload: dict[str, str]) -> dict[str, Any]:
-    app_secret = _require(settings.XIAOMI_PUSH_APP_SECRET, "XIAOMI_PUSH_APP_SECRET")
-    data = _request_form(
-        "POST",
-        settings.XIAOMI_PUSH_SEND_ENDPOINT,
-        data={
-            "registration_id": token,
-            "restricted_package_name": settings.ANDROID_PACKAGE_NAME,
-            "payload": json.dumps(payload, ensure_ascii=False),
-            "pass_through": "1",
-            "time_to_live": str(settings.PUSH_TIME_TO_LIVE_SEC * 1000),
-        },
-        headers={"Authorization": f"key={app_secret}"},
-    )
-    code = data.get("code")
-    if code not in (0, "0", None):
-        raise VendorPushError(f"xiaomi data push failed: {data}")
-    if str(data.get("result", "ok")).lower() not in ("ok", "success"):
-        raise VendorPushError(f"xiaomi data push failed: {data}")
-    return data
-
-
 def _click_extras(extras: dict[str, str]) -> dict[str, str]:
     """点击落地参数:消息中心推送(extras 带 notificationId)补 notif_id/notif_type 别名——
     客户端 MainActivity.consumeNavTarget 首选这两个键(厂商 receiver 路径的历史约定),原始键
@@ -636,24 +490,6 @@ def _click_intent_uri(extras: dict[str, str]) -> str:
     return ";".join(parts)
 
 
-def _vivo_click_intent_uri(extras: dict[str, str]) -> str:
-    """vivo skipType=4 使用官方要求的完整 intent deeplink 形态。
-
-    `?#Intent` 中的问号不能省；OriginOS 对缺少它的 URI 会退化为“仅打开应用”，
-    不把 S. 参数交给目标 Activity。
-    """
-    pkg = settings.ANDROID_PACKAGE_NAME
-    parts = [
-        "intent://push/detail?#Intent",
-        "scheme=shaguabijia",
-        f"component={pkg}/{pkg}.MainActivity",
-        "launchFlags=0x14000000",
-    ]
-    parts += [f"S.{key}={quote(str(value), safe='')}" for key, value in _click_extras(extras).items()]
-    parts.append("end")
-    return ";".join(parts)
-
-
 def _xiaomi_template_param(title: str, alert: str) -> str:
     rendered = (
         settings.XIAOMI_PUSH_TEMPLATE_PARAM_JSON
@@ -712,11 +548,6 @@ def _send_oppo(token: str, title: str, body: str, extras: dict[str, str]) -> dic
         "off_line_ttl": ttl_hours,
         "action_parameters": json.dumps(_click_extras(extras), ensure_ascii=False),
     }
-    if extras.get("notificationId"):
-        # 只有已落库的站内消息才计入角标；保活提醒等临时推送不应留下无法消除的未读数。
-        # 客户端打开/已读后会按服务端真实未读总数覆盖，避免长期漂移。
-        notification["badge_operation_type"] = 1
-        notification["badge_message_count"] = 1
     # 点击落地:OPPO SDK 没有点击回调,参数只能靠服务端点击动作配置送达——action_parameters 的
     # 键值对仅在 click_action_type=1/4 时才会注入目标 Activity 的 intent extras(type=0「启动应用」
     # 会忽略它,extras 全丢 → 点了没反应,与小米 notify_effect=1 同款坑)。

app/models/comparison.py

@@ -113,10 +113,6 @@ class ComparisonRecord(Base):
     items: Mapped[list] = mapped_column(_JSON, nullable=False, default=list)
     # 逐平台对比 [{platform_id, platform_name, package, price, is_source, rank, coupon_saved, coupon_name, applied_coupons}](price/coupon_saved 单位:元,原样存;coupon_name=优惠来源名;applied_coupons=[{name,amount}] 多券明细)
     comparison_results: Mapped[list] = mapped_column(_JSON, nullable=False, default=list)
-    # 展示模型统一数组(pricebot done.params.platforms 原样存): 每平台一行、自带
-    # status/is_best/display/display_order, 记录页据此直接渲染, 不再靠 comparison_results
-    # + 客户端合并 + 前端派生。老记录/旧客户端为空 → 前端回退老 comparison_results 渲染。
-    platforms: Mapped[list] = mapped_column(_JSON, nullable=False, default=list)
     # 目标平台未找到、跳过的菜名
     skipped_dish_names: Mapped[list] = mapped_column(_JSON, nullable=False, default=list)
     # 客户端上报的原始 payload(calibration + done.params 全量),未来取数兜底

app/models/guide_video.py

@@ -38,7 +38,7 @@ class GuideVideoPlay(Base):
         # start_play 捕获 IntegrityError 降级成"这次不放视频"。
         # 用 unique Index 而非 UniqueConstraint:与迁移里的 create_index 对齐(SQLite 加约束
         # 要整表重建),autogenerate 才不会每次报一条假 diff。
-        Index("uq_guide_video_play_user_scene_seq", "user_id", "scene", "seq", unique=True),
+        Index("uq_guide_video_play_user_seq", "user_id", "seq", unique=True),
     )
 
     id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)

app/repositories/comparison.py

@@ -164,9 +164,8 @@ def _derive(payload: ComparisonRecordIn) -> dict:
 
     is_source_best = best.is_source if best is not None else None
 
-    # status:优先 pricebot record_status(区分 below_minimum/store_closed) → 客户端显式 status
-    # → 兜底"非源且有价"=success/否则 failed。record_status 让"未满起送"不再塌缩成 failed。
-    status = payload.record_status or payload.status
+    # status:客户端显式给了就用;否则有"非源且有价"的结果=success,否则 failed
+    status = payload.status
     if status is None:
         has_valid_target = any(
             (not r.is_source) and r.price is not None for r in results
@@ -198,35 +197,16 @@ def upsert_record(
     灰度期老客户端 POST /compare/record 走这条,与后端 harvest 按 trace_id reconcile;
     新客户端不再 POST(改由 compare.py 透传壳 harvest 落库)。
     """
-    # 单源派生: 与 harvest_done 一致, payload 带 platforms 时从它派生(唯一真相源
-    # _derive_from_platforms), 老客户端不带 platforms 时回退 _derive(从 comparison_results)。
-    if payload.platforms:
-        derived = _derive_from_platforms(payload.platforms, payload.record_status)
-        # 对齐 _derive 返回键(#189 fail_reason): 两路径 fields 键集一致, 覆盖已有行时不残留旧值
-        derived["fail_reason"] = (
-            _derive_fail_display(payload.information, payload.platform_results or {})
-            if derived["status"] == "failed"
-            else None
-        )
-        # 单源派生取自 platforms 源行(常无源平台元数据/店名)→ 空则用 payload 兜底不丢字段。
-        # 下面 fields 不再显式写这四个键, 统一由 derived 提供(否则 dict(store_name=..., **derived)
-        # 与 _derive_from_platforms 同名键撞键 TypeError)。
-        for _k in ("store_name", "source_platform_id", "source_platform_name", "source_package"):
-            if not derived.get(_k):
-                derived[_k] = getattr(payload, _k)
-    else:
-        derived = _derive(payload)
-        # _derive 只从 comparison_results 派生, 不含源平台四件套 / store_name → 从 payload 补,
-        # 与上面 platforms 分支键集对齐(fields 统一靠 **derived 提供这些列)。
-        for _k in ("store_name", "source_platform_id", "source_platform_name", "source_package"):
-            derived[_k] = getattr(payload, _k)
+    derived = _derive(payload)
     items = [it.model_dump(exclude_none=True) for it in payload.items]
     fields = dict(
         device_id=payload.device_id,
         business_type=payload.business_type,
+        store_name=payload.store_name,
         product_names=_product_names_from_items(items),
-        # store_name / source_platform_id / source_platform_name / source_package 统一由
-        # derived 提供(见上方两分支补齐), 不在此显式写 —— 否则与 _derive_from_platforms 撞键。
+        source_platform_id=payload.source_platform_id,
+        source_platform_name=payload.source_platform_name,
+        source_package=payload.source_package,
         information=payload.information,
         best_deeplink=payload.best_deeplink,
         trace_url=payload.trace_url,
@@ -234,7 +214,6 @@ def upsert_record(
         skipped_dish_count=payload.skipped_dish_count,
         items=items,
         comparison_results=[r.model_dump() for r in payload.comparison_results],
-        platforms=list(payload.platforms or []),
         skipped_dish_names=list(payload.skipped_dish_names),
         # 客户端环境 / 性能(debug,客户端上报;旧客户端为 None)
         device_model=payload.device_model,
@@ -307,8 +286,7 @@ def upsert_record(
 
 
 def _derive_from_results(
-    results: list[dict], platform_results: dict | None = None,
-    record_status: str | None = None,
+    results: list[dict], platform_results: dict | None = None
 ) -> dict:
     """从 done 帧 comparison_results(pricebot 原始 dict 列表)派生结构化列。
     等价 _derive,但吃原始字段(is_source/price/rank/platform_id/store_name...)而非 pydantic 对象。
@@ -356,53 +334,7 @@ def _derive_from_results(
         "saved_amount_cents": saved_amount_cents,
         "is_source_best": best.get("is_source") if best else None,
         "store_name": (src_row or {}).get("store_name") or None,
-        # 记录级结局: 优先用 pricebot 下发的 record_status(区分 below_minimum/store_closed,
-        # 不再把"未满起送"塌缩成 failed → 记录页不再误报"网络开小差"); 旧 pricebot 未下发时
-        # 回退老的 success/failed 二态派生, 向后兼容。
-        "status": record_status or ("success" if has_valid_target else "failed"),
-    }
-
-
-def _derive_from_platforms(
-    platforms: list, record_status: str | None = None,
-) -> dict:
-    """从 done 帧 platforms(每平台一行、渲染就绪)派生结构化列——**单一真相源**。
-
-    best_* 直接取 platforms 里 is_best 的那一行、source_* 取 role=source 行,与前端读的
-    platforms 天然一致(不再像 _derive_from_results 那样从 comparison_results 二次评最优,
-    消除"标量列 vs platforms"双源不一致)。platforms 非空时优先走这里;老 pricebot 无
-    platforms 时调用方回退 _derive_from_results(向后兼容)。"""
-    rows = [p for p in (platforms or []) if isinstance(p, dict)]
-    src = next((p for p in rows if p.get("role") == "source"), None)
-    best = next((p for p in rows if p.get("is_best")), None)
-    source_price_cents = _yuan_to_cents(src.get("price")) if src else None
-    best_price_cents = _yuan_to_cents(best.get("price")) if best else None
-    saved_amount_cents = None
-    if source_price_cents is not None and best_price_cents is not None:
-        saved_amount_cents = source_price_cents - best_price_cents
-    has_valid_target = any(
-        p.get("role") != "source" and p.get("price") is not None for p in rows
-    )
-    # store_name: 优先源行; recompare 场景源平台自己当目标、源行被目标覆盖(pricebot
-    # _build_platform_rows 有意去重, platforms 无 role=source 行)→ 回退 best 行 → 首个有店名
-    # 的行(显示现场实际比到的店), 免得记录页店名空掉兜底显示成"比价"。正常比价有源行不走回退。
-    store_name = (
-        (src or {}).get("store_name")
-        or (best or {}).get("store_name")
-        or next((p.get("store_name") for p in rows if p.get("store_name")), None)
-    )
-    return {
-        "source_platform_id": (src or {}).get("platform_id"),
-        "source_platform_name": (src or {}).get("platform_name"),
-        "source_package": (src or {}).get("package"),
-        "source_price_cents": source_price_cents,
-        "best_platform_id": (best or {}).get("platform_id"),
-        "best_platform_name": (best or {}).get("platform_name"),
-        "best_price_cents": best_price_cents,
-        "saved_amount_cents": saved_amount_cents,
-        "is_source_best": (best.get("role") == "source") if best else None,
-        "store_name": store_name or None,
-        "status": record_status or ("success" if has_valid_target else "failed"),
+        "status": "success" if has_valid_target else "failed",
     }
 
 
@@ -561,29 +493,7 @@ def harvest_done(
     返回 (记录, 是否本次**新**落成 success)——供调用方据此幂等发一次邀请奖。
     行不存在(理论上帧0已建;防御)则新建。"""
     results = done_params.get("comparison_results") or []
-    # 展示模型统一数组(pricebot 新增, 每平台一行自带 status/is_best): 原样存, 记录页据此直渲染。
-    # record_status: 记录级结局(success/below_minimum/store_closed/failed), 覆盖老二态派生。
-    platforms = done_params.get("platforms") or []
-    record_status = done_params.get("record_status")
-    # 单源派生: platforms(含 pricebot 权威 is_best)是唯一真相源, best_*/source_*/saved/status
-    # 全从它取 → 与前端读的 platforms 天然一致; 菜品也取 platforms 源行。老 pricebot 无
-    # platforms 时回退从 comparison_results 派生(向后兼容)。
-    if platforms:
-        derived = _derive_from_platforms(platforms, record_status)
-        # 菜品优先源行; recompare 无源行 → 回退 best 行 → 首个有菜品的行(同 store_name 回退)
-        _item_row = (
-            next((p for p in platforms if isinstance(p, dict) and p.get("role") == "source"), None)
-            or next((p for p in platforms if isinstance(p, dict) and p.get("is_best")), None)
-            or next((p for p in platforms if isinstance(p, dict) and p.get("items")), None)
-        )
-        items = (_item_row or {}).get("items") or []
-    else:
-        derived = _derive_from_results(
-            results, done_params.get("platform_results"), record_status
-        )
-        # pricebot 已把源单菜品塞进 comparison_results[源行].items
-        items = next((r.get("items") or [] for r in results if r.get("is_source")), [])
-    # 失败展示原因(#189): platforms / results 两个派生分支的 status 都可能 failed, 统一在此算
+    derived = _derive_from_results(results, done_params.get("platform_results"))
     fail_reason = (
         _derive_fail_display(
             done_params.get("information"), done_params.get("platform_results")
@@ -591,6 +501,8 @@ def harvest_done(
         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,
@@ -602,7 +514,6 @@ def harvest_done(
         skipped_dish_count=done_params.get("skipped_dish_count"),
         skipped_dish_names=list(done_params.get("skipped_dish_names") or []),
         comparison_results=results,
-        platforms=platforms,
         items=items,
         product_names=_product_names_from_items(items),
         raw_payload=done_params,

app/repositories/guide_video.py

@@ -30,11 +30,7 @@ from app.models.app_config import AppConfig
 from app.models.guide_video import GuideVideoPlay
 from app.repositories import wallet as crud_wallet
 
-SCENES = ("coupon", "comparison")
-_KEY_BY_SCENE = {
-    "coupon": "coupon_guide_video",
-    "comparison": "comparison_guide_video",
-}
+_KEY = "coupon_guide_video"
 
 #: 金币流水 biz_type。客户端收益明细按它显示「新手引导视频奖励」。
 BIZ_TYPE = "guide_video"
@@ -45,7 +41,7 @@ _DEFAULTS: dict[str, Any] = {
     "enabled": True,
     "video_url": None,   # None/空 = 未配片 → 不下发,浮层照旧放广告
     "max_plays": 3,      # 每个账号前 N 次浮层放引导视频
-    "reward_coin": 100,  # 每次固定金币
+    "reward_coin": 120,  # 每次固定金币
 }
 
 _FIELDS = tuple(_DEFAULTS.keys())
@@ -69,28 +65,19 @@ def _merge(raw: Any) -> dict[str, Any]:
     return out
 
 
-def _config_key(scene: str) -> str:
-    if scene not in _KEY_BY_SCENE:
-        raise ValueError(f"unsupported guide video scene: {scene}")
-    return _KEY_BY_SCENE[scene]
-
-
-def get_config(db: Session, scene: str = "coupon") -> dict[str, Any]:
+def get_config(db: Session) -> dict[str, Any]:
     """完整配置 + updated_at(admin 读 / 业务读共用)。"""
-    row = db.get(AppConfig, _config_key(scene))
+    row = db.get(AppConfig, _KEY)
     cfg = _merge(row.value if row is not None else None)
     cfg["updated_at"] = row.updated_at.isoformat() if row is not None and row.updated_at else None
     return cfg
 
 
-def _write(
-    db: Session, value: dict[str, Any], *, scene: str, admin_id: int, commit: bool
-) -> dict[str, Any]:
+def _write(db: Session, value: dict[str, Any], *, admin_id: int, commit: bool) -> dict[str, Any]:
     """整体覆写该行(value 须为完整字段 dict),返回合并后的完整配置(含 updated_at)。"""
-    key = _config_key(scene)
-    row = db.get(AppConfig, key)
+    row = db.get(AppConfig, _KEY)
     if row is None:
-        row = AppConfig(key=key, value=value, updated_by_admin_id=admin_id)
+        row = AppConfig(key=_KEY, value=value, updated_by_admin_id=admin_id)
         db.add(row)
     else:
         row.value = value  # 整体重新赋值,SQLAlchemy 才侦测得到变更
@@ -111,12 +98,11 @@ def update_config(
     enabled: bool | None = None,
     max_plays: int | None = None,
     reward_coin: int | None = None,
-    scene: str = "coupon",
     admin_id: int,
     commit: bool = True,
 ) -> tuple[dict[str, Any], dict[str, Any]]:
     """改开关 / 次数 / 金币(只改传了的字段;视频走 set_video)。返回 (before, after) 供审计。"""
-    row = db.get(AppConfig, _config_key(scene))
+    row = db.get(AppConfig, _KEY)
     before = _merge(row.value if row is not None else None)
     new_value = {k: before[k] for k in _FIELDS}
     if enabled is not None:
@@ -125,52 +111,45 @@ def update_config(
         new_value["max_plays"] = max(0, min(int(max_plays), MAX_PLAYS_LIMIT))
     if reward_coin is not None:
         new_value["reward_coin"] = max(0, min(int(reward_coin), REWARD_COIN_LIMIT))
-    after = _write(db, new_value, scene=scene, admin_id=admin_id, commit=commit)
+    after = _write(db, new_value, admin_id=admin_id, commit=commit)
     return before, after
 
 
 def set_video(
-    db: Session, video_url: str | None, *, scene: str = "coupon",
-    admin_id: int, commit: bool = True
+    db: Session, video_url: str | None, *, admin_id: int, commit: bool = True
 ) -> tuple[dict[str, Any], dict[str, Any]]:
     """设置/清空引导视频地址。返回 (before, after);before['video_url'] 供调用方删旧文件。"""
-    row = db.get(AppConfig, _config_key(scene))
+    row = db.get(AppConfig, _KEY)
     before = _merge(row.value if row is not None else None)
     new_value = {k: before[k] for k in _FIELDS}
     new_value["video_url"] = video_url
-    after = _write(db, new_value, scene=scene, admin_id=admin_id, commit=commit)
+    after = _write(db, new_value, admin_id=admin_id, commit=commit)
     return before, after
 
 
 # ===== 播放计次 =====
 
 
-def used_plays(db: Session, user_id: int, scene: str = "coupon") -> int:
+def used_plays(db: Session, user_id: int) -> int:
     """该账号已用掉的引导视频次数(开播即算,含未发币的)。"""
     return int(
         db.execute(
             select(func.count()).select_from(GuideVideoPlay).where(
-                GuideVideoPlay.user_id == user_id,
-                GuideVideoPlay.scene == scene,
+                GuideVideoPlay.user_id == user_id
             )
         ).scalar_one()
     )
 
 
-def play_stats(db: Session, scene: str = "coupon") -> dict[str, int]:
+def play_stats(db: Session) -> dict[str, int]:
     """全站播放统计(admin 页展示):总播放次数 / 其中已发币次数。"""
     total = int(
-        db.execute(
-            select(func.count()).select_from(GuideVideoPlay).where(
-                GuideVideoPlay.scene == scene
-            )
-        ).scalar_one()
+        db.execute(select(func.count()).select_from(GuideVideoPlay)).scalar_one()
     )
     granted = int(
         db.execute(
             select(func.count()).select_from(GuideVideoPlay).where(
-                GuideVideoPlay.status == "granted",
-                GuideVideoPlay.scene == scene,
+                GuideVideoPlay.status == "granted"
             )
         ).scalar_one()
     )
@@ -189,12 +168,11 @@ def start_play(
       reward_coin  播完/中途关闭都发的固定金币
       seq / remaining  第几次 / 发完这次还剩几次(仅展示与排查用)
     """
-    _config_key(scene)
-    cfg = get_config(db, scene)
+    cfg = get_config(db)
     video_url = (cfg.get("video_url") or "").strip()
     max_plays = int(cfg.get("max_plays") or 0)
     reward_coin = int(cfg.get("reward_coin") or 0)
-    used = used_plays(db, user_id, scene)
+    used = used_plays(db, user_id)
 
     def _miss(used_now: int) -> dict[str, Any]:
         return {
@@ -216,7 +194,7 @@ def start_play(
         scene=scene,
         seq=seq,
         video_url=video_url,
-        coin=reward_coin,
+        coin=0,
         status="playing",
         completed=0,
         started_at=datetime.now(rewards.CN_TZ).replace(tzinfo=None),
@@ -232,7 +210,7 @@ def start_play(
             db.flush()
     except IntegrityError:
         db.rollback()
-        return _miss(used_plays(db, user_id, scene))
+        return _miss(used_plays(db, user_id))
     return {
         "should_play": True,
         "video_url": video_url,
@@ -263,8 +241,7 @@ def grant_play(
     """
     token = (play_token or "").strip()
     # 金币额度以**服务端配置**为准,不信客户端(客户端只上报"播完/关闭")。
-    play = _find_play(db, user_id, token)
-    coin = int(play.coin if play is not None else 0)
+    coin = int(get_config(db).get("reward_coin") or 0)
 
     # 幂等核心:把 status 放进 WHERE 做条件更新(compare-and-set),而不是"先读再判再写"。
     # 「播完」与「✕ 关闭」抢跑、或客户端超时重试时,两个请求会都读到 status='playing',

app/schemas/compare_record.py

@@ -107,13 +107,6 @@ class ComparisonRecordIn(BaseModel):
     # 明细
     items: list[ComparisonItemIn] = Field(default_factory=list)
     comparison_results: list[ComparisonResultIn] = Field(default_factory=list)
-    # 展示模型统一数组(pricebot done.params.platforms 原样透传): 每平台一行、自带
-    # status/is_best/display/display_order,记录页据此直渲染。宽松 list[dict] 存(结构由
-    # pricebot 定,server 只原样落库),前端读它、老记录空时回退 comparison_results。
-    platforms: list[dict] = Field(default_factory=list)
-    # 记录级结局(pricebot 下发): success/below_minimum/store_closed/failed。让"未满起送"不再
-    # 被塌缩成 failed。_derive 优先用它、其次客户端 status、再兜底二态派生。
-    record_status: str | None = None
     # 逐平台结局摘要(含失败平台的细分原因 status: store_not_found/items_not_found/below_minimum/
     # unsupported/...)。来自 done.params.platform_results,客户端透传;落 raw_payload(不单列),
     # admin「卡在哪一步」从这里读。dict{platform_id: {...}} 宽松存(结构由 pricebot 定——是
@@ -184,9 +177,6 @@ class ComparisonRecordOut(BaseModel):
     fail_reason: str | None = None
     items: list = []
     comparison_results: list = []
-    # 展示模型统一数组(每平台一行、自带 status/is_best/display/display_order): 记录页据此
-    # 直渲染, 不再靠 comparison_results + 前端派生。老记录为空 → 前端回退 comparison_results。
-    platforms: list = []
     skipped_dish_names: list = []
     total_ms: int | None = None
     # 「已下单」(店级):该店名在该用户真实下单(source='compare')里出现过即 True。

app/schemas/guide_video.py

@@ -1,15 +1,13 @@
 """新手引导视频(领券等候浮层前 N 次替代广告)的客户端请求/响应契约。"""
 from __future__ import annotations
 
-from typing import Literal
-
 from pydantic import BaseModel, Field
 
 
 class GuideVideoStartIn(BaseModel):
     """开播询问。scene 目前只有 coupon(领券浮层);预留给日后比价等场景。"""
 
-    scene: Literal["coupon", "comparison"] = "coupon"
+    scene: str = Field(default="coupon", max_length=16)
 
 
 class GuideVideoStartOut(BaseModel):

app/services/comparison_llm_backfill.py

@@ -135,7 +135,7 @@ def repair_missing_comparison_llm_costs(
                 select(ComparisonRecord.id, ComparisonRecord.trace_id)
                 .where(
                     *date_conditions,
-                    ComparisonRecord.status.in_(("success", "failed", "cancelled")),
+                    ComparisonRecord.status.in_(("success", "failed")),
                     ComparisonRecord.llm_cost_yuan.is_(None),
                 )
                 .order_by(ComparisonRecord.created_at.desc(), ComparisonRecord.id.desc())

app/services/notification_events.py

@@ -136,17 +136,9 @@ def _push_to_user_devices(db: Session, row: Notification, push_vars: dict[str, s
                 )
                 continue
             try:
-                response = vendor_push.send_notification(
+                vendor_push.send_notification(
                     vendor, dev.push_token, title=title, body=body, extras=extras
                 )
-                if vendor == "huawei" and row.type in {"feedback_reply", "feedback_reward"}:
-                    # 华为反馈推送联调日志：保留厂商返回码/requestId 等排障信息，
-                    # 请求本身的 token、Authorization 和应用密钥不会进入 response。
-                    logger.info(
-                        "huawei feedback push response user_id=%s type=%s "
-                        "notification_id=%s response=%s",
-                        row.user_id, row.type, row.id, response,
-                    )
                 logger.info(
                     "push sent user_id=%s type=%s vendor=%s notification_id=%s",
                     row.user_id, row.type, vendor, row.id,
@@ -155,19 +147,6 @@ def _push_to_user_devices(db: Session, row: Notification, push_vars: dict[str, s
                 logger.warning(
                     "push failed user_id=%s type=%s vendor=%s: %s", row.user_id, row.type, vendor, e
                 )
-            try:
-                vendor_push.send_data_event(
-                    vendor,
-                    dev.push_token,
-                    event=vendor_push.DATA_EVENT_NOTIFICATION_CREATED,
-                    notification_id=str(row.id),
-                )
-            except vendor_push.VendorPushError as e:
-                # 透传只负责前台铃铛实时刷新，失败不能影响通知栏消息或站内消息。
-                logger.warning(
-                    "push data event failed user_id=%s type=%s vendor=%s: %s",
-                    row.user_id, row.type, vendor, e,
-                )
     except Exception:  # noqa: BLE001 — 渲染/查设备等意外失败同样不外抛
         logger.exception("push notification failed user_id=%s type=%s", row.user_id, row.type)
 

tests/test_admin_read.py

@@ -219,115 +219,6 @@ def test_user_reward_stats_can_scope_withdrawals_by_account(
     assert invite.json()["cash_balance_cents"] == 456
 
 
-def test_user_reward_stats_draw_ecpm_uses_all_filtered_impressions(
-    admin_client: TestClient, admin_token: str
-) -> None:
-    """Draw 平均 eCPM 应与收益报表一致，不能只平均成功发奖记录。"""
-    from app.models.ad_ecpm import AdEcpmRecord
-    from app.models.ad_feed_reward import AdFeedRewardRecord
-
-    uid = _seed_user_with_data("13800000024")
-    created_at = datetime(2038, 1, 15, 4, tzinfo=UTC)
-    db = SessionLocal()
-    try:
-        db.add_all(
-            [
-                AdFeedRewardRecord(
-                    client_event_id="reward-stats-granted-high",
-                    ad_session_id="reward-stats-granted-high",
-                    user_id=uid,
-                    reward_date="2038-01-15",
-                    duration_seconds=10,
-                    unit_count=1,
-                    ecpm_raw="9000",
-                    ad_type="draw",
-                    feed_scene="coupon",
-                    app_env="prod",
-                    our_code_id="104098712",
-                    coin=9,
-                    status="granted",
-                    created_at=created_at,
-                ),
-                AdEcpmRecord(
-                    user_id=uid,
-                    ad_type="draw",
-                    feed_scene="coupon",
-                    ad_session_id="reward-stats-impression-low",
-                    app_env="prod",
-                    our_code_id="104098712",
-                    ecpm_raw="1000",
-                    report_date="2038-01-15",
-                    created_at=created_at,
-                ),
-                AdEcpmRecord(
-                    user_id=uid,
-                    ad_type="feed",
-                    feed_scene="coupon",
-                    ad_session_id="reward-stats-impression-mid",
-                    app_env="prod",
-                    our_code_id="104098712",
-                    ecpm_raw="3000",
-                    report_date="2038-01-15",
-                    created_at=created_at,
-                ),
-                # 同用户但不同场景/环境/非业务代码位，均不应进入本次详情筛选。
-                AdEcpmRecord(
-                    user_id=uid,
-                    ad_type="draw",
-                    feed_scene="comparison",
-                    ad_session_id="reward-stats-other-scene",
-                    app_env="prod",
-                    our_code_id="104098712",
-                    ecpm_raw="7000",
-                    report_date="2038-01-15",
-                    created_at=created_at,
-                ),
-                AdEcpmRecord(
-                    user_id=uid,
-                    ad_type="draw",
-                    feed_scene="coupon",
-                    ad_session_id="reward-stats-test-env",
-                    app_env="test",
-                    our_code_id="104127529",
-                    ecpm_raw="8000",
-                    report_date="2038-01-15",
-                    created_at=created_at,
-                ),
-                AdEcpmRecord(
-                    user_id=uid,
-                    ad_type="draw",
-                    feed_scene="coupon",
-                    ad_session_id="reward-stats-non-business",
-                    app_env="prod",
-                    our_code_id="demo-slot",
-                    ecpm_raw="9000",
-                    report_date="2038-01-15",
-                    created_at=created_at,
-                ),
-            ]
-        )
-        db.commit()
-    finally:
-        db.close()
-
-    response = admin_client.get(
-        f"/admin/api/users/{uid}/reward-stats",
-        params={
-            "date_from": "2038-01-15T00:00:00Z",
-            "date_to": "2038-01-15T23:59:59Z",
-            "app_env": "prod",
-            "revenue_scope": "business",
-            "feed_scene": "coupon",
-        },
-        headers=_auth(admin_token),
-    )
-    assert response.status_code == 200, response.text
-    data = response.json()
-    assert data["feed_count"] == 1
-    # 全部真实展示 (1000 + 3000) / 2；不能返回成功发奖记录的 9000。
-    assert data["feed_avg_ecpm"] == 2000.0
-
-
 def test_user_coin_record_sort_accepts_mixed_timezone_datetimes() -> None:
     """线上 PostgreSQL 返回 aware，SQLite/历史转换可能返回 naive，二者必须可混排。"""
     naive = datetime(2038, 1, 1, 8, 0)

tests/test_compare_harvest.py

@@ -263,9 +263,7 @@ def test_trace_finalize_harvests_abort(client) -> None:
     with SessionLocal() as db:                        # 先有 running 行(帧0建的)
         crud.harvest_running(db, trace_id=tid, user_id=None)
     p, _cap = _mock_pricebot({"trace_url": "https://price.shaguabijia.com/traces/fin/"})
-    with p, patch(
-        "app.api.v1.compare.backfill_comparison_llm_cost"
-    ) as backfill:
+    with p:
         r = client.post("/api/v1/trace/finalize",
                         json={"trace_id": tid, "status": "cancelled", "reason": "用户终止"})
     assert r.status_code == 200
@@ -273,7 +271,6 @@ def test_trace_finalize_harvests_abort(client) -> None:
         rec = _get(db, tid)
         assert rec is not None and rec.status == "cancelled"
         assert rec.trace_url.endswith("/fin/")
-        backfill.assert_called_once_with(rec.id, tid)
 
 
 def test_price_step_binds_user_when_authed(client) -> None:

tests/test_comparison_llm_backfill.py

@@ -72,7 +72,6 @@ def test_backfill_retries_then_persists_cost(monkeypatch):
 
 def test_repair_batch_only_targets_terminal_missing_rows(monkeypatch):
     missing_id = _record("llm-repair-missing")
-    cancelled_id = _record("llm-repair-cancelled", status="cancelled")
     running_id = _record("llm-repair-running", status="running")
     calls = [
         {
@@ -97,15 +96,12 @@ def test_repair_batch_only_targets_terminal_missing_rows(monkeypatch):
         )
         assert result["repaired"] >= 1
         assert "llm-repair-missing" in seen
-        assert "llm-repair-cancelled" in seen
         assert "llm-repair-running" not in seen
         with SessionLocal() as db:
             assert db.get(ComparisonRecord, missing_id).llm_cost_yuan is not None
-            assert db.get(ComparisonRecord, cancelled_id).llm_cost_yuan is not None
             assert db.get(ComparisonRecord, running_id).llm_cost_yuan is None
     finally:
         _delete(missing_id)
-        _delete(cancelled_id)
         _delete(running_id)
 
 

tests/test_guide_video.py

@@ -201,7 +201,7 @@ def test_start_loses_seq_race_degrades_to_ad(client, guide_configured, monkeypat
     assert first["should_play"] is True and first["seq"] == 1
 
     # 本次请求读到的是过期计数 → 仍会算出 seq=1
-    monkeypatch.setattr(crud_guide, "used_plays", lambda db, user_id, scene="coupon": 0)
+    monkeypatch.setattr(crud_guide, "used_plays", lambda db, user_id: 0)
     with SessionLocal() as db:
         result = crud_guide.start_play(db, uid)