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

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

alembic/versions/coupon_state_tables.py

@@ -0,0 +1,77 @@
+"""coupon state tables (领券今日状态:弹窗频控 engagement + 领券记录 claim)
+
+Revision ID: coupon_state_tables
+Revises: 11a1d08c6f55
+Create Date: 2026-06-08 00:00:00.000000
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+
+# revision identifiers, used by Alembic.
+revision: str = 'coupon_state_tables'
+down_revision: Union[str, Sequence[str], None] = '11a1d08c6f55'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+# PG 上为 JSONB,其它(SQLite dev)为 JSON——与模型层 with_variant 对齐
+_JSON = sa.JSON().with_variant(postgresql.JSONB(), 'postgresql')
+
+
+def upgrade() -> None:
+    # 领券记录(资产层,当前不参与去重判断)
+    op.create_table(
+        'coupon_claim_record',
+        sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
+        sa.Column('device_id', sa.String(length=64), nullable=False),
+        sa.Column('user_id', sa.Integer(), nullable=True),
+        sa.Column('coupon_id', sa.String(length=64), nullable=False),
+        sa.Column('claim_date', sa.Date(), nullable=False),
+        sa.Column('status', sa.String(length=24), nullable=False),
+        sa.Column('vendor', sa.String(length=48), nullable=True),
+        sa.Column('coupon_name', sa.String(length=128), nullable=True),
+        sa.Column('claimed_count', sa.Integer(), nullable=True),
+        sa.Column('trace_id', sa.String(length=64), nullable=True),
+        sa.Column('reason', sa.String(length=255), nullable=True),
+        sa.Column('extra', _JSON, nullable=True),
+        sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False),
+        sa.Column('updated_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False),
+        sa.PrimaryKeyConstraint('id'),
+        sa.UniqueConstraint('device_id', 'coupon_id', 'claim_date', name='uq_coupon_claim_device_coupon_date'),
+    )
+    with op.batch_alter_table('coupon_claim_record', schema=None) as batch_op:
+        batch_op.create_index('ix_coupon_claim_device_date', ['device_id', 'claim_date'], unique=False)
+        batch_op.create_index(batch_op.f('ix_coupon_claim_record_user_id'), ['user_id'], unique=False)
+        batch_op.create_index(batch_op.f('ix_coupon_claim_record_trace_id'), ['trace_id'], unique=False)
+
+    # 弹窗频控(今天 engage 过就不再弹)
+    op.create_table(
+        'coupon_prompt_engagement',
+        sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
+        sa.Column('device_id', sa.String(length=64), nullable=False),
+        sa.Column('user_id', sa.Integer(), nullable=True),
+        sa.Column('engage_date', sa.Date(), nullable=False),
+        sa.Column('engage_type', sa.String(length=16), nullable=False),
+        sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False),
+        sa.Column('updated_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False),
+        sa.PrimaryKeyConstraint('id'),
+        sa.UniqueConstraint('device_id', 'engage_date', name='uq_coupon_engage_device_date'),
+    )
+    with op.batch_alter_table('coupon_prompt_engagement', schema=None) as batch_op:
+        batch_op.create_index(batch_op.f('ix_coupon_prompt_engagement_user_id'), ['user_id'], unique=False)
+
+
+def downgrade() -> None:
+    with op.batch_alter_table('coupon_prompt_engagement', schema=None) as batch_op:
+        batch_op.drop_index(batch_op.f('ix_coupon_prompt_engagement_user_id'))
+    op.drop_table('coupon_prompt_engagement')
+
+    with op.batch_alter_table('coupon_claim_record', schema=None) as batch_op:
+        batch_op.drop_index(batch_op.f('ix_coupon_claim_record_trace_id'))
+        batch_op.drop_index(batch_op.f('ix_coupon_claim_record_user_id'))
+        batch_op.drop_index('ix_coupon_claim_device_date')
+    op.drop_table('coupon_claim_record')

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)
+    )

app/models/__init__.py

@@ -7,6 +7,10 @@ from app.models.admin import AdminAuditLog, AdminUser  # noqa: F401
 from app.models.app_config import AppConfig  # noqa: F401
 from app.models.comparison import ComparisonRecord  # noqa: F401
 from app.models.comparison_milestone import ComparisonMilestoneClaim  # noqa: F401
+from app.models.coupon_state import (  # noqa: F401
+    CouponClaimRecord,
+    CouponPromptEngagement,
+)
 from app.models.feedback import Feedback  # noqa: F401
 from app.models.meituan_coupon import MeituanCoupon  # noqa: F401
 from app.models.ops_marquee_seed import OpsMarqueeSeed  # noqa: F401

app/models/coupon_state.py

@@ -0,0 +1,130 @@
+"""领券今日状态(弹窗频控 + 领券记录)两张表。
+
+- `coupon_prompt_engagement`:按 (device, 自然日) 记"今天是否对领券引导窗表达过意向"
+  ——点「一键领取」(claim_started) 或 点拒绝/关闭 (dismissed) 都算。切到外卖 App 时
+  据此决定弹不弹:今天 engage 过就不再弹。判断维度是 **device_id**——券发到的是设备上
+  登录的那个外卖账号,device 比 user 更贴近"哪个登录环境",且 device_id 全链路现成、
+  不依赖领券鉴权。
+
+- `coupon_claim_record`:按 (device, 券, 自然日) 记每张券的领取结果,纯沉淀(资产/画像/
+  排查)。当前**不参与**"要不要领"的判断(MVP 先不去重:今天 engage 过就不弹,A 路径
+  主动领则全跑)。留作以后做按券去重 / CPS 归因 / 用户画像的数据源。
+
+口径:
+- 日期 = Asia/Shanghai 的自然日(claim_date / engage_date)。
+- user_id 领券登录态有就记(资产),可空、不进唯一键、不阻塞判断。
+- device_id 客户端生成存 SP,卸载重装会变 → 重装当新设备重新弹一次(产品预期)。
+"""
+from __future__ import annotations
+
+from datetime import date, datetime
+
+from sqlalchemy import (
+    JSON,
+    Date,
+    DateTime,
+    Index,
+    Integer,
+    String,
+    UniqueConstraint,
+    func,
+)
+from sqlalchemy.dialects.postgresql import JSONB
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.db.base import Base
+
+# PG 上用 JSONB,SQLite(本地/测试)退化为通用 JSON(同 price_observation / comparison)。
+_JSON = JSON().with_variant(JSONB(), "postgresql")
+
+
+class CouponClaimRecord(Base):
+    """单张券一天一条领取记录(资产层,当前不做去重判断)。"""
+
+    __tablename__ = "coupon_claim_record"
+    __table_args__ = (
+        # 同设备、同券、同一天只一条:领券每帧 last_coupon_result + done 帧 coupon_results
+        # 会重复上报同一张券,靠它幂等 upsert。
+        UniqueConstraint(
+            "device_id", "coupon_id", "claim_date",
+            name="uq_coupon_claim_device_coupon_date",
+        ),
+        # 去重/统计查询按 (device, 日) 取一天所有券。
+        Index("ix_coupon_claim_device_date", "device_id", "claim_date"),
+    )
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+
+    # 判断/聚合维度。client getOrCreateDeviceId 生成,重装会变。
+    device_id: Mapped[str] = mapped_column(String(64), nullable=False)
+    # 登录态有就记(资产/画像),可空、不进唯一键。
+    user_id: Mapped[int | None] = mapped_column(Integer, index=True, nullable=True)
+
+    coupon_id: Mapped[str] = mapped_column(String(64), nullable=False)
+    # Asia/Shanghai 自然日。每日可领的券(签到/天天红包)靠这天然每天一条。
+    claim_date: Mapped[date] = mapped_column(Date, nullable=False)
+
+    # success / already_claimed / failed / skipped(原样取 pricebot coupon 结果)
+    status: Mapped[str] = mapped_column(String(24), nullable=False)
+    vendor: Mapped[str | None] = mapped_column(String(48), nullable=True)
+    coupon_name: Mapped[str | None] = mapped_column(String(128), nullable=True)
+    # 这张领到几张(pricebot display_count;给不出时为 None)
+    claimed_count: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    # 哪次任务领的,回指 pricebot work_logs / 排查
+    trace_id: Mapped[str | None] = mapped_column(String(64), index=True, nullable=True)
+    # failed / skipped 原因
+    reason: Mapped[str | None] = mapped_column(String(255), nullable=True)
+    # 杂项兜底:券的结构化信息(面额/入口/关键节点摘要等),免得加字段就迁移。
+    # ⚠️ 别塞原始无障碍树(几十 KB → 行膨胀);原始大树看 trace_id 指过去的 work_logs。
+    extra: Mapped[dict | None] = mapped_column(_JSON, nullable=True)
+
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now(), nullable=False
+    )
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now(), onupdate=func.now(),
+        nullable=False,
+    )
+
+    def __repr__(self) -> str:  # pragma: no cover
+        return (
+            f"<CouponClaimRecord device={self.device_id} coupon={self.coupon_id} "
+            f"date={self.claim_date} status={self.status}>"
+        )
+
+
+class CouponPromptEngagement(Base):
+    """按 (device, 自然日) 记"今天是否对领券引导窗表达过意向"——弹窗频控源。"""
+
+    __tablename__ = "coupon_prompt_engagement"
+    __table_args__ = (
+        # 一台设备一天一条:今天 engage 过(领或拒)就不再弹。
+        UniqueConstraint(
+            "device_id", "engage_date",
+            name="uq_coupon_engage_device_date",
+        ),
+    )
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+
+    device_id: Mapped[str] = mapped_column(String(64), nullable=False)
+    user_id: Mapped[int | None] = mapped_column(Integer, index=True, nullable=True)
+    # Asia/Shanghai 自然日。
+    engage_date: Mapped[date] = mapped_column(Date, nullable=False)
+    # claim_started(点了一键领取)/ dismissed(点了拒绝/关闭)。仅记录区分,
+    # 判断只看"今天有没有这条",type 不影响弹不弹。
+    engage_type: Mapped[str] = mapped_column(String(16), nullable=False)
+
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now(), nullable=False
+    )
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now(), onupdate=func.now(),
+        nullable=False,
+    )
+
+    def __repr__(self) -> str:  # pragma: no cover
+        return (
+            f"<CouponPromptEngagement device={self.device_id} "
+            f"date={self.engage_date} type={self.engage_type}>"
+        )

app/repositories/coupon_state.py

@@ -0,0 +1,129 @@
+"""领券今日状态读写:弹窗频控(engagement)+ 领券记录(claim)。
+
+写操作按唯一键幂等 upsert,自带 commit + 并发 IntegrityError 兜底(对齐 price_observation)。
+日期口径 = Asia/Shanghai 的自然日。
+"""
+from __future__ import annotations
+
+import logging
+from datetime import date, datetime
+from zoneinfo import ZoneInfo
+
+from sqlalchemy import select
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.orm import Session
+
+from app.models.coupon_state import CouponClaimRecord, CouponPromptEngagement
+
+logger = logging.getLogger("shagua.coupon_state")
+
+_CN_TZ = ZoneInfo("Asia/Shanghai")
+
+
+def today_cn() -> date:
+    """Asia/Shanghai 的自然日(领券判断的"今天")。"""
+    return datetime.now(_CN_TZ).date()
+
+
+# ===== 弹窗频控(coupon_prompt_engagement)=====
+
+def has_engaged_today(db: Session, device_id: str) -> bool:
+    """这台设备今天是否已对领券引导窗表达过意向(领或拒)。有 = 不再弹。"""
+    row = db.execute(
+        select(CouponPromptEngagement.id).where(
+            CouponPromptEngagement.device_id == device_id,
+            CouponPromptEngagement.engage_date == today_cn(),
+        )
+    ).first()
+    return row is not None
+
+
+def mark_engagement(
+    db: Session, device_id: str, user_id: int | None, engage_type: str
+) -> None:
+    """记今日意向(claim_started / dismissed)。(device, 今天) 唯一,幂等 upsert。"""
+    today = today_cn()
+    row = db.execute(
+        select(CouponPromptEngagement).where(
+            CouponPromptEngagement.device_id == device_id,
+            CouponPromptEngagement.engage_date == today,
+        )
+    ).scalar_one_or_none()
+    if row is not None:
+        row.engage_type = engage_type
+        if user_id is not None:
+            row.user_id = user_id
+    else:
+        db.add(CouponPromptEngagement(
+            device_id=device_id, user_id=user_id,
+            engage_date=today, engage_type=engage_type,
+        ))
+    try:
+        db.commit()
+    except IntegrityError:
+        # 并发下另一请求刚插了同 (device, 日) → 唯一约束撞,回滚忽略(本就幂等)。
+        db.rollback()
+
+
+# ===== 领券记录(coupon_claim_record)=====
+
+def record_claims(
+    db: Session,
+    device_id: str,
+    user_id: int | None,
+    trace_id: str | None,
+    results: list[dict],
+) -> int:
+    """一批券领取结果幂等写入,返回写入(新增 + 更新)条数。
+
+    results 单项取自 pricebot 的 last_coupon_result / done.coupon_results,识别字段:
+    coupon_id(必需)/ status(必需)/ name / vendor / reason /(display_count)。
+    (device, coupon_id, 今天) 唯一:重复上报同张券走更新(status 以最后一次为准)。
+    """
+    today = today_cn()
+    written = 0
+    seen: set[str] = set()  # 同批去重防御:autoflush=False 下同 coupon_id 重复会两次 add → 撞唯一约束回滚整批
+    for r in results:
+        coupon_id = r.get("coupon_id")
+        status = r.get("status")
+        if not coupon_id or not status or coupon_id in seen:
+            continue  # 脏数据 / 同批重复跳过
+        seen.add(coupon_id)
+        count = r.get("display_count")
+        if count is None:
+            count = r.get("claimed_count")
+        row = db.execute(
+            select(CouponClaimRecord).where(
+                CouponClaimRecord.device_id == device_id,
+                CouponClaimRecord.coupon_id == coupon_id,
+                CouponClaimRecord.claim_date == today,
+            )
+        ).scalar_one_or_none()
+        if row is not None:
+            row.status = status
+            row.reason = r.get("reason")
+            if user_id is not None:
+                row.user_id = user_id
+            if count is not None:
+                row.claimed_count = count
+            row.extra = r
+        else:
+            db.add(CouponClaimRecord(
+                device_id=device_id, user_id=user_id,
+                coupon_id=coupon_id, claim_date=today,
+                status=status, vendor=r.get("vendor"), coupon_name=r.get("name"),
+                claimed_count=count, trace_id=trace_id, reason=r.get("reason"),
+                extra=r,
+            ))
+        written += 1
+    if written == 0:
+        return 0
+    try:
+        db.commit()
+    except IntegrityError:
+        db.rollback()
+        logger.warning(
+            "coupon_claim 并发幂等冲突 device=%s trace=%s,回滚", device_id, trace_id
+        )
+        return 0
+    return written

app/schemas/coupon_state.py

@@ -0,0 +1,21 @@
+"""领券今日状态端点的收发模型。"""
+from __future__ import annotations
+
+from pydantic import BaseModel
+
+
+class CouponPromptDismissIn(BaseModel):
+    """客户端拒绝/关闭领券引导窗的通知体。
+
+    server 据此记一条今日 engagement(dismissed)→ 今天这台设备不再弹引导窗。
+    MVP 不鉴权,按 device_id 判断;user_id 登录态带上就一并记(资产),可空。
+    """
+
+    device_id: str
+    user_id: int | None = None
+
+
+class CouponPromptShouldShowOut(BaseModel):
+    """切到外卖 App 时是否还应弹领券引导窗。今天已 engage(领或拒)过 → false。"""
+
+    should_show: bool