Compare commits

...

7 Commits

Author SHA1 Message Date
exinglang d484bd5801 Merge remote-tracking branch 'origin/main' into fix-mate20PageError
# Conflicts:
#	app/integrations/vendor_push.py
2026-07-30 10:44:15 +08:00
zuochenyong 675c7ecf81 fix(push): 完善厂商推送排障日志 (#198)
Co-authored-by: exinglang <exinglang@qq.com>
Reviewed-on: #198
Co-authored-by: zuochenyong <zuochenyong@wonderable.ai>
Co-committed-by: zuochenyong <zuochenyong@wonderable.ai>
2026-07-30 10:41:43 +08:00
exinglang 7474c2bebc fix(push): 补齐厂商请求日志参数 2026-07-30 10:26:26 +08:00
exinglang 7bfe703760 chore(tests): 取消推送测试文件改动 2026-07-30 10:25:05 +08:00
exinglang 69eeb43fe2 fix(push): 完善厂商推送参数与消息通知 2026-07-30 10:25:05 +08:00
exinglang a0196b8d64 fix(push): 完善厂商推送排障日志 2026-07-29 19:52:06 +08:00
zuochenyong 53c3b7f60f feat(guide-video): 支持领券和比价独立视频奖励配置 (#196)
Co-authored-by: exinglang <exinglang@qq.com>
Reviewed-on: #196
Co-authored-by: zuochenyong <zuochenyong@wonderable.ai>
Co-committed-by: zuochenyong <zuochenyong@wonderable.ai>
2026-07-29 16:12:05 +08:00
12 changed files with 666 additions and 86 deletions
+6
View File
@@ -175,3 +175,9 @@ PANGLE_REPORT_SITE_ID_TEST=5832303
# APPLOG_MAX_BATCH=500 # 单批最大条数(超 → 422;导入期常量,改需重启) # APPLOG_MAX_BATCH=500 # 单批最大条数(超 → 422;导入期常量,改需重启)
# APPLOG_MAX_BODY_BYTES=2097152 # 请求体上限 2MB(超 → 413;运行期可调) # APPLOG_MAX_BODY_BYTES=2097152 # 请求体上限 2MB(超 → 413;运行期可调)
# APPLOG_MAX_MSG_BYTES=8192 # 单条 msg 超此字节数截断 # APPLOG_MAX_MSG_BYTES=8192 # 单条 msg 超此字节数截断
# ===== 华为 Push =====
HUAWEI_PUSH_APP_ID=
HUAWEI_PUSH_APP_SECRET=
# 0=正式消息(默认);1=测试消息(仅开发联调,生产必须保持 0)
HUAWEI_PUSH_TARGET_USER_TYPE=0
@@ -0,0 +1,31 @@
"""guide video play count is independent for coupon and comparison
Revision ID: guide_video_scene_unique
Revises: 6d2309208549
"""
from alembic import op
revision = "guide_video_scene_unique"
down_revision = "6d2309208549"
branch_labels = None
depends_on = None
def upgrade() -> None:
op.drop_index("uq_guide_video_play_user_seq", table_name="guide_video_play")
op.create_index(
"uq_guide_video_play_user_scene_seq",
"guide_video_play",
["user_id", "scene", "seq"],
unique=True,
)
def downgrade() -> None:
op.drop_index("uq_guide_video_play_user_scene_seq", table_name="guide_video_play")
op.create_index(
"uq_guide_video_play_user_seq",
"guide_video_play",
["user_id", "seq"],
unique=True,
)
+29 -14
View File
@@ -9,7 +9,7 @@ client_max_body_size,见 shaguabijia-admin-web/deploy/nginx/admin.shaguabijia.co
""" """
from __future__ import annotations from __future__ import annotations
from typing import Annotated from typing import Annotated, Literal
from fastapi import APIRouter, Depends, File, HTTPException, Request, UploadFile from fastapi import APIRouter, Depends, File, HTTPException, Request, UploadFile
@@ -27,14 +27,21 @@ router = APIRouter(
) )
def _out(db: AdminDb) -> GuideVideoConfigOut: GuideScene = Literal["coupon", "comparison"]
def _out(db: AdminDb, scene: GuideScene) -> GuideVideoConfigOut:
"""配置 + 播放统计合成响应(四个写接口都以最新状态返回,前端一次同步到位)。""" """配置 + 播放统计合成响应(四个写接口都以最新状态返回,前端一次同步到位)。"""
return GuideVideoConfigOut(**guide_video.get_config(db), **guide_video.play_stats(db)) return GuideVideoConfigOut(
scene=scene,
**guide_video.get_config(db, scene),
**guide_video.play_stats(db, scene),
)
@router.get("", response_model=GuideVideoConfigOut, summary="新手引导视频配置(领券浮层)") @router.get("", response_model=GuideVideoConfigOut, summary="新手引导视频配置(领券浮层)")
def get_config(db: AdminDb) -> GuideVideoConfigOut: def get_config(db: AdminDb, scene: GuideScene = "coupon") -> GuideVideoConfigOut:
return _out(db) return _out(db, scene)
@router.patch("", response_model=GuideVideoConfigOut, summary="改开关/次数/金币(带审计)") @router.patch("", response_model=GuideVideoConfigOut, summary="改开关/次数/金币(带审计)")
@@ -43,21 +50,23 @@ def update_config(
request: Request, request: Request,
admin: Annotated[AdminUser, Depends(require_role("operator"))], admin: Annotated[AdminUser, Depends(require_role("operator"))],
db: AdminDb, db: AdminDb,
scene: GuideScene = "coupon",
) -> GuideVideoConfigOut: ) -> GuideVideoConfigOut:
before, after = guide_video.update_config( before, after = guide_video.update_config(
db, db,
enabled=body.enabled, enabled=body.enabled,
max_plays=body.max_plays, max_plays=body.max_plays,
reward_coin=body.reward_coin, reward_coin=body.reward_coin,
scene=scene,
admin_id=admin.id, admin_id=admin.id,
commit=False, commit=False,
) )
write_audit( write_audit(
db, admin, action="guide_video.update", target_type="guide_video", target_id=None, db, admin, action="guide_video.update", target_type="guide_video", target_id=None,
detail={"before": before, "after": after}, ip=get_client_ip(request), commit=False, detail={"scene": scene, "before": before, "after": after}, ip=get_client_ip(request), commit=False,
) )
db.commit() db.commit()
return _out(db) return _out(db, scene)
@router.post("/video", response_model=GuideVideoConfigOut, summary="上传新手引导视频(MP4,带审计)") @router.post("/video", response_model=GuideVideoConfigOut, summary="上传新手引导视频(MP4,带审计)")
@@ -65,23 +74,26 @@ async def upload_video(
request: Request, request: Request,
admin: Annotated[AdminUser, Depends(require_role("operator"))], admin: Annotated[AdminUser, Depends(require_role("operator"))],
db: AdminDb, db: AdminDb,
file: UploadFile = File(...), file: Annotated[UploadFile, File()],
scene: GuideScene = "coupon",
) -> GuideVideoConfigOut: ) -> GuideVideoConfigOut:
data = await file.read() data = await file.read()
try: try:
url = media.save_guide_video(data) url = media.save_guide_video(data)
except media.MediaError as e: except media.MediaError as e:
raise HTTPException(status_code=400, detail=str(e)) from e raise HTTPException(status_code=400, detail=str(e)) from e
before, after = guide_video.set_video(db, url, admin_id=admin.id, commit=False) before, after = guide_video.set_video(
db, url, scene=scene, admin_id=admin.id, commit=False
)
write_audit( write_audit(
db, admin, action="guide_video.set_video", target_type="guide_video", target_id=None, db, admin, action="guide_video.set_video", target_type="guide_video", target_id=None,
detail={"before": before.get("video_url"), "after": url, "bytes": len(data)}, detail={"scene": scene, "before": before.get("video_url"), "after": url, "bytes": len(data)},
ip=get_client_ip(request), commit=False, ip=get_client_ip(request), commit=False,
) )
db.commit() db.commit()
# 提交成功后再删旧片,避免新片没落库就把旧片丢了 # 提交成功后再删旧片,避免新片没落库就把旧片丢了
media.delete_guide_video(before.get("video_url")) media.delete_guide_video(before.get("video_url"))
return _out(db) return _out(db, scene)
@router.delete("/video", response_model=GuideVideoConfigOut, summary="移除新手引导视频(带审计)") @router.delete("/video", response_model=GuideVideoConfigOut, summary="移除新手引导视频(带审计)")
@@ -89,13 +101,16 @@ def delete_video(
request: Request, request: Request,
admin: Annotated[AdminUser, Depends(require_role("operator"))], admin: Annotated[AdminUser, Depends(require_role("operator"))],
db: AdminDb, db: AdminDb,
scene: GuideScene = "coupon",
) -> GuideVideoConfigOut: ) -> GuideVideoConfigOut:
"""移除后 /guide-video/start 一律返回 should_play=false,领券浮层回到「只放广告」。""" """移除后 /guide-video/start 一律返回 should_play=false,领券浮层回到「只放广告」。"""
before, after = guide_video.set_video(db, None, admin_id=admin.id, commit=False) before, after = guide_video.set_video(
db, None, scene=scene, admin_id=admin.id, commit=False
)
write_audit( write_audit(
db, admin, action="guide_video.delete_video", target_type="guide_video", target_id=None, db, admin, action="guide_video.delete_video", target_type="guide_video", target_id=None,
detail={"before": before.get("video_url")}, ip=get_client_ip(request), commit=False, detail={"scene": scene, "before": before.get("video_url")}, ip=get_client_ip(request), commit=False,
) )
db.commit() db.commit()
media.delete_guide_video(before.get("video_url")) media.delete_guide_video(before.get("video_url"))
return _out(db) return _out(db, scene)
+1
View File
@@ -7,6 +7,7 @@ from app.repositories.guide_video import MAX_PLAYS_LIMIT, REWARD_COIN_LIMIT
class GuideVideoConfigOut(BaseModel): class GuideVideoConfigOut(BaseModel):
scene: str
enabled: bool enabled: bool
video_url: str | None = None # 相对地址 /media/guide_video/xxx.mp4;未配片 = None video_url: str | None = None # 相对地址 /media/guide_video/xxx.mp4;未配片 = None
max_plays: int max_plays: int
+2
View File
@@ -88,6 +88,8 @@ class Settings(BaseSettings):
# (OAuth 换 token 时 client_id 即 AppId)。发送走 v1 messages:send,成功码 80000000。 # (OAuth 换 token 时 client_id 即 AppId)。发送走 v1 messages:send,成功码 80000000。
HUAWEI_PUSH_APP_ID: str = "" HUAWEI_PUSH_APP_ID: str = ""
HUAWEI_PUSH_APP_SECRET: str = "" HUAWEI_PUSH_APP_SECRET: str = ""
# 0=正式消息(默认,受正式消息频控);1=测试消息(仅开发联调,勿用于生产)。
HUAWEI_PUSH_TARGET_USER_TYPE: int = Field(default=0, ge=0, le=1)
HUAWEI_PUSH_TOKEN_ENDPOINT: str = "https://oauth-login.cloud.huawei.com/oauth2/v3/token" HUAWEI_PUSH_TOKEN_ENDPOINT: str = "https://oauth-login.cloud.huawei.com/oauth2/v3/token"
HUAWEI_PUSH_SEND_ENDPOINT_TEMPLATE: str = ( HUAWEI_PUSH_SEND_ENDPOINT_TEMPLATE: str = (
"https://push-api.cloud.huawei.com/v1/{app_id}/messages:send" "https://push-api.cloud.huawei.com/v1/{app_id}/messages:send"
+430 -46
View File
@@ -22,7 +22,7 @@ import uuid
from collections.abc import Callable from collections.abc import Callable
from dataclasses import dataclass from dataclasses import dataclass
from typing import Any from typing import Any
from urllib.parse import quote from urllib.parse import quote, urlsplit
import httpx import httpx
@@ -31,6 +31,7 @@ from app.core.config import settings
logger = logging.getLogger("shagua.vendor_push") logger = logging.getLogger("shagua.vendor_push")
TYPE_ACCESSIBILITY_DISABLED = "accessibility_disabled" TYPE_ACCESSIBILITY_DISABLED = "accessibility_disabled"
DATA_EVENT_NOTIFICATION_CREATED = "notification_created"
SUPPORTED_VENDORS = frozenset({"honor", "huawei", "vivo", "xiaomi", "oppo"}) SUPPORTED_VENDORS = frozenset({"honor", "huawei", "vivo", "xiaomi", "oppo"})
# vendor key → 中文名(测试/配置状态接口展示用) # vendor key → 中文名(测试/配置状态接口展示用)
@@ -69,6 +70,18 @@ class _CachedToken:
_token_cache: dict[str, _CachedToken] = {} _token_cache: dict[str, _CachedToken] = {}
_LOG_SUMMARY_MAX_CHARS = 1500
_LOG_STRING_MAX_CHARS = 200
_SENSITIVE_LOG_KEYS = {
"accesstoken",
"appkey",
"authtoken",
"authorization",
"clientsecret",
"mastersecret",
"sign",
}
def normalize_vendor(push_vendor: str | None) -> str | None: def normalize_vendor(push_vendor: str | None) -> str | None:
if not push_vendor: if not push_vendor:
@@ -115,8 +128,16 @@ def send_notification(
if mock: if mock:
logger.info( logger.info(
"[mock push] vendor=%s token=%s... title=%s body=%s extras=%s", "vendor push mock vendor=%s request=%s",
vendor, token[:12], title, body, extras, vendor,
_log_summary(
{
"push_token": token,
"title": title,
"body": body,
"extras": extras,
}
),
) )
return { return {
"mock": True, "mock": True,
@@ -133,7 +154,28 @@ def send_notification(
"xiaomi": _send_xiaomi, "xiaomi": _send_xiaomi,
"oppo": _send_oppo, "oppo": _send_oppo,
} }
return dispatch[vendor](token, title, body, extras) started = time.perf_counter()
try:
result = dispatch[vendor](token, title, body, extras)
except Exception:
logger.exception(
"vendor push dispatch failed vendor=%s token=%s notification_type=%s "
"elapsed_ms=%.1f",
vendor,
token,
extras.get("type", ""),
_elapsed_ms(started),
)
raise
logger.info(
"vendor push dispatch succeeded vendor=%s token=%s notification_type=%s "
"elapsed_ms=%.1f",
vendor,
token,
extras.get("type", ""),
_elapsed_ms(started),
)
return result
def send_accessibility_disabled( def send_accessibility_disabled(
@@ -153,19 +195,169 @@ def send_accessibility_disabled(
) )
def send_data_event(
push_vendor: str,
push_token: str,
*,
event: str,
notification_id: str,
mock: bool = False,
) -> dict[str, Any]:
"""发送无界面的轻量事件;当前只允许站内消息创建事件。"""
vendor = normalize_vendor(push_vendor)
token = push_token.strip() if push_token else ""
if not vendor or vendor not in SUPPORTED_VENDORS:
raise VendorPushError(f"unsupported push vendor: {push_vendor}")
if not token:
raise VendorPushError("push token is empty")
if event != DATA_EVENT_NOTIFICATION_CREATED:
raise VendorPushError(f"unsupported data event: {event}")
if not notification_id:
raise VendorPushError("notification_id is empty")
payload = {"event": event, "notificationId": str(notification_id)}
if mock:
return {"mock": True, "vendor": vendor, "payload": payload}
# vivo 的通知已设置 foregroundShow=falseApp 在前台时必走
# 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: def _require(value: str, name: str) -> str:
if not value: if not value:
raise VendorPushError(f"{name} not configured") raise VendorPushError(f"{name} not configured")
return value return value
def _elapsed_ms(started: float) -> float:
return (time.perf_counter() - started) * 1000
def _redacted_value(value: Any) -> str:
raw = json.dumps(value, ensure_ascii=False, sort_keys=True, default=str)
digest = hashlib.sha256(raw.encode()).hexdigest()[:10]
return f"<redacted len={len(raw)} sha256={digest}>"
def _sanitize_for_log(value: Any, *, key: str = "", depth: int = 0) -> Any:
"""生成有排障价值、但不暴露服务端鉴权凭据的紧凑日志摘要。"""
normalized_key = "".join(char for char in key.lower() if char.isalnum())
if normalized_key in _SENSITIVE_LOG_KEYS or normalized_key.endswith("secret"):
return _redacted_value(value)
if depth >= 10:
return f"<{type(value).__name__}>"
if isinstance(value, dict):
items = list(value.items())
sanitized = {
str(item_key): _sanitize_for_log(item_value, key=str(item_key), depth=depth + 1)
for item_key, item_value in items[:30]
}
if len(items) > 30:
sanitized["<omitted_keys>"] = len(items) - 30
return sanitized
if isinstance(value, (list, tuple)):
items = list(value)
sanitized = [_sanitize_for_log(item, key=key, depth=depth + 1) for item in items[:20]]
if len(items) > 20:
sanitized.append(f"<omitted_items={len(items) - 20}>")
return sanitized
if isinstance(value, str):
if value.lstrip().startswith(("{", "[")):
try:
decoded = json.loads(value)
except ValueError:
pass
else:
return _sanitize_for_log(decoded, key=key, depth=depth + 1)
if len(value) > _LOG_STRING_MAX_CHARS:
return f"{value[:_LOG_STRING_MAX_CHARS]}…<len={len(value)}>"
return value
def _log_summary(value: Any) -> str:
rendered = json.dumps(
_sanitize_for_log(value),
ensure_ascii=False,
sort_keys=True,
separators=(",", ":"),
default=str,
)
if len(rendered) > _LOG_SUMMARY_MAX_CHARS:
return f"{rendered[:_LOG_SUMMARY_MAX_CHARS]}…<len={len(rendered)}>"
return rendered
def _raw_log_summary(value: Any) -> str:
"""厂商响应摘要:保留原始字段和值,仅限制单条日志长度。"""
rendered = json.dumps(
value,
ensure_ascii=False,
sort_keys=True,
separators=(",", ":"),
default=str,
)
if len(rendered) > _LOG_SUMMARY_MAX_CHARS:
return f"{rendered[:_LOG_SUMMARY_MAX_CHARS]}…<len={len(rendered)}>"
return rendered
def _endpoint_for_log(url: str) -> str:
parsed = urlsplit(url)
return f"{parsed.scheme}://{parsed.netloc}{parsed.path}"
def _request_summary(kwargs: dict[str, Any]) -> str:
summary: dict[str, Any] = {}
if "json" in kwargs:
summary["body_type"] = "json"
summary["request_payload"] = kwargs["json"]
elif "data" in kwargs:
summary["body_type"] = "form"
summary["request_payload"] = kwargs["data"]
if kwargs.get("params"):
summary["params"] = kwargs["params"]
if kwargs.get("headers"):
summary["headers"] = kwargs["headers"]
return _log_summary(summary)
def _response_summary(resp: Any, parsed: Any | None = None) -> str:
if parsed is None:
try:
parsed = resp.json()
except ValueError:
pass
if parsed is not None:
return _raw_log_summary(parsed)
return _raw_log_summary(getattr(resp, "text", ""))
def _request_json( def _request_json(
method: str, method: str,
url: str, url: str,
*, *,
vendor: str,
operation: str,
expected_status: tuple[int, ...] = (200,), expected_status: tuple[int, ...] = (200,),
**kwargs: Any, **kwargs: Any,
) -> dict[str, Any]: ) -> dict[str, Any]:
started = time.perf_counter()
request_summary = _request_summary(kwargs)
try: try:
resp = httpx.request( resp = httpx.request(
method, method,
@@ -174,41 +366,82 @@ def _request_json(
**kwargs, **kwargs,
) )
except httpx.HTTPError as e: except httpx.HTTPError as e:
logger.error(
"vendor push http completed vendor=%s operation=%s method=%s endpoint=%s "
"outcome=network_error request=%s response=%s elapsed_ms=%.1f error=%s",
vendor,
operation,
method,
_endpoint_for_log(url),
request_summary,
"<no_response>",
_elapsed_ms(started),
type(e).__name__,
)
raise VendorPushError(f"push http error: {e}") from e raise VendorPushError(f"push http error: {e}") from e
if resp.status_code not in expected_status: if resp.status_code not in expected_status:
logger.error("vendor push http failed url=%s http=%s body=%s", url, resp.status_code, resp.text[:500]) logger.error(
"vendor push http completed vendor=%s operation=%s method=%s endpoint=%s "
"outcome=http_error http_status=%s request=%s response=%s elapsed_ms=%.1f",
vendor,
operation,
method,
_endpoint_for_log(url),
resp.status_code,
request_summary,
_response_summary(resp),
_elapsed_ms(started),
)
raise VendorPushError(f"push http {resp.status_code}") raise VendorPushError(f"push http {resp.status_code}")
try: try:
return resp.json() data = resp.json()
except ValueError as e: except ValueError as e:
logger.error(
"vendor push http completed vendor=%s operation=%s method=%s endpoint=%s "
"outcome=invalid_json http_status=%s request=%s response=%s elapsed_ms=%.1f",
vendor,
operation,
method,
_endpoint_for_log(url),
resp.status_code,
request_summary,
_response_summary(resp),
_elapsed_ms(started),
)
raise VendorPushError(f"push invalid json: {resp.text[:200]}") from e raise VendorPushError(f"push invalid json: {resp.text[:200]}") from e
logger.info(
"vendor push http completed vendor=%s operation=%s method=%s endpoint=%s "
"outcome=success http_status=%s request=%s response=%s elapsed_ms=%.1f",
vendor,
operation,
method,
_endpoint_for_log(url),
resp.status_code,
request_summary,
_response_summary(resp, data),
_elapsed_ms(started),
)
return data
def _request_form( def _request_form(
method: str, method: str,
url: str, url: str,
*, *,
vendor: str,
operation: str,
expected_status: tuple[int, ...] = (200,), expected_status: tuple[int, ...] = (200,),
**kwargs: Any, **kwargs: Any,
) -> dict[str, Any]: ) -> dict[str, Any]:
try: return _request_json(
resp = httpx.request( method,
method, url,
url, vendor=vendor,
timeout=settings.PUSH_REQUEST_TIMEOUT_SEC, operation=operation,
**kwargs, expected_status=expected_status,
) **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: def _cache_get(key: str) -> str | None:
@@ -234,6 +467,8 @@ def _honor_access_token() -> str:
data = _request_form( data = _request_form(
"POST", "POST",
settings.HONOR_PUSH_TOKEN_ENDPOINT, settings.HONOR_PUSH_TOKEN_ENDPOINT,
vendor="honor",
operation="authenticate",
data={ data={
"grant_type": "client_credentials", "grant_type": "client_credentials",
"client_id": client_id, "client_id": client_id,
@@ -243,16 +478,22 @@ def _honor_access_token() -> str:
) )
token = data.get("access_token") token = data.get("access_token")
if not token: if not token:
raise VendorPushError(f"honor auth failed: {data}") raise VendorPushError(f"honor auth failed: {_raw_log_summary(data)}")
return _cache_put(cache_key, str(token), data.get("expires_in")) 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]: 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") app_id = _require(settings.HONOR_PUSH_APP_ID, "HONOR_PUSH_APP_ID")
access_token = _honor_access_token() 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 = { payload = {
# clickAction type=3(打开应用首页)时,荣耀点击会把 data JSON 的键值对注入启动 intent 的
# extras(与 HMS 同机制)→ MainActivity.consumeNavTarget 读 notif_id/notif_type 直达落地。
"data": json.dumps(_click_extras(extras), ensure_ascii=False), "data": json.dumps(_click_extras(extras), ensure_ascii=False),
"notification": {"title": title, "body": body}, "notification": {"title": title, "body": body},
"android": { "android": {
@@ -261,7 +502,7 @@ def _send_honor(token: str, title: str, body: str, extras: dict[str, str]) -> di
"notification": { "notification": {
"title": title, "title": title,
"body": body, "body": body,
"clickAction": {"type": 3}, "clickAction": click_action,
"importance": "NORMAL", "importance": "NORMAL",
}, },
}, },
@@ -270,6 +511,8 @@ def _send_honor(token: str, title: str, body: str, extras: dict[str, str]) -> di
data = _request_json( data = _request_json(
"POST", "POST",
settings.HONOR_PUSH_SEND_ENDPOINT_TEMPLATE.format(app_id=app_id), settings.HONOR_PUSH_SEND_ENDPOINT_TEMPLATE.format(app_id=app_id),
vendor="honor",
operation="send_notification",
json=payload, json=payload,
headers={ headers={
"Content-Type": "application/json; charset=UTF-8", "Content-Type": "application/json; charset=UTF-8",
@@ -279,7 +522,28 @@ def _send_honor(token: str, title: str, body: str, extras: dict[str, str]) -> di
) )
code = data.get("code") code = data.get("code")
if code is not None and int(code) != 200: if code is not None and int(code) != 200:
raise VendorPushError(f"honor push failed: {data}") raise VendorPushError(f"honor push failed: {_raw_log_summary(data)}")
return data
def _send_honor_data(token: str, payload: dict[str, str]) -> dict[str, Any]:
app_id = _require(settings.HONOR_PUSH_APP_ID, "HONOR_PUSH_APP_ID")
access_token = _honor_access_token()
data = _request_json(
"POST",
settings.HONOR_PUSH_SEND_ENDPOINT_TEMPLATE.format(app_id=app_id),
vendor="honor",
operation="send_data_event",
json={"data": json.dumps(payload, ensure_ascii=False), "token": [token]},
headers={
"Content-Type": "application/json; charset=UTF-8",
"Authorization": f"Bearer {access_token}",
"timestamp": str(int(time.time() * 1000)),
},
)
code = data.get("code")
if code is not None and int(code) != 200:
raise VendorPushError(f"honor data push failed: {_raw_log_summary(data)}")
return data return data
@@ -294,6 +558,8 @@ def _huawei_access_token() -> str:
data = _request_form( data = _request_form(
"POST", "POST",
settings.HUAWEI_PUSH_TOKEN_ENDPOINT, settings.HUAWEI_PUSH_TOKEN_ENDPOINT,
vendor="huawei",
operation="authenticate",
data={ data={
"grant_type": "client_credentials", "grant_type": "client_credentials",
"client_id": app_id, "client_id": app_id,
@@ -303,7 +569,7 @@ def _huawei_access_token() -> str:
) )
token = data.get("access_token") token = data.get("access_token")
if not token: if not token:
raise VendorPushError(f"huawei auth failed: {data}") raise VendorPushError(f"huawei auth failed: {_raw_log_summary(data)}")
return _cache_put(cache_key, str(token), data.get("expires_in")) return _cache_put(cache_key, str(token), data.get("expires_in"))
@@ -312,18 +578,24 @@ def _send_huawei(token: str, title: str, body: str, extras: dict[str, str]) -> d
'80100000' 为部分成功(单 token 场景仍视为失败,错误里带原始响应便于排障)。""" '80100000' 为部分成功(单 token 场景仍视为失败,错误里带原始响应便于排障)。"""
app_id = _require(settings.HUAWEI_PUSH_APP_ID, "HUAWEI_PUSH_APP_ID") app_id = _require(settings.HUAWEI_PUSH_APP_ID, "HUAWEI_PUSH_APP_ID")
access_token = _huawei_access_token() 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 = { payload = {
"validate_only": False, "validate_only": False,
"message": { "message": {
# click_action type=3(打开应用首页)时,HMS 点击会把 data JSON 的键值对注入启动 intent
# 的 extras → MainActivity.consumeNavTarget 读 notif_id/notif_type 直达落地。
"data": json.dumps(_click_extras(extras), ensure_ascii=False), "data": json.dumps(_click_extras(extras), ensure_ascii=False),
"android": { "android": {
"target_user_type": settings.HUAWEI_PUSH_TARGET_USER_TYPE,
"ttl": f"{settings.PUSH_TIME_TO_LIVE_SEC}s", "ttl": f"{settings.PUSH_TIME_TO_LIVE_SEC}s",
"notification": { "notification": {
"title": title, "title": title,
"body": body, "body": body,
"click_action": {"type": 3}, "click_action": click_action,
"importance": "NORMAL", "importance": "NORMAL",
}, },
}, },
@@ -333,6 +605,8 @@ def _send_huawei(token: str, title: str, body: str, extras: dict[str, str]) -> d
data = _request_json( data = _request_json(
"POST", "POST",
settings.HUAWEI_PUSH_SEND_ENDPOINT_TEMPLATE.format(app_id=app_id), settings.HUAWEI_PUSH_SEND_ENDPOINT_TEMPLATE.format(app_id=app_id),
vendor="huawei",
operation="send_notification",
json=payload, json=payload,
headers={ headers={
"Content-Type": "application/json; charset=UTF-8", "Content-Type": "application/json; charset=UTF-8",
@@ -340,7 +614,32 @@ def _send_huawei(token: str, title: str, body: str, extras: dict[str, str]) -> d
}, },
) )
if str(data.get("code", "")) != "80000000": if str(data.get("code", "")) != "80000000":
raise VendorPushError(f"huawei push failed: {data}") raise VendorPushError(f"huawei push failed: {_raw_log_summary(data)}")
return data
def _send_huawei_data(token: str, payload: dict[str, str]) -> dict[str, Any]:
app_id = _require(settings.HUAWEI_PUSH_APP_ID, "HUAWEI_PUSH_APP_ID")
access_token = _huawei_access_token()
data = _request_json(
"POST",
settings.HUAWEI_PUSH_SEND_ENDPOINT_TEMPLATE.format(app_id=app_id),
vendor="huawei",
operation="send_data_event",
json={
"validate_only": False,
"message": {
"data": json.dumps(payload, ensure_ascii=False),
"token": [token],
},
},
headers={
"Content-Type": "application/json; charset=UTF-8",
"Authorization": f"Bearer {access_token}",
},
)
if str(data.get("code", "")) != "80000000":
raise VendorPushError(f"huawei data push failed: {_raw_log_summary(data)}")
return data return data
@@ -357,6 +656,8 @@ def _vivo_auth_token() -> str:
data = _request_json( data = _request_json(
"POST", "POST",
settings.VIVO_PUSH_AUTH_ENDPOINT, settings.VIVO_PUSH_AUTH_ENDPOINT,
vendor="vivo",
operation="authenticate",
json={ json={
"appId": app_id, "appId": app_id,
"appKey": app_key, "appKey": app_key,
@@ -366,10 +667,10 @@ def _vivo_auth_token() -> str:
headers={"Content-Type": "application/json"}, headers={"Content-Type": "application/json"},
) )
if int(data.get("result", -1)) != 0: if int(data.get("result", -1)) != 0:
raise VendorPushError(f"vivo auth failed: {data}") raise VendorPushError(f"vivo auth failed: {_raw_log_summary(data)}")
token = data.get("authToken") token = data.get("authToken")
if not token: if not token:
raise VendorPushError(f"vivo auth missing authToken: {data}") raise VendorPushError(f"vivo auth missing authToken: {_raw_log_summary(data)}")
return _cache_put(cache_key, str(token), 24 * 3600) return _cache_put(cache_key, str(token), 24 * 3600)
@@ -385,15 +686,22 @@ def _send_vivo(token: str, title: str, body: str, extras: dict[str, str]) -> dic
"timeToLive": settings.PUSH_TIME_TO_LIVE_SEC, "timeToLive": settings.PUSH_TIME_TO_LIVE_SEC,
"requestId": uuid.uuid4().hex, "requestId": uuid.uuid4().hex,
"pushMode": settings.VIVO_PUSH_MODE, "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, "clientCustomMap": extras,
} }
# 点击落地:消息中心推送(带 notificationId)→ skipType=4 + skipContent=intent uri,由 vivo # vivo Push SDK 480+ 已不再回调 skipType=3;必须使用 skipType=4 的完整 Intent URI。
# 系统直启 MainActivity 并携带 S. extras(与小米 notify_effect=2 同机制)。不依赖客户端 # URI 同时包含 data/scheme、显式 component 和 S. extrasOriginOS 才会把业务参数
# VivoPushReceiver.onNotificationMessageClicked 里的后台 startActivity——Android 10+ BAL # 原样交给 MainActivity(仅靠隐式 deeplink 会打开 App,但可能剥掉 extras)。
# 会静默拦掉,receiver 路径仅作兜底。无 notificationId 的召回类保持 skipType=1 仅打开首页。
if extras.get("notificationId"): if extras.get("notificationId"):
payload["skipType"] = 4 payload["skipType"] = 4
payload["skipContent"] = _click_intent_uri(extras) payload["skipContent"] = _vivo_click_intent_uri(extras)
else: else:
payload["skipType"] = 1 payload["skipType"] = 1
if settings.VIVO_PUSH_CATEGORY: if settings.VIVO_PUSH_CATEGORY:
@@ -401,14 +709,37 @@ def _send_vivo(token: str, title: str, body: str, extras: dict[str, str]) -> dic
data = _request_json( data = _request_json(
"POST", "POST",
settings.VIVO_PUSH_SEND_ENDPOINT, settings.VIVO_PUSH_SEND_ENDPOINT,
vendor="vivo",
operation="send_notification",
json=payload, json=payload,
headers={ headers={
"Content-Type": "application/json", "Content-Type": "application/json",
"authToken": auth_token, "authToken": auth_token,
}, },
) )
# 10089 = 当前 vivo 应用尚未开通 VPush「离线自动角标」能力。
# 角标只是通知中心未读数的镜像,不能因此让整条通知发送失败:去掉 addBadge
# 原样重试,通知到达/应用同步后仍由客户端系统 API 按统一未读数精确设置。
if int(data.get("result", -1)) == 10089:
logger.warning("vivo VPush badge not enabled (10089); retrying without addBadge")
fallback_payload = dict(payload)
fallback_payload.pop("addBadge", None)
fallback_payload["requestId"] = uuid.uuid4().hex
data = _request_json(
"POST",
settings.VIVO_PUSH_SEND_ENDPOINT,
vendor="vivo",
operation="send_notification",
json=fallback_payload,
headers={
"Content-Type": "application/json",
"authToken": auth_token,
},
)
if int(data.get("result", -1)) == 0:
data = {**data, "badgeFallback": True}
if int(data.get("result", -1)) != 0: if int(data.get("result", -1)) != 0:
raise VendorPushError(f"vivo push failed: {data}") raise VendorPushError(f"vivo push failed: {_raw_log_summary(data)}")
return data return data
@@ -446,14 +777,40 @@ def _send_xiaomi(token: str, title: str, body: str, extras: dict[str, str]) -> d
data = _request_form( data = _request_form(
"POST", "POST",
settings.XIAOMI_PUSH_SEND_ENDPOINT, settings.XIAOMI_PUSH_SEND_ENDPOINT,
vendor="xiaomi",
operation="send_notification",
data=form, data=form,
headers={"Authorization": f"key={app_secret}"}, headers={"Authorization": f"key={app_secret}"},
) )
code = data.get("code") code = data.get("code")
if code not in (0, "0", None): if code not in (0, "0", None):
raise VendorPushError(f"xiaomi push failed: {data}") raise VendorPushError(f"xiaomi push failed: {_raw_log_summary(data)}")
if str(data.get("result", "ok")).lower() not in ("ok", "success"): if str(data.get("result", "ok")).lower() not in ("ok", "success"):
raise VendorPushError(f"xiaomi push failed: {data}") raise VendorPushError(f"xiaomi push failed: {_raw_log_summary(data)}")
return data
def _send_xiaomi_data(token: str, payload: dict[str, str]) -> dict[str, Any]:
app_secret = _require(settings.XIAOMI_PUSH_APP_SECRET, "XIAOMI_PUSH_APP_SECRET")
data = _request_form(
"POST",
settings.XIAOMI_PUSH_SEND_ENDPOINT,
vendor="xiaomi",
operation="send_data_event",
data={
"registration_id": token,
"restricted_package_name": settings.ANDROID_PACKAGE_NAME,
"payload": json.dumps(payload, ensure_ascii=False),
"pass_through": "1",
"time_to_live": str(settings.PUSH_TIME_TO_LIVE_SEC * 1000),
},
headers={"Authorization": f"key={app_secret}"},
)
code = data.get("code")
if code not in (0, "0", None):
raise VendorPushError(f"xiaomi data push failed: {_raw_log_summary(data)}")
if str(data.get("result", "ok")).lower() not in ("ok", "success"):
raise VendorPushError(f"xiaomi data push failed: {_raw_log_summary(data)}")
return data return data
@@ -490,6 +847,24 @@ def _click_intent_uri(extras: dict[str, str]) -> str:
return ";".join(parts) 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: def _xiaomi_template_param(title: str, alert: str) -> str:
rendered = ( rendered = (
settings.XIAOMI_PUSH_TEMPLATE_PARAM_JSON settings.XIAOMI_PUSH_TEMPLATE_PARAM_JSON
@@ -522,6 +897,8 @@ def _oppo_auth_token() -> str:
data = _request_form( data = _request_form(
"POST", "POST",
settings.OPPO_PUSH_AUTH_ENDPOINT, settings.OPPO_PUSH_AUTH_ENDPOINT,
vendor="oppo",
operation="authenticate",
data={ data={
"app_key": app_key, "app_key": app_key,
"timestamp": timestamp, "timestamp": timestamp,
@@ -530,10 +907,10 @@ def _oppo_auth_token() -> str:
headers={"Content-Type": "application/x-www-form-urlencoded"}, headers={"Content-Type": "application/x-www-form-urlencoded"},
) )
if int(data.get("code", -1)) != 0: if int(data.get("code", -1)) != 0:
raise VendorPushError(f"oppo auth failed: {data}") raise VendorPushError(f"oppo auth failed: {_raw_log_summary(data)}")
token = (data.get("data") or {}).get("auth_token") or data.get("auth_token") token = (data.get("data") or {}).get("auth_token") or data.get("auth_token")
if not token: if not token:
raise VendorPushError(f"oppo auth missing auth_token: {data}") raise VendorPushError(f"oppo auth missing auth_token: {_raw_log_summary(data)}")
return _cache_put(cache_key, str(token), 24 * 3600) return _cache_put(cache_key, str(token), 24 * 3600)
@@ -548,6 +925,11 @@ def _send_oppo(token: str, title: str, body: str, extras: dict[str, str]) -> dic
"off_line_ttl": ttl_hours, "off_line_ttl": ttl_hours,
"action_parameters": json.dumps(_click_extras(extras), ensure_ascii=False), "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 的 # 点击落地:OPPO SDK 没有点击回调,参数只能靠服务端点击动作配置送达——action_parameters 的
# 键值对仅在 click_action_type=1/4 时才会注入目标 Activity 的 intent extras(type=0「启动应用」 # 键值对仅在 click_action_type=1/4 时才会注入目标 Activity 的 intent extras(type=0「启动应用」
# 会忽略它,extras 全丢 → 点了没反应,与小米 notify_effect=1 同款坑)。 # 会忽略它,extras 全丢 → 点了没反应,与小米 notify_effect=1 同款坑)。
@@ -573,6 +955,8 @@ def _send_oppo(token: str, title: str, body: str, extras: dict[str, str]) -> dic
data = _request_form( data = _request_form(
"POST", "POST",
settings.OPPO_PUSH_SEND_ENDPOINT, settings.OPPO_PUSH_SEND_ENDPOINT,
vendor="oppo",
operation="send_notification",
data={ data={
"auth_token": auth_token, "auth_token": auth_token,
"message": json.dumps(message, ensure_ascii=False), "message": json.dumps(message, ensure_ascii=False),
@@ -580,5 +964,5 @@ def _send_oppo(token: str, title: str, body: str, extras: dict[str, str]) -> dic
headers={"Content-Type": "application/x-www-form-urlencoded"}, headers={"Content-Type": "application/x-www-form-urlencoded"},
) )
if int(data.get("code", -1)) != 0: if int(data.get("code", -1)) != 0:
raise VendorPushError(f"oppo push failed: {data}") raise VendorPushError(f"oppo push failed: {_raw_log_summary(data)}")
return data return data
+1 -1
View File
@@ -38,7 +38,7 @@ class GuideVideoPlay(Base):
# start_play 捕获 IntegrityError 降级成"这次不放视频"。 # start_play 捕获 IntegrityError 降级成"这次不放视频"。
# 用 unique Index 而非 UniqueConstraint:与迁移里的 create_index 对齐(SQLite 加约束 # 用 unique Index 而非 UniqueConstraint:与迁移里的 create_index 对齐(SQLite 加约束
# 要整表重建),autogenerate 才不会每次报一条假 diff。 # 要整表重建),autogenerate 才不会每次报一条假 diff。
Index("uq_guide_video_play_user_seq", "user_id", "seq", unique=True), Index("uq_guide_video_play_user_scene_seq", "user_id", "scene", "seq", unique=True),
) )
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+45 -22
View File
@@ -30,7 +30,11 @@ from app.models.app_config import AppConfig
from app.models.guide_video import GuideVideoPlay from app.models.guide_video import GuideVideoPlay
from app.repositories import wallet as crud_wallet from app.repositories import wallet as crud_wallet
_KEY = "coupon_guide_video" SCENES = ("coupon", "comparison")
_KEY_BY_SCENE = {
"coupon": "coupon_guide_video",
"comparison": "comparison_guide_video",
}
#: 金币流水 biz_type。客户端收益明细按它显示「新手引导视频奖励」。 #: 金币流水 biz_type。客户端收益明细按它显示「新手引导视频奖励」。
BIZ_TYPE = "guide_video" BIZ_TYPE = "guide_video"
@@ -41,7 +45,7 @@ _DEFAULTS: dict[str, Any] = {
"enabled": True, "enabled": True,
"video_url": None, # None/空 = 未配片 → 不下发,浮层照旧放广告 "video_url": None, # None/空 = 未配片 → 不下发,浮层照旧放广告
"max_plays": 3, # 每个账号前 N 次浮层放引导视频 "max_plays": 3, # 每个账号前 N 次浮层放引导视频
"reward_coin": 120, # 每次固定金币 "reward_coin": 100, # 每次固定金币
} }
_FIELDS = tuple(_DEFAULTS.keys()) _FIELDS = tuple(_DEFAULTS.keys())
@@ -65,19 +69,28 @@ def _merge(raw: Any) -> dict[str, Any]:
return out return out
def get_config(db: Session) -> dict[str, Any]: def _config_key(scene: str) -> str:
if scene not in _KEY_BY_SCENE:
raise ValueError(f"unsupported guide video scene: {scene}")
return _KEY_BY_SCENE[scene]
def get_config(db: Session, scene: str = "coupon") -> dict[str, Any]:
"""完整配置 + updated_at(admin 读 / 业务读共用)。""" """完整配置 + updated_at(admin 读 / 业务读共用)。"""
row = db.get(AppConfig, _KEY) row = db.get(AppConfig, _config_key(scene))
cfg = _merge(row.value if row is not None else None) cfg = _merge(row.value if row is not None else None)
cfg["updated_at"] = row.updated_at.isoformat() if row is not None and row.updated_at else None cfg["updated_at"] = row.updated_at.isoformat() if row is not None and row.updated_at else None
return cfg return cfg
def _write(db: Session, value: dict[str, Any], *, admin_id: int, commit: bool) -> dict[str, Any]: def _write(
db: Session, value: dict[str, Any], *, scene: str, admin_id: int, commit: bool
) -> dict[str, Any]:
"""整体覆写该行(value 须为完整字段 dict),返回合并后的完整配置(含 updated_at)。""" """整体覆写该行(value 须为完整字段 dict),返回合并后的完整配置(含 updated_at)。"""
row = db.get(AppConfig, _KEY) key = _config_key(scene)
row = db.get(AppConfig, key)
if row is None: if row is None:
row = AppConfig(key=_KEY, value=value, updated_by_admin_id=admin_id) row = AppConfig(key=key, value=value, updated_by_admin_id=admin_id)
db.add(row) db.add(row)
else: else:
row.value = value # 整体重新赋值,SQLAlchemy 才侦测得到变更 row.value = value # 整体重新赋值,SQLAlchemy 才侦测得到变更
@@ -98,11 +111,12 @@ def update_config(
enabled: bool | None = None, enabled: bool | None = None,
max_plays: int | None = None, max_plays: int | None = None,
reward_coin: int | None = None, reward_coin: int | None = None,
scene: str = "coupon",
admin_id: int, admin_id: int,
commit: bool = True, commit: bool = True,
) -> tuple[dict[str, Any], dict[str, Any]]: ) -> tuple[dict[str, Any], dict[str, Any]]:
"""改开关 / 次数 / 金币(只改传了的字段;视频走 set_video)。返回 (before, after) 供审计。""" """改开关 / 次数 / 金币(只改传了的字段;视频走 set_video)。返回 (before, after) 供审计。"""
row = db.get(AppConfig, _KEY) row = db.get(AppConfig, _config_key(scene))
before = _merge(row.value if row is not None else None) before = _merge(row.value if row is not None else None)
new_value = {k: before[k] for k in _FIELDS} new_value = {k: before[k] for k in _FIELDS}
if enabled is not None: if enabled is not None:
@@ -111,45 +125,52 @@ def update_config(
new_value["max_plays"] = max(0, min(int(max_plays), MAX_PLAYS_LIMIT)) new_value["max_plays"] = max(0, min(int(max_plays), MAX_PLAYS_LIMIT))
if reward_coin is not None: if reward_coin is not None:
new_value["reward_coin"] = max(0, min(int(reward_coin), REWARD_COIN_LIMIT)) new_value["reward_coin"] = max(0, min(int(reward_coin), REWARD_COIN_LIMIT))
after = _write(db, new_value, admin_id=admin_id, commit=commit) after = _write(db, new_value, scene=scene, admin_id=admin_id, commit=commit)
return before, after return before, after
def set_video( def set_video(
db: Session, video_url: str | None, *, admin_id: int, commit: bool = True db: Session, video_url: str | None, *, scene: str = "coupon",
admin_id: int, commit: bool = True
) -> tuple[dict[str, Any], dict[str, Any]]: ) -> tuple[dict[str, Any], dict[str, Any]]:
"""设置/清空引导视频地址。返回 (before, after);before['video_url'] 供调用方删旧文件。""" """设置/清空引导视频地址。返回 (before, after);before['video_url'] 供调用方删旧文件。"""
row = db.get(AppConfig, _KEY) row = db.get(AppConfig, _config_key(scene))
before = _merge(row.value if row is not None else None) before = _merge(row.value if row is not None else None)
new_value = {k: before[k] for k in _FIELDS} new_value = {k: before[k] for k in _FIELDS}
new_value["video_url"] = video_url new_value["video_url"] = video_url
after = _write(db, new_value, admin_id=admin_id, commit=commit) after = _write(db, new_value, scene=scene, admin_id=admin_id, commit=commit)
return before, after return before, after
# ===== 播放计次 ===== # ===== 播放计次 =====
def used_plays(db: Session, user_id: int) -> int: def used_plays(db: Session, user_id: int, scene: str = "coupon") -> int:
"""该账号已用掉的引导视频次数(开播即算,含未发币的)。""" """该账号已用掉的引导视频次数(开播即算,含未发币的)。"""
return int( return int(
db.execute( db.execute(
select(func.count()).select_from(GuideVideoPlay).where( select(func.count()).select_from(GuideVideoPlay).where(
GuideVideoPlay.user_id == user_id GuideVideoPlay.user_id == user_id,
GuideVideoPlay.scene == scene,
) )
).scalar_one() ).scalar_one()
) )
def play_stats(db: Session) -> dict[str, int]: def play_stats(db: Session, scene: str = "coupon") -> dict[str, int]:
"""全站播放统计(admin 页展示):总播放次数 / 其中已发币次数。""" """全站播放统计(admin 页展示):总播放次数 / 其中已发币次数。"""
total = int( total = int(
db.execute(select(func.count()).select_from(GuideVideoPlay)).scalar_one() db.execute(
select(func.count()).select_from(GuideVideoPlay).where(
GuideVideoPlay.scene == scene
)
).scalar_one()
) )
granted = int( granted = int(
db.execute( db.execute(
select(func.count()).select_from(GuideVideoPlay).where( select(func.count()).select_from(GuideVideoPlay).where(
GuideVideoPlay.status == "granted" GuideVideoPlay.status == "granted",
GuideVideoPlay.scene == scene,
) )
).scalar_one() ).scalar_one()
) )
@@ -168,11 +189,12 @@ def start_play(
reward_coin 播完/中途关闭都发的固定金币 reward_coin 播完/中途关闭都发的固定金币
seq / remaining 第几次 / 发完这次还剩几次(仅展示与排查用) seq / remaining 第几次 / 发完这次还剩几次(仅展示与排查用)
""" """
cfg = get_config(db) _config_key(scene)
cfg = get_config(db, scene)
video_url = (cfg.get("video_url") or "").strip() video_url = (cfg.get("video_url") or "").strip()
max_plays = int(cfg.get("max_plays") or 0) max_plays = int(cfg.get("max_plays") or 0)
reward_coin = int(cfg.get("reward_coin") or 0) reward_coin = int(cfg.get("reward_coin") or 0)
used = used_plays(db, user_id) used = used_plays(db, user_id, scene)
def _miss(used_now: int) -> dict[str, Any]: def _miss(used_now: int) -> dict[str, Any]:
return { return {
@@ -194,7 +216,7 @@ def start_play(
scene=scene, scene=scene,
seq=seq, seq=seq,
video_url=video_url, video_url=video_url,
coin=0, coin=reward_coin,
status="playing", status="playing",
completed=0, completed=0,
started_at=datetime.now(rewards.CN_TZ).replace(tzinfo=None), started_at=datetime.now(rewards.CN_TZ).replace(tzinfo=None),
@@ -210,7 +232,7 @@ def start_play(
db.flush() db.flush()
except IntegrityError: except IntegrityError:
db.rollback() db.rollback()
return _miss(used_plays(db, user_id)) return _miss(used_plays(db, user_id, scene))
return { return {
"should_play": True, "should_play": True,
"video_url": video_url, "video_url": video_url,
@@ -241,7 +263,8 @@ def grant_play(
""" """
token = (play_token or "").strip() token = (play_token or "").strip()
# 金币额度以**服务端配置**为准,不信客户端(客户端只上报"播完/关闭")。 # 金币额度以**服务端配置**为准,不信客户端(客户端只上报"播完/关闭")。
coin = int(get_config(db).get("reward_coin") or 0) play = _find_play(db, user_id, token)
coin = int(play.coin if play is not None else 0)
# 幂等核心:把 status 放进 WHERE 做条件更新(compare-and-set),而不是"先读再判再写"。 # 幂等核心:把 status 放进 WHERE 做条件更新(compare-and-set),而不是"先读再判再写"。
# 「播完」与「✕ 关闭」抢跑、或客户端超时重试时,两个请求会都读到 status='playing', # 「播完」与「✕ 关闭」抢跑、或客户端超时重试时,两个请求会都读到 status='playing',
+3 -1
View File
@@ -1,13 +1,15 @@
"""新手引导视频(领券等候浮层前 N 次替代广告)的客户端请求/响应契约。""" """新手引导视频(领券等候浮层前 N 次替代广告)的客户端请求/响应契约。"""
from __future__ import annotations from __future__ import annotations
from typing import Literal
from pydantic import BaseModel, Field from pydantic import BaseModel, Field
class GuideVideoStartIn(BaseModel): class GuideVideoStartIn(BaseModel):
"""开播询问。scene 目前只有 coupon(领券浮层);预留给日后比价等场景。""" """开播询问。scene 目前只有 coupon(领券浮层);预留给日后比价等场景。"""
scene: str = Field(default="coupon", max_length=16) scene: Literal["coupon", "comparison"] = "coupon"
class GuideVideoStartOut(BaseModel): class GuideVideoStartOut(BaseModel):
+22 -1
View File
@@ -136,9 +136,17 @@ def _push_to_user_devices(db: Session, row: Notification, push_vars: dict[str, s
) )
continue continue
try: try:
vendor_push.send_notification( response = vendor_push.send_notification(
vendor, dev.push_token, title=title, body=body, extras=extras vendor, dev.push_token, title=title, body=body, extras=extras
) )
if vendor == "huawei" and row.type in {"feedback_reply", "feedback_reward"}:
# 华为反馈推送联调日志:保留厂商返回码/requestId 等排障信息,
# 请求本身的 token、Authorization 和应用密钥不会进入 response。
logger.info(
"huawei feedback push response user_id=%s type=%s "
"notification_id=%s response=%s",
row.user_id, row.type, row.id, response,
)
logger.info( logger.info(
"push sent user_id=%s type=%s vendor=%s notification_id=%s", "push sent user_id=%s type=%s vendor=%s notification_id=%s",
row.user_id, row.type, vendor, row.id, row.user_id, row.type, vendor, row.id,
@@ -147,6 +155,19 @@ def _push_to_user_devices(db: Session, row: Notification, push_vars: dict[str, s
logger.warning( logger.warning(
"push failed user_id=%s type=%s vendor=%s: %s", row.user_id, row.type, vendor, e "push failed user_id=%s type=%s vendor=%s: %s", row.user_id, row.type, vendor, e
) )
try:
vendor_push.send_data_event(
vendor,
dev.push_token,
event=vendor_push.DATA_EVENT_NOTIFICATION_CREATED,
notification_id=str(row.id),
)
except vendor_push.VendorPushError as e:
# 透传只负责前台铃铛实时刷新,失败不能影响通知栏消息或站内消息。
logger.warning(
"push data event failed user_id=%s type=%s vendor=%s: %s",
row.user_id, row.type, vendor, e,
)
except Exception: # noqa: BLE001 — 渲染/查设备等意外失败同样不外抛 except Exception: # noqa: BLE001 — 渲染/查设备等意外失败同样不外抛
logger.exception("push notification failed user_id=%s type=%s", row.user_id, row.type) logger.exception("push notification failed user_id=%s type=%s", row.user_id, row.type)
+1 -1
View File
@@ -201,7 +201,7 @@ def test_start_loses_seq_race_degrades_to_ad(client, guide_configured, monkeypat
assert first["should_play"] is True and first["seq"] == 1 assert first["should_play"] is True and first["seq"] == 1
# 本次请求读到的是过期计数 → 仍会算出 seq=1 # 本次请求读到的是过期计数 → 仍会算出 seq=1
monkeypatch.setattr(crud_guide, "used_plays", lambda db, user_id: 0) monkeypatch.setattr(crud_guide, "used_plays", lambda db, user_id, scene="coupon": 0)
with SessionLocal() as db: with SessionLocal() as db:
result = crud_guide.start_play(db, uid) result = crud_guide.start_play(db, uid)
+95
View File
@@ -7,7 +7,9 @@ mock/真发/变量覆盖/站内联动/设备反查/参数校验。厂商 HTTP
from __future__ import annotations from __future__ import annotations
import json import json
import logging
import httpx
import pytest import pytest
from fastapi.testclient import TestClient from fastapi.testclient import TestClient
@@ -110,6 +112,99 @@ def test_huawei_non_success_code_raises(monkeypatch) -> None:
vendor_push.send_notification("huawei", "bad-token", title="t", body="b") vendor_push.send_notification("huawei", "bad-token", title="t", body="b")
def test_vendor_http_logs_device_token_raw_response_and_elapsed(monkeypatch, caplog) -> None:
vendor_push._token_cache.clear()
def _fake_request(method, url, **kwargs): # noqa: ANN001
if url == vendor_push.settings.HUAWEI_PUSH_TOKEN_ENDPOINT:
return _Resp({"access_token": "raw-access-token", "expires_in": 3600})
return _Resp({"code": "80000000", "msg": "Success", "requestId": "vendor-request-1"})
monkeypatch.setattr(vendor_push.settings, "HUAWEI_PUSH_APP_ID", "10086001")
monkeypatch.setattr(vendor_push.settings, "HUAWEI_PUSH_APP_SECRET", "raw-app-secret")
monkeypatch.setattr(vendor_push.httpx, "request", _fake_request)
with caplog.at_level(logging.INFO, logger=vendor_push.logger.name):
vendor_push.send_notification(
"huawei",
"raw-device-token",
title="需要进入日志的标题",
body="需要进入日志的正文",
extras={"type": "withdraw_success", "notificationId": "90001"},
)
messages = "\n".join(record.getMessage() for record in caplog.records)
assert "operation=authenticate" in messages
assert "operation=send_notification" in messages
assert "request=" in messages
assert "response=" in messages
assert "elapsed_ms=" in messages
assert "vendor-request-1" in messages
assert "withdraw_success" in messages
assert "raw-access-token" in messages
assert "raw-device-token" in messages
assert "需要进入日志的标题" in messages
assert "需要进入日志的正文" in messages
assert "<redacted" in messages
assert "raw-app-secret" not in messages
send_http_log = next(
record.getMessage()
for record in caplog.records
if "operation=send_notification" in record.getMessage()
and "vendor push http completed" in record.getMessage()
)
assert "raw-device-token" in send_http_log
assert "raw-access-token" not in send_http_log # 请求头中的厂商鉴权 token 仍脱敏
def test_vendor_http_network_error_logs_request_context(monkeypatch, caplog) -> None:
def _fake_request(*args, **kwargs): # noqa: ANN002, ANN003
raise httpx.ConnectError("connection refused")
monkeypatch.setattr(vendor_push.settings, "XIAOMI_PUSH_APP_SECRET", "raw-xiaomi-secret")
monkeypatch.setattr(vendor_push.httpx, "request", _fake_request)
with (
caplog.at_level(logging.ERROR, logger=vendor_push.logger.name),
pytest.raises(vendor_push.VendorPushError, match="push http error"),
):
vendor_push.send_notification(
"xiaomi",
"raw-xiaomi-token",
title="title",
body="body",
extras={"type": "push_test"},
)
messages = "\n".join(record.getMessage() for record in caplog.records)
assert "outcome=network_error" in messages
assert "response=<no_response>" in messages
assert "operation=send_notification" in messages
assert "elapsed_ms=" in messages
assert "raw-xiaomi-secret" not in messages
assert "raw-xiaomi-token" in messages
@pytest.mark.parametrize(
("request_kwargs", "device_token"),
[
({"json": {"token": ["honor-device-token"]}}, "honor-device-token"),
({"json": {"message": {"token": ["huawei-device-token"]}}}, "huawei-device-token"),
({"json": {"regId": "vivo-device-token"}}, "vivo-device-token"),
({"data": {"registration_id": "xiaomi-device-token"}}, "xiaomi-device-token"),
(
{"data": {"message": '{"target_value":"oppo-device-token"}'}},
"oppo-device-token",
),
],
)
def test_all_vendor_device_token_fields_remain_visible_in_request_summary(
request_kwargs, device_token
) -> None:
summary = vendor_push._request_summary(request_kwargs)
assert device_token in summary
def test_vendor_aliases_normalize() -> None: def test_vendor_aliases_normalize() -> None:
assert vendor_push.normalize_vendor("华为") == "huawei" assert vendor_push.normalize_vendor("华为") == "huawei"
assert vendor_push.normalize_vendor("HMS") == "huawei" assert vendor_push.normalize_vendor("HMS") == "huawei"