OPPO 推送补新消息分类(channel_id/category/notify_level)

2024-11-20 后创建的 OPPO 应用发送通知必须携带 category(新规),
配置项 OPPO_PUSH_CHANNEL_ID / OPPO_PUSH_CATEGORY / OPPO_PUSH_NOTIFY_LEVEL(0=不传),
均留空时 payload 与原先完全一致。凭据已到位:荣耀/小米(含 channel 154219)/OPPO/vivo
四家配齐(.env,不入库),仅剩华为待补。

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

.env.example

@@ -27,9 +27,57 @@ JG_PRIVATE_KEY_PATH=./secrets/jverify_rsa_private.pem
 JG_VERIFY_ENDPOINT=https://api.verification.jpush.cn/v1/web/loginTokenVerify
 JG_REQUEST_TIMEOUT_SEC=15
 
-# ===== 无障碍保护存活监控(pull 后置检测;本期不接推送)=====
+# ===== 厂商直推(无障碍保护存活告警 + 消息中心 13 类通知)=====
+# 敏感密钥只放 .env / 服务器环境变量,不要提交到 git。
+# 各厂商配置状态可随时 GET /api/v1/push/vendors 查看(缺哪些键一目了然)。
+ANDROID_PACKAGE_NAME=com.jishisongfu.shaguabijia
+PUSH_REQUEST_TIMEOUT_SEC=15
+PUSH_TIME_TO_LIVE_SEC=86400
+
+HONOR_PUSH_APP_ID=
+HONOR_PUSH_CLIENT_ID=
+HONOR_PUSH_CLIENT_SECRET=
+HONOR_PUSH_TOKEN_ENDPOINT=https://iam.developer.honor.com/auth/token
+HONOR_PUSH_SEND_ENDPOINT_TEMPLATE=https://push-api.cloud.honor.com/api/v1/{app_id}/sendMessage
+
+# 华为 Push Kit:AGC 控制台 → 项目设置 → 常规 → 应用,AppId + AppSecret
+HUAWEI_PUSH_APP_ID=
+HUAWEI_PUSH_APP_SECRET=
+HUAWEI_PUSH_TOKEN_ENDPOINT=https://oauth-login.cloud.huawei.com/oauth2/v3/token
+HUAWEI_PUSH_SEND_ENDPOINT_TEMPLATE=https://push-api.cloud.huawei.com/v1/{app_id}/messages:send
+
+VIVO_PUSH_APP_ID=
+VIVO_PUSH_APP_KEY=
+VIVO_PUSH_APP_SECRET=
+VIVO_PUSH_AUTH_ENDPOINT=https://api-push.vivo.com.cn/message/auth
+VIVO_PUSH_SEND_ENDPOINT=https://api-push.vivo.com.cn/message/send
+# vivo 未上架测试时可用 push_mode=1; 上架正式推送改为 0。
+VIVO_PUSH_MODE=1
+VIVO_PUSH_NOTIFY_TYPE=4
+VIVO_PUSH_CATEGORY=DEVICE_REMINDER
+
+XIAOMI_PUSH_APP_SECRET=
+XIAOMI_PUSH_SEND_ENDPOINT=https://api.xmpush.xiaomi.com/v3/message/regid
+XIAOMI_PUSH_CHANNEL_ID=
+XIAOMI_PUSH_TEMPLATE_ID=
+XIAOMI_PUSH_TEMPLATE_TITLE=
+XIAOMI_PUSH_TEMPLATE_DESCRIPTION=
+# 可选: JSON 字符串，支持 {title}/{alert} 占位符，例如 {"title":"{title}","content":"{alert}"}
+XIAOMI_PUSH_TEMPLATE_PARAM_JSON=
+
+OPPO_PUSH_APP_KEY=
+OPPO_PUSH_MASTER_SECRET=
+OPPO_PUSH_AUTH_ENDPOINT=https://api.push.oppomobile.com/server/v1/auth
+OPPO_PUSH_SEND_ENDPOINT=https://api.push.oppomobile.com/server/v1/message/notification/unicast
+# OPPO 新消息分类(2024-11-20 后创建的应用必须携带 category;channel_id 为后台「通道ID」;
+# notify_level 0=不传走默认,内容营销类仅支持 1/2)
+OPPO_PUSH_CHANNEL_ID=
+OPPO_PUSH_CATEGORY=
+OPPO_PUSH_NOTIFY_LEVEL=0
+
+# ===== 无障碍保护存活监控(推送 + pull 后置兜底)=====
 HEARTBEAT_MONITOR_ENABLED=true
-HEARTBEAT_TIMEOUT_MINUTES=10
+HEARTBEAT_TIMEOUT_MINUTES=60
 HEARTBEAT_SCAN_INTERVAL_SEC=60
 
 # ===== 短信 (mock 模式) =====

alembic/versions/1a924c274fce_merge_direct_vendor_push_and_feedback_.py

@@ -0,0 +1,26 @@
+"""merge direct_vendor_push and feedback_type_reply heads
+
+Revision ID: 1a924c274fce
+Revises: direct_vendor_push_fields, feedback_type_reply
+Create Date: 2026-07-14 18:53:02.856979
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = '1a924c274fce'
+down_revision: Union[str, Sequence[str], None] = ('direct_vendor_push_fields', 'feedback_type_reply')
+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/761ef181ce7c_merge_jd_cps_order_fields_and_coupon_.py

@@ -0,0 +1,26 @@
+"""merge jd_cps_order_fields and coupon_session_origin_package heads
+
+Revision ID: 761ef181ce7c
+Revises: coupon_session_origin_package, jd_cps_order_fields
+Create Date: 2026-07-01 13:52:16.068808
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = '761ef181ce7c'
+down_revision: Union[str, Sequence[str], None] = ('coupon_session_origin_package', 'jd_cps_order_fields')
+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/coupon_session_origin_package.py

@@ -0,0 +1,32 @@
+"""coupon_session 加 origin_package 列(发起来源 App 包名 → admin「发起平台」)
+
+null=App 内(傻瓜比价首页)发起,非空=从美团/淘宝/京东弹券发起。与 trace_url 同理单独成迁移,
+已建表环境靠它补列、全新环境顺序应用,不重复加列。
+
+Revision ID: coupon_session_origin_package
+Revises: coupon_session_trace_url
+Create Date: 2026-06-30 00:00:00.000000
+
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "coupon_session_origin_package"
+down_revision: str | Sequence[str] | None = "coupon_session_trace_url"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    with op.batch_alter_table("coupon_session", schema=None) as batch_op:
+        batch_op.add_column(sa.Column("origin_package", sa.String(length=64), nullable=True))
+
+
+def downgrade() -> None:
+    with op.batch_alter_table("coupon_session", schema=None) as batch_op:
+        batch_op.drop_column("origin_package")

alembic/versions/coupon_session_table.py

@@ -0,0 +1,81 @@
+"""coupon session table(领券任务全程流水 → admin「领券数据」看板数据源)
+
+一次领券一行(trace_id 唯一):客户端 POST /api/v1/coupon/session 两段上报 —— 发起建行
+(status=started)、收尾(completed/failed/abandoned)按 trace_id 更新同一行。记全程耗时
+elapsed_ms + 各平台耗时 platform_elapsed + 机型/ROM,供 admin 算发起/完成数、耗时分位、机型维度。
+
+Revision ID: coupon_session_table
+Revises: feedback_submit_env
+Create Date: 2026-06-30 00:00:00.000000
+
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "coupon_session_table"
+down_revision: str | Sequence[str] | None = "feedback_submit_env"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+# PG 用 JSONB,SQLite(本地/测试)退化为通用 JSON(同 model 的 _JSON variant)。
+_JSON = sa.JSON().with_variant(postgresql.JSONB(), "postgresql")
+
+
+def upgrade() -> None:
+    op.create_table(
+        "coupon_session",
+        sa.Column("id", sa.Integer(), autoincrement=True, nullable=False),
+        sa.Column("trace_id", sa.String(length=64), nullable=False),
+        sa.Column("device_id", sa.String(length=64), nullable=False),
+        sa.Column("user_id", sa.Integer(), nullable=True),
+        sa.Column("status", sa.String(length=16), nullable=False),
+        sa.Column("app_env", sa.String(length=16), nullable=True),
+        sa.Column("platforms", _JSON, nullable=True),
+        sa.Column("device_model", sa.String(length=128), nullable=True),
+        sa.Column("rom", sa.String(length=64), nullable=True),
+        sa.Column("started_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("started_date", sa.Date(), nullable=False),
+        sa.Column("finished_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("elapsed_ms", sa.Integer(), nullable=True),
+        sa.Column("platform_elapsed", _JSON, nullable=True),
+        sa.Column("claimed_count", sa.Integer(), nullable=True),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("(CURRENT_TIMESTAMP)"),
+            nullable=False,
+        ),
+        sa.Column(
+            "updated_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("(CURRENT_TIMESTAMP)"),
+            nullable=False,
+        ),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint("trace_id", name="uq_coupon_session_trace"),
+    )
+    with op.batch_alter_table("coupon_session", schema=None) as batch_op:
+        batch_op.create_index(
+            batch_op.f("ix_coupon_session_user_id"), ["user_id"], unique=False
+        )
+        batch_op.create_index(
+            batch_op.f("ix_coupon_session_app_env"), ["app_env"], unique=False
+        )
+        # admin 主聚合/筛选:按上海自然日 + 环境。
+        batch_op.create_index(
+            "ix_coupon_session_date_env", ["started_date", "app_env"], unique=False
+        )
+
+
+def downgrade() -> None:
+    with op.batch_alter_table("coupon_session", schema=None) as batch_op:
+        batch_op.drop_index("ix_coupon_session_date_env")
+        batch_op.drop_index(batch_op.f("ix_coupon_session_app_env"))
+        batch_op.drop_index(batch_op.f("ix_coupon_session_user_id"))
+    op.drop_table("coupon_session")

alembic/versions/coupon_session_trace_url.py

@@ -0,0 +1,32 @@
+"""coupon_session 加 trace_url 列(pricebot done 帧公网调试链接)
+
+建表迁移 coupon_session_table 落地后才追加本列,故单独成一个迁移:已建表的环境(本地/已 upgrade 过)
+靠它补列,全新环境则「建表(无 trace_url)→ 本迁移加列」,两条路一致、不重复加列。
+
+Revision ID: coupon_session_trace_url
+Revises: coupon_session_table
+Create Date: 2026-06-30 00:00:00.000000
+
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "coupon_session_trace_url"
+down_revision: str | Sequence[str] | None = "coupon_session_table"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    with op.batch_alter_table("coupon_session", schema=None) as batch_op:
+        batch_op.add_column(sa.Column("trace_url", sa.String(length=512), nullable=True))
+
+
+def downgrade() -> None:
+    with op.batch_alter_table("coupon_session", schema=None) as batch_op:
+        batch_op.drop_column("trace_url")

alembic/versions/direct_vendor_push_fields.py

@@ -0,0 +1,30 @@
+"""add direct vendor push fields
+
+Revision ID: direct_vendor_push_fields
+Revises: jd_cps_order_fields
+Create Date: 2026-07-01 16:30:00.000000
+"""
+from __future__ import annotations
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision = "direct_vendor_push_fields"
+down_revision = "jd_cps_order_fields"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    with op.batch_alter_table("device_liveness") as batch_op:
+        batch_op.add_column(sa.Column("push_vendor", sa.String(length=32), nullable=True))
+        batch_op.add_column(sa.Column("push_token", sa.String(length=256), nullable=True))
+        batch_op.create_index("ix_device_liveness_push_vendor", ["push_vendor"])
+
+
+def downgrade() -> None:
+    with op.batch_alter_table("device_liveness") as batch_op:
+        batch_op.drop_index("ix_device_liveness_push_vendor")
+        batch_op.drop_column("push_token")
+        batch_op.drop_column("push_vendor")

alembic/versions/feedback_type_reply.py

@@ -0,0 +1,48 @@
+"""feedback 加反馈来源/场景/运营回复(source / scene / admin_reply)
+
+admin「用户反馈」页要展示并筛选「反馈类型」(比价反馈 / 普通反馈),并支持审核时给用户留言:
+  - source: 反馈来源入口(profile=「我的」页 / comparison=比价结果页)。NOT NULL,
+    旧数据 + 普通反馈默认 profile(server_default)。加索引供 admin 按类型筛选。
+  - scene: 比价反馈的问题场景(找错商品/优惠不对…),普通反馈为 NULL。
+  - admin_reply: 运营给用户的回复留言(用户端可见),随「我的反馈」历史下发。
+
+均为新增列(SQLite 原生支持 add_column);downgrade 的 drop_column 在 SQLite 走 batch 兜底。
+
+Revision ID: feedback_type_reply
+Revises: 761ef181ce7c
+Create Date: 2026-07-02 00:00:00.000000
+"""
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "feedback_type_reply"
+down_revision: str | Sequence[str] | None = "761ef181ce7c"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "feedback",
+        sa.Column(
+            "source",
+            sa.String(length=16),
+            nullable=False,
+            server_default="profile",
+        ),
+    )
+    op.add_column("feedback", sa.Column("scene", sa.String(length=32), nullable=True))
+    op.add_column("feedback", sa.Column("admin_reply", sa.String(length=256), nullable=True))
+    op.create_index("ix_feedback_source", "feedback", ["source"])
+
+
+def downgrade() -> None:
+    with op.batch_alter_table("feedback") as batch_op:
+        batch_op.drop_index("ix_feedback_source")
+        batch_op.drop_column("admin_reply")
+        batch_op.drop_column("scene")
+        batch_op.drop_column("source")

app/admin/main.py

@@ -21,6 +21,7 @@ from app.admin.routers.audit import router as audit_router
 from app.admin.routers.auth import router as auth_router
 from app.admin.routers.comparison import router as comparison_router
 from app.admin.routers.config import router as config_router
+from app.admin.routers.coupon_data import router as coupon_data_router
 from app.admin.routers.cps import router as cps_router
 from app.admin.routers.dashboard import router as dashboard_router
 from app.admin.routers.device_liveness import router as device_liveness_router
@@ -101,6 +102,7 @@ admin_app.include_router(audit_router)
 admin_app.include_router(config_router)
 admin_app.include_router(comparison_router)
 admin_app.include_router(cps_router)
+admin_app.include_router(coupon_data_router)
 admin_app.include_router(ad_audit_router)
 admin_app.include_router(ad_config_router)
 admin_app.include_router(ad_revenue_router)

app/admin/repositories/ad_audit.py

@@ -136,12 +136,16 @@ def _feed_scene_matches(rec: AdFeedRewardRecord, scene: str | None) -> bool:
     """该信息流记录是否落入请求的展示筛选 scene。
     - scene=="feed":ad_type in ("feed", NULL)(旧数据 NULL 视为 feed,向后兼容)
     - scene=="draw":ad_type=="draw"
+    - scene=="feed_all":所有信息流(feed/draw/NULL 都要)——业务已全切 Draw 信息流,收益报表把「Draw 信息流」
+      当作整个信息流口径(含历史误标 feed/NULL),用它避免筛选漏历史。
     - scene 为 None:不筛(两类都要)。
     """
     if scene == "feed":
         return rec.ad_type in (None, "feed")
     if scene == "draw":
         return rec.ad_type == "draw"
+    if scene == "feed_all":
+        return True
     return True
 
 
@@ -184,6 +188,7 @@ def _feed_rows(
                 "record_id": rec.id,
                 "user_id": rec.user_id,
                 "ad_session_id": rec.ad_session_id,
+                "trace_id": rec.trace_id,
                 "app_env": rec.app_env,
                 "our_code_id": rec.our_code_id,
                 "created_at": rec.created_at,
@@ -209,6 +214,7 @@ def _feed_rows(
                 "record_id": rec.id,
                 "user_id": rec.user_id,
                 "ad_session_id": rec.ad_session_id,
+                "trace_id": rec.trace_id,
                 "app_env": rec.app_env,
                 "our_code_id": rec.our_code_id,
                 "created_at": rec.created_at,
@@ -241,7 +247,7 @@ def audit_rows(
     rows: list[dict] = []
     if scene in (None, "reward_video"):
         rows.extend(_reward_video_rows(db, date=date, user_id=user_id))
-    if scene in (None, "feed", "draw"):
+    if scene in (None, "feed", "draw", "feed_all"):
         rows.extend(_feed_rows(db, date=date, user_id=user_id, scene=scene))
     return rows
 

app/admin/repositories/ad_revenue.py

@@ -3,9 +3,11 @@
 只读。每行 = 一次广告事件(不再按用户聚合):
 - **激励视频**:一次观看 = 1 条展示(ad_ecpm)+ 1 条发奖(ad_reward),按 ad_session_id 合并成一行,
   直接给出 eCPM / 收益 + 状态 / 应发 / 实发 / 一致;点开看该条金币复算因子。
-- **信息流**:轮播每条展示各一行(impressionId 各自独立);整场发奖(ad_feed_reward,client_event_id)
-  与逐条展示无法对应,单独成「纯发奖」行。
-- 兜底:有展示无发奖(中途关 / 未达发奖)、有发奖无展示(未上报 eCPM)都各自成行。
+- **信息流(比价/领券)**:一次比价 / 一次领券 = 一条整场发奖(ad_feed_reward)一行,给出 eCPM /
+  发奖金币 + 应发 / 实发 / 一致;点开看金币复算因子。⚠️ draw 的逐条展示(ad_ecpm,impressionId 各自
+  独立、与整场发奖无公共键、无法归到「哪一次」)**不再单独占行**(2026-07 按「一次比价/领券放一块」调整)——
+  其展示数 / eCPM / 预估收益仍进全量统计(合计 / 趋势 / 分类大盘 / 穿山甲对照),只是主表不逐条铺开。
+- 兜底:激励视频有展示无发奖(中途关 / 未达发奖)、有发奖无展示(未上报 eCPM)仍各自成行。
 
 展示与收益来自 ad_ecpm_record(收益 = eCPM元 ÷ 1000);应发 / 实发金币复用金币审计逐条复算
 (ad_audit.audit_rows,与正式发奖同一公式口径,不另写公式)。合计与对账在全量上统计,
@@ -58,14 +60,6 @@ def _date_range(date_from: str, date_to: str) -> list[str]:
 _AUDIT_SCENES = {"reward_video", "feed", "draw"}
 
 
-def _event_ad_type(row: dict) -> str:
-    """纯发奖事件行的 ad_type:信息流行用 audit 带回的真实 ad_type(feed/draw),回退 feed;
-    激励视频行恒 reward_video。不再用 scene 硬映射,避免把 draw 丢成 feed。"""
-    if row["scene"] == "reward_video":
-        return "reward_video"
-    return row.get("ad_type") or "feed"
-
-
 # 发奖复算明细字段(展开下钻看「金币怎么算出来的」)——从 audit 行原样取这些 key。
 _REWARD_DETAIL_KEYS = (
     "record_id", "created_at", "status", "ecpm", "ecpm_factor", "units",
@@ -108,9 +102,14 @@ def ad_revenue_report(
     #    同时保留全量列表,未被展示合并的成「纯发奖」事件。
     reward_by_session: dict[tuple[int, str], list[dict]] = {}
     all_reward_rows: list[dict] = []
-    # 报表 ad_type 直接当 audit scene 用(取值一致);未知/无效 ad_type 不取发奖行。draw 在此被
-    # 正确传成 scene="draw",audit 会按 ad_type 筛出 Draw 发奖,不再丢成 feed。
-    audit_scene = ad_type if ad_type in _AUDIT_SCENES else None
+    # 报表 ad_type → audit scene:reward_video/feed 直传;**draw(前端「Draw 信息流」)映射成 feed_all**
+    # ——业务已全切 Draw,把「Draw 信息流」当作整个信息流口径(含历史误标 feed/NULL),否则筛选会漏历史。
+    if ad_type == "draw":
+        audit_scene = "feed_all"
+    elif ad_type in _AUDIT_SCENES:
+        audit_scene = ad_type
+    else:
+        audit_scene = None
     if ad_type is None or audit_scene is not None:
         for d in _date_range(date_from, date_to):
             for row in ad_audit.audit_rows(db, date=d, user_id=user_id, scene=audit_scene):
@@ -140,7 +139,10 @@ def ad_revenue_report(
     )
     if user_id is not None:
         stmt = stmt.where(AdEcpmRecord.user_id == user_id)
-    if ad_type is not None:
+    if ad_type == "draw":
+        # draw = 所有信息流展示(业务已全 Draw,含历史误标 feed);展示行只进统计,不占主表行
+        stmt = stmt.where(AdEcpmRecord.ad_type.in_(["draw", "feed"]))
+    elif ad_type is not None:
         stmt = stmt.where(AdEcpmRecord.ad_type == ad_type)
     for rec in db.execute(stmt).scalars():
         rwd = _pop_reward(rec.user_id, rec.ad_session_id)
@@ -165,6 +167,8 @@ def ad_revenue_report(
             ),
             "adn": rec.adn,
             "slot_id": rec.slot_id,
+            "sub_rewards": [],
+            "sub_count": 1,
         }
         if rwd is not None:
             ev.update({
@@ -184,33 +188,88 @@ def ad_revenue_report(
             })
         events.append(ev)
 
-    # 3) 未被展示合并的发奖行 → 「纯发奖」事件(信息流整场发奖 / 有发奖无展示)。
-    #    收益恒 0(收益只算展示侧,避免与展示行重复计)。
+    # 3) 未被展示合并的发奖行 → 事件:
+    #    - 激励视频(reward_video):逐条成「纯发奖」事件(每次一个 ad_session_id;有发奖无展示等)。
+    #    - 信息流(feed/draw):同一次比价/领券的多条广告共享**整场 ad_session_id**(客户端整场复用),
+    #      按 (user_id, ad_session_id) 聚成**一次比价 / 一次领券**父事件;sub_rewards 为组内逐条明细,
+    #      应发/实发取组内合计;业务已全 Draw → 类型统一 "draw"。session 缺失(极少旧数据)各自单独成组。
+    feed_groups: dict[tuple[int, str], list[dict]] = {}
     for row in all_reward_rows:
         if row["record_id"] in used_reward_ids:
             continue
+        if row["scene"] == "reward_video":
+            events.append({
+                "event_key": f"rwd-{row['record_id']}",
+                "report_date": row["_report_date"],
+                "user_id": row["user_id"],
+                "ad_type": "reward_video",
+                "feed_scene": row.get("feed_scene"),
+                "app_env": row.get("app_env"),
+                "our_code_id": row.get("our_code_id"),
+                "created_at": row["created_at"],
+                "hour": _cn_hour(row["created_at"]) if by_hour else None,
+                "has_impression": False,
+                "impressions": 0,
+                "ecpm": row["ecpm"],
+                "revenue_yuan": 0.0,
+                "adn": None,
+                "slot_id": None,
+                "has_reward": True,
+                "status": row["status"],
+                "expected_coin": int(row["expected_coin"]),
+                "actual_coin": int(row["actual_coin"]),
+                "matched": bool(row["matched"]),
+                "reward_detail": _reward_detail(row),
+                "sub_rewards": [],
+                "sub_count": 1,
+            })
+        else:
+            # 聚合单位 = 一次完整比价/领券流程:优先用 trace_id(比价带 comparisonTraceId、领券带 sessionTraceId,
+            # 整个流程不变;即使中途点广告致浮层关闭重弹、ad_session_id 变了,trace_id 仍不变 → 全流程聚成一行)。
+            # 无 trace_id(历史领券未上报 / 旧数据)回退整场 ad_session_id;再无则 record_id 各自成组、不误并。
+            grp_key = row.get("trace_id") or row.get("ad_session_id") or f"_rid-{row['record_id']}"
+            feed_groups.setdefault((row["user_id"], grp_key), []).append(row)
+
+    # 信息流分组 → 「一次比价 / 一次领券」父事件(收益恒 0:收益只算展示侧,避免与展示行重复计)。
+    for (uid, grp_key), group in feed_groups.items():
+        group.sort(key=lambda r: (r["created_at"], r["record_id"]))
+        rep = group[-1]  # 代表条(最新一条):时间/场景/应用/代码位取它
+        expected_sum = sum(int(g["expected_coin"]) for g in group)
+        actual_sum = sum(int(g["actual_coin"]) for g in group)
+        # 父行 eCPM:组内各条 eCPM(分)均值(展示用,各条不同);无有效值则取代表条
+        ecpm_fens = [rewards.parse_ecpm_fen(g["ecpm"]) for g in group if g.get("ecpm")]
+        avg_ecpm = str(round(sum(ecpm_fens) / len(ecpm_fens))) if ecpm_fens else rep.get("ecpm")
+        # 主表逐行显示用:这次发奖广告的预估收益之和(发奖侧 eCPM 折算,钳顶同展示侧)。只放进
+        # row_revenue_yuan 给主表逐行展示,不进 revenue_yuan/合计/趋势——避免与展示侧 total 重复计。
+        row_revenue = round(sum(
+            min(rewards.parse_ecpm_yuan(g["ecpm"]), rewards.AD_ECPM_MAX_FEN / 100.0) / 1000.0
+            for g in group if g.get("ecpm")
+        ), 6)
         events.append({
-            "event_key": f"rwd-{row['record_id']}",
-            "report_date": row["_report_date"],
-            "user_id": row["user_id"],
-            "ad_type": _event_ad_type(row),
-            "feed_scene": row.get("feed_scene"),
-            "app_env": row.get("app_env"),
-            "our_code_id": row.get("our_code_id"),
-            "created_at": row["created_at"],
-            "hour": _cn_hour(row["created_at"]) if by_hour else None,
+            "event_key": f"feedgrp-{uid}-{grp_key}",
+            "report_date": rep["_report_date"],
+            "user_id": uid,
+            "ad_type": "draw",  # 业务已全切 Draw 信息流,聚合行统一 draw
+            "feed_scene": rep.get("feed_scene"),
+            "app_env": rep.get("app_env"),
+            "our_code_id": rep.get("our_code_id"),
+            "created_at": rep["created_at"],
+            "hour": _cn_hour(rep["created_at"]) if by_hour else None,
             "has_impression": False,
             "impressions": 0,
-            "ecpm": row["ecpm"],
+            "ecpm": avg_ecpm,
             "revenue_yuan": 0.0,
+            "row_revenue_yuan": row_revenue,
             "adn": None,
             "slot_id": None,
             "has_reward": True,
-            "status": row["status"],
-            "expected_coin": int(row["expected_coin"]),
-            "actual_coin": int(row["actual_coin"]),
-            "matched": bool(row["matched"]),
-            "reward_detail": _reward_detail(row),
+            "status": rep["status"],  # 代表状态(逐条见展开)
+            "expected_coin": expected_sum,
+            "actual_coin": actual_sum,
+            "matched": all(bool(g["matched"]) for g in group),
+            "reward_detail": None,
+            "sub_rewards": [_reward_detail(g) for g in group],
+            "sub_count": len(group),
         })
 
     # 「场景」作为全局筛选(与 user_id/ad_type 一致):同时作用于明细、合计与 daily/hourly 趋势。
@@ -331,9 +390,20 @@ def ad_revenue_report(
     is_today = date_from == date_to == rewards.cn_today().isoformat()
     dau = admin_stats.today_dau(db) if is_today else None
 
+    # 主表「逐行」= 单次广告行为(2026-07 按「一次比价/领券放一块」聚合):激励视频 = 一次观看一行(展示+发奖
+    # 按 ad_session_id 合并);一次比价 / 一次领券 = 该次整场多条广告按 ad_session_id 聚成一行(展开看逐条)。
+    # 信息流(draw/feed)的逐条展示(ad_ecpm,impressionId 各自独立、与整场发奖无公共键)不再单独占行
+    # ——其展示数 / eCPM / 预估收益已计入上面的全量统计(total_*、daily / hourly、type_stats、穿山甲对照),
+    # 只是主表不逐条铺开;逐条明细在父行展开里看(sub_rewards)。合计 / 趋势 / 分类大盘均基于全量 events,
+    # 不受此过滤影响;total / 分页只作用于主表行。
+    main_rows = [
+        e for e in events
+        if not (e["ad_type"] in ("draw", "feed") and e["has_impression"] and not e["has_reward"])
+    ]
+
     return {
-        "total": len(events),
-        "truncated": len(events) > offset + limit,
+        "total": len(main_rows),
+        "truncated": len(main_rows) > offset + limit,
         "total_impressions": total_impressions,
         "total_revenue_yuan": total_revenue_yuan,
         # 穿山甲后台收益合计(元):预估 revenue + 收益Api;非全量视图(带 user/类型/场景过滤)或无数据为 None。
@@ -347,5 +417,5 @@ def ad_revenue_report(
         "hourly": hourly,
         "type_stats": type_stats,
         "dau": dau,
-        "items": events[offset:offset + limit],
+        "items": main_rows[offset:offset + limit],
     }

app/admin/repositories/coupon_data.py

@@ -0,0 +1,225 @@
+"""admin「领券数据」看板聚合:发起/完成数、耗时均值与分位、按天/小时趋势、逐条明细。
+
+数据源 coupon_session(一次领券一行,客户端 /api/v1/coupon/session 两段上报)。量级不大,全量拉
+区间数据后 Python 聚合(分位 SQLite 无 percentile,统一 Python 算,PG 上也一致)。
+- 发起数 = 区间内全部 session(含 started/completed/failed/abandoned),= 流失统计的基数。
+- 完成数 / 耗时均值 / 分位 = 仅 status==completed 子集(成功跑完才有可比的"领券耗时")。
+- summary/daily/hourly/total 在全量上算,不受分页;items 为排序后当前页。
+"""
+from __future__ import annotations
+
+from datetime import UTC, date as _date, datetime
+
+from sqlalchemy import func, or_, select
+from sqlalchemy.orm import Session
+
+from app.core import rewards
+from app.models.coupon_state import CouponSession
+from app.models.user import User
+
+
+def _cn_hour(dt: datetime) -> int:
+    """started_at(UTC 口径)→ 北京时间小时(0–23)。naive 当 UTC(sqlite),tz-aware 直接换算(pg)。"""
+    if dt.tzinfo is None:
+        dt = dt.replace(tzinfo=UTC)
+    return dt.astimezone(rewards.CN_TZ).hour
+
+
+def _percentile(sorted_vals: list[int], q: float) -> int | None:
+    """线性插值分位(q=0..100,numpy 默认法)。sorted_vals 须已升序;空返回 None。"""
+    if not sorted_vals:
+        return None
+    if len(sorted_vals) == 1:
+        return sorted_vals[0]
+    idx = (len(sorted_vals) - 1) * q / 100.0
+    lo = int(idx)
+    hi = min(lo + 1, len(sorted_vals) - 1)
+    frac = idx - lo
+    return round(sorted_vals[lo] * (1 - frac) + sorted_vals[hi] * frac)
+
+
+def _avg(vals: list[int]) -> int | None:
+    return round(sum(vals) / len(vals)) if vals else None
+
+
+def _session_to_row(r, phone: str | None = None, nickname: str | None = None) -> dict:
+    """CouponSession ORM → 明细行 dict(主表「领券数据」与「用户全部领券」抽屉共用)。"""
+    return {
+        "id": r.id,
+        "trace_id": r.trace_id,
+        "user_id": r.user_id,
+        "user_phone": phone,
+        "user_nickname": nickname,
+        "status": r.status,
+        "platforms": r.platforms,
+        "origin_package": r.origin_package,
+        "elapsed_ms": r.elapsed_ms,
+        "platform_elapsed": r.platform_elapsed,
+        "device_model": r.device_model,
+        "rom": r.rom,
+        "app_env": r.app_env,
+        "started_at": r.started_at,
+        "claimed_count": r.claimed_count,
+        "trace_url": r.trace_url,
+    }
+
+
+def _empty_result() -> dict:
+    return {
+        "summary": {
+            "started_count": 0, "completed_count": 0, "avg_elapsed_ms": None,
+            "p5_ms": None, "p50_ms": None, "p95_ms": None, "p99_ms": None,
+        },
+        "daily": [],
+        "hourly": [],
+        "total": 0,
+        "items": [],
+    }
+
+
+def coupon_data_report(
+    db: Session,
+    *,
+    date_from: str,
+    date_to: str,
+    user: str | None = None,
+    app_env: str | None = None,
+    granularity: str = "day",
+    limit: int = 500,
+    offset: int = 0,
+    sort: str = "time",
+) -> dict:
+    """日期区间(北京自然日 started_date,闭区间)领券数据:汇总卡 + 趋势 + 逐条明细。
+
+    - user:手机号/昵称模糊搜(匹配不到任何用户 → 空结果)。
+    - app_env:prod/dev 精确;None=全部。
+    - sort:time=发起时刻倒序(默认) / elapsed=全程耗时倒序(None 末尾)。
+    """
+    by_hour = granularity == "hour"
+    d_from = _date.fromisoformat(date_from)
+    d_to = _date.fromisoformat(date_to)
+
+    # user 模糊 → 先定位匹配用户 id;匹配不到直接空结果(不全表扫)。
+    user_ids: set[int] | None = None
+    if user:
+        like = f"%{user}%"
+        user_ids = set(db.execute(
+            select(User.id).where(or_(User.phone.like(like), User.nickname.like(like)))
+        ).scalars().all())
+        if not user_ids:
+            return _empty_result()
+
+    stmt = select(CouponSession).where(
+        CouponSession.started_date >= d_from,
+        CouponSession.started_date <= d_to,
+    )
+    if app_env is not None:
+        stmt = stmt.where(CouponSession.app_env == app_env)
+    if user_ids is not None:
+        stmt = stmt.where(CouponSession.user_id.in_(user_ids))
+    rows = list(db.execute(stmt).scalars())
+
+    # ── 汇总卡 ──
+    completed_elapsed = sorted(
+        r.elapsed_ms for r in rows if r.status == "completed" and r.elapsed_ms is not None
+    )
+    summary = {
+        "started_count": len(rows),
+        "completed_count": sum(1 for r in rows if r.status == "completed"),
+        "avg_elapsed_ms": _avg(completed_elapsed),
+        "p5_ms": _percentile(completed_elapsed, 5),
+        "p50_ms": _percentile(completed_elapsed, 50),
+        "p95_ms": _percentile(completed_elapsed, 95),
+        "p99_ms": _percentile(completed_elapsed, 99),
+    }
+
+    # ── 按天趋势(柱=发起/完成数,线=平均耗时)──
+    daily_map: dict[str, dict] = {}
+    for r in rows:
+        d = r.started_date.isoformat()
+        b = daily_map.get(d)
+        if b is None:
+            b = {"date": d, "started_count": 0, "completed_count": 0, "_elapsed": []}
+            daily_map[d] = b
+        b["started_count"] += 1
+        if r.status == "completed":
+            b["completed_count"] += 1
+            if r.elapsed_ms is not None:
+                b["_elapsed"].append(r.elapsed_ms)
+    daily = [
+        {
+            "date": b["date"],
+            "started_count": b["started_count"],
+            "completed_count": b["completed_count"],
+            "avg_elapsed_ms": _avg(b["_elapsed"]),
+        }
+        for b in sorted(daily_map.values(), key=lambda x: x["date"])
+    ]
+
+    # ── 按小时趋势(单日 hour 粒度)──
+    hourly: list[dict] = []
+    if by_hour:
+        hour_map: dict[int, dict] = {}
+        for r in rows:
+            h = _cn_hour(r.started_at)
+            b = hour_map.get(h)
+            if b is None:
+                b = {"hour": h, "started_count": 0, "completed_count": 0, "_elapsed": []}
+                hour_map[h] = b
+            b["started_count"] += 1
+            if r.status == "completed":
+                b["completed_count"] += 1
+                if r.elapsed_ms is not None:
+                    b["_elapsed"].append(r.elapsed_ms)
+        hourly = [
+            {
+                "hour": b["hour"],
+                "started_count": b["started_count"],
+                "completed_count": b["completed_count"],
+                "avg_elapsed_ms": _avg(b["_elapsed"]),
+            }
+            for b in sorted(hour_map.values(), key=lambda x: x["hour"])
+        ]
+
+    # ── 明细:排序 + 分页 + 补用户手机号/昵称(批量,防 N+1)──
+    if sort == "elapsed":
+        rows.sort(key=lambda r: (r.elapsed_ms is None, -(r.elapsed_ms or 0)))
+    else:  # time:发起时刻倒序
+        rows.sort(key=lambda r: r.started_at, reverse=True)
+    page = rows[offset:offset + limit]
+
+    uids = {r.user_id for r in page if r.user_id is not None}
+    user_map: dict[int, tuple[str | None, str | None]] = {}
+    if uids:
+        user_map = {
+            uid: (phone, nickname)
+            for uid, phone, nickname in db.execute(
+                select(User.id, User.phone, User.nickname).where(User.id.in_(uids))
+            ).all()
+        }
+    items = []
+    for r in page:
+        phone, nickname = user_map.get(r.user_id, (None, None)) if r.user_id is not None else (None, None)
+        items.append(_session_to_row(r, phone, nickname))
+
+    return {
+        "summary": summary,
+        "daily": daily,
+        "hourly": hourly,
+        "total": len(rows),
+        "items": items,
+    }
+
+
+def coupon_user_records(db: Session, *, user_id: int, limit: int = 100) -> dict:
+    """某用户全部领券记录(点手机号抽屉用):按发起时刻倒序、不限日期,total=该用户领券总次数。"""
+    rows = list(db.execute(
+        select(CouponSession)
+        .where(CouponSession.user_id == user_id)
+        .order_by(CouponSession.started_at.desc())
+        .limit(limit)
+    ).scalars())
+    total = db.execute(
+        select(func.count()).select_from(CouponSession).where(CouponSession.user_id == user_id)
+    ).scalar_one()
+    return {"items": [_session_to_row(r) for r in rows], "total": int(total)}

app/admin/repositories/mutations.py

@@ -65,16 +65,19 @@ def review_feedback(
     reward_coins: int | None = None,
     reject_reason: str | None = None,
     review_note: str | None = None,
+    admin_reply: str | None = None,
     commit: bool = True,
 ) -> Feedback:
-    """审核反馈:置 adopted/rejected + 记录奖励/原因/审核人/审核时间。
+    """审核反馈:置 adopted/rejected + 记录奖励/原因/回复留言/审核人/审核时间。
 
     发金币(wallet.grant_coins)由 router 在同一事务里调,确保状态、金币流水、审计一起提交。
+    admin_reply=给用户的回复留言(用户端可见),与 review_note(内部备注)区分。
     """
     feedback.status = status
     feedback.reward_coins = reward_coins
     feedback.reject_reason = reject_reason
     feedback.review_note = review_note
+    feedback.admin_reply = admin_reply
     feedback.reviewed_by_admin_id = reviewed_by_admin_id
     feedback.reviewed_at = datetime.now(CN_TZ).replace(tzinfo=None)
     if commit:

app/admin/repositories/queries.py

@@ -390,6 +390,7 @@ def list_all_withdraw_orders(
     *,
     user_id: int | None = None,
     status: str | None = None,
+    source: str | None = None,
     keyword: str | None = None,
     date_from: datetime | None = None,
     date_to: datetime | None = None,
@@ -409,6 +410,9 @@ def list_all_withdraw_orders(
         stmt = stmt.where(WithdrawOrder.user_id == user_id)
     if status:
         stmt = stmt.where(WithdrawOrder.status == status)
+    # 提现类型:coin_cash(福利页提现)/ invite_cash(邀请提现);None=全部
+    if source:
+        stmt = stmt.where(WithdrawOrder.source == source)
 
     kw = (keyword or "").strip()
     if kw:
@@ -534,6 +538,7 @@ def list_feedbacks(
     db: Session,
     *,
     status: str | None = None,
+    source: str | None = None,
     user_id: int | None = None,
     content: str | None = None,
     created_from: datetime | None = None,
@@ -543,13 +548,15 @@ def list_feedbacks(
     limit: int = 20,
     cursor: int | None = None,
 ) -> tuple[list[Feedback], int | None, int]:
-    """反馈工单列表。支持 状态 / 用户ID / 内容模糊 / 提交时间范围 筛选,按 id·提交时间排序。
-    **offset 分页**(cursor=offset):任意列排序下游标语义统一(同 [list_users]),代价是翻页期间
-    数据变动可能错位一条——admin 低频场景可接受。返回 (items, next_cursor, total),total 供页码分页。
-    created_at 为 timestamptz,日期入参统一转 tz-aware UTC 比较。"""
+    """反馈工单列表。支持 状态 / 反馈类型(source) / 用户ID / 内容模糊 / 提交时间范围 筛选,
+    按 id·提交时间排序。**offset 分页**(cursor=offset):任意列排序下游标语义统一(同 [list_users]),
+    代价是翻页期间数据变动可能错位一条——admin 低频场景可接受。返回 (items, next_cursor, total),
+    total 供页码分页。created_at 为 timestamptz,日期入参统一转 tz-aware UTC 比较。"""
     stmt = select(Feedback)
     if status:
         stmt = stmt.where(Feedback.status == status)
+    if source:
+        stmt = stmt.where(Feedback.source == source)
     if user_id is not None:
         stmt = stmt.where(Feedback.user_id == user_id)
     if content and content.strip():
@@ -806,6 +813,12 @@ def get_user_overview(db: Session, user_id: int) -> dict | None:
     }
 
 
+def _as_utc_naive(value: datetime) -> datetime:
+    """窗口入参 → UTC naive(= _as_utc 去时区),与库里按 naive UTC 存取的 created_at 同口径比较。
+    历史遗留:_window_conds 一直引用本函数却未定义(自定义区间会 NameError),此处补上。"""
+    return _as_utc(value).replace(tzinfo=None)
+
+
 def _window_conds(col, date_from: datetime | None, date_to: datetime | None) -> list:
     """把 [date_from, date_to] 转成对 col(created_at)的过滤条件;都为 None = 全量(注册至今)。"""
     conds = []
@@ -895,6 +908,13 @@ def user_reward_stats(
     }
 
 
+def _cn_wall_to_utc(dt: datetime) -> datetime:
+    """coin_transaction 存的是北京 wall-clock(naive,见 wallet.grant_coins「存北京 wall-clock」),转成 UTC naive,
+    与广告表(func.now() UTC)统一 —— 让本函数按同一绝对时刻排序、且前端 apiTime(把无时区时间当 UTC 再 +8 展示)
+    口径一致;否则签到会比实际多显示 8 小时(北京时间又被 +8)。"""
+    return dt.replace(tzinfo=rewards.CN_TZ).astimezone(timezone.utc).replace(tzinfo=None)
+
+
 def user_coin_records(
     db: Session,
     user_id: int,
@@ -915,6 +935,9 @@ def user_coin_records(
     offset = max(cursor or 0, 0)
     fetch = offset + limit + 1
     rows: list[dict] = []
+    # coin_transaction 存北京 wall-clock(其余表存 UTC);签到窗口边界 +8h 对齐北京,过滤/计数才不偏移 8 小时
+    signin_from = date_from + timedelta(hours=8) if date_from is not None else None
+    signin_to = date_to + timedelta(hours=8) if date_to is not None else None
 
     for rec in db.execute(
         select(AdRewardRecord)
@@ -958,7 +981,7 @@ def user_coin_records(
         .where(
             CoinTransaction.user_id == user_id,
             CoinTransaction.biz_type == "signin",
-            *_window_conds(CoinTransaction.created_at, date_from, date_to),
+            *_window_conds(CoinTransaction.created_at, signin_from, signin_to),
         )
         .order_by(CoinTransaction.created_at.desc())
         .limit(fetch)
@@ -966,7 +989,8 @@ def user_coin_records(
         rows.append({
             "source": "signin",
             "source_label": "签到",
-            "created_at": rec.created_at,
+            # 北京 wall-clock → UTC,与广告记录统一(前端 apiTime 会 +8 回北京展示,不然签到会多 8 小时)
+            "created_at": _cn_wall_to_utc(rec.created_at),
             "ecpm": None,
             "coin": rec.amount,
         })
@@ -992,7 +1016,7 @@ def user_coin_records(
         + _count(
             CoinTransaction, CoinTransaction.user_id == user_id,
             CoinTransaction.biz_type == "signin",
-            *_window_conds(CoinTransaction.created_at, date_from, date_to),
+            *_window_conds(CoinTransaction.created_at, signin_from, signin_to),
         )
     )
     return rows[offset:offset + limit], (offset + limit if has_more else None), total

app/admin/routers/coupon_data.py

@@ -0,0 +1,106 @@
+"""admin「领券数据」看板:发起/完成数 + 领券耗时(均值 + P5/P50/P95/P99)+ 按天趋势 + 逐条明细。
+
+任意已登录 admin 可看(只读)。聚合逻辑在 app/admin/repositories/coupon_data.py。
+数据源 coupon_session(客户端 /api/v1/coupon/session 两段上报)。
+"""
+from __future__ import annotations
+
+from datetime import date as _date
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, HTTPException, Query
+
+from app.admin.deps import AdminDb, get_current_admin
+from app.admin.repositories import coupon_data
+from app.admin.schemas.coupon_data import (
+    CouponDataDaily,
+    CouponDataHourly,
+    CouponDataOut,
+    CouponDataRow,
+    CouponDataSummary,
+    CouponUserRecordsOut,
+)
+from app.core.rewards import cn_today
+
+router = APIRouter(
+    prefix="/admin/api/coupon-data",
+    tags=["admin-coupon-data"],
+    dependencies=[Depends(get_current_admin)],
+)
+
+# 区间最大跨度(天);超出拒绝,避免一次拉过多天拖垮接口(对齐广告收益报表)。
+_MAX_RANGE_DAYS = 92
+
+
+def _parse_day(value: str | None, *, field: str, default: _date) -> _date:
+    if value is None:
+        return default
+    try:
+        return _date.fromisoformat(value)
+    except ValueError as e:
+        raise HTTPException(status_code=422, detail=f"{field} 需为 YYYY-MM-DD") from e
+
+
+@router.get(
+    "",
+    response_model=CouponDataOut,
+    summary="领券数据看板(发起/完成数 + 耗时分位 + 按天趋势 + 逐条明细)",
+)
+def get_coupon_data(
+    db: AdminDb,
+    date_from: Annotated[str | None, Query(description="起始日 北京 YYYY-MM-DD,默认今天")] = None,
+    date_to: Annotated[str | None, Query(description="结束日 北京 YYYY-MM-DD,闭区间,默认=date_from")] = None,
+    user: Annotated[str | None, Query(description="用户手机号/昵称模糊搜;不传=全部")] = None,
+    app_env: Annotated[str, Query(description="prod(默认) / dev / all(全部环境)")] = "prod",
+    granularity: Annotated[
+        str, Query(description="day=按天 / hour=按小时(北京);区间>1 天建议 day")
+    ] = "day",
+    limit: Annotated[int, Query(ge=1, le=1000, description="每页条数(分页大小)")] = 500,
+    offset: Annotated[int, Query(ge=0, description="分页偏移(已跳过条数)=(页码-1)×每页条数")] = 0,
+    sort: Annotated[
+        str, Query(description="排序:time=发起时间倒序(默认) / elapsed=耗时倒序")
+    ] = "time",
+) -> CouponDataOut:
+    today = cn_today()
+    d_from = _parse_day(date_from, field="date_from", default=today)
+    d_to = _parse_day(date_to, field="date_to", default=d_from)
+    if d_to < d_from:
+        raise HTTPException(status_code=422, detail="date_to 不能早于 date_from")
+    if (d_to - d_from).days + 1 > _MAX_RANGE_DAYS:
+        raise HTTPException(status_code=422, detail=f"区间最长 {_MAX_RANGE_DAYS} 天")
+
+    # 报表默认只看 prod(对齐广告报表防串台口径);app_env=all 时不过滤、看全部环境。
+    env = None if app_env == "all" else app_env
+    result = coupon_data.coupon_data_report(
+        db, date_from=d_from.isoformat(), date_to=d_to.isoformat(),
+        user=user, app_env=env, granularity=granularity,
+        limit=limit, offset=offset, sort=sort,
+    )
+    return CouponDataOut(
+        date_from=d_from.isoformat(),
+        date_to=d_to.isoformat(),
+        summary=CouponDataSummary(**result["summary"]),
+        daily=[CouponDataDaily(**d) for d in result["daily"]],
+        hourly=[CouponDataHourly(**h) for h in result["hourly"]],
+        total=result["total"],
+        items=[CouponDataRow(**r) for r in result["items"]],
+    )
+
+
+@router.get(
+    "/user-records",
+    response_model=CouponUserRecordsOut,
+    summary="某用户全部领券记录(点手机号抽屉:领券次数 + 记录列表)",
+)
+def get_user_coupon_records(
+    db: AdminDb,
+    user_id: Annotated[int, Query(description="用户 id")],
+    limit: Annotated[int, Query(ge=1, le=500, description="最多返回条数")] = 100,
+    sort_by: Annotated[str, Query(description="兼容 UserRecordsDrawer 参数;固定按发起时间倒序")] = "created_at",
+    sort_order: Annotated[str, Query(description="兼容参数,忽略")] = "desc",
+) -> CouponUserRecordsOut:
+    result = coupon_data.coupon_user_records(db, user_id=user_id, limit=limit)
+    return CouponUserRecordsOut(
+        items=[CouponDataRow(**r) for r in result["items"]],
+        total=result["total"],
+    )

app/admin/routers/feedback.py

@@ -36,6 +36,8 @@ def _ensure_pending(fb: Feedback) -> None:
 def list_feedbacks(
     db: AdminDb,
     status: Annotated[str | None, Query()] = None,
+    # 反馈类型筛选:profile(普通反馈)/ comparison(比价反馈);None=全部
+    source: Annotated[str | None, Query(pattern="^(profile|comparison)$")] = None,
     user_id: Annotated[int | None, Query()] = None,
     content: Annotated[str | None, Query(max_length=100)] = None,
     created_from: Annotated[datetime | None, Query()] = None,
@@ -48,6 +50,7 @@ def list_feedbacks(
     items, next_cursor, total = queries.list_feedbacks(
         db,
         status=status,
+        source=source,
         user_id=user_id,
         content=content,
         created_from=created_from,
@@ -101,6 +104,7 @@ def approve_feedback(
         status="adopted",
         reward_coins=payload.reward_coins,
         review_note=payload.note,
+        admin_reply=payload.reply,
         reviewed_by_admin_id=admin.id,
         commit=False,
     )
@@ -123,6 +127,7 @@ def approve_feedback(
             "after": "adopted",
             "reward_coins": payload.reward_coins,
             "note": payload.note,
+            "reply": payload.reply,
         },
         ip=get_client_ip(request),
         commit=False,
@@ -152,6 +157,7 @@ def reject_feedback(
         status="rejected",
         reject_reason=payload.reason,
         review_note=payload.note,
+        admin_reply=payload.reply,
         reviewed_by_admin_id=admin.id,
         commit=False,
     )
@@ -166,6 +172,7 @@ def reject_feedback(
             "after": "rejected",
             "reason": payload.reason,
             "note": payload.note,
+            "reply": payload.reply,
         },
         ip=get_client_ip(request),
         commit=False,

app/admin/routers/withdraw.py

@@ -50,6 +50,8 @@ def list_withdraws(
     db: AdminDb,
     user_id: Annotated[int | None, Query()] = None,
     status: Annotated[str | None, Query()] = None,
+    # 提现类型筛选:coin_cash(福利页提现)/ invite_cash(邀请提现);None=全部
+    source: Annotated[str | None, Query(pattern="^(coin_cash|invite_cash)$")] = None,
     keyword: Annotated[str | None, Query(max_length=100)] = None,
     date_from: Annotated[datetime | None, Query()] = None,
     date_to: Annotated[datetime | None, Query()] = None,
@@ -69,6 +71,7 @@ def list_withdraws(
         db,
         user_id=user_id,
         status=status,
+        source=source,
         keyword=keyword,
         date_from=date_from,
         date_to=date_to,

app/admin/schemas/ad_revenue.py

@@ -94,6 +94,11 @@ class AdRevenueRow(BaseModel):
     impressions: int = Field(..., description="本行展示条数:有展示=1 / 纯发奖=0(供日汇总、趋势图复用)")
     ecpm: str | None = Field(None, description="eCPM 原始值(分/千次);展示行取展示值,纯发奖行取发奖采用值")
     revenue_yuan: float = Field(..., description="本次展示预估收益(元)= eCPM元 ÷ 1000;纯发奖行=0")
+    row_revenue_yuan: float | None = Field(
+        None,
+        description="主表逐行展示用的预估收益(元):一次比价/领券聚合行=该次发奖广告 eCPM 折算之和;"
+        "其它行为空(前端回退取 revenue_yuan)。不进合计/趋势,避免与展示侧重复计",
+    )
     adn: str | None = Field(None, description="实际填充 ADN 子渠道(pangle/gdt…);纯发奖行为空")
     slot_id: str | None = Field(None, description="底层 mediation rit(非我们配置的广告位 ID);纯发奖行为空")
     # ── 发奖侧 ──
@@ -106,6 +111,15 @@ class AdRevenueRow(BaseModel):
         None,
         description="发奖复算明细(eCPM/因子1/份数/LT/因子2/应发/实发/一致);点行展开下钻用,纯展示为空",
     )
+    sub_rewards: list[AdRevenueRecord] = Field(
+        default_factory=list,
+        description="一次比价/领券聚合行的组内逐条发奖明细(同一整场 ad_session_id 的多条广告);"
+        "点行展开渲染多行。激励视频/纯展示行为空(单条看 reward_detail)",
+    )
+    sub_count: int = Field(
+        1,
+        description="本行聚合的发奖条数:一次比价/领券=该次广告条数(≥1);激励视频/纯展示=1",
+    )
 
 
 class AdRevenueReportOut(BaseModel):

app/admin/schemas/coupon_data.py

@@ -0,0 +1,83 @@
+"""admin「领券数据」看板 schemas:汇总卡 + 按天/小时趋势 + 逐条领券明细。
+
+数据源 coupon_session(一次领券一行)。耗时单位 ms(前端按需折秒);均值/分位只统计 completed。
+"""
+from __future__ import annotations
+
+from datetime import datetime
+
+from pydantic import BaseModel, Field
+
+
+class CouponDataSummary(BaseModel):
+    """汇总卡:发起/完成数 + 耗时均值与分位(P5/P50/P95/P99,基于 completed 的 elapsed_ms)。"""
+
+    started_count: int = Field(..., description="发起数(区间内所有领券 session)")
+    completed_count: int = Field(..., description="完成数(status=completed)")
+    avg_elapsed_ms: int | None = Field(None, description="平均耗时(ms,仅 completed;无数据为空)")
+    p5_ms: int | None = Field(None, description="耗时 5 分位(ms)")
+    p50_ms: int | None = Field(None, description="耗时 50 分位(ms,中位数)")
+    p95_ms: int | None = Field(None, description="耗时 95 分位(ms)")
+    p99_ms: int | None = Field(None, description="耗时 99 分位(ms)")
+
+
+class CouponDataDaily(BaseModel):
+    """按天趋势(全量,不受分页影响):柱=发起/完成数,线=平均耗时。"""
+
+    date: str = Field(..., description="北京时间 YYYY-MM-DD")
+    started_count: int
+    completed_count: int
+    avg_elapsed_ms: int | None = Field(None, description="当天平均耗时(ms,仅 completed)")
+
+
+class CouponDataHourly(BaseModel):
+    """按北京小时(0–23)趋势(单日 granularity=hour 时非空)。"""
+
+    hour: int = Field(..., description="北京时间小时 0–23")
+    started_count: int
+    completed_count: int
+    avg_elapsed_ms: int | None = None
+
+
+class CouponDataRow(BaseModel):
+    """一条领券明细(一次领券任务)。"""
+
+    id: int = Field(..., description="coupon_session 主键(抽屉 rowKey 用)")
+    trace_id: str
+    user_id: int | None = None
+    user_phone: str | None = Field(None, description="手机号(admin 展示;匿名领券/查不到为空)")
+    user_nickname: str | None = Field(None, description="昵称")
+    status: str = Field(..., description="started / completed / failed / abandoned")
+    platforms: list[str] | None = Field(None, description="发起勾选平台")
+    origin_package: str | None = Field(None, description="发起来源 App 包名;null=App 内(傻瓜比价首页)发起")
+    elapsed_ms: int | None = Field(None, description="全程耗时(ms)")
+    platform_elapsed: dict[str, int] | None = Field(
+        None, description="各平台耗时 {meituan-waimai/taobao-shanguang/jd-waimai: ms}"
+    )
+    device_model: str | None = None
+    rom: str | None = None
+    app_env: str | None = None
+    started_at: datetime = Field(..., description="发起时刻(明细「时间」列)")
+    claimed_count: int | None = None
+    trace_url: str | None = Field(None, description="pricebot 公网 trace 链接(仅 completed 有);admin 渲染可点链接,无则显示可复制 trace_id")
+
+
+class CouponDataOut(BaseModel):
+    """领券数据看板响应:汇总卡 + 趋势 + 明细分页。"""
+
+    date_from: str
+    date_to: str
+    summary: CouponDataSummary
+    daily: list[CouponDataDaily] = Field(default_factory=list, description="按天趋势(全量)")
+    hourly: list[CouponDataHourly] = Field(
+        default_factory=list, description="按小时趋势(单日 hour 粒度时非空)"
+    )
+    total: int = Field(..., description="明细总条数(全量,不受分页)")
+    items: list[CouponDataRow] = Field(..., description="逐条领券明细(当前页)")
+
+
+class CouponUserRecordsOut(BaseModel):
+    """某用户全部领券记录(点手机号抽屉用):total=该用户领券总次数,items=记录列表(UserRecordsDrawer 渲染)。"""
+
+    items: list[CouponDataRow]
+    total: int

app/admin/schemas/device.py

@@ -24,7 +24,9 @@ class DeviceLivenessItem(BaseModel):
     device_model: str | None = None      # 由 device_id 解析(device_<机型>_<hash>);非 DB 列
     platform: str
     app_version: str | None = None
-    registration_id: str | None = None  # 非空 = 拿到极光 token、可推送
+    registration_id: str | None = None  # 旧极光字段,仅兼容历史数据
+    push_vendor: str | None = None
+    push_token: str | None = None
 
     ever_protected: bool                 # 是否开过无障碍(=该设备对功能有意义)
     first_protected_at: datetime | None = None   # 首次开无障碍时刻(老设备为 null)

app/admin/schemas/feedback.py

@@ -15,11 +15,17 @@ class FeedbackOut(BaseModel):
     user_id: int
     content: str
     contact: str
+    # 反馈来源入口:profile(「我的」页)/ comparison(比价结果页)——admin 展示/筛选「反馈类型」
+    source: str = "profile"
+    # 比价反馈的问题场景(找错商品/优惠不对…);普通反馈为 None
+    scene: str | None = None
     images: list[str] | None = None
     status: str
     reject_reason: str | None = None
     reward_coins: int | None = None
     review_note: str | None = None
+    # 运营给用户的回复留言(用户端可见,采纳/未采纳都可填)
+    admin_reply: str | None = None
     reviewed_by_admin_id: int | None = None
     reviewed_at: datetime | None = None
     created_at: datetime
@@ -39,12 +45,14 @@ class FeedbackApproveRequest(BaseModel):
         le=FEEDBACK_REWARD_MAX_COINS,
         description="采纳后发放金币数",
     )
-    note: str | None = Field(default=None, max_length=256, description="采纳要点/审核备注")
+    note: str | None = Field(default=None, max_length=256, description="采纳要点/审核备注(内部)")
+    reply: str | None = Field(default=None, max_length=256, description="给用户的回复留言,用户端可见")
 
 
 class FeedbackRejectRequest(BaseModel):
     reason: str = Field(min_length=1, max_length=256, description="未采纳原因,用户端可见")
     note: str | None = Field(default=None, max_length=256, description="运营内部审核备注")
+    reply: str | None = Field(default=None, max_length=256, description="给用户的回复留言,用户端可见")
 
 
 class FeedbackSummary(BaseModel):

app/admin/schemas/user.py

@@ -17,6 +17,7 @@ class AdminUserListItem(BaseModel):
     status: str
     debug_trace_enabled: bool = False
     wechat_openid: str | None = None
+    wechat_nickname: str | None = None
     created_at: datetime
     last_login_at: datetime
 

app/admin/schemas/wallet.py

@@ -41,6 +41,8 @@ class WithdrawOrderOut(BaseModel):
     user_id: int
     out_bill_no: str
     amount_cents: int
+    # 提现类型:coin_cash(福利页提现,金币兑换的现金)/ invite_cash(邀请提现,邀请奖励金)
+    source: str = "coin_cash"
     user_name: str | None = None  # 提现实名(审核核对 + 打款用)
     status: str
     wechat_state: str | None = None

app/api/v1/coupon.py

@@ -29,6 +29,7 @@ from app.schemas.coupon_state import (
     CouponPromptDismissIn,
     CouponPromptShouldShowOut,
     CouponPromptShownIn,
+    CouponSessionIn,
     CouponStatsOut,
 )
 
@@ -194,6 +195,34 @@ async def coupon_step(
     return resp_json
 
 
+@router.post("/session", summary="领券任务流水上报(admin 领券数据看板数据源)")
+def coupon_session(payload: CouponSessionIn, db: DbSession) -> dict[str, bool]:
+    """客户端两段上报一次领券流水(发起 started / 收尾 completed-failed-abandoned),按 trace_id upsert
+    到 coupon_session。不鉴权(同领券循环 MVP,按 device_id/trace_id);供 admin「领券数据」看板算
+    发起/完成数、耗时分位、机型维度。写库失败不应连累客户端(本就 fire-and-forget),吞掉返回 ok。"""
+    try:
+        coupon_repo.upsert_coupon_session(
+            db,
+            trace_id=payload.trace_id,
+            device_id=payload.device_id,
+            status=payload.status,
+            started_at_ms=payload.started_at_ms,
+            user_id=payload.user_id,
+            platforms=payload.platforms,
+            origin_package=payload.origin_package,
+            device_model=payload.device_model,
+            rom=payload.rom,
+            app_env=payload.app_env,
+            elapsed_ms=payload.elapsed_ms,
+            platform_elapsed=payload.platform_elapsed,
+            claimed_count=payload.claimed_count,
+            trace_url=payload.trace_url,
+        )
+    except Exception as e:  # noqa: BLE001
+        logger.warning("coupon session write failed: %s", e)
+    return {"ok": True}
+
+
 @router.post("/prompt/shown", summary="领券引导窗弹出即上报(按 App 记 shown)")
 def coupon_prompt_shown(payload: CouponPromptShownIn, db: DbSession) -> dict[str, bool]:
     """客户端弹出引导窗那刻调 → 记一条今日 engagement(shown),今天**这个 App** 不再自动弹。

app/api/v1/device.py

@@ -1,19 +1,22 @@
 """设备注册 / 心跳 endpoint(无障碍保护存活检测)。
 
 路由前缀 /api/v1/device,需 Bearer 鉴权(设备绑登录用户)。
-  POST /register   注册设备 / 更新 registration_id(App 前台、拿到 push token 时调)
+  POST /register   注册设备 / 更新厂商 push token(App 前台、拿到 push token 时调)
   POST /heartbeat  上报心跳(无障碍服务存活时周期调,刷新存活)
+  POST /push-test  开发验收:延迟发送厂商通道测试推送
 
-后端 heartbeat_monitor_worker 据此发现心跳超时的设备并极光推送告警。
+后端 heartbeat_monitor_worker 据此发现心跳超时的设备并厂商直推告警。
 见 spec: spec/accessibility-liveness-push.md。
 """
 from __future__ import annotations
 
 import logging
+import time
 
-from fastapi import APIRouter
+from fastapi import APIRouter, BackgroundTasks, HTTPException, status
 
 from app.api.deps import CurrentUser, DbSession
+from app.integrations import vendor_push
 from app.repositories import device as device_repo
 from app.schemas.device import (
     DeviceOut,
@@ -22,6 +25,8 @@ from app.schemas.device import (
     LivenessAckRequest,
     LivenessOut,
     OkResponse,
+    PushTestOut,
+    PushTestRequest,
 )
 
 logger = logging.getLogger("shagua.device")
@@ -29,6 +34,37 @@ logger = logging.getLogger("shagua.device")
 router = APIRouter(prefix="/api/v1/device", tags=["device"])
 
 
+def _send_push_test_after_delay(
+    push_vendor: str,
+    push_token: str,
+    delay_seconds: int,
+    user_id: int,
+    device_id: str,
+) -> None:
+    if delay_seconds > 0:
+        time.sleep(delay_seconds)
+    try:
+        vendor_push.send_accessibility_disabled(
+            push_vendor,
+            push_token,
+            title="测试推送",
+            alert="这是一条厂商通道测试推送。收到它说明 App 被划掉后仍可通过系统通知栏触达。",
+        )
+        logger.info(
+            "push test sent user_id=%d device_id=%s delay=%ds",
+            user_id,
+            device_id,
+            delay_seconds,
+        )
+    except vendor_push.VendorPushError as e:
+        logger.warning(
+            "push test failed user_id=%d device_id=%s error=%s",
+            user_id,
+            device_id,
+            e,
+        )
+
+
 @router.post("/register", response_model=DeviceOut, summary="注册设备/更新推送token")
 def register_device(
     req: DeviceRegisterRequest,
@@ -40,13 +76,17 @@ def register_device(
         user_id=user.id,
         device_id=req.device_id,
         registration_id=req.registration_id,
+        push_vendor=req.push_vendor,
+        push_token=req.push_token,
         platform=req.platform,
         app_version=req.app_version,
     )
     logger.info(
-        "device register user_id=%d device_id=%s reg=%s",
+        "device register user_id=%d device_id=%s vendor=%s token=%s legacy_reg=%s",
         user.id,
         req.device_id,
+        req.push_vendor,
+        bool(req.push_token),
         bool(req.registration_id),
     )
     return DeviceOut.model_validate(device)
@@ -64,10 +104,59 @@ def report_heartbeat(
         device_id=req.device_id,
         accessibility_enabled=req.accessibility_enabled,
         registration_id=req.registration_id,
+        push_vendor=req.push_vendor,
+        push_token=req.push_token,
     )
     return OkResponse()
 
 
+@router.post("/push-test", response_model=PushTestOut, summary="延迟发送厂商通道测试推送")
+def request_push_test(
+    req: PushTestRequest,
+    background_tasks: BackgroundTasks,
+    user: CurrentUser,
+    db: DbSession,
+) -> PushTestOut:
+    """开发验收用:App 内点一次,服务端延迟发厂商直推,验证离线通道。"""
+    push_vendor = req.push_vendor.strip() if req.push_vendor else None
+    push_token = req.push_token.strip() if req.push_token else None
+    if push_vendor and push_token:
+        device_repo.register_or_update(
+            db,
+            user_id=user.id,
+            device_id=req.device_id,
+            registration_id=req.registration_id,
+            push_vendor=push_vendor,
+            push_token=push_token,
+        )
+    else:
+        device = device_repo.get_device(db, user_id=user.id, device_id=req.device_id)
+        push_vendor = device.push_vendor if device is not None else None
+        push_token = device.push_token if device is not None else None
+
+    if not push_vendor or not push_token:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail="push vendor token not ready",
+        )
+
+    background_tasks.add_task(
+        _send_push_test_after_delay,
+        push_vendor,
+        push_token,
+        req.delay_seconds,
+        user.id,
+        req.device_id,
+    )
+    logger.info(
+        "push test scheduled user_id=%d device_id=%s delay=%ds",
+        user.id,
+        req.device_id,
+        req.delay_seconds,
+    )
+    return PushTestOut(delay_seconds=req.delay_seconds, has_push_token=True)
+
+
 @router.get("/liveness", response_model=LivenessOut, summary="查询本机掉线告警(后置检测)")
 def get_liveness(
     device_id: str,

app/api/v1/feedback.py

@@ -31,7 +31,18 @@ router = APIRouter(prefix="/api/v1/feedback", tags=["feedback"])
 _MAX_IMAGES = 6
 _CONTENT_MAX = 200
 _CONTACT_MAX = 128
+_SCENE_MAX = 32
 _VALID_RECORD_STATUS = {"pending", "adopted", "rejected"}
+_VALID_SOURCES = {"profile", "comparison"}
+
+
+def _resolve_source(source: str, scene: str) -> str:
+    """反馈来源:客户端显式传的 source 优先(profile / comparison);未传或非法时,
+    按 scene 有无派生——比价结果页反馈才带 scene,故 scene 非空即视作 comparison。"""
+    src = source.strip().lower()
+    if src in _VALID_SOURCES:
+        return src
+    return "comparison" if scene.strip() else "profile"
 
 
 def _app_status(db_status: str) -> str:
@@ -46,10 +57,12 @@ def _record_out(fb) -> FeedbackRecordOut:
     return FeedbackRecordOut(
         id=fb.id,
         content=fb.content,
+        scene=getattr(fb, "scene", None),
         images=fb.images or [],
         status=_app_status(fb.status),
         reject_reason=getattr(fb, "reject_reason", None),
         reward_coins=getattr(fb, "reward_coins", None),
+        admin_reply=getattr(fb, "admin_reply", None),
         created_at=fb.created_at,
     )
 
@@ -61,6 +74,11 @@ async def submit_feedback(
     content: str = Form(...),
     # 原型改版后客户端不再采集联系方式;保留字段以兼容旧端 + 后续可能复用,默认空串。
     contact: str = Form(default=""),
+    # 反馈来源入口:profile(「我的」页)/ comparison(比价结果页)。新端显式传;旧端不带 →
+    # 空串 → 按 scene 有无派生。admin 据此展示/筛选「反馈类型」。
+    source: str = Form(default=""),
+    # 比价结果页反馈的「问题场景」(找错商品/优惠不对…);普通反馈不带 → 空串 → 存 NULL。
+    scene: str = Form(default=""),
     # 提交端环境快照(admin 反馈页展示「提交版本号」「机型OS版本」);旧端不带 → 空串 → 存 NULL。
     app_version: str = Form(default=""),
     device_model: str = Form(default=""),
@@ -76,6 +94,8 @@ async def submit_feedback(
         raise HTTPException(status_code=400, detail="反馈内容过长")
     if len(contact) > _CONTACT_MAX:
         raise HTTPException(status_code=400, detail="联系方式过长")
+    scene = scene.strip()[:_SCENE_MAX]
+    resolved_source = _resolve_source(source, scene)
 
     files = [f for f in (images or []) if f is not None and f.filename]
     if len(files) > _MAX_IMAGES:
@@ -91,10 +111,13 @@ async def submit_feedback(
 
     fb = feedback_repo.create_feedback(
         db, user_id=user.id, content=content, contact=contact, images=urls,
+        source=resolved_source, scene=scene or None,
         app_version=app_version.strip(), device_model=device_model.strip(),
         rom_name=rom_name.strip(), android_version=android_version.strip(),
     )
-    logger.info("feedback id=%d user_id=%d images=%d", fb.id, user.id, len(urls))
+    logger.info(
+        "feedback id=%d user_id=%d source=%s images=%d", fb.id, user.id, resolved_source, len(urls)
+    )
     return FeedbackOut.model_validate(fb)
 
 

app/api/v1/notifications.py

@@ -0,0 +1,120 @@
+"""消息通知中心 endpoint(PRD《消息通知中心》)。
+
+路由前缀 `/api/v1/notifications`,需 Bearer 鉴权(消息按用户隔离)。
+  GET  /              消息列表(分页;全列表时间倒序,不分组——PRD 原文的分组已取消)
+  GET  /unread-count  未读总数(首页铃铛角标)
+  POST /read          标记已读({ids:[...]} 单条/多条 或 {all:true} 全量清零)
+
+⚠️ 虚拟数据阶段:数据来自 repositories/notification_mock.py 内存 mock(重启复位,
+每个用户一套覆盖全部 13 种类型的固定样例)。接真实数据时只换 repository 实现。
+
+⚠️ 字段命名:本组接口对外为 **camelCase**(sentAt / isRead / pageSize…,PRD 前端契约),
+详见 schemas/notification.py 顶部说明。
+"""
+from __future__ import annotations
+
+import logging
+
+from fastapi import APIRouter, HTTPException, Query
+
+from app.api.deps import CurrentUser
+from app.core import notification_catalog as catalog
+from app.repositories import notification_mock as mock_repo
+from app.schemas.notification import (
+    InfoRow,
+    MarkReadOut,
+    MarkReadRequest,
+    NotificationItem,
+    NotificationListOut,
+    UnreadCountOut,
+)
+
+logger = logging.getLogger("shagua.notifications")
+
+router = APIRouter(prefix="/api/v1/notifications", tags=["notifications"])
+
+
+def _to_item(n: mock_repo.MockNotification) -> NotificationItem:
+    """mock 记录 + 类型静态目录 → 接口出参。"""
+    ntype = catalog.get_type(n.type_key)
+    return NotificationItem(
+        id=n.id,
+        category=ntype.category,
+        category_label=catalog.category_label(ntype.category),
+        type=ntype.key,
+        card_style=ntype.card_style,
+        title=ntype.card_title,
+        coins=n.coins,
+        cash_cents=n.cash_cents,
+        cash_yuan=mock_repo.cash_yuan(n.cash_cents),
+        info_rows=[InfoRow(**row) for row in n.info_rows],
+        action_text=ntype.action_text,
+        extra=n.extra,
+        sent_at=n.sent_at,
+        is_read=n.is_read,
+    )
+
+
+@router.get("", response_model=NotificationListOut, summary="消息列表(分页)")
+def list_notifications(
+    user: CurrentUser,
+    page: int = Query(default=1, ge=1, description="页码,1 起"),
+    page_size: int = Query(
+        default=20, ge=1, le=100, alias="pageSize", description="每页条数,默认 20,最大 100"
+    ),
+) -> NotificationListOut:
+    """通知中心消息列表。
+
+    - 排序服务端已做好:**全列表按时间倒序**(最新在前,不做分类分组;PRD §1 的
+      "按分类分组"为笔误,已与需求方确认取消),前端按返回顺序渲染即可。
+    - 每条的字段构成与各版式说明见 NotificationItem schema。
+    - 响应同时带 unreadCount,进页面时可顺手刷新角标。
+    - 虚拟数据阶段:首次请求生成一套固定样例(13 类型全覆盖 + 今天/昨天/当年/跨年
+      四种时间 + 已读未读混合),已读状态在服务端内存维护,重启服务复位。
+    """
+    items, total, unread = mock_repo.list_notifications(user.id, page=page, page_size=page_size)
+    return NotificationListOut(
+        items=[_to_item(n) for n in items],
+        page=page,
+        page_size=page_size,
+        total=total,
+        has_more=page * page_size < total,
+        unread_count=unread,
+    )
+
+
+@router.get("/unread-count", response_model=UnreadCountOut, summary="未读总数(铃铛角标)")
+def get_unread_count(user: CurrentUser) -> UnreadCountOut:
+    """首页铃铛角标数据源。刷新时机(PRD §4):进入首页时、从通知中心/其他页面返回首页时。
+
+    - count:精确未读条数;
+    - badgeText:直接可展示的角标文案——超过 99 返回 "99+",等于 0 返回 null(隐藏整个角标)。
+    """
+    count = mock_repo.unread_count(user.id)
+    badge = None if count == 0 else ("99+" if count > 99 else str(count))
+    return UnreadCountOut(count=count, badge_text=badge)
+
+
+@router.post("/read", response_model=MarkReadOut, summary="标记已读(单条/多条/全量)")
+def mark_read(req: MarkReadRequest, user: CurrentUser) -> MarkReadOut:
+    """红点消除(PRD §4),两种调用模式:
+
+    1. `{"ids": [90001]}` —— 点击某张消息卡片(无论点击后是跳转/弹窗/无动作都算已读);
+       用户点击 push 直达落地页时,客户端也用它把对应站内消息同步置读(push extras 里带
+       notificationId);
+    2. `{"all": true}` —— 进入通知中心自动清零(只是浏览列表就消红点,无需逐条点击)。
+
+    幂等:不存在或已读的 id 忽略;重复调用 markedCount 为 0、不报错。
+    响应带 unreadCount(处理后剩余未读),可直接刷新铃铛角标。
+    """
+    if not req.all and not req.ids:
+        raise HTTPException(status_code=400, detail="ids 与 all 至少传一个:{ids:[...]} 或 {all:true}")
+    marked, unread = mock_repo.mark_read(user.id, ids=req.ids, mark_all=req.all)
+    logger.info(
+        "notifications read user_id=%d mode=%s marked=%d unread_left=%d",
+        user.id,
+        "all" if req.all else f"ids×{len(req.ids or [])}",
+        marked,
+        unread,
+    )
+    return MarkReadOut(ok=True, marked_count=marked, unread_count=unread)

app/api/v1/push.py

@@ -0,0 +1,186 @@
+"""厂商推送 测试/联调 endpoint。
+
+路由前缀 `/api/v1/push`,需 Bearer 鉴权。围绕「消息中心 13 类通知的厂商直推」提供三件套:
+  GET  /vendors    5 个厂商(荣耀/华为/小米/OPPO/vivo)服务端凭据配置状态,缺哪些键一目了然
+  GET  /templates  13 种通知类型的 push 标题/正文模板 + PRD 示例渲染效果
+  POST /test       测试发送:默认 mock(不真调厂商 API,回显渲染结果);mock=false 真发到手机
+
+与 `/api/v1/device/push-test`(无障碍召回通道的延迟自测)互补:本组面向消息中心 13 类
+push 的文案/参数/厂商通道联调。真实业务触发(奖励过期、提现回执等)接入后走
+integrations.vendor_push.send_notification,与本测试端点同一条发送链路。
+"""
+from __future__ import annotations
+
+import logging
+
+from fastapi import APIRouter, HTTPException, status
+
+from app.api.deps import CurrentUser, DbSession
+from app.core import notification_catalog as catalog
+from app.integrations import vendor_push
+from app.repositories import device as device_repo
+from app.repositories import notification_mock as mock_repo
+from app.schemas.push import (
+    PushTemplateOut,
+    PushTemplatesOut,
+    PushTestOut,
+    PushTestRequest,
+    PushVendorsOut,
+    PushVendorStatus,
+)
+
+logger = logging.getLogger("shagua.push")
+
+router = APIRouter(prefix="/api/v1/push", tags=["push"])
+
+# /vendors 的展示顺序(荣耀/华为/小米/OPPO/vivo)
+_VENDOR_ORDER = ("honor", "huawei", "xiaomi", "oppo", "vivo")
+
+_GENERIC_TEST_TITLE = "傻瓜比价测试推送"
+_GENERIC_TEST_BODY = "这是一条{label}通道的测试推送,收到说明服务端 → {label}厂商通道已打通。"
+
+
+@router.get("/vendors", response_model=PushVendorsOut, summary="厂商推送配置状态")
+def vendor_status(user: CurrentUser) -> PushVendorsOut:
+    """检查 5 个厂商的服务端推送凭据是否配齐(读 .env,不打厂商接口)。
+
+    missingKeys 列出的即还需要在 .env 里补的配置键;全空说明该厂商随时可真发。
+    mock 测试(POST /test 默认模式)不依赖任何凭据。
+    """
+    return PushVendorsOut(
+        vendors=[
+            PushVendorStatus(
+                vendor=v,
+                label=vendor_push.VENDOR_LABELS[v],
+                configured=not vendor_push.missing_settings(v),
+                missing_keys=vendor_push.missing_settings(v),
+            )
+            for v in _VENDOR_ORDER
+        ]
+    )
+
+
+@router.get("/templates", response_model=PushTemplatesOut, summary="13 类通知的 push 模板预览")
+def push_templates(user: CurrentUser) -> PushTemplatesOut:
+    """PRD §5 的 13 条 push 文案模板 + 用示例值渲染后的效果,联调对文案用。
+
+    标题固定(≤11 字不带变量);正文里 {var} 为变量,POST /test 的 vars 字段可覆盖。
+    """
+    templates: list[PushTemplateOut] = []
+    for key, ntype in catalog.TYPES.items():
+        title, body_sample = catalog.render_push(key)
+        templates.append(
+            PushTemplateOut(
+                type=key,
+                category=ntype.category,
+                category_label=catalog.category_label(ntype.category),
+                card_style=ntype.card_style,
+                push_title=title,
+                push_body_sample=body_sample,
+                push_body_template=ntype.push_body_template,
+                variables=catalog.push_variable_names(key),
+                sample_vars=ntype.sample_vars,
+            )
+        )
+    return PushTemplatesOut(templates=templates)
+
+
+@router.post("/test", response_model=PushTestOut, summary="测试发送厂商推送(默认 mock)")
+def send_test_push(req: PushTestRequest, user: CurrentUser, db: DbSession) -> PushTestOut:
+    """向指定厂商 token(或本用户已注册设备)发一条测试 push。
+
+    - **mock=true(默认)**:不真调厂商 API——校验参数、渲染文案后原样返回,并在
+      missingKeys 里提示真发前还缺哪些配置。虚拟数据阶段随便打,不会骚扰真机。
+    - **mock=false**:真发。要求该厂商凭据已配置(缺则 400 报缺失键);厂商 API 报错回 502。
+      注意 vivo 未上架前是测试推送模式(VIVO_PUSH_MODE=1),目标手机要先在 vivo 后台加为测试设备。
+    - **createNotification=true**:同时往消息中心 mock 列表插一条同类型未读通知并把
+      notificationId 放进 push extras → 客户端点击 push 后调 POST /notifications/read
+      {ids:[notificationId]} 即可闭环验证 PRD §4 的 push 已读联动。
+    """
+    # ---- 1. 解析推送目标(vendor + token):直填优先,缺则按 deviceId 反查已注册设备 ----
+    vendor_raw = req.vendor.strip()
+    push_token = req.push_token.strip()
+    if (not vendor_raw or not push_token) and req.device_id.strip():
+        device = device_repo.get_device(db, user_id=user.id, device_id=req.device_id.strip())
+        if device is not None:
+            vendor_raw = vendor_raw or (device.push_vendor or "")
+            push_token = push_token or (device.push_token or "")
+
+    vendor = vendor_push.normalize_vendor(vendor_raw)
+    if not vendor or vendor not in vendor_push.SUPPORTED_VENDORS:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=f"vendor 无效或无法从设备推断,支持: {', '.join(_VENDOR_ORDER)}",
+        )
+    if not push_token:
+        # mock 模式给个占位 token,让「只想看看渲染结果」的调用免造数据;真发必须给真 token。
+        if req.mock:
+            push_token = "mock-token"
+        else:
+            raise HTTPException(
+                status_code=status.HTTP_409_CONFLICT,
+                detail="push token 未知:请直传 pushToken,或先用该设备调 /api/v1/device/register 上报",
+            )
+
+    # ---- 2. 组装文案与 extras:直填 > type 模板 > 通用测试文案 ----
+    extras: dict[str, str] = {}
+    notification_id: int | None = None
+    if req.type:
+        try:
+            title, body = catalog.render_push(req.type, req.vars or None)
+        except catalog.UnknownNotificationType as e:
+            raise HTTPException(status_code=400, detail=str(e)) from e
+        extras["type"] = req.type
+        if req.create_notification:
+            item = mock_repo.insert_sample(user.id, req.type)
+            notification_id = item.id
+            extras.update({str(k): str(v) for k, v in item.extra.items()})
+            extras["notificationId"] = str(item.id)
+    else:
+        label = vendor_push.VENDOR_LABELS[vendor]
+        title = _GENERIC_TEST_TITLE
+        body = _GENERIC_TEST_BODY.format(label=label)
+        extras["type"] = "push_test"
+    if req.title.strip():
+        title = req.title.strip()
+    if req.content.strip():
+        body = req.content.strip()
+
+    # ---- 3. 发送(mock / 真发) ----
+    missing = vendor_push.missing_settings(vendor)
+    vendor_response = None
+    if req.mock:
+        vendor_push.send_notification(
+            vendor, push_token, title=title, body=body, extras=extras, mock=True
+        )
+    else:
+        if missing:
+            raise HTTPException(
+                status_code=400,
+                detail=f"{vendor_push.VENDOR_LABELS[vendor]}推送凭据未配置,先在 .env 补上: "
+                       f"{', '.join(missing)}",
+            )
+        try:
+            vendor_response = vendor_push.send_notification(
+                vendor, push_token, title=title, body=body, extras=extras
+            )
+        except vendor_push.VendorPushError as e:
+            raise HTTPException(
+                status_code=status.HTTP_502_BAD_GATEWAY, detail=f"厂商推送失败: {e}"
+            ) from e
+
+    logger.info(
+        "push test user_id=%d vendor=%s type=%s mock=%s notification_id=%s",
+        user.id, vendor, req.type or "generic", req.mock, notification_id,
+    )
+    return PushTestOut(
+        ok=True,
+        mock=req.mock,
+        vendor=vendor,
+        title=title,
+        body=body,
+        extras=extras,
+        notification_id=notification_id,
+        missing_keys=missing,
+        vendor_response=vendor_response,
+    )

app/core/config.py

@@ -66,9 +66,61 @@ class Settings(BaseSettings):
     JG_VERIFY_ENDPOINT: str = "https://api.verification.jpush.cn/v1/web/loginTokenVerify"
     JG_REQUEST_TIMEOUT_SEC: int = 15
 
-    # 无障碍保护存活监控后台任务(pull 后置检测;本期不接推送)
+    # ===== 厂商直推(无障碍保护存活告警)=====
+    ANDROID_PACKAGE_NAME: str = "com.jishisongfu.shaguabijia"
+    PUSH_REQUEST_TIMEOUT_SEC: int = 15
+    PUSH_TIME_TO_LIVE_SEC: int = 86400
+
+    HONOR_PUSH_APP_ID: str = ""
+    HONOR_PUSH_CLIENT_ID: str = ""
+    HONOR_PUSH_CLIENT_SECRET: str = ""
+    HONOR_PUSH_TOKEN_ENDPOINT: str = "https://iam.developer.honor.com/auth/token"
+    HONOR_PUSH_SEND_ENDPOINT_TEMPLATE: str = (
+        "https://push-api.cloud.honor.com/api/v1/{app_id}/sendMessage"
+    )
+
+    # 华为 Push Kit:AGC 控制台 → 项目设置 → 常规 → 应用,取 AppId + AppSecret
+    # (OAuth 换 token 时 client_id 即 AppId)。发送走 v1 messages:send,成功码 80000000。
+    HUAWEI_PUSH_APP_ID: str = ""
+    HUAWEI_PUSH_APP_SECRET: str = ""
+    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"
+    )
+
+    VIVO_PUSH_APP_ID: str = ""
+    VIVO_PUSH_APP_KEY: str = ""
+    VIVO_PUSH_APP_SECRET: str = ""
+    VIVO_PUSH_AUTH_ENDPOINT: str = "https://api-push.vivo.com.cn/message/auth"
+    VIVO_PUSH_SEND_ENDPOINT: str = "https://api-push.vivo.com.cn/message/send"
+    VIVO_PUSH_MODE: int = 1                  # 0=正式推送,1=测试推送(未上架 vivo 时用)
+    VIVO_PUSH_NOTIFY_TYPE: int = 4           # 1=无,2=响铃,3=振动,4=响铃+振动
+    VIVO_PUSH_CATEGORY: str = "DEVICE_REMINDER"
+
+    XIAOMI_PUSH_APP_SECRET: str = ""
+    XIAOMI_PUSH_SEND_ENDPOINT: str = "https://api.xmpush.xiaomi.com/v3/message/regid"
+    XIAOMI_PUSH_CHANNEL_ID: str = ""
+    XIAOMI_PUSH_TEMPLATE_ID: str = ""
+    XIAOMI_PUSH_TEMPLATE_TITLE: str = ""
+    XIAOMI_PUSH_TEMPLATE_DESCRIPTION: str = ""
+    XIAOMI_PUSH_TEMPLATE_PARAM_JSON: str = ""
+
+    OPPO_PUSH_APP_KEY: str = ""
+    OPPO_PUSH_MASTER_SECRET: str = ""
+    OPPO_PUSH_AUTH_ENDPOINT: str = "https://api.push.oppomobile.com/server/v1/auth"
+    OPPO_PUSH_SEND_ENDPOINT: str = (
+        "https://api.push.oppomobile.com/server/v1/message/notification/unicast"
+    )
+    # OPPO 新消息分类(2024-11-20 后创建的应用必须携带,否则可能被拒/限):
+    # channel_id=通知栏通道(OPPO 后台「通道ID」),category=消息分类 code(如 MARKETING 内容营销)。
+    # notify_level=提醒方式(0=不传走 OPPO 默认;内容营销类仅支持 1 通知栏/2 通知栏+锁屏)。
+    OPPO_PUSH_CHANNEL_ID: str = ""
+    OPPO_PUSH_CATEGORY: str = ""
+    OPPO_PUSH_NOTIFY_LEVEL: int = 0
+
+    # 无障碍保护存活监控后台任务(推送 + pull 后置兜底)
     HEARTBEAT_MONITOR_ENABLED: bool = True       # 总开关
-    HEARTBEAT_TIMEOUT_MINUTES: int = 10          # 多久没心跳算掉线(≈3 个客户端心跳周期)
+    HEARTBEAT_TIMEOUT_MINUTES: int = 60          # 多久没心跳算掉线(1 小时,避免短暂离线误判被杀)
     HEARTBEAT_SCAN_INTERVAL_SEC: int = 60        # 扫描周期
 
     # ===== 短信 =====

app/core/heartbeat_monitor_worker.py

@@ -1,7 +1,7 @@
 """无障碍保护存活监控后台任务。
 
 周期扫描「曾经保护过、当前 alive、心跳超时」的设备 = App 被彻底杀掉/无障碍已停(心跳断了),
-**命中即在服务器终端打印告警**(本期先不接推送,工程量大,用终端打印代替真实通知);并把状态机
+**命中即在服务器终端打印告警并尝试厂商直推**;并把状态机
 推进到 notified 防每轮重复打印(心跳恢复时由 repositories.device.touch_heartbeat 重置回 alive)。
 结构仿 withdraw_reconcile_worker(单实例锁 + asyncio 轮询 + 优雅退出)。
 
@@ -22,6 +22,7 @@ from sqlalchemy.exc import SQLAlchemyError
 
 from app.core.config import settings
 from app.db.session import SessionLocal
+from app.integrations import vendor_push
 from app.repositories import device as device_repo
 
 logger = logging.getLogger("shagua.heartbeat_monitor")
@@ -71,32 +72,66 @@ def _silent_seconds(last: datetime | None) -> int | None:
     """距上次心跳的秒数(兼容 sqlite 取回的 naive datetime)。"""
     if last is None:
         return None
-    ref = datetime.now(timezone.utc) if last.tzinfo is not None else datetime.utcnow()
+    ref = datetime.now(timezone.utc) if last.tzinfo is not None else datetime.utcnow()  # noqa: UP017
     return int((ref - last).total_seconds())
 
 
 def _scan_once(timeout_minutes: int) -> dict:
-    """扫描一轮:找出心跳超时(App 被彻底杀掉/无障碍已停)的设备,在**服务器终端打印**告警代替真实推送。
+    """扫描一轮:找出心跳超时(App 被彻底杀掉/无障碍已停)的设备并召回。
 
-    本期不接推送(极光/厂商通道工程量大),只做服务端掉线检测:命中即 logger.warning 打印到终端,
-    并把状态机推进到 notified 防每轮重复打印(心跳恢复时 touch_heartbeat 会重置回 alive)。
+    有 push_vendor + push_token 时先发厂商直推,无 token 或推送失败时仍置
+    kill_alert_pending,客户端下次进 App 继续走后置提醒兜底。
     """
     notified = 0
+    pushed = 0
+    push_failed = 0
     with SessionLocal() as db:
         overdue = device_repo.list_overdue(db, timeout_minutes=timeout_minutes)
         for device in overdue:
             silent = _silent_seconds(device.last_heartbeat_at)
             logger.warning(
                 "[掉线检测] user_id=%s device_id=%s 已 %s 秒无心跳(阈值 %d 分钟)"
-                " → 判定 App 已被杀/无障碍已停。【已置 kill_alert_pending: 用户下次进 App 将弹「开启自启动」引导(后置检测);推送本期未接】",
+                " → 判定 App 已被杀/无障碍已停。",
                 device.user_id,
                 device.device_id,
                 silent if silent is not None else "?",
                 timeout_minutes,
             )
+            if device.push_vendor and device.push_token:
+                try:
+                    vendor_push.send_accessibility_disabled(
+                        device.push_vendor,
+                        device.push_token,
+                    )
+                    pushed += 1
+                    logger.info(
+                        "[掉线检测] push sent user_id=%s device_id=%s vendor=%s",
+                        device.user_id,
+                        device.device_id,
+                        device.push_vendor,
+                    )
+                except vendor_push.VendorPushError as e:
+                    push_failed += 1
+                    logger.warning(
+                        "[掉线检测] push failed user_id=%s device_id=%s error=%s",
+                        device.user_id,
+                        device.device_id,
+                        e,
+                    )
+            else:
+                logger.info(
+                    "[掉线检测] device has no push vendor/token, skip push user_id=%s device_id=%s",
+                    device.user_id,
+                    device.device_id,
+                )
             device_repo.mark_notified(db, device_id_pk=device.id)
             notified += 1
-        return {"checked": len(overdue), "notified": notified}
+        return {
+            "checked": len(overdue),
+            "notified": notified,
+            "pushed": pushed,
+            "push_failed": push_failed,
+        }
 
 
 async def _run_loop() -> None:

app/core/notification_catalog.py

@@ -0,0 +1,241 @@
+"""消息通知中心:13 种通知类型的静态目录 + Push 文案模板。
+
+对应 PRD《消息通知中心》:§1 类型清单 / §3 字段元素 / §5 Push 文案。
+这里只放**静态定义**(分类、版式、标题、操作行、Push 模板),供两处消费:
+  - repositories/notification_mock.py  生成消息中心 mock 列表(虚拟数据阶段)
+  - api/v1/push.py                     渲染 13 类 push 标题/文案(厂商推送 + 测试端点)
+
+PRD 文案规范(§5):push 标题 ≤11 字、固定文案不带变量;变量只出现在正文里且尽量前置。
+模板变量用 `{name}` 占位,渲染时缺省回退 sample_vars(PRD 示例值),保证 mock 阶段随时可发。
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+
+# ---------------------------------------------------------------------------
+# 分类(仅作卡片头部的分类标签展示;列表不按分类分组——PRD §1 的分组已确认取消,全表时间倒序)
+# ---------------------------------------------------------------------------
+
+CATEGORY_WITHDRAW = "withdraw_assistant"
+CATEGORY_SYSTEM = "system"
+CATEGORY_FEEDBACK = "feedback"
+CATEGORY_REPORT = "report"
+CATEGORY_INVITE = "invite"
+
+# key → 中文标签
+CATEGORIES: dict[str, str] = {
+    CATEGORY_WITHDRAW: "提现助手",
+    CATEGORY_SYSTEM: "系统通知",
+    CATEGORY_FEEDBACK: "我的反馈",
+    CATEGORY_REPORT: "我的爆料",
+    CATEGORY_INVITE: "好友邀请",
+}
+
+
+def category_label(key: str) -> str:
+    return CATEGORIES[key]
+
+
+# ---------------------------------------------------------------------------
+# 卡片版式(PRD §3「版式」列;前端按此渲染五种卡)
+# ---------------------------------------------------------------------------
+
+CARD_DUAL_AMOUNT = "dual_amount"    # 双金额卡(金币数 + 现金数)
+CARD_WITHDRAW = "withdraw"          # 提现卡(¥金额)
+CARD_PLAIN_TEXT = "plain_text"      # 纯文本卡(无数值)
+CARD_COIN_REWARD = "coin_reward"    # 金币奖励卡(金币数 + 单位「金币」)
+CARD_FRIEND_CASH = "friend_cash"    # 好友现金卡(¥金额)
+
+
+@dataclass(frozen=True)
+class NotificationType:
+    """一种通知类型的静态定义(卡片元数据 + Push 模板)。"""
+
+    key: str                    # 类型 key(接口 type 字段;前端按它决定点击跳转,见 PRD §2)
+    category: str               # 分类 key(CATEGORIES 之一)
+    card_style: str             # 卡片版式(CARD_* 之一)
+    card_title: str             # 卡片标题(PRD §3「标题」列)
+    action_text: str | None     # 操作行文案;None = 无操作行(如「提现成功」)
+    push_title: str             # push 标题(≤11 字固定文案,PRD §5)
+    push_body_template: str     # push 正文模板,`{var}` 为变量
+    sample_vars: dict[str, str] = field(default_factory=dict)  # PRD 示例值,渲染缺省回退
+
+
+# 13 种类型,编号/文案与 PRD §1/§3/§5 一一对应(插入顺序 = PRD 编号顺序)。
+TYPES: dict[str, NotificationType] = {
+    t.key: t
+    for t in [
+        # -- 提现助手 -------------------------------------------------------
+        NotificationType(
+            key="reward_expiring",
+            category=CATEGORY_WITHDRAW,
+            card_style=CARD_DUAL_AMOUNT,
+            card_title="金币现金奖励即将失效",
+            action_text="立即激活您的收益",
+            push_title="您的奖励即将失效",
+            push_body_template="{coins}金币和{cash}元现金{days}天后失效，完成快来激活收益",
+            sample_vars={"coins": "86", "cash": "12.80", "days": "3"},
+        ),
+        NotificationType(
+            key="reward_expired",
+            category=CATEGORY_WITHDRAW,
+            card_style=CARD_DUAL_AMOUNT,
+            card_title="金币现金奖励已失效",
+            action_text="立即赚取新收益",
+            push_title="您的奖励已失效",
+            push_body_template="{coins}金币和{cash}元现金已过期，完成一次一键领券或一键比价可赚取新收益",
+            sample_vars={"coins": "35", "cash": "0.60"},
+        ),
+        NotificationType(
+            key="withdraw_success",
+            category=CATEGORY_WITHDRAW,
+            card_style=CARD_WITHDRAW,
+            card_title="提现成功",
+            action_text=None,  # PRD §3:提现成功卡无操作行,点击也无跳转、仅消红点
+            push_title="提现到账提醒",
+            push_body_template="¥{amount}已存入您的微信钱包，点击查看到账详情",
+            sample_vars={"amount": "0.50"},
+        ),
+        NotificationType(
+            key="withdraw_failed",
+            category=CATEGORY_WITHDRAW,
+            card_style=CARD_WITHDRAW,
+            card_title="提现失败，款项已退回",
+            action_text="重新提现",
+            push_title="提现失败，款项已退回",
+            push_body_template="¥{amount}因{reason}退回现金余额，点击重新提现",
+            sample_vars={"amount": "3.50", "reason": "微信零钱未实名"},
+        ),
+        # -- 系统通知(权限异常 ×4;标题里的功能名按类型写死,见 PRD §1/§3)----
+        NotificationType(
+            key="perm_accessibility",
+            category=CATEGORY_SYSTEM,
+            card_style=CARD_PLAIN_TEXT,
+            card_title="检测到您的比价功能已失效",
+            action_text="去开启",
+            push_title="检测到您的比价功能已失效",
+            push_body_template="未开启将导致核心功能不可用，请尽快来傻瓜比价开启",
+        ),
+        NotificationType(
+            key="perm_battery",
+            category=CATEGORY_SYSTEM,
+            card_style=CARD_PLAIN_TEXT,
+            card_title="检测到您的比价续航保护已失效",
+            action_text="去开启",
+            push_title="检测到您的比价续航保护已失效",
+            push_body_template="未开启将导致核心功能不可用，请尽快来傻瓜比价开启",
+        ),
+        NotificationType(
+            key="perm_autostart",
+            category=CATEGORY_SYSTEM,
+            card_style=CARD_PLAIN_TEXT,
+            card_title="检测到您的比价启动保护已失效",
+            action_text="去开启",
+            push_title="检测到您的比价启动保护已失效",
+            push_body_template="未开启将导致核心功能不可用，请尽快来傻瓜比价开启",
+        ),
+        NotificationType(
+            key="perm_overlay",
+            category=CATEGORY_SYSTEM,
+            card_style=CARD_PLAIN_TEXT,
+            card_title="检测到您的比价按钮已失效",
+            action_text="去开启",
+            push_title="检测到您的比价按钮已失效",
+            push_body_template="未开启将导致核心功能不可用，请尽快来傻瓜比价开启",
+        ),
+        # -- 我的反馈 -------------------------------------------------------
+        NotificationType(
+            key="feedback_reply",
+            category=CATEGORY_FEEDBACK,
+            card_style=CARD_PLAIN_TEXT,
+            card_title="傻瓜比价官方回复了您的反馈",
+            action_text="查看详情",
+            push_title="您的反馈有回复啦",
+            push_body_template="您提的建议我们认真看过了，来看看我们的回复吧～",
+        ),
+        NotificationType(
+            key="feedback_reward",
+            category=CATEGORY_FEEDBACK,
+            card_style=CARD_COIN_REWARD,
+            card_title="反馈奖励",
+            action_text="查看反馈详情",
+            push_title="反馈奖励已到账",
+            push_body_template="谢谢您帮傻瓜比价变得更好，{coins}金币已到账，还有一条给您的留言～",
+            sample_vars={"coins": "300"},
+        ),
+        # -- 我的爆料 -------------------------------------------------------
+        NotificationType(
+            key="report_approved",
+            category=CATEGORY_REPORT,
+            card_style=CARD_COIN_REWARD,
+            card_title="爆料审核通过",
+            action_text="查看爆料详情",
+            push_title="爆料审核通过",
+            push_body_template="您爆料的「{store}」更低价审核通过，{coins}金币已到账，感谢您的分享",
+            sample_vars={"store": "蜀大侠火锅", "coins": "1000"},
+        ),
+        # -- 好友邀请 -------------------------------------------------------
+        NotificationType(
+            key="invite_order_reward",
+            category=CATEGORY_INVITE,
+            card_style=CARD_FRIEND_CASH,
+            card_title="好友比价成功，现金已到账",
+            action_text="邀请更多好友赚现金",
+            push_title="您的邀请奖励已到账",
+            push_body_template="您的好友「{nickname}」完成首次下单，{amount}元现金已到账",
+            sample_vars={"nickname": "柚子", "amount": "2"},
+        ),
+        NotificationType(
+            key="invite_remind",
+            category=CATEGORY_INVITE,
+            card_style=CARD_PLAIN_TEXT,
+            card_title="你邀请的好友还差一步",
+            action_text="去提醒 TA",
+            push_title="提醒好友完成比价的奖励",
+            push_body_template="您的好友「{nickname}」还没完成比价下单，提醒TA完成，您可得{amount}元现金",
+            sample_vars={"nickname": "阿泽", "amount": "2"},
+        ),
+    ]
+}
+
+
+class UnknownNotificationType(ValueError):
+    """type key 不在 13 种类型之内。"""
+
+
+def get_type(type_key: str) -> NotificationType:
+    ntype = TYPES.get(type_key)
+    if ntype is None:
+        raise UnknownNotificationType(
+            f"unknown notification type: {type_key!r} (可选: {', '.join(TYPES)})"
+        )
+    return ntype
+
+
+def render_push(type_key: str, variables: dict[str, str] | None = None) -> tuple[str, str]:
+    """渲染某类型的 push (标题, 正文)。
+
+    variables 覆盖模板变量;缺的变量回退 sample_vars(PRD 示例值)——保证虚拟数据
+    阶段不传变量也能发出完整文案。多余的变量忽略。
+    """
+    ntype = get_type(type_key)
+    merged = {**ntype.sample_vars, **(variables or {})}
+
+    class _Fallback(dict):
+        def __missing__(self, key: str) -> str:  # 模板变量既没传也没示例值 → 保留 {key} 原样
+            return "{" + key + "}"
+
+    body = ntype.push_body_template.format_map(_Fallback(merged))
+    return ntype.push_title, body
+
+
+def push_variable_names(type_key: str) -> list[str]:
+    """列出模板里出现的变量名(给 /push/templates 预览用)。"""
+    import string
+
+    ntype = get_type(type_key)
+    return [
+        fname
+        for _, fname, _, _ in string.Formatter().parse(ntype.push_body_template)
+        if fname
+    ]

app/integrations/vendor_push.py

@@ -0,0 +1,519 @@
+"""厂商直推集成(荣耀 / 华为 / 小米 / OPPO / vivo)。
+
+服务端不经由 JPush Push API,而是按客户端上报的 push_vendor + push_token
+分发到各手机厂商的服务端 API。
+
+对外两个入口:
+  - send_notification()            通用:任意标题/正文/extras(消息中心 13 类推送走这里),
+                                   mock=True 时不真调厂商、返回渲染结果(虚拟数据联调用)
+  - send_accessibility_disabled()  旧:无障碍掉线召回(heartbeat_monitor_worker 在用),
+                                   已改为 send_notification 的薄封装,行为不变
+
+各厂商鉴权方式:荣耀/华为 OAuth client_credentials 换 access_token(进程内缓存);
+vivo/OPPO 签名换 authToken(缓存 24h);小米直接 AppSecret 走 Authorization 头。
+"""
+from __future__ import annotations
+
+import hashlib
+import json
+import logging
+import time
+import uuid
+from collections.abc import Callable
+from dataclasses import dataclass
+from typing import Any
+
+import httpx
+
+from app.core.config import settings
+
+logger = logging.getLogger("shagua.vendor_push")
+
+TYPE_ACCESSIBILITY_DISABLED = "accessibility_disabled"
+SUPPORTED_VENDORS = frozenset({"honor", "huawei", "vivo", "xiaomi", "oppo"})
+
+# vendor key → 中文名(测试/配置状态接口展示用)
+VENDOR_LABELS: dict[str, str] = {
+    "honor": "荣耀",
+    "huawei": "华为",
+    "xiaomi": "小米",
+    "oppo": "OPPO",
+    "vivo": "vivo",
+}
+
+# 各厂商真发推送所需的 settings 键(缺任一即视为未配置;/api/v1/push/vendors 据此报缺)
+REQUIRED_SETTINGS: dict[str, tuple[str, ...]] = {
+    "honor": ("HONOR_PUSH_APP_ID", "HONOR_PUSH_CLIENT_ID", "HONOR_PUSH_CLIENT_SECRET"),
+    "huawei": ("HUAWEI_PUSH_APP_ID", "HUAWEI_PUSH_APP_SECRET"),
+    "xiaomi": ("XIAOMI_PUSH_APP_SECRET",),
+    "oppo": ("OPPO_PUSH_APP_KEY", "OPPO_PUSH_MASTER_SECRET"),
+    "vivo": ("VIVO_PUSH_APP_ID", "VIVO_PUSH_APP_KEY", "VIVO_PUSH_APP_SECRET"),
+}
+
+
+def missing_settings(vendor: str) -> list[str]:
+    """该厂商还缺哪些配置键(全配齐返回空列表)。vendor 需已 normalize。"""
+    return [key for key in REQUIRED_SETTINGS.get(vendor, ()) if not getattr(settings, key, "")]
+
+
+class VendorPushError(Exception):
+    """厂商推送调用失败。"""
+
+
+@dataclass
+class _CachedToken:
+    value: str
+    expires_at: float
+
+
+_token_cache: dict[str, _CachedToken] = {}
+
+
+def normalize_vendor(push_vendor: str | None) -> str | None:
+    if not push_vendor:
+        return None
+    vendor = push_vendor.strip().lower()
+    aliases = {
+        "hihonor": "honor",
+        "荣耀": "honor",
+        "hms": "huawei",
+        "华为": "huawei",
+        "harmony": "huawei",
+        "harmonyos": "huawei",
+        "mi": "xiaomi",
+        "小米": "xiaomi",
+        "oneplus": "oppo",
+        "realme": "oppo",
+    }
+    return aliases.get(vendor, vendor)
+
+
+def send_notification(
+    push_vendor: str,
+    push_token: str,
+    *,
+    title: str,
+    body: str,
+    extras: dict[str, str] | None = None,
+    mock: bool = False,
+) -> dict[str, Any]:
+    """按厂商 token 向单台设备发送一条通知(通用入口)。
+
+    - extras:透传给客户端的自定义键值(值统一 string,兼容各厂商限制)。消息中心推送约定
+      至少带 {"type": <13 种类型 key>, "notificationId": <站内消息 id>},客户端据此
+      深链落地 + 调 /notifications/read 同步置读(PRD §4 push 联动)。
+    - mock=True:不真调厂商 API,校验参数后原样返回渲染结果(虚拟数据阶段联调/自动化测试用)。
+    """
+    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")
+    extras = {str(k): str(v) for k, v in (extras or {}).items()}
+
+    if mock:
+        logger.info(
+            "[mock push] vendor=%s token=%s... title=%s body=%s extras=%s",
+            vendor, token[:12], title, body, extras,
+        )
+        return {
+            "mock": True,
+            "vendor": vendor,
+            "title": title,
+            "body": body,
+            "extras": extras,
+        }
+
+    dispatch: dict[str, Callable[[str, str, str, dict[str, str]], dict[str, Any]]] = {
+        "honor": _send_honor,
+        "huawei": _send_huawei,
+        "vivo": _send_vivo,
+        "xiaomi": _send_xiaomi,
+        "oppo": _send_oppo,
+    }
+    return dispatch[vendor](token, title, body, extras)
+
+
+def send_accessibility_disabled(
+    push_vendor: str,
+    push_token: str,
+    *,
+    title: str = "保护已关闭",
+    alert: str = "傻瓜比价的无障碍保护被关了，点此重新开启，继续帮你自动比价省钱。",
+) -> dict[str, Any]:
+    """按厂商 token 向单台设备发送无障碍掉线通知(heartbeat_monitor_worker 在用,行为不变)。"""
+    return send_notification(
+        push_vendor,
+        push_token,
+        title=title,
+        body=alert,
+        extras={"type": TYPE_ACCESSIBILITY_DISABLED},
+    )
+
+
+def _require(value: str, name: str) -> str:
+    if not value:
+        raise VendorPushError(f"{name} not configured")
+    return value
+
+
+def _request_json(
+    method: str,
+    url: 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
+
+
+def _request_form(
+    method: str,
+    url: 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
+
+
+def _cache_get(key: str) -> str | None:
+    cached = _token_cache.get(key)
+    if cached and cached.expires_at > time.time() + 60:
+        return cached.value
+    return None
+
+
+def _cache_put(key: str, value: str, expires_in: int | float | None) -> str:
+    ttl = int(expires_in or 3600)
+    _token_cache[key] = _CachedToken(value=value, expires_at=time.time() + max(60, ttl - 60))
+    return value
+
+
+def _honor_access_token() -> str:
+    cache_key = "honor"
+    cached = _cache_get(cache_key)
+    if cached:
+        return cached
+    client_id = _require(settings.HONOR_PUSH_CLIENT_ID, "HONOR_PUSH_CLIENT_ID")
+    client_secret = _require(settings.HONOR_PUSH_CLIENT_SECRET, "HONOR_PUSH_CLIENT_SECRET")
+    data = _request_form(
+        "POST",
+        settings.HONOR_PUSH_TOKEN_ENDPOINT,
+        data={
+            "grant_type": "client_credentials",
+            "client_id": client_id,
+            "client_secret": client_secret,
+        },
+        headers={"Content-Type": "application/x-www-form-urlencoded"},
+    )
+    token = data.get("access_token")
+    if not token:
+        raise VendorPushError(f"honor auth failed: {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()
+    payload = {
+        "data": json.dumps(extras, ensure_ascii=False),
+        "notification": {"title": title, "body": body},
+        "android": {
+            "ttl": f"{settings.PUSH_TIME_TO_LIVE_SEC}s",
+            "targetUserType": 1,
+            "notification": {
+                "title": title,
+                "body": body,
+                "clickAction": {"type": 3},
+                "importance": "NORMAL",
+            },
+        },
+        "token": [token],
+    }
+    data = _request_json(
+        "POST",
+        settings.HONOR_PUSH_SEND_ENDPOINT_TEMPLATE.format(app_id=app_id),
+        json=payload,
+        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 push failed: {data}")
+    return data
+
+
+def _huawei_access_token() -> str:
+    """华为 OAuth2 client_credentials 换 access_token(client_id 即 AGC 应用的 AppId)。"""
+    cache_key = "huawei"
+    cached = _cache_get(cache_key)
+    if cached:
+        return cached
+    app_id = _require(settings.HUAWEI_PUSH_APP_ID, "HUAWEI_PUSH_APP_ID")
+    app_secret = _require(settings.HUAWEI_PUSH_APP_SECRET, "HUAWEI_PUSH_APP_SECRET")
+    data = _request_form(
+        "POST",
+        settings.HUAWEI_PUSH_TOKEN_ENDPOINT,
+        data={
+            "grant_type": "client_credentials",
+            "client_id": app_id,
+            "client_secret": app_secret,
+        },
+        headers={"Content-Type": "application/x-www-form-urlencoded"},
+    )
+    token = data.get("access_token")
+    if not token:
+        raise VendorPushError(f"huawei auth failed: {data}")
+    return _cache_put(cache_key, str(token), data.get("expires_in"))
+
+
+def _send_huawei(token: str, title: str, body: str, extras: dict[str, str]) -> dict[str, Any]:
+    """华为 Push Kit 下行消息(v1 messages:send)。成功码 '80000000';
+    '80100000' 为部分成功(单 token 场景仍视为失败,错误里带原始响应便于排障)。"""
+    app_id = _require(settings.HUAWEI_PUSH_APP_ID, "HUAWEI_PUSH_APP_ID")
+    access_token = _huawei_access_token()
+    payload = {
+        "validate_only": False,
+        "message": {
+            "data": json.dumps(extras, ensure_ascii=False),
+            "android": {
+                "ttl": f"{settings.PUSH_TIME_TO_LIVE_SEC}s",
+                "notification": {
+                    "title": title,
+                    "body": body,
+                    "click_action": {"type": 3},
+                    "importance": "NORMAL",
+                },
+            },
+            "token": [token],
+        },
+    }
+    data = _request_json(
+        "POST",
+        settings.HUAWEI_PUSH_SEND_ENDPOINT_TEMPLATE.format(app_id=app_id),
+        json=payload,
+        headers={
+            "Content-Type": "application/json; charset=UTF-8",
+            "Authorization": f"Bearer {access_token}",
+        },
+    )
+    if str(data.get("code", "")) != "80000000":
+        raise VendorPushError(f"huawei push failed: {data}")
+    return data
+
+
+def _vivo_auth_token() -> str:
+    cache_key = "vivo"
+    cached = _cache_get(cache_key)
+    if cached:
+        return cached
+    app_id = _require(settings.VIVO_PUSH_APP_ID, "VIVO_PUSH_APP_ID")
+    app_key = _require(settings.VIVO_PUSH_APP_KEY, "VIVO_PUSH_APP_KEY")
+    app_secret = _require(settings.VIVO_PUSH_APP_SECRET, "VIVO_PUSH_APP_SECRET")
+    timestamp = str(int(time.time() * 1000))
+    sign = hashlib.md5(f"{app_id}{app_key}{timestamp}{app_secret}".encode()).hexdigest()  # noqa: S324
+    data = _request_json(
+        "POST",
+        settings.VIVO_PUSH_AUTH_ENDPOINT,
+        json={
+            "appId": app_id,
+            "appKey": app_key,
+            "timestamp": timestamp,
+            "sign": sign,
+        },
+        headers={"Content-Type": "application/json"},
+    )
+    if int(data.get("result", -1)) != 0:
+        raise VendorPushError(f"vivo auth failed: {data}")
+    token = data.get("authToken")
+    if not token:
+        raise VendorPushError(f"vivo auth missing authToken: {data}")
+    return _cache_put(cache_key, str(token), 24 * 3600)
+
+
+def _send_vivo(token: str, title: str, body: str, extras: dict[str, str]) -> dict[str, Any]:
+    app_id = _require(settings.VIVO_PUSH_APP_ID, "VIVO_PUSH_APP_ID")
+    auth_token = _vivo_auth_token()
+    payload: dict[str, Any] = {
+        "appId": app_id,
+        "regId": token,
+        "notifyType": settings.VIVO_PUSH_NOTIFY_TYPE,
+        "title": title,
+        "content": body,
+        "timeToLive": settings.PUSH_TIME_TO_LIVE_SEC,
+        "skipType": 1,
+        "requestId": uuid.uuid4().hex,
+        "pushMode": settings.VIVO_PUSH_MODE,
+        "clientCustomMap": extras,
+    }
+    if settings.VIVO_PUSH_CATEGORY:
+        payload["category"] = settings.VIVO_PUSH_CATEGORY
+    data = _request_json(
+        "POST",
+        settings.VIVO_PUSH_SEND_ENDPOINT,
+        json=payload,
+        headers={
+            "Content-Type": "application/json",
+            "authToken": auth_token,
+        },
+    )
+    if int(data.get("result", -1)) != 0:
+        raise VendorPushError(f"vivo push failed: {data}")
+    return data
+
+
+def _send_xiaomi(token: str, title: str, body: str, extras: dict[str, str]) -> dict[str, Any]:
+    app_secret = _require(settings.XIAOMI_PUSH_APP_SECRET, "XIAOMI_PUSH_APP_SECRET")
+    message_title = settings.XIAOMI_PUSH_TEMPLATE_TITLE.strip() or title
+    message_description = settings.XIAOMI_PUSH_TEMPLATE_DESCRIPTION.strip() or body
+    form = {
+        "registration_id": token,
+        "restricted_package_name": settings.ANDROID_PACKAGE_NAME,
+        "title": message_title,
+        "description": message_description,
+        "payload": json.dumps(extras, ensure_ascii=False),
+        "pass_through": "0",
+        "notify_type": "-1",
+        "time_to_live": str(settings.PUSH_TIME_TO_LIVE_SEC * 1000),
+        "extra.notify_effect": "1",
+    }
+    if settings.XIAOMI_PUSH_CHANNEL_ID:
+        form["extra.channel_id"] = settings.XIAOMI_PUSH_CHANNEL_ID.strip()
+    if settings.XIAOMI_PUSH_TEMPLATE_ID:
+        form["extra.template_id"] = settings.XIAOMI_PUSH_TEMPLATE_ID.strip()
+    if settings.XIAOMI_PUSH_TEMPLATE_PARAM_JSON:
+        form["extra.template_param"] = _xiaomi_template_param(title, body)
+    data = _request_form(
+        "POST",
+        settings.XIAOMI_PUSH_SEND_ENDPOINT,
+        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}")
+    if str(data.get("result", "ok")).lower() not in ("ok", "success"):
+        raise VendorPushError(f"xiaomi push failed: {data}")
+    return data
+
+
+def _xiaomi_template_param(title: str, alert: str) -> str:
+    rendered = (
+        settings.XIAOMI_PUSH_TEMPLATE_PARAM_JSON
+        .replace("{title}", title)
+        .replace("{alert}", alert)
+    )
+    try:
+        payload = json.loads(rendered)
+    except ValueError as e:
+        raise VendorPushError("XIAOMI_PUSH_TEMPLATE_PARAM_JSON invalid json") from e
+    if not isinstance(payload, dict):
+        raise VendorPushError("XIAOMI_PUSH_TEMPLATE_PARAM_JSON must be a json object")
+    for key, value in payload.items():
+        if not isinstance(key, str) or not isinstance(value, str):
+            raise VendorPushError("xiaomi template params must be string key-value pairs")
+        if not value.strip() or len(value) > 128:
+            raise VendorPushError("xiaomi template param value length must be 1-128")
+    return json.dumps(payload, ensure_ascii=False, separators=(",", ":"))
+
+
+def _oppo_auth_token() -> str:
+    cache_key = "oppo"
+    cached = _cache_get(cache_key)
+    if cached:
+        return cached
+    app_key = _require(settings.OPPO_PUSH_APP_KEY, "OPPO_PUSH_APP_KEY")
+    master_secret = _require(settings.OPPO_PUSH_MASTER_SECRET, "OPPO_PUSH_MASTER_SECRET")
+    timestamp = str(int(time.time() * 1000))
+    sign = hashlib.sha256(f"{app_key}{timestamp}{master_secret}".encode()).hexdigest()
+    data = _request_form(
+        "POST",
+        settings.OPPO_PUSH_AUTH_ENDPOINT,
+        data={
+            "app_key": app_key,
+            "timestamp": timestamp,
+            "sign": sign,
+        },
+        headers={"Content-Type": "application/x-www-form-urlencoded"},
+    )
+    if int(data.get("code", -1)) != 0:
+        raise VendorPushError(f"oppo auth failed: {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}")
+    return _cache_put(cache_key, str(token), 24 * 3600)
+
+
+def _send_oppo(token: str, title: str, body: str, extras: dict[str, str]) -> dict[str, Any]:
+    auth_token = _oppo_auth_token()
+    ttl_hours = max(1, min(72, settings.PUSH_TIME_TO_LIVE_SEC // 3600))
+    notification: dict[str, Any] = {
+        "app_message_id": f"{extras.get('type', 'notify')}_{uuid.uuid4().hex}",
+        "title": title,
+        "content": body,
+        "click_action_type": 0,
+        "off_line": True,
+        "off_line_ttl": ttl_hours,
+        "action_parameters": json.dumps(extras, ensure_ascii=False),
+    }
+    # 新消息分类(2024-11-20 后创建的 OPPO 应用必须带 category,否则可能被拒收/降级)
+    if settings.OPPO_PUSH_CHANNEL_ID.strip():
+        notification["channel_id"] = settings.OPPO_PUSH_CHANNEL_ID.strip()
+    if settings.OPPO_PUSH_CATEGORY.strip():
+        notification["category"] = settings.OPPO_PUSH_CATEGORY.strip()
+    if settings.OPPO_PUSH_NOTIFY_LEVEL:
+        notification["notify_level"] = settings.OPPO_PUSH_NOTIFY_LEVEL
+    message = {
+        "target_type": 2,
+        "target_value": token,
+        "notification": notification,
+    }
+    data = _request_form(
+        "POST",
+        settings.OPPO_PUSH_SEND_ENDPOINT,
+        data={
+            "auth_token": auth_token,
+            "message": json.dumps(message, ensure_ascii=False),
+        },
+        headers={"Content-Type": "application/x-www-form-urlencoded"},
+    )
+    if int(data.get("code", -1)) != 0:
+        raise VendorPushError(f"oppo push failed: {data}")
+    return data

app/main.py

@@ -31,8 +31,10 @@ from app.api.v1.device import router as device_router
 from app.api.v1.feedback import router as feedback_router
 from app.api.v1.invite import router as invite_router
 from app.api.v1.meituan import router as meituan_router
+from app.api.v1.notifications import router as notifications_router
 from app.api.v1.order import router as order_router
 from app.api.v1.platform import router as platform_router
+from app.api.v1.push import router as push_router
 from app.api.v1.report import router as report_router
 from app.api.v1.savings import router as savings_router
 from app.api.v1.signin import router as signin_router
@@ -126,6 +128,10 @@ app.include_router(savings_router)
 app.include_router(ad_router)
 app.include_router(order_router)
 app.include_router(report_router)
+# 消息通知中心(PRD;虚拟数据阶段:mock 数据在内存,见 repositories/notification_mock.py)
+app.include_router(notifications_router)
+# 厂商推送测试三件套(配置状态/模板预览/测试发送,支持 mock 与真发)
+app.include_router(push_router)
 # 内部(server→server)端点:pricebot 上报价格观测 / 店铺映射,靠共享密钥头校验,不对客户端开放。
 app.include_router(internal_price_router)
 app.include_router(internal_store_router)

app/models/__init__.py

@@ -19,6 +19,7 @@ from app.models.coupon_state import (  # noqa: F401
     CouponClaimRecord,
     CouponDailyCompletion,
     CouponPromptEngagement,
+    CouponSession,
 )
 from app.models.feedback import Feedback  # noqa: F401
 from app.models.invite import InviteRelation  # noqa: F401

app/models/coupon_state.py

@@ -181,3 +181,79 @@ class CouponPromptEngagement(Base):
             f"<CouponPromptEngagement device={self.device_id} "
             f"date={self.engage_date} type={self.engage_type}>"
         )
+
+
+class CouponSession(Base):
+    """一次领券任务的全程流水(admin「领券数据」看板数据源,2026-06-30)。
+
+    与 coupon_claim_record(按券一天一条去重)、coupon_daily_completion(按设备一天一条)都不同:
+    本表**一次领券一条**(trace_id 唯一),记从发起(started)到收尾(completed/failed/abandoned)的
+    全程耗时 + 各平台耗时 + 机型/ROM。客户端 POST /api/v1/coupon/session 两段上报:发起建行、
+    收尾按 trace_id 更新同一行。发起即落库 → admin 可算「发起数」与中途流失(started 无终态=未完成)。
+
+    口径:elapsed_ms 由客户端全程计时(点发起→收尾)、权威;started_at/finished_at 为时刻留痕。
+    started_date = started_at 的 Asia/Shanghai 自然日,供 admin 按天聚合 / 日期筛选(索引)。
+    """
+
+    __tablename__ = "coupon_session"
+    __table_args__ = (
+        # 一次领券一行:trace_id 幂等 upsert(发起建、收尾更新同一行)。
+        UniqueConstraint("trace_id", name="uq_coupon_session_trace"),
+        # admin 主聚合/筛选:按上海自然日 + 环境(报表默认只看 prod,避免测试数据串台)。
+        Index("ix_coupon_session_date_env", "started_date", "app_env"),
+    )
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+
+    # 一次领券唯一 id(客户端 UUID,全程贯穿),upsert 键。
+    trace_id: Mapped[str] = mapped_column(String(64), nullable=False)
+    device_id: Mapped[str] = mapped_column(String(64), nullable=False)
+    # 登录态才带(admin join 用户表出手机号/昵称);匿名领券为空。
+    user_id: Mapped[int | None] = mapped_column(Integer, index=True, nullable=True)
+
+    # started / completed / failed / abandoned。started 无终态 = 中途流失(未完成)。
+    status: Mapped[str] = mapped_column(String(16), nullable=False)
+    # prod / dev(客户端 BuildConfig.DEBUG)。admin 报表默认只看 prod(对齐广告报表防串台口径)。
+    app_env: Mapped[str | None] = mapped_column(String(16), index=True, nullable=True)
+
+    # 发起勾选平台 ["meituan-waimai", ...](空=全领)。
+    platforms: Mapped[list | None] = mapped_column(_JSON, nullable=True)
+    # 发起来源外卖 App 包名;null=App 内(傻瓜比价首页)发起,非空=从美团/淘宝/京东弹券发起。
+    # admin「发起平台」列据此区分(空→傻瓜比价,包名→对应平台)。
+    origin_package: Mapped[str | None] = mapped_column(String(64), nullable=True)
+    # 机型(Build.MANUFACTURER + MODEL)与 ROM(OemDetector,如 "ColorOS 14")。明细「机型/ROM」列 + 维度筛选。
+    device_model: Mapped[str | None] = mapped_column(String(128), nullable=True)
+    rom: Mapped[str | None] = mapped_column(String(64), nullable=True)
+
+    # 发起时刻(客户端墙钟):明细「时间」列、趋势 X 轴。
+    started_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
+    # 发起的 Asia/Shanghai 自然日:按天聚合 / 日期范围筛选(索引)。
+    started_date: Mapped[date] = mapped_column(Date, nullable=False)
+    # 收尾时刻(服务端 now);未收尾(流失)为空。
+    finished_at: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True), nullable=True
+    )
+
+    # 全程耗时(ms,客户端点发起→收尾):平均 / 分位都基于它(只统计 completed)。
+    elapsed_ms: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    # 各平台领券耗时 {"meituan-waimai":3200,...}(ms)。明细美团/淘宝/京东耗时列。
+    platform_elapsed: Mapped[dict | None] = mapped_column(_JSON, nullable=True)
+    # 领到总张数(收尾帧带)。
+    claimed_count: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    # pricebot done 帧回传的公网调试链接(price.shaguabijia.com/traces/{dir});含落盘时分秒、拼不出,只能存
+    # (同 ComparisonRecord.trace_url)。admin「领券数据」明细据此渲染可点 trace 链接;未到 done(failed/abandoned)为空。
+    trace_url: Mapped[str | None] = mapped_column(String(512), nullable=True)
+
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now(), nullable=False
+    )
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now(), onupdate=func.now(),
+        nullable=False,
+    )
+
+    def __repr__(self) -> str:  # pragma: no cover
+        return (
+            f"<CouponSession trace={self.trace_id} status={self.status} "
+            f"elapsed_ms={self.elapsed_ms}>"
+        )

app/models/device.py

@@ -1,9 +1,9 @@
-"""设备表(无障碍保护存活检测 + 极光推送)。
+"""设备表(无障碍保护存活检测 + 厂商直推)。
 
 每条 = 一个用户的一台设备(per-install,device_id 由客户端 DeviceId.get() 生成)。
 客户端的无障碍服务存活时周期上报心跳刷新 last_heartbeat_at;App 前台/登录时上报
-registration_id(极光推送目标)。后端 heartbeat_monitor_worker 扫描「曾经保护过、
-现在心跳超时」的设备,通过极光推送提醒用户重开无障碍。
+push_vendor + push_token(厂商推送目标)。后端 heartbeat_monitor_worker 扫描「曾经保护过、
+现在心跳超时」的设备,通过厂商直推提醒用户重开无障碍。
 
 liveness_state 状态机(防刷屏,一次掉线只推一条):
   unknown → alive(收到 service 心跳)→ silent/notified(扫描发现超时并已推送)
@@ -30,7 +30,7 @@ from app.db.base import Base
 
 class DeviceLiveness(Base):
     # 表名不叫 device:device 易被当成「设备信息(品牌/型号/系统)」表;本表实为**无障碍存活监控状态**
-    # (心跳 last_heartbeat_at + liveness_state + kill_alert_pending + 推送目标 registration_id),故名 device_liveness。
+    # (心跳 last_heartbeat_at + liveness_state + kill_alert_pending + 厂商推送目标),故名 device_liveness。
     __tablename__ = "device_liveness"
     __table_args__ = (
         UniqueConstraint("user_id", "device_id", name="uq_device_liveness_user_device"),
@@ -42,8 +42,12 @@ class DeviceLiveness(Base):
     )
     # 客户端 DeviceId.get() 生成的 per-install id(如 device_Pixel_ab12cd34)
     device_id: Mapped[str] = mapped_column(String(128), index=True, nullable=False)
-    # 极光推送 registration id;拿到才填(JCollectionAuth 同意后才下发)
+    # 旧极光推送 registration id,仅为兼容历史客户端/数据保留;新链路使用 push_vendor + push_token。
     registration_id: Mapped[str | None] = mapped_column(String(64), nullable=True)
+    # 厂商推送类型:honor/vivo/xiaomi/oppo 等;客户端按实际 SDK token 来源上报。
+    push_vendor: Mapped[str | None] = mapped_column(String(32), nullable=True)
+    # 厂商 push token / regId / registration_id;不同厂商命名不同,后端统一存这里。
+    push_token: Mapped[str | None] = mapped_column(String(256), nullable=True)
     platform: Mapped[str] = mapped_column(String(16), nullable=False, default="android")
     app_version: Mapped[str | None] = mapped_column(String(32), nullable=True)
 

app/models/feedback.py

@@ -23,6 +23,14 @@ class Feedback(Base):
     )
     content: Mapped[str] = mapped_column(Text, nullable=False)
     contact: Mapped[str] = mapped_column(String(128), nullable=False)
+    # 反馈来源入口:profile(「我的」页反馈入口)/ comparison(比价结果页反馈入口)。
+    # admin 据此展示/筛选「反馈类型」。旧数据与「我的」页反馈均为 profile(server_default)。
+    # 客户端显式传 source 优先;未传时由 scene 有无派生(见 api/v1/feedback.submit_feedback)。
+    source: Mapped[str] = mapped_column(
+        String(16), nullable=False, default="profile", server_default="profile", index=True
+    )
+    # 比价反馈的「问题场景」(找错商品/优惠不对/比价太慢…),比价结果页反馈才有;普通反馈为 NULL。
+    scene: Mapped[str | None] = mapped_column(String(32), nullable=True)
     # 截图 URL 列表(相对路径,如 ["/media/feedback/u1_ab12.jpg"]);无图为 None
     images: Mapped[list[str] | None] = mapped_column(JSON, nullable=True)
     # 提交时的端环境快照(admin 排查用;客户端改版带上后的新反馈才有,历史数据为 NULL)
@@ -36,6 +44,9 @@ class Feedback(Base):
     reward_coins: Mapped[int | None] = mapped_column(Integer, nullable=True)
     # 审核批注:采纳时可写采纳要点,未采纳时也可保留运营侧备注
     review_note: Mapped[str | None] = mapped_column(String(256), nullable=True)
+    # 运营给用户的回复留言(**用户端可见**,采纳/未采纳都可填,随「我的反馈」历史下发)。
+    # 与 review_note(内部备注,用户不可见)、reject_reason(未采纳原因)区分开。
+    admin_reply: Mapped[str | None] = mapped_column(String(256), nullable=True)
     reviewed_by_admin_id: Mapped[int | None] = mapped_column(
         Integer, ForeignKey("admin_user.id"), nullable=True
     )

app/repositories/coupon_state.py

@@ -6,7 +6,7 @@
 from __future__ import annotations
 
 import logging
-from datetime import date, datetime
+from datetime import date, datetime, timezone
 from zoneinfo import ZoneInfo
 
 from sqlalchemy import delete, func, select
@@ -17,6 +17,7 @@ from app.models.coupon_state import (
     CouponClaimRecord,
     CouponDailyCompletion,
     CouponPromptEngagement,
+    CouponSession,
 )
 
 logger = logging.getLogger("shagua.coupon_state")
@@ -232,3 +233,91 @@ def sum_claimed_count(db: Session, user_id: int) -> int:
         )
     ).scalar_one()
     return int(total or 0)
+
+
+# ===== 领券任务流水(coupon_session,admin「领券数据」看板数据源)=====
+
+def upsert_coupon_session(
+    db: Session,
+    *,
+    trace_id: str,
+    device_id: str,
+    status: str,
+    started_at_ms: int,
+    user_id: int | None = None,
+    platforms: list[str] | None = None,
+    origin_package: str | None = None,
+    device_model: str | None = None,
+    rom: str | None = None,
+    app_env: str | None = None,
+    elapsed_ms: int | None = None,
+    platform_elapsed: dict | None = None,
+    claimed_count: int | None = None,
+    trace_url: str | None = None,
+) -> None:
+    """一条领券流水按 trace_id 幂等 upsert(发起 started 建行、收尾终态更新同一行)。
+
+    - 乱序/重复兜底:终态(completed/failed/abandoned)先到也建行;started 重复到不覆盖已有终态
+      (状态只前进,不降级)。
+    - started_at 由客户端墙钟毫秒转;started_date 取其 Asia/Shanghai 自然日(admin 按天聚合/筛选)。
+    - 终态帧补 finished_at=服务端 now;各字段非空才写(避免 started 帧的 None 抹掉收尾值,反之亦然)。
+    并发 IntegrityError 回滚忽略(本就幂等)。
+    """
+    started_at = datetime.fromtimestamp(started_at_ms / 1000, tz=timezone.utc)
+    started_date = started_at.astimezone(_CN_TZ).date()
+    is_terminal = status in ("completed", "failed", "abandoned")
+
+    row = db.execute(
+        select(CouponSession).where(CouponSession.trace_id == trace_id)
+    ).scalar_one_or_none()
+
+    if row is None:
+        db.add(CouponSession(
+            trace_id=trace_id,
+            device_id=device_id,
+            user_id=user_id,
+            status=status,
+            app_env=app_env,
+            platforms=platforms,
+            origin_package=origin_package,
+            device_model=device_model,
+            rom=rom,
+            started_at=started_at,
+            started_date=started_date,
+            finished_at=datetime.now(timezone.utc) if is_terminal else None,
+            elapsed_ms=elapsed_ms,
+            platform_elapsed=platform_elapsed,
+            claimed_count=claimed_count,
+            trace_url=trace_url,
+        ))
+    else:
+        # 状态只前进:started 帧重复到(如 START_STICKY 重启)不把已有终态降级回 started。
+        if not (status == "started" and row.status in ("completed", "failed", "abandoned")):
+            row.status = status
+        if is_terminal:
+            row.finished_at = datetime.now(timezone.utc)
+        if user_id is not None:
+            row.user_id = user_id
+        if platforms is not None:
+            row.platforms = platforms
+        if origin_package is not None:
+            row.origin_package = origin_package
+        if device_model is not None:
+            row.device_model = device_model
+        if rom is not None:
+            row.rom = rom
+        if app_env is not None:
+            row.app_env = app_env
+        if elapsed_ms is not None:
+            row.elapsed_ms = elapsed_ms
+        if platform_elapsed is not None:
+            row.platform_elapsed = platform_elapsed
+        if claimed_count is not None:
+            row.claimed_count = claimed_count
+        if trace_url is not None:
+            row.trace_url = trace_url
+    try:
+        db.commit()
+    except IntegrityError:
+        # 并发下另一请求刚插了同 trace_id → 唯一约束撞,回滚忽略(本就幂等)。
+        db.rollback()

app/repositories/device.py

@@ -21,17 +21,23 @@ def register_or_update(
     *,
     user_id: int,
     device_id: str,
-    registration_id: str | None,
+    registration_id: str | None = None,
+    push_vendor: str | None = None,
+    push_token: str | None = None,
     platform: str = "android",
     app_version: str | None = None,
 ) -> DeviceLiveness:
-    """注册设备或更新其 registration_id / 元信息。upsert by (user_id, device_id)。"""
+    """注册设备或更新其厂商 push token / 元信息。upsert by (user_id, device_id)。"""
+    normalized_vendor = _normalize_push_vendor(push_vendor)
+    normalized_token = push_token.strip() if push_token else None
     device = _get(db, user_id=user_id, device_id=device_id)
     if device is None:
         device = DeviceLiveness(
             user_id=user_id,
             device_id=device_id,
             registration_id=registration_id,
+            push_vendor=normalized_vendor,
+            push_token=normalized_token,
             platform=platform or "android",
             app_version=app_version,
         )
@@ -39,6 +45,10 @@ def register_or_update(
     else:
         if registration_id:
             device.registration_id = registration_id
+        if normalized_vendor:
+            device.push_vendor = normalized_vendor
+        if normalized_token:
+            device.push_token = normalized_token
         if platform:
             device.platform = platform
         if app_version:
@@ -54,7 +64,9 @@ def touch_heartbeat(
     user_id: int,
     device_id: str,
     accessibility_enabled: bool,
-    registration_id: str | None,
+    registration_id: str | None = None,
+    push_vendor: str | None = None,
+    push_token: str | None = None,
 ) -> DeviceLiveness:
     """处理一次心跳(心跳也能自注册)。
 
@@ -69,6 +81,12 @@ def touch_heartbeat(
 
     if registration_id:
         device.registration_id = registration_id
+    normalized_vendor = _normalize_push_vendor(push_vendor)
+    normalized_token = push_token.strip() if push_token else None
+    if normalized_vendor:
+        device.push_vendor = normalized_vendor
+    if normalized_token:
+        device.push_token = normalized_token
     device.last_report_protection_on = accessibility_enabled
 
     if accessibility_enabled:
@@ -87,7 +105,7 @@ def touch_heartbeat(
 def list_overdue(db: Session, *, timeout_minutes: int) -> list[DeviceLiveness]:
     """掉线设备:曾经保护过、当前 alive、心跳超时。
 
-    本期只做终端打印检测、不推送 → 不再要求有 registration_id(没接极光 token 的设备也要检出)。
+    即使没有厂商 token 也要检出,后续由 kill_alert_pending 走客户端进 App 后兜底提醒。
     """
     cutoff = datetime.now(timezone.utc) - timedelta(minutes=timeout_minutes)
     stmt = select(DeviceLiveness).where(
@@ -124,3 +142,27 @@ def ack_kill_alert(db: Session, *, user_id: int, device_id: str) -> None:
     if device is not None and device.kill_alert_pending:
         device.kill_alert_pending = False
         db.commit()
+
+
+def has_push_target(device: DeviceLiveness | None) -> bool:
+    """是否已有厂商直推所需的 vendor + token。"""
+    return bool(device and device.push_vendor and device.push_token)
+
+
+def _normalize_push_vendor(push_vendor: str | None) -> str | None:
+    if not push_vendor:
+        return None
+    vendor = push_vendor.strip().lower()
+    aliases = {
+        "honor": "honor",
+        "hihonor": "honor",
+        "荣耀": "honor",
+        "vivo": "vivo",
+        "xiaomi": "xiaomi",
+        "mi": "xiaomi",
+        "小米": "xiaomi",
+        "oppo": "oppo",
+        "oneplus": "oppo",
+        "realme": "oppo",
+    }
+    return aliases.get(vendor, vendor)

app/repositories/feedback.py

@@ -17,6 +17,8 @@ def create_feedback(
     content: str,
     contact: str,
     images: list[str] | None,
+    source: str = "profile",
+    scene: str | None = None,
     app_version: str | None = None,
     device_model: str | None = None,
     rom_name: str | None = None,
@@ -26,6 +28,8 @@ def create_feedback(
         user_id=user_id,
         content=content,
         contact=contact,
+        source=source,
+        scene=scene or None,
         images=images or None,
         app_version=app_version or None,
         device_model=device_model or None,

app/repositories/notification_mock.py

@@ -0,0 +1,328 @@
+"""消息通知中心 mock 数据仓库(内存版,不落库)。
+
+⚠️ 虚拟数据阶段专用:
+  - 每个用户**首次访问时**生成一套固定样例(覆盖全部 13 种类型 + 「今天/昨天/当年/跨年」
+    四种时间显示分支 + 已读/未读混合),供前端联调列表渲染、角标、已读消除全流程。
+  - 已读状态维护在进程内存 → **重启服务即复位**;多 worker 部署时各 worker 各一份
+    (联调期单 worker 跑即可,与 core/ratelimit.py 同款局限)。
+  - 后续接真实数据:建 notification 表 + 本文件同名函数改为查库,API 层无需改动。
+
+排序规则:**全列表按时间倒序**(最新在前),不做分类分组(需求方 2026-07-14 确认
+PRD §1 的"按分类分组"写错了,已取消);分页作用在排序后的整表上。
+"""
+from __future__ import annotations
+
+import itertools
+import threading
+from dataclasses import dataclass, field
+from datetime import datetime, timedelta, timezone
+
+from app.core import notification_catalog as catalog
+
+# 北京时间:sentAt 统一带 +08:00 偏移下发,前端直接按本地时区渲染「今天/昨天/M月D日」。
+_CST = timezone(timedelta(hours=8))
+
+# mock id 从 90001 起步,一眼可辨"这是虚拟数据"(真实表主键自增不会从这里开始)。
+_id_counter = itertools.count(90001)
+_lock = threading.Lock()
+# user_id → 该用户的 mock 通知列表(生成后仅 is_read 会变化)
+_stores: dict[int, list[MockNotification]] = {}
+
+
+@dataclass
+class MockNotification:
+    """一条 mock 通知的动态部分;静态部分(标题/分类/版式/操作行)看 catalog.TYPES。"""
+
+    id: int
+    type_key: str
+    sent_at: datetime
+    is_read: bool
+    coins: int | None = None            # 金币数(整数;dual_amount / coin_reward 卡)
+    cash_cents: int | None = None       # 现金金额,单位分(dual_amount / withdraw / friend_cash 卡)
+    info_rows: list[dict[str, str]] = field(default_factory=list)  # [{label, value}]
+    extra: dict[str, str] = field(default_factory=dict)            # 点击跳转所需业务参数
+
+
+def cash_yuan(cents: int | None) -> str | None:
+    """分 → 保留两位小数的元字符串(PRD §3:现金/提现金额保留两位小数)。"""
+    if cents is None:
+        return None
+    return f"{cents // 100}.{cents % 100:02d}"
+
+
+def _fmt_time(dt: datetime) -> str:
+    """信息行里「到账时间」等 value 的展示格式。"""
+    return dt.strftime("%Y-%m-%d %H:%M")
+
+
+# ---------------------------------------------------------------------------
+# 各类型的样例卡片内容(数值 + 信息行 + extra),文案对齐 PRD §3
+# ---------------------------------------------------------------------------
+
+
+def _card_reward_expiring(sent_at: datetime, coins: int = 86, cash: int = 1280, days: int = 3) -> dict:
+    return {
+        "coins": coins,
+        "cash_cents": cash,
+        "info_rows": [
+            {
+                "label": "过期说明",
+                "value": f"您有{coins}金币和{cash_yuan(cash)}元现金即将失效，"
+                         "完成一次一键领券或一键比价即可激活收益",
+            },
+            {"label": "过期时间", "value": f"{days}天后失效"},
+        ],
+        # batchId:同一批次激活成功后不再重复推送(PRD §2 激活逻辑)
+        "extra": {"batchId": f"batch_{sent_at:%Y%m%d}"},
+    }
+
+
+def _card_reward_expired(sent_at: datetime, coins: int = 35, cash: int = 60) -> dict:
+    return {
+        "coins": coins,
+        "cash_cents": cash,
+        "info_rows": [
+            {
+                "label": "过期说明",
+                "value": f"您的{coins}金币和{cash_yuan(cash)}元现金已失效，"
+                         "完成一次一键领券或一键比价可赚取新收益",
+            },
+            {"label": "过期时间", "value": f"已过期 {sent_at.month}月{sent_at.day}日失效"},
+        ],
+        "extra": {},  # 点击跳赚钱页(tab),无需参数
+    }
+
+
+def _card_withdraw_success(sent_at: datetime, cash: int = 50) -> dict:
+    return {
+        "cash_cents": cash,
+        "info_rows": [
+            {"label": "到账账户", "value": "微信钱包"},
+            {"label": "到账时间", "value": _fmt_time(sent_at)},
+        ],
+        "extra": {},  # 无跳转,仅消红点
+    }
+
+
+def _card_withdraw_failed(sent_at: datetime, cash: int = 350, reason: str = "微信零钱未实名") -> dict:
+    return {
+        "cash_cents": cash,
+        "info_rows": [
+            {"label": "失败原因", "value": reason},
+            {"label": "退回说明", "value": "款项已原路退回现金余额"},
+        ],
+        "extra": {"withdrawId": "88001"},  # 点击跳提现页
+    }
+
+
+def _card_permission(permission: str) -> dict:
+    # permission ∈ accessibility(无障碍)/ battery(省电策略)/ autostart(自启动)/ overlay(悬浮窗)
+    # 客户端点击时按此 key 实时检测该权限并弹对应开启弹窗(PRD §2 权限逻辑)。
+    return {
+        "info_rows": [
+            {"label": "说明文案", "value": "未开启将导致核心功能不可用，请尽快开启"},
+        ],
+        "extra": {"permission": permission},
+    }
+
+
+def _card_feedback_reply(feedback_id: str) -> dict:
+    return {
+        "info_rows": [
+            {"label": "说明文案", "value": "快去看看官方给您的回复吧～"},
+        ],
+        "extra": {"feedbackId": feedback_id},  # 跳反馈历史页并滚动高亮该条(PRD §2)
+    }
+
+
+def _card_feedback_reward(sent_at: datetime, coins: int = 300,
+                          reply: str = "感谢反馈，您说的问题已经修复上线，送您的金币请查收～") -> dict:
+    return {
+        "coins": coins,
+        "info_rows": [
+            {"label": "奖励说明", "value": "感谢您的反馈，您的金币奖励已到账"},
+            {"label": "官方留言", "value": reply},  # PRD §3:官方留言必填(发奖励必带留言)
+            {"label": "到账时间", "value": _fmt_time(sent_at)},
+        ],
+        "extra": {"feedbackId": "3002"},
+    }
+
+
+def _card_report_approved(sent_at: datetime, coins: int = 1000, store: str = "蜀大侠火锅") -> dict:
+    return {
+        "coins": coins,
+        "info_rows": [
+            {"label": "奖励说明", "value": f"您爆料的「{store}」更低价已通过审核，金币奖励已到账"},
+            {"label": "到账时间", "value": _fmt_time(sent_at)},
+        ],
+        "extra": {"reportId": "5001"},  # 跳爆料记录页并滚动高亮该条
+    }
+
+
+def _card_invite_order_reward(sent_at: datetime, cash: int = 200, nickname: str = "柚子") -> dict:
+    return {
+        "cash_cents": cash,
+        "info_rows": [
+            {"label": "奖励说明", "value": f"好友「{nickname}」完成首次下单"},
+            {"label": "到账时间", "value": _fmt_time(sent_at)},
+        ],
+        "extra": {"inviteeNickname": nickname},  # 跳邀请页(welfare/invite.html?from=notifications)
+    }
+
+
+def _card_invite_remind(nickname: str = "阿泽") -> dict:
+    return {
+        "info_rows": [
+            {
+                "label": "说明文案",
+                "value": f"好友「{nickname}」已注册，还没完成比价下单，提醒TA完成后你可得2元现金",
+            },
+        ],
+        # scrollTo=remind:跳邀请页并自动滚动到底部「提醒好友」模块(PRD §2 #13)
+        "extra": {"inviteeNickname": nickname, "scrollTo": "remind"},
+    }
+
+
+def build_sample(type_key: str, sent_at: datetime | None = None) -> MockNotification:
+    """按类型生成一条默认样例(push 测试联动等场景插入新 mock 通知用)。"""
+    catalog.get_type(type_key)  # 校验 type 合法
+    now = sent_at or datetime.now(_CST)
+    default_builders = {
+        "reward_expiring": lambda: _card_reward_expiring(now),
+        "reward_expired": lambda: _card_reward_expired(now),
+        "withdraw_success": lambda: _card_withdraw_success(now),
+        "withdraw_failed": lambda: _card_withdraw_failed(now),
+        "perm_accessibility": lambda: _card_permission("accessibility"),
+        "perm_battery": lambda: _card_permission("battery"),
+        "perm_autostart": lambda: _card_permission("autostart"),
+        "perm_overlay": lambda: _card_permission("overlay"),
+        "feedback_reply": lambda: _card_feedback_reply("3001"),
+        "feedback_reward": lambda: _card_feedback_reward(now),
+        "report_approved": lambda: _card_report_approved(now),
+        "invite_order_reward": lambda: _card_invite_order_reward(now),
+        "invite_remind": lambda: _card_invite_remind(),
+    }
+    return MockNotification(
+        id=next(_id_counter),
+        type_key=type_key,
+        sent_at=now,
+        is_read=False,
+        **default_builders[type_key](),
+    )
+
+
+def _seed_user(user_id: int) -> list[MockNotification]:
+    """给一个用户生成整套 mock 样例:13 类型全覆盖 + 时间显示四分支 + 已读/未读混合。"""
+    now = datetime.now(_CST)
+    today = now - timedelta(hours=2)            # 今天
+    today_recent = now - timedelta(minutes=10)  # 今天(更近,验证组内时间倒序)
+    yesterday = now - timedelta(days=1)         # 昨天
+    this_year = now - timedelta(days=40)        # 当年 → 「M月D日」
+    last_year = now - timedelta(days=370)       # 跨年 → 「YYYY年M月D日」
+
+    def one(type_key: str, sent_at: datetime, card: dict, *, read: bool = False) -> MockNotification:
+        return MockNotification(
+            id=next(_id_counter), type_key=type_key, sent_at=sent_at, is_read=read, **card
+        )
+
+    return [
+        # ---- 提现助手(4 类 + 2 条补充,验证组内倒序与跨年展示)----
+        one("reward_expiring", today, _card_reward_expiring(today)),
+        one("reward_expired", yesterday, _card_reward_expired(yesterday)),
+        one("withdraw_success", today_recent, _card_withdraw_success(today_recent)),
+        one("withdraw_failed", yesterday, _card_withdraw_failed(yesterday)),
+        one("withdraw_success", now - timedelta(days=3),
+            _card_withdraw_success(now - timedelta(days=3), cash=3000)),
+        one("withdraw_success", last_year, _card_withdraw_success(last_year, cash=120), read=True),
+        # ---- 系统通知(权限异常 ×4;同一权限未读期间只更新时间不新增,故各一条)----
+        one("perm_accessibility", now - timedelta(hours=1), _card_permission("accessibility")),
+        one("perm_battery", now - timedelta(days=3), _card_permission("battery"), read=True),
+        one("perm_autostart", now - timedelta(days=5), _card_permission("autostart")),
+        one("perm_overlay", now - timedelta(days=6), _card_permission("overlay"), read=True),
+        # ---- 我的反馈 ----
+        one("feedback_reply", now - timedelta(hours=4), _card_feedback_reply("3001")),
+        one("feedback_reward", yesterday, _card_feedback_reward(yesterday)),
+        one("feedback_reply", now - timedelta(days=60), _card_feedback_reply("2001"), read=True),
+        # ---- 我的爆料 ----
+        one("report_approved", this_year, _card_report_approved(this_year)),
+        # ---- 好友邀请 ----
+        one("invite_order_reward", today_recent, _card_invite_order_reward(today_recent)),
+        one("invite_remind", now - timedelta(days=2), _card_invite_remind()),
+    ]
+
+
+def _get_store(user_id: int) -> list[MockNotification]:
+    """取该用户的 mock 列表,首次访问时生成。调用方须持有 _lock。"""
+    store = _stores.get(user_id)
+    if store is None:
+        store = _seed_user(user_id)
+        _stores[user_id] = store
+    return store
+
+
+def _sorted(store: list[MockNotification]) -> list[MockNotification]:
+    """排序:全列表时间倒序(同秒再按 id 倒序稳定化)。不分组(PRD 原文写错,已取消)。"""
+    return sorted(store, key=lambda n: (-n.sent_at.timestamp(), -n.id))
+
+
+def list_notifications(
+    user_id: int, *, page: int, page_size: int
+) -> tuple[list[MockNotification], int, int]:
+    """分页取通知列表。返回 (当前页条目, 总条数, 未读条数)。"""
+    with _lock:
+        store = _get_store(user_id)
+        ordered = _sorted(store)
+        total = len(ordered)
+        unread = sum(1 for n in store if not n.is_read)
+        start = (page - 1) * page_size
+        return ordered[start : start + page_size], total, unread
+
+
+def unread_count(user_id: int) -> int:
+    """未读总数(首页铃铛角标)。"""
+    with _lock:
+        return sum(1 for n in _get_store(user_id) if not n.is_read)
+
+
+def mark_read(
+    user_id: int, *, ids: list[int] | None = None, mark_all: bool = False
+) -> tuple[int, int]:
+    """标记已读。mark_all=True 全量清零,否则按 ids 逐条置读(不存在的 id 忽略,幂等)。
+
+    返回 (本次实际由未读→已读的条数, 剩余未读数)。
+    """
+    with _lock:
+        store = _get_store(user_id)
+        marked = 0
+        if mark_all:
+            for n in store:
+                if not n.is_read:
+                    n.is_read = True
+                    marked += 1
+        else:
+            wanted = set(ids or [])
+            for n in store:
+                if n.id in wanted and not n.is_read:
+                    n.is_read = True
+                    marked += 1
+        unread = sum(1 for n in store if not n.is_read)
+        return marked, unread
+
+
+def insert_sample(user_id: int, type_key: str) -> MockNotification:
+    """插入一条该类型的新 mock 通知(未读)。push 测试联动用:push extras 带上它的 id,
+    客户端点击 push 后调 POST /notifications/read {ids:[id]} 即可闭环验证已读联动。"""
+    with _lock:
+        store = _get_store(user_id)
+        item = build_sample(type_key)
+        store.append(item)
+        return item
+
+
+def reset_store(user_id: int | None = None) -> None:
+    """清空 mock 状态(测试/联调用;user_id=None 清全部)。下次访问重新生成。"""
+    with _lock:
+        if user_id is None:
+            _stores.clear()
+        else:
+            _stores.pop(user_id, None)

app/schemas/coupon_state.py

@@ -49,3 +49,28 @@ class CouponStatsOut(BaseModel):
     """
 
     coupon_count: int
+
+
+class CouponSessionIn(BaseModel):
+    """客户端领券流水上报体(admin「领券数据」看板数据源,POST /api/v1/coupon/session)。
+
+    一次领券两段上报,按 trace_id upsert 到 coupon_session:
+    - 发起(status=started):带勾选平台 + 机型/ROM/app_env + started_at_ms(发起墙钟毫秒)。
+    - 收尾(completed/failed/abandoned):带 elapsed_ms(全程耗时)+ platform_elapsed(各平台耗时)+ claimed_count。
+    不鉴权(同领券循环 MVP,按 device_id/trace_id),user_id 登录态带上做留痕(可空)。
+    """
+
+    trace_id: str
+    device_id: str
+    status: str  # started / completed / failed / abandoned
+    started_at_ms: int  # 发起墙钟毫秒(客户端 System.currentTimeMillis)
+    user_id: int | None = None
+    platforms: list[str] | None = None
+    origin_package: str | None = None
+    device_model: str | None = None
+    rom: str | None = None
+    app_env: str | None = None
+    elapsed_ms: int | None = None
+    platform_elapsed: dict[str, int] | None = None
+    claimed_count: int | None = None
+    trace_url: str | None = None

app/schemas/device.py

@@ -3,12 +3,15 @@ from __future__ import annotations
 
 from datetime import datetime
 
-from pydantic import BaseModel, ConfigDict
+from pydantic import BaseModel, ConfigDict, Field
 
 
 class DeviceRegisterRequest(BaseModel):
     device_id: str
+    # registration_id 为旧极光字段,新推送链路统一使用 push_vendor + push_token。
     registration_id: str | None = None
+    push_vendor: str | None = None
+    push_token: str | None = None
     platform: str = "android"
     app_version: str | None = None
 
@@ -18,6 +21,8 @@ class HeartbeatRequest(BaseModel):
     source: str = "service"          # service | app
     accessibility_enabled: bool = True
     registration_id: str | None = None
+    push_vendor: str | None = None
+    push_token: str | None = None
 
 
 class DeviceOut(BaseModel):
@@ -26,6 +31,8 @@ class DeviceOut(BaseModel):
     id: int
     device_id: str
     registration_id: str | None
+    push_vendor: str | None
+    push_token: str | None
     ever_protected: bool
     liveness_state: str
     last_heartbeat_at: datetime | None
@@ -46,3 +53,17 @@ class LivenessOut(BaseModel):
 
 class LivenessAckRequest(BaseModel):
     device_id: str
+
+
+class PushTestRequest(BaseModel):
+    device_id: str
+    delay_seconds: int = Field(default=10, ge=0, le=60)
+    push_vendor: str | None = None
+    push_token: str | None = None
+    registration_id: str | None = None
+
+
+class PushTestOut(BaseModel):
+    ok: bool = True
+    delay_seconds: int
+    has_push_token: bool

app/schemas/feedback.py

@@ -30,10 +30,14 @@ class FeedbackQrConfigOut(BaseModel):
 class FeedbackRecordOut(BaseModel):
     id: int
     content: str
+    # 比价反馈的问题场景(找错商品/优惠不对…);普通反馈为 None
+    scene: str | None = None
     images: list[str] = Field(default_factory=list)
     status: str
     reject_reason: str | None = None
     reward_coins: int | None = None
+    # 运营给用户的回复留言(用户端可见,采纳/未采纳都可能有)
+    admin_reply: str | None = None
     created_at: datetime
 
 

app/schemas/notification.py

@@ -0,0 +1,128 @@
+"""消息通知中心 请求/响应契约。
+
+⚠️ 命名约定:本组接口按 PRD 前端契约使用 **camelCase**(sentAt / isRead / pageSize …),
+与库内其他 snake_case 接口不同——PRD 与前端原型(notifications.html)按 camelCase 对接,
+需求方接口清单亦明确写作 sentAt / isRead,故整组遵循之。响应序列化走 pydantic alias。
+
+字段说明都写在 Field(description=...) 里,起服务后打开 /docs 即是给前端的在线文档。
+"""
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Any
+
+from pydantic import BaseModel, ConfigDict, Field
+from pydantic.alias_generators import to_camel
+
+
+class _CamelModel(BaseModel):
+    """出参统一 camelCase(alias);populate_by_name 允许服务端代码仍用 snake_case 构造。"""
+
+    model_config = ConfigDict(alias_generator=to_camel, populate_by_name=True)
+
+
+class InfoRow(_CamelModel):
+    """卡片信息行(PRD §3「信息行」列),前端按 label: value 逐行渲染。"""
+
+    label: str = Field(description="行标签,如「过期说明」「到账账户」「失败原因」")
+    value: str = Field(description="行内容(已按 PRD 文案拼好变量,前端直接展示)")
+
+
+class NotificationItem(_CamelModel):
+    """一条通知卡片。
+
+    卡片头部三要素:categoryLabel(分类标签)+ 未读红点(isRead=false 时展示)+ 时间(sentAt)。
+    时间显示规则(前端处理):今天→「今天」;昨天→「昨天」;当年→「M月D日」;跨年→「YYYY年M月D日」。
+    """
+
+    id: int = Field(description="通知 id(未读消除、push 联动都用它)")
+    category: str = Field(
+        description="分类 key:withdraw_assistant=提现助手 / system=系统通知 / "
+                    "feedback=我的反馈 / report=我的爆料 / invite=好友邀请"
+    )
+    category_label: str = Field(description="分类中文标签(卡片头部直接展示)")
+    type: str = Field(
+        description="类型 key(13 种,决定点击行为,见 PRD §2):reward_expiring 即将失效 / "
+                    "reward_expired 已失效 / withdraw_success 提现成功 / withdraw_failed 提现失败 / "
+                    "perm_accessibility 无障碍异常 / perm_battery 省电策略异常 / "
+                    "perm_autostart 自启动异常 / perm_overlay 悬浮窗异常 / "
+                    "feedback_reply 官方回复 / feedback_reward 反馈奖励 / "
+                    "report_approved 爆料审核通过 / invite_order_reward 好友下单奖励 / "
+                    "invite_remind 好友催单提醒"
+    )
+    card_style: str = Field(
+        description="卡片版式:dual_amount 双金额卡 / withdraw 提现卡 / plain_text 纯文本卡 / "
+                    "coin_reward 金币奖励卡 / friend_cash 好友现金卡"
+    )
+    title: str = Field(description="卡片标题(双金额/提现/金币奖励/好友现金卡标题居中)")
+    coins: int | None = Field(
+        default=None,
+        description="金币数(整数,不带小数)。dual_amount / coin_reward 卡有值,其余 null",
+    )
+    cash_cents: int | None = Field(
+        default=None,
+        description="现金金额,单位【分】。dual_amount / withdraw / friend_cash 卡有值,其余 null",
+    )
+    cash_yuan: str | None = Field(
+        default=None,
+        description="现金金额展示串(元,保留两位小数,如 \"12.80\"),与 cashCents 同源,可直接展示",
+    )
+    info_rows: list[InfoRow] = Field(
+        description="信息行列表(label: value),内容已按 PRD §3 拼好,前端逐行渲染即可"
+    )
+    action_text: str | None = Field(
+        default=None,
+        description="操作行文案(如「立即激活您的收益」「去开启」);null=无操作行(提现成功卡)。"
+                    "注意:点击目标是整张卡片,不区分卡片主体和操作行",
+    )
+    extra: dict[str, Any] = Field(
+        description="点击跳转所需业务参数,按 type 取用:perm_* → {permission: accessibility|battery|"
+                    "autostart|overlay}(点击时实时检测该权限);feedback_* → {feedbackId};"
+                    "report_approved → {reportId};withdraw_failed → {withdrawId};"
+                    "invite_order_reward → {inviteeNickname};invite_remind → "
+                    "{inviteeNickname, scrollTo:\"remind\"};reward_expiring → {batchId}"
+    )
+    sent_at: datetime = Field(description="下发时间(ISO8601 带 +08:00 时区),前端按显示规则格式化")
+    is_read: bool = Field(description="是否已读;false 时分类标签右侧展示 6px 红点(#E53935)")
+
+
+class NotificationListOut(_CamelModel):
+    """GET /api/v1/notifications 出参。列表已按时间倒序排好(最新在前,**不分组**;
+    PRD §1 的"按分类分组"为笔误,已确认取消),前端无需再排。"""
+
+    items: list[NotificationItem] = Field(description="当前页通知卡片")
+    page: int = Field(description="当前页码(1 起)")
+    page_size: int = Field(description="每页条数")
+    total: int = Field(description="全部通知总条数(含已读)")
+    has_more: bool = Field(description="是否还有下一页")
+    unread_count: int = Field(description="当前未读总数(与 /notifications/unread-count 同口径,省一次请求)")
+
+
+class UnreadCountOut(_CamelModel):
+    """GET /api/v1/notifications/unread-count 出参(首页铃铛角标)。"""
+
+    count: int = Field(description="未读总条数(精确值)")
+    badge_text: str | None = Field(
+        description="角标展示文案:超过 99 返回 \"99+\";等于 0 返回 null(整个角标隐藏,不展示空红点)"
+    )
+
+
+class MarkReadRequest(_CamelModel):
+    """POST /api/v1/notifications/read 入参,两种模式二选一:
+
+    - `{"ids": [90001, 90002]}` 单条/多条置读——点击某张卡片、点击 push 落地后同步置读;
+    - `{"all": true}`           全量清零——进入通知中心(或退出时)自动清零(PRD §4)。
+
+    同时传时 all=true 优先;不存在/已读的 id 自动忽略(幂等,可放心重试)。
+    """
+
+    ids: list[int] | None = Field(default=None, description="要置为已读的通知 id 列表")
+    all: bool = Field(default=False, description="true=清空该用户全部未读")
+
+
+class MarkReadOut(_CamelModel):
+    """POST /api/v1/notifications/read 出参。"""
+
+    ok: bool = Field(description="固定 true(参数非法时走 400,不会到这里)")
+    marked_count: int = Field(description="本次实际由未读变为已读的条数(重复请求会是 0)")
+    unread_count: int = Field(description="处理后的剩余未读总数,可直接刷新铃铛角标")

app/schemas/push.py

@@ -0,0 +1,99 @@
+"""厂商推送(测试/联调)接口契约。与消息中心同族,出参统一 camelCase。"""
+from __future__ import annotations
+
+from typing import Any
+
+from pydantic import BaseModel, ConfigDict, Field
+from pydantic.alias_generators import to_camel
+
+
+class _CamelModel(BaseModel):
+    model_config = ConfigDict(alias_generator=to_camel, populate_by_name=True)
+
+
+class PushVendorStatus(_CamelModel):
+    vendor: str = Field(description="厂商 key:honor / huawei / xiaomi / oppo / vivo")
+    label: str = Field(description="厂商中文名")
+    configured: bool = Field(description="服务端凭据是否齐全(齐全才能真发,mock 不受影响)")
+    missing_keys: list[str] = Field(description="缺失的 .env 配置键;configured=true 时为空")
+
+
+class PushVendorsOut(_CamelModel):
+    vendors: list[PushVendorStatus] = Field(description="5 个厂商的配置状态")
+
+
+class PushTemplateOut(_CamelModel):
+    type: str = Field(description="通知类型 key(13 种,与消息中心 type 一致)")
+    category: str = Field(description="分类 key")
+    category_label: str = Field(description="分类中文标签")
+    card_style: str = Field(description="站内卡片版式")
+    push_title: str = Field(description="push 标题(≤11 字固定文案,PRD §5)")
+    push_body_sample: str = Field(description="push 正文示例(模板用 PRD 示例值渲染后的效果)")
+    push_body_template: str = Field(description="push 正文模板原文,{var} 为变量占位")
+    variables: list[str] = Field(description="模板变量名列表(调 /push/test 时可在 vars 里覆盖)")
+    sample_vars: dict[str, str] = Field(description="各变量的 PRD 示例值(vars 未覆盖时的缺省)")
+
+
+class PushTemplatesOut(_CamelModel):
+    templates: list[PushTemplateOut] = Field(description="13 种通知类型的 push 模板(PRD 编号顺序)")
+
+
+class PushTestRequest(_CamelModel):
+    """POST /api/v1/push/test 入参。三种发送内容来源(优先级从高到低):
+
+    1. 直接指定 title + content;
+    2. 指定 type(13 种之一)→ 按 PRD §5 模板渲染,vars 可覆盖模板变量;
+    3. 都不传 → 发一条通用测试文案。
+
+    推送目标:pushToken 直填,或 deviceId 反查该用户已注册设备(/api/v1/device/register 上报过的)。
+    """
+
+    vendor: str = Field(
+        default="",
+        description="厂商:honor/huawei/xiaomi/oppo/vivo(中文「华为」「小米」等别名也识别)。"
+                    "留空时用 deviceId 对应设备上报的 push_vendor",
+    )
+    push_token: str = Field(default="", description="厂商 push token / regId;留空则走 deviceId 反查")
+    device_id: str = Field(default="", description="设备 id(客户端 DeviceId.get());用于反查 token")
+    type: str = Field(
+        default="",
+        description="通知类型 key(13 种,见 GET /push/templates);留空且未直接给 title/content 时发通用测试文案",
+    )
+    vars: dict[str, str] = Field(
+        default_factory=dict,
+        description="覆盖 push 模板变量,如 {\"coins\":\"520\",\"cash\":\"6.66\"};缺省用 PRD 示例值",
+    )
+    title: str = Field(default="", description="直接指定标题(优先于 type 模板)")
+    content: str = Field(default="", description="直接指定正文(优先于 type 模板)")
+    create_notification: bool = Field(
+        default=False,
+        description="true=同时往该用户的消息中心 mock 列表插入一条同类型未读通知,push extras 带上它的"
+                    " notificationId → 可闭环验证「点 push → 落地 → 调 /notifications/read 消红点」联动"
+                    "(仅 type 为 13 种类型之一时生效)",
+    )
+    mock: bool = Field(
+        default=True,
+        description="true(默认)=不真调厂商 API,返回渲染结果(联调安全);false=真发,要求该厂商凭据已配置",
+    )
+
+
+class PushTestOut(_CamelModel):
+    ok: bool = Field(description="发送(或 mock 渲染)成功")
+    mock: bool = Field(description="本次是否 mock(未真调厂商 API)")
+    vendor: str = Field(description="实际使用的厂商 key(已归一化)")
+    title: str = Field(description="实际下发的 push 标题")
+    body: str = Field(description="实际下发的 push 正文")
+    extras: dict[str, str] = Field(
+        description="随 push 下发的自定义键值(客户端深链用):type 必有;createNotification=true 时带"
+                    " notificationId 及该通知的业务参数(feedbackId / permission / …)"
+    )
+    notification_id: int | None = Field(
+        default=None, description="createNotification=true 时新插入的站内 mock 通知 id"
+    )
+    missing_keys: list[str] = Field(
+        default_factory=list,
+        description="该厂商仍缺失的配置键(mock 发送时提示「真发前还需配什么」;真发时必为空)",
+    )
+    vendor_response: dict[str, Any] | None = Field(
+        default=None, description="真发时厂商 API 的原始响应(mock 时为 null)"
+    )

deploy/pangle-revenue.md

@@ -0,0 +1,67 @@
+# 穿山甲 GroMore 收益拉取 定时任务 — 运维手册
+
+> 对象:维护「每天拉穿山甲后台收益入库」这套定时任务的同事。
+> 🔒 服务器登录信息见**私密交接清单**,不入库。
+
+## 它是什么
+admin「广告收益报表」里的「穿山甲后台收益(T+1)」读的是**本地表 `ad_pangle_daily_revenue` 的快照,不是实时查穿山甲**。穿山甲只通过 GroMore 数据 API 给数、且 **T+1**(次日约 10:00 出昨天的数),所以每天得拉一次入库,报表才会往前走。
+
+- 每天 10:30 跑一轮 `scripts/sync_pangle_revenue.py`,默认 `--days 3` 回补近 3 天。
+- 维度 = 日期 × 应用(site_id)× 广告位(ad_unit_id);指标 = `revenue`(预估)+ `api_revenue`(结算口径)。
+- **幂等 upsert**:同一(日期×应用×代码位)重跑只覆盖、不重复,故回补 / 重跑 / catch-up 都安全。
+- 穿山甲无用户/设备维度 → 只能落「汇总/趋势级」,报表带 user_id 过滤时这块收益置空(显示「-」)。
+
+## 文件
+| 项 | 路径 |
+|---|---|
+| 脚本入口 | `scripts/sync_pangle_revenue.py` |
+| 拉取 / 签名 | `app/integrations/pangle_report.py`(签名=参数字典序拼接+secure_key 后 MD5) |
+| 入库表 | `ad_pangle_daily_revenue`(读写在 `app/repositories/ad_pangle_revenue.py`) |
+| 凭证 | `.env` 的 `PANGLE_REPORT_USER_ID` / `PANGLE_REPORT_ROLE_ID` / `PANGLE_REPORT_SECURITY_KEY` |
+| 应用映射 | `.env` 的 `PANGLE_REPORT_SITE_ID_PROD`(5830519)/ `PANGLE_REPORT_SITE_ID_TEST`(5832303) |
+| systemd 单元 | `deploy/pangle-revenue.{service,timer}` |
+
+## 上线前置(只做一次)
+1. **填凭证**:后台「接入中心 → GroMore-API」领 user_id / role_id / Security Key(`secure_key`,**≠ 发奖 m-key**),填进线上 `.env` 的 `PANGLE_REPORT_*`。三项填齐脚本才工作,缺任一 → no-op 退出。
+2. **子账号要授权**:线上若用子账号(role_id ≠ user_id),需主账号给它授「查看全部数据」,否则接口报 **118**(无权限)。本人自查自己(user_id=role_id)无此问题。
+3. 凭证泄露:后台「重置 key 值」即可,改完同步线上 `.env` 重跑。
+
+## 部署(Linux 服务器,需 root)
+```bash
+sudo cp deploy/pangle-revenue.{service,timer} /etc/systemd/system/
+sudo systemctl daemon-reload && sudo systemctl enable --now pangle-revenue.timer
+systemctl list-timers pangle-revenue.timer   # 确认下次触发时间(应是次日 10:30)
+```
+
+## 怎么看健康 / 手动跑一次
+```bash
+sudo systemctl start pangle-revenue.service     # 立即手动跑一轮(不等 10:30)
+journalctl -u pangle-revenue -n 30 --no-pager   # 看日志:拉取区间 / 入库行数 / 新增更新 / 预估收益合计
+```
+成功日志形如:`✅ 完成:接口 N 行 → 入库 M 行(跳过 x),新增 a / 更新 b;预估收益合计 ¥19.42`。
+> 看不到收益、提示 `PANGLE_REPORT_* 未配置`→ 回「上线前置」补 `.env`;报 118 → 子账号没授「查看全部数据」。
+
+## 本机 Windows 开发(无 systemd)
+直接手动跑:
+```
+.venv\Scripts\python -m scripts.sync_pangle_revenue            # 拉昨天
+.venv\Scripts\python -m scripts.sync_pangle_revenue --days 3   # 回补近 3 天
+.venv\Scripts\python -m scripts.sync_pangle_revenue --date 2026-06-27          # 指定单天
+.venv\Scripts\python -m scripts.sync_pangle_revenue --start 2026-06-01 --end 2026-06-27   # 区间回补
+```
+
+## 脚本参数
+- 无参:拉**昨天**(北京时间)。timer 用的是 `--days 3`。
+- `--days N`:从昨天起往前回补 N 天(含昨天)。
+- `--date YYYY-MM-DD`:指定单天。
+- `--start / --end`:指定闭区间(跨度 ≤ 31 天,接口上限 1 个月,超了报 114)。
+
+## 注意事项
+- **触发时间**:`OnCalendar=*-*-* 10:30:00`。穿山甲 ~10:00 出数,故别早于 10:00 跑(会拉到空/不全)。
+- **catch-up**:`Persistent=true` 补跑错过的那一轮;叠加 `--days 3`,漏一两天重新触发即自愈。
+- **今天 / 今天以前要分开查**:脚本默认只拉昨天及更早,不混查今天(接口约束),无需关心。
+- **join key 是 `ad_unit_id`(我们配的 104xxx)不是 `code_id`**:`code_id` 是底层各 ADN 代码位,对不上口径;`ad_unit_id='-1'` 是未归因桶。改维度时务必注意(详见脚本头注释)。
+- **`api_revenue` 很稀疏**:测试应用 ADN 没配 Reporting → 全 0,仅 prod 个别位有;`revenue`(预估)才是稳的主力。
+- **DB 无关**:sqlite / postgres 均可(upsert 逐行 select-then-write,不像美团 ETL 需要 PG)。
+- **别和别的触发方式双跑**:本 systemd timer 与「手动 cron / 进程内任务」二选一,虽幂等不会重复入库,纯属多余。
+- **改脚本 / 改部署**:走 git + PR,由有 root 的人部署。

deploy/pangle-revenue.service

@@ -0,0 +1,38 @@
+# 每天拉穿山甲 GroMore T+1 天级收益入库 —— 单轮跑,由 pangle-revenue.timer 每天 10:30 触发。
+# 落 ad_pangle_daily_revenue 表,供 admin 广告收益报表的「穿山甲后台收益(T+1)」区块。
+#
+# 仅用于 Linux 服务器;本机 Windows 开发无 systemd,直接手动跑脚本即可:
+#   .venv\Scripts\python -m scripts.sync_pangle_revenue            # 拉昨天(北京时间)
+#   .venv\Scripts\python -m scripts.sync_pangle_revenue --days 3   # 回补近 3 天
+#
+# 部署(服务器):
+#   sudo cp deploy/pangle-revenue.{service,timer} /etc/systemd/system/
+#   sudo systemctl daemon-reload && sudo systemctl enable --now pangle-revenue.timer
+#   # 手动跑一次验证: sudo systemctl start pangle-revenue.service && journalctl -u pangle-revenue -n 30
+#
+# 前置:.env 配好 PANGLE_REPORT_USER_ID / PANGLE_REPORT_ROLE_ID / PANGLE_REPORT_SECURITY_KEY
+#       (后台「接入中心 → GroMore-API」领;子账号需主账号授「查看全部数据」否则接口 118)。
+#       未配齐这三项 → 脚本自动 no-op 退出,免动 timer。
+[Unit]
+Description=Sync Pangle GroMore daily revenue (T+1, one-shot, driven by timer)
+After=network-online.target
+Wants=network-online.target
+
+[Service]
+Type=oneshot
+User=root
+WorkingDirectory=/opt/shaguabijia-app-server
+Environment="PATH=/opt/shaguabijia-app-server/.venv/bin:/usr/bin:/bin"
+EnvironmentFile=/opt/shaguabijia-app-server/.env
+# 默认拉昨天;--days 3 回补近 3 天(幂等 upsert,应对偶发漏跑 + 穿山甲对历史数据订正,重跑无害)。
+ExecStart=/opt/shaguabijia-app-server/.venv/bin/python -m scripts.sync_pangle_revenue --days 3
+SyslogIdentifier=pangle-revenue
+# 仅几个 HTTP 请求 + 小批量入库,通常数秒;给 10min 硬超时防穿山甲接口卡死。
+TimeoutStartSec=600
+
+# 与主服务 shaguabijia-app-server.service 同款加固。
+NoNewPrivileges=true
+PrivateTmp=true
+ProtectSystem=strict
+ReadWritePaths=/opt/shaguabijia-app-server
+ProtectHome=true

deploy/pangle-revenue.timer

@@ -0,0 +1,14 @@
+# 每天 10:30 触发一次穿山甲 GroMore T+1 收益拉取入库(Linux 服务器用)。
+# 见 pangle-revenue.service 顶部注释的部署步骤。
+[Unit]
+Description=Run Pangle GroMore daily revenue sync at 10:30
+
+[Timer]
+# 穿山甲 T+1、次日约 10:00 出数;10:30 触发留 30min 余量。要错开整点扎堆可微调到 10:35。
+OnCalendar=*-*-* 10:30:00
+# 服务器宕机/重启后,补跑错过的那一轮(而不是干等次日);叠加 --days 3 回补,漏一两天能自愈。
+Persistent=true
+AccuracySec=1min
+
+[Install]
+WantedBy=timers.target

docs/api/README.md

@@ -1,9 +1,9 @@
 # 傻瓜比价 App 后端 — API 接口文档（索引）
 
 > Base URL:生产 `https://app-api.shaguabijia.com`;本地联调 `http://<开发机>:8770`
-> 协议:HTTP / JSON,请求与响应体均 `application/json`,字段统一 **snake_case**
+> 协议:HTTP / JSON,请求与响应体均 `application/json`,字段统一 **snake_case**(⚠️ 例外:消息通知中心 `notifications` 族与厂商推送 `push` 族按 PRD 前端契约用 **camelCase**,见各自文档)
 > 鉴权:需鉴权的接口在请求头带 `Authorization: Bearer <access_token>`
-> 最后更新:2026-06-23（补全此前缺整族的端点:device 无障碍存活监控、internal 内部回写、CPS 短链落地、report/invite/wxpay,及 platform/coupon/feedback/user 的零散读端点;device/internal/cps-redirect 三族新建单文件文档。上一次 2026-06-18 补录比价透传端点与 `meituan/top-sales`）
+> 最后更新:2026-07-14（新增 **消息通知中心** 3 端点(M1-M3,虚拟数据阶段)与 **厂商推送测试** 3 端点(P1-P3,荣耀/华为/小米/OPPO/vivo);上一次 2026-06-23 补全 device/internal/CPS 短链等整族端点）
 > 架构:`app/api/v1/` 只放很轻的接口层;穿山甲/微信支付/极光/短信/美团等 SDK 集成的重逻辑在 `app/integrations/`,实现细节见 [docs/integrations/](../integrations/README.md)。
 
 ---
@@ -99,6 +99,14 @@
 | 38 | `POST /api/v1/feedback` | Bearer | [详情](./feedback.md) |
 | 38a | `GET /api/v1/feedback/config` | Bearer | 反馈页「加群二维码」卡配置（开关 + 二维码图 + 三行文案）(无单独文档) |
 | 38b | `GET /api/v1/feedback/records` | Bearer | 我的反馈历史（pending/adopted/rejected）(无单独文档) |
+| **消息通知中心**（前缀 `/api/v1/notifications`;⚠️ 本族对外 **camelCase**;虚拟数据阶段:内存 mock,重启复位） |||
+| M1 | `GET /api/v1/notifications` | Bearer | [详情](./notifications.md)（消息列表,分页;13 类型卡片字段 + sentAt/isRead;服务端已按时间倒序排好,不分组） |
+| M2 | `GET /api/v1/notifications/unread-count` | Bearer | [详情](./notifications.md)（未读总数,首页铃铛角标;>99 → "99+",0 → null 隐藏） |
+| M3 | `POST /api/v1/notifications/read` | Bearer | [详情](./notifications.md)（标记已读:`{ids:[...]}` 单条/多条 或 `{all:true}` 进通知中心全量清零;幂等） |
+| **厂商推送测试**（前缀 `/api/v1/push`;荣耀/华为/小米/OPPO/vivo 五通道联调三件套,同为 camelCase） |||
+| P1 | `GET /api/v1/push/vendors` | Bearer | [详情](./push-vendor-test.md)（5 厂商服务端凭据配置状态,缺哪些 .env 键一目了然） |
+| P2 | `GET /api/v1/push/templates` | Bearer | [详情](./push-vendor-test.md)（13 类通知的 push 标题/正文模板 + PRD 示例渲染效果） |
+| P3 | `POST /api/v1/push/test` | Bearer | [详情](./push-vendor-test.md)（测试发送:默认 mock 不真发;mock=false 真发;可联动插一条站内 mock 通知闭环验证已读） |
 | **首页门面数据 / 客户端配置**（前缀 `/api/v1/platform`；全平台展示数字 + 运营开关,**全部不鉴权**,登录前可读） |||
 | 39 | `GET /api/v1/platform/stats` | 无 | [详情](./platform-stats.md) |
 | 40 | `GET /api/v1/platform/savings-feed` | 无 | [详情](./platform-savings-feed.md) |

docs/api/notifications.md

@@ -0,0 +1,121 @@
+# 消息通知中心(notifications 族)
+
+> 所属:notifications 组(前缀 `/api/v1/notifications`,源 `app/api/v1/notifications.py`) | 鉴权:**全部 Bearer**(消息按用户隔离) | [← 返回 API 索引](./README.md)
+>
+> 对应 PRD《消息通知中心》(通知类型清单 / 点击跳转 / 字段元素 / 未读红点 / Push 文案)。
+> Push 侧(厂商直推 + 测试)见 [push-vendor-test.md](./push-vendor-test.md)。
+>
+> ⚠️ **虚拟数据阶段**:数据来自内存 mock(`app/repositories/notification_mock.py`),每个用户首次访问生成一套固定样例(13 种类型全覆盖 + 今天/昨天/当年/跨年四种时间 + 已读未读混合,共 16 条,id 从 90001 起);已读状态在服务端内存维护,**重启服务复位**。后续接真实数据只换 repository 实现,接口契约不变。
+>
+> ⚠️ **字段命名**:本组接口(含 push 测试组)对外为 **camelCase**(`sentAt` / `isRead` / `pageSize`…),与库内其他 snake_case 接口不同——按 PRD 前端契约对接,勿混用。
+
+## 通知类型速查(13 种)
+
+列表**服务端已排好序:全列表按时间倒序**(最新在前,**不做分类分组**——PRD §1 的"按分类分组"为笔误,2026-07-14 需求方确认取消),前端按返回顺序渲染即可。category 仅用于卡片头部的分类标签展示。
+
+| category | 分类标签 | type | 类型 | cardStyle 版式 | actionText 操作行 | extra 里带什么 |
+|---|---|---|---|---|---|---|
+| withdraw_assistant | 提现助手 | `reward_expiring` | 金币现金奖励即将失效 | dual_amount 双金额卡 | 立即激活您的收益 | `batchId` |
+| withdraw_assistant | 提现助手 | `reward_expired` | 金币现金奖励已失效 | dual_amount 双金额卡 | 立即赚取新收益 | — |
+| withdraw_assistant | 提现助手 | `withdraw_success` | 提现成功 | withdraw 提现卡 | **null(无操作行,点击仅消红点)** | — |
+| withdraw_assistant | 提现助手 | `withdraw_failed` | 提现失败,款项已退回 | withdraw 提现卡 | 重新提现 | `withdrawId` |
+| system | 系统通知 | `perm_accessibility` | 比价功能异常(无障碍) | plain_text 纯文本卡 | 去开启 | `permission:"accessibility"` |
+| system | 系统通知 | `perm_battery` | 比价续航保护异常(省电策略) | plain_text 纯文本卡 | 去开启 | `permission:"battery"` |
+| system | 系统通知 | `perm_autostart` | 比价启动保护异常(自启动) | plain_text 纯文本卡 | 去开启 | `permission:"autostart"` |
+| system | 系统通知 | `perm_overlay` | 比价按钮异常(悬浮窗) | plain_text 纯文本卡 | 去开启 | `permission:"overlay"` |
+| feedback | 我的反馈 | `feedback_reply` | 官方回复 | plain_text 纯文本卡 | 查看详情 | `feedbackId` |
+| feedback | 我的反馈 | `feedback_reward` | 反馈奖励(必带官方留言行) | coin_reward 金币奖励卡 | 查看反馈详情 | `feedbackId` |
+| report | 我的爆料 | `report_approved` | 爆料审核通过 | coin_reward 金币奖励卡 | 查看爆料详情 | `reportId` |
+| invite | 好友邀请 | `invite_order_reward` | 好友下单奖励到账 | friend_cash 好友现金卡 | 邀请更多好友赚现金 | `inviteeNickname` |
+| invite | 好友邀请 | `invite_remind` | 好友催单提醒 | plain_text 纯文本卡 | 去提醒 TA | `inviteeNickname`, `scrollTo:"remind"` |
+
+点击跳转逻辑按 PRD §2 由客户端按 `type` 分发;点击目标 = 整张卡片(不区分主体和操作行),任何点击都先调 `POST /read` 消该条红点。
+
+## GET /api/v1/notifications — 消息列表(分页)
+
+**入参(query)**
+
+| 参数 | 类型 | 必填 | 说明 |
+|---|---|---|---|
+| `page` | int | ❌ | 页码,1 起,默认 1 |
+| `pageSize` | int | ❌ | 每页条数,默认 20,最大 100 |
+
+**出参**
+
+```jsonc
+{
+  "items": [
+    {
+      "id": 90001,
+      "category": "withdraw_assistant",   // 分类 key(5 种,见上表)
+      "categoryLabel": "提现助手",         // 卡片头部左上角分类标签
+      "type": "reward_expiring",          // 类型 key(13 种,决定点击行为)
+      "cardStyle": "dual_amount",         // 版式:dual_amount/withdraw/plain_text/coin_reward/friend_cash
+      "title": "金币现金奖励即将失效",      // 卡片标题
+      "coins": 86,                        // 金币数,整数;无金币的版式为 null
+      "cashCents": 1280,                  // 现金金额(分);无现金的版式为 null
+      "cashYuan": "12.80",                // 现金展示串(元,两位小数),与 cashCents 同源
+      "infoRows": [                       // 信息行,已按 PRD 拼好文案,逐行 label: value 渲染
+        { "label": "过期说明", "value": "您有86金币和12.80元现金即将失效，完成一次一键领券或一键比价即可激活收益" },
+        { "label": "过期时间", "value": "3天后失效" }
+      ],
+      "actionText": "立即激活您的收益",    // 操作行;null = 无操作行(提现成功卡)
+      "extra": { "batchId": "batch_20260714" },  // 跳转/联动参数,按 type 取用(见上表)
+      "sentAt": "2026-07-14T14:59:58+08:00",     // ISO8601 带时区
+      "isRead": false                     // false → 分类标签右侧显示 6px 红点(#E53935)
+    }
+  ],
+  "page": 1,
+  "pageSize": 20,
+  "total": 16,
+  "hasMore": false,
+  "unreadCount": 12    // 与 /unread-count 同口径,进页面可顺手刷角标
+}
+```
+
+**时间显示规则(前端处理 `sentAt`)**:今天 →「今天」;昨天 →「昨天」;当年 →「M月D日」(不补零);跨年 →「YYYY年M月D日」。mock 样例特意覆盖了这四种情况。
+
+**数值约束(PRD §3)**:金币整数不带小数;现金/提现金额两位小数(直接用 `cashYuan`)。
+
+## GET /api/v1/notifications/unread-count — 未读总数(首页铃铛角标)
+
+无入参。**出参**:
+
+```jsonc
+{ "count": 12, "badgeText": "12" }   // count>99 时 badgeText="99+";count=0 时 badgeText=null → 整个角标隐藏
+```
+
+刷新时机(PRD §4):进入首页时、从通知中心/其他页面返回首页时(原型监听 `pageshow`)。
+
+## POST /api/v1/notifications/read — 标记已读
+
+**入参(JSON),两种模式二选一(同时传时 `all` 优先)**
+
+| 模式 | body | 使用场景 |
+|---|---|---|
+| 单条/多条 | `{ "ids": [90001, 90003] }` | ① 点击某张消息卡片(点击后无论跳转/弹窗/无动作都算已读);② 用户点击 push 直达落地页后,客户端拿 push extras 里的 `notificationId` 同步置读 |
+| 全量清零 | `{ "all": true }` | 进入通知中心自动清零(只浏览列表就消红点,无需逐条点击;退出通知中心时也可再调一次兜底) |
+
+**出参**
+
+```jsonc
+{ "ok": true, "markedCount": 2, "unreadCount": 10 }   // unreadCount = 处理后剩余未读,可直接刷新角标
+```
+
+幂等:不存在/已读的 id 忽略,重复调用 `markedCount=0` 不报错。
+
+**错误**:`400` ids 与 all 都没传(或 ids 为空数组);`401` 未鉴权。
+
+## 联调小抄
+
+```bash
+# 1. 登录拿 token(SMS mock:任意手机号 + 任意 6 位验证码)
+curl -X POST :8770/api/v1/auth/sms/send  -d '{"phone":"13800001234"}'
+curl -X POST :8770/api/v1/auth/sms/login -d '{"phone":"13800001234","code":"123456"}'
+# 2. 列表 / 角标 / 置读
+curl ":8770/api/v1/notifications?page=1&pageSize=20" -H "Authorization: Bearer $TOKEN"
+curl ":8770/api/v1/notifications/unread-count"        -H "Authorization: Bearer $TOKEN"
+curl -X POST ":8770/api/v1/notifications/read" -d '{"all":true}' -H "Authorization: Bearer $TOKEN"
+```
+
+要重置 mock 数据(重新生成整套样例):重启服务即可。想造「push → 站内联动」数据,用 [POST /api/v1/push/test](./push-vendor-test.md) 的 `createNotification:true`。

docs/api/push-vendor-test.md

@@ -0,0 +1,102 @@
+# 厂商推送测试三件套(push 族)
+
+> 所属:push 组(前缀 `/api/v1/push`,源 `app/api/v1/push.py`) | 鉴权:**全部 Bearer** | [← 返回 API 索引](./README.md)
+>
+> 发送实现:`app/integrations/vendor_push.py`(荣耀 / **华为** / 小米 / OPPO / vivo 五通道,
+> `send_notification()` 通用入口)。站内消息中心见 [notifications.md](./notifications.md)。
+> 与 `POST /api/v1/device/push-test`(无障碍召回通道延迟自测)互补:本组面向消息中心 13 类 push 的文案/参数/通道联调。
+>
+> 字段命名同 notifications 族:**camelCase**。
+
+## 链路总览
+
+```
+真实业务事件(奖励过期/提现回执/审核通过/…)
+      └→ vendor_push.send_notification(vendor, token, title, body, extras)
+              extras = { type, notificationId, ...业务参数 }        ← 客户端深链 + 已读联动的钥匙
+客户端点击 push → 按 extras.type 直达落地页(与站内点击一致)
+              → 调 POST /notifications/read {ids:[extras.notificationId]} 同步消红点(PRD §4)
+```
+
+推送目标来源:客户端集成各厂商 push SDK 拿到 regId/token 后,通过 `POST /api/v1/device/register` 上报 `push_vendor` + `push_token`,服务端存 `device_liveness` 表。
+
+## GET /api/v1/push/vendors — 厂商配置状态
+
+检查 5 家厂商服务端凭据是否配齐(只读 .env,不打厂商接口)。`missingKeys` 即还要补的配置键;mock 测试不依赖任何凭据。
+
+```jsonc
+{ "vendors": [
+    { "vendor": "honor",  "label": "荣耀", "configured": false, "missingKeys": ["HONOR_PUSH_APP_ID", "HONOR_PUSH_CLIENT_ID", "HONOR_PUSH_CLIENT_SECRET"] },
+    { "vendor": "huawei", "label": "华为", "configured": false, "missingKeys": ["HUAWEI_PUSH_APP_ID", "HUAWEI_PUSH_APP_SECRET"] },
+    { "vendor": "xiaomi", "label": "小米", "configured": true,  "missingKeys": [] },
+    { "vendor": "oppo",   "label": "OPPO", "configured": false, "missingKeys": ["OPPO_PUSH_APP_KEY", "OPPO_PUSH_MASTER_SECRET"] },
+    { "vendor": "vivo",   "label": "vivo", "configured": false, "missingKeys": ["VIVO_PUSH_APP_ID", "VIVO_PUSH_APP_KEY", "VIVO_PUSH_APP_SECRET"] }
+] }
+```
+
+## GET /api/v1/push/templates — 13 类通知的 push 模板预览
+
+PRD §5 的 13 条 push 文案(标题固定 ≤11 字不带变量;正文 `{var}` 为变量,示例值即 PRD 示例)。对文案、看变量名用。
+
+```jsonc
+{ "templates": [
+    {
+      "type": "withdraw_success",
+      "category": "withdraw_assistant", "categoryLabel": "提现助手", "cardStyle": "withdraw",
+      "pushTitle": "提现到账提醒",
+      "pushBodySample": "¥0.50已存入您的微信钱包，点击查看到账详情",   // 用示例值渲染后的效果
+      "pushBodyTemplate": "¥{amount}已存入您的微信钱包，点击查看到账详情",
+      "variables": ["amount"],
+      "sampleVars": { "amount": "0.50" }
+    }
+    // ... 共 13 条,顺序即 PRD 编号
+] }
+```
+
+## POST /api/v1/push/test — 测试发送(默认 mock)
+
+**入参(JSON)**
+
+| 字段 | 类型 | 必填 | 说明 |
+|---|---|---|---|
+| `vendor` | string | ❌* | `honor/huawei/xiaomi/oppo/vivo`,中文「华为」「小米」等别名也识别;留空时用 `deviceId` 设备上报的 vendor |
+| `pushToken` | string | ❌* | 厂商 push token/regId;留空则按 `deviceId` 反查已注册设备(*mock 模式两者都缺时用占位 token,只看渲染结果*) |
+| `deviceId` | string | ❌ | 客户端 `DeviceId.get()` 的设备 id,用于反查 vendor+token |
+| `type` | string | ❌ | 13 种类型 key 之一 → 按 PRD 模板渲染;不传且没直给文案 → 发通用测试文案 |
+| `vars` | object | ❌ | 覆盖模板变量,如 `{"coins":"520","cash":"6.66"}`;缺省用 PRD 示例值 |
+| `title` / `content` | string | ❌ | 直接指定标题/正文(优先于 type 模板) |
+| `createNotification` | bool | ❌ | true = 同时往该用户消息中心插一条同类型未读 mock 通知,extras 带其 `notificationId` → 可闭环验证「点 push → 调 /notifications/read 消红点」(仅 type 合法时生效) |
+| `mock` | bool | ❌ | **默认 true = 不真调厂商 API**,回显渲染结果;false = 真发到手机(要求该厂商凭据已配) |
+
+**出参**
+
+```jsonc
+{
+  "ok": true, "mock": true, "vendor": "huawei",
+  "title": "反馈奖励已到账",
+  "body": "谢谢您帮傻瓜比价变得更好，300金币已到账，还有一条给您的留言～",
+  "extras": { "type": "feedback_reward", "feedbackId": "3002", "notificationId": "90017" },
+  "notificationId": 90017,          // createNotification=true 时的站内 mock 通知 id
+  "missingKeys": ["HUAWEI_PUSH_APP_ID", "HUAWEI_PUSH_APP_SECRET"],  // 真发前还缺的配置(真发成功时必为空)
+  "vendorResponse": null            // 真发时为厂商 API 原始响应
+}
+```
+
+**错误**:`400` vendor/type 非法、真发但凭据未配(detail 列缺失键);`409` 真发但拿不到 pushToken;`502` 厂商 API 返回失败(detail 带厂商原始错误)。
+
+**真发注意**:
+- 目标手机必须先装 App 且客户端已集成对应厂商 SDK、`/device/register` 上报过 token;
+- vivo 未上架前走测试推送(`VIVO_PUSH_MODE=1`),目标手机需在 vivo 开放平台加入测试设备;
+- 小米新设备需在开放平台把签名/包名配好,token 才有效。
+
+## 厂商凭据怎么拿(.env 键名)
+
+| 厂商 | 后台 | 需要的键 |
+|---|---|---|
+| 华为 | AGC 控制台 → 项目设置 → 常规 → 应用 | `HUAWEI_PUSH_APP_ID`、`HUAWEI_PUSH_APP_SECRET`(OAuth client_id 即 AppId) |
+| 荣耀 | 荣耀开发者服务平台 → 推送服务 | `HONOR_PUSH_APP_ID`、`HONOR_PUSH_CLIENT_ID`、`HONOR_PUSH_CLIENT_SECRET` |
+| 小米 | 开放平台 → 消息推送 → 应用秘钥 | `XIAOMI_PUSH_APP_SECRET`(服务端只要这个;AppID/AppKey 是客户端 SDK 用) |
+| OPPO | 开放平台 → 推送服务 | `OPPO_PUSH_APP_KEY`、`OPPO_PUSH_MASTER_SECRET`(注意是**服务端 MasterSecret**) |
+| vivo | 开放平台 → 推送 | `VIVO_PUSH_APP_ID`、`VIVO_PUSH_APP_KEY`、`VIVO_PUSH_APP_SECRET` |
+
+各家发送协议差异(鉴权方式/成功码/payload 结构)封装在 `integrations/vendor_push.py`,业务侧只面对 `send_notification()`。

tests/test_admin_read.py

@@ -190,6 +190,67 @@ def test_feedback_list(admin_client: TestClient, admin_token: str) -> None:
     assert all(f["status"] == "pending" for f in r.json()["items"])
 
 
+def test_feedback_list_filter_by_source(admin_client: TestClient, admin_token: str) -> None:
+    """反馈列表按反馈类型(source)筛选;返回项带 source/scene 字段。"""
+    db = SessionLocal()
+    try:
+        uid = user_repo.upsert_user_for_login(db, phone="13800009001", register_channel="sms").id
+        db.add(Feedback(user_id=uid, content="比价反馈", contact="", status="pending",
+                        source="comparison", scene="找错商品"))
+        db.add(Feedback(user_id=uid, content="普通反馈", contact="", status="pending",
+                        source="profile"))
+        db.commit()
+    finally:
+        db.close()
+
+    r = admin_client.get(
+        "/admin/api/feedbacks", params={"user_id": uid, "source": "comparison"},
+        headers=_auth(admin_token),
+    )
+    assert r.status_code == 200, r.text
+    items = r.json()["items"]
+    assert items and all(it["source"] == "comparison" for it in items)
+    assert any(it["scene"] == "找错商品" for it in items)
+
+    r = admin_client.get(
+        "/admin/api/feedbacks", params={"user_id": uid, "source": "profile"},
+        headers=_auth(admin_token),
+    )
+    assert all(it["source"] == "profile" for it in r.json()["items"])
+
+    # 非法 source 值 → 422
+    assert admin_client.get(
+        "/admin/api/feedbacks", params={"source": "bogus"}, headers=_auth(admin_token)
+    ).status_code == 422
+
+
+def test_withdraw_list_exposes_source_and_filters(
+    admin_client: TestClient, admin_token: str
+) -> None:
+    """提现列表带 source 字段(coin_cash/invite_cash),并可按提现类型筛选。"""
+    db = SessionLocal()
+    try:
+        uid = user_repo.upsert_user_for_login(db, phone="13800009002", register_channel="sms").id
+        db.add(WithdrawOrder(user_id=uid, out_bill_no=f"src{uid}coin0001",
+                             amount_cents=100, status="success", source="coin_cash"))
+        db.add(WithdrawOrder(user_id=uid, out_bill_no=f"src{uid}invite001",
+                             amount_cents=200, status="success", source="invite_cash"))
+        db.commit()
+    finally:
+        db.close()
+
+    r = admin_client.get("/admin/api/withdraws", params={"user_id": uid}, headers=_auth(admin_token))
+    assert r.status_code == 200, r.text
+    assert {it["source"] for it in r.json()["items"]} == {"coin_cash", "invite_cash"}
+
+    r = admin_client.get(
+        "/admin/api/withdraws", params={"user_id": uid, "source": "invite_cash"},
+        headers=_auth(admin_token),
+    )
+    items = r.json()["items"]
+    assert items and all(it["source"] == "invite_cash" for it in items)
+
+
 def test_ad_coin_audit_full_count_truncate_and_only_mismatch(
     admin_client: TestClient, admin_token: str
 ) -> None:

tests/test_admin_write.py

@@ -368,6 +368,35 @@ def test_reject_feedback_records_reason_and_no_coin(
         db.close()
 
 
+def test_feedback_review_stores_admin_reply(
+    admin_client: TestClient, operator_token: str
+) -> None:
+    """审核可给用户留言(admin_reply,用户端可见),采纳/拒绝都能带,并回显在响应里。"""
+    fid = _seed_feedback("13900000021")
+    r = admin_client.post(
+        f"/admin/api/feedbacks/{fid}/approve",
+        json={"reward_coins": 100, "note": "内部备注", "reply": "感谢反馈,已采纳~"},
+        headers=_auth(operator_token),
+    )
+    assert r.status_code == 200, r.text
+    assert r.json()["admin_reply"] == "感谢反馈,已采纳~"
+    db = SessionLocal()
+    try:
+        fb = db.get(Feedback, fid)
+        assert fb.admin_reply == "感谢反馈,已采纳~" and fb.review_note == "内部备注"
+    finally:
+        db.close()
+
+    fid2 = _seed_feedback("13900000022")
+    r = admin_client.post(
+        f"/admin/api/feedbacks/{fid2}/reject",
+        json={"reason": "无法复现", "reply": "已收到,后续跟进"},
+        headers=_auth(operator_token),
+    )
+    assert r.status_code == 200, r.text
+    assert r.json()["admin_reply"] == "已收到,后续跟进"
+
+
 # ===== admin 账号管理(super_admin) =====
 
 def test_create_and_update_admin(admin_client: TestClient, super_token: str) -> None:

tests/test_device_push.py

@@ -0,0 +1,317 @@
+from __future__ import annotations
+
+import json
+from datetime import datetime, timedelta, timezone
+
+from fastapi.testclient import TestClient
+
+from app.api.v1 import device as device_api
+from app.core import heartbeat_monitor_worker
+from app.db.session import SessionLocal
+from app.integrations import vendor_push
+from app.models.device import DeviceLiveness
+from app.repositories import user as user_repo
+
+
+class _Resp:
+    status_code = 200
+    text = "{}"
+
+    def __init__(self, data: dict) -> None:
+        self._data = data
+
+    def json(self) -> dict:
+        return self._data
+
+
+def test_xiaomi_accessibility_payload(monkeypatch) -> None:
+    captured: dict = {}
+
+    def _fake_request(method, url, **kwargs):  # noqa: ANN001
+        captured.update(method=method, url=url, **kwargs)
+        return _Resp({"code": 0, "result": "ok", "data": {"id": "xm-msg"}})
+
+    monkeypatch.setattr(vendor_push.settings, "XIAOMI_PUSH_APP_SECRET", "xiaomi-secret")
+    monkeypatch.setattr(vendor_push.settings, "XIAOMI_PUSH_CHANNEL_ID", "")
+    monkeypatch.setattr(vendor_push.settings, "XIAOMI_PUSH_TEMPLATE_ID", "")
+    monkeypatch.setattr(vendor_push.settings, "XIAOMI_PUSH_TEMPLATE_TITLE", "")
+    monkeypatch.setattr(vendor_push.settings, "XIAOMI_PUSH_TEMPLATE_DESCRIPTION", "")
+    monkeypatch.setattr(vendor_push.settings, "XIAOMI_PUSH_TEMPLATE_PARAM_JSON", "")
+    monkeypatch.setattr(vendor_push.httpx, "request", _fake_request)
+
+    data = vendor_push.send_accessibility_disabled("xiaomi", "xm-regid")
+
+    assert data["data"]["id"] == "xm-msg"
+    assert captured["method"] == "POST"
+    assert captured["url"] == vendor_push.settings.XIAOMI_PUSH_SEND_ENDPOINT
+    assert captured["headers"]["Authorization"] == "key=xiaomi-secret"
+    body = captured["data"]
+    assert body["registration_id"] == "xm-regid"
+    assert body["restricted_package_name"] == "com.jishisongfu.shaguabijia"
+    assert json.loads(body["payload"]) == {"type": "accessibility_disabled"}
+    assert "extra.channel_id" not in body
+
+
+def test_xiaomi_payload_with_channel_and_template(monkeypatch) -> None:
+    captured: dict = {}
+
+    def _fake_request(method, url, **kwargs):  # noqa: ANN001
+        captured.update(method=method, url=url, **kwargs)
+        return _Resp({"code": 0, "result": "ok", "data": {"id": "xm-msg"}})
+
+    monkeypatch.setattr(vendor_push.settings, "XIAOMI_PUSH_APP_SECRET", "xiaomi-secret")
+    monkeypatch.setattr(vendor_push.settings, "XIAOMI_PUSH_CHANNEL_ID", "130")
+    monkeypatch.setattr(vendor_push.settings, "XIAOMI_PUSH_TEMPLATE_ID", "1001")
+    monkeypatch.setattr(vendor_push.settings, "XIAOMI_PUSH_TEMPLATE_TITLE", "{$app_name$}提醒")
+    monkeypatch.setattr(vendor_push.settings, "XIAOMI_PUSH_TEMPLATE_DESCRIPTION", "{$content$}")
+    monkeypatch.setattr(
+        vendor_push.settings,
+        "XIAOMI_PUSH_TEMPLATE_PARAM_JSON",
+        '{"app_name":"傻瓜比价","content":"{alert}"}',
+    )
+    monkeypatch.setattr(vendor_push.httpx, "request", _fake_request)
+
+    vendor_push.send_accessibility_disabled(
+        "xiaomi",
+        "xm-regid",
+        title="测试标题",
+        alert="测试内容",
+    )
+
+    body = captured["data"]
+    assert body["title"] == "{$app_name$}提醒"
+    assert body["description"] == "{$content$}"
+    assert body["extra.channel_id"] == "130"
+    assert body["extra.template_id"] == "1001"
+    assert body["extra.template_param"] == '{"app_name":"傻瓜比价","content":"测试内容"}'
+
+
+def test_vivo_auth_and_send_payload(monkeypatch) -> None:
+    vendor_push._token_cache.clear()
+    calls: list[dict] = []
+
+    def _fake_request(method, url, **kwargs):  # noqa: ANN001
+        calls.append({"method": method, "url": url, **kwargs})
+        if url == vendor_push.settings.VIVO_PUSH_AUTH_ENDPOINT:
+            return _Resp({"result": 0, "authToken": "vivo-auth"})
+        return _Resp({"result": 0, "taskId": "vivo-task"})
+
+    monkeypatch.setattr(vendor_push.settings, "VIVO_PUSH_APP_ID", "106072775")
+    monkeypatch.setattr(vendor_push.settings, "VIVO_PUSH_APP_KEY", "vivo-key")
+    monkeypatch.setattr(vendor_push.settings, "VIVO_PUSH_APP_SECRET", "vivo-secret")
+    monkeypatch.setattr(vendor_push.httpx, "request", _fake_request)
+
+    data = vendor_push.send_accessibility_disabled("vivo", "vivo-regid")
+
+    assert data["taskId"] == "vivo-task"
+    assert calls[0]["url"] == vendor_push.settings.VIVO_PUSH_AUTH_ENDPOINT
+    assert calls[0]["json"]["appId"] == "106072775"
+    assert calls[0]["json"]["sign"]
+    assert calls[1]["url"] == vendor_push.settings.VIVO_PUSH_SEND_ENDPOINT
+    assert calls[1]["headers"]["authToken"] == "vivo-auth"
+    body = calls[1]["json"]
+    assert body["regId"] == "vivo-regid"
+    assert body["pushMode"] == vendor_push.settings.VIVO_PUSH_MODE
+    assert body["clientCustomMap"] == {"type": "accessibility_disabled"}
+
+
+def test_oppo_auth_and_send_payload(monkeypatch) -> None:
+    vendor_push._token_cache.clear()
+    calls: list[dict] = []
+
+    def _fake_request(method, url, **kwargs):  # noqa: ANN001
+        calls.append({"method": method, "url": url, **kwargs})
+        if url == vendor_push.settings.OPPO_PUSH_AUTH_ENDPOINT:
+            return _Resp({"code": 0, "data": {"auth_token": "oppo-auth"}})
+        return _Resp({"code": 0, "data": {"message_id": "oppo-msg"}})
+
+    monkeypatch.setattr(vendor_push.settings, "OPPO_PUSH_APP_KEY", "oppo-key")
+    monkeypatch.setattr(vendor_push.settings, "OPPO_PUSH_MASTER_SECRET", "oppo-master")
+    monkeypatch.setattr(vendor_push.httpx, "request", _fake_request)
+
+    data = vendor_push.send_accessibility_disabled("oppo", "oppo-regid")
+
+    assert data["data"]["message_id"] == "oppo-msg"
+    assert calls[0]["data"]["app_key"] == "oppo-key"
+    assert calls[0]["data"]["sign"]
+    message = json.loads(calls[1]["data"]["message"])
+    assert calls[1]["data"]["auth_token"] == "oppo-auth"
+    assert message["target_type"] == 2
+    assert message["target_value"] == "oppo-regid"
+    assert json.loads(message["notification"]["action_parameters"]) == {
+        "type": "accessibility_disabled"
+    }
+
+
+def test_honor_auth_and_send_payload(monkeypatch) -> None:
+    vendor_push._token_cache.clear()
+    calls: list[dict] = []
+
+    def _fake_request(method, url, **kwargs):  # noqa: ANN001
+        calls.append({"method": method, "url": url, **kwargs})
+        if url == vendor_push.settings.HONOR_PUSH_TOKEN_ENDPOINT:
+            return _Resp({"access_token": "honor-access", "expires_in": 3600})
+        return _Resp({"code": 200, "message": "successful!", "data": {"sendResult": True}})
+
+    monkeypatch.setattr(vendor_push.settings, "HONOR_PUSH_APP_ID", "104559789")
+    monkeypatch.setattr(vendor_push.settings, "HONOR_PUSH_CLIENT_ID", "honor-client")
+    monkeypatch.setattr(vendor_push.settings, "HONOR_PUSH_CLIENT_SECRET", "honor-secret")
+    monkeypatch.setattr(vendor_push.httpx, "request", _fake_request)
+
+    data = vendor_push.send_accessibility_disabled("honor", "honor-token")
+
+    assert data["code"] == 200
+    assert calls[0]["data"]["client_id"] == "honor-client"
+    assert calls[1]["headers"]["Authorization"] == "Bearer honor-access"
+    assert calls[1]["headers"]["timestamp"]
+    assert calls[1]["url"].endswith("/api/v1/104559789/sendMessage")
+    body = calls[1]["json"]
+    assert body["token"] == ["honor-token"]
+    assert body["android"]["targetUserType"] == 1
+    assert body["android"]["notification"]["clickAction"] == {"type": 3}
+    assert json.loads(body["data"]) == {"type": "accessibility_disabled"}
+
+
+def _seed_overdue_device(
+    *,
+    phone: str,
+    device_id: str,
+    push_vendor: str | None,
+    push_token: str | None,
+) -> int:
+    with SessionLocal() as db:
+        user = user_repo.upsert_user_for_login(db, phone=phone, register_channel="sms")
+        device = DeviceLiveness(
+            user_id=user.id,
+            device_id=device_id,
+            push_vendor=push_vendor,
+            push_token=push_token,
+            platform="android",
+            ever_protected=True,
+            last_heartbeat_at=datetime.now(timezone.utc) - timedelta(minutes=30),  # noqa: UP017
+            last_report_protection_on=True,
+            liveness_state="alive",
+            kill_alert_pending=False,
+        )
+        db.add(device)
+        db.commit()
+        db.refresh(device)
+        return device.id
+
+
+def _login(client: TestClient, phone: str) -> str:
+    client.post("/api/v1/auth/sms/send", json={"phone": phone})
+    r = client.post("/api/v1/auth/sms/login", json={"phone": phone, "code": "123456"})
+    assert r.status_code == 200, r.text
+    return r.json()["access_token"]
+
+
+def _auth(token: str) -> dict[str, str]:
+    return {"Authorization": f"Bearer {token}"}
+
+
+def test_heartbeat_monitor_pushes_overdue_device(monkeypatch) -> None:
+    device_pk = _seed_overdue_device(
+        phone="13900009001",
+        device_id="dev-push-honor",
+        push_vendor="honor",
+        push_token="honor-token-1",
+    )
+    calls: list[tuple[str, str]] = []
+
+    def _fake_send(push_vendor: str, push_token: str) -> dict:
+        calls.append((push_vendor, push_token))
+        return {"msg_id": "m1"}
+
+    monkeypatch.setattr(
+        heartbeat_monitor_worker.vendor_push,
+        "send_accessibility_disabled",
+        _fake_send,
+    )
+
+    result = heartbeat_monitor_worker._scan_once(timeout_minutes=10)
+
+    assert result["pushed"] >= 1
+    assert ("honor", "honor-token-1") in calls
+    with SessionLocal() as db:
+        device = db.get(DeviceLiveness, device_pk)
+        assert device is not None
+        assert device.liveness_state == "notified"
+        assert device.kill_alert_pending is True
+
+
+def test_heartbeat_monitor_skips_push_without_vendor_token(monkeypatch) -> None:
+    device_pk = _seed_overdue_device(
+        phone="13900009002",
+        device_id="dev-push-no-token",
+        push_vendor=None,
+        push_token=None,
+    )
+
+    def _fake_send(push_vendor: str, push_token: str) -> dict:
+        raise AssertionError(f"should not push without token: {push_vendor}/{push_token}")
+
+    monkeypatch.setattr(
+        heartbeat_monitor_worker.vendor_push,
+        "send_accessibility_disabled",
+        _fake_send,
+    )
+
+    result = heartbeat_monitor_worker._scan_once(timeout_minutes=10)
+
+    assert result["checked"] >= 1
+    with SessionLocal() as db:
+        device = db.get(DeviceLiveness, device_pk)
+        assert device is not None
+        assert device.liveness_state == "notified"
+        assert device.kill_alert_pending is True
+
+
+def test_push_test_endpoint_schedules_vendor_push(client: TestClient, monkeypatch) -> None:
+    token = _login(client, "13900009003")
+    calls: list[tuple[str, str, str, str]] = []
+
+    def _fake_send(push_vendor: str, push_token: str, *, title: str, alert: str) -> dict:
+        calls.append((push_vendor, push_token, title, alert))
+        return {"msg_id": "m-test"}
+
+    monkeypatch.setattr(device_api.vendor_push, "send_accessibility_disabled", _fake_send)
+
+    r = client.post(
+        "/api/v1/device/push-test",
+        json={
+            "device_id": "dev-push-test",
+            "push_vendor": "honor",
+            "push_token": "honor-test-token",
+            "delay_seconds": 0,
+        },
+        headers=_auth(token),
+    )
+
+    assert r.status_code == 200, r.text
+    assert r.json() == {
+        "ok": True,
+        "delay_seconds": 0,
+        "has_push_token": True,
+    }
+    assert calls == [
+        (
+            "honor",
+            "honor-test-token",
+            "测试推送",
+            "这是一条厂商通道测试推送。收到它说明 App 被划掉后仍可通过系统通知栏触达。",
+        )
+    ]
+
+
+def test_push_test_endpoint_requires_vendor_token(client: TestClient) -> None:
+    token = _login(client, "13900009004")
+
+    r = client.post(
+        "/api/v1/device/push-test",
+        json={"device_id": "dev-push-test-no-token", "delay_seconds": 0},
+        headers=_auth(token),
+    )
+
+    assert r.status_code == 409
+    assert r.json()["detail"] == "push vendor token not ready"

tests/test_feedback.py

@@ -36,6 +36,43 @@ def test_feedback_config_returns_default(client) -> None:
     assert body["group_name"] == "傻瓜比价官方群"
 
 
+def test_submit_feedback_derives_source_from_scene(client) -> None:
+    """比价结果页反馈带 scene → source=comparison 并存 scene;「我的」页不带 → profile;
+    客户端显式传 source 优先(即便没 scene 也判 comparison)。历史接口带回 scene。"""
+    token = _login(client, "13622000010")
+    h = {"Authorization": f"Bearer {token}"}
+
+    r = client.post("/api/v1/feedback", data={"content": "比价太慢了", "scene": "比价太慢"}, headers=h)
+    assert r.status_code == 200, r.text
+    fid_cmp = r.json()["id"]
+
+    r = client.post("/api/v1/feedback", data={"content": "普通建议"}, headers=h)
+    assert r.status_code == 200, r.text
+    fid_prof = r.json()["id"]
+
+    r = client.post(
+        "/api/v1/feedback", data={"content": "只填了描述", "source": "comparison"}, headers=h
+    )
+    assert r.status_code == 200, r.text
+    fid_explicit = r.json()["id"]
+
+    db = SessionLocal()
+    try:
+        cmp_fb = db.get(Feedback, fid_cmp)
+        assert cmp_fb.source == "comparison" and cmp_fb.scene == "比价太慢"
+        prof_fb = db.get(Feedback, fid_prof)
+        assert prof_fb.source == "profile" and prof_fb.scene is None
+        assert db.get(Feedback, fid_explicit).source == "comparison"
+    finally:
+        db.close()
+
+    r = client.get("/api/v1/feedback/records", headers=h)
+    assert r.status_code == 200, r.text
+    scenes = {item["id"]: item["scene"] for item in r.json()["records"]}
+    assert scenes[fid_cmp] == "比价太慢"
+    assert scenes[fid_prof] is None
+
+
 def test_feedback_records_maps_existing_statuses(client) -> None:
     phone = "13622000003"
     token = _login(client, phone)

tests/test_notifications.py

@@ -0,0 +1,193 @@
+"""消息通知中心 3 接口(mock 数据阶段)。
+
+覆盖:列表字段/排序/分页、未读角标、标记已读(ids / all / 幂等 / 参数校验)、
+鉴权与用户间隔离。数据来自 repositories/notification_mock.py 的内存 mock。
+"""
+from __future__ import annotations
+
+from fastapi.testclient import TestClient
+
+from app.core import notification_catalog as catalog
+from app.repositories import notification_mock as mock_repo
+
+
+def _login(client: TestClient, phone: str) -> str:
+    client.post("/api/v1/auth/sms/send", json={"phone": phone})
+    r = client.post("/api/v1/auth/sms/login", json={"phone": phone, "code": "123456"})
+    assert r.status_code == 200, r.text
+    return r.json()["access_token"]
+
+
+def _auth(token: str) -> dict[str, str]:
+    return {"Authorization": f"Bearer {token}"}
+
+
+def _fetch_all(client: TestClient, token: str) -> dict:
+    r = client.get("/api/v1/notifications?pageSize=100", headers=_auth(token))
+    assert r.status_code == 200, r.text
+    return r.json()
+
+
+def test_requires_auth(client: TestClient) -> None:
+    assert client.get("/api/v1/notifications").status_code == 401
+    assert client.get("/api/v1/notifications/unread-count").status_code == 401
+    assert client.post("/api/v1/notifications/read", json={"all": True}).status_code == 401
+
+
+def test_list_covers_all_types_with_camel_case_fields(client: TestClient) -> None:
+    token = _login(client, "13900010001")
+    data = _fetch_all(client, token)
+
+    assert data["total"] >= 13
+    assert data["unreadCount"] > 0
+    items = data["items"]
+    assert len(items) == data["total"]
+
+    # 13 种类型全覆盖
+    assert {i["type"] for i in items} == set(catalog.TYPES)
+
+    # 字段按 PRD 契约 camelCase,且卡片要素齐全
+    first = items[0]
+    for key in (
+        "id", "category", "categoryLabel", "type", "cardStyle", "title",
+        "coins", "cashCents", "cashYuan", "infoRows", "actionText",
+        "extra", "sentAt", "isRead",
+    ):
+        assert key in first, f"missing field {key}"
+    assert "+08:00" in first["sentAt"]
+
+    # 双金额卡:金币 + 现金都有;金币为整数、现金串两位小数
+    expiring = next(i for i in items if i["type"] == "reward_expiring")
+    assert isinstance(expiring["coins"], int)
+    assert expiring["cashYuan"] == "12.80"
+    assert expiring["cashCents"] == 1280
+    assert expiring["actionText"] == "立即激活您的收益"
+    assert [row["label"] for row in expiring["infoRows"]] == ["过期说明", "过期时间"]
+
+    # 提现成功卡:无操作行、无跳转
+    ws = next(i for i in items if i["type"] == "withdraw_success")
+    assert ws["actionText"] is None
+    assert ws["coins"] is None
+
+    # 权限异常卡带 permission 参数(客户端点击时实时检测用)
+    perm = next(i for i in items if i["type"] == "perm_accessibility")
+    assert perm["extra"] == {"permission": "accessibility"}
+
+    # 反馈奖励卡:官方留言必填(PRD §3)
+    reward = next(i for i in items if i["type"] == "feedback_reward")
+    assert any(row["label"] == "官方留言" and row["value"] for row in reward["infoRows"])
+
+
+def test_list_sorted_by_time_desc_without_grouping(client: TestClient) -> None:
+    """全列表时间倒序、不按分类分组(PRD 原文的分组已确认取消)。"""
+    token = _login(client, "13900010002")
+    items = _fetch_all(client, token)["items"]
+
+    sent_ats = [i["sentAt"] for i in items]
+    assert sent_ats == sorted(sent_ats, reverse=True), "全列表时间倒序,最新在前"
+
+    # mock 样例里「好友下单(10 分钟前)」比「昨天的提现失败」新但分类靠后 →
+    # 若还在按分类分组,它不可能排进前 3;以此钉死"确实没分组"。
+    newest_types = {i["type"] for i in items[:3]}
+    assert "invite_order_reward" in newest_types
+
+
+def test_pagination(client: TestClient) -> None:
+    token = _login(client, "13900010003")
+    total = _fetch_all(client, token)["total"]
+
+    page_size = 5
+    seen_ids: list[int] = []
+    page = 1
+    while True:
+        r = client.get(
+            f"/api/v1/notifications?page={page}&pageSize={page_size}", headers=_auth(token)
+        )
+        assert r.status_code == 200
+        data = r.json()
+        assert data["page"] == page
+        assert data["pageSize"] == page_size
+        assert data["total"] == total
+        seen_ids.extend(i["id"] for i in data["items"])
+        if not data["hasMore"]:
+            assert len(data["items"]) <= page_size
+            break
+        assert len(data["items"]) == page_size
+        page += 1
+
+    assert len(seen_ids) == total
+    assert len(set(seen_ids)) == total, "翻页不重不漏"
+
+    # 超出末页 → 空页而非报错
+    r = client.get("/api/v1/notifications?page=99&pageSize=50", headers=_auth(token))
+    assert r.status_code == 200
+    assert r.json()["items"] == []
+    assert r.json()["hasMore"] is False
+
+
+def test_unread_count_and_badge(client: TestClient) -> None:
+    token = _login(client, "13900010004")
+    data = _fetch_all(client, token)
+    expect_unread = sum(1 for i in data["items"] if not i["isRead"])
+
+    r = client.get("/api/v1/notifications/unread-count", headers=_auth(token))
+    assert r.status_code == 200
+    body = r.json()
+    assert body["count"] == expect_unread == data["unreadCount"]
+    assert body["badgeText"] == str(expect_unread)  # mock 样例未读数在 1~99 区间
+
+    # 全部读完 → count=0,badgeText=null(整个角标隐藏)
+    client.post("/api/v1/notifications/read", json={"all": True}, headers=_auth(token))
+    r = client.get("/api/v1/notifications/unread-count", headers=_auth(token))
+    assert r.json() == {"count": 0, "badgeText": None}
+
+
+def test_mark_read_by_ids_idempotent(client: TestClient) -> None:
+    token = _login(client, "13900010005")
+    items = _fetch_all(client, token)["items"]
+    unread_ids = [i["id"] for i in items if not i["isRead"]]
+    before = len(unread_ids)
+    picked = unread_ids[:2]
+
+    r = client.post("/api/v1/notifications/read", json={"ids": picked}, headers=_auth(token))
+    assert r.status_code == 200
+    assert r.json() == {"ok": True, "markedCount": 2, "unreadCount": before - 2}
+
+    # 列表状态同步翻转
+    items = _fetch_all(client, token)["items"]
+    assert all(i["isRead"] for i in items if i["id"] in picked)
+
+    # 重复置读 + 不存在的 id → 幂等,不报错
+    r = client.post(
+        "/api/v1/notifications/read", json={"ids": [*picked, 123456789]}, headers=_auth(token)
+    )
+    assert r.status_code == 200
+    assert r.json()["markedCount"] == 0
+    assert r.json()["unreadCount"] == before - 2
+
+
+def test_mark_read_requires_ids_or_all(client: TestClient) -> None:
+    token = _login(client, "13900010006")
+    r = client.post("/api/v1/notifications/read", json={}, headers=_auth(token))
+    assert r.status_code == 400
+    r = client.post("/api/v1/notifications/read", json={"ids": []}, headers=_auth(token))
+    assert r.status_code == 400
+
+
+def test_stores_isolated_between_users(client: TestClient) -> None:
+    token_a = _login(client, "13900010007")
+    token_b = _login(client, "13900010008")
+
+    client.post("/api/v1/notifications/read", json={"all": True}, headers=_auth(token_a))
+
+    r = client.get("/api/v1/notifications/unread-count", headers=_auth(token_b))
+    assert r.json()["count"] > 0, "A 清零不影响 B"
+
+
+def test_reset_store_regenerates(client: TestClient) -> None:
+    token = _login(client, "13900010009")
+    client.post("/api/v1/notifications/read", json={"all": True}, headers=_auth(token))
+    assert client.get("/api/v1/notifications/unread-count", headers=_auth(token)).json()["count"] == 0
+
+    mock_repo.reset_store()  # 联调辅助:清空后重新生成整套样例
+    assert client.get("/api/v1/notifications/unread-count", headers=_auth(token)).json()["count"] > 0

tests/test_push_center.py

@@ -0,0 +1,447 @@
+"""厂商推送(5 家)+ 推送测试三件套。
+
+覆盖:华为 Push Kit 发送链路(OAuth + messages:send payload)、send_notification 通用入口
+与 mock 模式、/push/vendors 配置状态、/push/templates 模板渲染、/push/test 的
+mock/真发/变量覆盖/站内联动/设备反查/参数校验。厂商 HTTP 全部 monkeypatch,不真发。
+"""
+from __future__ import annotations
+
+import json
+
+import pytest
+from fastapi.testclient import TestClient
+
+from app.integrations import vendor_push
+
+_ALL_VENDOR_SETTINGS = [key for keys in vendor_push.REQUIRED_SETTINGS.values() for key in keys]
+
+
+class _Resp:
+    status_code = 200
+    text = "{}"
+
+    def __init__(self, data: dict) -> None:
+        self._data = data
+
+    def json(self) -> dict:
+        return self._data
+
+
+@pytest.fixture()
+def _no_vendor_creds(monkeypatch) -> None:
+    """把 5 家厂商凭据全部清空(隔离本机 .env 里已填的真实密钥,保证用例确定性)。"""
+    for key in _ALL_VENDOR_SETTINGS:
+        monkeypatch.setattr(vendor_push.settings, key, "")
+
+
+def _login(client: TestClient, phone: str) -> str:
+    client.post("/api/v1/auth/sms/send", json={"phone": phone})
+    r = client.post("/api/v1/auth/sms/login", json={"phone": phone, "code": "123456"})
+    assert r.status_code == 200, r.text
+    return r.json()["access_token"]
+
+
+def _auth(token: str) -> dict[str, str]:
+    return {"Authorization": f"Bearer {token}"}
+
+
+# ---------------------------------------------------------------------------
+# integrations.vendor_push:华为链路 + 通用入口
+# ---------------------------------------------------------------------------
+
+
+def test_huawei_auth_and_send_payload(monkeypatch) -> None:
+    vendor_push._token_cache.clear()
+    calls: list[dict] = []
+
+    def _fake_request(method, url, **kwargs):  # noqa: ANN001
+        calls.append({"method": method, "url": url, **kwargs})
+        if url == vendor_push.settings.HUAWEI_PUSH_TOKEN_ENDPOINT:
+            return _Resp({"access_token": "hw-access", "expires_in": 3600})
+        return _Resp({"code": "80000000", "msg": "Success", "requestId": "req-1"})
+
+    monkeypatch.setattr(vendor_push.settings, "HUAWEI_PUSH_APP_ID", "10086001")
+    monkeypatch.setattr(vendor_push.settings, "HUAWEI_PUSH_APP_SECRET", "hw-secret")
+    monkeypatch.setattr(vendor_push.httpx, "request", _fake_request)
+
+    data = vendor_push.send_notification(
+        "huawei",
+        "hw-token",
+        title="测试标题",
+        body="测试内容",
+        extras={"type": "withdraw_success", "notificationId": "90001"},
+    )
+
+    assert data["code"] == "80000000"
+    # OAuth:client_id 即 AppId
+    assert calls[0]["url"] == vendor_push.settings.HUAWEI_PUSH_TOKEN_ENDPOINT
+    assert calls[0]["data"]["grant_type"] == "client_credentials"
+    assert calls[0]["data"]["client_id"] == "10086001"
+    # 发送:v1 messages:send,Bearer 鉴权,token 数组 + data 透传 extras
+    assert calls[1]["url"].endswith("/v1/10086001/messages:send")
+    assert calls[1]["headers"]["Authorization"] == "Bearer hw-access"
+    message = calls[1]["json"]["message"]
+    assert message["token"] == ["hw-token"]
+    assert message["android"]["notification"]["title"] == "测试标题"
+    assert message["android"]["notification"]["click_action"] == {"type": 3}
+    assert json.loads(message["data"]) == {"type": "withdraw_success", "notificationId": "90001"}
+
+
+def test_huawei_non_success_code_raises(monkeypatch) -> 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": "hw-access", "expires_in": 3600})
+        return _Resp({"code": "80300007", "msg": "all tokens are invalid"})
+
+    monkeypatch.setattr(vendor_push.settings, "HUAWEI_PUSH_APP_ID", "10086001")
+    monkeypatch.setattr(vendor_push.settings, "HUAWEI_PUSH_APP_SECRET", "hw-secret")
+    monkeypatch.setattr(vendor_push.httpx, "request", _fake_request)
+
+    with pytest.raises(vendor_push.VendorPushError, match="huawei push failed"):
+        vendor_push.send_notification("huawei", "bad-token", title="t", body="b")
+
+
+def test_vendor_aliases_normalize() -> None:
+    assert vendor_push.normalize_vendor("华为") == "huawei"
+    assert vendor_push.normalize_vendor("HMS") == "huawei"
+    assert vendor_push.normalize_vendor("荣耀") == "honor"
+    assert vendor_push.normalize_vendor("小米") == "xiaomi"
+    assert vendor_push.SUPPORTED_VENDORS == {"honor", "huawei", "xiaomi", "oppo", "vivo"}
+
+
+def test_send_notification_mock_skips_http(monkeypatch) -> None:
+    def _boom(*args, **kwargs):  # noqa: ANN001, ANN002, ANN003
+        raise AssertionError("mock 模式不应发起任何 HTTP 请求")
+
+    monkeypatch.setattr(vendor_push.httpx, "request", _boom)
+
+    data = vendor_push.send_notification(
+        "oppo", "any-token", title="标题", body="正文", extras={"type": "push_test"}, mock=True
+    )
+    assert data == {
+        "mock": True,
+        "vendor": "oppo",
+        "title": "标题",
+        "body": "正文",
+        "extras": {"type": "push_test"},
+    }
+
+
+def test_send_notification_rejects_unknown_vendor() -> None:
+    with pytest.raises(vendor_push.VendorPushError, match="unsupported push vendor"):
+        vendor_push.send_notification("nokia", "t", title="a", body="b", mock=True)
+    with pytest.raises(vendor_push.VendorPushError, match="token is empty"):
+        vendor_push.send_notification("huawei", "  ", title="a", body="b", mock=True)
+
+
+def test_accessibility_wrapper_keeps_legacy_extras(monkeypatch) -> None:
+    """旧入口 send_accessibility_disabled 仍传 {"type":"accessibility_disabled"}(worker 兼容)。"""
+    captured: dict = {}
+
+    def _fake_request(method, url, **kwargs):  # noqa: ANN001
+        captured.update(method=method, url=url, **kwargs)
+        return _Resp({"code": 0, "result": "ok"})
+
+    monkeypatch.setattr(vendor_push.settings, "XIAOMI_PUSH_APP_SECRET", "xiaomi-secret")
+    monkeypatch.setattr(vendor_push.settings, "XIAOMI_PUSH_TEMPLATE_PARAM_JSON", "")
+    monkeypatch.setattr(vendor_push.settings, "XIAOMI_PUSH_TEMPLATE_TITLE", "")
+    monkeypatch.setattr(vendor_push.settings, "XIAOMI_PUSH_TEMPLATE_DESCRIPTION", "")
+    monkeypatch.setattr(vendor_push.httpx, "request", _fake_request)
+
+    vendor_push.send_accessibility_disabled("xiaomi", "regid-1")
+
+    assert json.loads(captured["data"]["payload"]) == {"type": "accessibility_disabled"}
+
+
+def test_oppo_payload_includes_new_message_category(monkeypatch) -> None:
+    """OPPO 新消息分类:配置了 channel_id/category 时随通知体下发(2024-11 新规必带)。"""
+    vendor_push._token_cache.clear()
+    calls: list[dict] = []
+
+    def _fake_request(method, url, **kwargs):  # noqa: ANN001
+        calls.append({"method": method, "url": url, **kwargs})
+        if url == vendor_push.settings.OPPO_PUSH_AUTH_ENDPOINT:
+            return _Resp({"code": 0, "data": {"auth_token": "oppo-auth"}})
+        return _Resp({"code": 0, "data": {"message_id": "oppo-msg"}})
+
+    monkeypatch.setattr(vendor_push.settings, "OPPO_PUSH_APP_KEY", "oppo-key")
+    monkeypatch.setattr(vendor_push.settings, "OPPO_PUSH_MASTER_SECRET", "oppo-master")
+    monkeypatch.setattr(vendor_push.settings, "OPPO_PUSH_CHANNEL_ID", "push_oplus_category_content")
+    monkeypatch.setattr(vendor_push.settings, "OPPO_PUSH_CATEGORY", "MARKETING")
+    monkeypatch.setattr(vendor_push.settings, "OPPO_PUSH_NOTIFY_LEVEL", 0)
+    monkeypatch.setattr(vendor_push.httpx, "request", _fake_request)
+
+    vendor_push.send_notification(
+        "oppo", "oppo-regid", title="标题", body="正文", extras={"type": "push_test"}
+    )
+
+    notification = json.loads(calls[1]["data"]["message"])["notification"]
+    assert notification["channel_id"] == "push_oplus_category_content"
+    assert notification["category"] == "MARKETING"
+    assert "notify_level" not in notification  # 0=不传,走 OPPO 默认
+
+
+def test_oppo_payload_omits_category_when_unconfigured(monkeypatch) -> None:
+    vendor_push._token_cache.clear()
+    calls: list[dict] = []
+
+    def _fake_request(method, url, **kwargs):  # noqa: ANN001
+        calls.append({"method": method, "url": url, **kwargs})
+        if url == vendor_push.settings.OPPO_PUSH_AUTH_ENDPOINT:
+            return _Resp({"code": 0, "data": {"auth_token": "oppo-auth"}})
+        return _Resp({"code": 0, "data": {"message_id": "oppo-msg"}})
+
+    monkeypatch.setattr(vendor_push.settings, "OPPO_PUSH_APP_KEY", "oppo-key")
+    monkeypatch.setattr(vendor_push.settings, "OPPO_PUSH_MASTER_SECRET", "oppo-master")
+    monkeypatch.setattr(vendor_push.settings, "OPPO_PUSH_CHANNEL_ID", "")
+    monkeypatch.setattr(vendor_push.settings, "OPPO_PUSH_CATEGORY", "")
+    monkeypatch.setattr(vendor_push.settings, "OPPO_PUSH_NOTIFY_LEVEL", 0)
+    monkeypatch.setattr(vendor_push.httpx, "request", _fake_request)
+
+    vendor_push.send_notification(
+        "oppo", "oppo-regid", title="标题", body="正文", extras={"type": "push_test"}
+    )
+
+    notification = json.loads(calls[1]["data"]["message"])["notification"]
+    assert "channel_id" not in notification
+    assert "category" not in notification
+
+
+# ---------------------------------------------------------------------------
+# /api/v1/push 三件套
+# ---------------------------------------------------------------------------
+
+
+def test_vendors_status_reports_missing_keys(client: TestClient, monkeypatch, _no_vendor_creds) -> None:
+    monkeypatch.setattr(vendor_push.settings, "XIAOMI_PUSH_APP_SECRET", "xiaomi-secret")
+    token = _login(client, "13900011001")
+
+    r = client.get("/api/v1/push/vendors", headers=_auth(token))
+    assert r.status_code == 200
+    vendors = {v["vendor"]: v for v in r.json()["vendors"]}
+    assert list(vendors) == ["honor", "huawei", "xiaomi", "oppo", "vivo"]
+
+    assert vendors["xiaomi"]["configured"] is True
+    assert vendors["xiaomi"]["missingKeys"] == []
+    assert vendors["huawei"]["configured"] is False
+    assert vendors["huawei"]["missingKeys"] == ["HUAWEI_PUSH_APP_ID", "HUAWEI_PUSH_APP_SECRET"]
+    assert vendors["honor"]["label"] == "荣耀"
+    assert vendors["vivo"]["missingKeys"] == [
+        "VIVO_PUSH_APP_ID", "VIVO_PUSH_APP_KEY", "VIVO_PUSH_APP_SECRET",
+    ]
+
+
+def test_templates_render_all_13_types(client: TestClient) -> None:
+    token = _login(client, "13900011002")
+    r = client.get("/api/v1/push/templates", headers=_auth(token))
+    assert r.status_code == 200
+    templates = r.json()["templates"]
+    assert len(templates) == 13
+
+    by_type = {t["type"]: t for t in templates}
+    ws = by_type["withdraw_success"]
+    assert ws["pushTitle"] == "提现到账提醒"
+    assert ws["pushBodySample"] == "¥0.50已存入您的微信钱包，点击查看到账详情"
+    assert ws["variables"] == ["amount"]
+
+    expiring = by_type["reward_expiring"]
+    assert "86金币" in expiring["pushBodySample"]
+    assert "{coins}" in expiring["pushBodyTemplate"]
+    assert expiring["sampleVars"]["cash"] == "12.80"
+
+    # 权限类标题按类型写死功能名
+    assert by_type["perm_overlay"]["pushTitle"] == "检测到您的比价按钮已失效"
+
+
+def test_push_test_mock_renders_template(client: TestClient, _no_vendor_creds) -> None:
+    token = _login(client, "13900011003")
+    r = client.post(
+        "/api/v1/push/test",
+        json={"vendor": "华为", "pushToken": "hw-token-1", "type": "withdraw_success"},
+        headers=_auth(token),
+    )
+    assert r.status_code == 200, r.text
+    body = r.json()
+    assert body["ok"] is True
+    assert body["mock"] is True
+    assert body["vendor"] == "huawei"  # 中文别名已归一化
+    assert body["title"] == "提现到账提醒"
+    assert body["body"] == "¥0.50已存入您的微信钱包，点击查看到账详情"
+    assert body["extras"] == {"type": "withdraw_success"}
+    assert body["missingKeys"] == ["HUAWEI_PUSH_APP_ID", "HUAWEI_PUSH_APP_SECRET"]
+    assert body["vendorResponse"] is None
+
+
+def test_push_test_vars_override(client: TestClient, _no_vendor_creds) -> None:
+    token = _login(client, "13900011004")
+    r = client.post(
+        "/api/v1/push/test",
+        json={
+            "vendor": "xiaomi",
+            "pushToken": "xm-1",
+            "type": "invite_order_reward",
+            "vars": {"nickname": "小王", "amount": "6.66"},
+        },
+        headers=_auth(token),
+    )
+    assert r.status_code == 200
+    assert r.json()["body"] == "您的好友「小王」完成首次下单，6.66元现金已到账"
+
+
+def test_push_test_generic_copy_without_type(client: TestClient, _no_vendor_creds) -> None:
+    token = _login(client, "13900011005")
+    r = client.post(
+        "/api/v1/push/test",
+        json={"vendor": "oppo", "pushToken": "op-1"},
+        headers=_auth(token),
+    )
+    assert r.status_code == 200
+    body = r.json()
+    assert body["extras"]["type"] == "push_test"
+    assert "OPPO" in body["body"]
+
+
+def test_push_test_create_notification_links_message_center(
+    client: TestClient, _no_vendor_creds
+) -> None:
+    token = _login(client, "13900011006")
+    before = client.get("/api/v1/notifications/unread-count", headers=_auth(token)).json()["count"]
+
+    r = client.post(
+        "/api/v1/push/test",
+        json={
+            "vendor": "vivo",
+            "pushToken": "vv-1",
+            "type": "feedback_reward",
+            "createNotification": True,
+        },
+        headers=_auth(token),
+    )
+    assert r.status_code == 200
+    body = r.json()
+    nid = body["notificationId"]
+    assert isinstance(nid, int)
+    assert body["extras"]["notificationId"] == str(nid)
+    assert body["extras"]["type"] == "feedback_reward"
+    assert body["extras"]["feedbackId"]  # 业务参数一并带上,客户端可直达反馈详情
+
+    # 站内多了一条未读;按 push extras 的 id 置读 → 闭环
+    after = client.get("/api/v1/notifications/unread-count", headers=_auth(token)).json()["count"]
+    assert after == before + 1
+    r = client.post("/api/v1/notifications/read", json={"ids": [nid]}, headers=_auth(token))
+    assert r.json()["markedCount"] == 1
+
+
+def test_push_test_real_send_requires_credentials(client: TestClient, _no_vendor_creds) -> None:
+    token = _login(client, "13900011007")
+    r = client.post(
+        "/api/v1/push/test",
+        json={"vendor": "huawei", "pushToken": "hw-1", "mock": False},
+        headers=_auth(token),
+    )
+    assert r.status_code == 400
+    assert "HUAWEI_PUSH_APP_ID" in r.json()["detail"]
+
+
+def test_push_test_real_send_xiaomi(client: TestClient, monkeypatch, _no_vendor_creds) -> None:
+    captured: dict = {}
+
+    def _fake_request(method, url, **kwargs):  # noqa: ANN001
+        captured.update(method=method, url=url, **kwargs)
+        return _Resp({"code": 0, "result": "ok", "data": {"id": "xm-real"}})
+
+    monkeypatch.setattr(vendor_push.settings, "XIAOMI_PUSH_APP_SECRET", "xiaomi-secret")
+    monkeypatch.setattr(vendor_push.httpx, "request", _fake_request)
+    token = _login(client, "13900011008")
+
+    r = client.post(
+        "/api/v1/push/test",
+        json={
+            "vendor": "xiaomi",
+            "pushToken": "xm-regid-9",
+            "type": "report_approved",
+            "mock": False,
+        },
+        headers=_auth(token),
+    )
+    assert r.status_code == 200, r.text
+    body = r.json()
+    assert body["mock"] is False
+    assert body["missingKeys"] == []
+    assert body["vendorResponse"]["data"]["id"] == "xm-real"
+    assert captured["data"]["registration_id"] == "xm-regid-9"
+    assert captured["data"]["title"] == "爆料审核通过"
+    assert "蜀大侠火锅" in captured["data"]["description"]
+    assert json.loads(captured["data"]["payload"]) == {"type": "report_approved"}
+
+
+def test_push_test_real_send_vendor_error_maps_502(
+    client: TestClient, monkeypatch, _no_vendor_creds
+) -> None:
+    def _fake_request(method, url, **kwargs):  # noqa: ANN001
+        return _Resp({"code": 500, "result": "error", "reason": "invalid regid"})
+
+    monkeypatch.setattr(vendor_push.settings, "XIAOMI_PUSH_APP_SECRET", "xiaomi-secret")
+    monkeypatch.setattr(vendor_push.httpx, "request", _fake_request)
+    token = _login(client, "13900011009")
+
+    r = client.post(
+        "/api/v1/push/test",
+        json={"vendor": "xiaomi", "pushToken": "bad", "mock": False},
+        headers=_auth(token),
+    )
+    assert r.status_code == 502
+    assert "厂商推送失败" in r.json()["detail"]
+
+
+def test_push_test_resolves_token_from_registered_device(
+    client: TestClient, _no_vendor_creds
+) -> None:
+    token = _login(client, "13900011010")
+    r = client.post(
+        "/api/v1/device/register",
+        json={"device_id": "dev-push-center-1", "push_vendor": "honor", "push_token": "honor-t1"},
+        headers=_auth(token),
+    )
+    assert r.status_code == 200, r.text
+
+    r = client.post(
+        "/api/v1/push/test",
+        json={"deviceId": "dev-push-center-1", "type": "perm_accessibility"},
+        headers=_auth(token),
+    )
+    assert r.status_code == 200, r.text
+    body = r.json()
+    assert body["vendor"] == "honor"
+    assert body["title"] == "检测到您的比价功能已失效"
+
+
+def test_push_test_validation_errors(client: TestClient, _no_vendor_creds) -> None:
+    token = _login(client, "13900011011")
+
+    # 未知厂商
+    r = client.post(
+        "/api/v1/push/test",
+        json={"vendor": "nokia", "pushToken": "t"},
+        headers=_auth(token),
+    )
+    assert r.status_code == 400
+
+    # 未知类型
+    r = client.post(
+        "/api/v1/push/test",
+        json={"vendor": "xiaomi", "pushToken": "t", "type": "bogus"},
+        headers=_auth(token),
+    )
+    assert r.status_code == 400
+    assert "unknown notification type" in r.json()["detail"]
+
+    # 真发但没有 token 可用
+    r = client.post(
+        "/api/v1/push/test",
+        json={"vendor": "xiaomi", "mock": False},
+        headers=_auth(token),
+    )
+    assert r.status_code == 409