Compare commits
5 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 0ed1b08a85 | |||
| 566a201354 | |||
| 7871f1f6b6 | |||
| 97613eca8e | |||
| f902f3011c |
@@ -0,0 +1,28 @@
|
||||
"""add planned coupon count to coupon session
|
||||
|
||||
Revision ID: coupon_session_planned_count
|
||||
Revises: 6d2309208549
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Sequence
|
||||
|
||||
import sqlalchemy as sa
|
||||
|
||||
from alembic import op
|
||||
|
||||
revision: str = "coupon_session_planned_count"
|
||||
down_revision: str | None = "6d2309208549"
|
||||
branch_labels: str | Sequence[str] | None = None
|
||||
depends_on: str | Sequence[str] | None = None
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
op.add_column(
|
||||
"coupon_session",
|
||||
sa.Column("planned_coupon_count", sa.Integer(), nullable=True),
|
||||
)
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
op.drop_column("coupon_session", "planned_coupon_count")
|
||||
@@ -1,31 +0,0 @@
|
||||
"""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,
|
||||
)
|
||||
@@ -23,6 +23,7 @@ from app.repositories.coupon_state import DEFAULT_PLATFORMS, coupon_id_to_platfo
|
||||
|
||||
_SLOT_OK = ("success", "already_claimed")
|
||||
_SLOT_TRIED = ("success", "already_claimed", "failed")
|
||||
_SESSION_POINT_TOTAL = (*_SLOT_TRIED, "skipped")
|
||||
|
||||
|
||||
def _cn_hour(dt: datetime) -> int:
|
||||
@@ -166,6 +167,10 @@ def _session_to_row(
|
||||
point_stats: dict | None = None,
|
||||
) -> dict:
|
||||
"""CouponSession ORM → 明细行 dict(主表「领券数据」与「用户全部领券」抽屉共用)。"""
|
||||
succeeded = point_stats["succeeded"] if point_stats else 0
|
||||
event_total = point_stats["tried"] if point_stats else 0
|
||||
planned_total = r.planned_coupon_count or 0
|
||||
point_total = max(event_total, planned_total)
|
||||
return {
|
||||
"id": r.id,
|
||||
"trace_id": r.trace_id,
|
||||
@@ -182,8 +187,8 @@ def _session_to_row(
|
||||
"app_env": r.app_env,
|
||||
"started_at": r.started_at,
|
||||
"claimed_count": r.claimed_count,
|
||||
"point_success_count": point_stats["succeeded"] if point_stats else None,
|
||||
"point_total_count": point_stats["tried"] if point_stats else None,
|
||||
"point_success_count": succeeded if point_total > 0 else None,
|
||||
"point_total_count": point_total if point_total > 0 else None,
|
||||
"trace_url": r.trace_url,
|
||||
"ad_revenue_yuan": ad_revenue_yuan,
|
||||
}
|
||||
@@ -202,7 +207,7 @@ def _point_scores_by_trace(db: Session, trace_ids: list[str]) -> dict[str, dict[
|
||||
)
|
||||
.where(
|
||||
CouponClaimEvent.trace_id.in_(trace_ids),
|
||||
CouponClaimEvent.status.in_(_SLOT_TRIED),
|
||||
CouponClaimEvent.status.in_(_SESSION_POINT_TOTAL),
|
||||
)
|
||||
.group_by(CouponClaimEvent.trace_id)
|
||||
).all()
|
||||
@@ -401,11 +406,13 @@ def coupon_user_records(db: Session, *, user_id: int, limit: int = 100) -> dict:
|
||||
select(func.count()).select_from(CouponSession).where(CouponSession.user_id == user_id)
|
||||
).scalar_one()
|
||||
rev_map = crud_ecpm.revenue_yuan_by_trace(db, [r.trace_id for r in rows])
|
||||
point_stats_map = _point_scores_by_trace(db, [r.trace_id for r in rows])
|
||||
return {
|
||||
"items": [
|
||||
_session_to_row(
|
||||
r,
|
||||
ad_revenue_yuan=rev_map.get(r.trace_id, 0.0),
|
||||
point_stats=point_stats_map.get(r.trace_id),
|
||||
)
|
||||
for r in rows
|
||||
],
|
||||
|
||||
@@ -9,7 +9,7 @@ client_max_body_size,见 shaguabijia-admin-web/deploy/nginx/admin.shaguabijia.co
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Annotated, Literal
|
||||
from typing import Annotated
|
||||
|
||||
from fastapi import APIRouter, Depends, File, HTTPException, Request, UploadFile
|
||||
|
||||
@@ -27,21 +27,14 @@ router = APIRouter(
|
||||
)
|
||||
|
||||
|
||||
GuideScene = Literal["coupon", "comparison"]
|
||||
|
||||
|
||||
def _out(db: AdminDb, scene: GuideScene) -> GuideVideoConfigOut:
|
||||
def _out(db: AdminDb) -> GuideVideoConfigOut:
|
||||
"""配置 + 播放统计合成响应(四个写接口都以最新状态返回,前端一次同步到位)。"""
|
||||
return GuideVideoConfigOut(
|
||||
scene=scene,
|
||||
**guide_video.get_config(db, scene),
|
||||
**guide_video.play_stats(db, scene),
|
||||
)
|
||||
return GuideVideoConfigOut(**guide_video.get_config(db), **guide_video.play_stats(db))
|
||||
|
||||
|
||||
@router.get("", response_model=GuideVideoConfigOut, summary="新手引导视频配置(领券浮层)")
|
||||
def get_config(db: AdminDb, scene: GuideScene = "coupon") -> GuideVideoConfigOut:
|
||||
return _out(db, scene)
|
||||
def get_config(db: AdminDb) -> GuideVideoConfigOut:
|
||||
return _out(db)
|
||||
|
||||
|
||||
@router.patch("", response_model=GuideVideoConfigOut, summary="改开关/次数/金币(带审计)")
|
||||
@@ -50,23 +43,21 @@ def update_config(
|
||||
request: Request,
|
||||
admin: Annotated[AdminUser, Depends(require_role("operator"))],
|
||||
db: AdminDb,
|
||||
scene: GuideScene = "coupon",
|
||||
) -> GuideVideoConfigOut:
|
||||
before, after = guide_video.update_config(
|
||||
db,
|
||||
enabled=body.enabled,
|
||||
max_plays=body.max_plays,
|
||||
reward_coin=body.reward_coin,
|
||||
scene=scene,
|
||||
admin_id=admin.id,
|
||||
commit=False,
|
||||
)
|
||||
write_audit(
|
||||
db, admin, action="guide_video.update", target_type="guide_video", target_id=None,
|
||||
detail={"scene": scene, "before": before, "after": after}, ip=get_client_ip(request), commit=False,
|
||||
detail={"before": before, "after": after}, ip=get_client_ip(request), commit=False,
|
||||
)
|
||||
db.commit()
|
||||
return _out(db, scene)
|
||||
return _out(db)
|
||||
|
||||
|
||||
@router.post("/video", response_model=GuideVideoConfigOut, summary="上传新手引导视频(MP4,带审计)")
|
||||
@@ -74,26 +65,23 @@ async def upload_video(
|
||||
request: Request,
|
||||
admin: Annotated[AdminUser, Depends(require_role("operator"))],
|
||||
db: AdminDb,
|
||||
file: Annotated[UploadFile, File()],
|
||||
scene: GuideScene = "coupon",
|
||||
file: UploadFile = File(...),
|
||||
) -> GuideVideoConfigOut:
|
||||
data = await file.read()
|
||||
try:
|
||||
url = media.save_guide_video(data)
|
||||
except media.MediaError as e:
|
||||
raise HTTPException(status_code=400, detail=str(e)) from e
|
||||
before, after = guide_video.set_video(
|
||||
db, url, scene=scene, admin_id=admin.id, commit=False
|
||||
)
|
||||
before, after = guide_video.set_video(db, url, admin_id=admin.id, commit=False)
|
||||
write_audit(
|
||||
db, admin, action="guide_video.set_video", target_type="guide_video", target_id=None,
|
||||
detail={"scene": scene, "before": before.get("video_url"), "after": url, "bytes": len(data)},
|
||||
detail={"before": before.get("video_url"), "after": url, "bytes": len(data)},
|
||||
ip=get_client_ip(request), commit=False,
|
||||
)
|
||||
db.commit()
|
||||
# 提交成功后再删旧片,避免新片没落库就把旧片丢了
|
||||
media.delete_guide_video(before.get("video_url"))
|
||||
return _out(db, scene)
|
||||
return _out(db)
|
||||
|
||||
|
||||
@router.delete("/video", response_model=GuideVideoConfigOut, summary="移除新手引导视频(带审计)")
|
||||
@@ -101,16 +89,13 @@ def delete_video(
|
||||
request: Request,
|
||||
admin: Annotated[AdminUser, Depends(require_role("operator"))],
|
||||
db: AdminDb,
|
||||
scene: GuideScene = "coupon",
|
||||
) -> GuideVideoConfigOut:
|
||||
"""移除后 /guide-video/start 一律返回 should_play=false,领券浮层回到「只放广告」。"""
|
||||
before, after = guide_video.set_video(
|
||||
db, None, scene=scene, admin_id=admin.id, commit=False
|
||||
)
|
||||
before, after = guide_video.set_video(db, None, admin_id=admin.id, commit=False)
|
||||
write_audit(
|
||||
db, admin, action="guide_video.delete_video", target_type="guide_video", target_id=None,
|
||||
detail={"scene": scene, "before": before.get("video_url")}, ip=get_client_ip(request), commit=False,
|
||||
detail={"before": before.get("video_url")}, ip=get_client_ip(request), commit=False,
|
||||
)
|
||||
db.commit()
|
||||
media.delete_guide_video(before.get("video_url"))
|
||||
return _out(db, scene)
|
||||
return _out(db)
|
||||
|
||||
@@ -79,10 +79,15 @@ class CouponDataRow(BaseModel):
|
||||
started_at: datetime = Field(..., description="发起时刻(明细「时间」列)")
|
||||
claimed_count: int | None = None
|
||||
point_success_count: int | None = Field(
|
||||
None, description="本次成功单券数(success+already_claimed);无逐券事件为空"
|
||||
None,
|
||||
description="本次成功单券数(success+already_claimed);有计划数但尚无成功事件时为 0",
|
||||
)
|
||||
point_total_count: int | None = Field(
|
||||
None, description="本次尝试单券数(success+already_claimed+failed,不含 skipped);无逐券事件为空"
|
||||
None,
|
||||
description=(
|
||||
"本次计划单券数(max(逐券事件数,计划数),事件含 skipped);"
|
||||
"无计划数且无逐券事件时为空"
|
||||
),
|
||||
)
|
||||
trace_url: str | None = Field(None, description="pricebot 公网 trace 链接(仅 completed 有);admin 渲染可点链接,无则显示可复制 trace_id")
|
||||
ad_revenue_yuan: float = Field(
|
||||
|
||||
@@ -7,7 +7,6 @@ from app.repositories.guide_video import MAX_PLAYS_LIMIT, REWARD_COIN_LIMIT
|
||||
|
||||
|
||||
class GuideVideoConfigOut(BaseModel):
|
||||
scene: str
|
||||
enabled: bool
|
||||
video_url: str | None = None # 相对地址 /media/guide_video/xxx.mp4;未配片 = None
|
||||
max_plays: int
|
||||
|
||||
@@ -93,6 +93,13 @@ def _record_claims_blocking(
|
||||
)
|
||||
|
||||
|
||||
def _merge_planned_count_blocking(
|
||||
trace_id: str | None, planned_count: int | None
|
||||
) -> None:
|
||||
with SessionLocal() as db:
|
||||
coupon_repo.merge_session_planned_count(db, trace_id, planned_count)
|
||||
|
||||
|
||||
def _mark_completed_blocking(
|
||||
device_id: str, user_id: int | None, trace_id: str | None
|
||||
) -> None:
|
||||
@@ -176,6 +183,18 @@ async def coupon_step(
|
||||
|
||||
resp_json = resp.json()
|
||||
|
||||
# pricebot 每帧 status.progress.total 都带本轮计划券数。独立回写 session 后,
|
||||
# 即使用户在第一张出结果前退出,admin 也能显示 0/N,而不是空值。
|
||||
progress = (resp_json.get("status") or {}).get("progress") or {}
|
||||
planned_count = progress.get("total")
|
||||
if trace_id and isinstance(planned_count, int) and planned_count > 0:
|
||||
try:
|
||||
await run_in_threadpool(
|
||||
_merge_planned_count_blocking, trace_id, planned_count
|
||||
)
|
||||
except Exception as e: # noqa: BLE001
|
||||
logger.warning("coupon planned count write failed: %s", e)
|
||||
|
||||
# 领券结果沉淀:每日资产 + 逐次事件;中间帧和 done 全量帧均幂等写库。
|
||||
# 当前只记录、不参与"要不要领"判断(MVP 先不去重)。写库失败不影响返回。
|
||||
if device_id:
|
||||
|
||||
+35
-242
@@ -22,7 +22,7 @@ import uuid
|
||||
from collections.abc import Callable
|
||||
from dataclasses import dataclass
|
||||
from typing import Any
|
||||
from urllib.parse import quote, urlsplit
|
||||
from urllib.parse import quote
|
||||
|
||||
import httpx
|
||||
|
||||
@@ -69,18 +69,6 @@ class _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:
|
||||
if not push_vendor:
|
||||
@@ -127,16 +115,8 @@ def send_notification(
|
||||
|
||||
if mock:
|
||||
logger.info(
|
||||
"vendor push mock vendor=%s request=%s",
|
||||
vendor,
|
||||
_log_summary(
|
||||
{
|
||||
"push_token": token,
|
||||
"title": title,
|
||||
"body": body,
|
||||
"extras": extras,
|
||||
}
|
||||
),
|
||||
"[mock push] vendor=%s token=%s... title=%s body=%s extras=%s",
|
||||
vendor, token[:12], title, body, extras,
|
||||
)
|
||||
return {
|
||||
"mock": True,
|
||||
@@ -153,28 +133,7 @@ def send_notification(
|
||||
"xiaomi": _send_xiaomi,
|
||||
"oppo": _send_oppo,
|
||||
}
|
||||
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
|
||||
return dispatch[vendor](token, title, body, extras)
|
||||
|
||||
|
||||
def send_accessibility_disabled(
|
||||
@@ -200,120 +159,13 @@ def _require(value: str, name: str) -> str:
|
||||
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(
|
||||
method: str,
|
||||
url: str,
|
||||
*,
|
||||
vendor: str,
|
||||
operation: str,
|
||||
expected_status: tuple[int, ...] = (200,),
|
||||
**kwargs: Any,
|
||||
) -> dict[str, Any]:
|
||||
started = time.perf_counter()
|
||||
request_summary = _request_summary(kwargs)
|
||||
try:
|
||||
resp = httpx.request(
|
||||
method,
|
||||
@@ -322,82 +174,41 @@ def _request_json(
|
||||
**kwargs,
|
||||
)
|
||||
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
|
||||
|
||||
if resp.status_code not in expected_status:
|
||||
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),
|
||||
)
|
||||
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:
|
||||
data = resp.json()
|
||||
return resp.json()
|
||||
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
|
||||
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(
|
||||
method: str,
|
||||
url: str,
|
||||
*,
|
||||
vendor: str,
|
||||
operation: str,
|
||||
expected_status: tuple[int, ...] = (200,),
|
||||
**kwargs: Any,
|
||||
) -> dict[str, Any]:
|
||||
return _request_json(
|
||||
method,
|
||||
url,
|
||||
vendor=vendor,
|
||||
operation=operation,
|
||||
expected_status=expected_status,
|
||||
**kwargs,
|
||||
)
|
||||
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:
|
||||
@@ -423,8 +234,6 @@ def _honor_access_token() -> str:
|
||||
data = _request_form(
|
||||
"POST",
|
||||
settings.HONOR_PUSH_TOKEN_ENDPOINT,
|
||||
vendor="honor",
|
||||
operation="authenticate",
|
||||
data={
|
||||
"grant_type": "client_credentials",
|
||||
"client_id": client_id,
|
||||
@@ -434,7 +243,7 @@ def _honor_access_token() -> str:
|
||||
)
|
||||
token = data.get("access_token")
|
||||
if not token:
|
||||
raise VendorPushError(f"honor auth failed: {_raw_log_summary(data)}")
|
||||
raise VendorPushError(f"honor auth failed: {data}")
|
||||
return _cache_put(cache_key, str(token), data.get("expires_in"))
|
||||
|
||||
|
||||
@@ -461,8 +270,6 @@ def _send_honor(token: str, title: str, body: str, extras: dict[str, str]) -> di
|
||||
data = _request_json(
|
||||
"POST",
|
||||
settings.HONOR_PUSH_SEND_ENDPOINT_TEMPLATE.format(app_id=app_id),
|
||||
vendor="honor",
|
||||
operation="send_notification",
|
||||
json=payload,
|
||||
headers={
|
||||
"Content-Type": "application/json; charset=UTF-8",
|
||||
@@ -472,7 +279,7 @@ def _send_honor(token: str, title: str, body: str, extras: dict[str, str]) -> di
|
||||
)
|
||||
code = data.get("code")
|
||||
if code is not None and int(code) != 200:
|
||||
raise VendorPushError(f"honor push failed: {_raw_log_summary(data)}")
|
||||
raise VendorPushError(f"honor push failed: {data}")
|
||||
return data
|
||||
|
||||
|
||||
@@ -487,8 +294,6 @@ def _huawei_access_token() -> str:
|
||||
data = _request_form(
|
||||
"POST",
|
||||
settings.HUAWEI_PUSH_TOKEN_ENDPOINT,
|
||||
vendor="huawei",
|
||||
operation="authenticate",
|
||||
data={
|
||||
"grant_type": "client_credentials",
|
||||
"client_id": app_id,
|
||||
@@ -498,7 +303,7 @@ def _huawei_access_token() -> str:
|
||||
)
|
||||
token = data.get("access_token")
|
||||
if not token:
|
||||
raise VendorPushError(f"huawei auth failed: {_raw_log_summary(data)}")
|
||||
raise VendorPushError(f"huawei auth failed: {data}")
|
||||
return _cache_put(cache_key, str(token), data.get("expires_in"))
|
||||
|
||||
|
||||
@@ -528,8 +333,6 @@ def _send_huawei(token: str, title: str, body: str, extras: dict[str, str]) -> d
|
||||
data = _request_json(
|
||||
"POST",
|
||||
settings.HUAWEI_PUSH_SEND_ENDPOINT_TEMPLATE.format(app_id=app_id),
|
||||
vendor="huawei",
|
||||
operation="send_notification",
|
||||
json=payload,
|
||||
headers={
|
||||
"Content-Type": "application/json; charset=UTF-8",
|
||||
@@ -537,7 +340,7 @@ def _send_huawei(token: str, title: str, body: str, extras: dict[str, str]) -> d
|
||||
},
|
||||
)
|
||||
if str(data.get("code", "")) != "80000000":
|
||||
raise VendorPushError(f"huawei push failed: {_raw_log_summary(data)}")
|
||||
raise VendorPushError(f"huawei push failed: {data}")
|
||||
return data
|
||||
|
||||
|
||||
@@ -554,8 +357,6 @@ def _vivo_auth_token() -> str:
|
||||
data = _request_json(
|
||||
"POST",
|
||||
settings.VIVO_PUSH_AUTH_ENDPOINT,
|
||||
vendor="vivo",
|
||||
operation="authenticate",
|
||||
json={
|
||||
"appId": app_id,
|
||||
"appKey": app_key,
|
||||
@@ -565,10 +366,10 @@ def _vivo_auth_token() -> str:
|
||||
headers={"Content-Type": "application/json"},
|
||||
)
|
||||
if int(data.get("result", -1)) != 0:
|
||||
raise VendorPushError(f"vivo auth failed: {_raw_log_summary(data)}")
|
||||
raise VendorPushError(f"vivo auth failed: {data}")
|
||||
token = data.get("authToken")
|
||||
if not token:
|
||||
raise VendorPushError(f"vivo auth missing authToken: {_raw_log_summary(data)}")
|
||||
raise VendorPushError(f"vivo auth missing authToken: {data}")
|
||||
return _cache_put(cache_key, str(token), 24 * 3600)
|
||||
|
||||
|
||||
@@ -600,8 +401,6 @@ def _send_vivo(token: str, title: str, body: str, extras: dict[str, str]) -> dic
|
||||
data = _request_json(
|
||||
"POST",
|
||||
settings.VIVO_PUSH_SEND_ENDPOINT,
|
||||
vendor="vivo",
|
||||
operation="send_notification",
|
||||
json=payload,
|
||||
headers={
|
||||
"Content-Type": "application/json",
|
||||
@@ -609,7 +408,7 @@ def _send_vivo(token: str, title: str, body: str, extras: dict[str, str]) -> dic
|
||||
},
|
||||
)
|
||||
if int(data.get("result", -1)) != 0:
|
||||
raise VendorPushError(f"vivo push failed: {_raw_log_summary(data)}")
|
||||
raise VendorPushError(f"vivo push failed: {data}")
|
||||
return data
|
||||
|
||||
|
||||
@@ -647,16 +446,14 @@ def _send_xiaomi(token: str, title: str, body: str, extras: dict[str, str]) -> d
|
||||
data = _request_form(
|
||||
"POST",
|
||||
settings.XIAOMI_PUSH_SEND_ENDPOINT,
|
||||
vendor="xiaomi",
|
||||
operation="send_notification",
|
||||
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: {_raw_log_summary(data)}")
|
||||
raise VendorPushError(f"xiaomi push failed: {data}")
|
||||
if str(data.get("result", "ok")).lower() not in ("ok", "success"):
|
||||
raise VendorPushError(f"xiaomi push failed: {_raw_log_summary(data)}")
|
||||
raise VendorPushError(f"xiaomi push failed: {data}")
|
||||
return data
|
||||
|
||||
|
||||
@@ -725,8 +522,6 @@ def _oppo_auth_token() -> str:
|
||||
data = _request_form(
|
||||
"POST",
|
||||
settings.OPPO_PUSH_AUTH_ENDPOINT,
|
||||
vendor="oppo",
|
||||
operation="authenticate",
|
||||
data={
|
||||
"app_key": app_key,
|
||||
"timestamp": timestamp,
|
||||
@@ -735,10 +530,10 @@ def _oppo_auth_token() -> str:
|
||||
headers={"Content-Type": "application/x-www-form-urlencoded"},
|
||||
)
|
||||
if int(data.get("code", -1)) != 0:
|
||||
raise VendorPushError(f"oppo auth failed: {_raw_log_summary(data)}")
|
||||
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: {_raw_log_summary(data)}")
|
||||
raise VendorPushError(f"oppo auth missing auth_token: {data}")
|
||||
return _cache_put(cache_key, str(token), 24 * 3600)
|
||||
|
||||
|
||||
@@ -778,8 +573,6 @@ def _send_oppo(token: str, title: str, body: str, extras: dict[str, str]) -> dic
|
||||
data = _request_form(
|
||||
"POST",
|
||||
settings.OPPO_PUSH_SEND_ENDPOINT,
|
||||
vendor="oppo",
|
||||
operation="send_notification",
|
||||
data={
|
||||
"auth_token": auth_token,
|
||||
"message": json.dumps(message, ensure_ascii=False),
|
||||
@@ -787,5 +580,5 @@ def _send_oppo(token: str, title: str, body: str, extras: dict[str, str]) -> dic
|
||||
headers={"Content-Type": "application/x-www-form-urlencoded"},
|
||||
)
|
||||
if int(data.get("code", -1)) != 0:
|
||||
raise VendorPushError(f"oppo push failed: {_raw_log_summary(data)}")
|
||||
raise VendorPushError(f"oppo push failed: {data}")
|
||||
return data
|
||||
|
||||
@@ -277,6 +277,9 @@ class CouponSession(Base):
|
||||
platform_elapsed: Mapped[dict | None] = mapped_column(_JSON, nullable=True)
|
||||
# 领到总张数(收尾帧带)。
|
||||
claimed_count: Mapped[int | None] = mapped_column(Integer, nullable=True)
|
||||
# pricebot 本轮计划处理的券数。由 /coupon/step 每帧 status.progress.total 回写;
|
||||
# 即使用户中途退出、尚无任何单券终态,admin 也能以计划数作为成功率分母。
|
||||
planned_coupon_count: Mapped[int | None] = mapped_column(Integer, nullable=True)
|
||||
# 本次 session 至少领到一张(status∈{success,already_claimed})的平台 id 列表,如 ["meituan-waimai","jd-waimai"]。
|
||||
# admin「领券数据」据此算整单成功率(②)/点位成功率(③);服务端 /step 逐帧按 trace_id 并集写入
|
||||
# (见 coupon_state.merge_session_platform_success)。旧行=NULL → 视作空集。
|
||||
|
||||
@@ -38,7 +38,7 @@ class GuideVideoPlay(Base):
|
||||
# start_play 捕获 IntegrityError 降级成"这次不放视频"。
|
||||
# 用 unique Index 而非 UniqueConstraint:与迁移里的 create_index 对齐(SQLite 加约束
|
||||
# 要整表重建),autogenerate 才不会每次报一条假 diff。
|
||||
Index("uq_guide_video_play_user_scene_seq", "user_id", "scene", "seq", unique=True),
|
||||
Index("uq_guide_video_play_user_seq", "user_id", "seq", unique=True),
|
||||
)
|
||||
|
||||
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
||||
|
||||
@@ -157,6 +157,22 @@ def session_app_env(db: Session, trace_id: str | None) -> str | None:
|
||||
).scalar_one_or_none()
|
||||
|
||||
|
||||
def merge_session_planned_count(
|
||||
db: Session, trace_id: str | None, planned_count: int | None
|
||||
) -> None:
|
||||
"""把 pricebot 队列总数回写 session,供中途退出/全跳过场次计算逐券分母。"""
|
||||
if not trace_id or planned_count is None or planned_count <= 0:
|
||||
return
|
||||
row = db.execute(
|
||||
select(CouponSession).where(CouponSession.trace_id == trace_id)
|
||||
).scalar_one_or_none()
|
||||
if row is None:
|
||||
return
|
||||
# 队列在一个 trace 内固定;取大值可防乱序旧帧覆盖,也兼容调度层补入预跳券。
|
||||
row.planned_coupon_count = max(row.planned_coupon_count or 0, planned_count)
|
||||
db.commit()
|
||||
|
||||
|
||||
def record_claims(
|
||||
db: Session,
|
||||
device_id: str,
|
||||
|
||||
@@ -30,11 +30,7 @@ from app.models.app_config import AppConfig
|
||||
from app.models.guide_video import GuideVideoPlay
|
||||
from app.repositories import wallet as crud_wallet
|
||||
|
||||
SCENES = ("coupon", "comparison")
|
||||
_KEY_BY_SCENE = {
|
||||
"coupon": "coupon_guide_video",
|
||||
"comparison": "comparison_guide_video",
|
||||
}
|
||||
_KEY = "coupon_guide_video"
|
||||
|
||||
#: 金币流水 biz_type。客户端收益明细按它显示「新手引导视频奖励」。
|
||||
BIZ_TYPE = "guide_video"
|
||||
@@ -45,7 +41,7 @@ _DEFAULTS: dict[str, Any] = {
|
||||
"enabled": True,
|
||||
"video_url": None, # None/空 = 未配片 → 不下发,浮层照旧放广告
|
||||
"max_plays": 3, # 每个账号前 N 次浮层放引导视频
|
||||
"reward_coin": 100, # 每次固定金币
|
||||
"reward_coin": 120, # 每次固定金币
|
||||
}
|
||||
|
||||
_FIELDS = tuple(_DEFAULTS.keys())
|
||||
@@ -69,28 +65,19 @@ def _merge(raw: Any) -> dict[str, Any]:
|
||||
return out
|
||||
|
||||
|
||||
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]:
|
||||
def get_config(db: Session) -> dict[str, Any]:
|
||||
"""完整配置 + updated_at(admin 读 / 业务读共用)。"""
|
||||
row = db.get(AppConfig, _config_key(scene))
|
||||
row = db.get(AppConfig, _KEY)
|
||||
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
|
||||
return cfg
|
||||
|
||||
|
||||
def _write(
|
||||
db: Session, value: dict[str, Any], *, scene: str, admin_id: int, commit: bool
|
||||
) -> dict[str, Any]:
|
||||
def _write(db: Session, value: dict[str, Any], *, admin_id: int, commit: bool) -> dict[str, Any]:
|
||||
"""整体覆写该行(value 须为完整字段 dict),返回合并后的完整配置(含 updated_at)。"""
|
||||
key = _config_key(scene)
|
||||
row = db.get(AppConfig, key)
|
||||
row = db.get(AppConfig, _KEY)
|
||||
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)
|
||||
else:
|
||||
row.value = value # 整体重新赋值,SQLAlchemy 才侦测得到变更
|
||||
@@ -111,12 +98,11 @@ def update_config(
|
||||
enabled: bool | None = None,
|
||||
max_plays: int | None = None,
|
||||
reward_coin: int | None = None,
|
||||
scene: str = "coupon",
|
||||
admin_id: int,
|
||||
commit: bool = True,
|
||||
) -> tuple[dict[str, Any], dict[str, Any]]:
|
||||
"""改开关 / 次数 / 金币(只改传了的字段;视频走 set_video)。返回 (before, after) 供审计。"""
|
||||
row = db.get(AppConfig, _config_key(scene))
|
||||
row = db.get(AppConfig, _KEY)
|
||||
before = _merge(row.value if row is not None else None)
|
||||
new_value = {k: before[k] for k in _FIELDS}
|
||||
if enabled is not None:
|
||||
@@ -125,52 +111,45 @@ def update_config(
|
||||
new_value["max_plays"] = max(0, min(int(max_plays), MAX_PLAYS_LIMIT))
|
||||
if reward_coin is not None:
|
||||
new_value["reward_coin"] = max(0, min(int(reward_coin), REWARD_COIN_LIMIT))
|
||||
after = _write(db, new_value, scene=scene, admin_id=admin_id, commit=commit)
|
||||
after = _write(db, new_value, admin_id=admin_id, commit=commit)
|
||||
return before, after
|
||||
|
||||
|
||||
def set_video(
|
||||
db: Session, video_url: str | None, *, scene: str = "coupon",
|
||||
admin_id: int, commit: bool = True
|
||||
db: Session, video_url: str | None, *, admin_id: int, commit: bool = True
|
||||
) -> tuple[dict[str, Any], dict[str, Any]]:
|
||||
"""设置/清空引导视频地址。返回 (before, after);before['video_url'] 供调用方删旧文件。"""
|
||||
row = db.get(AppConfig, _config_key(scene))
|
||||
row = db.get(AppConfig, _KEY)
|
||||
before = _merge(row.value if row is not None else None)
|
||||
new_value = {k: before[k] for k in _FIELDS}
|
||||
new_value["video_url"] = video_url
|
||||
after = _write(db, new_value, scene=scene, admin_id=admin_id, commit=commit)
|
||||
after = _write(db, new_value, admin_id=admin_id, commit=commit)
|
||||
return before, after
|
||||
|
||||
|
||||
# ===== 播放计次 =====
|
||||
|
||||
|
||||
def used_plays(db: Session, user_id: int, scene: str = "coupon") -> int:
|
||||
def used_plays(db: Session, user_id: int) -> int:
|
||||
"""该账号已用掉的引导视频次数(开播即算,含未发币的)。"""
|
||||
return int(
|
||||
db.execute(
|
||||
select(func.count()).select_from(GuideVideoPlay).where(
|
||||
GuideVideoPlay.user_id == user_id,
|
||||
GuideVideoPlay.scene == scene,
|
||||
GuideVideoPlay.user_id == user_id
|
||||
)
|
||||
).scalar_one()
|
||||
)
|
||||
|
||||
|
||||
def play_stats(db: Session, scene: str = "coupon") -> dict[str, int]:
|
||||
def play_stats(db: Session) -> dict[str, int]:
|
||||
"""全站播放统计(admin 页展示):总播放次数 / 其中已发币次数。"""
|
||||
total = int(
|
||||
db.execute(
|
||||
select(func.count()).select_from(GuideVideoPlay).where(
|
||||
GuideVideoPlay.scene == scene
|
||||
)
|
||||
).scalar_one()
|
||||
db.execute(select(func.count()).select_from(GuideVideoPlay)).scalar_one()
|
||||
)
|
||||
granted = int(
|
||||
db.execute(
|
||||
select(func.count()).select_from(GuideVideoPlay).where(
|
||||
GuideVideoPlay.status == "granted",
|
||||
GuideVideoPlay.scene == scene,
|
||||
GuideVideoPlay.status == "granted"
|
||||
)
|
||||
).scalar_one()
|
||||
)
|
||||
@@ -189,12 +168,11 @@ def start_play(
|
||||
reward_coin 播完/中途关闭都发的固定金币
|
||||
seq / remaining 第几次 / 发完这次还剩几次(仅展示与排查用)
|
||||
"""
|
||||
_config_key(scene)
|
||||
cfg = get_config(db, scene)
|
||||
cfg = get_config(db)
|
||||
video_url = (cfg.get("video_url") or "").strip()
|
||||
max_plays = int(cfg.get("max_plays") or 0)
|
||||
reward_coin = int(cfg.get("reward_coin") or 0)
|
||||
used = used_plays(db, user_id, scene)
|
||||
used = used_plays(db, user_id)
|
||||
|
||||
def _miss(used_now: int) -> dict[str, Any]:
|
||||
return {
|
||||
@@ -216,7 +194,7 @@ def start_play(
|
||||
scene=scene,
|
||||
seq=seq,
|
||||
video_url=video_url,
|
||||
coin=reward_coin,
|
||||
coin=0,
|
||||
status="playing",
|
||||
completed=0,
|
||||
started_at=datetime.now(rewards.CN_TZ).replace(tzinfo=None),
|
||||
@@ -232,7 +210,7 @@ def start_play(
|
||||
db.flush()
|
||||
except IntegrityError:
|
||||
db.rollback()
|
||||
return _miss(used_plays(db, user_id, scene))
|
||||
return _miss(used_plays(db, user_id))
|
||||
return {
|
||||
"should_play": True,
|
||||
"video_url": video_url,
|
||||
@@ -263,8 +241,7 @@ def grant_play(
|
||||
"""
|
||||
token = (play_token or "").strip()
|
||||
# 金币额度以**服务端配置**为准,不信客户端(客户端只上报"播完/关闭")。
|
||||
play = _find_play(db, user_id, token)
|
||||
coin = int(play.coin if play is not None else 0)
|
||||
coin = int(get_config(db).get("reward_coin") or 0)
|
||||
|
||||
# 幂等核心:把 status 放进 WHERE 做条件更新(compare-and-set),而不是"先读再判再写"。
|
||||
# 「播完」与「✕ 关闭」抢跑、或客户端超时重试时,两个请求会都读到 status='playing',
|
||||
|
||||
@@ -1,15 +1,13 @@
|
||||
"""新手引导视频(领券等候浮层前 N 次替代广告)的客户端请求/响应契约。"""
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Literal
|
||||
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
|
||||
class GuideVideoStartIn(BaseModel):
|
||||
"""开播询问。scene 目前只有 coupon(领券浮层);预留给日后比价等场景。"""
|
||||
|
||||
scene: Literal["coupon", "comparison"] = "coupon"
|
||||
scene: str = Field(default="coupon", max_length=16)
|
||||
|
||||
|
||||
class GuideVideoStartOut(BaseModel):
|
||||
|
||||
@@ -10,14 +10,16 @@ from app.admin.repositories.coupon_data import (
|
||||
_point_scores_by_trace,
|
||||
coupon_data_report,
|
||||
coupon_point_details,
|
||||
coupon_user_records,
|
||||
)
|
||||
from app.admin.security import create_admin_token
|
||||
from app.db.session import SessionLocal
|
||||
from app.models.coupon_state import CouponClaimEvent, CouponSession
|
||||
from app.repositories.coupon_state import merge_session_planned_count
|
||||
|
||||
|
||||
def test_point_scores_by_trace() -> None:
|
||||
"""已领算成功、失败算尝试、跳过不进分母。"""
|
||||
"""已领算成功;失败和跳过均属于本轮计划券,进入分母。"""
|
||||
db = SessionLocal()
|
||||
trace = "point-score-trace"
|
||||
try:
|
||||
@@ -37,7 +39,7 @@ def test_point_scores_by_trace() -> None:
|
||||
|
||||
stats = _point_scores_by_trace(db, [trace])[trace]
|
||||
assert stats["succeeded"] == 2
|
||||
assert stats["tried"] == 3
|
||||
assert stats["tried"] == 4
|
||||
details = coupon_point_details(db, trace_id=trace)
|
||||
assert [item["status"] for item in details] == [
|
||||
"success", "already_claimed", "failed", "skipped"
|
||||
@@ -48,8 +50,8 @@ def test_point_scores_by_trace() -> None:
|
||||
db.close()
|
||||
|
||||
|
||||
def test_skipped_detail_does_not_create_a_score() -> None:
|
||||
"""仅有 skipped 时按需明细仍可查到,但列表没有虚假的 0/0 分数。"""
|
||||
def test_skipped_detail_creates_zero_score() -> None:
|
||||
"""仅有 skipped 时也应显示 0/1,而不是把计划券静默成空值。"""
|
||||
db = SessionLocal()
|
||||
trace = "point-score-skipped"
|
||||
try:
|
||||
@@ -63,7 +65,7 @@ def test_skipped_detail_does_not_create_a_score() -> None:
|
||||
db.flush()
|
||||
|
||||
scores = _point_scores_by_trace(db, [trace, "missing-trace"])
|
||||
assert trace not in scores
|
||||
assert scores[trace] == {"succeeded": 0, "tried": 1}
|
||||
assert "missing-trace" not in scores
|
||||
assert coupon_point_details(db, trace_id=trace)[0]["status"] == "skipped"
|
||||
finally:
|
||||
@@ -114,6 +116,126 @@ def test_coupon_data_report_returns_scores_without_embedding_details() -> None:
|
||||
db.close()
|
||||
|
||||
|
||||
def test_report_uses_planned_count_when_events_are_partial_or_missing() -> None:
|
||||
"""退出前只有部分/没有逐券终态时,计划数仍是稳定分母。"""
|
||||
db = SessionLocal()
|
||||
report_date = date(2020, 1, 6)
|
||||
partial_trace = "point-score-planned-partial"
|
||||
empty_trace = "point-score-planned-empty"
|
||||
try:
|
||||
db.add_all([
|
||||
CouponSession(
|
||||
trace_id=partial_trace,
|
||||
device_id="planned-partial-device",
|
||||
status="abandoned",
|
||||
app_env="prod",
|
||||
platforms=[],
|
||||
planned_coupon_count=8,
|
||||
started_at=datetime(2020, 1, 6, tzinfo=UTC),
|
||||
started_date=report_date,
|
||||
),
|
||||
CouponSession(
|
||||
trace_id=empty_trace,
|
||||
device_id="planned-empty-device",
|
||||
status="abandoned",
|
||||
app_env="prod",
|
||||
platforms=[],
|
||||
planned_coupon_count=8,
|
||||
started_at=datetime(2020, 1, 6, tzinfo=UTC),
|
||||
started_date=report_date,
|
||||
),
|
||||
CouponClaimEvent(
|
||||
trace_id=partial_trace,
|
||||
device_id="planned-partial-device",
|
||||
coupon_id="mt-planned-success",
|
||||
claim_date=report_date,
|
||||
status="success",
|
||||
),
|
||||
])
|
||||
db.flush()
|
||||
|
||||
report = coupon_data_report(
|
||||
db,
|
||||
date_from=report_date.isoformat(),
|
||||
date_to=report_date.isoformat(),
|
||||
app_env="prod",
|
||||
)
|
||||
rows = {item["trace_id"]: item for item in report["items"]}
|
||||
assert rows[partial_trace]["point_success_count"] == 1
|
||||
assert rows[partial_trace]["point_total_count"] == 8
|
||||
assert rows[empty_trace]["point_success_count"] == 0
|
||||
assert rows[empty_trace]["point_total_count"] == 8
|
||||
finally:
|
||||
db.rollback()
|
||||
db.close()
|
||||
|
||||
|
||||
def test_user_records_include_actual_successes_with_planned_count() -> None:
|
||||
"""用户领券记录也必须加载逐券统计,不能把有成功事件的场次显示成 0/N。"""
|
||||
db = SessionLocal()
|
||||
trace = "point-score-user-records"
|
||||
user_id = 987656
|
||||
report_date = date(2020, 1, 8)
|
||||
try:
|
||||
db.add_all([
|
||||
CouponSession(
|
||||
trace_id=trace,
|
||||
device_id="planned-user-records-device",
|
||||
user_id=user_id,
|
||||
status="abandoned",
|
||||
app_env="prod",
|
||||
platforms=[],
|
||||
planned_coupon_count=8,
|
||||
started_at=datetime(2020, 1, 8, tzinfo=UTC),
|
||||
started_date=report_date,
|
||||
),
|
||||
CouponClaimEvent(
|
||||
trace_id=trace,
|
||||
device_id="planned-user-records-device",
|
||||
coupon_id="mt-user-records-success",
|
||||
claim_date=report_date,
|
||||
status="success",
|
||||
),
|
||||
])
|
||||
db.flush()
|
||||
|
||||
records = coupon_user_records(db, user_id=user_id)
|
||||
row = next(item for item in records["items"] if item["trace_id"] == trace)
|
||||
assert row["point_success_count"] == 1
|
||||
assert row["point_total_count"] == 8
|
||||
finally:
|
||||
db.rollback()
|
||||
db.close()
|
||||
|
||||
|
||||
def test_planned_count_only_grows_for_a_trace() -> None:
|
||||
"""乱序帧不得用较小的队列总数覆盖已经观测到的计划数。"""
|
||||
db = SessionLocal()
|
||||
trace = "point-score-planned-upsert"
|
||||
try:
|
||||
row = CouponSession(
|
||||
trace_id=trace,
|
||||
device_id="planned-upsert-device",
|
||||
status="started",
|
||||
app_env="prod",
|
||||
platforms=[],
|
||||
started_at=datetime(2020, 1, 7, tzinfo=UTC),
|
||||
started_date=date(2020, 1, 7),
|
||||
)
|
||||
db.add(row)
|
||||
db.commit()
|
||||
|
||||
merge_session_planned_count(db, trace, 8)
|
||||
merge_session_planned_count(db, trace, 3)
|
||||
db.refresh(row)
|
||||
assert row.planned_coupon_count == 8
|
||||
finally:
|
||||
db.rollback()
|
||||
db.execute(delete(CouponSession).where(CouponSession.trace_id == trace))
|
||||
db.commit()
|
||||
db.close()
|
||||
|
||||
|
||||
def test_coupon_point_details_endpoint() -> None:
|
||||
"""前端点击使用的接口按约定返回 trace_id 和逐券 items。"""
|
||||
db = SessionLocal()
|
||||
|
||||
@@ -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
|
||||
|
||||
# 本次请求读到的是过期计数 → 仍会算出 seq=1
|
||||
monkeypatch.setattr(crud_guide, "used_plays", lambda db, user_id, scene="coupon": 0)
|
||||
monkeypatch.setattr(crud_guide, "used_plays", lambda db, user_id: 0)
|
||||
with SessionLocal() as db:
|
||||
result = crud_guide.start_play(db, uid)
|
||||
|
||||
|
||||
@@ -7,9 +7,7 @@ mock/真发/变量覆盖/站内联动/设备反查/参数校验。厂商 HTTP
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
@@ -112,99 +110,6 @@ def test_huawei_non_success_code_raises(monkeypatch) -> None:
|
||||
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:
|
||||
assert vendor_push.normalize_vendor("华为") == "huawei"
|
||||
assert vendor_push.normalize_vendor("HMS") == "huawei"
|
||||
|
||||
Reference in New Issue
Block a user