接入厂商直推服务端并补充推送验收接口

- 新增荣耀、vivo、小米、OPPO 服务端推送分发
- 设备注册和心跳支持 push_vendor/push_token
- 增加 /api/v1/device/push-test 延迟验收接口
- 补充数据库迁移、配置示例和厂商推送测试
- 保留本分支已有微信登录联调改动

验证：python -m pytest tests/test_device_push.py tests/test_auth.py tests/test_health.py

.env.example

@@ -18,7 +18,7 @@ JWT_ACCESS_TOKEN_EXPIRE_MINUTES=120
 # refresh token 有效期(天),默认 30 天
 JWT_REFRESH_TOKEN_EXPIRE_DAYS=30
 
-# ===== 极光一键登录 =====
+# ===== 极光一键登录 / 短信 =====
 # 控制台 → 应用设置 → 应用信息 拿到
 JG_APP_KEY=
 JG_MASTER_SECRET=
@@ -27,7 +27,43 @@ JG_PRIVATE_KEY_PATH=./secrets/jverify_rsa_private.pem
 JG_VERIFY_ENDPOINT=https://api.verification.jpush.cn/v1/web/loginTokenVerify
 JG_REQUEST_TIMEOUT_SEC=15
 
-# ===== 无障碍保护存活监控(pull 后置检测;本期不接推送)=====
+# ===== 厂商直推(无障碍保护存活告警)=====
+# 敏感密钥只放 .env / 服务器环境变量,不要提交到 git。
+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
+
+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
+
+# ===== 无障碍保护存活监控(推送 + pull 后置兜底)=====
 HEARTBEAT_MONITOR_ENABLED=true
 HEARTBEAT_TIMEOUT_MINUTES=10
 HEARTBEAT_SCAN_INTERVAL_SEC=60
@@ -58,6 +94,13 @@ MT_CPS_DEFAULT_SID=sgbjia
 # 线上国内服务器留空(=直连)。留空且本机直连失败时 /feed、/coupons、/top-sales 会返回空。
 MT_CPS_PROXY=
 
+# ===== 京东联盟 CPS =====
+# 京东联盟/京东宙斯开放平台创建应用后填写。AUTH_KEY 是工具商授权 key,自有应用可留空。
+JD_UNION_APP_KEY=
+JD_UNION_APP_SECRET=
+JD_UNION_SITE_ID=
+JD_UNION_AUTH_KEY=
+
 # ===== Pricebot 上游 (领券/比价业务透传目标) =====
 # 客户端调本服务的 /api/v1/coupon/step 等,我们透传到 pricebot-backend。
 # 本地开发用 localhost:8000。生产部署改成内网地址(如 http://pricebot.internal:8000)。
@@ -82,8 +125,9 @@ INTERNAL_API_SECRET=
 # 逗号分隔,生产留空(只让 app 调,不开放 web)。本地开发可加 http://localhost:5173 之类
 CORS_ALLOW_ORIGINS=
 
-# ===== 微信支付(商家转账到零钱 / 提现)=====
-# appid/secret 来自微信开放平台移动应用;mch/序列号/公钥ID 来自微信支付商户平台。
+# ===== 微信开放平台 App + 微信支付(登录 / 商家转账到零钱 / 提现)=====
+# WECHAT_APP_ID/WECHAT_APP_SECRET 来自微信开放平台移动应用,用于 App 微信一键登录(code 换 openid)。
+# mch/序列号/公钥ID 来自微信支付商户平台,仅提现/转账需要。
 # 证书 .pem 放 secrets/(已 gitignore)。
 WECHAT_APP_ID=wxxxxxxxxxxxxxxxxx
 WECHAT_APP_SECRET=your_app_secret

alembic/versions/direct_vendor_push_fields.py

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

alembic/versions/jd_cps_order_fields.py

@@ -0,0 +1,56 @@
+"""add jd cps order fields
+
+Revision ID: jd_cps_order_fields
+Revises: 7db22acee504
+Create Date: 2026-06-28 20:30:00.000000
+"""
+from __future__ import annotations
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision = "jd_cps_order_fields"
+down_revision = "7db22acee504"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    with op.batch_alter_table("cps_order") as batch_op:
+        batch_op.add_column(
+            sa.Column("platform", sa.String(length=20), nullable=False, server_default="meituan")
+        )
+        batch_op.add_column(sa.Column("external_order_id", sa.String(length=128), nullable=True))
+        batch_op.add_column(sa.Column("external_row_id", sa.String(length=128), nullable=True))
+        batch_op.add_column(sa.Column("estimated_commission_cents", sa.Integer(), nullable=True))
+        batch_op.add_column(sa.Column("actual_commission_cents", sa.Integer(), nullable=True))
+        batch_op.add_column(sa.Column("jd_valid_code", sa.String(length=16), nullable=True))
+        batch_op.add_column(sa.Column("settle_month", sa.String(length=16), nullable=True))
+        batch_op.add_column(sa.Column("site_id", sa.String(length=128), nullable=True))
+        batch_op.add_column(sa.Column("position_id", sa.String(length=128), nullable=True))
+        batch_op.add_column(sa.Column("pid", sa.String(length=128), nullable=True))
+        batch_op.add_column(sa.Column("sub_union_id", sa.String(length=128), nullable=True))
+        batch_op.create_index("ix_cps_order_platform", ["platform"])
+        batch_op.create_index("ix_cps_order_external_order_id", ["external_order_id"])
+        batch_op.create_index("ix_cps_order_external_row_id", ["external_row_id"])
+        batch_op.create_index("ix_cps_order_jd_valid_code", ["jd_valid_code"])
+
+
+def downgrade() -> None:
+    with op.batch_alter_table("cps_order") as batch_op:
+        batch_op.drop_index("ix_cps_order_jd_valid_code")
+        batch_op.drop_index("ix_cps_order_external_row_id")
+        batch_op.drop_index("ix_cps_order_external_order_id")
+        batch_op.drop_index("ix_cps_order_platform")
+        batch_op.drop_column("sub_union_id")
+        batch_op.drop_column("pid")
+        batch_op.drop_column("position_id")
+        batch_op.drop_column("site_id")
+        batch_op.drop_column("settle_month")
+        batch_op.drop_column("jd_valid_code")
+        batch_op.drop_column("actual_commission_cents")
+        batch_op.drop_column("estimated_commission_cents")
+        batch_op.drop_column("external_row_id")
+        batch_op.drop_column("external_order_id")
+        batch_op.drop_column("platform")

app/admin/repositories/ad_revenue.py

@@ -20,7 +20,7 @@ report_date / reward_date 归日。
 """
 from __future__ import annotations
 
-from datetime import UTC, datetime, timedelta
+from datetime import datetime, timedelta, timezone
 from datetime import date as _date
 
 from sqlalchemy import select
@@ -36,7 +36,7 @@ from app.models.user import User
 def _cn_hour(dt: datetime) -> int:
     """created_at(UTC 口径)→ 北京时间小时(0–23)。naive 当 UTC 处理(sqlite),tz-aware 直接换算(pg)。"""
     if dt.tzinfo is None:
-        dt = dt.replace(tzinfo=UTC)
+        dt = dt.replace(tzinfo=timezone.utc)
     return dt.astimezone(rewards.CN_TZ).hour
 
 

app/admin/repositories/cps.py

@@ -7,13 +7,14 @@ from __future__ import annotations
 
 from datetime import datetime, timedelta, timezone
 from decimal import Decimal, InvalidOperation
+from typing import Any
 from uuid import uuid4
 
 from sqlalchemy import desc, func, select
 from sqlalchemy.orm import Session
 
 from app.admin.repositories.queries import _as_utc, offset_paginate
-from app.integrations import meituan
+from app.integrations import jd_union, meituan
 from app.repositories import cps_link as cps_link_repo
 from app.models.cps_activity import CpsActivity
 from app.models.cps_group import CpsGroup
@@ -24,6 +25,11 @@ from app.models.cps_wx_user import CpsWxUser
 # 美团订单状态:取消(4)/风控(5)不计佣金;结算(6)为佣金真正到账
 _INVALID_STATUS = {"4", "5"}
 _SETTLED_STATUS = "6"
+_JD_INVALID_CODES = {
+    "2", "3", "4", "5", "6", "7", "8", "9", "11", "13", "14", "19", "20", "21",
+    "22", "23", "25", "26", "27", "28", "29", "30", "31", "34", "35", "36",
+}
+_JD_UNPAID_CODES = {"15"}
 
 # CPS 点击时序按北京时区分桶(运营看的是北京时间)
 _BJ_TZ = timezone(timedelta(hours=8))
@@ -45,8 +51,58 @@ def _yuan_to_cents(v: object) -> int | None:
 
 def _ts_to_dt(ts: object) -> datetime | None:
     """秒级时间戳 → tz-aware UTC datetime(绝对时刻,前端按北京展示)。"""
-    if not ts:
+    if ts is None:
         return None
+    if isinstance(ts, datetime):
+        return ts if ts.tzinfo else ts.replace(tzinfo=_BJ_TZ).astimezone(timezone.utc)
+    s = str(ts).strip()
+    if not s or s.lower() == "null":
+        return None
+    try:
+        seconds = float(Decimal(s))
+    except (InvalidOperation, ValueError):
+        return None
+    if seconds == 0:
+        return None
+    # 美团文档是秒级时间戳，这里顺手兼容毫秒/微秒，避免上游格式变化导致时间再次落空。
+    if abs(seconds) > 10_000_000_000_000:
+        seconds /= 1_000_000
+    elif abs(seconds) > 10_000_000_000:
+        seconds /= 1_000
+    try:
+        return datetime.fromtimestamp(seconds, tz=timezone.utc)
+    except (OverflowError, OSError, ValueError):
+        return None
+
+
+def _jd_dt_to_utc(value: object) -> datetime | None:
+    """京东时间字符串(北京时间) → UTC aware datetime。"""
+    if value is None:
+        return None
+    s = str(value).strip()
+    if not s:
+        return None
+    for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d"):
+        try:
+            dt = datetime.strptime(s, fmt)
+            return dt.replace(tzinfo=_BJ_TZ).astimezone(timezone.utc)
+        except ValueError:
+            continue
+    return None
+
+
+def _text(value: object) -> str | None:
+    if value is None:
+        return None
+    s = str(value).strip()
+    return s or None
+
+
+def _pick(row: dict[str, Any], *keys: str) -> Any:
+    for key in keys:
+        if key in row and row[key] is not None:
+            return row[key]
+    return None
     try:
         return datetime.fromtimestamp(int(ts), tz=timezone.utc)
     except (ValueError, OSError, TypeError):
@@ -249,6 +305,80 @@ def _map_order_fields(r: dict) -> dict:
     }
 
 
+def _jd_order_key(r: dict[str, Any]) -> str | None:
+    row_id = _text(_pick(r, "id", "rowId", "orderRowId"))
+    if row_id:
+        return f"jd:{row_id}"
+    order_id = _text(_pick(r, "orderId", "parentOrderId"))
+    sku_id = _text(_pick(r, "skuId"))
+    if order_id and sku_id:
+        return f"jd:{order_id}:{sku_id}"
+    if order_id:
+        return f"jd:{order_id}"
+    return None
+
+
+def _map_jd_order_fields(r: dict[str, Any]) -> dict:
+    """京东 order.row.query 单条订单行 → CpsOrder 字段。"""
+    sku_name = _text(_pick(r, "skuName", "goodsName", "productName"))
+    if sku_name and len(sku_name) > 500:
+        sku_name = sku_name[:500]
+    valid_code = _text(_pick(r, "validCode", "valid_code"))
+    actual_fee = _yuan_to_cents(_pick(r, "actualFee", "actual_fee"))
+    estimate_fee = _yuan_to_cents(_pick(r, "estimateFee", "estimate_fee"))
+    commission = actual_fee if actual_fee not in (None, 0) else estimate_fee
+    order_time = _jd_dt_to_utc(_pick(r, "orderTime", "order_time"))
+    return {
+        "platform": "jd",
+        "external_order_id": _text(_pick(r, "orderId", "parentOrderId")),
+        "external_row_id": _text(_pick(r, "id", "rowId", "orderRowId")),
+        "sid": _text(_pick(r, "subUnionId", "sub_union_id")),
+        "act_id": None,
+        "biz_line": None,
+        "trade_type": None,
+        "pay_price_cents": _yuan_to_cents(
+            _pick(r, "actualCosPrice", "estimateCosPrice", "price")
+        ),
+        "commission_cents": commission,
+        "commission_rate": _text(_pick(r, "commissionRate", "commission_rate")),
+        "refund_price_cents": None,
+        "refund_profit_cents": None,
+        "estimated_commission_cents": estimate_fee,
+        "actual_commission_cents": actual_fee,
+        "mt_status": None,
+        "jd_valid_code": valid_code,
+        "invalid_reason": None if _is_jd_valid_code(valid_code) else f"validCode={valid_code}",
+        "product_name": sku_name,
+        "settle_month": _text(_pick(r, "payMonth", "settleMonth", "pay_month")),
+        "site_id": _text(_pick(r, "siteId", "site_id")),
+        "position_id": _text(_pick(r, "positionId", "position_id")),
+        "pid": _text(_pick(r, "pid")),
+        "sub_union_id": _text(_pick(r, "subUnionId", "sub_union_id")),
+        "pay_time": order_time,
+        "mt_update_time": _jd_dt_to_utc(_pick(r, "modifyTime", "updateTime", "modify_time"))
+        or order_time,
+        "raw": r,
+    }
+
+
+def _is_jd_valid_code(valid_code: str | None) -> bool:
+    code = str(valid_code).strip() if valid_code is not None else ""
+    return bool(code and code not in _JD_INVALID_CODES and code not in _JD_UNPAID_CODES)
+
+
+def is_jd_order_valid(order: CpsOrder) -> bool:
+    return _is_jd_valid_code(order.jd_valid_code)
+
+
+def effective_commission_cents(order: CpsOrder) -> int:
+    if order.platform == "jd":
+        if order.actual_commission_cents not in (None, 0):
+            return order.actual_commission_cents or 0
+        if order.estimated_commission_cents is not None:
+            return order.estimated_commission_cents or 0
+    return order.commission_cents or 0
+
+
 def reconcile_orders(
     db: Session, *, start_time: int, end_time: int,
     query_time_type: int = 1, sid: str | None = None, max_pages: int = 200,
@@ -274,6 +404,11 @@ def reconcile_orders(
                 continue
             fetched += 1
             fields = _map_order_fields(r)
+            fields.setdefault("platform", "meituan")
+            fields.setdefault("external_order_id", order_id)
+            fields.setdefault("external_row_id", None)
+            fields.setdefault("estimated_commission_cents", fields.get("commission_cents"))
+            fields.setdefault("actual_commission_cents", None)
             existing = db.execute(
                 select(CpsOrder).where(CpsOrder.order_id == order_id)
             ).scalar_one_or_none()
@@ -291,6 +426,56 @@ def reconcile_orders(
     return {"fetched": fetched, "inserted": inserted, "updated": updated, "pages": pages}
 
 
+def reconcile_jd_orders(
+    db: Session, *, start_time: datetime, end_time: datetime,
+    query_time_type: int = 3, max_pages: int = 100,
+) -> dict:
+    """调京东 order.row.query 拉单 → 按订单行 upsert。
+
+    京东单次查询窗口最多 1 小时,这里按北京自然时间切窗并逐页拉取。
+    """
+    fetched = inserted = updated = pages = 0
+    cur = start_time
+    while cur < end_time:
+        win_end = min(cur + timedelta(hours=1), end_time)
+        page = 1
+        while page <= max_pages:
+            resp = jd_union.query_order_rows(
+                start_time=cur,
+                end_time=win_end,
+                query_time_type=query_time_type,
+                page_index=page,
+                page_size=200,
+            )
+            rows = resp.get("rows") or []
+            has_more = bool(resp.get("has_more"))
+            if not rows:
+                break
+            pages += 1
+            for r in rows:
+                order_id = _jd_order_key(r)
+                if not order_id:
+                    continue
+                fetched += 1
+                fields = _map_jd_order_fields(r)
+                existing = db.execute(
+                    select(CpsOrder).where(CpsOrder.order_id == order_id)
+                ).scalar_one_or_none()
+                if existing is None:
+                    db.add(CpsOrder(order_id=order_id, **fields))
+                    inserted += 1
+                else:
+                    for k, v in fields.items():
+                        setattr(existing, k, v)
+                    updated += 1
+            if not has_more or len(rows) < 200:
+                break
+            page += 1
+        cur = win_end
+    db.commit()
+    return {"fetched": fetched, "inserted": inserted, "updated": updated, "pages": pages}
+
+
 def list_orders(
     db: Session, *, sid: str | None = None, mt_status: str | None = None,
     limit: int = 20, cursor: int | None = None,

app/admin/repositories/stats.py

@@ -13,6 +13,7 @@ from sqlalchemy.orm import Session
 
 from app.models.ad_feed_reward import AdFeedRewardRecord
 from app.models.ad_reward import AdRewardRecord
+from app.models.analytics_event import AnalyticsEvent
 from app.models.comparison import ComparisonRecord
 from app.models.coupon_state import CouponPromptEngagement
 from app.models.cps_order import CpsOrder
@@ -23,11 +24,13 @@ from app.models.user import User
 from app.models.wallet import CoinTransaction, WithdrawOrder
 
 _BEIJING = timezone(timedelta(hours=8))
-COUPON_REWARD_BIZ_TYPES = ("reward_video", "ad_reward", "coupon", "coupon_reward")
+REWARD_VIDEO_BIZ_TYPES = ("reward_video", "ad_reward")
+COUPON_REWARD_BIZ_TYPES = ("coupon", "coupon_reward")
 COMPARISON_REWARD_BIZ_TYPES = ("comparison", "compare_reward", "comparison_reward")
 EXCLUDED_REWARD_BIZ_TYPES = ("invite_inviter", "invite_invitee", "admin_grant")
 UNCLASSIFIED_FEED_BIZ_TYPES = ("feed_ad_reward",)
 REGULAR_TASK_EXCLUDED_BIZ_TYPES = (
+    *REWARD_VIDEO_BIZ_TYPES,
     *COUPON_REWARD_BIZ_TYPES,
     *COMPARISON_REWARD_BIZ_TYPES,
     *EXCLUDED_REWARD_BIZ_TYPES,
@@ -35,6 +38,13 @@ REGULAR_TASK_EXCLUDED_BIZ_TYPES = (
 )
 MEITUAN_CPS_INVALID_STATUSES = ("4", "5")
 MEITUAN_CPS_SETTLED_STATUS = "6"
+COMPARE_START_EVENT = "real_compare_start"
+COUPON_START_EVENT = "real_coupon_start"
+JD_CPS_INVALID_CODES = {
+    "2", "3", "4", "5", "6", "7", "8", "9", "11", "13", "14", "19", "20", "21",
+    "22", "23", "25", "26", "27", "28", "29", "30", "31", "34", "35", "36",
+}
+JD_CPS_UNPAID_CODES = {"15"}
 
 
 def _beijing_today_start_utc() -> datetime:
@@ -45,14 +55,31 @@ def _beijing_today_start_utc() -> datetime:
 
 
 def today_dau(db: Session) -> int:
-    """今日活跃用户数(DAU):北京时区今天 0 点后登录过(last_login_at)。
+    """今日活跃用户数(DAU):登录 + 开始比价 + 开始领券,按用户去重。
 
     广告收益报表复用这个函数;历史窗口 DAU 由 dashboard_overview 的 period 口径另算。
     """
+    today_bj = datetime.now(_BEIJING).date()
     today_start = _beijing_today_start_utc()
-    return int(
-        db.execute(select(func.count(User.id)).where(User.last_login_at >= today_start)).scalar_one()
+    tomorrow_start = today_start + timedelta(days=1)
+    login_user_ids = _id_set(
+        db,
+        select(User.id).where(User.last_login_at >= today_start, User.last_login_at < tomorrow_start),
     )
+    compare_start_user_ids = _event_user_ids(
+        db, (COMPARE_START_EVENT,), today_start, tomorrow_start
+    )
+    coupon_event_user_ids = _event_user_ids(
+        db, (COUPON_START_EVENT,), today_start, tomorrow_start
+    )
+    coupon_claim_user_ids = _id_set(
+        db,
+        select(CouponPromptEngagement.user_id).where(
+            CouponPromptEngagement.engage_date == today_bj,
+            CouponPromptEngagement.engage_type == "claim_started",
+        ),
+    )
+    return len(login_user_ids | compare_start_user_ids | coupon_event_user_ids | coupon_claim_user_ids)
 
 
 def _default_period_end() -> date:
@@ -91,6 +118,24 @@ def _date_range(date_from: date, date_to: date) -> list[date]:
     return [date_from + timedelta(days=i) for i in range(days + 1)]
 
 
+def _id_set(db: Session, stmt) -> set[int]:
+    return {int(v) for v in db.execute(stmt).scalars().all() if v is not None}
+
+
+def _event_user_ids(
+    db: Session, event_names: tuple[str, ...], start_utc: datetime, end_utc: datetime
+) -> set[int]:
+    return _id_set(
+        db,
+        select(AnalyticsEvent.user_id).where(
+            AnalyticsEvent.user_id.is_not(None),
+            AnalyticsEvent.event.in_(event_names),
+            AnalyticsEvent.created_at >= start_utc,
+            AnalyticsEvent.created_at < end_utc,
+        ),
+    )
+
+
 def _commission_rate_percent(raw: str | None) -> Decimal | None:
     """美团 commissionRate 原值: "300"=3%, "10"=0.1%;也兼容 "3%"。"""
     if raw is None:
@@ -107,6 +152,11 @@ def _commission_rate_percent(raw: str | None) -> Decimal | None:
     return val / Decimal("100")
 
 
+def _jd_valid_order(order: CpsOrder) -> bool:
+    code = str(order.jd_valid_code).strip() if order.jd_valid_code is not None else ""
+    return bool(code and code not in JD_CPS_INVALID_CODES and code not in JD_CPS_UNPAID_CODES)
+
+
 def dashboard_overview(
     db: Session, *, date_from: date | None = None, date_to: date | None = None
 ) -> dict:
@@ -216,17 +266,22 @@ def dashboard_overview(
     login_user_ids = _user_id_set(
         select(User.id).where(User.last_login_at >= start_utc, User.last_login_at < end_utc)
     )
-    compare_user_ids = _user_id_set(
-        select(ComparisonRecord.user_id).where(*period_comparison_conds)
+    compare_start_user_ids = _event_user_ids(
+        db, (COMPARE_START_EVENT,), start_utc, end_utc
     )
-    coupon_user_ids = _user_id_set(
+    coupon_event_user_ids = _event_user_ids(
+        db, (COUPON_START_EVENT,), start_utc, end_utc
+    )
+    coupon_claim_user_ids = _user_id_set(
         select(CouponPromptEngagement.user_id).where(
             CouponPromptEngagement.engage_date >= period_from,
             CouponPromptEngagement.engage_date <= period_to,
             CouponPromptEngagement.engage_type == "claim_started",
         )
     )
-    period_active_user_ids = login_user_ids | compare_user_ids | coupon_user_ids
+    period_active_user_ids = (
+        login_user_ids | compare_start_user_ids | coupon_event_user_ids | coupon_claim_user_ids
+    )
     period_retained_new_user_ids = period_new_user_ids & period_active_user_ids
     period_retention_rate = (
         round(len(period_retained_new_user_ids) / len(period_new_user_ids), 4)
@@ -248,10 +303,13 @@ def dashboard_overview(
                 User.last_login_at < day_end_utc,
             )
         )
-        daily_compare_user_ids = _user_id_set(
-            select(ComparisonRecord.user_id).where(*daily_comparison_conds)
+        daily_compare_start_user_ids = _event_user_ids(
+            db, (COMPARE_START_EVENT,), day_start_utc, day_end_utc
         )
-        daily_coupon_user_ids = _user_id_set(
+        daily_coupon_event_user_ids = _event_user_ids(
+            db, (COUPON_START_EVENT,), day_start_utc, day_end_utc
+        )
+        daily_coupon_claim_user_ids = _user_id_set(
             select(CouponPromptEngagement.user_id).where(
                 CouponPromptEngagement.engage_date == cur_date,
                 CouponPromptEngagement.engage_type == "claim_started",
@@ -261,7 +319,10 @@ def dashboard_overview(
             {
                 "date": cur_date,
                 "active_users": len(
-                    daily_login_user_ids | daily_compare_user_ids | daily_coupon_user_ids
+                    daily_login_user_ids
+                    | daily_compare_start_user_ids
+                    | daily_coupon_event_user_ids
+                    | daily_coupon_claim_user_ids
                 ),
                 "new_users": _count(
                     User,
@@ -280,13 +341,28 @@ def dashboard_overview(
     period_reward_video_coin_total = _sum(
         CoinTransaction.amount,
         *period_coin_conds,
-        CoinTransaction.biz_type.in_(("reward_video", "ad_reward")),
+        CoinTransaction.biz_type.in_(REWARD_VIDEO_BIZ_TYPES),
     )
     period_feed_ad_coin_total = _sum(
         CoinTransaction.amount,
         *period_coin_conds,
         CoinTransaction.biz_type == "feed_ad_reward",
     )
+    period_feed_ad_conds = (
+        AdFeedRewardRecord.created_at >= start_utc,
+        AdFeedRewardRecord.created_at < end_utc,
+        AdFeedRewardRecord.status == "granted",
+    )
+    period_coupon_feed_coin_total = _sum(
+        AdFeedRewardRecord.coin,
+        *period_feed_ad_conds,
+        AdFeedRewardRecord.feed_scene == "coupon",
+    )
+    period_comparison_feed_coin_total = _sum(
+        AdFeedRewardRecord.coin,
+        *period_feed_ad_conds,
+        AdFeedRewardRecord.feed_scene == "comparison",
+    )
     period_signin_coin_total = _sum(
         CoinTransaction.amount,
         *period_coin_conds,
@@ -306,18 +382,24 @@ def dashboard_overview(
         CoinTransaction.amount,
         *period_coin_conds,
         CoinTransaction.biz_type.in_(COUPON_REWARD_BIZ_TYPES),
-    )
+    ) + period_coupon_feed_coin_total
     period_comparison_reward_coin_total = _sum(
         CoinTransaction.amount,
         *period_coin_conds,
         CoinTransaction.biz_type.in_(COMPARISON_REWARD_BIZ_TYPES),
-    )
+    ) + period_comparison_feed_coin_total
     period_regular_task_coin_total = _sum(
         CoinTransaction.amount,
         *period_coin_conds,
         CoinTransaction.biz_type.notin_(REGULAR_TASK_EXCLUDED_BIZ_TYPES),
     )
-    period_meituan_orders = list(
+    period_classified_coin_total = (
+        period_coupon_reward_coin_total
+        + period_comparison_reward_coin_total
+        + period_reward_video_coin_total
+        + period_regular_task_coin_total
+    )
+    period_cps_orders = list(
         db.execute(
             select(CpsOrder).where(
                 CpsOrder.pay_time >= start_utc,
@@ -325,9 +407,17 @@ def dashboard_overview(
             )
         ).scalars()
     )
+    period_meituan_orders = [
+        o for o in period_cps_orders if (o.platform or "meituan") == "meituan"
+    ]
     period_meituan_valid_orders = [
         o for o in period_meituan_orders if o.mt_status not in MEITUAN_CPS_INVALID_STATUSES
     ]
+    period_jd_orders = [o for o in period_cps_orders if o.platform == "jd"]
+    period_jd_valid_orders = [o for o in period_jd_orders if _jd_valid_order(o)]
+    period_jd_invalid_orders = [
+        o for o in period_jd_orders if o.jd_valid_code and not _jd_valid_order(o)
+    ]
     period_meituan_hit_count = 0
     period_meituan_miss_count = 0
     period_meituan_unknown_rate_count = 0
@@ -346,6 +436,41 @@ def dashboard_overview(
         else None
     )
 
+    total_reward_video_coin_total = _sum(
+        CoinTransaction.amount,
+        CoinTransaction.amount > 0,
+        CoinTransaction.biz_type.in_(REWARD_VIDEO_BIZ_TYPES),
+    )
+    total_coupon_reward_coin_total = _sum(
+        CoinTransaction.amount,
+        CoinTransaction.amount > 0,
+        CoinTransaction.biz_type.in_(COUPON_REWARD_BIZ_TYPES),
+    ) + _sum(
+        AdFeedRewardRecord.coin,
+        AdFeedRewardRecord.status == "granted",
+        AdFeedRewardRecord.feed_scene == "coupon",
+    )
+    total_comparison_reward_coin_total = _sum(
+        CoinTransaction.amount,
+        CoinTransaction.amount > 0,
+        CoinTransaction.biz_type.in_(COMPARISON_REWARD_BIZ_TYPES),
+    ) + _sum(
+        AdFeedRewardRecord.coin,
+        AdFeedRewardRecord.status == "granted",
+        AdFeedRewardRecord.feed_scene == "comparison",
+    )
+    total_regular_task_coin_total = _sum(
+        CoinTransaction.amount,
+        CoinTransaction.amount > 0,
+        CoinTransaction.biz_type.notin_(REGULAR_TASK_EXCLUDED_BIZ_TYPES),
+    )
+    total_classified_coin_total = (
+        total_coupon_reward_coin_total
+        + total_comparison_reward_coin_total
+        + total_reward_video_coin_total
+        + total_regular_task_coin_total
+    )
+
     return {
         "users": {
             "total": _count(User),
@@ -356,12 +481,12 @@ def dashboard_overview(
             "dau": today_dau(db),
         },
         "coins": {
-            # 累计发放金币(coin_transaction 里所有 amount>0 之和;负数是兑换/扣减不计)
-            "granted_total": _sum(CoinTransaction.amount, CoinTransaction.amount > 0),
+            # 累计发放金币按大盘 4 个展示分类加总;未分类/运营手动/邀请奖励不进入本指标。
+            "granted_total": total_classified_coin_total,
             "reward_video_coin_total": _sum(
                 CoinTransaction.amount,
                 CoinTransaction.amount > 0,
-                CoinTransaction.biz_type.in_(("reward_video", "ad_reward")),
+                CoinTransaction.biz_type.in_(REWARD_VIDEO_BIZ_TYPES),
             ),
             "reward_video_watch_count": _count(
                 AdRewardRecord,
@@ -389,6 +514,10 @@ def dashboard_overview(
                 CoinTransaction.biz_type == "signin_boost",
             ),
             "signin_boost_watch_count": _count(SigninBoostRecord),
+            "task_coin_total": total_regular_task_coin_total,
+            "coupon_reward_coin_total": total_coupon_reward_coin_total,
+            "comparison_reward_coin_total": total_comparison_reward_coin_total,
+            "regular_task_coin_total": total_regular_task_coin_total,
         },
         "cash": {
             "withdraw_success_cents": _sum(
@@ -412,8 +541,8 @@ def dashboard_overview(
                 "retained_new_users": len(period_retained_new_user_ids),
                 "retention_rate": period_retention_rate,
                 "retention_note": (
-                    "近似口径:登录(last_login_at)+已上报比价记录+领券claim_started;"
-                    "尚不包含未完成上报的比价开始事件"
+                    "口径:登录(last_login_at)+开始比价(real_compare_start)+"
+                    "开始领券(real_coupon_start/claim_started),按用户去重"
                 ),
             },
             "comparison": {
@@ -425,7 +554,7 @@ def dashboard_overview(
                 "average_saved_cents": period_avg_saved_cents,
             },
             "coins": {
-                "granted_total": _sum(CoinTransaction.amount, *period_coin_conds),
+                "granted_total": period_classified_coin_total,
                 "reward_video_coin_total": period_reward_video_coin_total,
                 "feed_ad_coin_total": period_feed_ad_coin_total,
                 "signin_coin_total": period_signin_coin_total,
@@ -450,7 +579,7 @@ def dashboard_overview(
         },
         "cps": {
             "available": True,
-            "note": "美团 CPS 读 cps_order 对账订单;淘宝/京东佣金暂空",
+            "note": "美团/JD CPS 读 cps_order 对账订单;淘宝佣金暂空",
             "meituan_order_count": len(period_meituan_valid_orders),
             "meituan_commission_cents": sum(
                 o.commission_cents or 0 for o in period_meituan_valid_orders
@@ -459,5 +588,17 @@ def dashboard_overview(
             "meituan_miss_count": period_meituan_miss_count,
             "meituan_unknown_rate_count": period_meituan_unknown_rate_count,
             "meituan_hit_rate": period_meituan_hit_rate,
+            "jd_order_count": len(period_jd_valid_orders),
+            # 数据大盘京东 CPS 只看实际佣金,不再用预估佣金兜底。
+            "jd_commission_cents": sum(
+                o.actual_commission_cents or 0 for o in period_jd_valid_orders
+            ),
+            "jd_actual_commission_cents": sum(
+                o.actual_commission_cents or 0 for o in period_jd_valid_orders
+            ),
+            "jd_estimated_commission_cents": sum(
+                o.estimated_commission_cents or 0 for o in period_jd_valid_orders
+            ),
+            "jd_invalid_count": len(period_jd_invalid_orders),
         },
     }

app/admin/routers/cps.py

@@ -1,7 +1,7 @@
-"""admin CPS 分发与对账:群/活动管理 + 生成落地页短链 + 美团订单对账 + 统计。
+"""admin CPS 分发与对账:群/活动管理 + 生成落地页短链 + 联盟订单对账 + 统计。
 
 平台:meituan(actId+sid 转链 + query_order 对账) / taobao(整段淘口令) / jd(链接)。
-淘宝/京东无 API → 只统计点击(咱落地页 PV/UV + 淘宝复制),对账字段显示 "-"。
+淘宝暂未接 API → 只统计点击(咱落地页 PV/UV + 淘宝复制),对账字段显示 "-"。
 群/活动管理 = operator;订单对账(涉佣金) = finance;只读列表/统计 = 登录即可。
 """
 from __future__ import annotations
@@ -34,6 +34,7 @@ from app.admin.schemas.cps import (
 from app.core import media
 from app.core.config import settings
 from app.integrations import meituan
+from app.integrations.jd_union import JdUnionError
 from app.integrations.meituan import MeituanCpsError
 from app.models.admin import AdminUser
 from app.models.cps_activity import CpsActivity
@@ -373,7 +374,22 @@ def _reconcile_range_to_ts(
     return int(start_dt.timestamp()), int(end_dt.timestamp())
 
 
-@router.post("/orders/reconcile", response_model=CpsReconcileResult, summary="拉取美团订单对账")
+def _reconcile_range_to_bj_dt(
+    date_from: _date | None, date_to: _date | None, days: int
+) -> tuple[datetime, datetime]:
+    start_ts, end_ts = _reconcile_range_to_ts(date_from, date_to, days)
+    return (
+        datetime.fromtimestamp(start_ts, tz=_BEIJING),
+        datetime.fromtimestamp(end_ts, tz=_BEIJING),
+    )
+
+
+def _merge_reconcile_result(total: dict, current: dict) -> None:
+    for key in ("fetched", "inserted", "updated", "pages"):
+        total[key] = int(total.get(key, 0)) + int(current.get(key, 0))
+
+
+@router.post("/orders/reconcile", response_model=CpsReconcileResult, summary="拉取联盟订单对账")
 def reconcile_orders(
     request: Request,
     admin: Annotated[AdminUser, Depends(require_role("finance"))],
@@ -382,26 +398,42 @@ def reconcile_orders(
     date_to: Annotated[str | None, Query(description="结束日 YYYY-MM-DD")] = None,
     days: Annotated[int, Query(ge=1, le=90, description="未传日期时默认拉近 N 天")] = 7,
     sid: Annotated[str | None, Query(max_length=64)] = None,
-    query_time_type: Annotated[int, Query(ge=1, le=2)] = 1,
+    query_time_type: Annotated[int, Query(ge=1, le=3)] = 1,
+    platform: Annotated[str, Query(pattern="^(all|meituan|jd)$")] = "all",
 ) -> CpsReconcileResult:
-    start_ts, end_ts = _reconcile_range_to_ts(
-        _parse_day(date_from, field="date_from"),
-        _parse_day(date_to, field="date_to"),
-        days,
-    )
+    parsed_from = _parse_day(date_from, field="date_from")
+    parsed_to = _parse_day(date_to, field="date_to")
+    result = {"fetched": 0, "inserted": 0, "updated": 0, "pages": 0}
     try:
-        result = cps_repo.reconcile_orders(
-            db,
-            start_time=start_ts,
-            end_time=end_ts,
-            query_time_type=query_time_type,
-            sid=sid,
-        )
+        if platform in {"all", "meituan"}:
+            start_ts, end_ts = _reconcile_range_to_ts(parsed_from, parsed_to, days)
+            mt_result = cps_repo.reconcile_orders(
+                db,
+                start_time=start_ts,
+                end_time=end_ts,
+                query_time_type=query_time_type if query_time_type in (1, 2) else 2,
+                sid=sid,
+            )
+            _merge_reconcile_result(result, mt_result)
+        if platform in {"all", "jd"}:
+            if sid:
+                raise HTTPException(status_code=422, detail="京东订单刷新不支持 sid 筛选")
+            start_dt, end_dt = _reconcile_range_to_bj_dt(parsed_from, parsed_to, days)
+            jd_result = cps_repo.reconcile_jd_orders(
+                db,
+                start_time=start_dt,
+                end_time=end_dt,
+                query_time_type=query_time_type,
+            )
+            _merge_reconcile_result(result, jd_result)
     except MeituanCpsError as e:
         raise HTTPException(status_code=502, detail=f"美团拉单失败: {e}") from e
+    except JdUnionError as e:
+        raise HTTPException(status_code=502, detail=f"京东拉单失败: {e}") from e
     write_audit(
         db, admin, action="cps.order.reconcile", target_type="cps_order", target_id=None,
         detail={
+            "platform": platform,
             "date_from": date_from,
             "date_to": date_to,
             "days": days,

app/admin/schemas/cps.py

@@ -1,7 +1,7 @@
 """admin CPS 分发与对账 schemas。金额统一「分」(cents),前端 yuan() 展示。
 
-平台:meituan(actId+sid 转链对账) / taobao(淘口令,只统计点击) / jd(链接,只统计点击)。
-对账类字段对淘宝/京东为 None → 前端显示 "-"(无法对账)。
+平台:meituan(actId+sid 转链对账) / taobao(淘口令,只统计点击) / jd(链接 + 订单 API 对账)。
+对账类字段对淘宝为 None → 前端显示 "-"(暂未对账)。
 """
 from __future__ import annotations
 
@@ -116,15 +116,22 @@ class CpsOrderOut(BaseModel):
     model_config = ConfigDict(from_attributes=True)
 
     id: int
+    platform: str = "meituan"
     order_id: str
+    external_order_id: str | None = None
+    external_row_id: str | None = None
     sid: str | None = None
     act_id: str | None = None
     pay_price_cents: int | None = None
     commission_cents: int | None = None
+    estimated_commission_cents: int | None = None
+    actual_commission_cents: int | None = None
     commission_rate: str | None = None
     mt_status: str | None = None
+    jd_valid_code: str | None = None
     invalid_reason: str | None = None
     product_name: str | None = None
+    settle_month: str | None = None
     pay_time: datetime | None = None
 
 

app/admin/schemas/dashboard.py

@@ -103,6 +103,11 @@ class DashboardCps(BaseModel):
     meituan_miss_count: int = 0
     meituan_unknown_rate_count: int = 0
     meituan_hit_rate: float | None = None
+    jd_order_count: int = 0
+    jd_commission_cents: int = 0
+    jd_actual_commission_cents: int = 0
+    jd_estimated_commission_cents: int = 0
+    jd_invalid_count: int = 0
 
 
 class DashboardOverview(BaseModel):

app/admin/schemas/device.py

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

app/api/v1/auth.py

@@ -2,6 +2,7 @@
 
 路由前缀 `/api/v1/auth`,包含:
   POST /jverify-login         极光一键登录(loginToken → 手机号 → 注册即登录 → 签 JWT)
+  POST /wechat-login          微信 App 授权登录(code → openid → 注册即登录 → 签 JWT)
   POST /sms/send              发短信验证码(mock 阶段任意 6 位通过)
   POST /sms/login             手机号 + 验证码登录
   POST /refresh               用 refresh_token 换新的 token 对
@@ -20,6 +21,7 @@ from app.core.ratelimit import enforce_rate_limit, rate_limit
 from app.core.security import TokenError, decode_token, issue_token_pair
 from app.integrations.jiguang import JiguangError, mask_phone, verify_and_get_phone
 from app.integrations.sms import SmsError, send_code, verify_code
+from app.integrations.wechat_login import WechatLoginError, code_to_userinfo
 from app.repositories import onboarding as onboarding_repo
 from app.repositories import user as user_repo
 from app.schemas.auth import (
@@ -32,6 +34,7 @@ from app.schemas.auth import (
     TokenPair,
     TokenWithUser,
     UserOut,
+    WechatLoginRequest,
 )
 
 logger = logging.getLogger("shagua.auth")
@@ -84,6 +87,40 @@ def jverify_login(req: JverifyLoginRequest, db: DbSession) -> TokenWithUser:
     return _login_response(user, onboarding_completed=completed)
 
 
+# ===================== 微信一键登录 =====================
+
+@router.post(
+    "/wechat-login",
+    response_model=TokenWithUser,
+    summary="微信一键登录",
+    dependencies=[Depends(rate_limit(20, 60, "wechat-login"))],
+)
+def wechat_login(req: WechatLoginRequest, db: DbSession) -> TokenWithUser:
+    try:
+        info = code_to_userinfo(req.code)
+    except WechatLoginError as e:
+        logger.warning("wechat_login failed: %s", e)
+        raise HTTPException(status_code=502, detail=str(e)) from e
+
+    user = user_repo.upsert_user_for_wechat_login(
+        db,
+        openid=info["openid"],
+        nickname=info.get("nickname"),
+        avatar_url=info.get("avatar_url"),
+    )
+    if user.status != "active":
+        raise HTTPException(status_code=403, detail="account disabled")
+
+    completed = onboarding_repo.is_completed(db, user_id=user.id, device_id=req.device_id)
+    logger.info(
+        "wechat_login ok user_id=%d openid=%s*** onboarded=%s",
+        user.id,
+        info["openid"][:6],
+        completed,
+    )
+    return _login_response(user, onboarding_completed=completed)
+
+
 # ===================== 短信登录 =====================
 
 @router.post(

app/api/v1/device.py

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

app/core/config.py

@@ -9,7 +9,7 @@ from functools import lru_cache
 from pathlib import Path
 from typing import Literal
 
-from pydantic import Field, model_validator
+from pydantic import model_validator
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
 _PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent
@@ -59,14 +59,51 @@ class Settings(BaseSettings):
             return []
         return [ip.strip() for ip in self.ADMIN_IP_ALLOWLIST.split(",") if ip.strip()]
 
-    # ===== 极光 =====
+    # ===== 极光一键登录 / 短信 =====
     JG_APP_KEY: str = ""
     JG_MASTER_SECRET: str = ""
     JG_PRIVATE_KEY_PATH: str = "./secrets/jverify_rsa_private.pem"
     JG_VERIFY_ENDPOINT: str = "https://api.verification.jpush.cn/v1/web/loginTokenVerify"
     JG_REQUEST_TIMEOUT_SEC: int = 15
 
-    # 无障碍保护存活监控后台任务(pull 后置检测;本期不接推送)
+    # ===== 厂商直推(无障碍保护存活告警)=====
+    ANDROID_PACKAGE_NAME: str = "com.jishisongfu.shaguabijia"
+    PUSH_REQUEST_TIMEOUT_SEC: int = 15
+    PUSH_TIME_TO_LIVE_SEC: int = 86400
+
+    HONOR_PUSH_APP_ID: str = ""
+    HONOR_PUSH_CLIENT_ID: str = ""
+    HONOR_PUSH_CLIENT_SECRET: str = ""
+    HONOR_PUSH_TOKEN_ENDPOINT: str = "https://iam.developer.honor.com/auth/token"
+    HONOR_PUSH_SEND_ENDPOINT_TEMPLATE: str = (
+        "https://push-api.cloud.honor.com/api/v1/{app_id}/sendMessage"
+    )
+
+    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"
+    )
+
+    # 无障碍保护存活监控后台任务(推送 + pull 后置兜底)
     HEARTBEAT_MONITOR_ENABLED: bool = True       # 总开关
     HEARTBEAT_TIMEOUT_MINUTES: int = 10          # 多久没心跳算掉线(≈3 个客户端心跳周期)
     HEARTBEAT_SCAN_INTERVAL_SEC: int = 60        # 扫描周期
@@ -113,6 +150,21 @@ class Settings(BaseSettings):
         """美团 CPS 凭证齐全(缺则接口返空,而非 502)。"""
         return bool(self.MT_CPS_APP_KEY and self.MT_CPS_APP_SECRET)
 
+    # ===== 京东联盟 CPS =====
+    # app_key/app_secret 来自京东联盟应用;site_id 是推广管理里的 APP/网站 ID;
+    # auth_key 是工具商授权 key,自有应用查询可留空。
+    JD_UNION_APP_KEY: str = ""
+    JD_UNION_APP_SECRET: str = ""
+    JD_UNION_SITE_ID: str = ""
+    JD_UNION_AUTH_KEY: str = ""
+    JD_UNION_GATEWAY: str = "https://api.jd.com/routerjson"
+    JD_UNION_TIMEOUT_SEC: int = 15
+
+    @property
+    def jd_union_configured(self) -> bool:
+        """京东联盟订单查询凭证齐全。"""
+        return bool(self.JD_UNION_APP_KEY and self.JD_UNION_APP_SECRET)
+
     # ===== 微信服务号(网页授权) =====
     # CPS 落地页在微信内拿用户 openid(base 静默)/昵称头像(userinfo),做用户级群统计。
     # ⚠️ 区别于 WECHAT_APP_ID(那是 App 移动应用,用于微信支付);这是【已认证服务号】。
@@ -272,7 +324,7 @@ class Settings(BaseSettings):
         return self.APP_ENV == "prod"
 
     @model_validator(mode="after")
-    def _enforce_prod_secrets(self) -> "Settings":
+    def _enforce_prod_secrets(self) -> Settings:
         """prod 下强校验 JWT secret,弱/默认/空即启动报错(fail-fast,挡住 token 被伪造)。
 
         只校验两个签发凭证:App 用户的 JWT_SECRET_KEY、后台的 ADMIN_JWT_SECRET——它们沿用默认值

app/core/heartbeat_monitor_worker.py

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

app/integrations/jd_union.py

@@ -0,0 +1,163 @@
+"""京东联盟 OpenAPI 客户端。
+
+当前只接数据大盘需要的订单明细接口:
+`jd.union.open.order.row.query`。京东要求订单查询时间窗最长 1 小时,
+调用方负责切窗分页。
+"""
+from __future__ import annotations
+
+import hashlib
+import json
+import logging
+from datetime import datetime, timedelta, timezone
+from typing import Any
+
+import httpx
+
+from app.core.config import settings
+
+logger = logging.getLogger(__name__)
+
+_BEIJING = timezone(timedelta(hours=8))
+
+
+class JdUnionError(RuntimeError):
+    """京东联盟 API 调用失败。"""
+
+
+def _parse_json_maybe(value: Any) -> Any:
+    if not isinstance(value, str):
+        return value
+    text = value.strip()
+    if not text:
+        return value
+    try:
+        return json.loads(text)
+    except json.JSONDecodeError:
+        return value
+
+
+def _sign(params: dict[str, Any], secret: str) -> str:
+    pieces = [secret]
+    for key in sorted(k for k in params if k != "sign"):
+        value = params[key]
+        if value is None:
+            continue
+        pieces.append(f"{key}{value}")
+    pieces.append(secret)
+    raw = "".join(pieces)
+    return hashlib.md5(raw.encode("utf-8")).hexdigest().upper()
+
+
+def _unwrap_response(data: dict[str, Any]) -> dict[str, Any]:
+    if "error_response" in data:
+        err = data["error_response"] or {}
+        msg = err.get("zh_desc") or err.get("en_desc") or err.get("msg") or err
+        raise JdUnionError(f"京东 API 错误: {msg}")
+
+    body: Any = data
+    for key, value in data.items():
+        if key.endswith("_responce") or key.endswith("_response"):
+            body = value
+            break
+
+    body = _parse_json_maybe(body)
+    if not isinstance(body, dict):
+        raise JdUnionError("京东 API 返回格式异常")
+
+    result = body.get("queryResult", body.get("result", body))
+    result = _parse_json_maybe(result)
+    if not isinstance(result, dict):
+        raise JdUnionError("京东 API 业务结果格式异常")
+
+    code = str(result.get("code", result.get("resultCode", "200")))
+    if code not in {"0", "200"}:
+        msg = result.get("message") or result.get("msg") or result.get("resultMsg") or result
+        raise JdUnionError(f"京东 API 业务错误: {msg}")
+    return result
+
+
+def call(method: str, payload: dict[str, Any], *, version: str = "1.0") -> dict[str, Any]:
+    if not settings.jd_union_configured:
+        raise JdUnionError("京东联盟凭证未配置")
+
+    biz_json = json.dumps(payload, ensure_ascii=False, separators=(",", ":"))
+    params: dict[str, Any] = {
+        "method": method,
+        "app_key": settings.JD_UNION_APP_KEY,
+        "timestamp": datetime.now(_BEIJING).strftime("%Y-%m-%d %H:%M:%S"),
+        "format": "json",
+        "v": version,
+        "sign_method": "md5",
+        "360buy_param_json": biz_json,
+    }
+    params["sign"] = _sign(params, settings.JD_UNION_APP_SECRET)
+
+    try:
+        with httpx.Client(timeout=settings.JD_UNION_TIMEOUT_SEC, trust_env=False) as client:
+            resp = client.post(settings.JD_UNION_GATEWAY, data=params)
+            resp.raise_for_status()
+            data = resp.json()
+    except httpx.HTTPError as e:
+        raise JdUnionError(f"京东 API 网络错误: {e}") from e
+    except json.JSONDecodeError as e:
+        raise JdUnionError("京东 API 返回非 JSON") from e
+
+    return _unwrap_response(data)
+
+
+def _extract_rows(result: dict[str, Any]) -> tuple[list[dict[str, Any]], bool]:
+    payload = _parse_json_maybe(result.get("data", result.get("result", result)))
+    has_more = bool(result.get("hasMore") or result.get("has_more"))
+
+    if isinstance(payload, dict):
+        for key in ("orderRowResp", "orderRows", "orderList", "orders", "list", "rows"):
+            rows = _parse_json_maybe(payload.get(key))
+            if isinstance(rows, list):
+                return [r for r in rows if isinstance(r, dict)], bool(
+                    payload.get("hasMore") or payload.get("has_more") or has_more
+                )
+        return [], bool(payload.get("hasMore") or payload.get("has_more") or has_more)
+    if isinstance(payload, list):
+        return [r for r in payload if isinstance(r, dict)], has_more
+    return [], has_more
+
+
+def query_order_rows(
+    *,
+    start_time: datetime,
+    end_time: datetime,
+    query_time_type: int = 3,
+    page_index: int = 1,
+    page_size: int = 200,
+) -> dict[str, Any]:
+    """查询京东 CPS 订单行。
+
+    query_time_type: 1 下单时间, 2 完成时间, 3 更新时间。
+    start_time/end_time 用北京时间展示给京东;调用方需保证窗口不超过 1 小时。
+    """
+    start_bj = start_time.astimezone(_BEIJING)
+    end_bj = end_time.astimezone(_BEIJING)
+    if end_bj <= start_bj:
+        return {"rows": [], "has_more": False}
+    if end_bj - start_bj > timedelta(hours=1):
+        raise JdUnionError("京东订单查询单次时间窗不能超过 1 小时")
+
+    order_req: dict[str, Any] = {
+        "pageIndex": page_index,
+        "pageSize": min(max(page_size, 1), 200),
+        "type": query_time_type,
+        "startTime": start_bj.strftime("%Y-%m-%d %H:%M:%S"),
+        "endTime": end_bj.strftime("%Y-%m-%d %H:%M:%S"),
+    }
+    if settings.JD_UNION_AUTH_KEY:
+        order_req["key"] = settings.JD_UNION_AUTH_KEY
+    result = call("jd.union.open.order.row.query", {"orderReq": order_req})
+    rows, has_more = _extract_rows(result)
+    logger.info(
+        "jd.union.open.order.row.query fetched rows=%s page=%s has_more=%s",
+        len(rows),
+        page_index,
+        has_more,
+    )
+    return {"rows": rows, "has_more": has_more, "raw": result}

app/integrations/vendor_push.py

@@ -0,0 +1,380 @@
+"""厂商直推集成。
+
+服务端不再经由 JPush Push API 发送无障碍召回通知,而是按客户端上报的
+push_vendor + push_token 分发到各手机厂商的服务端 API。
+"""
+from __future__ import annotations
+
+import hashlib
+import json
+import logging
+import time
+import uuid
+from collections.abc import Callable
+from dataclasses import dataclass
+from typing import Any
+
+import httpx
+
+from app.core.config import settings
+
+logger = logging.getLogger("shagua.vendor_push")
+
+TYPE_ACCESSIBILITY_DISABLED = "accessibility_disabled"
+SUPPORTED_VENDORS = frozenset({"honor", "vivo", "xiaomi", "oppo"})
+
+
+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",
+        "mi": "xiaomi",
+        "小米": "xiaomi",
+        "oneplus": "oppo",
+        "realme": "oppo",
+    }
+    return aliases.get(vendor, vendor)
+
+
+def send_accessibility_disabled(
+    push_vendor: str,
+    push_token: str,
+    *,
+    title: str = "保护已关闭",
+    alert: str = "傻瓜比价的无障碍保护被关了，点此重新开启，继续帮你自动比价省钱。",
+) -> dict[str, Any]:
+    """按厂商 token 向单台设备发送无障碍掉线通知。"""
+    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")
+
+    dispatch: dict[str, Callable[[str, str, str], dict[str, Any]]] = {
+        "honor": _send_honor,
+        "vivo": _send_vivo,
+        "xiaomi": _send_xiaomi,
+        "oppo": _send_oppo,
+    }
+    return dispatch[vendor](token, title, alert)
+
+
+def _extras() -> dict[str, str]:
+    return {"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, alert: str) -> dict[str, Any]:
+    app_id = _require(settings.HONOR_PUSH_APP_ID, "HONOR_PUSH_APP_ID")
+    access_token = _honor_access_token()
+    payload = {
+        "data": json.dumps(_extras(), ensure_ascii=False),
+        "notification": {"title": title, "body": alert},
+        "android": {
+            "ttl": f"{settings.PUSH_TIME_TO_LIVE_SEC}s",
+            "targetUserType": 1,
+            "notification": {
+                "title": title,
+                "body": alert,
+                "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 _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, alert: str) -> dict[str, Any]:
+    app_id = _require(settings.VIVO_PUSH_APP_ID, "VIVO_PUSH_APP_ID")
+    auth_token = _vivo_auth_token()
+    body: dict[str, Any] = {
+        "appId": app_id,
+        "regId": token,
+        "notifyType": settings.VIVO_PUSH_NOTIFY_TYPE,
+        "title": title,
+        "content": alert,
+        "timeToLive": settings.PUSH_TIME_TO_LIVE_SEC,
+        "skipType": 1,
+        "requestId": uuid.uuid4().hex,
+        "pushMode": settings.VIVO_PUSH_MODE,
+        "clientCustomMap": _extras(),
+    }
+    if settings.VIVO_PUSH_CATEGORY:
+        body["category"] = settings.VIVO_PUSH_CATEGORY
+    data = _request_json(
+        "POST",
+        settings.VIVO_PUSH_SEND_ENDPOINT,
+        json=body,
+        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, alert: 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 alert
+    body = {
+        "registration_id": token,
+        "restricted_package_name": settings.ANDROID_PACKAGE_NAME,
+        "title": message_title,
+        "description": message_description,
+        "payload": json.dumps(_extras(), ensure_ascii=False),
+        "pass_through": "0",
+        "notify_type": "-1",
+        "time_to_live": str(settings.PUSH_TIME_TO_LIVE_SEC * 1000),
+        "extra.notify_effect": "1",
+    }
+    if settings.XIAOMI_PUSH_CHANNEL_ID:
+        body["extra.channel_id"] = settings.XIAOMI_PUSH_CHANNEL_ID.strip()
+    if settings.XIAOMI_PUSH_TEMPLATE_ID:
+        body["extra.template_id"] = settings.XIAOMI_PUSH_TEMPLATE_ID.strip()
+    if settings.XIAOMI_PUSH_TEMPLATE_PARAM_JSON:
+        body["extra.template_param"] = _xiaomi_template_param(title, alert)
+    data = _request_form(
+        "POST",
+        settings.XIAOMI_PUSH_SEND_ENDPOINT,
+        data=body,
+        headers={"Authorization": f"key={app_secret}"},
+    )
+    code = data.get("code")
+    if code not in (0, "0", None):
+        raise VendorPushError(f"xiaomi push failed: {data}")
+    if str(data.get("result", "ok")).lower() not in ("ok", "success"):
+        raise VendorPushError(f"xiaomi push failed: {data}")
+    return data
+
+
+def _xiaomi_template_param(title: str, alert: str) -> str:
+    rendered = (
+        settings.XIAOMI_PUSH_TEMPLATE_PARAM_JSON
+        .replace("{title}", title)
+        .replace("{alert}", alert)
+    )
+    try:
+        payload = json.loads(rendered)
+    except ValueError as e:
+        raise VendorPushError("XIAOMI_PUSH_TEMPLATE_PARAM_JSON invalid json") from e
+    if not isinstance(payload, dict):
+        raise VendorPushError("XIAOMI_PUSH_TEMPLATE_PARAM_JSON must be a json object")
+    for key, value in payload.items():
+        if not isinstance(key, str) or not isinstance(value, str):
+            raise VendorPushError("xiaomi template params must be string key-value pairs")
+        if not value.strip() or len(value) > 128:
+            raise VendorPushError("xiaomi template param value length must be 1-128")
+    return json.dumps(payload, ensure_ascii=False, separators=(",", ":"))
+
+
+def _oppo_auth_token() -> str:
+    cache_key = "oppo"
+    cached = _cache_get(cache_key)
+    if cached:
+        return cached
+    app_key = _require(settings.OPPO_PUSH_APP_KEY, "OPPO_PUSH_APP_KEY")
+    master_secret = _require(settings.OPPO_PUSH_MASTER_SECRET, "OPPO_PUSH_MASTER_SECRET")
+    timestamp = str(int(time.time() * 1000))
+    sign = hashlib.sha256(f"{app_key}{timestamp}{master_secret}".encode()).hexdigest()
+    data = _request_form(
+        "POST",
+        settings.OPPO_PUSH_AUTH_ENDPOINT,
+        data={
+            "app_key": app_key,
+            "timestamp": timestamp,
+            "sign": sign,
+        },
+        headers={"Content-Type": "application/x-www-form-urlencoded"},
+    )
+    if int(data.get("code", -1)) != 0:
+        raise VendorPushError(f"oppo auth failed: {data}")
+    token = (data.get("data") or {}).get("auth_token") or data.get("auth_token")
+    if not token:
+        raise VendorPushError(f"oppo auth missing auth_token: {data}")
+    return _cache_put(cache_key, str(token), 24 * 3600)
+
+
+def _send_oppo(token: str, title: str, alert: str) -> dict[str, Any]:
+    auth_token = _oppo_auth_token()
+    ttl_hours = max(1, min(72, settings.PUSH_TIME_TO_LIVE_SEC // 3600))
+    message = {
+        "target_type": 2,
+        "target_value": token,
+        "notification": {
+            "app_message_id": f"accessibility_disabled_{uuid.uuid4().hex}",
+            "title": title,
+            "content": alert,
+            "click_action_type": 0,
+            "off_line": True,
+            "off_line_ttl": ttl_hours,
+            "action_parameters": json.dumps(_extras(), ensure_ascii=False),
+        },
+    }
+    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/integrations/wechat_login.py

@@ -0,0 +1,89 @@
+"""微信开放平台 App 登录集成。
+
+客户端只把微信 SDK 回调拿到的临时 code 传给服务端;AppSecret、access_token
+与用户 openid 都留在服务端处理,避免敏感凭证落到 APK 里。
+"""
+from __future__ import annotations
+
+import certifi
+import httpx
+
+from app.core.config import settings
+
+_API_BASE = "https://api.weixin.qq.com"
+
+
+class WechatLoginError(Exception):
+    """微信授权换取用户信息失败。"""
+
+
+def _ensure_configured() -> None:
+    if not settings.WECHAT_APP_ID or not settings.WECHAT_APP_SECRET:
+        raise WechatLoginError("wechat app not configured")
+
+
+def _http_client() -> httpx.Client:
+    return httpx.Client(verify=certifi.where())
+
+
+def code_to_userinfo(code: str) -> dict:
+    """用授权 code 换 openid/unionid,尽力补昵称头像。
+
+    返回字段:
+      - openid: 同一开放平台移动应用下的用户唯一标识
+      - unionid: 同一开放平台主体下的用户唯一标识,微信可能不返回
+      - nickname/avatar_url: sns/userinfo 可用时返回,失败不阻断登录
+      - raw: userinfo 原始响应,便于排障
+    """
+    _ensure_configured()
+    try:
+        with _http_client() as client:
+            token_resp = client.get(
+                f"{_API_BASE}/sns/oauth2/access_token",
+                params={
+                    "appid": settings.WECHAT_APP_ID,
+                    "secret": settings.WECHAT_APP_SECRET,
+                    "code": code,
+                    "grant_type": "authorization_code",
+                },
+                timeout=settings.WXPAY_REQUEST_TIMEOUT_SEC,
+            )
+        token_data = token_resp.json()
+    except Exception as e:  # noqa: BLE001
+        raise WechatLoginError("微信授权请求失败,请稍后重试") from e
+    if "openid" not in token_data or "access_token" not in token_data:
+        raise WechatLoginError(f"微信授权失败: {token_data.get('errmsg', token_data)}")
+
+    openid = token_data["openid"]
+    unionid = token_data.get("unionid")
+    nickname = None
+    avatar_url = None
+    raw: dict = {}
+
+    try:
+        with _http_client() as client:
+            info_resp = client.get(
+                f"{_API_BASE}/sns/userinfo",
+                params={
+                    "access_token": token_data["access_token"],
+                    "openid": openid,
+                    "lang": "zh_CN",
+                },
+                timeout=settings.WXPAY_REQUEST_TIMEOUT_SEC,
+            )
+        raw = info_resp.json()
+        if "errcode" not in raw:
+            nickname = raw.get("nickname") or None
+            avatar_url = raw.get("headimgurl") or None
+            unionid = unionid or raw.get("unionid")
+    except Exception:  # noqa: BLE001
+        # openid 已拿到即可登录;昵称头像只是展示增强。
+        pass
+
+    return {
+        "openid": openid,
+        "unionid": unionid,
+        "nickname": nickname,
+        "avatar_url": avatar_url,
+        "raw": raw,
+    }

app/models/cps_order.py

@@ -1,11 +1,13 @@
 """CPS 对账订单(cps_order)。
 
-从美团联盟 query_order 按时间窗拉回、按 sid 归群的订单明细。字段对齐 query_order
+从联盟 API 按时间窗拉回、按平台落库的 CPS 订单明细。字段最初对齐美团 query_order,
+后续兼容京东订单报表:
 实测返回:
   - payPrice / profit 是「元」字符串 → 入库统一转「分」(与全站口径一致)
   - payTime / updateTime 是秒级时间戳 → 入库转 tz-aware datetime
   - status: 2付款 3完成 4取消 5风控 6结算(取消/风控不计佣金)
-order_id 全局唯一,reconcile 按它 upsert(订单状态会变,重复拉则更新)。
+order_id 全局唯一,reconcile 按它 upsert(订单状态会变,重复拉则更新)。京东订单用
+`jd:<row_id>` 前缀避免与美团订单号碰撞。
 """
 from __future__ import annotations
 
@@ -22,8 +24,13 @@ class CpsOrder(Base):
     __tablename__ = "cps_order"
 
     id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
-    # 美团订单号(加密串),全局唯一,upsert 幂等键。
+    # 平台:meituan / jd。历史数据迁移默认 meituan。
+    platform: Mapped[str] = mapped_column(String(20), default="meituan", index=True, nullable=False)
+    # 平台订单号/行号包装后的全局唯一键,upsert 幂等。
     order_id: Mapped[str] = mapped_column(String(64), unique=True, index=True, nullable=False)
+    # 平台原始订单号/行号。京东一笔订单多 SKU 时可按行号区分。
+    external_order_id: Mapped[str | None] = mapped_column(String(128), index=True, nullable=True)
+    external_row_id: Mapped[str | None] = mapped_column(String(128), index=True, nullable=True)
     # 渠道追踪位 = 群 sid(历史无 sid 订单为空)。按它归群聚合。
     sid: Mapped[str | None] = mapped_column(String(64), index=True, nullable=True)
     act_id: Mapped[str | None] = mapped_column(String(64), index=True, nullable=True)
@@ -35,11 +42,21 @@ class CpsOrder(Base):
     commission_rate: Mapped[str | None] = mapped_column(String(16), nullable=True)  # "300"=3% "10"=0.1%
     refund_price_cents: Mapped[int | None] = mapped_column(Integer, nullable=True)
     refund_profit_cents: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    # 通用佣金拆分。美团只有预估 profit;京东有预估/实际佣金。
+    estimated_commission_cents: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    actual_commission_cents: Mapped[int | None] = mapped_column(Integer, nullable=True)
 
     # 美团订单状态: 2付款 3完成 4取消 5风控 6结算
     mt_status: Mapped[str | None] = mapped_column(String(8), index=True, nullable=True)
+    # 京东订单有效码(validCode),用于判断是否有效/已完成。
+    jd_valid_code: Mapped[str | None] = mapped_column(String(16), index=True, nullable=True)
     invalid_reason: Mapped[str | None] = mapped_column(String(128), nullable=True)
     product_name: Mapped[str | None] = mapped_column(String(512), nullable=True)
+    settle_month: Mapped[str | None] = mapped_column(String(16), nullable=True)
+    site_id: Mapped[str | None] = mapped_column(String(128), nullable=True)
+    position_id: Mapped[str | None] = mapped_column(String(128), nullable=True)
+    pid: Mapped[str | None] = mapped_column(String(128), nullable=True)
+    sub_union_id: Mapped[str | None] = mapped_column(String(128), nullable=True)
 
     pay_time: Mapped[datetime | None] = mapped_column(
         DateTime(timezone=True), index=True, nullable=True
@@ -59,5 +76,6 @@ class CpsOrder(Base):
     def __repr__(self) -> str:  # pragma: no cover
         return (
             f"<CpsOrder id={self.id} order_id={self.order_id!r} "
-            f"sid={self.sid!r} status={self.mt_status} profit_cents={self.commission_cents}>"
+            f"platform={self.platform!r} sid={self.sid!r} "
+            f"status={self.mt_status or self.jd_valid_code} profit_cents={self.commission_cents}>"
         )

app/models/device.py

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

app/repositories/device.py

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

app/repositories/user.py

@@ -6,6 +6,7 @@ from __future__ import annotations
 
 import secrets
 import string
+import hashlib
 from datetime import datetime, timezone
 
 from sqlalchemy import select
@@ -70,6 +71,21 @@ def get_user_by_phone(db: Session, phone: str) -> User | None:
     return db.execute(stmt).scalar_one_or_none()
 
 
+def get_user_by_wechat_openid(db: Session, openid: str) -> User | None:
+    stmt = select(User).where(User.wechat_openid == openid)
+    return db.execute(stmt).scalar_one_or_none()
+
+
+def _wechat_placeholder_phone(openid: str) -> str:
+    """给纯微信注册用户生成内部登录号。
+
+    当前 user.phone 非空唯一,而微信 App 授权不会返回手机号。这里生成一个不可能
+    命中手机号正则的内部值,只用于满足既有表结构和客户端登录态判断。
+    """
+    digest = hashlib.sha256(openid.encode("utf-8")).hexdigest()[:16]
+    return f"wx_{digest}"
+
+
 def upsert_user_for_login(
     db: Session,
     *,
@@ -99,6 +115,52 @@ def upsert_user_for_login(
     return user
 
 
+def upsert_user_for_wechat_login(
+    db: Session,
+    *,
+    openid: str,
+    nickname: str | None = None,
+    avatar_url: str | None = None,
+) -> User:
+    """微信 App 登录:openid 已存在则回到原账号,不存在则注册新账号。
+
+    如果用户之前在提现页绑定过微信,该 openid 已在 user.wechat_openid 上,因此会
+    直接登录到同一账号;新微信用户则创建 register_channel=wechat 的账号。
+    """
+    user = get_user_by_wechat_openid(db, openid)
+    now = datetime.now(timezone.utc)
+    if user is None:
+        phone = _wechat_placeholder_phone(openid)
+        # 极小概率 hash 前缀碰撞时,后缀追加随机段兜底。
+        if get_user_by_phone(db, phone) is not None:
+            phone = f"wx_{secrets.token_hex(8)}"
+        user = User(
+            phone=phone,
+            username=_gen_unique_username(db),
+            nickname=nickname or _gen_nickname(),
+            avatar_url=avatar_url,
+            register_channel="wechat",
+            wechat_openid=openid,
+            wechat_nickname=nickname,
+            wechat_avatar_url=avatar_url,
+            last_login_at=now,
+        )
+        db.add(user)
+    else:
+        user.last_login_at = now
+        if nickname:
+            user.wechat_nickname = nickname
+            if not user.nickname:
+                user.nickname = nickname
+        if avatar_url:
+            user.wechat_avatar_url = avatar_url
+            if not user.avatar_url:
+                user.avatar_url = avatar_url
+    db.commit()
+    db.refresh(user)
+    return user
+
+
 def update_nickname(db: Session, user: User, *, nickname: str) -> User:
     user.nickname = nickname
     db.commit()

app/schemas/auth.py

@@ -67,6 +67,14 @@ class JverifyLoginRequest(BaseModel):
     )
 
 
+class WechatLoginRequest(BaseModel):
+    code: str = Field(..., min_length=1, description="微信 SDK SendAuth.Resp 返回的一次性授权 code")
+    device_id: str = Field(
+        "", max_length=64,
+        description="硬件级设备标识(Android ANDROID_ID),用于新手引导按 设备+账号 去重;空=按未完成处理",
+    )
+
+
 # ===== 短信验证码 =====
 
 class SmsSendRequest(BaseModel):

app/schemas/device.py

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

tests/test_admin_read.py

@@ -66,6 +66,7 @@ def test_dashboard_overview(admin_client: TestClient, admin_token: str) -> None:
     assert "success_rate" in data["comparison"]
     assert data["cps"]["available"] is True
     assert "meituan_order_count" in data["cps"]
+    assert "jd_order_count" in data["cps"]
 
 
 def test_user_list_and_detail(admin_client: TestClient, admin_token: str) -> None:

tests/test_auth.py

@@ -68,6 +68,64 @@ def test_sms_login_and_me_flow(client) -> None:
     assert r.json()["ok"] is True
 
 
+def test_wechat_login_creates_and_reuses_user(client, monkeypatch) -> None:
+    """微信 code 登录:新 openid 注册,同 openid 重登回同一账号。"""
+    monkeypatch.setattr(
+        "app.api.v1.auth.code_to_userinfo",
+        lambda code: {
+            "openid": "openid_login_1",
+            "unionid": "union_1",
+            "nickname": "微信用户A",
+            "avatar_url": "https://wx/avatar-a.png",
+            "raw": {},
+        },
+    )
+
+    r = client.post("/api/v1/auth/wechat-login", json={"code": "code-a", "device_id": "dev-wx"})
+    assert r.status_code == 200, r.text
+    first = r.json()
+    assert first["user"]["register_channel"] == "wechat"
+    assert first["user"]["phone"].startswith("wx_")
+    assert first["user"]["nickname"] == "微信用户A"
+
+    r = client.post("/api/v1/auth/wechat-login", json={"code": "code-b", "device_id": "dev-wx"})
+    assert r.status_code == 200, r.text
+    second = r.json()
+    assert second["user"]["id"] == first["user"]["id"]
+
+
+def test_wechat_login_uses_existing_bound_user(client, monkeypatch) -> None:
+    """如果该微信已在提现页绑定过,微信登录应回到原手机号账号。"""
+    from app.db.session import SessionLocal
+    from app.repositories import user as user_repo
+
+    db = SessionLocal()
+    try:
+        user = user_repo.upsert_user_for_login(db, phone="13600136010", register_channel="sms")
+        user.wechat_openid = "openid_bound_1"
+        db.commit()
+        uid = user.id
+    finally:
+        db.close()
+
+    monkeypatch.setattr(
+        "app.api.v1.auth.code_to_userinfo",
+        lambda code: {
+            "openid": "openid_bound_1",
+            "unionid": None,
+            "nickname": "绑定微信",
+            "avatar_url": None,
+            "raw": {},
+        },
+    )
+
+    r = client.post("/api/v1/auth/wechat-login", json={"code": "code-bound", "device_id": "dev-wx"})
+    assert r.status_code == 200, r.text
+    body = r.json()
+    assert body["user"]["id"] == uid
+    assert body["user"]["phone"] == "13600136010"
+
+
 def test_sms_send_too_frequent(client) -> None:
     phone = "13900139000"
     assert client.post("/api/v1/auth/sms/send", json={"phone": phone}).status_code == 200

tests/test_cps_admin.py

@@ -11,6 +11,7 @@ from app.admin.repositories import admin_user as admin_repo
 from app.admin.repositories import cps as cps_repo
 from app.db.session import SessionLocal
 from app.models.cps_link import CpsClick
+from app.models.cps_order import CpsOrder
 from app.models.cps_wx_user import CpsWxUser
 from app.repositories import cps_link as cps_link_repo
 
@@ -180,3 +181,73 @@ def test_day_users_cross_year(admin_client: TestClient, admin_token: str) -> Non
     assert len(users) == 1
     assert users[0]["openid"] == openid
     assert users[0]["visit_count"] == 5  # 次年 01-01 那条被时间窗排除
+
+
+def test_jd_reconcile_updates_dashboard(
+    admin_client: TestClient, admin_token: str, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """京东拉单:按订单行入库;大盘只统计有效 validCode 的佣金。"""
+
+    def fake_query_order_rows(**kwargs):
+        start = kwargs["start_time"].astimezone(_BJ)
+        if start.hour != 10 or kwargs["page_index"] != 1:
+            return {"rows": [], "has_more": False}
+        return {
+            "rows": [
+                {
+                    "id": "pytest-jd-row-valid",
+                    "orderId": "pytest-jd-order-1",
+                    "skuId": "sku-1",
+                    "skuName": "京东测试商品",
+                    "orderTime": "2026-06-25 10:10:00",
+                    "modifyTime": "2026-06-25 10:20:00",
+                    "validCode": "16",
+                    "estimateCosPrice": "19.90",
+                    "estimateFee": "1.23",
+                    "actualFee": "2.34",
+                    "commissionRate": "10.00",
+                },
+                {
+                    "id": "pytest-jd-row-invalid",
+                    "orderId": "pytest-jd-order-2",
+                    "skuId": "sku-2",
+                    "skuName": "京东无效订单",
+                    "orderTime": "2026-06-25 10:15:00",
+                    "modifyTime": "2026-06-25 10:25:00",
+                    "validCode": "4",
+                    "estimateCosPrice": "9.90",
+                    "estimateFee": "0.50",
+                    "commissionRate": "5.00",
+                },
+            ],
+            "has_more": False,
+        }
+
+    monkeypatch.setattr("app.integrations.jd_union.query_order_rows", fake_query_order_rows)
+    r = admin_client.post(
+        "/admin/api/cps/orders/reconcile",
+        params={
+            "platform": "jd",
+            "date_from": "2026-06-25",
+            "date_to": "2026-06-25",
+            "query_time_type": 1,
+        },
+        headers=_auth(admin_token),
+    )
+    assert r.status_code == 200, r.text
+    assert r.json()["fetched"] == 2
+
+    with SessionLocal() as db:
+        rows = db.query(CpsOrder).filter(CpsOrder.platform == "jd").all()
+        assert len([o for o in rows if o.order_id.startswith("jd:pytest-jd-row")]) == 2
+
+    overview = admin_client.get(
+        "/admin/api/stats/overview",
+        params={"date_from": "2026-06-25", "date_to": "2026-06-25"},
+        headers=_auth(admin_token),
+    )
+    assert overview.status_code == 200, overview.text
+    cps = overview.json()["cps"]
+    assert cps["jd_order_count"] >= 1
+    assert cps["jd_commission_cents"] >= 234
+    assert cps["jd_invalid_count"] >= 1

tests/test_device_push.py

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

tests/test_health.py

@@ -14,6 +14,7 @@ def test_openapi_loads(client) -> None:
     paths = resp.json()["paths"]
     # auth endpoints 都注册了
     assert "/api/v1/auth/jverify-login" in paths
+    assert "/api/v1/auth/wechat-login" in paths
     assert "/api/v1/auth/sms/send" in paths
     assert "/api/v1/auth/sms/login" in paths
     assert "/api/v1/auth/refresh" in paths