exinglang
754 lines
30 KiB
Python
754 lines
30 KiB
Python
"""厂商直推集成(荣耀 / 华为 / 小米 / OPPO / vivo)。
|
||
|
||
服务端不经由 JPush Push API,而是按客户端上报的 push_vendor + push_token
|
||
分发到各手机厂商的服务端 API。
|
||
|
||
对外两个入口:
|
||
- send_notification() 通用:任意标题/正文/extras(消息中心 13 类推送走这里),
|
||
mock=True 时不真调厂商、返回渲染结果(虚拟数据联调用)
|
||
- send_accessibility_disabled() 旧:无障碍掉线召回(heartbeat_monitor_worker 在用),
|
||
已改为 send_notification 的薄封装,行为不变
|
||
|
||
各厂商鉴权方式:荣耀/华为 OAuth client_credentials 换 access_token(进程内缓存);
|
||
vivo/OPPO 签名换 authToken(缓存 24h);小米直接 AppSecret 走 Authorization 头。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import json
|
||
import logging
|
||
import time
|
||
import uuid
|
||
from collections.abc import Callable
|
||
from dataclasses import dataclass
|
||
from typing import Any
|
||
from urllib.parse import quote
|
||
|
||
import httpx
|
||
|
||
from app.core.config import settings
|
||
|
||
logger = logging.getLogger("shagua.vendor_push")
|
||
|
||
TYPE_ACCESSIBILITY_DISABLED = "accessibility_disabled"
|
||
DATA_EVENT_NOTIFICATION_CREATED = "notification_created"
|
||
SUPPORTED_VENDORS = frozenset({"honor", "huawei", "vivo", "xiaomi", "oppo"})
|
||
|
||
# vendor key → 中文名(测试/配置状态接口展示用)
|
||
VENDOR_LABELS: dict[str, str] = {
|
||
"honor": "荣耀",
|
||
"huawei": "华为",
|
||
"xiaomi": "小米",
|
||
"oppo": "OPPO",
|
||
"vivo": "vivo",
|
||
}
|
||
|
||
# 各厂商真发推送所需的 settings 键(缺任一即视为未配置;/api/v1/push/vendors 据此报缺)
|
||
REQUIRED_SETTINGS: dict[str, tuple[str, ...]] = {
|
||
"honor": ("HONOR_PUSH_APP_ID", "HONOR_PUSH_CLIENT_ID", "HONOR_PUSH_CLIENT_SECRET"),
|
||
"huawei": ("HUAWEI_PUSH_APP_ID", "HUAWEI_PUSH_APP_SECRET"),
|
||
"xiaomi": ("XIAOMI_PUSH_APP_SECRET",),
|
||
"oppo": ("OPPO_PUSH_APP_KEY", "OPPO_PUSH_MASTER_SECRET"),
|
||
"vivo": ("VIVO_PUSH_APP_ID", "VIVO_PUSH_APP_KEY", "VIVO_PUSH_APP_SECRET"),
|
||
}
|
||
|
||
|
||
def missing_settings(vendor: str) -> list[str]:
|
||
"""该厂商还缺哪些配置键(全配齐返回空列表)。vendor 需已 normalize。"""
|
||
return [key for key in REQUIRED_SETTINGS.get(vendor, ()) if not getattr(settings, key, "")]
|
||
|
||
|
||
class VendorPushError(Exception):
|
||
"""厂商推送调用失败。"""
|
||
|
||
|
||
@dataclass
|
||
class _CachedToken:
|
||
value: str
|
||
expires_at: float
|
||
|
||
|
||
_token_cache: dict[str, _CachedToken] = {}
|
||
|
||
|
||
def normalize_vendor(push_vendor: str | None) -> str | None:
|
||
if not push_vendor:
|
||
return None
|
||
vendor = push_vendor.strip().lower()
|
||
aliases = {
|
||
"hihonor": "honor",
|
||
"荣耀": "honor",
|
||
"hms": "huawei",
|
||
"华为": "huawei",
|
||
"harmony": "huawei",
|
||
"harmonyos": "huawei",
|
||
"mi": "xiaomi",
|
||
"小米": "xiaomi",
|
||
"oneplus": "oppo",
|
||
"realme": "oppo",
|
||
}
|
||
return aliases.get(vendor, vendor)
|
||
|
||
|
||
def send_notification(
|
||
push_vendor: str,
|
||
push_token: str,
|
||
*,
|
||
title: str,
|
||
body: str,
|
||
extras: dict[str, str] | None = None,
|
||
mock: bool = False,
|
||
) -> dict[str, Any]:
|
||
"""按厂商 token 向单台设备发送一条通知(通用入口)。
|
||
|
||
- extras:透传给客户端的自定义键值(值统一 string,兼容各厂商限制)。消息中心推送约定
|
||
至少带 {"type": <13 种类型 key>, "notificationId": <站内消息 id>},客户端据此
|
||
深链落地 + 调 /notifications/read 同步置读(PRD §4 push 联动)。
|
||
- mock=True:不真调厂商 API,校验参数后原样返回渲染结果(虚拟数据阶段联调/自动化测试用)。
|
||
"""
|
||
vendor = normalize_vendor(push_vendor)
|
||
token = push_token.strip() if push_token else ""
|
||
if not vendor or vendor not in SUPPORTED_VENDORS:
|
||
raise VendorPushError(f"unsupported push vendor: {push_vendor}")
|
||
if not token:
|
||
raise VendorPushError("push token is empty")
|
||
extras = {str(k): str(v) for k, v in (extras or {}).items()}
|
||
|
||
if mock:
|
||
logger.info(
|
||
"[mock push] vendor=%s token=%s... title=%s body=%s extras=%s",
|
||
vendor, token[:12], title, body, extras,
|
||
)
|
||
return {
|
||
"mock": True,
|
||
"vendor": vendor,
|
||
"title": title,
|
||
"body": body,
|
||
"extras": extras,
|
||
}
|
||
|
||
dispatch: dict[str, Callable[[str, str, str, dict[str, str]], dict[str, Any]]] = {
|
||
"honor": _send_honor,
|
||
"huawei": _send_huawei,
|
||
"vivo": _send_vivo,
|
||
"xiaomi": _send_xiaomi,
|
||
"oppo": _send_oppo,
|
||
}
|
||
return dispatch[vendor](token, title, body, extras)
|
||
|
||
|
||
def send_accessibility_disabled(
|
||
push_vendor: str,
|
||
push_token: str,
|
||
*,
|
||
title: str = "保护已关闭",
|
||
alert: str = "傻瓜比价的无障碍保护被关了，点此重新开启，继续帮你自动比价省钱。",
|
||
) -> dict[str, Any]:
|
||
"""按厂商 token 向单台设备发送无障碍掉线通知(heartbeat_monitor_worker 在用,行为不变)。"""
|
||
return send_notification(
|
||
push_vendor,
|
||
push_token,
|
||
title=title,
|
||
body=alert,
|
||
extras={"type": TYPE_ACCESSIBILITY_DISABLED},
|
||
)
|
||
|
||
|
||
def send_data_event(
|
||
push_vendor: str,
|
||
push_token: str,
|
||
*,
|
||
event: str,
|
||
notification_id: str,
|
||
mock: bool = False,
|
||
) -> dict[str, Any]:
|
||
"""发送无界面的轻量事件；当前只允许站内消息创建事件。"""
|
||
vendor = normalize_vendor(push_vendor)
|
||
token = push_token.strip() if push_token else ""
|
||
if not vendor or vendor not in SUPPORTED_VENDORS:
|
||
raise VendorPushError(f"unsupported push vendor: {push_vendor}")
|
||
if not token:
|
||
raise VendorPushError("push token is empty")
|
||
if event != DATA_EVENT_NOTIFICATION_CREATED:
|
||
raise VendorPushError(f"unsupported data event: {event}")
|
||
if not notification_id:
|
||
raise VendorPushError("notification_id is empty")
|
||
|
||
payload = {"event": event, "notificationId": str(notification_id)}
|
||
if mock:
|
||
return {"mock": True, "vendor": vendor, "payload": payload}
|
||
|
||
# vivo 的通知已设置 foregroundShow=false：App 在前台时必走
|
||
# onForegroundMessageArrived，正好就是本事件需要的刷新信号；不重复占用一次推送配额。
|
||
if vendor == "vivo":
|
||
return {"skipped": True, "reason": "foreground notification callback"}
|
||
|
||
# OPush 当前只支持通知栏消息，没有服务端透传单推接口。
|
||
# 客户端在 OPPO 首页可见时轮询未读数；这里必须安全跳过，不能请求不存在的
|
||
# /message/transparent/unicast（该地址会稳定返回 HTTP 404）。
|
||
if vendor == "oppo":
|
||
return {"skipped": True, "reason": "oppo does not support data messages"}
|
||
|
||
dispatch: dict[str, Callable[[str, dict[str, str]], dict[str, Any]]] = {
|
||
"honor": _send_honor_data,
|
||
"huawei": _send_huawei_data,
|
||
"xiaomi": _send_xiaomi_data,
|
||
}
|
||
return dispatch[vendor](token, payload)
|
||
|
||
|
||
def _require(value: str, name: str) -> str:
|
||
if not value:
|
||
raise VendorPushError(f"{name} not configured")
|
||
return value
|
||
|
||
|
||
def _request_json(
|
||
method: str,
|
||
url: str,
|
||
*,
|
||
expected_status: tuple[int, ...] = (200,),
|
||
**kwargs: Any,
|
||
) -> dict[str, Any]:
|
||
try:
|
||
resp = httpx.request(
|
||
method,
|
||
url,
|
||
timeout=settings.PUSH_REQUEST_TIMEOUT_SEC,
|
||
**kwargs,
|
||
)
|
||
except httpx.HTTPError as e:
|
||
raise VendorPushError(f"push http error: {e}") from e
|
||
|
||
if resp.status_code not in expected_status:
|
||
logger.error("vendor push http failed url=%s http=%s body=%s", url, resp.status_code, resp.text[:500])
|
||
raise VendorPushError(f"push http {resp.status_code}")
|
||
try:
|
||
return resp.json()
|
||
except ValueError as e:
|
||
raise VendorPushError(f"push invalid json: {resp.text[:200]}") from e
|
||
|
||
|
||
def _request_form(
|
||
method: str,
|
||
url: str,
|
||
*,
|
||
expected_status: tuple[int, ...] = (200,),
|
||
**kwargs: Any,
|
||
) -> dict[str, Any]:
|
||
try:
|
||
resp = httpx.request(
|
||
method,
|
||
url,
|
||
timeout=settings.PUSH_REQUEST_TIMEOUT_SEC,
|
||
**kwargs,
|
||
)
|
||
except httpx.HTTPError as e:
|
||
raise VendorPushError(f"push http error: {e}") from e
|
||
|
||
if resp.status_code not in expected_status:
|
||
logger.error("vendor push http failed url=%s http=%s body=%s", url, resp.status_code, resp.text[:500])
|
||
raise VendorPushError(f"push http {resp.status_code}")
|
||
try:
|
||
return resp.json()
|
||
except ValueError as e:
|
||
raise VendorPushError(f"push invalid json: {resp.text[:200]}") from e
|
||
|
||
|
||
def _cache_get(key: str) -> str | None:
|
||
cached = _token_cache.get(key)
|
||
if cached and cached.expires_at > time.time() + 60:
|
||
return cached.value
|
||
return None
|
||
|
||
|
||
def _cache_put(key: str, value: str, expires_in: int | float | None) -> str:
|
||
ttl = int(expires_in or 3600)
|
||
_token_cache[key] = _CachedToken(value=value, expires_at=time.time() + max(60, ttl - 60))
|
||
return value
|
||
|
||
|
||
def _honor_access_token() -> str:
|
||
cache_key = "honor"
|
||
cached = _cache_get(cache_key)
|
||
if cached:
|
||
return cached
|
||
client_id = _require(settings.HONOR_PUSH_CLIENT_ID, "HONOR_PUSH_CLIENT_ID")
|
||
client_secret = _require(settings.HONOR_PUSH_CLIENT_SECRET, "HONOR_PUSH_CLIENT_SECRET")
|
||
data = _request_form(
|
||
"POST",
|
||
settings.HONOR_PUSH_TOKEN_ENDPOINT,
|
||
data={
|
||
"grant_type": "client_credentials",
|
||
"client_id": client_id,
|
||
"client_secret": client_secret,
|
||
},
|
||
headers={"Content-Type": "application/x-www-form-urlencoded"},
|
||
)
|
||
token = data.get("access_token")
|
||
if not token:
|
||
raise VendorPushError(f"honor auth failed: {data}")
|
||
return _cache_put(cache_key, str(token), data.get("expires_in"))
|
||
|
||
|
||
def _send_honor(token: str, title: str, body: str, extras: dict[str, str]) -> dict[str, Any]:
|
||
app_id = _require(settings.HONOR_PUSH_APP_ID, "HONOR_PUSH_APP_ID")
|
||
access_token = _honor_access_token()
|
||
click_action: dict[str, Any]
|
||
if extras.get("notificationId"):
|
||
# type=3 只负责打开首页，不保证 data 会变成目标 Activity 的 extras。消息中心通知必须
|
||
# 用自定义页面(type=1)+ intent URI，把 notificationId/type/feedbackId 等直接带给
|
||
# MainActivity；否则华为/荣耀系统能展示通知，但用户点击后客户端收不到任何导航参数。
|
||
click_action = {"type": 1, "intent": _click_intent_uri(extras)}
|
||
else:
|
||
click_action = {"type": 3}
|
||
payload = {
|
||
"data": json.dumps(_click_extras(extras), ensure_ascii=False),
|
||
"notification": {"title": title, "body": body},
|
||
"android": {
|
||
"ttl": f"{settings.PUSH_TIME_TO_LIVE_SEC}s",
|
||
"targetUserType": 1,
|
||
"notification": {
|
||
"title": title,
|
||
"body": body,
|
||
"clickAction": click_action,
|
||
"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 _send_honor_data(token: str, payload: dict[str, str]) -> dict[str, Any]:
|
||
app_id = _require(settings.HONOR_PUSH_APP_ID, "HONOR_PUSH_APP_ID")
|
||
access_token = _honor_access_token()
|
||
data = _request_json(
|
||
"POST",
|
||
settings.HONOR_PUSH_SEND_ENDPOINT_TEMPLATE.format(app_id=app_id),
|
||
json={"data": json.dumps(payload, ensure_ascii=False), "token": [token]},
|
||
headers={
|
||
"Content-Type": "application/json; charset=UTF-8",
|
||
"Authorization": f"Bearer {access_token}",
|
||
"timestamp": str(int(time.time() * 1000)),
|
||
},
|
||
)
|
||
code = data.get("code")
|
||
if code is not None and int(code) != 200:
|
||
raise VendorPushError(f"honor data push failed: {data}")
|
||
return data
|
||
|
||
|
||
def _huawei_access_token() -> str:
|
||
"""华为 OAuth2 client_credentials 换 access_token(client_id 即 AGC 应用的 AppId)。"""
|
||
cache_key = "huawei"
|
||
cached = _cache_get(cache_key)
|
||
if cached:
|
||
return cached
|
||
app_id = _require(settings.HUAWEI_PUSH_APP_ID, "HUAWEI_PUSH_APP_ID")
|
||
app_secret = _require(settings.HUAWEI_PUSH_APP_SECRET, "HUAWEI_PUSH_APP_SECRET")
|
||
data = _request_form(
|
||
"POST",
|
||
settings.HUAWEI_PUSH_TOKEN_ENDPOINT,
|
||
data={
|
||
"grant_type": "client_credentials",
|
||
"client_id": app_id,
|
||
"client_secret": app_secret,
|
||
},
|
||
headers={"Content-Type": "application/x-www-form-urlencoded"},
|
||
)
|
||
token = data.get("access_token")
|
||
if not token:
|
||
raise VendorPushError(f"huawei auth failed: {data}")
|
||
return _cache_put(cache_key, str(token), data.get("expires_in"))
|
||
|
||
|
||
def _send_huawei(token: str, title: str, body: str, extras: dict[str, str]) -> dict[str, Any]:
|
||
"""华为 Push Kit 下行消息(v1 messages:send)。成功码 '80000000';
|
||
'80100000' 为部分成功(单 token 场景仍视为失败,错误里带原始响应便于排障)。"""
|
||
app_id = _require(settings.HUAWEI_PUSH_APP_ID, "HUAWEI_PUSH_APP_ID")
|
||
access_token = _huawei_access_token()
|
||
click_action: dict[str, Any]
|
||
if extras.get("notificationId"):
|
||
# type=3 只打开首页，Mate 20/EMUI 不会把 message.data 自动拆成启动 Intent extras。
|
||
# type=1 的自定义 intent 才能稳定携带反馈记录 id，且冷启动/onNewIntent 都走同一路由。
|
||
click_action = {"type": 1, "intent": _click_intent_uri(extras)}
|
||
else:
|
||
click_action = {"type": 3}
|
||
payload = {
|
||
"validate_only": False,
|
||
"message": {
|
||
"data": json.dumps(_click_extras(extras), ensure_ascii=False),
|
||
"android": {
|
||
"target_user_type": settings.HUAWEI_PUSH_TARGET_USER_TYPE,
|
||
"ttl": f"{settings.PUSH_TIME_TO_LIVE_SEC}s",
|
||
"notification": {
|
||
"title": title,
|
||
"body": body,
|
||
"click_action": click_action,
|
||
"importance": "NORMAL",
|
||
},
|
||
},
|
||
"token": [token],
|
||
},
|
||
}
|
||
data = _request_json(
|
||
"POST",
|
||
settings.HUAWEI_PUSH_SEND_ENDPOINT_TEMPLATE.format(app_id=app_id),
|
||
json=payload,
|
||
headers={
|
||
"Content-Type": "application/json; charset=UTF-8",
|
||
"Authorization": f"Bearer {access_token}",
|
||
},
|
||
)
|
||
if str(data.get("code", "")) != "80000000":
|
||
raise VendorPushError(f"huawei push failed: {data}")
|
||
return data
|
||
|
||
|
||
def _send_huawei_data(token: str, payload: dict[str, str]) -> dict[str, Any]:
|
||
app_id = _require(settings.HUAWEI_PUSH_APP_ID, "HUAWEI_PUSH_APP_ID")
|
||
access_token = _huawei_access_token()
|
||
data = _request_json(
|
||
"POST",
|
||
settings.HUAWEI_PUSH_SEND_ENDPOINT_TEMPLATE.format(app_id=app_id),
|
||
json={
|
||
"validate_only": False,
|
||
"message": {
|
||
"data": json.dumps(payload, ensure_ascii=False),
|
||
"token": [token],
|
||
},
|
||
},
|
||
headers={
|
||
"Content-Type": "application/json; charset=UTF-8",
|
||
"Authorization": f"Bearer {access_token}",
|
||
},
|
||
)
|
||
if str(data.get("code", "")) != "80000000":
|
||
raise VendorPushError(f"huawei data push failed: {data}")
|
||
return data
|
||
|
||
|
||
def _vivo_auth_token() -> str:
|
||
cache_key = "vivo"
|
||
cached = _cache_get(cache_key)
|
||
if cached:
|
||
return cached
|
||
app_id = _require(settings.VIVO_PUSH_APP_ID, "VIVO_PUSH_APP_ID")
|
||
app_key = _require(settings.VIVO_PUSH_APP_KEY, "VIVO_PUSH_APP_KEY")
|
||
app_secret = _require(settings.VIVO_PUSH_APP_SECRET, "VIVO_PUSH_APP_SECRET")
|
||
timestamp = str(int(time.time() * 1000))
|
||
sign = hashlib.md5(f"{app_id}{app_key}{timestamp}{app_secret}".encode()).hexdigest() # noqa: S324
|
||
data = _request_json(
|
||
"POST",
|
||
settings.VIVO_PUSH_AUTH_ENDPOINT,
|
||
json={
|
||
"appId": app_id,
|
||
"appKey": app_key,
|
||
"timestamp": timestamp,
|
||
"sign": sign,
|
||
},
|
||
headers={"Content-Type": "application/json"},
|
||
)
|
||
if int(data.get("result", -1)) != 0:
|
||
raise VendorPushError(f"vivo auth failed: {data}")
|
||
token = data.get("authToken")
|
||
if not token:
|
||
raise VendorPushError(f"vivo auth missing authToken: {data}")
|
||
return _cache_put(cache_key, str(token), 24 * 3600)
|
||
|
||
|
||
def _send_vivo(token: str, title: str, body: str, extras: dict[str, str]) -> dict[str, Any]:
|
||
app_id = _require(settings.VIVO_PUSH_APP_ID, "VIVO_PUSH_APP_ID")
|
||
auth_token = _vivo_auth_token()
|
||
payload: dict[str, Any] = {
|
||
"appId": app_id,
|
||
"regId": token,
|
||
"notifyType": settings.VIVO_PUSH_NOTIFY_TYPE,
|
||
"title": title,
|
||
"content": body,
|
||
"timeToLive": settings.PUSH_TIME_TO_LIVE_SEC,
|
||
"requestId": uuid.uuid4().hex,
|
||
"pushMode": settings.VIVO_PUSH_MODE,
|
||
# vivo SDK 4.1.5 只有在前台展示关闭时才调用
|
||
# OpenClientPushMessageReceiver.onForegroundMessageArrived。
|
||
# App 在该回调中刷新服务端未读数并补发一条本地通知；后台/锁屏时仍由
|
||
# vivo 系统展示，因而任何场景都只会出现一条通知。
|
||
"foregroundShow": False,
|
||
# vivo 官方 VPush 角标字段：离线收到通知时先由桌面自动 +1；
|
||
# App 启动/同步后再由统一未读数通过系统 API 精确校准。
|
||
"addBadge": True,
|
||
"clientCustomMap": extras,
|
||
}
|
||
# vivo Push SDK 480+ 已不再回调 skipType=3；必须使用 skipType=4 的完整 Intent URI。
|
||
# URI 同时包含 data/scheme、显式 component 和 S. extras，OriginOS 才会把业务参数
|
||
# 原样交给 MainActivity（仅靠隐式 deeplink 会打开 App，但可能剥掉 extras）。
|
||
if extras.get("notificationId"):
|
||
payload["skipType"] = 4
|
||
payload["skipContent"] = _vivo_click_intent_uri(extras)
|
||
else:
|
||
payload["skipType"] = 1
|
||
if settings.VIVO_PUSH_CATEGORY:
|
||
payload["category"] = settings.VIVO_PUSH_CATEGORY
|
||
data = _request_json(
|
||
"POST",
|
||
settings.VIVO_PUSH_SEND_ENDPOINT,
|
||
json=payload,
|
||
headers={
|
||
"Content-Type": "application/json",
|
||
"authToken": auth_token,
|
||
},
|
||
)
|
||
# 10089 = 当前 vivo 应用尚未开通 VPush「离线自动角标」能力。
|
||
# 角标只是通知中心未读数的镜像，不能因此让整条通知发送失败：去掉 addBadge
|
||
# 原样重试，通知到达/应用同步后仍由客户端系统 API 按统一未读数精确设置。
|
||
if int(data.get("result", -1)) == 10089:
|
||
logger.warning("vivo VPush badge not enabled (10089); retrying without addBadge")
|
||
fallback_payload = dict(payload)
|
||
fallback_payload.pop("addBadge", None)
|
||
fallback_payload["requestId"] = uuid.uuid4().hex
|
||
data = _request_json(
|
||
"POST",
|
||
settings.VIVO_PUSH_SEND_ENDPOINT,
|
||
json=fallback_payload,
|
||
headers={
|
||
"Content-Type": "application/json",
|
||
"authToken": auth_token,
|
||
},
|
||
)
|
||
if int(data.get("result", -1)) == 0:
|
||
data = {**data, "badgeFallback": True}
|
||
if int(data.get("result", -1)) != 0:
|
||
raise VendorPushError(f"vivo push failed: {data}")
|
||
return data
|
||
|
||
|
||
def _send_xiaomi(token: str, title: str, body: str, extras: dict[str, str]) -> dict[str, Any]:
|
||
app_secret = _require(settings.XIAOMI_PUSH_APP_SECRET, "XIAOMI_PUSH_APP_SECRET")
|
||
message_title = settings.XIAOMI_PUSH_TEMPLATE_TITLE.strip() or title
|
||
message_description = settings.XIAOMI_PUSH_TEMPLATE_DESCRIPTION.strip() or body
|
||
form = {
|
||
"registration_id": token,
|
||
"restricted_package_name": settings.ANDROID_PACKAGE_NAME,
|
||
"title": message_title,
|
||
"description": message_description,
|
||
"payload": json.dumps(extras, ensure_ascii=False),
|
||
"pass_through": "0",
|
||
"notify_type": "-1",
|
||
"time_to_live": str(settings.PUSH_TIME_TO_LIVE_SEC * 1000),
|
||
}
|
||
# 点击落地:带 notificationId 的消息中心推送 → notify_effect=2 + intent_uri,MiPush 直接打开
|
||
# MainActivity 并把 extras 作为 String extra 传入(客户端 MainActivity.consumeNavTarget 读
|
||
# notif_id/notif_type,兜底 notificationId/type)→ 置读 + 刷角标 + 按 type 直达对应页(PRD §5)。
|
||
# ⚠️ 早前用 notify_effect=1(仅打开 Launcher),小米自身不会把 payload 拆成普通 extra、而是塞进
|
||
# 序列化的 MiPushMessage(key_message),客户端读不到 → 点击后停在首页「没反应」。
|
||
# 无 notificationId 的系统召回类(如无障碍掉线)保持 notify_effect=1 仅拉起 App,行为不变。
|
||
if extras.get("notificationId"):
|
||
form["extra.notify_effect"] = "2"
|
||
form["extra.intent_uri"] = _click_intent_uri(extras)
|
||
else:
|
||
form["extra.notify_effect"] = "1"
|
||
if settings.XIAOMI_PUSH_CHANNEL_ID:
|
||
form["extra.channel_id"] = settings.XIAOMI_PUSH_CHANNEL_ID.strip()
|
||
if settings.XIAOMI_PUSH_TEMPLATE_ID:
|
||
form["extra.template_id"] = settings.XIAOMI_PUSH_TEMPLATE_ID.strip()
|
||
if settings.XIAOMI_PUSH_TEMPLATE_PARAM_JSON:
|
||
form["extra.template_param"] = _xiaomi_template_param(title, body)
|
||
data = _request_form(
|
||
"POST",
|
||
settings.XIAOMI_PUSH_SEND_ENDPOINT,
|
||
data=form,
|
||
headers={"Authorization": f"key={app_secret}"},
|
||
)
|
||
code = data.get("code")
|
||
if code not in (0, "0", None):
|
||
raise VendorPushError(f"xiaomi push failed: {data}")
|
||
if str(data.get("result", "ok")).lower() not in ("ok", "success"):
|
||
raise VendorPushError(f"xiaomi push failed: {data}")
|
||
return data
|
||
|
||
|
||
def _send_xiaomi_data(token: str, payload: dict[str, str]) -> dict[str, Any]:
|
||
app_secret = _require(settings.XIAOMI_PUSH_APP_SECRET, "XIAOMI_PUSH_APP_SECRET")
|
||
data = _request_form(
|
||
"POST",
|
||
settings.XIAOMI_PUSH_SEND_ENDPOINT,
|
||
data={
|
||
"registration_id": token,
|
||
"restricted_package_name": settings.ANDROID_PACKAGE_NAME,
|
||
"payload": json.dumps(payload, ensure_ascii=False),
|
||
"pass_through": "1",
|
||
"time_to_live": str(settings.PUSH_TIME_TO_LIVE_SEC * 1000),
|
||
},
|
||
headers={"Authorization": f"key={app_secret}"},
|
||
)
|
||
code = data.get("code")
|
||
if code not in (0, "0", None):
|
||
raise VendorPushError(f"xiaomi data push failed: {data}")
|
||
if str(data.get("result", "ok")).lower() not in ("ok", "success"):
|
||
raise VendorPushError(f"xiaomi data push failed: {data}")
|
||
return data
|
||
|
||
|
||
def _click_extras(extras: dict[str, str]) -> dict[str, str]:
|
||
"""点击落地参数:消息中心推送(extras 带 notificationId)补 notif_id/notif_type 别名——
|
||
客户端 MainActivity.consumeNavTarget 首选这两个键(厂商 receiver 路径的历史约定),原始键
|
||
(notificationId/type/feedbackId/reportId/…)保留作兜底与业务跳转参数。
|
||
无 notificationId(如无障碍召回)原样返回,不喂点击路由参数。"""
|
||
if not extras.get("notificationId"):
|
||
return dict(extras)
|
||
merged = dict(extras)
|
||
merged.setdefault("notif_id", extras["notificationId"])
|
||
if extras.get("type"):
|
||
merged.setdefault("notif_type", extras["type"])
|
||
return merged
|
||
|
||
|
||
def _click_intent_uri(extras: dict[str, str]) -> str:
|
||
"""构造「系统直启 MainActivity 并带 extras」的 intent uri(小米 notify_effect=2 的
|
||
extra.intent_uri、vivo skipType=4 的 skipContent 共用):点击后厂商系统用 Intent.parseUri
|
||
解析并 startActivity,extras 作为 String extra 原样送达。
|
||
|
||
- component 显式指向本包 MainActivity(exported=true、singleTask)→ 已运行则走 onNewIntent、
|
||
未运行则 onCreate,两条都会执行 consumeNavTarget。
|
||
- 参数 = _click_extras(补 notif_id/notif_type 别名 + 透传 feedbackId/reportId 等跳转参数)。
|
||
- 值按 Android Uri.encode 规则百分号编码(quote(safe="")):中文/分号/等号都不会破坏 intent uri
|
||
结构;客户端 Intent.parseUri 侧 Uri.decode 无损还原。表单/JSON 传输层的编码与本层相互独立、
|
||
各自解码,不会双重转义(2026-07-15 小米联调结论)。
|
||
"""
|
||
pkg = settings.ANDROID_PACKAGE_NAME
|
||
parts = ["intent:#Intent", f"component={pkg}/{pkg}.MainActivity"]
|
||
parts += [f"S.{key}={quote(str(value), safe='')}" for key, value in _click_extras(extras).items()]
|
||
parts.append("end")
|
||
return ";".join(parts)
|
||
|
||
|
||
def _vivo_click_intent_uri(extras: dict[str, str]) -> str:
|
||
"""vivo skipType=4 使用官方要求的完整 intent deeplink 形态。
|
||
|
||
`?#Intent` 中的问号不能省；OriginOS 对缺少它的 URI 会退化为“仅打开应用”，
|
||
不把 S. 参数交给目标 Activity。
|
||
"""
|
||
pkg = settings.ANDROID_PACKAGE_NAME
|
||
parts = [
|
||
"intent://push/detail?#Intent",
|
||
"scheme=shaguabijia",
|
||
f"component={pkg}/{pkg}.MainActivity",
|
||
"launchFlags=0x14000000",
|
||
]
|
||
parts += [f"S.{key}={quote(str(value), safe='')}" for key, value in _click_extras(extras).items()]
|
||
parts.append("end")
|
||
return ";".join(parts)
|
||
|
||
|
||
def _xiaomi_template_param(title: str, alert: str) -> str:
|
||
rendered = (
|
||
settings.XIAOMI_PUSH_TEMPLATE_PARAM_JSON
|
||
.replace("{title}", title)
|
||
.replace("{alert}", alert)
|
||
)
|
||
try:
|
||
payload = json.loads(rendered)
|
||
except ValueError as e:
|
||
raise VendorPushError("XIAOMI_PUSH_TEMPLATE_PARAM_JSON invalid json") from e
|
||
if not isinstance(payload, dict):
|
||
raise VendorPushError("XIAOMI_PUSH_TEMPLATE_PARAM_JSON must be a json object")
|
||
for key, value in payload.items():
|
||
if not isinstance(key, str) or not isinstance(value, str):
|
||
raise VendorPushError("xiaomi template params must be string key-value pairs")
|
||
if not value.strip() or len(value) > 128:
|
||
raise VendorPushError("xiaomi template param value length must be 1-128")
|
||
return json.dumps(payload, ensure_ascii=False, separators=(",", ":"))
|
||
|
||
|
||
def _oppo_auth_token() -> str:
|
||
cache_key = "oppo"
|
||
cached = _cache_get(cache_key)
|
||
if cached:
|
||
return cached
|
||
app_key = _require(settings.OPPO_PUSH_APP_KEY, "OPPO_PUSH_APP_KEY")
|
||
master_secret = _require(settings.OPPO_PUSH_MASTER_SECRET, "OPPO_PUSH_MASTER_SECRET")
|
||
timestamp = str(int(time.time() * 1000))
|
||
sign = hashlib.sha256(f"{app_key}{timestamp}{master_secret}".encode()).hexdigest()
|
||
data = _request_form(
|
||
"POST",
|
||
settings.OPPO_PUSH_AUTH_ENDPOINT,
|
||
data={
|
||
"app_key": app_key,
|
||
"timestamp": timestamp,
|
||
"sign": sign,
|
||
},
|
||
headers={"Content-Type": "application/x-www-form-urlencoded"},
|
||
)
|
||
if int(data.get("code", -1)) != 0:
|
||
raise VendorPushError(f"oppo auth failed: {data}")
|
||
token = (data.get("data") or {}).get("auth_token") or data.get("auth_token")
|
||
if not token:
|
||
raise VendorPushError(f"oppo auth missing auth_token: {data}")
|
||
return _cache_put(cache_key, str(token), 24 * 3600)
|
||
|
||
|
||
def _send_oppo(token: str, title: str, body: str, extras: dict[str, str]) -> dict[str, Any]:
|
||
auth_token = _oppo_auth_token()
|
||
ttl_hours = max(1, min(72, settings.PUSH_TIME_TO_LIVE_SEC // 3600))
|
||
notification: dict[str, Any] = {
|
||
"app_message_id": f"{extras.get('type', 'notify')}_{uuid.uuid4().hex}",
|
||
"title": title,
|
||
"content": body,
|
||
"off_line": True,
|
||
"off_line_ttl": ttl_hours,
|
||
"action_parameters": json.dumps(_click_extras(extras), ensure_ascii=False),
|
||
}
|
||
if extras.get("notificationId"):
|
||
# 只有已落库的站内消息才计入角标；保活提醒等临时推送不应留下无法消除的未读数。
|
||
# 客户端打开/已读后会按服务端真实未读总数覆盖，避免长期漂移。
|
||
notification["badge_operation_type"] = 1
|
||
notification["badge_message_count"] = 1
|
||
# 点击落地:OPPO SDK 没有点击回调,参数只能靠服务端点击动作配置送达——action_parameters 的
|
||
# 键值对仅在 click_action_type=1/4 时才会注入目标 Activity 的 intent extras(type=0「启动应用」
|
||
# 会忽略它,extras 全丢 → 点了没反应,与小米 notify_effect=1 同款坑)。
|
||
# 消息中心推送(带 notificationId)→ type=4(打开应用内页面,Activity 全路径,exported=true);
|
||
# 无 notificationId 的召回类保持 type=0 仅打开应用。
|
||
if extras.get("notificationId"):
|
||
notification["click_action_type"] = 4
|
||
notification["click_action_activity"] = f"{settings.ANDROID_PACKAGE_NAME}.MainActivity"
|
||
else:
|
||
notification["click_action_type"] = 0
|
||
# 新消息分类(2024-11-20 后创建的 OPPO 应用必须带 category,否则可能被拒收/降级)
|
||
if settings.OPPO_PUSH_CHANNEL_ID.strip():
|
||
notification["channel_id"] = settings.OPPO_PUSH_CHANNEL_ID.strip()
|
||
if settings.OPPO_PUSH_CATEGORY.strip():
|
||
notification["category"] = settings.OPPO_PUSH_CATEGORY.strip()
|
||
if settings.OPPO_PUSH_NOTIFY_LEVEL:
|
||
notification["notify_level"] = settings.OPPO_PUSH_NOTIFY_LEVEL
|
||
message = {
|
||
"target_type": 2,
|
||
"target_value": token,
|
||
"notification": notification,
|
||
}
|
||
data = _request_form(
|
||
"POST",
|
||
settings.OPPO_PUSH_SEND_ENDPOINT,
|
||
data={
|
||
"auth_token": auth_token,
|
||
"message": json.dumps(message, ensure_ascii=False),
|
||
},
|
||
headers={"Content-Type": "application/x-www-form-urlencoded"},
|
||
)
|
||
if int(data.get("code", -1)) != 0:
|
||
raise VendorPushError(f"oppo push failed: {data}")
|
||
return data
|