feat(coupon): 领券今日状态后端(弹窗频控 + 领券记录) (#23)

- 两张表 + 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

app/api/v1/coupon.py

@@ -16,15 +16,66 @@ 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,
@@ -45,6 +96,21 @@ async def coupon_step(
     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"
@@ -80,4 +146,44 @@ async def coupon_step(
             detail=f"pricebot upstream returned {resp.status_code}",
         )
 
-    return resp.json()
+    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)
+    )