shaguabijia-app-server/app/models/coupon_state.py

"""领券今日状态(弹窗频控 + 领券记录)两张表。

- `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}>"
        )