357ba27499
- 两张表 + Alembic 迁移:coupon_prompt_engagement(device×日,弹窗频控源)/ coupon_claim_record(device×券×日,领券记录·资产) - coupon/step 透传顺手写库:step0 写 engagement(claim_started);响应解析 last_coupon_result/coupon_results 幂等写 claim record(按 coupon_id 去重,避免 done 帧 last 与 coupon_results 重复在 autoflush=False 下撞唯一约束回滚整批) - 新增 GET /coupon/prompt/should-show(弹窗判据)+ POST /coupon/prompt/dismiss(拒绝通知,透传链路看不到拒绝) - 判断按 device_id(券发到设备登录的外卖账号,比 user 更贴近登录环境);user_id 仅资产留痕、可空 - MVP 先不去重(队列照旧全发);pricebot 未改 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Reviewed-on: #23
190 lines
7.3 KiB
Python
190 lines
7.3 KiB
Python
"""领券业务透传端点。
|
|
|
|
把客户端的 POST /api/v1/coupon/step 请求原样转发给 pricebot-backend
|
|
的 /api/coupon/step。MVP 阶段**不鉴权**(见 docs/待办与技术债.md P1:
|
|
device_id 透传给 pricebot 区分设备,待补 JWT + device_id↔user_id 绑定后
|
|
才能做用户级画像)。
|
|
|
|
pricebot 协议文档:
|
|
pricebot-backend/docs/projects/领券-客户端对接协议.md
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
from typing import Any
|
|
|
|
import httpx
|
|
from fastapi import APIRouter, HTTPException, Request, status
|
|
from fastapi.concurrency import run_in_threadpool
|
|
|
|
from app.api.deps import DbSession
|
|
from app.core.config import settings
|
|
from app.core.pricebot_router import pick_pricebot
|
|
from app.db.session import SessionLocal
|
|
from app.repositories import coupon_state as coupon_repo
|
|
from app.schemas.coupon_state import CouponPromptDismissIn, CouponPromptShouldShowOut
|
|
|
|
logger = logging.getLogger("shagua.coupon")
|
|
|
|
router = APIRouter(prefix="/api/v1/coupon", tags=["coupon"])
|
|
|
|
|
|
def _to_int(v: object) -> int | None:
|
|
"""user_id 协议是字符串(登录态才带),转 int 存库;缺失/非法 → None。"""
|
|
if v is None:
|
|
return None
|
|
try:
|
|
return int(v) # type: ignore[arg-type]
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
def _extract_coupon_results(resp_json: dict) -> list[dict]:
|
|
"""从 pricebot 响应抽本帧券结果,并**按 coupon_id 去重**。
|
|
|
|
⚠️ done/单券帧同时带 last_coupon_result(最后一张)+ action.params.coupon_results
|
|
(全量含最后一张)→ 最后一张出现两次。必须去重:同批里同 coupon_id 两次,
|
|
record_claims 在 autoflush=False 下两次 select 都查不到刚 add 的行 → 重复 add →
|
|
commit 撞唯一约束 IntegrityError 回滚整批(done/单券记录全丢)。
|
|
coupon_results(全量权威)覆盖 last_coupon_result。"""
|
|
by_id: dict[str, dict] = {}
|
|
last = resp_json.get("last_coupon_result")
|
|
if isinstance(last, dict) and last.get("coupon_id"):
|
|
by_id[last["coupon_id"]] = last
|
|
params = (resp_json.get("action") or {}).get("params") or {}
|
|
cr = params.get("coupon_results")
|
|
if isinstance(cr, list):
|
|
for x in cr:
|
|
if isinstance(x, dict) and x.get("coupon_id"):
|
|
by_id[x["coupon_id"]] = x
|
|
return list(by_id.values())
|
|
|
|
|
|
def _mark_engagement_blocking(
|
|
device_id: str, user_id: int | None, engage_type: str
|
|
) -> None:
|
|
"""独立 session 写 engagement(async 端点经 run_in_threadpool 调,不阻塞事件循环)。"""
|
|
with SessionLocal() as db:
|
|
coupon_repo.mark_engagement(db, device_id, user_id, engage_type)
|
|
|
|
|
|
def _record_claims_blocking(
|
|
device_id: str, user_id: int | None, trace_id: str | None, results: list[dict]
|
|
) -> None:
|
|
with SessionLocal() as db:
|
|
coupon_repo.record_claims(db, device_id, user_id, trace_id, results)
|
|
|
|
|
|
@router.post("/step", summary="领券任务步进 (透传到 pricebot)")
|
|
async def coupon_step(
|
|
request: Request,
|
|
) -> dict[str, Any]:
|
|
"""把请求体原样转发给 pricebot-backend 的 /api/coupon/step。
|
|
|
|
- 鉴权: MVP 阶段不鉴权(待补 JWT,见 docs/待办与技术债.md P1)
|
|
- 透传: 不做 schema 校验,pricebot 自己校验
|
|
- 失败: 网络不可达 / pricebot 5xx → 502 + 友好 message
|
|
"""
|
|
# 读原始字节,避免"反序列化→再序列化"的双重 JSON(省透传 CPU)。只 loads 一次拿
|
|
# trace_id 做亲和 + 打日志,转发时直接发原始 bytes,不重新 dumps。
|
|
raw = await request.body()
|
|
try:
|
|
meta = json.loads(raw)
|
|
except Exception as e:
|
|
raise HTTPException(status_code=400, detail=f"invalid json body: {e}") from e
|
|
if not isinstance(meta, dict):
|
|
meta = {}
|
|
|
|
device_id = meta.get("device_id")
|
|
user_id = _to_int(meta.get("user_id")) # 登录态才带;判断不靠它,资产留痕用
|
|
trace_id = meta.get("trace_id")
|
|
|
|
# 领券任务首帧(step=0)= 用户已发起领券 → 记一条今日 engagement(claim_started),
|
|
# 今天这台设备不再弹引导窗(对齐前台"点一键领取即 markEngaged")。写库失败绝不能
|
|
# 连累领券主流程,整段吞掉。
|
|
if device_id and meta.get("step") == 0:
|
|
try:
|
|
await run_in_threadpool(
|
|
_mark_engagement_blocking, device_id, user_id, "claim_started"
|
|
)
|
|
except Exception as e: # noqa: BLE001
|
|
logger.warning("coupon engagement write failed: %s", e)
|
|
|
|
# 按 trace_id 一致性 hash 选 pricebot 实例(同一领券任务的所有帧落同一进程)
|
|
base = pick_pricebot(meta.get("trace_id"))
|
|
url = f"{base.rstrip('/')}/api/coupon/step"
|
|
timeout = settings.PRICEBOT_REQUEST_TIMEOUT_SEC
|
|
|
|
logger.info(
|
|
"coupon_step device_id=%s trace_id=%s step=%s",
|
|
meta.get("device_id"),
|
|
meta.get("trace_id"),
|
|
meta.get("step"),
|
|
)
|
|
|
|
try:
|
|
async with httpx.AsyncClient(timeout=timeout) as client:
|
|
resp = await client.post(
|
|
url, content=raw, headers={"Content-Type": "application/json"}
|
|
)
|
|
except httpx.RequestError as e:
|
|
logger.error("[pricebot] request failed: %s", e)
|
|
raise HTTPException(
|
|
status_code=status.HTTP_502_BAD_GATEWAY,
|
|
detail=f"pricebot upstream unreachable: {e}",
|
|
) from e
|
|
|
|
if resp.status_code >= 500:
|
|
logger.error(
|
|
"[pricebot] 5xx status=%d body=%s",
|
|
resp.status_code,
|
|
resp.text[:500],
|
|
)
|
|
raise HTTPException(
|
|
status_code=status.HTTP_502_BAD_GATEWAY,
|
|
detail=f"pricebot upstream returned {resp.status_code}",
|
|
)
|
|
|
|
resp_json = resp.json()
|
|
|
|
# 领券结果沉淀(资产):中间帧 last_coupon_result + done 帧 coupon_results 幂等写库。
|
|
# 当前只记录、不参与"要不要领"判断(MVP 先不去重)。写库失败不影响返回。
|
|
if device_id:
|
|
results = _extract_coupon_results(resp_json)
|
|
if results:
|
|
try:
|
|
await run_in_threadpool(
|
|
_record_claims_blocking, device_id, user_id, trace_id, results
|
|
)
|
|
except Exception as e: # noqa: BLE001
|
|
logger.warning("coupon claim write failed: %s", e)
|
|
|
|
return resp_json
|
|
|
|
|
|
@router.post("/prompt/dismiss", summary="用户拒绝/关闭领券引导窗(记今日已 engage)")
|
|
def coupon_prompt_dismiss(payload: CouponPromptDismissIn, db: DbSession) -> dict[str, bool]:
|
|
"""客户端点关闭引导窗时调用 → 记一条今日 engagement(dismissed),今天不再弹。
|
|
|
|
server 在透传链路里看不到"用户拒绝"(拒绝不发起领券),故必须客户端通知。
|
|
MVP 不鉴权,按 device_id 记。
|
|
"""
|
|
coupon_repo.mark_engagement(db, payload.device_id, payload.user_id, "dismissed")
|
|
return {"ok": True}
|
|
|
|
|
|
@router.get(
|
|
"/prompt/should-show",
|
|
response_model=CouponPromptShouldShowOut,
|
|
summary="切到外卖 App 时是否还应弹领券引导窗",
|
|
)
|
|
def coupon_prompt_should_show(
|
|
device_id: str, db: DbSession
|
|
) -> CouponPromptShouldShowOut:
|
|
"""今天这台设备已 engage(领或拒)过 → should_show=false。客户端据此决定弹不弹
|
|
(前台 SP 缓存做快速路径,这里是权威)。"""
|
|
return CouponPromptShouldShowOut(
|
|
should_show=not coupon_repo.has_engaged_today(db, device_id)
|
|
)
|