From c48a958110d47cd33a1b70377f3d28abf35265b7 Mon Sep 17 00:00:00 2001 From: marco Date: Mon, 8 Jun 2026 02:53:51 +0800 Subject: [PATCH] =?UTF-8?q?feat(coupon):=20=E9=A2=86=E5=88=B8=E4=BB=8A?= =?UTF-8?q?=E6=97=A5=E7=8A=B6=E6=80=81=E5=90=8E=E7=AB=AF(=E5=BC=B9?= =?UTF-8?q?=E7=AA=97=E9=A2=91=E6=8E=A7=20+=20=E9=A2=86=E5=88=B8=E8=AE=B0?= =?UTF-8?q?=E5=BD=95)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 两张表 + 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) --- alembic/versions/coupon_state_tables.py | 77 ++++++++++++++ app/api/v1/coupon.py | 108 +++++++++++++++++++- app/models/__init__.py | 4 + app/models/coupon_state.py | 130 ++++++++++++++++++++++++ app/repositories/coupon_state.py | 129 +++++++++++++++++++++++ app/schemas/coupon_state.py | 21 ++++ 6 files changed, 468 insertions(+), 1 deletion(-) create mode 100644 alembic/versions/coupon_state_tables.py create mode 100644 app/models/coupon_state.py create mode 100644 app/repositories/coupon_state.py create mode 100644 app/schemas/coupon_state.py diff --git a/alembic/versions/coupon_state_tables.py b/alembic/versions/coupon_state_tables.py new file mode 100644 index 0000000..1b3a2c1 --- /dev/null +++ b/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') diff --git a/app/api/v1/coupon.py b/app/api/v1/coupon.py index 0ed84ef..4de69eb 100644 --- a/app/api/v1/coupon.py +++ b/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) + ) diff --git a/app/models/__init__.py b/app/models/__init__.py index 2f95779..3c5846c 100644 --- a/app/models/__init__.py +++ b/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 diff --git a/app/models/coupon_state.py b/app/models/coupon_state.py new file mode 100644 index 0000000..4374048 --- /dev/null +++ b/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"" + ) + + +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"" + ) diff --git a/app/repositories/coupon_state.py b/app/repositories/coupon_state.py new file mode 100644 index 0000000..d36be0a --- /dev/null +++ b/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 diff --git a/app/schemas/coupon_state.py b/app/schemas/coupon_state.py new file mode 100644 index 0000000..92043ce --- /dev/null +++ b/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