feat(observe): 生产部署硬化 + nginx 反代(Certbot+BasicAuth) + 恢复采集中间件

- app/main.py: 恢复 merge 丢失的 RequestMetricsMiddleware 注册,接口埋点重新生效
- docker-compose.prod.yml: 镜像 pin v0.91.2、数据保留 30 天、关匿名遥测(ZO_TELEMETRY)、容器日志上限、资源硬上限(2C/3G)、数据改 bind-mount 到宿主 /data
- observe.shaguabijia.com.conf: 改用 Certbot 证书(与 admin-web 一致)+ Basic Auth 兜底(办公网无固定 IP)+ 去掉与 admin-web 冲突的 ipv6only=on
- dashboard-api-metrics.json: 补 dashboardId 修复新版 OpenObserve 导入校验
- app-api.conf: client_max_body_size 4m->32m,修带截图反馈被 nginx 413(顺带)

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

.env.example

@@ -27,55 +27,7 @@ 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
 
-# ===== 厂商直推(无障碍保护存活告警 + 消息中心 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 后置兜底)=====
+# ===== 无障碍保护存活监控(pull 后置检测;本期不接推送)=====
 HEARTBEAT_MONITOR_ENABLED=true
 HEARTBEAT_TIMEOUT_MINUTES=60
 HEARTBEAT_SCAN_INTERVAL_SEC=60

AGENTS.md

@@ -1,116 +0,0 @@
-# AGENTS.md
-
-This file provides guidance to Codex (Codex.ai/code) when working with code in this repository.
-
-## Project overview
-
-Shaguabijia (傻瓜比价) App backend — FastAPI + SQLAlchemy 2.0 + JWT. Covers user auth (Jiguang one-click / SMS), welfare wallet (coins/cash/signin/tasks/savings), WeChat Pay withdrawals, ad-reward callbacks (Pangle/GroMore S2S), Meituan CPS (coupon forwarding / price comparison), and an admin backend.
-
-## Commands
-
-```bash
-# Install
-pip install -e ".[dev]"
-
-# Run app server (port 8770, auto-migrates, auto-reload)
-./run.sh                    # or: uvicorn app.main:app --reload --port 8770
-
-# Run admin server (port 8771, separate process)
-uvicorn app.admin.main:admin_app --reload --port 8771
-
-# Database
-alembic upgrade head        # apply all migrations (idempotent)
-alembic revision --autogenerate -m "description"  # generate new migration
-
-# Tests
-pytest -q                   # all tests
-pytest tests/test_auth.py -q        # single file
-pytest -k "test_sms_login" -q       # single test by name
-
-# Lint
-ruff check .
-ruff check --fix .
-```
-
-## Architecture: two FastAPI apps
-
-This repo runs **two separate FastAPI processes** sharing the same `app/` codebase (models, repositories, integrations, config):
-
-| | App server | Admin server |
-|---|---|---|
-| Entry | `app/main.py` → `app:app` | `app/admin/main.py` → `admin_app` |
-| Port | 8770 | 8771 |
-| Auth | User JWT (`JWT_SECRET_KEY`) | Admin JWT (`ADMIN_JWT_SECRET`, independent) |
-| Audience | Mobile app clients | Internal admin dashboard |
-| Docs | `/docs` (non-prod only) | `/admin/docs` (non-prod only) |
-
-The two apps are intentionally decoupled — `app.main` never imports `app.admin`. Admin has its own auth chain (`app/admin/deps.py`, `app/admin/security.py`), role-based guards (`require_role`), and routers under `app/admin/routers/`.
-
-## Layered request flow
-
-```
-api/v1/  (thin: parse → delegate → respond + HTTP errors)
-  ├── integrations/   (external SDKs: signature, encryption, HTTP calls)
-  └── repositories/   (data access + transactions)
-        └── models/   (SQLAlchemy ORM, DeclarativeBase)
-```
-
-- **`api/v1/`**: Route handlers. Keep these thin — parse request, call repository or integration, return response. Never put business logic or external HTTP here.
-- **`api/deps.py`**: Shared FastAPI dependencies — `get_current_user` (Bearer JWT → User ORM object), `get_db` (request-scoped session).
-- **`integrations/`**: All external service logic — Jiguang REST + RSA decryption, WeChat Pay V3 signing/encryption, Meituan CPS gateway signing, Pangle callback signature verification, SMS sending. This is the layer you change when swapping vendors.
-- **`repositories/`**: Data access. Each file wraps SQLAlchemy queries + transactions for one domain (user, wallet, signin, savings, ad_reward, etc.). Some repositories also call integrations (e.g., `wallet.py` calls `integrations/wxpay.py` for withdrawals).
-- **`models/`**: ORM table definitions (SQLAlchemy 2.0 `Mapped` style, `DeclarativeBase`). Every new model must be imported in `app/models/__init__.py` so Alembic can discover it.
-- **`schemas/`**: Pydantic request/response contracts.
-- **`core/`**: Infrastructure — config (`pydantic-settings`), JWT (`security.py`), in-memory rate limiter (`ratelimit.py`), reward constants (`rewards.py`), logging setup, pricebot router (consistent-hash load balancing), withdraw reconcile worker.
-
-## Internal (server-to-server) endpoints
-
-Endpoints under `app/api/internal/` are for server-to-server communication (pricebot → app-server), NOT for clients. They use a shared secret header `X-Internal-Secret` (compared via `hmac.compare_digest`) instead of user JWT. If `INTERNAL_API_SECRET` is empty, these endpoints return 503.
-
-## Auth system
-
-- **User login**: Jiguang one-click (`integrations/jiguang.py` — REST token verification + RSA decryption with multi-padding retry) or SMS code (mock by default; `SMS_MOCK=true`).
-- **Tokens**: JWT access (2h) + refresh (30d). Both are JWT with `typ` claim (`"access"` vs `"refresh"`) to prevent refresh-as-access. See `core/security.py`.
-- **Admin auth**: Separate JWT secret (`ADMIN_JWT_SECRET`), 12h expiry, no refresh. Username + bcrypt password login. Role-based access via `require_role()` guard in `app/admin/deps.py` (`super_admin` bypasses all role checks).
-- **Rate limiting**: In-memory fixed-window by client IP (`core/ratelimit.py`). Single-worker only; disabled in tests via `RATE_LIMIT_ENABLED=false`.
-
-## Database
-
-- **Dev**: SQLite (`sqlite:///./data/app.db`), `check_same_thread=False`, no connection pool.
-- **Prod**: PostgreSQL — just change `DATABASE_URL` in `.env`. Pool size 10 + max overflow 20, pool_recycle 3600.
-- **Migrations**: Alembic with `render_as_batch` for SQLite compatibility. ~60+ migration files in `alembic/versions/` (filenames are descriptive, not hex prefixes). Migration chain uses `down_revision` within each file.
-- **New models**: Define in `app/models/`, import in `app/models/__init__.py`, then run `alembic revision --autogenerate`.
-
-## Config
-
-All config via `pydantic-settings` in `app/core/config.py`. Single `Settings` class with env vars / `.env` file. Access anywhere via `from app.core.config import settings`. Key patterns:
-- `*_configured` properties gate features gracefully (e.g., `mt_cps_configured`, `wxpay_configured`, `pangle_callback_configured`) — missing credentials → endpoints return empty/503 rather than crashing at startup.
-- Prod validation: `_enforce_prod_secrets` model validator blocks startup if `APP_ENV=prod` with weak JWT secrets.
-
-## Testing
-
-- `tests/conftest.py`: Sets env vars BEFORE imports, creates temp SQLite file, builds all tables with `Base.metadata.create_all()`, tears down with `drop_all()` + unlink.
-- External integrations are monkeypatched in tests (e.g., WeChat Pay, Jiguang, Pangle callbacks) — tests never make real HTTP calls.
-- `TestClient` from FastAPI is used for all tests. Rate limiting is disabled globally in tests.
-
-## Key integration details
-
-- **Jiguang one-click login**: REST call to verify `loginToken`, then RSA decrypt the returned phone number. Multiple padding schemes tried in order (PKCS1v15, OAEP with SHA1/SHA256) because Jiguang's encryption padding varies.
-- **WeChat Pay withdrawals**: V3 API merchant transfer to user WeChat balance. Lazy-loads merchant certificates from `secrets/`. Withdrawal flow: bind WeChat → create withdraw order → auto-reconcile worker polls pending orders.
-- **Pangle ad rewards**: S2S callback verification via SHA256 signature. Multiple `m-key` secrets supported (one per ad placement). Callback is idempotent by `trans_id`. Test grant endpoint (`AD_REWARD_TEST_GRANT_ENABLED`) for local debugging — must be false in prod.
-- **Meituan CPS**: Gateway signature-based API calls. Proxy support (`MT_CPS_PROXY`) for local dev (direct connection causes SSL EOF). Coupon endpoints gracefully return empty when credentials are missing.
-- **Pricebot forwarding**: `/api/v1/coupon/step` and `/api/v1/compare/*` proxy to pricebot-backend. Multi-instance support with consistent-hash routing by `trace_id` (see `core/pricebot_router.py`).
-- **CPS redirect**: `/c/{code}` is a public (no auth) short-link redirect — records a click then 302s to Meituan. Click recording failure never blocks the redirect.
-
-## Money and units
-
-All monetary amounts are in **cents** (`*_cents` fields). Coins/gold have their own unit. Conversion constants are in `core/rewards.py`.
-
-## Scripts
-
-Key operational scripts in `scripts/`:
-- `migrate.sh` — run migrations standalone
-- `create_admin.py` — create admin user
-- `daily_auto_exchange.py` — auto-convert coins to cash (triggered by systemd timer)
-- `reconcile_withdraws.py` — reconcile withdrawal orders with WeChat Pay
-- `sim_pangle_callback.py` — simulate Pangle S2S callback for testing

alembic/versions/1a924c274fce_merge_direct_vendor_push_and_feedback_.py

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

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

@@ -1,68 +0,0 @@
-"""notification table (消息通知中心 站内消息)
-
-Revision ID: notification_table
-Revises: 1a924c274fce
-Create Date: 2026-07-15 12:00:00.000000
-
-"""
-from typing import Sequence, Union
-
-from alembic import op
-import sqlalchemy as sa
-from sqlalchemy.dialects import postgresql
-
-
-# revision identifiers, used by Alembic.
-revision: str = 'notification_table'
-down_revision: Union[str, Sequence[str], None] = '1a924c274fce'
-branch_labels: Union[str, Sequence[str], None] = None
-depends_on: Union[str, Sequence[str], None] = None
-
-# PG 用 JSONB,SQLite 退化为通用 JSON(与 models/notification._JSON 一致)。
-_JSON = sa.JSON().with_variant(postgresql.JSONB(), 'postgresql')
-
-
-def upgrade() -> None:
-    op.create_table(
-        'notification',
-        sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
-        sa.Column('user_id', sa.Integer(), nullable=False),
-        sa.Column('type', sa.String(length=32), nullable=False),
-        sa.Column('coins', sa.Integer(), nullable=True),
-        sa.Column('cash_cents', sa.Integer(), nullable=True),
-        sa.Column('info_rows', _JSON, nullable=False),
-        sa.Column('extra', _JSON, nullable=False),
-        sa.Column('is_read', sa.Boolean(), nullable=False),
-        sa.Column('read_at', sa.DateTime(timezone=True), nullable=True),
-        sa.Column('dedup_key', sa.String(length=64), nullable=True),
-        sa.Column('sent_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False),
-        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.ForeignKeyConstraint(['user_id'], ['user.id'], ),
-        sa.PrimaryKeyConstraint('id'),
-    )
-    with op.batch_alter_table('notification', schema=None) as batch_op:
-        batch_op.create_index('ix_notification_type', ['type'], unique=False)
-        # 列表分页:按用户取 + sent_at 倒序
-        batch_op.create_index('ix_notification_user_sent', ['user_id', 'sent_at'], unique=False)
-        # 铃铛角标:count where user_id=? and is_read=false —— 部分索引只覆盖未读行
-        batch_op.create_index(
-            'ix_notification_user_unread', ['user_id'], unique=False,
-            sqlite_where=sa.text('is_read = 0'),
-            postgresql_where=sa.text('is_read = false'),
-        )
-        # 去重/合并:同一 (user, type, dedup_key) 未读期间只允许一条(已读后可再生成)
-        batch_op.create_index(
-            'uq_notification_user_type_dedup', ['user_id', 'type', 'dedup_key'], unique=True,
-            sqlite_where=sa.text('dedup_key IS NOT NULL AND is_read = 0'),
-            postgresql_where=sa.text('dedup_key IS NOT NULL AND is_read = false'),
-        )
-
-
-def downgrade() -> None:
-    with op.batch_alter_table('notification', schema=None) as batch_op:
-        batch_op.drop_index('uq_notification_user_type_dedup')
-        batch_op.drop_index('ix_notification_user_unread')
-        batch_op.drop_index('ix_notification_user_sent')
-        batch_op.drop_index('ix_notification_type')
-    op.drop_table('notification')

app/admin/routers/feedback.py

@@ -19,7 +19,6 @@ from app.admin.schemas.feedback import (
 from app.models.admin import AdminUser
 from app.models.feedback import Feedback
 from app.repositories import wallet as wallet_repo
-from app.services import notification_events
 
 router = APIRouter(
     prefix="/admin/api/feedbacks",
@@ -135,11 +134,7 @@ def approve_feedback(
     )
     db.commit()
     db.refresh(fb)
-    out = FeedbackOut.model_validate(fb)
-    # PRD #10 反馈奖励:采纳发金币后通知用户(站内 + push,必带官方留言)。
-    # 业务已 commit,通知失败只 log 不影响审核结果。
-    notification_events.notify_feedback_reward(db, fb)
-    return out
+    return FeedbackOut.model_validate(fb)
 
 
 @router.post("/{feedback_id}/reject", response_model=FeedbackOut, summary="拒绝采纳反馈")
@@ -184,7 +179,4 @@ def reject_feedback(
     )
     db.commit()
     db.refresh(fb)
-    out = FeedbackOut.model_validate(fb)
-    # PRD #9 官方回复:未采纳也回复了用户(原因/留言用户端可见),通知去反馈历史页查看。
-    notification_events.notify_feedback_reply(db, fb)
-    return out
+    return FeedbackOut.model_validate(fb)

app/admin/routers/price_report.py

@@ -3,8 +3,7 @@
 数据由客户端 POST /api/v1/report 写入 price_report 表(提交即 pending);本路由是运营后台
 对它的人工审核窗口。**通过** → 给上报用户钱包发固定金币(PRICE_REPORT_REWARD_COINS):
 改状态 + 发金币(wallet.grant_coins)+ 审计同一事务一起 commit(原子,仿 users.grant_user_coins),
-绝不只改状态不发钱或反之。通过后下发「爆料审核通过」通知(站内 + push,PRD #11);
-客户端也可轮询 GET /api/v1/report/records 看到结果。
+绝不只改状态不发钱或反之。客户端轮询 GET /api/v1/report/records 自动看到结果(无需推送)。
 """
 from __future__ import annotations
 
@@ -25,7 +24,6 @@ from app.core.rewards import PRICE_REPORT_REWARD_COINS
 from app.models.admin import AdminUser
 from app.models.price_report import PriceReport
 from app.repositories import wallet as wallet_repo
-from app.services import notification_events
 
 router = APIRouter(
     prefix="/admin/api/price-reports",
@@ -86,8 +84,6 @@ def approve_price_report(
         detail={"reward_coins": coins, "user_id": rep.user_id}, ip=get_client_ip(request), commit=False,
     )
     db.commit()
-    # PRD #11 爆料审核通过:发金币后通知用户(站内 + push)。业务已 commit,通知失败只 log。
-    notification_events.notify_report_approved(db, rep)
     return OkResponse()
 
 

app/admin/schemas/device.py

@@ -24,9 +24,7 @@ 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  # 旧极光字段,仅兼容历史数据
-    push_vendor: str | None = None
-    push_token: str | None = None
+    registration_id: str | None = None  # 非空 = 拿到极光 token、可推送
 
     ever_protected: bool                 # 是否开过无障碍(=该设备对功能有意义)
     first_protected_at: datetime | None = None   # 首次开无障碍时刻(老设备为 null)

app/api/v1/device.py

@@ -1,22 +1,19 @@
 """设备注册 / 心跳 endpoint(无障碍保护存活检测)。
 
 路由前缀 /api/v1/device,需 Bearer 鉴权(设备绑登录用户)。
-  POST /register   注册设备 / 更新厂商 push token(App 前台、拿到 push token 时调)
+  POST /register   注册设备 / 更新 registration_id(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, BackgroundTasks, HTTPException, status
+from fastapi import APIRouter
 
 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,
@@ -25,8 +22,6 @@ from app.schemas.device import (
     LivenessAckRequest,
     LivenessOut,
     OkResponse,
-    PushTestOut,
-    PushTestRequest,
 )
 
 logger = logging.getLogger("shagua.device")
@@ -34,37 +29,6 @@ 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,
@@ -76,17 +40,13 @@ 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 vendor=%s token=%s legacy_reg=%s",
+        "device register user_id=%d device_id=%s reg=%s",
         user.id,
         req.device_id,
-        req.push_vendor,
-        bool(req.push_token),
         bool(req.registration_id),
     )
     return DeviceOut.model_validate(device)
@@ -104,59 +64,10 @@ 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/notifications.py

@@ -1,124 +0,0 @@
-"""消息通知中心 endpoint(PRD《消息通知中心》)。
-
-路由前缀 `/api/v1/notifications`,需 Bearer 鉴权(消息按用户隔离)。
-  GET  /              消息列表(分页;全列表时间倒序,不分组——PRD 原文的分组已取消)
-  GET  /unread-count  未读总数(首页铃铛角标)
-  POST /read          标记已读({ids:[...]} 单条/多条 或 {all:true} 全量清零)
-
-数据落库 `notification` 表(repositories/notification.py,按用户隔离)。业务事件(奖励过期、
-提现回执、反馈回复……)调 `create_notification` 下发;未接入业务前列表为空,可用
-`/api/v1/push/test` 的 createNotification 造联调数据。
-
-⚠️ 字段命名:本组接口对外为 **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, DbSession
-from app.core import notification_catalog as catalog
-from app.models.notification import Notification
-from app.repositories import notification as notif_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: Notification) -> NotificationItem:
-    """通知行 + 类型静态目录 → 接口出参。"""
-    ntype = catalog.get_type(n.type)
-    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=notif_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=notif_repo.as_cst(n.sent_at),
-        is_read=n.is_read,
-    )
-
-
-@router.get("", response_model=NotificationListOut, summary="消息列表(分页)")
-def list_notifications(
-    user: CurrentUser,
-    db: DbSession,
-    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,进页面时可顺手刷新角标。
-    - 无消息时返回空列表(total=0);数据由业务事件下发,联调可用 /push/test 造。
-    """
-    items, total, unread = notif_repo.list_notifications(
-        db, 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, db: DbSession) -> UnreadCountOut:
-    """首页铃铛角标数据源。刷新时机(PRD §4):进入首页时、从通知中心/其他页面返回首页时。
-
-    - count:精确未读条数;
-    - badgeText:直接可展示的角标文案——超过 99 返回 "99+",等于 0 返回 null(隐藏整个角标)。
-    """
-    count = notif_repo.unread_count(db, 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, db: DbSession) -> 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 = notif_repo.mark_read(db, 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

@@ -1,187 +0,0 @@
-"""厂商推送 测试/联调 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 的文案/参数/厂商通道联调。真实业务触发统一走 services/notification_events
-(提现回执/反馈审核/爆料通过/好友下单已接入),底层与本测试端点同一条
-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 as notif_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**:同时往消息中心(notification 表)插一条同类型未读通知并把
-      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 = notif_repo.insert_sample(db, 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

@@ -71,59 +71,7 @@ class Settings(BaseSettings):
     JG_VERIFY_ENDPOINT: str = "https://api.verification.jpush.cn/v1/web/loginTokenVerify"
     JG_REQUEST_TIMEOUT_SEC: int = 15
 
-    # ===== 厂商直推(无障碍保护存活告警)=====
-    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 后置兜底)
+    # 无障碍保护存活监控后台任务(pull 后置检测;本期不接推送)
     HEARTBEAT_MONITOR_ENABLED: bool = True       # 总开关
     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,7 +22,6 @@ 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")
@@ -72,66 +71,32 @@ 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()  # noqa: UP017
+    ref = datetime.now(timezone.utc) if last.tzinfo is not None else datetime.utcnow()
     return int((ref - last).total_seconds())
 
 
 def _scan_once(timeout_minutes: int) -> dict:
-    """扫描一轮:找出心跳超时(App 被彻底杀掉/无障碍已停)的设备并召回。
+    """扫描一轮:找出心跳超时(App 被彻底杀掉/无障碍已停)的设备,在**服务器终端打印**告警代替真实推送。
 
-    有 push_vendor + push_token 时先发厂商直推,无 token 或推送失败时仍置
-    kill_alert_pending,客户端下次进 App 继续走后置提醒兜底。
+    本期不接推送(极光/厂商通道工程量大),只做服务端掉线检测:命中即 logger.warning 打印到终端,
+    并把状态机推进到 notified 防每轮重复打印(心跳恢复时 touch_heartbeat 会重置回 alive)。
     """
     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 已被杀/无障碍已停。",
+                " → 判定 App 已被杀/无障碍已停。【已置 kill_alert_pending: 用户下次进 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,
-            "pushed": pushed,
-            "push_failed": push_failed,
-        }
+        return {"checked": len(overdue), "notified": notified}
 
 
 async def _run_loop() -> None:

app/core/notification_catalog.py

@@ -1,241 +0,0 @@
-"""消息通知中心:13 种通知类型的静态目录 + Push 文案模板。
-
-对应 PRD《消息通知中心》:§1 类型清单 / §3 字段元素 / §5 Push 文案。
-这里只放**静态定义**(分类、版式、标题、操作行、Push 模板),供两处消费:
-  - repositories/notification.py       消息中心列表按 type 派生分类/版式/标题/操作行
-  - 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

@@ -1,584 +0,0 @@
-"""厂商直推集成(荣耀 / 华为 / 小米 / 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
-from urllib.parse import quote
-
-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 = {
-        # clickAction type=3(打开应用首页)时,荣耀点击会把 data JSON 的键值对注入启动 intent 的
-        # extras(与 HMS 同机制)→ MainActivity.consumeNavTarget 读 notif_id/notif_type 直达落地。
-        "data": json.dumps(_click_extras(extras), ensure_ascii=False),
-        "notification": {"title": title, "body": body},
-        "android": {
-            "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": {
-            # click_action type=3(打开应用首页)时,HMS 点击会把 data JSON 的键值对注入启动 intent
-            # 的 extras → MainActivity.consumeNavTarget 读 notif_id/notif_type 直达落地。
-            "data": json.dumps(_click_extras(extras), ensure_ascii=False),
-            "android": {
-                "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,
-        "requestId": uuid.uuid4().hex,
-        "pushMode": settings.VIVO_PUSH_MODE,
-        "clientCustomMap": extras,
-    }
-    # 点击落地:消息中心推送(带 notificationId)→ skipType=4 + skipContent=intent uri,由 vivo
-    # 系统直启 MainActivity 并携带 S. extras(与小米 notify_effect=2 同机制)。不依赖客户端
-    # VivoPushReceiver.onNotificationMessageClicked 里的后台 startActivity——Android 10+ BAL
-    # 会静默拦掉,receiver 路径仅作兜底。无 notificationId 的召回类保持 skipType=1 仅打开首页。
-    if extras.get("notificationId"):
-        payload["skipType"] = 4
-        payload["skipContent"] = _click_intent_uri(extras)
-    else:
-        payload["skipType"] = 1
-    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),
-    }
-    # 点击落地:带 notificationId 的消息中心推送 → notify_effect=2 + intent_uri,MiPush 直接打开
-    # MainActivity 并把 extras 作为 String extra 传入(客户端 MainActivity.consumeNavTarget 读
-    # notif_id/notif_type,兜底 notificationId/type)→ 置读 + 刷角标 + 按 type 直达对应页(PRD §5)。
-    # ⚠️ 早前用 notify_effect=1(仅打开 Launcher),小米自身不会把 payload 拆成普通 extra、而是塞进
-    #    序列化的 MiPushMessage(key_message),客户端读不到 → 点击后停在首页「没反应」。
-    # 无 notificationId 的系统召回类(如无障碍掉线)保持 notify_effect=1 仅拉起 App,行为不变。
-    if extras.get("notificationId"):
-        form["extra.notify_effect"] = "2"
-        form["extra.intent_uri"] = _click_intent_uri(extras)
-    else:
-        form["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 _click_extras(extras: dict[str, str]) -> dict[str, str]:
-    """点击落地参数:消息中心推送(extras 带 notificationId)补 notif_id/notif_type 别名——
-    客户端 MainActivity.consumeNavTarget 首选这两个键(厂商 receiver 路径的历史约定),原始键
-    (notificationId/type/feedbackId/reportId/…)保留作兜底与业务跳转参数。
-    无 notificationId(如无障碍召回)原样返回,不喂点击路由参数。"""
-    if not extras.get("notificationId"):
-        return dict(extras)
-    merged = dict(extras)
-    merged.setdefault("notif_id", extras["notificationId"])
-    if extras.get("type"):
-        merged.setdefault("notif_type", extras["type"])
-    return merged
-
-
-def _click_intent_uri(extras: dict[str, str]) -> str:
-    """构造「系统直启 MainActivity 并带 extras」的 intent uri(小米 notify_effect=2 的
-    extra.intent_uri、vivo skipType=4 的 skipContent 共用):点击后厂商系统用 Intent.parseUri
-    解析并 startActivity,extras 作为 String extra 原样送达。
-
-    - component 显式指向本包 MainActivity(exported=true、singleTask)→ 已运行则走 onNewIntent、
-      未运行则 onCreate,两条都会执行 consumeNavTarget。
-    - 参数 = _click_extras(补 notif_id/notif_type 别名 + 透传 feedbackId/reportId 等跳转参数)。
-    - 值按 Android Uri.encode 规则百分号编码(quote(safe="")):中文/分号/等号都不会破坏 intent uri
-      结构;客户端 Intent.parseUri 侧 Uri.decode 无损还原。表单/JSON 传输层的编码与本层相互独立、
-      各自解码,不会双重转义(2026-07-15 小米联调结论)。
-    """
-    pkg = settings.ANDROID_PACKAGE_NAME
-    parts = ["intent:#Intent", f"component={pkg}/{pkg}.MainActivity"]
-    parts += [f"S.{key}={quote(str(value), safe='')}" for key, value in _click_extras(extras).items()]
-    parts.append("end")
-    return ";".join(parts)
-
-
-def _xiaomi_template_param(title: str, alert: str) -> str:
-    rendered = (
-        settings.XIAOMI_PUSH_TEMPLATE_PARAM_JSON
-        .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,
-        "off_line": True,
-        "off_line_ttl": ttl_hours,
-        "action_parameters": json.dumps(_click_extras(extras), ensure_ascii=False),
-    }
-    # 点击落地:OPPO SDK 没有点击回调,参数只能靠服务端点击动作配置送达——action_parameters 的
-    # 键值对仅在 click_action_type=1/4 时才会注入目标 Activity 的 intent extras(type=0「启动应用」
-    # 会忽略它,extras 全丢 → 点了没反应,与小米 notify_effect=1 同款坑)。
-    # 消息中心推送(带 notificationId)→ type=4(打开应用内页面,Activity 全路径,exported=true);
-    # 无 notificationId 的召回类保持 type=0 仅打开应用。
-    if extras.get("notificationId"):
-        notification["click_action_type"] = 4
-        notification["click_action_activity"] = f"{settings.ANDROID_PACKAGE_NAME}.MainActivity"
-    else:
-        notification["click_action_type"] = 0
-    # 新消息分类(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,10 +31,8 @@ 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
@@ -150,10 +148,6 @@ app.include_router(savings_router)
 app.include_router(ad_router)
 app.include_router(order_router)
 app.include_router(report_router)
-# 消息通知中心(PRD;数据落库 notification 表,见 repositories/notification.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

@@ -35,7 +35,6 @@ from app.models.invite import InviteRelation  # noqa: F401
 from app.models.invite_fingerprint import InviteFingerprint  # noqa: F401
 from app.models.launch_confirm_sample import LaunchConfirmSample  # noqa: F401
 from app.models.meituan_coupon import MeituanCoupon  # noqa: F401
-from app.models.notification import Notification  # noqa: F401
 from app.models.onboarding import OnboardingCompletion  # noqa: F401
 from app.models.phone_rebind_log import PhoneRebindLog  # noqa: F401
 from app.models.ops_marquee_seed import OpsMarqueeSeed  # noqa: F401

app/models/device.py

@@ -1,9 +1,9 @@
-"""设备表(无障碍保护存活检测 + 厂商直推)。
+"""设备表(无障碍保护存活检测 + 极光推送)。
 
 每条 = 一个用户的一台设备(per-install,device_id 由客户端 DeviceId.get() 生成)。
 客户端的无障碍服务存活时周期上报心跳刷新 last_heartbeat_at;App 前台/登录时上报
-push_vendor + push_token(厂商推送目标)。后端 heartbeat_monitor_worker 扫描「曾经保护过、
-现在心跳超时」的设备,通过厂商直推提醒用户重开无障碍。
+registration_id(极光推送目标)。后端 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 + 厂商推送目标),故名 device_liveness。
+    # (心跳 last_heartbeat_at + liveness_state + kill_alert_pending + 推送目标 registration_id),故名 device_liveness。
     __tablename__ = "device_liveness"
     __table_args__ = (
         UniqueConstraint("user_id", "device_id", name="uq_device_liveness_user_device"),
@@ -42,12 +42,8 @@ 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,仅为兼容历史客户端/数据保留;新链路使用 push_vendor + push_token。
+    # 极光推送 registration id;拿到才填(JCollectionAuth 同意后才下发)
     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/notification.py

@@ -1,95 +0,0 @@
-"""消息通知中心:站内消息表(一行 = 一条下发给某用户的站内消息)。
-
-13 类通知的**静态定义**(分类 / 版式 / 标题 / 操作行 / push 模板)在
-`app/core/notification_catalog.py`,是代码常量,**不入库**;本表只存**每条消息的动态部分**
-(与接口 NotificationItem 的动态字段一一对应):type + 金额 + 信息行 + extra + 已读态 + 时间。
-category / card_style / title / action_text 都由 `type` 经 catalog 派生,不冗余存库。
-
-- 写:`repositories/notification.create_notification`(业务事件下发站内消息的统一入口)。
-- 读:`api/v1/notifications.py`(列表 / 未读数 / 标记已读),均按 user 隔离、sent_at 倒序。
-"""
-from __future__ import annotations
-
-from datetime import datetime
-
-from sqlalchemy import (
-    JSON,
-    Boolean,
-    DateTime,
-    ForeignKey,
-    Index,
-    Integer,
-    String,
-    func,
-    text,
-)
-from sqlalchemy.dialects.postgresql import JSONB
-from sqlalchemy.orm import Mapped, mapped_column
-
-from app.db.base import Base
-
-# PG 用 JSONB,SQLite(本地/测试)退化为通用 JSON(同 comparison_record.raw_payload 等)。
-_JSON = JSON().with_variant(JSONB(), "postgresql")
-
-
-class Notification(Base):
-    __tablename__ = "notification"
-    __table_args__ = (
-        # 列表分页:按用户取 + sent_at 倒序(核心查询,覆盖 user_id 前缀查找,故不再单独索引 user_id)
-        Index("ix_notification_user_sent", "user_id", "sent_at"),
-        # 铃铛角标:count where user_id=? and is_read=false —— 部分索引只覆盖未读行
-        Index(
-            "ix_notification_user_unread",
-            "user_id",
-            sqlite_where=text("is_read = 0"),
-            postgresql_where=text("is_read = false"),
-        ),
-        # 去重/合并:同一 (user, type, dedup_key) 未读期间只允许一条(perm_* 权限异常、
-        # reward_expiring 同批次即用它);消息一旦已读即离开索引,之后可再生成新的未读消息。
-        Index(
-            "uq_notification_user_type_dedup",
-            "user_id",
-            "type",
-            "dedup_key",
-            unique=True,
-            sqlite_where=text("dedup_key IS NOT NULL AND is_read = 0"),
-            postgresql_where=text("dedup_key IS NOT NULL AND is_read = false"),
-        ),
-    )
-
-    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
-    user_id: Mapped[int] = mapped_column(Integer, ForeignKey("user.id"), nullable=False)
-    # 13 类之一(catalog.TYPES 的 key);category/card_style/title/action_text 由它派生,不入库
-    type: Mapped[str] = mapped_column(String(32), nullable=False, index=True)
-    # 金币数(dual_amount / coin_reward 卡);其余类型 None
-    coins: Mapped[int | None] = mapped_column(Integer, nullable=True)
-    # 现金,单位【分】(dual_amount / withdraw / friend_cash 卡);其余 None
-    cash_cents: Mapped[int | None] = mapped_column(Integer, nullable=True)
-    # 信息行 [{label, value}](已渲染好文案,前端逐行展示)
-    info_rows: Mapped[list] = mapped_column(_JSON, nullable=False, default=list)
-    # 点击跳转/联动参数(feedbackId / withdrawId / permission / inviteeNickname / batchId …)
-    extra: Mapped[dict] = mapped_column(_JSON, nullable=False, default=dict)
-    is_read: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
-    # 置读时刻(未读时为 None;埋点/分析用)
-    read_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
-    # 去重键(可空):perm_*→permission、reward_expiring→batchId 等;配合部分唯一索引防重复未读
-    dedup_key: Mapped[str | None] = mapped_column(String(64), nullable=True)
-    # 下发/业务时间;列表排序与展示都用它(带 +08:00 下发)
-    sent_at: Mapped[datetime] = mapped_column(
-        DateTime(timezone=True), server_default=func.now(), nullable=False
-    )
-    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"<Notification id={self.id} user_id={self.user_id} "
-            f"type={self.type} read={self.is_read}>"
-        )

app/repositories/device.py

@@ -21,23 +21,17 @@ def register_or_update(
     *,
     user_id: int,
     device_id: str,
-    registration_id: str | None = None,
-    push_vendor: str | None = None,
-    push_token: str | None = None,
+    registration_id: str | None,
     platform: str = "android",
     app_version: str | None = None,
 ) -> DeviceLiveness:
-    """注册设备或更新其厂商 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
+    """注册设备或更新其 registration_id / 元信息。upsert by (user_id, device_id)。"""
     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,
         )
@@ -45,10 +39,6 @@ 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:
@@ -64,9 +54,7 @@ def touch_heartbeat(
     user_id: int,
     device_id: str,
     accessibility_enabled: bool,
-    registration_id: str | None = None,
-    push_vendor: str | None = None,
-    push_token: str | None = None,
+    registration_id: str | None,
 ) -> DeviceLiveness:
     """处理一次心跳(心跳也能自注册)。
 
@@ -81,12 +69,6 @@ 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:
@@ -105,7 +87,7 @@ def touch_heartbeat(
 def list_overdue(db: Session, *, timeout_minutes: int) -> list[DeviceLiveness]:
     """掉线设备:曾经保护过、当前 alive、心跳超时。
 
-    即使没有厂商 token 也要检出,后续由 kill_alert_pending 走客户端进 App 后兜底提醒。
+    本期只做终端打印检测、不推送 → 不再要求有 registration_id(没接极光 token 的设备也要检出)。
     """
     cutoff = datetime.now(timezone.utc) - timedelta(minutes=timeout_minutes)
     stmt = select(DeviceLiveness).where(
@@ -142,55 +124,3 @@ 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 list_push_targets(db: Session, *, user_id: int) -> list[DeviceLiveness]:
-    """该用户全部可用厂商推送目标(push_vendor + push_token 双非空),最近更新在前。
-
-    同 (vendor, token) 只留最新一行:同一台手机重装 App 后 device_id 会变、
-    留下 token 相同的旧行,去重防一次业务事件对同一台手机重复推送。
-    """
-    stmt = (
-        select(DeviceLiveness)
-        .where(
-            DeviceLiveness.user_id == user_id,
-            DeviceLiveness.push_vendor.is_not(None),
-            DeviceLiveness.push_token.is_not(None),
-        )
-        .order_by(DeviceLiveness.updated_at.desc(), DeviceLiveness.id.desc())
-    )
-    seen: set[tuple[str, str]] = set()
-    targets: list[DeviceLiveness] = []
-    for dev in db.execute(stmt).scalars():
-        if not dev.push_vendor or not dev.push_token:  # 空串兜底(旧数据)
-            continue
-        key = (dev.push_vendor, dev.push_token)
-        if key in seen:
-            continue
-        seen.add(key)
-        targets.append(dev)
-    return targets
-
-
-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/invite.py

@@ -25,7 +25,6 @@ from app.models.invite import InviteRelation
 from app.models.invite_fingerprint import InviteFingerprint
 from app.models.user import User
 from app.repositories import wallet as crud_wallet
-from app.services import notification_events
 
 # 邀请码字符集:去掉易混字符(0/O/1/I/L/B/8/S/5/Z/2),用户口述/手输不易错
 _CODE_ALPHABET = "ACDEFGHJKMNPQRTUVWXY34679"
@@ -198,13 +197,12 @@ def try_reward_on_compare(db: Session, invitee_user_id: int) -> CompareRewardRes
         return CompareRewardResult("inviter_inactive", rel.inviter_user_id)
 
     reward = rewards.INVITE_COMPARE_REWARD_CENTS
-    inviter_id = inviter.id
     rel.compare_reward_granted = True
     rel.compare_reward_cents = reward
     rel.compare_rewarded_at = datetime.now(timezone.utc)
     # 发邀请奖励金到邀请人的独立账户(与金币隔离),ref_id 指向被邀请人便于对账
     crud_wallet.grant_invite_cash(
-        db, inviter_id, reward,
+        db, inviter.id, reward,
         biz_type="invite_reward", ref_id=str(invitee_user_id), remark="好友比价奖励",
     )
     try:
@@ -212,11 +210,7 @@ def try_reward_on_compare(db: Session, invitee_user_id: int) -> CompareRewardRes
     except Exception:
         db.rollback()
         raise
-    # PRD #12 好友下单到账:发奖已 commit,通知邀请人(站内 + push;失败只 log 不影响发奖)
-    notification_events.notify_invite_order_reward(
-        db, inviter_user_id=inviter_id, invitee_user_id=invitee_user_id, cash_cents=reward
-    )
-    return CompareRewardResult("granted", inviter_id, reward)
+    return CompareRewardResult("granted", inviter.id, reward)
 
 
 def get_stats(db: Session, inviter_id: int) -> tuple[int, int]:

app/repositories/notification.py

@@ -1,318 +0,0 @@
-"""消息通知中心 数据仓库(落库版,查/写 `notification` 表)。
-
-沿用原 notification_mock 的同名函数(list_notifications / unread_count / mark_read /
-insert_sample),由内存 mock 迁到落库,**API 契约不变**。
-
-- 读:按 user 隔离、sent_at 倒序;未读数 / 标记已读同口径。
-- 写:`create_notification` 是落库统一入口。**业务事件请走 services/notification_events**
-  (站内消息 + 厂商 push 一起下发,已接入提现回执/反馈审核/爆料通过/好友下单);
-  `build_sample_card` / `insert_sample` 按类型造样例内容,供
-  `/api/v1/push/test` 的 createNotification 做「push → 站内已读联动」联调。
-
-排序规则:全列表按 sent_at 倒序(最新在前;同秒再按 id 倒序稳定化),不分组。
-"""
-from __future__ import annotations
-
-from datetime import datetime, timedelta, timezone
-
-from sqlalchemy import func, select
-from sqlalchemy.orm import Session
-
-from app.core import notification_catalog as catalog
-from app.models.notification import Notification
-
-# 北京时间:sent_at 统一带 +08:00 下发,前端直接按本地时区渲染「今天/昨天/M月D日」。
-_CST = timezone(timedelta(hours=8))
-
-
-def cash_yuan(cents: int | None) -> str | None:
-    """分 → 保留两位小数的元字符串(PRD §3:现金/提现金额保留两位小数)。"""
-    if cents is None:
-        return None
-    return f"{cents // 100}.{cents % 100:02d}"
-
-
-def as_cst(dt: datetime) -> datetime:
-    """把库里取出的时间归一到北京时间(+08:00)再下发,保证接口 sentAt 恒带 +08:00。
-
-    SQLite 的 DateTime 不存时区,取出为 naive(存的就是写入时的 CST 墙上时间)→ 直接贴 +08:00;
-    PostgreSQL 的 timestamptz 取出为 aware(通常 UTC)→ 转到 +08:00。两端下发口径一致。
-    """
-    if dt.tzinfo is None:
-        return dt.replace(tzinfo=_CST)
-    return dt.astimezone(_CST)
-
-
-def _fmt_time(dt: datetime) -> str:
-    """信息行里「到账时间」等 value 的展示格式。"""
-    return dt.strftime("%Y-%m-%d %H:%M")
-
-
-# ---------------------------------------------------------------------------
-# 读:列表 / 未读数 / 标记已读
-# ---------------------------------------------------------------------------
-
-
-def _unread_count(db: Session, user_id: int) -> int:
-    return int(
-        db.execute(
-            select(func.count())
-            .select_from(Notification)
-            .where(Notification.user_id == user_id, Notification.is_read.is_(False))
-        ).scalar_one()
-    )
-
-
-def list_notifications(
-    db: Session, user_id: int, *, page: int, page_size: int
-) -> tuple[list[Notification], int, int]:
-    """分页取通知列表。返回 (当前页条目, 总条数, 未读条数)。"""
-    total = int(
-        db.execute(
-            select(func.count())
-            .select_from(Notification)
-            .where(Notification.user_id == user_id)
-        ).scalar_one()
-    )
-    unread = _unread_count(db, user_id)
-    rows = (
-        db.execute(
-            select(Notification)
-            .where(Notification.user_id == user_id)
-            .order_by(Notification.sent_at.desc(), Notification.id.desc())
-            .offset((page - 1) * page_size)
-            .limit(page_size)
-        )
-        .scalars()
-        .all()
-    )
-    return list(rows), total, unread
-
-
-def unread_count(db: Session, user_id: int) -> int:
-    """未读总数(首页铃铛角标)。"""
-    return _unread_count(db, user_id)
-
-
-def mark_read(
-    db: Session, user_id: int, *, ids: list[int] | None = None, mark_all: bool = False
-) -> tuple[int, int]:
-    """标记已读。mark_all=True 全量清零,否则按 ids 逐条置读(不存在的 id 忽略,幂等)。
-
-    返回 (本次实际由未读→已读的条数, 剩余未读数)。
-    """
-    if not mark_all:
-        wanted = set(ids or [])
-        if not wanted:
-            return 0, _unread_count(db, user_id)
-
-    stmt = select(Notification).where(
-        Notification.user_id == user_id, Notification.is_read.is_(False)
-    )
-    if not mark_all:
-        stmt = stmt.where(Notification.id.in_(wanted))
-
-    now = datetime.now(timezone.utc)
-    marked = 0
-    for n in db.execute(stmt).scalars().all():
-        n.is_read = True
-        n.read_at = now
-        marked += 1
-    db.commit()
-    return marked, _unread_count(db, user_id)
-
-
-# ---------------------------------------------------------------------------
-# 写:业务下发入口
-# ---------------------------------------------------------------------------
-
-
-def create_notification(
-    db: Session,
-    *,
-    user_id: int,
-    type_key: str,
-    coins: int | None = None,
-    cash_cents: int | None = None,
-    info_rows: list[dict[str, str]] | None = None,
-    extra: dict[str, str] | None = None,
-    sent_at: datetime | None = None,
-    dedup_key: str | None = None,
-) -> Notification:
-    """下发一条站内消息(业务事件统一入口)。type_key 必须是 catalog 的 13 类之一。
-
-    dedup_key 非空时受部分唯一索引约束(同 user+type+dedup_key 未读期间仅一条);
-    需要「同批次/同权限只保留一条未读」的调用方,应捕获 IntegrityError 或先查已存在的未读再决定
-    更新 sent_at,而非重复插入(见 models/notification 的 uq_notification_user_type_dedup)。
-    """
-    catalog.get_type(type_key)  # 校验类型合法(未知类型抛 UnknownNotificationType)
-    row = Notification(
-        user_id=user_id,
-        type=type_key,
-        coins=coins,
-        cash_cents=cash_cents,
-        info_rows=info_rows or [],
-        extra=extra or {},
-        sent_at=sent_at or datetime.now(_CST),
-        dedup_key=dedup_key,
-    )
-    db.add(row)
-    db.commit()
-    db.refresh(row)
-    return row
-
-
-# ---------------------------------------------------------------------------
-# 样例内容(供 /push/test createNotification 联调;文案对齐 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_card(type_key: str, sent_at: datetime | None = None) -> dict:
-    """按类型生成一份样例卡片内容({coins?, cash_cents?, info_rows, extra}),/push/test 联调用。"""
-    catalog.get_type(type_key)  # 校验 type 合法
-    now = sent_at or datetime.now(_CST)
-    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 builders[type_key]()
-
-
-def insert_sample(db: Session, user_id: int, type_key: str) -> Notification:
-    """插入一条该类型的样例未读通知并落库(/push/test createNotification 联调:push extras 带上
-    它的 id,客户端点击 push 后调 POST /notifications/read {ids:[id]} 即闭环验证已读联动)。"""
-    return create_notification(db, user_id=user_id, type_key=type_key, **build_sample_card(type_key))

app/repositories/wallet.py

@@ -29,7 +29,6 @@ from app.models.wallet import (
     WechatTransferAuthorization,
     WithdrawOrder,
 )
-from app.services import notification_events
 
 # 微信转账终态:成功 / 失败(失败/取消/关闭都退款)
 _WX_STATE_SUCCESS = "SUCCESS"
@@ -526,8 +525,6 @@ def _refund_withdraw(
         order.status = final_status
         order.fail_reason = reason[:256]
         db.commit()
-        # 上次退款后没走完终态(如中途崩溃)的补账路径:这里补发通知(dedup 防重)
-        notification_events.notify_withdraw_failed(db, order)
         return
     bal = _add_cash(db, order.user_id, order.amount_cents, order.source)
     db.add(
@@ -571,11 +568,6 @@ def _refund_withdraw(
             fresh_order.status = final_status
             fresh_order.fail_reason = reason[:256]
             db.commit()
-            notification_events.notify_withdraw_failed(db, fresh_order)
-        return
-    # PRD #4 提现失败通知:所有退款终态(failed/rejected)在此收口下发;
-    # dedup=out_bill_no,与上面并发路径重复触发时未读期间只落一条。
-    notification_events.notify_withdraw_failed(db, order)
 
 
 def _wx_not_found(result: dict) -> bool:
@@ -616,7 +608,6 @@ def _settle_after_ambiguous(db: Session, order: WithdrawOrder, reason: str) -> N
         order.status = "success"
         order.transfer_bill_no = q["data"].get("transfer_bill_no")
         db.commit()
-        notification_events.notify_withdraw_success(db, order)  # PRD #3 提现到账
     elif state in _WX_STATE_FAILED:
         _refund_withdraw(db, order, reason=reason)
     else:
@@ -990,8 +981,6 @@ def _apply_transfer_result(db: Session, order: WithdrawOrder, data: dict) -> Wit
         order.status = "success"
     db.commit()
     db.refresh(order)
-    if order.status == "success":  # 免确认转账直接到账 → PRD #3 提现到账
-        notification_events.notify_withdraw_success(db, order)
     return order
 
 
@@ -1141,7 +1130,6 @@ def refresh_withdraw_status(
     if state == _WX_STATE_SUCCESS:
         order.status = "success"
         db.commit()
-        notification_events.notify_withdraw_success(db, order)  # PRD #3 提现到账
     elif state in _WX_STATE_FAILED:
         _refund_withdraw(db, order, reason=f"微信转账状态 {state}")
     elif state == _WX_STATE_WAIT_CONFIRM and cancel_if_unconfirmed:

app/schemas/device.py

@@ -3,15 +3,12 @@ from __future__ import annotations
 
 from datetime import datetime
 
-from pydantic import BaseModel, ConfigDict, Field
+from pydantic import BaseModel, ConfigDict
 
 
 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
 
@@ -21,8 +18,6 @@ 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):
@@ -31,8 +26,6 @@ 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
@@ -53,17 +46,3 @@ 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/notification.py

@@ -1,128 +0,0 @@
-"""消息通知中心 请求/响应契约。
-
-⚠️ 命名约定:本组接口按 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

@@ -1,99 +0,0 @@
-"""厂商推送(测试/联调)接口契约。与消息中心同族,出参统一 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)"
-    )

app/services/notification_events.py

@@ -1,269 +0,0 @@
-"""消息通知中心:业务事件 → 站内消息 + 厂商 push 的统一下发口。
-
-PRD《消息通知中心》真实业务触发在此收口(替代 /push/test 的样例数据),已接入:
-  #3  withdraw_success    提现到账(repositories/wallet 各「pending→success」转换点)
-  #4  withdraw_failed     提现失败/退回(repositories/wallet._refund_withdraw,含审核拒绝)
-  #9  feedback_reply      官方回复(admin 反馈审核「拒绝」,带用户可见原因/留言)
-  #10 feedback_reward     反馈奖励(admin 反馈审核「采纳」发金币,必带官方留言)
-  #11 report_approved     爆料审核通过(admin 上报更低价「通过」发金币)
-  #12 invite_order_reward 好友下单到账(repositories/invite.try_reward_on_compare 发奖后)
-
-行为约定(调用方唯一需要知道的两条):
-  1. **绝不抛异常**——通知只是业务的副产物,站内消息落库失败/推送失败只 log,
-     绝不让提现退款、审核发奖等主流程回滚或报错。
-  2. **必须在业务事务 commit 之后调用**——内部会再 commit(写 notification 表);
-     若在业务半途调用,会把调用方未提交的脏状态一并提交。
-
-去重:各事件用业务主键做 dedup_key(提现单号/反馈 id/爆料 id/被邀请人 id),配合
-notification 表的部分唯一索引,同一事件并发重复触发时未读期间只落一条、只推一次。
-
-推送:向该用户所有已上报厂商 token 的设备直推(integrations/vendor_push);
-厂商凭据未配置(本地/测试环境)时自动跳过推送、只落站内消息。extras 按
-PRD §4 约定带 {type, notificationId, ...跳转参数},客户端点击 push 深链落地
-并调 POST /notifications/read 同步置读。
-"""
-from __future__ import annotations
-
-import logging
-from datetime import datetime
-from typing import TYPE_CHECKING
-
-from sqlalchemy.exc import IntegrityError
-from sqlalchemy.orm import Session
-
-from app.core import notification_catalog as catalog
-from app.core.rewards import CN_TZ
-from app.integrations import vendor_push
-from app.models.user import User
-from app.repositories import device as device_repo
-from app.repositories import notification as notif_repo
-
-if TYPE_CHECKING:
-    from app.models.feedback import Feedback
-    from app.models.notification import Notification
-    from app.models.price_report import PriceReport
-    from app.models.wallet import WithdrawOrder
-
-logger = logging.getLogger("shagua.notification_events")
-
-
-def _fmt_time(dt: datetime) -> str:
-    """信息行「到账时间」的展示格式(与 repositories/notification 样例卡一致)。"""
-    return dt.strftime("%Y-%m-%d %H:%M")
-
-
-def _yuan_trim(cents: int) -> str:
-    """分 → 元,去掉多余的 0(200→"2"、1280→"12.80")。push 正文用(PRD §5 示例口径:
-    「{2}元现金已到账」);卡片数值仍走 cash_cents 由前端按两位小数渲染。"""
-    yuan = cents / 100
-    return f"{yuan:.2f}".rstrip("0").rstrip(".")
-
-
-def _display_name(user: User | None) -> str:
-    """好友昵称展示:昵称 → 微信昵称 → 手机尾号,全无则「好友」。"""
-    name = ((user.nickname if user else None) or (user.wechat_nickname if user else None) or "").strip()
-    if not name and user and user.phone:
-        name = f"用户{user.phone[-4:]}"
-    return name or "好友"
-
-
-# ---------------------------------------------------------------------------
-# 内核:落站内消息 + 厂商推送(全程吞异常)
-# ---------------------------------------------------------------------------
-
-
-def _dispatch(
-    db: Session,
-    *,
-    user_id: int,
-    type_key: str,
-    coins: int | None = None,
-    cash_cents: int | None = None,
-    info_rows: list[dict[str, str]] | None = None,
-    extra: dict[str, str] | None = None,
-    dedup_key: str | None = None,
-    push_vars: dict[str, str] | None = None,
-) -> Notification | None:
-    """落一条站内消息并向该用户设备直推。返回落库行;去重命中/失败返回 None。"""
-    try:
-        row = notif_repo.create_notification(
-            db,
-            user_id=user_id,
-            type_key=type_key,
-            coins=coins,
-            cash_cents=cash_cents,
-            info_rows=info_rows,
-            extra=extra,
-            dedup_key=dedup_key,
-        )
-    except IntegrityError:
-        # 同 (user, type, dedup_key) 已有未读消息 = 同一事件并发/重复触发 → 不重复落、不重复推
-        db.rollback()
-        logger.info(
-            "notification dedup hit user_id=%s type=%s dedup_key=%s", user_id, type_key, dedup_key
-        )
-        return None
-    except Exception:  # noqa: BLE001 — 通知失败绝不影响业务主流程
-        logger.exception("create notification failed user_id=%s type=%s", user_id, type_key)
-        try:
-            db.rollback()
-        except Exception:  # noqa: BLE001 — 回滚失败也不外抛,session 由请求生命周期兜底
-            logger.exception("rollback after notification failure also failed")
-        return None
-
-    _push_to_user_devices(db, row, push_vars)
-    return row
-
-
-def _push_to_user_devices(db: Session, row: Notification, push_vars: dict[str, str] | None) -> None:
-    """向消息归属用户的全部厂商推送目标直推(best-effort,单设备失败不影响其余)。"""
-    try:
-        title, body = catalog.render_push(row.type, push_vars)
-        # PRD §4 push 联动:extras 至少带 type + notificationId,外加该类型的跳转参数(extra 列)
-        extras: dict[str, str] = {"type": row.type}
-        extras.update({str(k): str(v) for k, v in (row.extra or {}).items()})
-        extras["notificationId"] = str(row.id)
-
-        for dev in device_repo.list_push_targets(db, user_id=row.user_id):
-            vendor = vendor_push.normalize_vendor(dev.push_vendor)
-            if not vendor or vendor not in vendor_push.SUPPORTED_VENDORS:
-                continue
-            if vendor_push.missing_settings(vendor):
-                # 本地/测试环境凭据不齐 → 只落站内消息,不发真推送(与 push/vendors 的报缺口径一致)
-                logger.info(
-                    "skip push (vendor %s not configured) user_id=%s type=%s",
-                    vendor, row.user_id, row.type,
-                )
-                continue
-            try:
-                vendor_push.send_notification(
-                    vendor, dev.push_token, title=title, body=body, extras=extras
-                )
-                logger.info(
-                    "push sent user_id=%s type=%s vendor=%s notification_id=%s",
-                    row.user_id, row.type, vendor, row.id,
-                )
-            except vendor_push.VendorPushError as e:
-                logger.warning(
-                    "push failed user_id=%s type=%s vendor=%s: %s", row.user_id, row.type, vendor, e
-                )
-    except Exception:  # noqa: BLE001 — 渲染/查设备等意外失败同样不外抛
-        logger.exception("push notification failed user_id=%s type=%s", row.user_id, row.type)
-
-
-# ---------------------------------------------------------------------------
-# 六个业务事件(PRD §1/§3/§5 编号见文件头)
-# ---------------------------------------------------------------------------
-
-
-def notify_withdraw_success(db: Session, order: WithdrawOrder) -> None:
-    """#3 提现成功:款项已存入微信零钱。点击无跳转仅消红点(extra 空)。"""
-    _dispatch(
-        db,
-        user_id=order.user_id,
-        type_key="withdraw_success",
-        cash_cents=order.amount_cents,
-        info_rows=[
-            {"label": "到账账户", "value": "微信钱包"},
-            {"label": "到账时间", "value": _fmt_time(datetime.now(CN_TZ))},
-        ],
-        extra={},
-        dedup_key=order.out_bill_no,
-        push_vars={"amount": notif_repo.cash_yuan(order.amount_cents)},
-    )
-
-
-def notify_withdraw_failed(db: Session, order: WithdrawOrder) -> None:
-    """#4 提现失败/退回:含微信侧失败、审核拒绝、解绑退回。点击跳提现页重新提现。
-
-    失败原因用 order.fail_reason(与 /withdraw/status 下发的用户可读原因同源)。
-    """
-    reason = (order.fail_reason or "").strip() or "提现未成功"
-    _dispatch(
-        db,
-        user_id=order.user_id,
-        type_key="withdraw_failed",
-        cash_cents=order.amount_cents,
-        info_rows=[
-            {"label": "失败原因", "value": reason},
-            {"label": "退回说明", "value": "款项已原路退回现金余额"},
-        ],
-        extra={"withdrawId": order.out_bill_no},
-        dedup_key=order.out_bill_no,
-        push_vars={"amount": notif_repo.cash_yuan(order.amount_cents), "reason": reason},
-    )
-
-
-def notify_feedback_reply(db: Session, feedback: Feedback) -> None:
-    """#9 官方回复:运营审核了反馈且未采纳(用户可见原因/留言落在反馈记录上)。
-    点击跳反馈历史页滚动高亮该条(extra.feedbackId)。"""
-    _dispatch(
-        db,
-        user_id=feedback.user_id,
-        type_key="feedback_reply",
-        info_rows=[{"label": "说明文案", "value": "快去看看官方给您的回复吧～"}],
-        extra={"feedbackId": str(feedback.id)},
-        dedup_key=str(feedback.id),
-    )
-
-
-def notify_feedback_reward(db: Session, feedback: Feedback) -> None:
-    """#10 反馈奖励:反馈被采纳,金币已到账。PRD 约定发奖必带官方留言(admin_reply);
-    运营漏填时省略该信息行,不硬造文案。"""
-    coins = int(feedback.reward_coins or 0)
-    info_rows = [{"label": "奖励说明", "value": "感谢您的反馈，您的金币奖励已到账"}]
-    reply = (feedback.admin_reply or "").strip()
-    if reply:
-        info_rows.append({"label": "官方留言", "value": reply})
-    info_rows.append({"label": "到账时间", "value": _fmt_time(datetime.now(CN_TZ))})
-    _dispatch(
-        db,
-        user_id=feedback.user_id,
-        type_key="feedback_reward",
-        coins=coins,
-        info_rows=info_rows,
-        extra={"feedbackId": str(feedback.id)},
-        dedup_key=str(feedback.id),
-        push_vars={"coins": str(coins)},
-    )
-
-
-def notify_report_approved(db: Session, report: PriceReport) -> None:
-    """#11 爆料审核通过:上报的更低价过审,金币已到账。点击跳爆料记录页高亮该条。"""
-    coins = int(report.reward_coins or 0)
-    store = (report.store_name or "").strip() or "该店铺"
-    _dispatch(
-        db,
-        user_id=report.user_id,
-        type_key="report_approved",
-        coins=coins,
-        info_rows=[
-            {"label": "奖励说明", "value": f"您爆料的「{store}」更低价已通过审核，金币奖励已到账"},
-            {"label": "到账时间", "value": _fmt_time(datetime.now(CN_TZ))},
-        ],
-        extra={"reportId": str(report.id)},
-        dedup_key=str(report.id),
-        push_vars={"store": store, "coins": str(coins)},
-    )
-
-
-def notify_invite_order_reward(
-    db: Session, *, inviter_user_id: int, invitee_user_id: int, cash_cents: int
-) -> None:
-    """#12 好友下单到账:被邀请好友完成首次下单(比价),现金奖励已入邀请人账户。
-    通知发给【邀请人】;每个好友只发一次奖 → dedup 按被邀请人。"""
-    invitee = db.get(User, invitee_user_id)
-    nickname = _display_name(invitee)
-    _dispatch(
-        db,
-        user_id=inviter_user_id,
-        type_key="invite_order_reward",
-        cash_cents=cash_cents,
-        info_rows=[
-            {"label": "奖励说明", "value": f"好友「{nickname}」完成首次下单"},
-            {"label": "到账时间", "value": _fmt_time(datetime.now(CN_TZ))},
-        ],
-        extra={"inviteeNickname": nickname},
-        dedup_key=str(invitee_user_id),
-        push_vars={"nickname": nickname, "amount": _yuan_trim(cash_cents)},
-    )

docs/api/README.md

@@ -1,13 +1,9 @@
 # 傻瓜比价 App 后端 — API 接口文档（索引）
 
 > Base URL:生产 `https://app-api.shaguabijia.com`;本地联调 `http://<开发机>:8770`
-> 协议:HTTP / JSON,请求与响应体均 `application/json`,字段统一 **snake_case**(⚠️ 例外:消息通知中心 `notifications` 族与厂商推送 `push` 族按 PRD 前端契约用 **camelCase**,见各自文档)
+> 协议:HTTP / JSON,请求与响应体均 `application/json`,字段统一 **snake_case**
 > 鉴权:需鉴权的接口在请求头带 `Authorization: Bearer <access_token>`
-<<<<<<< HEAD
-> 最后更新:2026-07-14（新增 **消息通知中心** 3 端点(M1-M3,虚拟数据阶段)与 **厂商推送测试** 3 端点(P1-P3,荣耀/华为/小米/OPPO/vivo);上一次 2026-06-23 补全 device/internal/CPS 短链等整族端点）
-=======
 > 最后更新:2026-07-09（① 比价透传改「软鉴权 + trace_id 签发 + harvest 落库」(#112 尾声帧 `trace/epilogue` 一并补录);② 新端点:`user/onboarding/reset`(#114)、`GET /internal/launch-confirm-samples`(#91);③ 参数更新:提现族 `source` 分账(#82/#121)、`wallet/account` 邀请奖励金余额、美团 feed/top-sales 按城市过滤(#116)、admin 调现金 `account` 目标账户(#95);④ **Admin 索引补全到当前全量**:新家族 roles(#117/#126)/coupon-data(#99)/device-liveness(#80)/event-logs(#83)/price-reports(#94)/CPS 运营台/提现审核族,及 feedbacks 采纳拒绝(#94/#105)、marquee 模式与真实条浏览(#122/#123)等。上一次 2026-07-03）
->>>>>>> origin/main
 > 架构:`app/api/v1/` 只放很轻的接口层;穿山甲/微信支付/极光/短信/美团等 SDK 集成的重逻辑在 `app/integrations/`,实现细节见 [docs/integrations/](../integrations/README.md)。
 
 ---
@@ -107,26 +103,12 @@
 | 36c | `POST /api/v1/user/onboarding/reset` | Bearer | [详情](./user/user-onboarding.md)（重置本设备引导标记,下次登录重走,#114） |
 | 37 | `DELETE /api/v1/user` | Bearer | [详情](./user/user-delete.md) |
 | **帮助与反馈**（前缀 `/api/v1/feedback`） |||
-<<<<<<< HEAD
-| 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 通知闭环验证已读） |
-=======
 | 38 | `POST /api/v1/feedback` | Bearer | [详情](./other/feedback.md) |
 | 38a | `GET /api/v1/feedback/config` | Bearer | [详情](./other/feedback-config.md)（反馈页「加群二维码」卡配置:开关+二维码图+三行文案） |
 | 38b | `GET /api/v1/feedback/records` | Bearer | [详情](./other/feedback-records.md)（我的反馈历史,pending/adopted/rejected） |
 | **埋点 & 订单上报**（前缀分散;全部 Bearer 除 analytics/events 不强制登录） |||
 | E1 | `POST /api/v1/analytics/events` | 无 | [详情](./other/analytics-events.md)（批量上报埋点事件,不强制登录,每批最多200条） |
 | E2 | `POST /api/v1/order/report` | Bearer | [详情](./other/order-report.md)（上报归因订单,比价后5分钟内点链接+支付金额与比价价相差≤1元） |
->>>>>>> origin/main
 | **首页门面数据 / 客户端配置**（前缀 `/api/v1/platform`；全平台展示数字 + 运营开关,**全部不鉴权**,登录前可读） |||
 | 39 | `GET /api/v1/platform/stats` | 无 | [详情](./platform/platform-stats.md) |
 | 40 | `GET /api/v1/platform/savings-feed` | 无 | [详情](./savings/platform-savings-feed.md) |

docs/api/notifications.md

@@ -1,132 +0,0 @@
-# 消息通知中心(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)。
->
-> **数据落库**:消息存 `notification` 表(`app/repositories/notification.py`,按用户隔离,`sentAt` 倒序)。业务事件统一走 `app/services/notification_events.py` 下发(站内消息 + 厂商 push 一条链路,业务事务 commit 后触发、失败只 log 不影响业务)。**已接入 6 类真实触发**:
->
-> | type | 触发点 |
-> |---|---|
-> | `withdraw_success` | 提现单转账到账(免确认直达 / 查单归一化 / 对账兜底,`repositories/wallet.py`) |
-> | `withdraw_failed` | 提现退款收口 `_refund_withdraw`(微信侧失败、审核拒绝、解绑退回) |
-> | `feedback_reply` | admin 反馈审核「拒绝」(原因/留言用户可见,`admin/routers/feedback.py`) |
-> | `feedback_reward` | admin 反馈审核「采纳」发金币(必带官方留言) |
-> | `report_approved` | admin 上报更低价「通过」发金币(`admin/routers/price_report.py`) |
-> | `invite_order_reward` | 被邀请好友首次成功比价 → 邀请人发 2 元(`repositories/invite.try_reward_on_compare`) |
->
-> 其余类型(奖励过期 ×2、权限异常 ×4、好友催单)业务侧尚未接入。要造联调数据,用 [POST /api/v1/push/test](./push-vendor-test.md) 的 `createNotification:true` 逐条插入。
->
-> ⚠️ **字段命名**:本组接口(含 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日」。`sentAt` 恒带 +08:00(服务端已归一,与库底层用 SQLite/PostgreSQL 无关)。
-
-**数值约束(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"
-```
-
-列表初始为空,登录后先用 [POST /api/v1/push/test](./push-vendor-test.md) 的 `createNotification:true` 插几条(可指定 `type` 覆盖不同版式),再验列表 / 角标 / 置读全流程;它同时把 `notificationId` 放进 push extras,可闭环验证「push → 站内已读联动」。

docs/api/push-vendor-test.md

@@ -1,103 +0,0 @@
-# 厂商推送测试三件套(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**。
-
-## 链路总览
-
-```
-真实业务事件(提现回执/反馈审核/爆料通过/好友下单 已接入;奖励过期等待接)
-      └→ services/notification_events(先落 notification 表,再向该用户全部已注册设备直推)
-              └→ 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()`。

run.bat

@@ -38,8 +38,4 @@ if errorlevel 1 (
 )
 
 REM Long-running foreground process. Ctrl+C to stop.
-REM --timeout-keep-alive 120: real-device debugging over `adb reverse` — uvicorn's default 5s
-REM   closes idle keep-alive connections, but the adb-reverse pipe doesn't propagate the close,
-REM   so okhttp reuses a dead connection and the next request fails with "unexpected end of
-REM   stream" (esp. login / message-center calls after an idle gap). Bump to 120s to avoid it.
-"%PY%" -m uvicorn app.main:app --host 0.0.0.0 --port 8770 --reload --timeout-keep-alive 120
+"%PY%" -m uvicorn app.main:app --host 0.0.0.0 --port 8770 --reload

run.sh

@@ -23,7 +23,4 @@ mkdir -p data            # sqlite 文件所在目录
 
 # --reload 只盯源码目录 app/:别去监视 logs/(日志写入触发"检测→再写日志"回环)和
 # data/(sqlite 频繁写)。改 alembic/、.env、本脚本后请手动重启。
-# --timeout-keep-alive 120:真机经 adb reverse 联调时,uvicorn 默认 5s 就关闭空闲 keep-alive
-#   连接,但 adb reverse 管道不把关闭事件透传回设备侧 → okhttp 复用"已死"的连接、下一次请求
-#   报 "unexpected end of stream"(尤其登录/消息中心等间隔较久的调用)。调大到 120s 规避。
-exec "$PY" -m uvicorn app.main:app --host 0.0.0.0 --port 8770 --reload --reload-dir app --timeout-keep-alive 120
+exec "$PY" -m uvicorn app.main:app --host 0.0.0.0 --port 8770 --reload --reload-dir app

scripts/fire_push_events.py

@@ -1,141 +0,0 @@
-"""直接触发「消息通知中心」真实推送链路,给指定用户(默认 11111111111)的已注册设备发 push。
-
-用于**后台无法驱动**的事件联调(本环境:wxpay 未配 → 提现成功打不通、提现单唯一约束 →
-造不了多张待审单、好友下单后台无入口)。本脚本直接调 services/notification_events 的真实
-下发函数,走的就是生产同一条链路:落 notification 表(站内消息) + 厂商直推(honor/huawei/
-xiaomi/oppo/vivo)到该用户 device_liveness 里已注册的 push token。
-
-默认只发这 3 类(后台驱动不了的):
-  #3  withdraw_success       提现到账
-  #4  withdraw_failed        提现失败,款项已退回
-  #12 invite_order_reward    好友下单奖励到账
-可用 --types 指定;--types all 追加后台能驱动的 #9/#10/#11(注意:这几类的点击跳转 id 是假的,
-仅验证「推送到达手机」,真实跳转请走后台审核流程)。
-
-  .venv\\Scripts\\python.exe scripts\\fire_push_events.py                    # 3 类各 10 条
-  .venv\\Scripts\\python.exe scripts\\fire_push_events.py --count 1          # 各 1 条(先小量验证通道)
-  .venv\\Scripts\\python.exe scripts\\fire_push_events.py --types withdraw_failed --count 3
-  .venv\\Scripts\\python.exe scripts\\fire_push_events.py --types all --count 2
-
-推送成败看输出里的 `shagua.vendor_push` 日志(push sent / push failed);2 台设备则每条各推 2 次。
-凭据缺失或 token 失效时 notification_events 只记日志、不抛错(站内消息仍会落库)。
-"""
-from __future__ import annotations
-
-import argparse
-import logging
-import random
-import sys
-import uuid
-
-from app.core.rewards import INVITE_COMPARE_REWARD_CENTS, PRICE_REPORT_REWARD_COINS
-from app.db.session import SessionLocal, engine
-from app.models.feedback import Feedback
-from app.models.price_report import PriceReport
-from app.models.wallet import WithdrawOrder
-from app.repositories import device as device_repo
-from app.repositories import user as user_repo
-from app.services import notification_events
-
-# SQL 回显静音;shagua.* 开到 INFO,好看到「push sent / push failed」结果
-# dev 下 APP_DEBUG=true → engine echo=True,echo 走 InstanceLogger 直写、无视 logger level,只能关 echo 本身
-logging.basicConfig(level=logging.INFO, format="%(levelname)s %(name)s: %(message)s")
-logging.getLogger("sqlalchemy.engine").setLevel(logging.WARNING)
-engine.echo = False
-
-if hasattr(sys.stdout, "reconfigure"):
-    sys.stdout.reconfigure(encoding="utf-8")
-
-DEFAULT_PHONE = "11111111111"
-DEFAULT_TYPES = ["withdraw_success", "withdraw_failed", "invite_order_reward"]
-ADMIN_DRIVEN = ["feedback_reward", "feedback_reply", "report_approved"]  # --types all 追加
-ALL_TYPES = DEFAULT_TYPES + ADMIN_DRIVEN
-
-_FAIL_REASONS = [
-    "微信零钱未实名,款项已退回",
-    "收款账户异常,款项已退回",
-    "超出微信零钱收款限额,款项已退回",
-]
-
-
-def _fire_one(db, uid: int, type_key: str, i: int) -> None:
-    """构造一条该类型的瞬态业务对象(不落业务表,只为给 notify 函数读字段),触发真实推送。"""
-    if type_key == "withdraw_success":
-        order = WithdrawOrder(user_id=uid, out_bill_no=uuid.uuid4().hex, amount_cents=50, source="coin_cash")
-        notification_events.notify_withdraw_success(db, order)
-    elif type_key == "withdraw_failed":
-        order = WithdrawOrder(
-            user_id=uid, out_bill_no=uuid.uuid4().hex, amount_cents=350, source="coin_cash",
-            fail_reason=random.choice(_FAIL_REASONS),
-        )
-        notification_events.notify_withdraw_failed(db, order)
-    elif type_key == "invite_order_reward":
-        # 假被邀请人 id(> 真实用户范围,避重):昵称回退「好友」。真实昵称请走 API 流程(见文末说明)。
-        fake_invitee = random.randint(900000, 999999)
-        notification_events.notify_invite_order_reward(
-            db, inviter_user_id=uid, invitee_user_id=fake_invitee, cash_cents=INVITE_COMPARE_REWARD_CENTS
-        )
-    elif type_key == "feedback_reward":
-        fb = Feedback(user_id=uid, content="(直发)", contact="", status="adopted",
-                      reward_coins=300, admin_reply="感谢反馈,您说的问题已修复上线,金币请查收～")
-        fb.id = random.randint(900000, 999999)
-        notification_events.notify_feedback_reward(db, fb)
-    elif type_key == "feedback_reply":
-        fb = Feedback(user_id=uid, content="(直发)", contact="", status="rejected",
-                      admin_reply="您的建议我们记录啦,会在后续版本评估~")
-        fb.id = random.randint(900000, 999999)
-        notification_events.notify_feedback_reply(db, fb)
-    elif type_key == "report_approved":
-        rep = PriceReport(
-            user_id=uid, reported_platform_id="jd", reported_platform_name="京东外卖",
-            reported_price_cents=8800, images=[], status="approved",
-            reward_coins=PRICE_REPORT_REWARD_COINS, store_name=f"测试火锅店{i:02d}",
-        )
-        rep.id = random.randint(900000, 999999)
-        notification_events.notify_report_approved(db, rep)
-    else:
-        raise SystemExit(f"未知类型: {type_key}(可选: {', '.join(ALL_TYPES)})")
-
-
-def main() -> None:
-    parser = argparse.ArgumentParser(description="直接触发消息通知中心真实推送(后台驱动不了的事件用)")
-    parser.add_argument("--phone", default=DEFAULT_PHONE, help=f"目标用户手机号(默认 {DEFAULT_PHONE})")
-    parser.add_argument("--count", type=int, default=10, help="每类发多少条(默认 10)")
-    parser.add_argument(
-        "--types", default=",".join(DEFAULT_TYPES),
-        help=f"逗号分隔的类型;'all' = {', '.join(ALL_TYPES)}。默认 {', '.join(DEFAULT_TYPES)}",
-    )
-    args = parser.parse_args()
-
-    types = ALL_TYPES if args.types.strip() == "all" else [t.strip() for t in args.types.split(",") if t.strip()]
-    bad = [t for t in types if t not in ALL_TYPES]
-    if bad:
-        print(f"❌ 未知类型: {', '.join(bad)}(可选: {', '.join(ALL_TYPES)})")
-        return
-
-    db = SessionLocal()
-    try:
-        user = user_repo.get_user_by_phone(db, args.phone)
-        if user is None:
-            print(f"❌ 用户 {args.phone} 不存在。请先用该手机号在 App 登录一次再跑本脚本。")
-            return
-        uid = user.id
-
-        targets = device_repo.list_push_targets(db, user_id=uid)
-        print(f"目标用户 {args.phone}(id={uid});已注册推送设备 {len(targets)} 台:"
-              f"{[t.push_vendor for t in targets] or '无(手机收不到!先在 App 上报 push token)'}")
-        print(f"即将触发:{types},每类 {args.count} 条 → 共 {len(types) * args.count} 条\n")
-
-        for t in types:
-            print(f"── {t} ×{args.count} " + "─" * 30)
-            for i in range(1, args.count + 1):
-                _fire_one(db, uid, t, i)
-
-        print(f"\n✅ 已触发完。站内消息已落 notification 表(用 {args.phone} 登录 App 可在消息中心看到);"
-              "\n   手机推送成败见上方 `shagua.vendor_push` 日志(push sent=成功 / push failed=失败)。")
-    finally:
-        db.close()
-
-
-if __name__ == "__main__":
-    main()

scripts/seed_mock_notifications.py

@@ -1,420 +0,0 @@
-"""给指定用户(默认手机号 11111111111)造一整套「消息通知中心」联调数据。
-
-不只是 notification 本身,还把 13 种类型**点击后要跳转的落地页数据**一起造齐,保证每条都能点开看到真实内容:
-
-  notification 类型            点击落地              需要的业务数据(本脚本一并造)
-  ─────────────────────────────────────────────────────────────────────────────
-  reward_expiring/expired      赚钱页(tab)          —(金额在通知里,无需外部记录)
-  withdraw_success             无跳转,仅消红点        —
-  withdraw_failed              提现页(withdrawId)    withdraw_order(failed 一单)
-  perm_*(4 种)                客户端权限检测弹窗       —(纯客户端)
-  feedback_reply               我的反馈(feedbackId)  feedback(rejected + 官方回复)
-  feedback_reward              我的反馈(feedbackId)  feedback(adopted + 官方留言 + 奖励金币)
-  report_approved              我的爆料(reportId)    price_report(approved + 截图 + 奖励)
-  invite_order_reward          邀请页                invite_relation + 好友 user(已完成比价)
-  invite_remind                邀请页(scrollTo)     invite_relation + 好友 user(未完成)
-
-配套还造:钱包余额 + 金币/现金/邀请奖励金流水(让赚钱页 / 金币明细 / 现金明细 / 邀请战绩都有内容)。
-
-幂等:每次先清掉该用户上一轮由本脚本造的全部数据(通知 + 上述业务记录 + mock 好友 + mock 截图)再重建。
-  .venv\\Scripts\\python.exe scripts\\seed_mock_notifications.py
-  .venv\\Scripts\\python.exe scripts\\seed_mock_notifications.py --phone 11111111111
-  .venv\\Scripts\\python.exe scripts\\seed_mock_notifications.py --clean-only
-
-时间口径按各域现有约定:notification.sent_at 用东八区(带 +08:00 下发);feedback / withdraw /
-钱包流水 / 邀请关系用 naive UTC(= func.now() 在 SQLite 的口径,与真实数据一致);price_report 用
-naive 北京时间(与 report_repo.create_report 一致)。
-"""
-from __future__ import annotations
-
-import argparse
-import struct
-import sys
-import uuid
-import zlib
-from datetime import datetime, timedelta, timezone
-from pathlib import Path
-
-from sqlalchemy import delete, select
-
-from app.core import rewards
-from app.core.config import settings
-from app.db.session import SessionLocal
-from app.models.feedback import Feedback
-from app.models.invite import InviteRelation
-from app.models.notification import Notification
-from app.models.price_report import PriceReport
-from app.models.user import User
-from app.models.wallet import (
-    CashTransaction,
-    CoinAccount,
-    CoinTransaction,
-    InviteCashTransaction,
-    WithdrawOrder,
-)
-from app.repositories import notification as notif_repo
-from app.repositories import user as user_repo
-
-if hasattr(sys.stdout, "reconfigure"):
-    sys.stdout.reconfigure(encoding="utf-8")  # Windows GBK 控制台也能打印中文/¥
-
-_CST = timezone(timedelta(hours=8))
-
-DEFAULT_PHONE = "11111111111"
-
-# mock 好友(被邀请人):固定手机号,便于幂等清理。(phone, 昵称, 是否已完成比价)
-FRIEND_SPECS = [
-    ("12000000001", "柚子", True),   # 已完成 → 驱动 invite_order_reward,计入邀请战绩
-    ("12000000003", "小美", True),   # 已完成 → 让邀请列表 / 战绩更丰满
-    ("12000000002", "阿泽", False),  # 未完成 → 驱动 invite_remind(去催单)
-]
-FRIEND_PHONES = [p for p, _, _ in FRIEND_SPECS]
-
-_REPORT_DIR = Path(settings.MEDIA_ROOT) / "price_report"
-_MOCK_IMG_GLOB = "mock_notif_*.png"  # 本脚本生成的截图前缀,清理按此删
-
-
-# ---------------------------------------------------------------------------
-# 时间口径小工具
-# ---------------------------------------------------------------------------
-def _utc() -> datetime:
-    """naive UTC now(与 func.now() 在 SQLite 一致:feedback / withdraw / 流水 / 邀请关系用)。"""
-    return datetime.now(timezone.utc).replace(tzinfo=None)
-
-
-def _bj_naive() -> datetime:
-    """naive 北京 wall-clock(price_report 用,与 report_repo.create_report 一致)。"""
-    return datetime.now(_CST).replace(tzinfo=None)
-
-
-# ---------------------------------------------------------------------------
-# mock 截图(纯色 PNG,无需 Pillow;抄 seed_mock_price_reports 的手写字节法)
-# ---------------------------------------------------------------------------
-def _solid_png(width: int, height: int, rgb: tuple[int, int, int]) -> bytes:
-    def _chunk(typ: bytes, data: bytes) -> bytes:
-        body = typ + data
-        return struct.pack(">I", len(data)) + body + struct.pack(">I", zlib.crc32(body) & 0xFFFFFFFF)
-
-    ihdr = struct.pack(">IIBBBBB", width, height, 8, 2, 0, 0, 0)  # RGB truecolor
-    row = b"\x00" + bytes(rgb) * width
-    idat = zlib.compress(row * height, 9)
-    return b"\x89PNG\r\n\x1a\n" + _chunk(b"IHDR", ihdr) + _chunk(b"IDAT", idat) + _chunk(b"IEND", b"")
-
-
-def _write_mock_image(name: str, rgb: tuple[int, int, int]) -> str:
-    _REPORT_DIR.mkdir(parents=True, exist_ok=True)
-    (_REPORT_DIR / name).write_bytes(_solid_png(320, 320, rgb))
-    return f"{settings.MEDIA_URL_PREFIX}/price_report/{name}"
-
-
-# ---------------------------------------------------------------------------
-# 清理(幂等)
-# ---------------------------------------------------------------------------
-def clean(db, target: User) -> None:
-    uid = target.id
-    friend_ids = list(
-        db.execute(select(User.id).where(User.phone.in_(FRIEND_PHONES))).scalars()
-    )
-
-    # 1) 目标用户的通知 + 业务记录 + 钱包
-    for model in (
-        Notification, Feedback, PriceReport, WithdrawOrder,
-        CashTransaction, CoinTransaction, InviteCashTransaction, CoinAccount,
-    ):
-        db.execute(delete(model).where(model.user_id == uid))
-    # 2) 邀请关系(目标作为邀请人 + mock 好友作为被邀请人)
-    db.execute(delete(InviteRelation).where(InviteRelation.inviter_user_id == uid))
-    if friend_ids:
-        db.execute(delete(InviteRelation).where(InviteRelation.invitee_user_id.in_(friend_ids)))
-        db.execute(delete(User).where(User.id.in_(friend_ids)))
-    db.commit()
-
-    # 3) mock 截图文件
-    if _REPORT_DIR.exists():
-        for f in _REPORT_DIR.glob(_MOCK_IMG_GLOB):
-            f.unlink(missing_ok=True)
-
-
-# ---------------------------------------------------------------------------
-# 造业务记录(通知的点击落地数据)
-# ---------------------------------------------------------------------------
-def _make_friends(db, inviter: User) -> dict[str, User]:
-    """建 mock 好友 user + 邀请关系(注册即生效;完成比价的置 compare_reward_granted 并发奖励金)。"""
-    now = _utc()
-    friends: dict[str, User] = {}
-    for i, (phone, nickname, _completed) in enumerate(FRIEND_SPECS):
-        u = User(
-            phone=phone,
-            username=user_repo._gen_unique_username(db),
-            nickname=nickname,
-            register_channel="sms",
-            status="active",
-            created_at=now - timedelta(days=6 - i),
-            last_login_at=now - timedelta(hours=2),
-        )
-        db.add(u)
-        friends[nickname] = u
-    db.flush()  # 拿 friend.id
-
-    for i, (_phone, nickname, completed) in enumerate(FRIEND_SPECS):
-        f = friends[nickname]
-        db.add(InviteRelation(
-            inviter_user_id=inviter.id,
-            invitee_user_id=f.id,
-            channel="clipboard",
-            status="effective",
-            compare_reward_granted=completed,
-            compare_reward_cents=rewards.INVITE_COMPARE_REWARD_CENTS if completed else 0,
-            compare_rewarded_at=(now - timedelta(days=5 - i)) if completed else None,
-            created_at=now - timedelta(days=6 - i),
-        ))
-    return friends
-
-
-def _make_feedbacks(db, uid: int) -> dict[str, Feedback]:
-    """两条反馈:一条(rejected)带官方回复 → feedback_reply;一条(adopted)带留言+奖励 → feedback_reward。"""
-    now = _utc()
-    reply = Feedback(
-        user_id=uid,
-        content="比价结果页希望能一键复制到微信分享给朋友。",
-        contact="",
-        source="profile",
-        status="rejected",
-        admin_reply="您反馈的分享功能我们记录啦,会在后续版本评估上线,感谢支持～",
-        review_note="需求已进池",
-        reviewed_at=now - timedelta(hours=5),
-        created_at=now - timedelta(days=1, hours=2),
-    )
-    reward = Feedback(
-        user_id=uid,
-        content="点某些店铺比价偶尔会闪退,机型 Redmi K60。",
-        contact="",
-        source="comparison",
-        scene="compare_slow",
-        status="adopted",
-        admin_reply="感谢反馈,您说的闪退问题已修复上线,送您的金币请查收～",
-        review_note="已修复:比价页空指针",
-        reward_coins=300,
-        reviewed_at=now - timedelta(days=1),
-        created_at=now - timedelta(days=3),
-    )
-    db.add_all([reply, reward])
-    db.flush()
-    return {"reply": reply, "reward": reward}
-
-
-def _make_report(db, uid: int) -> PriceReport:
-    """一条 approved 上报(带真实可加载截图 + 奖励金币)→ report_approved 点击可看爆料详情。"""
-    now = _bj_naive()
-    img = _write_mock_image("mock_notif_report.png", (250, 173, 20))
-    rep = PriceReport(
-        user_id=uid,
-        comparison_record_id=None,
-        store_name="蜀大侠火锅(春熙路店)",
-        dish_summary="招牌牛油锅 × 1、鲜毛肚 × 2",
-        original_platform_id="meituan-waimai",
-        original_platform_name="美团外卖",
-        original_price_cents=13800,
-        reported_platform_id="jd-waimai",
-        reported_platform_name="京东外卖",
-        reported_price_cents=11800,
-        images=[img],
-        status="approved",
-        reward_coins=1000,
-        reviewed_at=now - timedelta(days=39, hours=-1),
-        created_at=now - timedelta(days=40),
-    )
-    db.add(rep)
-    db.flush()
-    return rep
-
-
-def _make_withdraws(db, uid: int) -> dict[str, WithdrawOrder]:
-    """两单提现:success(历史)+ failed(驱动 withdraw_failed 点击去提现页)。不造在审单,避活动单唯一约束。"""
-    now = _utc()
-    success = WithdrawOrder(
-        user_id=uid, out_bill_no=uuid.uuid4().hex, amount_cents=500, source="coin_cash",
-        user_name="测试用户", status="success", wechat_state="SUCCESS",
-        transfer_bill_no="1330" + str(uuid.uuid4().int)[:26],
-        created_at=now - timedelta(days=5), updated_at=now - timedelta(days=5),
-    )
-    failed = WithdrawOrder(
-        user_id=uid, out_bill_no=uuid.uuid4().hex, amount_cents=350, source="coin_cash",
-        user_name="测试用户", status="failed", wechat_state="FAIL",
-        transfer_bill_no="1330" + str(uuid.uuid4().int)[:26],
-        fail_reason="微信实名与提现实名不一致,款项已原路退回现金余额",
-        created_at=now - timedelta(days=2), updated_at=now - timedelta(days=2) + timedelta(hours=1),
-    )
-    db.add_all([success, failed])
-    db.flush()
-    return {"success": success, "failed": failed}
-
-
-def _make_wallet(db, uid: int, friends: dict[str, User], withdraws: dict[str, WithdrawOrder]) -> None:
-    """钱包余额 + 三本流水(金币 / 现金 / 邀请奖励金),让赚钱页与各明细页都有内容。"""
-    now = _utc()
-
-    # 金币流水(只增,链上 balance_after)
-    coin_events = [
-        (2000, "signin", (now - timedelta(days=6)).date().isoformat(), "每日签到"),
-        (160, "reward_video", uuid.uuid4().hex, "看视频奖励"),
-        (500, "task_enable_notification", "task_enable_notification", "开启消息提醒奖励"),
-        (1000, "report_reward", None, "爆料审核通过奖励"),
-        (300, "feedback_reward", None, "反馈采纳奖励"),
-    ]
-    coin_bal = 0
-    for amt, biz, ref, remark in coin_events:
-        coin_bal += amt
-        db.add(CoinTransaction(
-            user_id=uid, amount=amt, balance_after=coin_bal, biz_type=biz,
-            ref_id=ref, remark=remark, created_at=now - timedelta(days=4),
-        ))
-
-    # 现金流水:兑入 + 两单提现扣款 + 失败退款 → 期末 1500
-    cash_events = [
-        (now - timedelta(days=10), 2000, "exchange_in", None, "金币兑入"),
-        (withdraws["success"].created_at, -500, "withdraw", withdraws["success"].out_bill_no, "提现扣款"),
-        (withdraws["failed"].created_at, -350, "withdraw", withdraws["failed"].out_bill_no, "提现扣款"),
-        (withdraws["failed"].updated_at, 350, "withdraw_refund", withdraws["failed"].out_bill_no, "提现退款"),
-    ]
-    cash_events.sort(key=lambda e: e[0])
-    cash_bal = 0
-    for t, amt, biz, ref, remark in cash_events:
-        cash_bal += amt
-        db.add(CashTransaction(
-            user_id=uid, amount_cents=amt, balance_after_cents=cash_bal,
-            biz_type=biz, ref_id=ref, remark=remark, created_at=t,
-        ))
-
-    # 邀请奖励金流水:每个已完成好友发一笔 → 期末 = 已完成好友数 × 单笔奖励
-    invite_bal = 0
-    reward_cents = rewards.INVITE_COMPARE_REWARD_CENTS
-    for i, (_phone, nickname, completed) in enumerate(FRIEND_SPECS):
-        if not completed:
-            continue
-        invite_bal += reward_cents
-        db.add(InviteCashTransaction(
-            user_id=uid, amount_cents=reward_cents, balance_after_cents=invite_bal,
-            biz_type="invite_reward", ref_id=str(friends[nickname].id),
-            remark="好友比价奖励", created_at=now - timedelta(days=5 - i),
-        ))
-
-    total_earned = sum(a for a, *_ in coin_events)
-    db.add(CoinAccount(
-        user_id=uid,
-        coin_balance=coin_bal,
-        cash_balance_cents=cash_bal,
-        invite_cash_balance_cents=invite_bal,
-        total_coin_earned=total_earned,
-    ))
-
-
-# ---------------------------------------------------------------------------
-# 造 13 类通知(extra 指向上面真实记录的 id)
-# ---------------------------------------------------------------------------
-def _notif(uid: int, type_key: str, sent_at: datetime, *, read: bool = False,
-           extra_override: dict | None = None, dedup_key: str | None = None) -> Notification:
-    card = notif_repo.build_sample_card(type_key, sent_at=sent_at)
-    extra = dict(card.get("extra", {}))
-    if extra_override:
-        extra.update(extra_override)
-    return Notification(
-        user_id=uid, type=type_key, is_read=read,
-        read_at=(sent_at + timedelta(minutes=5)) if read else None,
-        sent_at=sent_at, dedup_key=dedup_key,
-        coins=card.get("coins"), cash_cents=card.get("cash_cents"),
-        info_rows=card.get("info_rows", []), extra=extra,
-    )
-
-
-def _make_notifications(
-    db, uid: int, fb: dict[str, Feedback], rep: PriceReport,
-    wd: dict[str, WithdrawOrder], friends: dict[str, User],
-) -> list[Notification]:
-    n = datetime.now(_CST)
-
-    def ago(**kw) -> datetime:
-        return n - timedelta(**kw)
-
-    rows = [
-        # —— 提现助手 ——（金额/现金卡;withdraw_failed 指向真实失败单)
-        _notif(uid, "reward_expiring", ago(hours=2), dedup_key=f"batch_{n:%Y%m%d}"),
-        _notif(uid, "reward_expired", ago(days=1, hours=3), read=True),
-        _notif(uid, "withdraw_success", ago(minutes=10)),
-        _notif(uid, "withdraw_success", ago(days=3), read=True),  # 额外一条(历史,已读)
-        _notif(uid, "withdraw_failed", ago(days=1, hours=1),
-               extra_override={"withdrawId": str(wd["failed"].id)}),
-        # —— 系统通知(权限异常 ×4;dedup_key=权限名,未读期间只保留一条)——
-        _notif(uid, "perm_accessibility", ago(hours=1), dedup_key="accessibility"),
-        _notif(uid, "perm_battery", ago(days=3), read=True, dedup_key="battery"),
-        _notif(uid, "perm_autostart", ago(days=5), dedup_key="autostart"),
-        _notif(uid, "perm_overlay", ago(days=6), read=True, dedup_key="overlay"),
-        # —— 我的反馈(feedbackId 指向真实反馈)——
-        _notif(uid, "feedback_reply", ago(hours=4),
-               extra_override={"feedbackId": str(fb["reply"].id)}),
-        _notif(uid, "feedback_reward", ago(days=1),
-               extra_override={"feedbackId": str(fb["reward"].id)}),
-        _notif(uid, "feedback_reply", ago(days=380), read=True,  # 跨年(测「YYYY年M月D日」),已读
-               extra_override={"feedbackId": str(fb["reply"].id)}),
-        # —— 我的爆料(reportId 指向真实上报)——
-        _notif(uid, "report_approved", ago(days=40),  # 当年(测「M月D日」)
-               extra_override={"reportId": str(rep.id)}),
-        # —— 好友邀请(inviteeNickname 指向真实好友)——
-        _notif(uid, "invite_order_reward", ago(minutes=20),
-               extra_override={"inviteeNickname": "柚子"}),
-        _notif(uid, "invite_remind", ago(days=2),
-               extra_override={"inviteeNickname": "阿泽", "scrollTo": "remind"}),
-    ]
-    db.add_all(rows)
-    return rows
-
-
-def seed(db, target: User) -> list[Notification]:
-    uid = target.id
-    friends = _make_friends(db, target)
-    fb = _make_feedbacks(db, uid)
-    rep = _make_report(db, uid)
-    wd = _make_withdraws(db, uid)
-    _make_wallet(db, uid, friends, wd)
-    rows = _make_notifications(db, uid, fb, rep, wd, friends)
-    db.commit()
-    return rows
-
-
-# ---------------------------------------------------------------------------
-def main() -> None:
-    parser = argparse.ArgumentParser(description="给指定用户造消息通知中心 + 点击落地页 mock 数据")
-    parser.add_argument("--phone", default=DEFAULT_PHONE, help=f"目标用户手机号(默认 {DEFAULT_PHONE})")
-    parser.add_argument("--clean-only", action="store_true", help="只清理,不重建")
-    args = parser.parse_args()
-
-    db = SessionLocal()
-    try:
-        target = user_repo.get_user_by_phone(db, args.phone)
-        if target is None:
-            print(f"❌ 用户 {args.phone} 不存在。请先用该手机号在 App 登录一次(SMS mock:任意 6 位验证码)再跑本脚本。")
-            return
-
-        clean(db, target)
-        print(f"🧹 已清理用户 {args.phone}(id={target.id})上一轮 mock 通知 + 业务记录 + mock 好友/截图")
-        if args.clean_only:
-            print("✅ 仅清理,已完成。")
-            return
-
-        rows = seed(db, target)
-        unread = sum(1 for r in rows if not r.is_read)
-        print(f"\n✅ 已为用户 {args.phone}(id={target.id})生成 {len(rows)} 条通知(未读 {unread}):")
-        for r in sorted(rows, key=lambda x: x.sent_at, reverse=True):
-            flag = "  " if r.is_read else "●"
-            print(f"   {flag} {r.type:<20} {r.sent_at:%Y-%m-%d %H:%M}  extra={r.extra}")
-        print(
-            "\n👉 用 11111111111 登录 App(SMS mock:任意 6 位验证码)看消息通知中心;"
-            "\n   逐条点击验证跳转:反馈→我的反馈、爆料→我的爆料、提现失败→提现页、邀请→邀请页、权限→检测弹窗。"
-            "\n   后端若没带 --reload,改了数据也无需重启(本脚本直接写库,接口实时读)。"
-        )
-    finally:
-        db.close()
-
-
-if __name__ == "__main__":
-    main()

scripts/seed_push_admin_test.py

@@ -1,135 +0,0 @@
-"""给指定用户(默认 11111111111)造「后台可驱动」的推送联调数据。
-
-覆盖能从**管理后台点一下就触发手机推送**的 3 类事件,每类 10 条待审记录:
-
-  事件(PRD #)      后台动作                              造的数据
-  ────────────────────────────────────────────────────────────────────
-  #10 反馈奖励       反馈工单 → 采纳(填回复留言 + 金币)   10 条 pending feedback(标「请采纳」)
-  #9  官方回复       反馈工单 → 拒绝(填未采纳原因/留言)   10 条 pending feedback(标「请拒绝」)
-  #11 爆料审核通过   上报更低价 → 通过                     10 条 pending price_report
-
-触发链路:admin 审核 → 发金币/改状态 → services/notification_events 落站内消息 + 厂商直推
-→ 该用户已注册设备(device_liveness)收到 push。
-
-其余 3 类(#3 提现成功 / #4 提现失败 / #12 好友下单到账)后台无法在本环境驱动
-(wxpay 未配 / 提现单唯一约束 / 后台无入口),用 scripts/fire_push_events.py 直接触发。
-
-幂等:每次先删掉本脚本上一轮造的记录(按内容标记 [PUSH测试] 识别,不动用户真实反馈/爆料),再重建。
-  .venv\\Scripts\\python.exe scripts\\seed_push_admin_test.py
-  .venv\\Scripts\\python.exe scripts\\seed_push_admin_test.py --phone 11111111111 --count 10
-  .venv\\Scripts\\python.exe scripts\\seed_push_admin_test.py --clean-only
-"""
-from __future__ import annotations
-
-import argparse
-import logging
-import sys
-
-from sqlalchemy import delete, select
-
-from app.db.session import SessionLocal
-from app.models.feedback import Feedback
-from app.models.price_report import PriceReport
-from app.repositories import user as user_repo
-
-logging.getLogger("sqlalchemy.engine").setLevel(logging.WARNING)  # 静音 SQL 回显,输出更干净
-
-if hasattr(sys.stdout, "reconfigure"):
-    sys.stdout.reconfigure(encoding="utf-8")
-
-DEFAULT_PHONE = "11111111111"
-MARK = "[PUSH测试]"  # 本脚本造的数据统一带此标记,幂等清理按它识别(不误删真实数据)
-
-
-def clean(db, uid: int) -> tuple[int, int]:
-    """删掉本脚本上一轮造的带标记记录(任何状态都删,彻底重置)。返回 (删反馈数, 删爆料数)。"""
-    fb_ids = list(db.execute(
-        select(Feedback.id).where(Feedback.user_id == uid, Feedback.content.like(f"{MARK}%"))
-    ).scalars())
-    rep_ids = list(db.execute(
-        select(PriceReport.id).where(
-            PriceReport.user_id == uid, PriceReport.store_name.like(f"{MARK}%")
-        )
-    ).scalars())
-    if fb_ids:
-        db.execute(delete(Feedback).where(Feedback.id.in_(fb_ids)))
-    if rep_ids:
-        db.execute(delete(PriceReport).where(PriceReport.id.in_(rep_ids)))
-    db.commit()
-    return len(fb_ids), len(rep_ids)
-
-
-def seed(db, uid: int, count: int) -> None:
-    # #10 反馈奖励:采纳这些 → 手机收「反馈奖励已到账」。采纳时记得在后台填「给用户的回复留言」
-    # (PRD 要求发奖必带官方留言),否则通知里不带留言行。
-    for i in range(1, count + 1):
-        db.add(Feedback(
-            user_id=uid,
-            content=f"{MARK} 请【采纳】我 → 触发 #10 反馈奖励推送。测试反馈内容 {i:02d}:比价页能加个历史记录就好了。",
-            contact="",
-            source="profile",
-            status="pending",
-        ))
-    # #9 官方回复:拒绝这些 → 手机收「您的反馈有回复啦」。拒绝时填「未采纳原因」+「回复留言」。
-    for i in range(1, count + 1):
-        db.add(Feedback(
-            user_id=uid,
-            content=f"{MARK} 请【拒绝】我 → 触发 #9 官方回复推送。测试反馈内容 {i:02d}:希望支持某某小众平台比价。",
-            contact="",
-            source="comparison",
-            scene="other",
-            status="pending",
-        ))
-    # #11 爆料审核通过:通过这些 → 手机收「爆料审核通过」(发固定金币)。
-    for i in range(1, count + 1):
-        db.add(PriceReport(
-            user_id=uid,
-            comparison_record_id=None,
-            store_name=f"{MARK}测试火锅店{i:02d}",
-            dish_summary="招牌套餐 × 1",
-            original_platform_id="meituan-waimai",
-            original_platform_name="美团外卖",
-            original_price_cents=9900,
-            reported_platform_id="jd-waimai",
-            reported_platform_name="京东外卖",
-            reported_price_cents=8800,
-            images=[],
-            status="pending",
-        ))
-    db.commit()
-
-
-def main() -> None:
-    parser = argparse.ArgumentParser(description="造后台可驱动的推送联调数据(反馈×2 + 爆料)")
-    parser.add_argument("--phone", default=DEFAULT_PHONE, help=f"目标用户手机号(默认 {DEFAULT_PHONE})")
-    parser.add_argument("--count", type=int, default=10, help="每类造多少条(默认 10)")
-    parser.add_argument("--clean-only", action="store_true", help="只清理本脚本造的数据,不重建")
-    args = parser.parse_args()
-
-    db = SessionLocal()
-    try:
-        user = user_repo.get_user_by_phone(db, args.phone)
-        if user is None:
-            print(f"❌ 用户 {args.phone} 不存在。请先用该手机号在 App 登录一次再跑本脚本。")
-            return
-        uid = user.id
-
-        nf, nr = clean(db, uid)
-        print(f"🧹 已清理上一轮 [PUSH测试] 数据:反馈 {nf} 条、爆料 {nr} 条")
-        if args.clean_only:
-            print("✅ 仅清理,已完成。")
-            return
-
-        seed(db, uid, args.count)
-        print(f"\n✅ 已为 {args.phone}(id={uid})造好后台联调数据(每类 {args.count} 条):")
-        print(f"   • 反馈工单「请采纳」× {args.count} → 后台【采纳】(填回复留言+金币)→ 手机收 #10 反馈奖励")
-        print(f"   • 反馈工单「请拒绝」× {args.count} → 后台【拒绝】(填未采纳原因/留言)→ 手机收 #9 官方回复")
-        print(f"   • 上报更低价    × {args.count} → 后台【通过】→ 手机收 #11 爆料审核通过")
-        print("\n👉 打开管理后台(:8771)对应列表即可看到这些待审记录,逐条审核就会推到手机。")
-        print("   #3/#4/#12 本环境后台驱动不了,用:.venv\\Scripts\\python.exe scripts\\fire_push_events.py")
-    finally:
-        db.close()
-
-
-if __name__ == "__main__":
-    main()

scripts/show_device_regids.bat

@@ -1,13 +0,0 @@
-@echo off
-REM Show all device_id + push regIds (push_token / registration_id) from the
-REM local SQLite DB. Double-click this file to see everything, or run from a
-REM console. Optional substring filter:  show_device_regids.bat xiaomi
-REM Works from ANY directory (locates project root + venv python by itself).
-REM ASCII-only comments: cmd parses .bat in the console codepage (GBK); UTF-8
-REM Chinese here gets mangled into bogus commands.
-cd /d "%~dp0.."
-set "PY=python"
-if exist ".venv\Scripts\python.exe" set "PY=.venv\Scripts\python.exe"
-"%PY%" "scripts\show_device_regids.py" %*
-echo.
-pause

scripts/show_device_regids.py

@@ -1,136 +0,0 @@
-# -*- coding: utf-8 -*-
-"""导出 device_liveness 表里所有 device_id 及其对应的推送 regId,按 push_vendor 分组(本地开发工具)。
-
-用法:
-  双击 scripts/show_device_regids.bat,或命令行:
-    python scripts/show_device_regids.py [filter]
-
-  [filter] 可选:大小写不敏感的子串,匹配 device_id / push_vendor / push_token /
-  registration_id 任一列;不传则列出全部。
-    show_device_regids.py xiaomi          # 只看小米
-    show_device_regids.py device_Pixel    # 按 device_id 片段找
-
-说明:
-- 直接以**只读**方式读 SQLite(server 在跑也不会抢写锁),所以开不开服务器都能用。
-- push_token = 各厂商的 regId/pushToken(新链路,当前在用);registration_id = 旧极光
-  regId(历史兼容)。两列都打出来,方便跟 logcat 现役值逐字比对。
-- 输出刻意保持纯 ASCII 排版并竖排展示**完整值**:双击弹出的 cmd 走 GBK 码页,竖排纯
-  ASCII 不会乱码;完整值不截断,才能直接复制去比对。
-- 若 DATABASE_URL 不是 sqlite(生产 postgres),这里只提示改用 psql。
-"""
-from __future__ import annotations
-
-import os
-import sqlite3
-import sys
-from pathlib import Path
-
-REPO_ROOT = Path(__file__).resolve().parent.parent
-
-
-def _resolve_sqlite_path() -> Path | None:
-    """按 env > .env > 默认 的顺序拿 DATABASE_URL,解析出 sqlite 文件路径。"""
-    url = os.environ.get("DATABASE_URL", "").strip()
-    if not url:
-        env = REPO_ROOT / ".env"
-        if env.exists():
-            for line in env.read_text(encoding="utf-8", errors="ignore").splitlines():
-                s = line.strip()
-                if s.startswith("DATABASE_URL="):
-                    url = s.split("=", 1)[1].strip().strip('"').strip("'")
-                    break
-    if not url:
-        url = "sqlite:///./data/app.db"
-    if not url.startswith("sqlite"):
-        print(f"[!] DATABASE_URL not sqlite: {url}")
-        print("    This tool only reads a local SQLite dev DB. For prod use psql.")
-        return None
-    raw = url.split("///", 1)[1] if "///" in url else "./data/app.db"
-    p = Path(raw)
-    if not p.is_absolute():
-        p = (REPO_ROOT / raw).resolve()
-    return p
-
-
-def main() -> int:
-    needle = sys.argv[1].lower() if len(sys.argv) > 1 else None
-
-    db = _resolve_sqlite_path()
-    if db is None:
-        return 2
-    if not db.exists():
-        print(f"[X] DB file not found: {db}")
-        return 2
-
-    # 只读打开,避免和正在运行的 server 抢写锁。URI 路径用正斜杠(as_posix)规避
-    # Windows 反斜杠/盘符在 file: URI 里的歧义;万一 URI 形式打不开再回退普通连接。
-    try:
-        con = sqlite3.connect(f"file:{db.as_posix()}?mode=ro", uri=True)
-    except sqlite3.OperationalError:
-        con = sqlite3.connect(str(db))
-    con.row_factory = sqlite3.Row
-    rows = con.execute(
-        """
-        SELECT id, user_id, device_id, push_vendor, push_token, registration_id,
-               platform, app_version, liveness_state, last_heartbeat_at, updated_at
-        FROM device_liveness
-        ORDER BY updated_at DESC
-        """
-    ).fetchall()
-    con.close()
-
-    if needle:
-        def hit(r: sqlite3.Row) -> bool:
-            for k in ("device_id", "push_vendor", "push_token", "registration_id"):
-                v = r[k]
-                if v and needle in str(v).lower():
-                    return True
-            return False
-
-        rows = [r for r in rows if hit(r)]
-
-    with_token = sum(1 for r in rows if (r["push_token"] or "").strip())
-    with_reg = sum(1 for r in rows if (r["registration_id"] or "").strip())
-
-    # 按 push_vendor 分组;None/空归入 "(none)"。组顺序:设备数多的在前,(none) 垫底;
-    # 组内沿用 updated_at DESC(rows 查询时已如此排序,dict 保序即可)。
-    groups: dict[str, list] = {}
-    for r in rows:
-        groups.setdefault(r["push_vendor"] or "(none)", []).append(r)
-    ordered = sorted(
-        groups.items(), key=lambda kv: (kv[0] == "(none)", -len(kv[1]), kv[0])
-    )
-    tally = "  ".join(f"{v}={len(items)}" for v, items in ordered) or "-"
-
-    print(f"DB : {db}")
-    line = f"device_liveness : {len(rows)} row(s)"
-    if needle:
-        line += f'   filter="{sys.argv[1]}"'
-    print(line)
-    print(f"has push_token(vendor regId) : {with_token}"
-          f"    has registration_id(jiguang) : {with_reg}")
-    print(f"vendors : {tally}")
-    print("=" * 72)
-
-    if not rows:
-        print("(no rows)")
-        return 0
-
-    for vendor, items in ordered:
-        print()
-        print(f"===== vendor={vendor} : {len(items)} device(s) =====")
-        for r in items:
-            print(f"  [#{r['id']}]  user_id={r['user_id']}  platform={r['platform']}"
-                  f"  state={r['liveness_state']}  app={r['app_version'] or '-'}")
-            print(f"    device_id       : {r['device_id']}")
-            print(f"    push_token(regId): {r['push_token'] or '(empty)'}")
-            print(f"    registration_id : {r['registration_id'] or '(empty)'}")
-            print(f"    last_heartbeat  : {r['last_heartbeat_at'] or '-'}"
-                  f"    updated : {r['updated_at']}")
-            print("    " + "-" * 68)
-
-    return 0
-
-
-if __name__ == "__main__":
-    raise SystemExit(main())

scripts/test_push_invite_order_reward.bat

@@ -1,9 +0,0 @@
-@echo off
-REM Push test #12 invite_order_reward (1 notification per run, random amount).
-REM Works from ANY directory (locates project root + venv python by itself):
-REM   scripts\test_push_invite_order_reward.bat
-REM Extra args pass through, e.g.:  test_push_invite_order_reward.bat --invitee-phone 12000000001
-REM ASCII-only comments: cmd parses .bat in the console codepage (GBK), UTF-8
-REM Chinese here gets mangled into bogus commands.
-cd /d "%~dp0.."
-".venv\Scripts\python.exe" "scripts\test_push_invite_order_reward.py" %*

scripts/test_push_invite_order_reward.py

@@ -1,110 +0,0 @@
-"""#12 好友下单到账(invite_order_reward)推送联调脚本 —— 每次执行只发 1 条,金额随机。
-
-后台没有驱动这个事件的入口(真实链路要好友注册 + 完成首次比价);本脚本直接调
-services/notification_events.notify_invite_order_reward —— 与生产同一条下发链路:
-落 notification 表(站内消息)+ 向【邀请人】全部已注册厂商 token 直推(honor/huawei/xiaomi/oppo/vivo)。
-
-可重复性:
-  默认用随机假「被邀请人 id」做 dedup_key → 永不去重,想跑多少次都行(昵称兜底显示「好友」)。
-  金额默认每次随机(0.01 ~ 99.99 元)→ 手机上按金额认出这条通知;--cents 可固定(线上真实值 200 = 2 元)。
-  带 --invitee-phone 指定真实用户(如种子好友 12000000001 柚子)→ 通知里显示真实昵称;
-  ⚠️ 但 dedup_key = 被邀请人 id:上一条还未读时重复发会命中去重(日志出现 dedup hit,不落库不推送),
-  在 App 里把那条读掉(或换号)即可再次触发。
-
-用法(服务端进程无需在跑,脚本自己连库 + 直调厂商接口):
-  .venv\\Scripts\\python.exe scripts\\test_push_invite_order_reward.py                        # 随机金额发 1 条
-  .venv\\Scripts\\python.exe scripts\\test_push_invite_order_reward.py --cents 200            # 固定 2.00 元
-  .venv\\Scripts\\python.exe scripts\\test_push_invite_order_reward.py --invitee-phone 12000000001  # 真实昵称
-
-结果判读(看输出日志):
-  push sent   = 厂商接口受理成功,手机应弹「好友下单奖励到账」通知
-  push failed = 厂商拒绝(原因见日志:token 失效/凭据错误等)
-  skip push   = 该厂商凭据未配置,只落站内消息
-站内消息用 11111111111 登录 App → 消息中心「好友邀请」可见;点击应跳邀请页。
-"""
-from __future__ import annotations
-
-import argparse
-import logging
-import random
-import sys
-
-from sqlalchemy import func, select
-
-from app.db.session import SessionLocal, engine
-from app.models.notification import Notification
-from app.repositories import device as device_repo
-from app.repositories import user as user_repo
-from app.services import notification_events
-
-# SQL 回显静音;shagua.* 开到 INFO 才看得到 push sent / push failed / dedup hit
-# dev 下 APP_DEBUG=true → engine echo=True,echo 走 InstanceLogger 直写、无视 logger level,只能关 echo 本身
-logging.basicConfig(level=logging.INFO, format="%(levelname)s %(name)s: %(message)s")
-logging.getLogger("sqlalchemy.engine").setLevel(logging.WARNING)
-engine.echo = False
-
-if hasattr(sys.stdout, "reconfigure"):
-    sys.stdout.reconfigure(encoding="utf-8")
-
-TYPE_KEY = "invite_order_reward"
-
-
-def _notif_count(db, uid: int) -> int:
-    return int(
-        db.execute(
-            select(func.count())
-            .select_from(Notification)
-            .where(Notification.user_id == uid, Notification.type == TYPE_KEY)
-        ).scalar_one()
-    )
-
-
-def main() -> None:
-    parser = argparse.ArgumentParser(description="#12 好友下单到账 推送联调(每次 1 条,金额默认随机)")
-    parser.add_argument("--phone", default="11111111111", help="邀请人(收通知方)手机号,默认 11111111111")
-    parser.add_argument("--cents", type=int, default=None,
-                        help="奖励金额,单位分(默认随机 1~9999;线上真实值 200)")
-    parser.add_argument("--invitee-phone", default="",
-                        help="被邀请人手机号(可选):显示真实昵称;不传用随机假 id,昵称显示「好友」")
-    args = parser.parse_args()
-
-    cents = args.cents if args.cents is not None else random.randint(1, 9999)
-
-    db = SessionLocal()
-    try:
-        user = user_repo.get_user_by_phone(db, args.phone)
-        if user is None:
-            print(f"❌ 用户 {args.phone} 不存在。先用该手机号在 App 登录一次再跑。")
-            return
-
-        if args.invitee_phone.strip():
-            invitee = user_repo.get_user_by_phone(db, args.invitee_phone.strip())
-            if invitee is None:
-                print(f"❌ 被邀请人 {args.invitee_phone} 不存在(种子好友可用 12000000001 柚子 / 12000000003 小美)。")
-                return
-            invitee_id = invitee.id
-        else:
-            invitee_id = random.randint(900000, 999999)  # 假 id,昵称兜底「好友」,永不去重
-
-        targets = device_repo.list_push_targets(db, user_id=user.id)
-        vendors = [t.push_vendor for t in targets]
-        print(f"邀请人 {args.phone}(id={user.id});推送设备 {len(targets)} 台:{vendors or '无 ← 手机收不到!先在 App 上报 push token'}")
-
-        before = _notif_count(db, user.id)
-        print(f"→ 本次奖励 【{cents / 100:.2f} 元】(invitee_user_id={invitee_id}),手机上按金额认领这条通知")
-        notification_events.notify_invite_order_reward(
-            db, inviter_user_id=user.id, invitee_user_id=invitee_id, cash_cents=cents
-        )
-
-        created = _notif_count(db, user.id) - before
-        if created == 1:
-            verdict = "✅ 已落库 1 条站内消息"
-        else:
-            verdict = "⚠️ 未落库(若日志有 dedup hit:该被邀请人上一条还未读,先在 App 里读掉再发)"
-        print(f"\n{verdict};推送成败见上方 shagua.notification_events 日志。")
-    finally:
-        db.close()
-
-
-if __name__ == "__main__":
-    main()

scripts/test_push_withdraw_failed.bat

@@ -1,9 +0,0 @@
-@echo off
-REM Push test #4 withdraw_failed (1 notification per run, random amount + reason).
-REM Works from ANY directory (locates project root + venv python by itself):
-REM   scripts\test_push_withdraw_failed.bat
-REM Extra args pass through, e.g.:  test_push_withdraw_failed.bat --cents 350
-REM ASCII-only comments: cmd parses .bat in the console codepage (GBK), UTF-8
-REM Chinese here gets mangled into bogus commands.
-cd /d "%~dp0.."
-".venv\Scripts\python.exe" "scripts\test_push_withdraw_failed.py" %*

scripts/test_push_withdraw_failed.py

@@ -1,106 +0,0 @@
-"""#4 提现失败(withdraw_failed)推送联调脚本 —— 每次执行只发 1 条,金额 + 失败原因随机。
-
-后台虽能驱动提现拒绝,但受「同一用户同时只能有 1 张待审单」约束,批量测试凑不齐单子;
-本脚本直接调 services/notification_events.notify_withdraw_failed —— 与生产同一条下发链路:
-落 notification 表(站内消息)+ 向该用户全部已注册厂商 token 直推(honor/huawei/xiaomi/oppo/vivo)。
-
-可重复性:每次执行用全新 uuid 单号做 dedup_key,永不命中「未读去重」,想跑多少次都行。
-金额默认每次随机(0.01 ~ 99.99 元)、失败原因随机 → 手机上按金额/原因就能认出这条通知;
---cents / --reason 可固定。
-
-用法(服务端进程无需在跑,脚本自己连库 + 直调厂商接口):
-  .venv\\Scripts\\python.exe scripts\\test_push_withdraw_failed.py                       # 随机金额+原因发 1 条
-  .venv\\Scripts\\python.exe scripts\\test_push_withdraw_failed.py --cents 350           # 固定 3.50 元
-  .venv\\Scripts\\python.exe scripts\\test_push_withdraw_failed.py --reason "自定义原因"  # 固定失败原因
-
-结果判读(看输出日志):
-  push sent   = 厂商接口受理成功,手机应弹「提现失败」通知
-  push failed = 厂商拒绝(原因见日志:token 失效/凭据错误等)
-  skip push   = 该厂商凭据未配置,只落站内消息
-站内消息用 11111111111 登录 App → 消息中心「提现助手」可见;点击应跳提现页(extra.withdrawId)。
-"""
-from __future__ import annotations
-
-import argparse
-import logging
-import random
-import sys
-import uuid
-
-from sqlalchemy import func, select
-
-from app.db.session import SessionLocal, engine
-from app.models.notification import Notification
-from app.models.wallet import WithdrawOrder
-from app.repositories import device as device_repo
-from app.repositories import user as user_repo
-from app.services import notification_events
-
-# SQL 回显静音;shagua.* 开到 INFO 才看得到 push sent / push failed
-# dev 下 APP_DEBUG=true → engine echo=True,echo 走 InstanceLogger 直写、无视 logger level,只能关 echo 本身
-logging.basicConfig(level=logging.INFO, format="%(levelname)s %(name)s: %(message)s")
-logging.getLogger("sqlalchemy.engine").setLevel(logging.WARNING)
-engine.echo = False
-
-if hasattr(sys.stdout, "reconfigure"):
-    sys.stdout.reconfigure(encoding="utf-8")
-
-TYPE_KEY = "withdraw_failed"
-
-# 失败原因池:不带 --reason 时随机抽一条,模拟不同失败场景(文案与 /withdraw/status 用户可读口径一致)
-FAIL_REASONS = [
-    "微信零钱未实名,款项已退回",
-    "收款账户异常,款项已退回",
-    "超出微信零钱收款限额,款项已退回",
-    "微信实名与提现实名不一致,款项已原路退回现金余额",
-]
-
-
-def _notif_count(db, uid: int) -> int:
-    return int(
-        db.execute(
-            select(func.count())
-            .select_from(Notification)
-            .where(Notification.user_id == uid, Notification.type == TYPE_KEY)
-        ).scalar_one()
-    )
-
-
-def main() -> None:
-    parser = argparse.ArgumentParser(description="#4 提现失败 推送联调(每次 1 条,金额/原因默认随机)")
-    parser.add_argument("--phone", default="11111111111", help="目标用户手机号(默认 11111111111)")
-    parser.add_argument("--cents", type=int, default=None, help="退回金额,单位分(默认随机 1~9999)")
-    parser.add_argument("--reason", default="", help="失败原因(默认从内置原因池随机抽)")
-    args = parser.parse_args()
-
-    cents = args.cents if args.cents is not None else random.randint(1, 9999)
-    reason = args.reason.strip() or random.choice(FAIL_REASONS)
-
-    db = SessionLocal()
-    try:
-        user = user_repo.get_user_by_phone(db, args.phone)
-        if user is None:
-            print(f"❌ 用户 {args.phone} 不存在。先用该手机号在 App 登录一次再跑。")
-            return
-
-        targets = device_repo.list_push_targets(db, user_id=user.id)
-        vendors = [t.push_vendor for t in targets]
-        print(f"目标用户 {args.phone}(id={user.id});推送设备 {len(targets)} 台:{vendors or '无 ← 手机收不到!先在 App 上报 push token'}")
-
-        before = _notif_count(db, user.id)
-        order = WithdrawOrder(
-            user_id=user.id, out_bill_no=uuid.uuid4().hex,
-            amount_cents=cents, source="coin_cash", fail_reason=reason,
-        )
-        print(f"→ 本次金额 【{cents / 100:.2f} 元】,原因【{reason}】(单号 {order.out_bill_no[:8]}…)")
-        notification_events.notify_withdraw_failed(db, order)
-
-        created = _notif_count(db, user.id) - before
-        verdict = "✅ 已落库 1 条站内消息" if created == 1 else "⚠️ 未落库(见上方日志)"
-        print(f"\n{verdict};推送成败见上方 shagua.notification_events 日志。")
-    finally:
-        db.close()
-
-
-if __name__ == "__main__":
-    main()

scripts/test_push_withdraw_success.bat

@@ -1,9 +0,0 @@
-@echo off
-REM Push test #3 withdraw_success (1 notification per run, random amount).
-REM Works from ANY directory (locates project root + venv python by itself):
-REM   scripts\test_push_withdraw_success.bat
-REM Extra args pass through, e.g.:  test_push_withdraw_success.bat --cents 1280
-REM ASCII-only comments: cmd parses .bat in the console codepage (GBK), UTF-8
-REM Chinese here gets mangled into bogus commands.
-cd /d "%~dp0.."
-".venv\Scripts\python.exe" "scripts\test_push_withdraw_success.py" %*

scripts/test_push_withdraw_success.py

@@ -1,94 +0,0 @@
-"""#3 提现到账(withdraw_success)推送联调脚本 —— 每次执行只发 1 条,金额随机。
-
-本环境 wxpay 未配置,后台审核通过发不出微信转账,打不通真实提现链路;本脚本直接调
-services/notification_events.notify_withdraw_success —— 与生产同一条下发链路:
-落 notification 表(站内消息)+ 向该用户全部已注册厂商 token 直推(honor/huawei/xiaomi/oppo/vivo)。
-
-可重复性:每次执行用全新 uuid 单号做 dedup_key,永不命中「未读去重」,想跑多少次都行。
-金额默认每次随机(0.01 ~ 99.99 元)→ 手机上按金额就能认出这条通知是哪次跑出来的;--cents 可固定。
-
-用法(服务端进程无需在跑,脚本自己连库 + 直调厂商接口):
-  .venv\\Scripts\\python.exe scripts\\test_push_withdraw_success.py               # 随机金额发 1 条
-  .venv\\Scripts\\python.exe scripts\\test_push_withdraw_success.py --cents 1280  # 固定 12.80 元
-
-结果判读(看输出日志):
-  push sent   = 厂商接口受理成功,手机应弹「提现到账」通知
-  push failed = 厂商拒绝(原因见日志:token 失效/凭据错误等)
-  skip push   = 该厂商凭据未配置,只落站内消息
-站内消息用 11111111111 登录 App → 消息中心「提现助手」可见。
-"""
-from __future__ import annotations
-
-import argparse
-import logging
-import random
-import sys
-import uuid
-
-from sqlalchemy import func, select
-
-from app.db.session import SessionLocal, engine
-from app.models.notification import Notification
-from app.models.wallet import WithdrawOrder
-from app.repositories import device as device_repo
-from app.repositories import user as user_repo
-from app.services import notification_events
-
-# SQL 回显静音;shagua.* 开到 INFO 才看得到 push sent / push failed
-# dev 下 APP_DEBUG=true → engine echo=True,echo 走 InstanceLogger 直写、无视 logger level,只能关 echo 本身
-logging.basicConfig(level=logging.INFO, format="%(levelname)s %(name)s: %(message)s")
-logging.getLogger("sqlalchemy.engine").setLevel(logging.WARNING)
-engine.echo = False
-
-if hasattr(sys.stdout, "reconfigure"):
-    sys.stdout.reconfigure(encoding="utf-8")
-
-TYPE_KEY = "withdraw_success"
-
-
-def _notif_count(db, uid: int) -> int:
-    return int(
-        db.execute(
-            select(func.count())
-            .select_from(Notification)
-            .where(Notification.user_id == uid, Notification.type == TYPE_KEY)
-        ).scalar_one()
-    )
-
-
-def main() -> None:
-    parser = argparse.ArgumentParser(description="#3 提现到账 推送联调(每次 1 条,金额默认随机)")
-    parser.add_argument("--phone", default="11111111111", help="目标用户手机号(默认 11111111111)")
-    parser.add_argument("--cents", type=int, default=None, help="到账金额,单位分(默认随机 1~9999)")
-    args = parser.parse_args()
-
-    cents = args.cents if args.cents is not None else random.randint(1, 9999)
-
-    db = SessionLocal()
-    try:
-        user = user_repo.get_user_by_phone(db, args.phone)
-        if user is None:
-            print(f"❌ 用户 {args.phone} 不存在。先用该手机号在 App 登录一次再跑。")
-            return
-
-        targets = device_repo.list_push_targets(db, user_id=user.id)
-        vendors = [t.push_vendor for t in targets]
-        print(f"目标用户 {args.phone}(id={user.id});推送设备 {len(targets)} 台:{vendors or '无 ← 手机收不到!先在 App 上报 push token'}")
-
-        before = _notif_count(db, user.id)
-        order = WithdrawOrder(
-            user_id=user.id, out_bill_no=uuid.uuid4().hex,
-            amount_cents=cents, source="coin_cash",
-        )
-        print(f"→ 本次金额 【{cents / 100:.2f} 元】(单号 {order.out_bill_no[:8]}…),手机上按金额认领这条通知")
-        notification_events.notify_withdraw_success(db, order)
-
-        created = _notif_count(db, user.id) - before
-        verdict = "✅ 已落库 1 条站内消息" if created == 1 else "⚠️ 未落库(见上方日志)"
-        print(f"\n{verdict};推送成败见上方 shagua.notification_events 日志。")
-    finally:
-        db.close()
-
-
-if __name__ == "__main__":
-    main()

tests/test_device_push.py

@@ -1,317 +0,0 @@
-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_notification_events.py

@@ -1,358 +0,0 @@
-"""业务事件 → 站内通知 + 厂商推送 联动测试(services/notification_events)。
-
-覆盖 PRD 六类真实触发:
-  #3 提现成功 / #4 提现失败(含审核拒绝) / #9 官方回复 / #10 反馈奖励 /
-  #11 爆料审核通过 / #12 好友下单到账。
-厂商推送不真发:测试环境无凭据默认跳过;推送链路用 monkeypatch 捕获/注错验证
-「有设备则推、推挂了业务不受影响」。wxpay 网络调用照旧全 monkeypatch。
-"""
-from __future__ import annotations
-
-import pytest
-from fastapi.testclient import TestClient
-from sqlalchemy import select
-
-from app.admin.main import admin_app
-from app.admin.repositories import admin_user as admin_repo
-from app.core.rewards import INVITE_COMPARE_REWARD_CENTS, PRICE_REPORT_REWARD_COINS
-from app.core.security import decode_token, hash_password
-from app.db.session import SessionLocal
-from app.models.feedback import Feedback
-from app.models.price_report import PriceReport
-from app.models.wallet import CoinAccount, WithdrawOrder
-from app.repositories import device as device_repo
-from app.repositories import wallet as crud_wallet
-from app.services import notification_events
-
-# ===== 用户侧 helpers(同 test_withdraw / test_notifications)=====
-
-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 _uid(token: str) -> int:
-    return int(decode_token(token, expected_type="access")["sub"])
-
-
-def _notifications(client: TestClient, token: str) -> list[dict]:
-    r = client.get("/api/v1/notifications?pageSize=50", headers=_auth(token))
-    assert r.status_code == 200, r.text
-    return r.json()["items"]
-
-
-def _seed_cash(client: TestClient, token: str, cents: int) -> None:
-    """先访问 /account 触发建账户,再直接灌现金余额。"""
-    client.get("/api/v1/wallet/account", headers=_auth(token))
-    with SessionLocal() as db:
-        acc = db.get(CoinAccount, _uid(token))
-        acc.cash_balance_cents = cents
-        db.commit()
-
-
-def _create_withdraw(client: TestClient, token: str, monkeypatch, cents: int = 50) -> str:
-    """绑微信 + 发起提现(进入 reviewing),返回 out_bill_no。"""
-    monkeypatch.setattr(
-        "app.integrations.wxpay.code_to_userinfo",
-        lambda code: {"openid": f"openid_{_uid(token)}", "nickname": None, "avatar_url": None, "raw": {}},
-    )
-    _seed_cash(client, token, cents * 2)
-    client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token))
-    r = client.post("/api/v1/wallet/withdraw", json={"amount_cents": cents}, headers=_auth(token))
-    assert r.status_code == 200, r.text
-    assert r.json()["status"] == "reviewing"
-    return r.json()["out_bill_no"]
-
-
-# ===== admin 侧 helpers(同 test_admin_write)=====
-
-@pytest.fixture()
-def admin_client() -> TestClient:
-    return TestClient(admin_app)
-
-
-@pytest.fixture()
-def operator_token() -> str:
-    with SessionLocal() as db:
-        a = admin_repo.get_by_username(db, "ne_operator")
-        if a is None:
-            admin_repo.create_admin(db, username="ne_operator", password="pass1234", role="operator")
-        else:
-            a.password_hash = hash_password("pass1234")
-            a.role = "operator"
-            a.status = "active"
-            db.commit()
-    c = TestClient(admin_app)
-    return c.post(
-        "/admin/api/auth/login", json={"username": "ne_operator", "password": "pass1234"}
-    ).json()["access_token"]
-
-
-def _seed_feedback(uid: int) -> int:
-    with SessionLocal() as db:
-        fb = Feedback(user_id=uid, content="比价按钮找不到", contact="wx123", status="pending")
-        db.add(fb)
-        db.commit()
-        return fb.id
-
-
-def _seed_price_report(uid: int, store: str = "蜀大侠火锅") -> int:
-    with SessionLocal() as db:
-        rep = PriceReport(
-            user_id=uid,
-            store_name=store,
-            reported_platform_id="mt",
-            reported_platform_name="美团",
-            reported_price_cents=990,
-            images=[],
-            status="pending",
-        )
-        db.add(rep)
-        db.commit()
-        return rep.id
-
-
-# ===== #4 提现失败(审核拒绝路径,_refund_withdraw 收口)=====
-
-def test_withdraw_reject_creates_failed_notification(client: TestClient, monkeypatch) -> None:
-    token = _login(client, "13800005001")
-    bill = _create_withdraw(client, token, monkeypatch)
-
-    with SessionLocal() as db:
-        crud_wallet.reject_withdraw(db, bill, "微信零钱未实名")
-
-    items = _notifications(client, token)
-    failed = [i for i in items if i["type"] == "withdraw_failed"]
-    assert len(failed) == 1
-    n = failed[0]
-    assert n["cashCents"] == 50
-    assert n["cashYuan"] == "0.50"
-    assert n["title"] == "提现失败，款项已退回"
-    assert n["extra"]["withdrawId"] == bill
-    assert n["isRead"] is False
-    rows = {r["label"]: r["value"] for r in n["infoRows"]}
-    assert rows["失败原因"] == "微信零钱未实名"
-    assert rows["退回说明"] == "款项已原路退回现金余额"
-
-
-def test_withdraw_failed_event_dedup_single_unread(client: TestClient, monkeypatch) -> None:
-    """同一提现单重复触发失败事件(并发查单等)→ 未读期间只落一条。"""
-    token = _login(client, "13800005002")
-    bill = _create_withdraw(client, token, monkeypatch)
-    with SessionLocal() as db:
-        crud_wallet.reject_withdraw(db, bill, "审核未通过")
-        order = db.execute(
-            select(WithdrawOrder).where(WithdrawOrder.out_bill_no == bill)
-        ).scalar_one()
-        notification_events.notify_withdraw_failed(db, order)  # 人为重复触发
-
-    items = [i for i in _notifications(client, token) if i["type"] == "withdraw_failed"]
-    assert len(items) == 1
-
-
-# ===== #3 提现成功(查单归一化路径)=====
-
-def test_withdraw_success_notification_on_status_query(client: TestClient, monkeypatch) -> None:
-    token = _login(client, "13800005003")
-    monkeypatch.setattr(
-        "app.integrations.wxpay.create_transfer",
-        lambda openid, amount_fen, out_bill_no, user_name=None: {
-            "status_code": 200,
-            "data": {"state": "WAIT_USER_CONFIRM", "package_info": "pkg", "transfer_bill_no": "tb"},
-        },
-    )
-    bill = _create_withdraw(client, token, monkeypatch)
-    with SessionLocal() as db:
-        crud_wallet.approve_withdraw(db, bill)  # 审核通过 → 转账进 pending(等用户确认)
-
-    assert [i for i in _notifications(client, token) if i["type"] == "withdraw_success"] == []
-
-    # 用户确认后查单 → SUCCESS → success + 下发「提现到账」通知
-    monkeypatch.setattr(
-        "app.integrations.wxpay.query_transfer",
-        lambda out_bill_no: {"status_code": 200, "data": {"state": "SUCCESS"}},
-    )
-    r = client.get("/api/v1/wallet/withdraw/status", params={"out_bill_no": bill}, headers=_auth(token))
-    assert r.json()["status"] == "success"
-
-    ok = [i for i in _notifications(client, token) if i["type"] == "withdraw_success"]
-    assert len(ok) == 1
-    n = ok[0]
-    assert n["cashCents"] == 50
-    assert n["actionText"] is None  # PRD:提现成功卡无操作行
-    rows = {r["label"]: r["value"] for r in n["infoRows"]}
-    assert rows["到账账户"] == "微信钱包"
-    assert "到账时间" in rows
-
-    # 再查一次(已终态,早退)→ 不重复下发
-    client.get("/api/v1/wallet/withdraw/status", params={"out_bill_no": bill}, headers=_auth(token))
-    assert len([i for i in _notifications(client, token) if i["type"] == "withdraw_success"]) == 1
-
-
-# ===== #10 反馈奖励(admin 采纳发金币)=====
-
-def test_feedback_approve_sends_reward_notification(
-    client: TestClient, admin_client: TestClient, operator_token: str
-) -> None:
-    token = _login(client, "13800005004")
-    fb_id = _seed_feedback(_uid(token))
-
-    r = admin_client.post(
-        f"/admin/api/feedbacks/{fb_id}/approve",
-        json={"reward_coins": 300, "note": "好建议", "reply": "问题已修复上线，送您的金币请查收～"},
-        headers=_auth(operator_token),
-    )
-    assert r.status_code == 200, r.text
-
-    items = [i for i in _notifications(client, token) if i["type"] == "feedback_reward"]
-    assert len(items) == 1
-    n = items[0]
-    assert n["coins"] == 300
-    assert n["extra"]["feedbackId"] == str(fb_id)
-    rows = {r["label"]: r["value"] for r in n["infoRows"]}
-    assert rows["官方留言"] == "问题已修复上线，送您的金币请查收～"  # PRD:发奖必带官方留言
-    assert "到账时间" in rows
-
-
-# ===== #9 官方回复(admin 拒绝,原因/留言用户可见)=====
-
-def test_feedback_reject_sends_reply_notification(
-    client: TestClient, admin_client: TestClient, operator_token: str
-) -> None:
-    token = _login(client, "13800005005")
-    fb_id = _seed_feedback(_uid(token))
-
-    r = admin_client.post(
-        f"/admin/api/feedbacks/{fb_id}/reject",
-        json={"reason": "无法复现", "reply": "麻烦补个录屏,我们再看看~"},
-        headers=_auth(operator_token),
-    )
-    assert r.status_code == 200, r.text
-
-    items = [i for i in _notifications(client, token) if i["type"] == "feedback_reply"]
-    assert len(items) == 1
-    n = items[0]
-    assert n["title"] == "傻瓜比价官方回复了您的反馈"
-    assert n["extra"]["feedbackId"] == str(fb_id)
-    assert n["coins"] is None
-
-
-# ===== #11 爆料审核通过(admin 通过发固定金币)=====
-
-def test_price_report_approve_sends_notification(
-    client: TestClient, admin_client: TestClient, operator_token: str
-) -> None:
-    token = _login(client, "13800005006")
-    rep_id = _seed_price_report(_uid(token), store="蜀大侠火锅")
-
-    r = admin_client.post(
-        f"/admin/api/price-reports/{rep_id}/approve", headers=_auth(operator_token)
-    )
-    assert r.status_code == 200, r.text
-
-    items = [i for i in _notifications(client, token) if i["type"] == "report_approved"]
-    assert len(items) == 1
-    n = items[0]
-    assert n["coins"] == PRICE_REPORT_REWARD_COINS
-    assert n["extra"]["reportId"] == str(rep_id)
-    rows = {r["label"]: r["value"] for r in n["infoRows"]}
-    assert "蜀大侠火锅" in rows["奖励说明"]
-
-
-# ===== #12 好友下单到账(好友首次成功比价 → 通知邀请人)=====
-
-def test_invite_compare_reward_sends_notification(client: TestClient) -> None:
-    a = _login(client, "13800005007")
-    b = _login(client, "13800005008")
-    code = client.get("/api/v1/invite/me", headers=_auth(a)).json()["invite_code"]
-    client.post("/api/v1/invite/bind", json={"invite_code": code}, headers=_auth(b))
-
-    r = client.post(
-        "/api/v1/compare/record",
-        json={"trace_id": "trace-notif-1", "status": "success"},
-        headers=_auth(b),
-    )
-    assert r.status_code == 200, r.text
-
-    items = [i for i in _notifications(client, a) if i["type"] == "invite_order_reward"]
-    assert len(items) == 1
-    n = items[0]
-    assert n["cashCents"] == INVITE_COMPARE_REWARD_CENTS
-    assert n["extra"]["inviteeNickname"]  # 好友昵称兜底(昵称/尾号/「好友」)非空
-    # 被邀请人自己不收该通知
-    assert [i for i in _notifications(client, b) if i["type"] == "invite_order_reward"] == []
-
-    # 好友再比价 → 不再发奖也不再通知(发奖幂等 + 通知 dedup 双保险)
-    client.post(
-        "/api/v1/compare/record",
-        json={"trace_id": "trace-notif-2", "status": "success"},
-        headers=_auth(b),
-    )
-    assert len([i for i in _notifications(client, a) if i["type"] == "invite_order_reward"]) == 1
-
-
-# ===== 厂商推送联动(有设备则推;推送失败不伤业务)=====
-
-def _register_device(uid: int, vendor: str = "xiaomi", token: str = "regid-1") -> None:
-    with SessionLocal() as db:
-        device_repo.register_or_update(
-            db, user_id=uid, device_id=f"dev_{uid}", push_vendor=vendor, push_token=token
-        )
-
-
-def test_push_sent_to_registered_device(client: TestClient, monkeypatch) -> None:
-    token = _login(client, "13800005009")
-    _register_device(_uid(token))
-
-    sent: list[dict] = []
-    monkeypatch.setattr(notification_events.vendor_push, "missing_settings", lambda vendor: [])
-
-    def _capture(vendor, push_token, *, title, body, extras=None, mock=False):
-        sent.append({"vendor": vendor, "token": push_token, "title": title, "body": body, "extras": extras})
-        return {"ok": True}
-
-    monkeypatch.setattr(notification_events.vendor_push, "send_notification", _capture)
-
-    bill = _create_withdraw(client, token, monkeypatch)
-    with SessionLocal() as db:
-        crud_wallet.reject_withdraw(db, bill, "微信零钱未实名")
-
-    assert len(sent) == 1
-    p = sent[0]
-    assert p["vendor"] == "xiaomi" and p["token"] == "regid-1"
-    assert p["title"] == "提现失败，款项已退回"
-    assert "0.50" in p["body"] and "微信零钱未实名" in p["body"]
-    # PRD §4 push 已读联动:extras 带 type + notificationId + 跳转参数
-    assert p["extras"]["type"] == "withdraw_failed"
-    assert p["extras"]["withdrawId"] == bill
-    nid = int(p["extras"]["notificationId"])
-    assert any(i["id"] == nid for i in _notifications(client, token))
-
-
-def test_push_failure_does_not_break_business(client: TestClient, monkeypatch) -> None:
-    """推送炸了(哪怕不是 VendorPushError)→ 提现拒绝照常退款,站内消息照常落库。"""
-    token = _login(client, "13800005010")
-    _register_device(_uid(token), token="regid-2")
-
-    monkeypatch.setattr(notification_events.vendor_push, "missing_settings", lambda vendor: [])
-
-    def _boom(*args, **kwargs):
-        raise RuntimeError("vendor api down")
-
-    monkeypatch.setattr(notification_events.vendor_push, "send_notification", _boom)
-
-    bill = _create_withdraw(client, token, monkeypatch)
-    with SessionLocal() as db:
-        crud_wallet.reject_withdraw(db, bill, "审核未通过")
-
-    r = client.get("/api/v1/wallet/withdraw/status", params={"out_bill_no": bill}, headers=_auth(token))
-    assert r.json()["status"] == "rejected"  # 业务不受影响
-    r = client.get("/api/v1/wallet/account", headers=_auth(token))
-    assert r.json()["cash_balance_cents"] == 100  # 已退回(seed 100 扣 50 退 50)
-    assert len([i for i in _notifications(client, token) if i["type"] == "withdraw_failed"]) == 1

tests/test_notifications.py

@@ -1,251 +0,0 @@
-"""消息通知中心 3 接口(落库版)。
-
-覆盖:空列表(虚拟数据已清除)、列表字段/派生/排序/分页、未读角标、标记已读(ids / all /
-幂等 / 参数校验)、鉴权与用户隔离、去重键部分唯一索引。数据直接写 notification 表
-(repositories/notification),不再有内存 mock。
-"""
-from __future__ import annotations
-
-from datetime import datetime, timedelta, timezone
-
-import pytest
-from fastapi.testclient import TestClient
-from sqlalchemy.exc import IntegrityError
-
-from app.core.security import decode_token
-from app.db.session import SessionLocal
-from app.repositories import notification as notif_repo
-
-_CST = timezone(timedelta(hours=8))
-
-
-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 _uid(token: str) -> int:
-    return int(decode_token(token, expected_type="access")["sub"])
-
-
-def _seed_sample(token: str, type_key: str):
-    """按类型插一条样例通知(复用 repo 的样例卡片内容),返回落库行。"""
-    with SessionLocal() as db:
-        return notif_repo.insert_sample(db, _uid(token), type_key)
-
-
-def _seed(token: str, type_key: str, **kw):
-    """按显式内容插一条通知(排序/分页用,可指定 sent_at)。"""
-    with SessionLocal() as db:
-        return notif_repo.create_notification(db, user_id=_uid(token), type_key=type_key, **kw)
-
-
-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_fresh_user_has_no_notifications(client: TestClient) -> None:
-    """虚拟数据已清除:新用户初始为空列表、未读 0、角标隐藏。"""
-    token = _login(client, "13900010000")
-    data = _fetch_all(client, token)
-    assert data["items"] == []
-    assert data["total"] == 0
-    assert data["unreadCount"] == 0
-    assert data["hasMore"] is False
-
-    r = client.get("/api/v1/notifications/unread-count", headers=_auth(token))
-    assert r.json() == {"count": 0, "badgeText": None}
-
-
-def test_list_item_fields_camel_case_and_derived(client: TestClient) -> None:
-    token = _login(client, "13900010001")
-    _seed_sample(token, "reward_expiring")
-    _seed_sample(token, "withdraw_success")
-    _seed_sample(token, "perm_accessibility")
-    _seed_sample(token, "feedback_reward")
-
-    items = _fetch_all(client, token)["items"]
-    assert len(items) == 4
-
-    # 字段按 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"]  # 落库后仍恒带 +08:00
-
-    by_type = {i["type"]: i for i in items}
-
-    # 双金额卡:金币整数 + 现金两位小数;category/cardStyle/title/actionText 均由 catalog 派生
-    expiring = by_type["reward_expiring"]
-    assert expiring["category"] == "withdraw_assistant"
-    assert expiring["categoryLabel"] == "提现助手"
-    assert expiring["cardStyle"] == "dual_amount"
-    assert expiring["title"] == "金币现金奖励即将失效"
-    assert expiring["actionText"] == "立即激活您的收益"
-    assert isinstance(expiring["coins"], int)
-    assert expiring["cashCents"] == 1280
-    assert expiring["cashYuan"] == "12.80"
-    assert [row["label"] for row in expiring["infoRows"]] == ["过期说明", "过期时间"]
-
-    # 提现成功卡:无操作行、无金币
-    ws = by_type["withdraw_success"]
-    assert ws["actionText"] is None
-    assert ws["coins"] is None
-
-    # 权限异常卡带 permission 参数(客户端点击时实时检测用)
-    assert by_type["perm_accessibility"]["extra"] == {"permission": "accessibility"}
-
-    # 反馈奖励卡:官方留言必填(PRD §3)
-    reward = by_type["feedback_reward"]
-    assert any(row["label"] == "官方留言" and row["value"] for row in reward["infoRows"])
-
-    # 新插入默认未读
-    assert all(i["isRead"] is False for i in items)
-
-
-def test_list_sorted_by_time_desc(client: TestClient) -> None:
-    """全列表 sent_at 倒序(最新在前),不分组。"""
-    token = _login(client, "13900010002")
-    base = datetime(2026, 7, 1, 12, 0, tzinfo=_CST)
-    _seed(token, "withdraw_success", sent_at=base - timedelta(days=2))
-    newest = _seed(token, "invite_order_reward", sent_at=base)
-    _seed(token, "feedback_reply", sent_at=base - timedelta(days=1))
-
-    items = _fetch_all(client, token)["items"]
-    sent_ats = [i["sentAt"] for i in items]
-    assert sent_ats == sorted(sent_ats, reverse=True), "最新在前"
-    assert items[0]["id"] == newest.id
-    assert items[0]["type"] == "invite_order_reward"
-
-
-def test_pagination(client: TestClient) -> None:
-    token = _login(client, "13900010003")
-    base = datetime(2026, 7, 1, 12, 0, tzinfo=_CST)
-    n = 12
-    for i in range(n):
-        _seed(token, "withdraw_success", sent_at=base - timedelta(minutes=i))
-
-    total = _fetch_all(client, token)["total"]
-    assert total == n
-
-    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")
-    for _ in range(3):
-        _seed_sample(token, "withdraw_success")
-
-    r = client.get("/api/v1/notifications/unread-count", headers=_auth(token))
-    assert r.json() == {"count": 3, "badgeText": "3"}
-
-    # 全部读完 → 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")
-    ids = [_seed_sample(token, "withdraw_success").id for _ in range(3)]
-    picked = 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": 1}
-
-    # 列表状态同步翻转
-    items = {i["id"]: i for i in _fetch_all(client, token)["items"]}
-    assert all(items[i]["isRead"] for i in picked)
-    assert items[ids[2]]["isRead"] is False
-
-    # 重复置读 + 不存在的 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"] == 1
-
-
-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_isolated_between_users(client: TestClient) -> None:
-    token_a = _login(client, "13900010007")
-    token_b = _login(client, "13900010008")
-    _seed_sample(token_a, "withdraw_success")
-    _seed_sample(token_b, "withdraw_success")
-
-    client.post("/api/v1/notifications/read", json={"all": True}, headers=_auth(token_a))
-    assert client.get("/api/v1/notifications/unread-count", headers=_auth(token_a)).json()["count"] == 0
-    assert (
-        client.get("/api/v1/notifications/unread-count", headers=_auth(token_b)).json()["count"] == 1
-    ), "A 清零不影响 B"
-
-
-def test_dedup_key_blocks_duplicate_unread(client: TestClient) -> None:
-    """同一 (user, type, dedup_key) 未读期间只允许一条(部分唯一索引拦重复未读)。"""
-    token = _login(client, "13900010009")
-    uid = _uid(token)
-    with SessionLocal() as db:
-        notif_repo.create_notification(
-            db, user_id=uid, type_key="perm_accessibility", dedup_key="accessibility"
-        )
-    # 同键第二条(仍未读)→ 唯一索引拦截
-    with pytest.raises(IntegrityError):
-        with SessionLocal() as db:
-            notif_repo.create_notification(
-                db, user_id=uid, type_key="perm_accessibility", dedup_key="accessibility"
-            )
-    # 只落了一条
-    assert _fetch_all(client, token)["total"] == 1

tests/test_push_center.py

@@ -1,454 +0,0 @@
-"""厂商推送(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
-    # (消息中心推送带 notificationId → data 额外补 notif_id/notif_type 点击路由别名,
-    #  点击时 HMS 把 data 键值对注入启动 intent,客户端首选这两个键落地)
-    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",
-        "notif_id": "90001",
-        "notif_type": "withdraw_success",
-    }
-
-
-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