Merge remote-tracking branch 'origin/main' into fix-mate20PageError

# Conflicts:
#	app/integrations/vendor_push.py

.env.example

@@ -175,3 +175,9 @@ 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/coupon_session_planned_count.py

@@ -1,28 +0,0 @@
-"""add planned coupon count to coupon session
-
-Revision ID: coupon_session_planned_count
-Revises: 6d2309208549
-"""
-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 = "6d2309208549"
-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")

alembic/versions/guide_video_scene_unique.py

@@ -0,0 +1,31 @@
+"""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/coupon_data.py

@@ -23,7 +23,6 @@ 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:
@@ -167,10 +166,6 @@ 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,
@@ -187,8 +182,8 @@ def _session_to_row(
         "app_env": r.app_env,
         "started_at": r.started_at,
         "claimed_count": r.claimed_count,
-        "point_success_count": succeeded if point_total > 0 else None,
-        "point_total_count": point_total if point_total > 0 else None,
+        "point_success_count": point_stats["succeeded"] if point_stats else None,
+        "point_total_count": point_stats["tried"] if point_stats else None,
         "trace_url": r.trace_url,
         "ad_revenue_yuan": ad_revenue_yuan,
     }
@@ -207,7 +202,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_(_SESSION_POINT_TOTAL),
+            CouponClaimEvent.status.in_(_SLOT_TRIED),
         )
         .group_by(CouponClaimEvent.trace_id)
     ).all()
@@ -406,13 +401,11 @@ def coupon_user_records(db: Session, *, user_id: int, limit: int = 100) -> dict:
         select(func.count()).select_from(CouponSession).where(CouponSession.user_id == user_id)
     ).scalar_one()
     rev_map = crud_ecpm.revenue_yuan_by_trace(db, [r.trace_id for r in rows])
-    point_stats_map = _point_scores_by_trace(db, [r.trace_id for r in rows])
     return {
         "items": [
             _session_to_row(
                 r,
                 ad_revenue_yuan=rev_map.get(r.trace_id, 0.0),
-                point_stats=point_stats_map.get(r.trace_id),
             )
             for r in rows
         ],

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
+from typing import Annotated, Literal
 
 from fastapi import APIRouter, Depends, File, HTTPException, Request, UploadFile
 
@@ -27,14 +27,21 @@ router = APIRouter(
 )
 
 
-def _out(db: AdminDb) -> GuideVideoConfigOut:
+GuideScene = Literal["coupon", "comparison"]
+
+
+def _out(db: AdminDb, scene: GuideScene) -> GuideVideoConfigOut:
     """配置 + 播放统计合成响应(四个写接口都以最新状态返回,前端一次同步到位)。"""
-    return GuideVideoConfigOut(**guide_video.get_config(db), **guide_video.play_stats(db))
+    return GuideVideoConfigOut(
+        scene=scene,
+        **guide_video.get_config(db, scene),
+        **guide_video.play_stats(db, scene),
+    )
 
 
 @router.get("", response_model=GuideVideoConfigOut, summary="新手引导视频配置(领券浮层)")
-def get_config(db: AdminDb) -> GuideVideoConfigOut:
-    return _out(db)
+def get_config(db: AdminDb, scene: GuideScene = "coupon") -> GuideVideoConfigOut:
+    return _out(db, scene)
 
 
 @router.patch("", response_model=GuideVideoConfigOut, summary="改开关/次数/金币(带审计)")
@@ -43,21 +50,23 @@ 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={"before": before, "after": after}, ip=get_client_ip(request), commit=False,
+        detail={"scene": scene, "before": before, "after": after}, ip=get_client_ip(request), commit=False,
     )
     db.commit()
-    return _out(db)
+    return _out(db, scene)
 
 
 @router.post("/video", response_model=GuideVideoConfigOut, summary="上传新手引导视频(MP4,带审计)")
@@ -65,23 +74,26 @@ async def upload_video(
     request: Request,
     admin: Annotated[AdminUser, Depends(require_role("operator"))],
     db: AdminDb,
-    file: UploadFile = File(...),
+    file: Annotated[UploadFile, File()],
+    scene: GuideScene = "coupon",
 ) -> 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, admin_id=admin.id, commit=False)
+    before, after = guide_video.set_video(
+        db, url, scene=scene, admin_id=admin.id, commit=False
+    )
     write_audit(
         db, admin, action="guide_video.set_video", target_type="guide_video", target_id=None,
-        detail={"before": before.get("video_url"), "after": url, "bytes": len(data)},
+        detail={"scene": scene, "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)
+    return _out(db, scene)
 
 
 @router.delete("/video", response_model=GuideVideoConfigOut, summary="移除新手引导视频(带审计)")
@@ -89,13 +101,16 @@ 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, admin_id=admin.id, commit=False)
+    before, after = guide_video.set_video(
+        db, None, scene=scene, admin_id=admin.id, commit=False
+    )
     write_audit(
         db, admin, action="guide_video.delete_video", target_type="guide_video", target_id=None,
-        detail={"before": before.get("video_url")}, ip=get_client_ip(request), commit=False,
+        detail={"scene": scene, "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)
+    return _out(db, scene)

app/admin/schemas/coupon_data.py

@@ -79,15 +79,10 @@ class CouponDataRow(BaseModel):
     started_at: datetime = Field(..., description="发起时刻(明细「时间」列)")
     claimed_count: int | None = None
     point_success_count: int | None = Field(
-        None,
-        description="本次成功单券数(success+already_claimed);有计划数但尚无成功事件时为 0",
+        None, description="本次成功单券数(success+already_claimed);无逐券事件为空"
     )
     point_total_count: int | None = Field(
-        None,
-        description=(
-            "本次计划单券数(max(逐券事件数,计划数),事件含 skipped);"
-            "无计划数且无逐券事件时为空"
-        ),
+        None, description="本次尝试单券数(success+already_claimed+failed,不含 skipped);无逐券事件为空"
     )
     trace_url: str | None = Field(None, description="pricebot 公网 trace 链接(仅 completed 有);admin 渲染可点链接,无则显示可复制 trace_id")
     ad_revenue_yuan: float = Field(

app/admin/schemas/guide_video.py

@@ -7,6 +7,7 @@ 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/api/v1/coupon.py

@@ -93,13 +93,6 @@ 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:
@@ -183,18 +176,6 @@ 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/core/config.py

@@ -88,6 +88,8 @@ 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

@@ -22,7 +22,7 @@ import uuid
 from collections.abc import Callable
 from dataclasses import dataclass
 from typing import Any
-from urllib.parse import quote
+from urllib.parse import quote, urlsplit
 
 import httpx
 
@@ -31,6 +31,7 @@ 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 → 中文名(测试/配置状态接口展示用)
@@ -69,6 +70,18 @@ class _CachedToken:
 
 _token_cache: dict[str, _CachedToken] = {}
 
+_LOG_SUMMARY_MAX_CHARS = 1500
+_LOG_STRING_MAX_CHARS = 200
+_SENSITIVE_LOG_KEYS = {
+    "accesstoken",
+    "appkey",
+    "authtoken",
+    "authorization",
+    "clientsecret",
+    "mastersecret",
+    "sign",
+}
+
 
 def normalize_vendor(push_vendor: str | None) -> str | None:
     if not push_vendor:
@@ -115,8 +128,16 @@ def send_notification(
 
     if mock:
         logger.info(
-            "[mock push] vendor=%s token=%s... title=%s body=%s extras=%s",
-            vendor, token[:12], title, body, extras,
+            "vendor push mock vendor=%s request=%s",
+            vendor,
+            _log_summary(
+                {
+                    "push_token": token,
+                    "title": title,
+                    "body": body,
+                    "extras": extras,
+                }
+            ),
         )
         return {
             "mock": True,
@@ -133,7 +154,28 @@ def send_notification(
         "xiaomi": _send_xiaomi,
         "oppo": _send_oppo,
     }
-    return dispatch[vendor](token, title, body, extras)
+    started = time.perf_counter()
+    try:
+        result = dispatch[vendor](token, title, body, extras)
+    except Exception:
+        logger.exception(
+            "vendor push dispatch failed vendor=%s token=%s notification_type=%s "
+            "elapsed_ms=%.1f",
+            vendor,
+            token,
+            extras.get("type", ""),
+            _elapsed_ms(started),
+        )
+        raise
+    logger.info(
+        "vendor push dispatch succeeded vendor=%s token=%s notification_type=%s "
+        "elapsed_ms=%.1f",
+        vendor,
+        token,
+        extras.get("type", ""),
+        _elapsed_ms(started),
+    )
+    return result
 
 
 def send_accessibility_disabled(
@@ -153,19 +195,169 @@ 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")
     return value
 
 
+def _elapsed_ms(started: float) -> float:
+    return (time.perf_counter() - started) * 1000
+
+
+def _redacted_value(value: Any) -> str:
+    raw = json.dumps(value, ensure_ascii=False, sort_keys=True, default=str)
+    digest = hashlib.sha256(raw.encode()).hexdigest()[:10]
+    return f"<redacted len={len(raw)} sha256={digest}>"
+
+
+def _sanitize_for_log(value: Any, *, key: str = "", depth: int = 0) -> Any:
+    """生成有排障价值、但不暴露服务端鉴权凭据的紧凑日志摘要。"""
+    normalized_key = "".join(char for char in key.lower() if char.isalnum())
+    if normalized_key in _SENSITIVE_LOG_KEYS or normalized_key.endswith("secret"):
+        return _redacted_value(value)
+    if depth >= 10:
+        return f"<{type(value).__name__}>"
+    if isinstance(value, dict):
+        items = list(value.items())
+        sanitized = {
+            str(item_key): _sanitize_for_log(item_value, key=str(item_key), depth=depth + 1)
+            for item_key, item_value in items[:30]
+        }
+        if len(items) > 30:
+            sanitized["<omitted_keys>"] = len(items) - 30
+        return sanitized
+    if isinstance(value, (list, tuple)):
+        items = list(value)
+        sanitized = [_sanitize_for_log(item, key=key, depth=depth + 1) for item in items[:20]]
+        if len(items) > 20:
+            sanitized.append(f"<omitted_items={len(items) - 20}>")
+        return sanitized
+    if isinstance(value, str):
+        if value.lstrip().startswith(("{", "[")):
+            try:
+                decoded = json.loads(value)
+            except ValueError:
+                pass
+            else:
+                return _sanitize_for_log(decoded, key=key, depth=depth + 1)
+        if len(value) > _LOG_STRING_MAX_CHARS:
+            return f"{value[:_LOG_STRING_MAX_CHARS]}…<len={len(value)}>"
+    return value
+
+
+def _log_summary(value: Any) -> str:
+    rendered = json.dumps(
+        _sanitize_for_log(value),
+        ensure_ascii=False,
+        sort_keys=True,
+        separators=(",", ":"),
+        default=str,
+    )
+    if len(rendered) > _LOG_SUMMARY_MAX_CHARS:
+        return f"{rendered[:_LOG_SUMMARY_MAX_CHARS]}…<len={len(rendered)}>"
+    return rendered
+
+
+def _raw_log_summary(value: Any) -> str:
+    """厂商响应摘要：保留原始字段和值，仅限制单条日志长度。"""
+    rendered = json.dumps(
+        value,
+        ensure_ascii=False,
+        sort_keys=True,
+        separators=(",", ":"),
+        default=str,
+    )
+    if len(rendered) > _LOG_SUMMARY_MAX_CHARS:
+        return f"{rendered[:_LOG_SUMMARY_MAX_CHARS]}…<len={len(rendered)}>"
+    return rendered
+
+
+def _endpoint_for_log(url: str) -> str:
+    parsed = urlsplit(url)
+    return f"{parsed.scheme}://{parsed.netloc}{parsed.path}"
+
+
+def _request_summary(kwargs: dict[str, Any]) -> str:
+    summary: dict[str, Any] = {}
+    if "json" in kwargs:
+        summary["body_type"] = "json"
+        summary["request_payload"] = kwargs["json"]
+    elif "data" in kwargs:
+        summary["body_type"] = "form"
+        summary["request_payload"] = kwargs["data"]
+    if kwargs.get("params"):
+        summary["params"] = kwargs["params"]
+    if kwargs.get("headers"):
+        summary["headers"] = kwargs["headers"]
+    return _log_summary(summary)
+
+
+def _response_summary(resp: Any, parsed: Any | None = None) -> str:
+    if parsed is None:
+        try:
+            parsed = resp.json()
+        except ValueError:
+            pass
+    if parsed is not None:
+        return _raw_log_summary(parsed)
+    return _raw_log_summary(getattr(resp, "text", ""))
+
+
 def _request_json(
     method: str,
     url: str,
     *,
+    vendor: str,
+    operation: str,
     expected_status: tuple[int, ...] = (200,),
     **kwargs: Any,
 ) -> dict[str, Any]:
+    started = time.perf_counter()
+    request_summary = _request_summary(kwargs)
     try:
         resp = httpx.request(
             method,
@@ -174,41 +366,82 @@ def _request_json(
             **kwargs,
         )
     except httpx.HTTPError as e:
+        logger.error(
+            "vendor push http completed vendor=%s operation=%s method=%s endpoint=%s "
+            "outcome=network_error request=%s response=%s elapsed_ms=%.1f error=%s",
+            vendor,
+            operation,
+            method,
+            _endpoint_for_log(url),
+            request_summary,
+            "<no_response>",
+            _elapsed_ms(started),
+            type(e).__name__,
+        )
         raise VendorPushError(f"push http error: {e}") from e
 
     if resp.status_code not in expected_status:
-        logger.error("vendor push http failed url=%s http=%s body=%s", url, resp.status_code, resp.text[:500])
+        logger.error(
+            "vendor push http completed vendor=%s operation=%s method=%s endpoint=%s "
+            "outcome=http_error http_status=%s request=%s response=%s elapsed_ms=%.1f",
+            vendor,
+            operation,
+            method,
+            _endpoint_for_log(url),
+            resp.status_code,
+            request_summary,
+            _response_summary(resp),
+            _elapsed_ms(started),
+        )
         raise VendorPushError(f"push http {resp.status_code}")
     try:
-        return resp.json()
+        data = resp.json()
     except ValueError as e:
+        logger.error(
+            "vendor push http completed vendor=%s operation=%s method=%s endpoint=%s "
+            "outcome=invalid_json http_status=%s request=%s response=%s elapsed_ms=%.1f",
+            vendor,
+            operation,
+            method,
+            _endpoint_for_log(url),
+            resp.status_code,
+            request_summary,
+            _response_summary(resp),
+            _elapsed_ms(started),
+        )
         raise VendorPushError(f"push invalid json: {resp.text[:200]}") from e
+    logger.info(
+        "vendor push http completed vendor=%s operation=%s method=%s endpoint=%s "
+        "outcome=success http_status=%s request=%s response=%s elapsed_ms=%.1f",
+        vendor,
+        operation,
+        method,
+        _endpoint_for_log(url),
+        resp.status_code,
+        request_summary,
+        _response_summary(resp, data),
+        _elapsed_ms(started),
+    )
+    return data
 
 
 def _request_form(
     method: str,
     url: str,
     *,
+    vendor: str,
+    operation: str,
     expected_status: tuple[int, ...] = (200,),
     **kwargs: Any,
 ) -> dict[str, Any]:
-    try:
-        resp = httpx.request(
-            method,
-            url,
-            timeout=settings.PUSH_REQUEST_TIMEOUT_SEC,
-            **kwargs,
-        )
-    except httpx.HTTPError as e:
-        raise VendorPushError(f"push http error: {e}") from e
-
-    if resp.status_code not in expected_status:
-        logger.error("vendor push http failed url=%s http=%s body=%s", url, resp.status_code, resp.text[:500])
-        raise VendorPushError(f"push http {resp.status_code}")
-    try:
-        return resp.json()
-    except ValueError as e:
-        raise VendorPushError(f"push invalid json: {resp.text[:200]}") from e
+    return _request_json(
+        method,
+        url,
+        vendor=vendor,
+        operation=operation,
+        expected_status=expected_status,
+        **kwargs,
+    )
 
 
 def _cache_get(key: str) -> str | None:
@@ -234,6 +467,8 @@ def _honor_access_token() -> str:
     data = _request_form(
         "POST",
         settings.HONOR_PUSH_TOKEN_ENDPOINT,
+        vendor="honor",
+        operation="authenticate",
         data={
             "grant_type": "client_credentials",
             "client_id": client_id,
@@ -243,16 +478,22 @@ def _honor_access_token() -> str:
     )
     token = data.get("access_token")
     if not token:
-        raise VendorPushError(f"honor auth failed: {data}")
+        raise VendorPushError(f"honor auth failed: {_raw_log_summary(data)}")
     return _cache_put(cache_key, str(token), data.get("expires_in"))
 
 
 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": {
@@ -261,7 +502,7 @@ def _send_honor(token: str, title: str, body: str, extras: dict[str, str]) -> di
             "notification": {
                 "title": title,
                 "body": body,
-                "clickAction": {"type": 3},
+                "clickAction": click_action,
                 "importance": "NORMAL",
             },
         },
@@ -270,6 +511,8 @@ def _send_honor(token: str, title: str, body: str, extras: dict[str, str]) -> di
     data = _request_json(
         "POST",
         settings.HONOR_PUSH_SEND_ENDPOINT_TEMPLATE.format(app_id=app_id),
+        vendor="honor",
+        operation="send_notification",
         json=payload,
         headers={
             "Content-Type": "application/json; charset=UTF-8",
@@ -279,7 +522,28 @@ def _send_honor(token: str, title: str, body: str, extras: dict[str, str]) -> di
     )
     code = data.get("code")
     if code is not None and int(code) != 200:
-        raise VendorPushError(f"honor push failed: {data}")
+        raise VendorPushError(f"honor push failed: {_raw_log_summary(data)}")
+    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),
+        vendor="honor",
+        operation="send_data_event",
+        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: {_raw_log_summary(data)}")
     return data
 
 
@@ -294,6 +558,8 @@ def _huawei_access_token() -> str:
     data = _request_form(
         "POST",
         settings.HUAWEI_PUSH_TOKEN_ENDPOINT,
+        vendor="huawei",
+        operation="authenticate",
         data={
             "grant_type": "client_credentials",
             "client_id": app_id,
@@ -303,7 +569,7 @@ def _huawei_access_token() -> str:
     )
     token = data.get("access_token")
     if not token:
-        raise VendorPushError(f"huawei auth failed: {data}")
+        raise VendorPushError(f"huawei auth failed: {_raw_log_summary(data)}")
     return _cache_put(cache_key, str(token), data.get("expires_in"))
 
 
@@ -312,18 +578,24 @@ 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": {"type": 3},
+                    "click_action": click_action,
                     "importance": "NORMAL",
                 },
             },
@@ -333,6 +605,8 @@ def _send_huawei(token: str, title: str, body: str, extras: dict[str, str]) -> d
     data = _request_json(
         "POST",
         settings.HUAWEI_PUSH_SEND_ENDPOINT_TEMPLATE.format(app_id=app_id),
+        vendor="huawei",
+        operation="send_notification",
         json=payload,
         headers={
             "Content-Type": "application/json; charset=UTF-8",
@@ -340,7 +614,32 @@ def _send_huawei(token: str, title: str, body: str, extras: dict[str, str]) -> d
         },
     )
     if str(data.get("code", "")) != "80000000":
-        raise VendorPushError(f"huawei push failed: {data}")
+        raise VendorPushError(f"huawei push failed: {_raw_log_summary(data)}")
+    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),
+        vendor="huawei",
+        operation="send_data_event",
+        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: {_raw_log_summary(data)}")
     return data
 
 
@@ -357,6 +656,8 @@ def _vivo_auth_token() -> str:
     data = _request_json(
         "POST",
         settings.VIVO_PUSH_AUTH_ENDPOINT,
+        vendor="vivo",
+        operation="authenticate",
         json={
             "appId": app_id,
             "appKey": app_key,
@@ -366,10 +667,10 @@ def _vivo_auth_token() -> str:
         headers={"Content-Type": "application/json"},
     )
     if int(data.get("result", -1)) != 0:
-        raise VendorPushError(f"vivo auth failed: {data}")
+        raise VendorPushError(f"vivo auth failed: {_raw_log_summary(data)}")
     token = data.get("authToken")
     if not token:
-        raise VendorPushError(f"vivo auth missing authToken: {data}")
+        raise VendorPushError(f"vivo auth missing authToken: {_raw_log_summary(data)}")
     return _cache_put(cache_key, str(token), 24 * 3600)
 
 
@@ -385,15 +686,22 @@ 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,
     }
-    # 点击落地:消息中心推送(带 notificationId)→ skipType=4 + skipContent=intent uri,由 vivo
-    # 系统直启 MainActivity 并携带 S. extras(与小米 notify_effect=2 同机制)。不依赖客户端
-    # VivoPushReceiver.onNotificationMessageClicked 里的后台 startActivity——Android 10+ BAL
-    # 会静默拦掉,receiver 路径仅作兜底。无 notificationId 的召回类保持 skipType=1 仅打开首页。
+    # vivo Push SDK 480+ 已不再回调 skipType=3；必须使用 skipType=4 的完整 Intent URI。
+    # URI 同时包含 data/scheme、显式 component 和 S. extras，OriginOS 才会把业务参数
+    # 原样交给 MainActivity（仅靠隐式 deeplink 会打开 App，但可能剥掉 extras）。
     if extras.get("notificationId"):
         payload["skipType"] = 4
-        payload["skipContent"] = _click_intent_uri(extras)
+        payload["skipContent"] = _vivo_click_intent_uri(extras)
     else:
         payload["skipType"] = 1
     if settings.VIVO_PUSH_CATEGORY:
@@ -401,14 +709,37 @@ def _send_vivo(token: str, title: str, body: str, extras: dict[str, str]) -> dic
     data = _request_json(
         "POST",
         settings.VIVO_PUSH_SEND_ENDPOINT,
+        vendor="vivo",
+        operation="send_notification",
         json=payload,
         headers={
             "Content-Type": "application/json",
             "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,
+            vendor="vivo",
+            operation="send_notification",
+            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}")
+        raise VendorPushError(f"vivo push failed: {_raw_log_summary(data)}")
     return data
 
 
@@ -446,14 +777,40 @@ def _send_xiaomi(token: str, title: str, body: str, extras: dict[str, str]) -> d
     data = _request_form(
         "POST",
         settings.XIAOMI_PUSH_SEND_ENDPOINT,
+        vendor="xiaomi",
+        operation="send_notification",
         data=form,
         headers={"Authorization": f"key={app_secret}"},
     )
     code = data.get("code")
     if code not in (0, "0", None):
-        raise VendorPushError(f"xiaomi push failed: {data}")
+        raise VendorPushError(f"xiaomi push failed: {_raw_log_summary(data)}")
     if str(data.get("result", "ok")).lower() not in ("ok", "success"):
-        raise VendorPushError(f"xiaomi push failed: {data}")
+        raise VendorPushError(f"xiaomi push failed: {_raw_log_summary(data)}")
+    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,
+        vendor="xiaomi",
+        operation="send_data_event",
+        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: {_raw_log_summary(data)}")
+    if str(data.get("result", "ok")).lower() not in ("ok", "success"):
+        raise VendorPushError(f"xiaomi data push failed: {_raw_log_summary(data)}")
     return data
 
 
@@ -490,6 +847,24 @@ 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
@@ -522,6 +897,8 @@ def _oppo_auth_token() -> str:
     data = _request_form(
         "POST",
         settings.OPPO_PUSH_AUTH_ENDPOINT,
+        vendor="oppo",
+        operation="authenticate",
         data={
             "app_key": app_key,
             "timestamp": timestamp,
@@ -530,10 +907,10 @@ def _oppo_auth_token() -> str:
         headers={"Content-Type": "application/x-www-form-urlencoded"},
     )
     if int(data.get("code", -1)) != 0:
-        raise VendorPushError(f"oppo auth failed: {data}")
+        raise VendorPushError(f"oppo auth failed: {_raw_log_summary(data)}")
     token = (data.get("data") or {}).get("auth_token") or data.get("auth_token")
     if not token:
-        raise VendorPushError(f"oppo auth missing auth_token: {data}")
+        raise VendorPushError(f"oppo auth missing auth_token: {_raw_log_summary(data)}")
     return _cache_put(cache_key, str(token), 24 * 3600)
 
 
@@ -548,6 +925,11 @@ 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 同款坑)。
@@ -573,6 +955,8 @@ def _send_oppo(token: str, title: str, body: str, extras: dict[str, str]) -> dic
     data = _request_form(
         "POST",
         settings.OPPO_PUSH_SEND_ENDPOINT,
+        vendor="oppo",
+        operation="send_notification",
         data={
             "auth_token": auth_token,
             "message": json.dumps(message, ensure_ascii=False),
@@ -580,5 +964,5 @@ def _send_oppo(token: str, title: str, body: str, extras: dict[str, str]) -> dic
         headers={"Content-Type": "application/x-www-form-urlencoded"},
     )
     if int(data.get("code", -1)) != 0:
-        raise VendorPushError(f"oppo push failed: {data}")
+        raise VendorPushError(f"oppo push failed: {_raw_log_summary(data)}")
     return data

app/models/coupon_state.py

@@ -277,9 +277,6 @@ 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/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_seq", "user_id", "seq", unique=True),
+        Index("uq_guide_video_play_user_scene_seq", "user_id", "scene", "seq", unique=True),
     )
 
     id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)

app/repositories/coupon_state.py

@@ -157,22 +157,6 @@ 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/repositories/guide_video.py

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

app/schemas/guide_video.py

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

app/services/notification_events.py

@@ -136,9 +136,17 @@ def _push_to_user_devices(db: Session, row: Notification, push_vars: dict[str, s
                 )
                 continue
             try:
-                vendor_push.send_notification(
+                response = 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,
@@ -147,6 +155,19 @@ 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_coupon_point_score.py

@@ -10,16 +10,14 @@ from app.admin.repositories.coupon_data import (
     _point_scores_by_trace,
     coupon_data_report,
     coupon_point_details,
-    coupon_user_records,
 )
 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:
@@ -39,7 +37,7 @@ def test_point_scores_by_trace() -> None:
 
         stats = _point_scores_by_trace(db, [trace])[trace]
         assert stats["succeeded"] == 2
-        assert stats["tried"] == 4
+        assert stats["tried"] == 3
         details = coupon_point_details(db, trace_id=trace)
         assert [item["status"] for item in details] == [
             "success", "already_claimed", "failed", "skipped"
@@ -50,8 +48,8 @@ def test_point_scores_by_trace() -> None:
         db.close()
 
 
-def test_skipped_detail_creates_zero_score() -> None:
-    """仅有 skipped 时也应显示 0/1，而不是把计划券静默成空值。"""
+def test_skipped_detail_does_not_create_a_score() -> None:
+    """仅有 skipped 时按需明细仍可查到，但列表没有虚假的 0/0 分数。"""
     db = SessionLocal()
     trace = "point-score-skipped"
     try:
@@ -65,7 +63,7 @@ def test_skipped_detail_creates_zero_score() -> None:
         db.flush()
 
         scores = _point_scores_by_trace(db, [trace, "missing-trace"])
-        assert scores[trace] == {"succeeded": 0, "tried": 1}
+        assert trace not in scores
         assert "missing-trace" not in scores
         assert coupon_point_details(db, trace_id=trace)[0]["status"] == "skipped"
     finally:
@@ -116,126 +114,6 @@ 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_user_records_include_actual_successes_with_planned_count() -> None:
-    """用户领券记录也必须加载逐券统计，不能把有成功事件的场次显示成 0/N。"""
-    db = SessionLocal()
-    trace = "point-score-user-records"
-    user_id = 987656
-    report_date = date(2020, 1, 8)
-    try:
-        db.add_all([
-            CouponSession(
-                trace_id=trace,
-                device_id="planned-user-records-device",
-                user_id=user_id,
-                status="abandoned",
-                app_env="prod",
-                platforms=[],
-                planned_coupon_count=8,
-                started_at=datetime(2020, 1, 8, tzinfo=UTC),
-                started_date=report_date,
-            ),
-            CouponClaimEvent(
-                trace_id=trace,
-                device_id="planned-user-records-device",
-                coupon_id="mt-user-records-success",
-                claim_date=report_date,
-                status="success",
-            ),
-        ])
-        db.flush()
-
-        records = coupon_user_records(db, user_id=user_id)
-        row = next(item for item in records["items"] if item["trace_id"] == trace)
-        assert row["point_success_count"] == 1
-        assert row["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_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: 0)
+    monkeypatch.setattr(crud_guide, "used_plays", lambda db, user_id, scene="coupon": 0)
     with SessionLocal() as db:
         result = crud_guide.start_play(db, uid)
 

tests/test_push_center.py

@@ -7,7 +7,9 @@ mock/真发/变量覆盖/站内联动/设备反查/参数校验。厂商 HTTP
 from __future__ import annotations
 
 import json
+import logging
 
+import httpx
 import pytest
 from fastapi.testclient import TestClient
 
@@ -110,6 +112,99 @@ def test_huawei_non_success_code_raises(monkeypatch) -> None:
         vendor_push.send_notification("huawei", "bad-token", title="t", body="b")
 
 
+def test_vendor_http_logs_device_token_raw_response_and_elapsed(monkeypatch, caplog) -> None:
+    vendor_push._token_cache.clear()
+
+    def _fake_request(method, url, **kwargs):  # noqa: ANN001
+        if url == vendor_push.settings.HUAWEI_PUSH_TOKEN_ENDPOINT:
+            return _Resp({"access_token": "raw-access-token", "expires_in": 3600})
+        return _Resp({"code": "80000000", "msg": "Success", "requestId": "vendor-request-1"})
+
+    monkeypatch.setattr(vendor_push.settings, "HUAWEI_PUSH_APP_ID", "10086001")
+    monkeypatch.setattr(vendor_push.settings, "HUAWEI_PUSH_APP_SECRET", "raw-app-secret")
+    monkeypatch.setattr(vendor_push.httpx, "request", _fake_request)
+
+    with caplog.at_level(logging.INFO, logger=vendor_push.logger.name):
+        vendor_push.send_notification(
+            "huawei",
+            "raw-device-token",
+            title="需要进入日志的标题",
+            body="需要进入日志的正文",
+            extras={"type": "withdraw_success", "notificationId": "90001"},
+        )
+
+    messages = "\n".join(record.getMessage() for record in caplog.records)
+    assert "operation=authenticate" in messages
+    assert "operation=send_notification" in messages
+    assert "request=" in messages
+    assert "response=" in messages
+    assert "elapsed_ms=" in messages
+    assert "vendor-request-1" in messages
+    assert "withdraw_success" in messages
+    assert "raw-access-token" in messages
+    assert "raw-device-token" in messages
+    assert "需要进入日志的标题" in messages
+    assert "需要进入日志的正文" in messages
+    assert "<redacted" in messages
+    assert "raw-app-secret" not in messages
+    send_http_log = next(
+        record.getMessage()
+        for record in caplog.records
+        if "operation=send_notification" in record.getMessage()
+        and "vendor push http completed" in record.getMessage()
+    )
+    assert "raw-device-token" in send_http_log
+    assert "raw-access-token" not in send_http_log  # 请求头中的厂商鉴权 token 仍脱敏
+
+
+def test_vendor_http_network_error_logs_request_context(monkeypatch, caplog) -> None:
+    def _fake_request(*args, **kwargs):  # noqa: ANN002, ANN003
+        raise httpx.ConnectError("connection refused")
+
+    monkeypatch.setattr(vendor_push.settings, "XIAOMI_PUSH_APP_SECRET", "raw-xiaomi-secret")
+    monkeypatch.setattr(vendor_push.httpx, "request", _fake_request)
+
+    with (
+        caplog.at_level(logging.ERROR, logger=vendor_push.logger.name),
+        pytest.raises(vendor_push.VendorPushError, match="push http error"),
+    ):
+        vendor_push.send_notification(
+            "xiaomi",
+            "raw-xiaomi-token",
+            title="title",
+            body="body",
+            extras={"type": "push_test"},
+        )
+
+    messages = "\n".join(record.getMessage() for record in caplog.records)
+    assert "outcome=network_error" in messages
+    assert "response=<no_response>" in messages
+    assert "operation=send_notification" in messages
+    assert "elapsed_ms=" in messages
+    assert "raw-xiaomi-secret" not in messages
+    assert "raw-xiaomi-token" in messages
+
+
+@pytest.mark.parametrize(
+    ("request_kwargs", "device_token"),
+    [
+        ({"json": {"token": ["honor-device-token"]}}, "honor-device-token"),
+        ({"json": {"message": {"token": ["huawei-device-token"]}}}, "huawei-device-token"),
+        ({"json": {"regId": "vivo-device-token"}}, "vivo-device-token"),
+        ({"data": {"registration_id": "xiaomi-device-token"}}, "xiaomi-device-token"),
+        (
+            {"data": {"message": '{"target_value":"oppo-device-token"}'}},
+            "oppo-device-token",
+        ),
+    ],
+)
+def test_all_vendor_device_token_fields_remain_visible_in_request_summary(
+    request_kwargs, device_token
+) -> None:
+    summary = vendor_push._request_summary(request_kwargs)
+    assert device_token in summary
+
+
 def test_vendor_aliases_normalize() -> None:
     assert vendor_push.normalize_vendor("华为") == "huawei"
     assert vendor_push.normalize_vendor("HMS") == "huawei"