diff --git a/alembic/versions/coin_transaction_trace_id.py b/alembic/versions/coin_transaction_trace_id.py new file mode 100644 index 0000000..b914688 --- /dev/null +++ b/alembic/versions/coin_transaction_trace_id.py @@ -0,0 +1,60 @@ +"""coin_transaction.trace_id (金币记录按会话聚合比价/领券看广告金币) + +Revision ID: coin_transaction_trace_id +Revises: savings_record_trace_id +Create Date: 2026-08-07 + +比价/领券信息流发奖时把本场 trace_id 一并写入 coin_transaction;金币变动记录接口按 +trace_id 把一次比价/领券的多条广告金币聚合成一条。历史行从 ad_feed_reward_record +回填(ref_id == client_event_id)。 +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision: str = 'coin_transaction_trace_id' +down_revision: Union[str, Sequence[str], None] = 'savings_record_trace_id' +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + # SQLite 下 ADD COLUMN(可空) 与 CREATE INDEX 均原生支持,无需 batch_alter_table。 + op.add_column( + 'coin_transaction', + sa.Column('trace_id', sa.String(length=64), nullable=True), + ) + op.create_index( + op.f('ix_coin_transaction_trace_id'), + 'coin_transaction', + ['trace_id'], + unique=False, + ) + # 历史回填:从 ad_feed_reward_record 按 ref_id==client_event_id 补 trace_id。仅比价/领券两类、 + # 仅当前为空、且广告行确有 trace_id 时补(EXISTS 守护);`trace_id IS NULL` 保证重跑幂等。 + # 相关子查询 SQLite/PG 通用。两类 biz_type 在应用侧为 rewards.FEED_AD_SESSION_BIZ_TYPES, + # 此处按「迁移不可变」原则硬编码历史快照(勿改为 import 应用常量)。 + # 大表(prod PG)如需可改按 id 区间分批;此处一次性 UPDATE。 + op.execute( + """ + UPDATE coin_transaction SET trace_id = ( + SELECT r.trace_id FROM ad_feed_reward_record r + WHERE r.client_event_id = coin_transaction.ref_id) + WHERE biz_type IN ('feed_ad_reward_comparison', 'feed_ad_reward_coupon') + AND trace_id IS NULL + AND EXISTS (SELECT 1 FROM ad_feed_reward_record r2 + WHERE r2.client_event_id = coin_transaction.ref_id + AND r2.trace_id IS NOT NULL) + """ + ) + + +def downgrade() -> None: + op.drop_index( + op.f('ix_coin_transaction_trace_id'), + table_name='coin_transaction', + ) + op.drop_column('coin_transaction', 'trace_id') diff --git a/alembic/versions/savings_record_trace_id.py b/alembic/versions/savings_record_trace_id.py new file mode 100644 index 0000000..8ff2fd8 --- /dev/null +++ b/alembic/versions/savings_record_trace_id.py @@ -0,0 +1,34 @@ +"""savings_record.trace_id (下单归因到的比价 trace_id) + +Revision ID: savings_record_trace_id +Revises: comparison_updated_at +Create Date: 2026-08-07 00:00:00.000000 + +「已下单」从店级改单次级:下单上报带上本次比价的 trace_id,落这一列,读取时按 +trace_id 精确对齐 comparison_record.trace_id —— 同一家店比价多次,只有真正下单的 +那一条标「已下单」。可空:demo 行 / 老客户端 / 历史订单没有 trace_id(→ 不进任何 +记录的「已下单」,不做回填)。见 repositories.comparison._ordered_trace_id_select。 +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision: str = 'savings_record_trace_id' +down_revision: Union[str, Sequence[str], None] = 'comparison_updated_at' +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + with op.batch_alter_table('savings_record', schema=None) as batch_op: + batch_op.add_column(sa.Column('trace_id', sa.String(length=64), nullable=True)) + batch_op.create_index(batch_op.f('ix_savings_record_trace_id'), ['trace_id'], unique=False) + + +def downgrade() -> None: + with op.batch_alter_table('savings_record', schema=None) as batch_op: + batch_op.drop_index(batch_op.f('ix_savings_record_trace_id')) + batch_op.drop_column('trace_id') diff --git a/app/admin/repositories/queries.py b/app/admin/repositories/queries.py index 28d10ed..ae455ba 100644 --- a/app/admin/repositories/queries.py +++ b/app/admin/repositories/queries.py @@ -76,30 +76,30 @@ def _device_marketing_name(model: str | None) -> str | None: def _attach_comparison_order_status(db: Session, items: list[ComparisonRecord]) -> None: - """按 C 端既有口径给比价记录批量补充是否真实下单。""" + """按 C 端既有口径给比价记录批量补充是否真实下单(按 trace_id 精确对齐真实下单上报)。""" user_ids = {item.user_id for item in items if item.user_id is not None} - shop_names = {item.store_name for item in items if item.store_name} + trace_ids = {item.trace_id for item in items if item.trace_id} ordered_pairs: set[tuple[int, str]] = set() - if user_ids and shop_names: + if user_ids and trace_ids: rows = db.execute( - select(SavingsRecord.user_id, SavingsRecord.shop_name) + select(SavingsRecord.user_id, SavingsRecord.trace_id) .where( SavingsRecord.user_id.in_(user_ids), SavingsRecord.source == "compare", - SavingsRecord.shop_name.in_(shop_names), + SavingsRecord.trace_id.in_(trace_ids), ) .distinct() ).all() ordered_pairs = { - (row.user_id, row.shop_name) + (row.user_id, row.trace_id) for row in rows - if row.shop_name is not None + if row.trace_id is not None } for item in items: item.ordered = bool( item.user_id is not None - and item.store_name - and (item.user_id, item.store_name) in ordered_pairs + and item.trace_id + and (item.user_id, item.trace_id) in ordered_pairs ) diff --git a/app/admin/repositories/stats.py b/app/admin/repositories/stats.py index 94b8beb..d2fbb3f 100644 --- a/app/admin/repositories/stats.py +++ b/app/admin/repositories/stats.py @@ -376,15 +376,14 @@ def dashboard_overview( .where( SavingsRecord.user_id == ComparisonRecord.user_id, SavingsRecord.source == "compare", - SavingsRecord.shop_name.is_not(None), - SavingsRecord.shop_name == ComparisonRecord.store_name, + SavingsRecord.trace_id.is_not(None), + SavingsRecord.trace_id == ComparisonRecord.trace_id, ) .exists() ) period_ordered_count = _count( ComparisonRecord, *period_comparison_conds, - ComparisonRecord.store_name.is_not(None), ordered_exists, ) diff --git a/app/core/rewards.py b/app/core/rewards.py index 842f0ae..f0bc598 100644 --- a/app/core/rewards.py +++ b/app/core/rewards.py @@ -26,6 +26,16 @@ SIGNIN_REWARDS: tuple[int, ...] = ( SIGNIN_CYCLE_LEN: int = len(SIGNIN_REWARDS) +# ===== 信息流广告「按会话聚合」的 biz_type ===== +# 比价 / 领券等候期看的信息流广告,每条各写一条 coin_transaction;这两类在「金币变动记录」 +# 里按 trace_id 聚合成一条展示(见 repositories/wallet.list_coin_transactions)。其余类型 +# (reward_video / guide_video / 通用 feed_ad_reward / signin / task_* 等)不聚合。 +FEED_AD_SESSION_BIZ_TYPES: tuple[str, str] = ( + "feed_ad_reward_comparison", + "feed_ad_reward_coupon", +) + + def signin_reward(cycle_day: int) -> int: """cycle_day 取值 1..SIGNIN_CYCLE_LEN。""" return SIGNIN_REWARDS[cycle_day - 1] diff --git a/app/models/ad_feed_reward.py b/app/models/ad_feed_reward.py index aa662d4..815e39e 100644 --- a/app/models/ad_feed_reward.py +++ b/app/models/ad_feed_reward.py @@ -36,8 +36,8 @@ class AdFeedRewardRecord(Base): # 点位场景:comparison(比价等待) / coupon(领券) / welfare(福利页)。比价与领券共用同一信息流 # 代码位,slot_id/our_code_id 分不出,只能客户端各调用点显式打标;NULL=历史/未升级客户端=未分类。 feed_scene: Mapped[str | None] = mapped_column(String(16), nullable=True) - # 本次比价 trace_id(仅 comparison 场景由客户端带上):把这场广告金币归属到对应比价记录, - # 比价记录页按 trace_id 聚合本场实发金币显示「比价赚 N 金币」。领券/福利/旧客户端 = NULL。 + # 本次会话 trace_id:比价一直带;领券自 2026-07-15(客户端 a98cab8)起也带;福利/旧客户端 = NULL。 + # 用途:比价记录页按 trace_id 聚合「比价赚 N 金币」;金币记录列表把一次比价/领券的多条广告金币聚合成一条。 trace_id: Mapped[str | None] = mapped_column(String(64), index=True, nullable=True) # 来源(广告收益报表用):我们的应用环境 prod/test + 我们配置的代码位 104xxx,由客户端 feed-reward 上报带上。旧数据为 NULL。 app_env: Mapped[str | None] = mapped_column(String(16), nullable=True) diff --git a/app/models/savings.py b/app/models/savings.py index 0a85c3f..656f13a 100644 --- a/app/models/savings.py +++ b/app/models/savings.py @@ -56,6 +56,9 @@ class SavingsRecord(Base): source_deeplink: Mapped[str | None] = mapped_column(String(512), nullable=True) # 客户端幂等键(UUID);demo 行为 NULL client_event_id: Mapped[str | None] = mapped_column(String(64), nullable=True) + # 本次下单归因到的那次比价的 trace_id(客户端归因时从比价会话缓存带来;demo/老客户端/历史行为 NULL)。 + # 「已下单」按它精确对齐 comparison_record.trace_id —— 同店多次比价只标真正下单的那一条。 + trace_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True) device_id: Mapped[str | None] = mapped_column(String(128), nullable=True) created_at: Mapped[datetime] = mapped_column( diff --git a/app/models/wallet.py b/app/models/wallet.py index dcdff4a..f5c22b1 100644 --- a/app/models/wallet.py +++ b/app/models/wallet.py @@ -84,6 +84,9 @@ class CoinTransaction(Base): # 关联业务 id(签到日期、任务 key 等),可空 ref_id: Mapped[str | None] = mapped_column(String(64), nullable=True) remark: Mapped[str | None] = mapped_column(String(128), nullable=True) + # 会话键:仅比价/领券信息流发奖(feed_ad_reward_comparison/coupon)时写入本场 trace_id, + # 金币记录列表据此把一次比价/领券的多条广告金币聚合成一条。其余类型 = NULL。 + trace_id: Mapped[str | None] = mapped_column(String(64), index=True, nullable=True) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), server_default=func.now(), index=True, nullable=False diff --git a/app/repositories/ad_feed_reward.py b/app/repositories/ad_feed_reward.py index a90c0c7..1649b89 100644 --- a/app/repositories/ad_feed_reward.py +++ b/app/repositories/ad_feed_reward.py @@ -219,7 +219,7 @@ def grant_feed_reward( crud_wallet.grant_coins( db, user_id, coin, biz_type=reward_biz, ref_id=client_event_id, - remark=reward_remark, + remark=reward_remark, trace_id=trace_id, ) rec = AdFeedRewardRecord( client_event_id=client_event_id, diff --git a/app/repositories/comparison.py b/app/repositories/comparison.py index 2703f92..01d5cfa 100644 --- a/app/repositories/comparison.py +++ b/app/repositories/comparison.py @@ -737,18 +737,18 @@ def harvest_abort( return rec -def _ordered_shop_name_select(user_id: int): - """该用户「真实下单」(source='compare')覆盖到的店名 select,给「已下单」筛选当子查询。 +def _ordered_trace_id_select(user_id: int): + """该用户「真实下单」(source='compare')覆盖到的比价 trace_id 的 select,给「已下单」筛选当子查询。 - 口径与 [_ordered_shop_names] 完全一致,只是时机不同:那边是**拿到本页之后**按 candidates + 口径与 [_ordered_trace_ids] 完全一致,只是时机不同:那边是**拿到本页之后**按 candidates 反查打标;这边是**分页之前**就要过滤,拿不到 candidates,只能整段下推成子查询。 - 没有先捞成集合再展开 IN (...) 字面量 —— 重度用户下单过的店名可能上千,展开会撞 SQLite + 没有先捞成集合再展开 IN (...) 字面量 —— 重度用户下单过的 trace_id 可能上千,展开会撞 SQLite 的绑定变量上限,而且又变回了那个「随下单量线性变慢」的老写法。 """ - return select(SavingsRecord.shop_name).where( + return select(SavingsRecord.trace_id).where( SavingsRecord.user_id == user_id, SavingsRecord.source == "compare", - SavingsRecord.shop_name.is_not(None), + SavingsRecord.trace_id.is_not(None), ) @@ -760,27 +760,27 @@ def _like_escape(kw: str) -> str: return kw.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") -def _ordered_shop_names(db: Session, user_id: int, candidates: set[str]) -> set[str]: - """[candidates] 里哪些店名被该用户「真实下单」(source='compare')覆盖过,用来打「已下单」。 +def _ordered_trace_ids(db: Session, user_id: int, candidates: set[str]) -> set[str]: + """[candidates] 里哪些 trace_id 被该用户「真实下单」(source='compare')覆盖过,用来打「已下单」。 - 只认 compare(归因命中后真实上报),demo 演示数据不算。下单上报不带 trace_id, - 只能按店名对齐——两边店名同源(都来自比价意图识别阶段的门店名 query),精确相等即视为同店。 - 语义=店级:同一家店比价过多次,这些记录会一并标「已下单」。 + 只认 compare(归因命中后真实上报),demo 演示数据不算。下单上报带上本次比价的 trace_id → + 按 trace_id 精确对齐 comparison_record.trace_id:同一家店比价多次,只有真正下单的那一条会被 + 标「已下单」。没带 trace_id 的下单(历史 / 老客户端)对齐不上任何记录 → 不进「已下单」。 - ⚠️ 只查**本页出现过的店名**(candidates ≤ limit 条),不再把该用户全部下单店名捞回内存: - 老写法随下单量线性增长,重度用户几千行全读一遍只为跟 50 条记录取交集。空集合直接返回 + ⚠️ 只查**本页出现过的 trace_id**(candidates ≤ limit 条),不再把该用户全部下单 trace_id 捞回 + 内存:老写法随下单量线性增长,重度用户几千行全读一遍只为跟 50 条记录取交集。空集合直接返回 (避免 IN () 非法)。 """ if not candidates: return set() rows = db.execute( - select(SavingsRecord.shop_name).where( + select(SavingsRecord.trace_id).where( SavingsRecord.user_id == user_id, SavingsRecord.source == "compare", - SavingsRecord.shop_name.in_(candidates), + SavingsRecord.trace_id.in_(candidates), ).distinct() ).scalars().all() - return {s for s in rows if s} + return {t for t in rows if t} def _ad_coins_by_trace(db: Session, user_id: int, trace_ids: list[str]) -> dict[str, int]: @@ -832,7 +832,7 @@ def list_records( ordered: bool | None = None, keyword: str | None = None, ) -> tuple[list[ComparisonRecord], int | None]: - """比价记录分页(按创建时间倒序、id 兜底,游标式)。附「已下单」店级标记 + 「看广告赚的金币」(瞬态,不写库)。""" + """比价记录分页(按创建时间倒序、id 兜底,游标式)。附「已下单」标记(按 trace_id 精确对齐真实下单)+ 「看广告赚的金币」(瞬态,不写库)。""" stmt = ( select(ComparisonRecord) .where(ComparisonRecord.user_id == user_id) @@ -844,7 +844,7 @@ def list_records( # 分页之后一页里可能一条都不命中,列表看着就是空的/卡住的,得翻很多页才蹦出一条。 if ordered: stmt = stmt.where( - ComparisonRecord.store_name.in_(_ordered_shop_name_select(user_id)) + ComparisonRecord.trace_id.in_(_ordered_trace_id_select(user_id)) ) kw = (keyword or "").strip() if kw: @@ -865,16 +865,16 @@ def list_records( items = list(db.execute(stmt).scalars().all()) next_cursor = items[-1].id if len(items) == limit else None - # 「已下单」标记:本页记录的 store_name 若落在该用户真实下单的店名集合里即 True。 + # 「已下单」标记:本页记录的 trace_id 若落在该用户真实下单(带 trace_id)的集合里即 True。 # ordered / ad_coins_earned 均非 ORM 列,仅挂实例上供 ComparisonRecordOut(from_attributes) 读出,不持久化。 - page_shops = {it.store_name for it in items if it.store_name} - # ordered=True 时上面已按同一口径(_ordered_shop_name_select)筛过,本页必然全是已下单, - # 省掉这次反查;其余情况照旧按本页店名反查 savings。 - ordered_shops = page_shops if ordered else _ordered_shop_names(db, user_id, page_shops) + page_traces = {it.trace_id for it in items} + # ordered=True 时上面已按同一口径(_ordered_trace_id_select)筛过,本页必然全是已下单, + # 省掉这次反查;其余情况按本页 trace_id 反查 savings。 + ordered_traces = page_traces if ordered else _ordered_trace_ids(db, user_id, page_traces) # 「本次比价看广告赚的金币」:按本页 trace_id 一次性聚合(同 ordered 范式)。 ad_coins = _ad_coins_by_trace(db, user_id, [it.trace_id for it in items]) for it in items: - it.ordered = bool(it.store_name and it.store_name in ordered_shops) + it.ordered = it.trace_id in ordered_traces it.ad_coins_earned = ad_coins.get(it.trace_id, 0) return items, next_cursor diff --git a/app/repositories/savings.py b/app/repositories/savings.py index 5c8574d..1d3f53d 100644 --- a/app/repositories/savings.py +++ b/app/repositories/savings.py @@ -177,6 +177,7 @@ def create_from_report( source_platform_name=req.source_platform_name, source_deeplink=req.source_deeplink, client_event_id=req.client_event_id, + trace_id=req.trace_id, device_id=req.device_id, source="compare", # created_at 显式存 naive 北京 wall-clock(与 demo 行、聚合 _local_date 的 naive 分支一致)。 diff --git a/app/repositories/wallet.py b/app/repositories/wallet.py index fbaebbb..2213f1c 100644 --- a/app/repositories/wallet.py +++ b/app/repositories/wallet.py @@ -10,9 +10,10 @@ import logging import re import unicodedata import uuid +from dataclasses import dataclass from datetime import datetime, timedelta, timezone -from sqlalchemy import func, select, update +from sqlalchemy import String, and_, case, cast, func, literal, select, update from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session @@ -170,6 +171,7 @@ def grant_coins( biz_type: str, ref_id: str | None = None, remark: str | None = None, + trace_id: str | None = None, ) -> tuple[CoinAccount, CoinTransaction]: """金币变动入口(正数入账 / 负数出账)。更新余额 + 写流水,不 commit。 @@ -189,6 +191,7 @@ def grant_coins( biz_type=biz_type, ref_id=ref_id, remark=remark, + trace_id=trace_id, created_at=datetime.now(rewards.CN_TZ).replace(tzinfo=None), # 存北京 wall-clock(客户端原样切片显示) ) db.add(txn) @@ -257,24 +260,94 @@ def grant_invite_cash( return acc, txn +@dataclass(frozen=True) +class CoinLedgerRow: + """`list_coin_transactions` 的返回行。广告类按 trace_id 聚合后的展示行, + **非 ORM 对象**(避免把聚合后的 amount 误写回底层流水)。""" + id: int + amount: int + balance_after: int + biz_type: str + ref_id: str | None + remark: str | None + created_at: datetime + merged_count: int + + def list_coin_transactions( db: Session, user_id: int, *, limit: int = 20, cursor: int | None = None, -) -> tuple[list[CoinTransaction], int | None]: - """金币流水分页(按 id 倒序,游标式)。 +) -> tuple[list[CoinLedgerRow], int | None]: + """金币流水分页(游标式,按 id 倒序)。 - cursor 为上一页最后一条的 id;返回 (本页列表, next_cursor)。 - next_cursor 为 None 表示没有下一页。 + 比价/领券两类信息流广告(rewards.FEED_AD_SESSION_BIZ_TYPES)且带 trace_id 的行, + 按 trace_id 聚合成一条(一次比价/领券 = 一行):金额合计、代表行取组内最新一条 + (MAX(id))的余额/时间、merged_count=组内条数。其余每条一行。 + + 分组必须在该用户**全量**行上算真实 rep_id 后再按 rep_id 过滤——**不可**把 CTE 输入 + 裁成 id **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** App「金币变动记录」里，一次比价 / 一次领券连续看广告获得的多条金币，按 `trace_id` 聚合成一条展示。 + +**Architecture:** 纯后端。给 `coin_transaction` 加 `trace_id` 列并在发奖时写入；`GET /api/v1/wallet/coin-transactions` 改为按 `trace_id` 分组的游标分页（比价/领券两类 feed 广告按会话合并，其余每条一行）；历史行从 `ad_feed_reward_record` 回填。领券自 2026-07-15 起结算已带 trace_id，无需改 Android。仅动 App 用户接口，admin 审计接口不变。 + +**Tech Stack:** FastAPI · SQLAlchemy 2.0 · Alembic · Pydantic v2 · pytest（SQLite）。 + +**Spec:** `docs/superpowers/specs/2026-08-07-coin-ledger-aggregate-ad-rewards-design.md` + +**约定：** dev / pytest 均 **SQLite**，prod PostgreSQL——所有 SQL 保持 SQLite/PG 通用（不用 PG 专有语法）。金额单位为金币（非分）。 + +--- + +## File Structure + +| 文件 | 改动 | 职责 | +|---|---|---| +| `app/core/rewards.py` | 加常量 `FEED_AD_SESSION_BIZ_TYPES` | 「按会话聚合」的两类 biz_type 单一来源（查询 + 写入 + 测试共用） | +| `app/models/wallet.py` | `CoinTransaction` 加 `trace_id` 列 | 金币流水会话键 | +| `app/repositories/wallet.py` | `grant_coins` 加 `trace_id` 参；`list_coin_transactions` 改分组分页；加 `CoinLedgerRow` | 写入透传 + 聚合读取 | +| `app/repositories/ad_feed_reward.py` | `grant_feed_reward` 调 `grant_coins` 时传 `trace_id` | 把本场 trace_id 写进金币流水 | +| `app/schemas/welfare.py` | `CoinTransactionOut` 加 `merged_count` | 下发合并条数 | +| `app/models/ad_feed_reward.py` | 更正 `trace_id` 注释 | 文档（领券自 2026-07-15 也带） | +| `alembic/versions/coin_transaction_trace_id.py` | 新建迁移 | 加列 + 索引 + 回填历史 | +| `tests/test_welfare.py` | 追加测试 | 覆盖写入 / 回填 / 发奖透传 / 聚合 / 分页防残组 | + +--- + +## Task 1: 常量 + `CoinTransaction.trace_id` 列 + `grant_coins` 透传 + +**Files:** +- Modify: `app/core/rewards.py`（加常量） +- Modify: `app/models/wallet.py:86`（`remark` 后加 `trace_id` 列） +- Modify: `app/repositories/wallet.py:165-196`（`grant_coins` 加参数 + 写入） +- Test: `tests/test_welfare.py`（追加） + +- [ ] **Step 1: 写失败测试** + +追加到 `tests/test_welfare.py` 末尾： + +```python +def test_grant_coins_persists_trace_id(client) -> None: + """grant_coins 传 trace_id 落库;不传则为 None。""" + phone = "13800002001" + _login(client, phone) + with SessionLocal() as db: + user = get_user_by_phone(db, phone) + assert user is not None + _, txn1 = crud_wallet.grant_coins( + db, user.id, 5, biz_type="feed_ad_reward_comparison", + ref_id="evt1", remark="比价奖励", trace_id="trace-A", + ) + _, txn2 = crud_wallet.grant_coins( + db, user.id, 30, biz_type="signin", remark="每日签到奖励", + ) + db.commit() + assert txn1.trace_id == "trace-A" + assert txn2.trace_id is None +``` + +- [ ] **Step 2: 跑测试确认失败** + +Run: `pytest tests/test_welfare.py::test_grant_coins_persists_trace_id -q` +Expected: FAIL —— `TypeError: grant_coins() got an unexpected keyword argument 'trace_id'` + +- [ ] **Step 3: 加常量** + +在 `app/core/rewards.py` 的签到常量段之后（约第 32 行 `SIGNIN_CYCLE_LEN` 之后）插入： + +```python +# ===== 信息流广告「按会话聚合」的 biz_type ===== +# 比价 / 领券等候期看的信息流广告,每条各写一条 coin_transaction;这两类在「金币变动记录」 +# 里按 trace_id 聚合成一条展示(见 repositories/wallet.list_coin_transactions)。其余类型 +# (reward_video / guide_video / 通用 feed_ad_reward / signin / task_* 等)不聚合。 +FEED_AD_SESSION_BIZ_TYPES: tuple[str, str] = ( + "feed_ad_reward_comparison", + "feed_ad_reward_coupon", +) +``` + +- [ ] **Step 4: 加模型列** + +`app/models/wallet.py`，把 `remark` 行（第 86 行）后面补一列： + +```python + remark: Mapped[str | None] = mapped_column(String(128), nullable=True) + # 会话键:仅比价/领券信息流发奖(feed_ad_reward_comparison/coupon)时写入本场 trace_id, + # 金币记录列表据此把一次比价/领券的多条广告金币聚合成一条。其余类型 = NULL。 + trace_id: Mapped[str | None] = mapped_column(String(64), index=True, nullable=True) +``` + +（`String` 在该文件已导入。`index=True` 会在测试 `create_all` 时自动建 `ix_coin_transaction_trace_id`。） + +- [ ] **Step 5: `grant_coins` 加参数 + 写入** + +`app/repositories/wallet.py` 的 `grant_coins`：签名加 `trace_id`，构造 `CoinTransaction` 时带上。 + +签名（第 165-173 行）改为： + +```python +def grant_coins( + db: Session, + user_id: int, + amount: int, + *, + biz_type: str, + ref_id: str | None = None, + remark: str | None = None, + trace_id: str | None = None, +) -> tuple[CoinAccount, CoinTransaction]: +``` + +`CoinTransaction(...)` 构造（第 185-193 行）改为： + +```python + txn = CoinTransaction( + user_id=user_id, + amount=amount, + balance_after=acc.coin_balance, + biz_type=biz_type, + ref_id=ref_id, + remark=remark, + trace_id=trace_id, + created_at=datetime.now(rewards.CN_TZ).replace(tzinfo=None), # 存北京 wall-clock(客户端原样切片显示) + ) +``` + +- [ ] **Step 6: 跑测试确认通过** + +Run: `pytest tests/test_welfare.py::test_grant_coins_persists_trace_id -q` +Expected: PASS + +- [ ] **Step 7: 回归 + lint** + +Run: `pytest tests/test_welfare.py -q && ruff check app/core/rewards.py app/models/wallet.py app/repositories/wallet.py tests/test_welfare.py` +Expected: 原有用例仍 PASS，无新增 lint。 + +- [ ] **Step 8: 提交** + +```bash +git add app/core/rewards.py app/models/wallet.py app/repositories/wallet.py tests/test_welfare.py +git commit -m "feat(wallet): coin_transaction 增 trace_id 列 + grant_coins 透传 + +Co-Authored-By: Claude Opus 4.8 (1M context) " +``` + +--- + +## Task 2: 迁移（加列 + 索引 + 回填历史） + +**Files:** +- Create: `alembic/versions/coin_transaction_trace_id.py` +- Test: `tests/test_welfare.py`（追加回填语义测试） + +- [ ] **Step 1: 写失败测试（回填语义）** + +在 `tests/test_welfare.py` 顶部 import 区补两行（若尚无）： + +```python +from sqlalchemy import text +from app.models.ad_feed_reward import AdFeedRewardRecord +``` + +追加测试： + +```python +def test_backfill_coin_trace_id_from_ad_record(client) -> None: + """回填:coin_transaction(trace_id 空)按 ref_id==client_event_id 从 ad_feed_reward_record 补 trace_id; + 只补比价/领券两类,无关类型与无匹配的不动。SQL 与迁移 coin_transaction_trace_id 保持同步。""" + phone = "13800002002" + _login(client, phone) + with SessionLocal() as db: + user = get_user_by_phone(db, phone) + assert user is not None + # 广告表:两条带 trace_id 的记录(模拟客户端已上报) + db.add(AdFeedRewardRecord( + client_event_id="evt-cmp", user_id=user.id, reward_date="2026-08-07", + duration_seconds=10, unit_count=1, ecpm_raw="1000", + feed_scene="comparison", trace_id="trace-CMP", coin=5, status="granted", + )) + db.add(AdFeedRewardRecord( + client_event_id="evt-cpn", user_id=user.id, reward_date="2026-08-07", + duration_seconds=10, unit_count=1, ecpm_raw="1000", + feed_scene="coupon", trace_id="trace-CPN", coin=7, status="granted", + )) + db.commit() + # 老金币流水:trace_id 全空(模拟改动前),ref_id 指向上面的广告事件 + _, c1 = crud_wallet.grant_coins(db, user.id, 5, biz_type="feed_ad_reward_comparison", ref_id="evt-cmp", remark="比价奖励") + _, c2 = crud_wallet.grant_coins(db, user.id, 7, biz_type="feed_ad_reward_coupon", ref_id="evt-cpn", remark="领券奖励") + _, c3 = crud_wallet.grant_coins(db, user.id, 30, biz_type="signin", remark="每日签到奖励") + db.commit() + assert c1.trace_id is None and c2.trace_id is None + + # 执行与迁移 upgrade() 等价的回填 SQL(务必与迁移保持一致) + db.execute(text( + """ + UPDATE coin_transaction SET trace_id = ( + SELECT r.trace_id FROM ad_feed_reward_record r + WHERE r.client_event_id = coin_transaction.ref_id) + WHERE biz_type IN ('feed_ad_reward_comparison', 'feed_ad_reward_coupon') + AND trace_id IS NULL + AND EXISTS (SELECT 1 FROM ad_feed_reward_record r2 + WHERE r2.client_event_id = coin_transaction.ref_id + AND r2.trace_id IS NOT NULL) + """ + )) + db.commit() + db.refresh(c1); db.refresh(c2); db.refresh(c3) + assert c1.trace_id == "trace-CMP" # 比价补上 + assert c2.trace_id == "trace-CPN" # 领券补上 + assert c3.trace_id is None # 签到不动 +``` + +- [ ] **Step 2: 跑测试确认失败** + +Run: `pytest tests/test_welfare.py::test_backfill_coin_trace_id_from_ad_record -q` +Expected: FAIL —— `sqlite3.OperationalError: no such column: trace_id`（Task 1 已加列则此步应改为直接 PASS；若已 PASS 说明 create_all 已含列，跳到 Step 4 建迁移）。 + +> 说明：本测试验证的是回填 SQL 的 JOIN 语义，不经 Alembic。Task 1 完成后列已在测试库存在，测试可能直接 PASS——这是预期的（回填 SQL 本身是正确逻辑）。真正要新建的产物是迁移文件（Step 4），供 dev/prod 使用。 + +- [ ] **Step 3: 跑测试确认通过** + +Run: `pytest tests/test_welfare.py::test_backfill_coin_trace_id_from_ad_record -q` +Expected: PASS + +- [ ] **Step 4: 建迁移文件** + +Create `alembic/versions/coin_transaction_trace_id.py`： + +```python +"""coin_transaction.trace_id (金币记录按会话聚合比价/领券看广告金币) + +Revision ID: coin_transaction_trace_id +Revises: comparison_updated_at +Create Date: 2026-08-07 + +比价/领券信息流发奖时把本场 trace_id 一并写入 coin_transaction;金币变动记录接口按 +trace_id 把一次比价/领券的多条广告金币聚合成一条。历史行从 ad_feed_reward_record +回填(ref_id == client_event_id)。 +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision: str = 'coin_transaction_trace_id' +down_revision: Union[str, Sequence[str], None] = 'comparison_updated_at' +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + # SQLite 下 ADD COLUMN(可空) 与 CREATE INDEX 均原生支持,无需 batch_alter_table。 + op.add_column( + 'coin_transaction', + sa.Column('trace_id', sa.String(length=64), nullable=True), + ) + op.create_index( + op.f('ix_coin_transaction_trace_id'), + 'coin_transaction', + ['trace_id'], + unique=False, + ) + # 历史回填:从 ad_feed_reward_record 按 ref_id==client_event_id 补 trace_id。仅比价/领券两类、 + # 仅当前为空、且广告行确有 trace_id 时补(EXISTS 守护);`trace_id IS NULL` 保证重跑幂等。 + # 相关子查询 SQLite/PG 通用。两类 biz_type 在应用侧为 rewards.FEED_AD_SESSION_BIZ_TYPES, + # 此处按「迁移不可变」原则硬编码历史快照(勿改为 import 应用常量)。 + # 大表(prod PG)如需可改按 id 区间分批;此处一次性 UPDATE。 + op.execute( + """ + UPDATE coin_transaction SET trace_id = ( + SELECT r.trace_id FROM ad_feed_reward_record r + WHERE r.client_event_id = coin_transaction.ref_id) + WHERE biz_type IN ('feed_ad_reward_comparison', 'feed_ad_reward_coupon') + AND trace_id IS NULL + AND EXISTS (SELECT 1 FROM ad_feed_reward_record r2 + WHERE r2.client_event_id = coin_transaction.ref_id + AND r2.trace_id IS NOT NULL) + """ + ) + + +def downgrade() -> None: + op.drop_index( + op.f('ix_coin_transaction_trace_id'), + table_name='coin_transaction', + ) + op.drop_column('coin_transaction', 'trace_id') +``` + +- [ ] **Step 5: 验证迁移可正反向应用（用一次性 scratch SQLite，不碰 dev 库）** + +Run: +```bash +DATABASE_URL="sqlite:///./data/_migcheck.db" python -m alembic upgrade head && \ +DATABASE_URL="sqlite:///./data/_migcheck.db" python -m alembic downgrade -1 && \ +DATABASE_URL="sqlite:///./data/_migcheck.db" python -m alembic upgrade head && \ +rm -f ./data/_migcheck.db +``` +Expected: 三条 alembic 命令均无报错，最终 head 落在 `coin_transaction_trace_id`；`rm` 清掉临时库。 + +- [ ] **Step 6: lint** + +Run: `ruff check alembic/versions/coin_transaction_trace_id.py tests/test_welfare.py` +Expected: 无新增 lint。 + +- [ ] **Step 7: 提交** + +```bash +git add alembic/versions/coin_transaction_trace_id.py tests/test_welfare.py +git commit -m "feat(wallet): 迁移加 coin_transaction.trace_id + 索引 + 回填历史 + +Co-Authored-By: Claude Opus 4.8 (1M context) " +``` + +--- + +## Task 3: `grant_feed_reward` 把 trace_id 写进金币流水 + +**Files:** +- Modify: `app/repositories/ad_feed_reward.py:219-223`（`grant_coins` 调用加 `trace_id`） +- Test: `tests/test_welfare.py`（追加） + +- [ ] **Step 1: 写失败测试** + +在 `tests/test_welfare.py` 顶部 import 区补（若尚无）： + +```python +from sqlalchemy import select +from app.models.wallet import CoinTransaction +from app.repositories import ad_feed_reward as crud_feed +``` + +追加测试： + +```python +def test_grant_feed_reward_sets_coin_trace_id(client) -> None: + """grant_feed_reward(comparison) 把 trace_id 透传给 grant_coins,coin_transaction 带上本场 trace_id。""" + phone = "13800002003" + _login(client, phone) + with SessionLocal() as db: + user = get_user_by_phone(db, phone) + assert user is not None + rec = crud_feed.grant_feed_reward( + db, user.id, + client_event_id="evt-fr-1", ecpm="1000", duration_seconds=10, + feed_scene="comparison", trace_id="trace-FR", display_coin=5, + ) + assert rec.status == "granted", rec.status + txn = db.execute( + select(CoinTransaction).where( + CoinTransaction.user_id == user.id, + CoinTransaction.ref_id == "evt-fr-1", + ) + ).scalar_one() + assert txn.biz_type == "feed_ad_reward_comparison" + assert txn.trace_id == "trace-FR" +``` + +- [ ] **Step 2: 跑测试确认失败** + +Run: `pytest tests/test_welfare.py::test_grant_feed_reward_sets_coin_trace_id -q` +Expected: FAIL —— `AssertionError: assert None == 'trace-FR'`（`grant_coins` 尚未收到 trace_id）。 + +- [ ] **Step 3: 传 trace_id** + +`app/repositories/ad_feed_reward.py` 第 219-223 行 `grant_coins` 调用改为： + +```python + crud_wallet.grant_coins( + db, user_id, coin, + biz_type=reward_biz, ref_id=client_event_id, + remark=reward_remark, trace_id=trace_id, + ) +``` + +- [ ] **Step 4: 跑测试确认通过** + +Run: `pytest tests/test_welfare.py::test_grant_feed_reward_sets_coin_trace_id -q` +Expected: PASS + +- [ ] **Step 5: lint** + +Run: `ruff check app/repositories/ad_feed_reward.py tests/test_welfare.py` +Expected: 无新增 lint。 + +- [ ] **Step 6: 提交** + +```bash +git add app/repositories/ad_feed_reward.py tests/test_welfare.py +git commit -m "feat(ad-feed): 发奖把本场 trace_id 写入金币流水 + +Co-Authored-By: Claude Opus 4.8 (1M context) " +``` + +--- + +## Task 4: 聚合查询 —— `list_coin_transactions` 分组 + `CoinLedgerRow` + `merged_count` + +**Files:** +- Modify: `app/repositories/wallet.py`（imports + 加 `CoinLedgerRow` + 重写 `list_coin_transactions`） +- Modify: `app/schemas/welfare.py:23-32`（`CoinTransactionOut` 加 `merged_count`） +- Test: `tests/test_welfare.py`（追加 3 个聚合测试 + 一个 `_seed_coin` 帮手） + +- [ ] **Step 1: 写失败测试** + +在 `tests/test_welfare.py` 追加帮手 + 测试： + +```python +def _seed_coin(db, user_id, amount, biz_type, *, trace_id=None, ref_id=None, remark=None): + crud_wallet.grant_coins( + db, user_id, amount, biz_type=biz_type, ref_id=ref_id, remark=remark, trace_id=trace_id + ) + + +def test_coin_transactions_aggregate_by_trace(client) -> None: + """一次比价的多条广告金币聚合成一条:金额合计、merged_count=条数、balance_after 取最后一条。""" + phone = "13800002004" + token = _login(client, phone) + with SessionLocal() as db: + user = get_user_by_phone(db, phone) + assert user is not None + _seed_coin(db, user.id, 5, "feed_ad_reward_comparison", trace_id="t1", ref_id="e1", remark="比价奖励") + _seed_coin(db, user.id, 3, "feed_ad_reward_comparison", trace_id="t1", ref_id="e2", remark="比价奖励") + _seed_coin(db, user.id, 4, "feed_ad_reward_comparison", trace_id="t1", ref_id="e3", remark="比价奖励") + db.commit() + r = client.get("/api/v1/wallet/coin-transactions", headers=_auth(token)) + assert r.status_code == 200, r.text + items = r.json()["items"] + assert len(items) == 1 + row = items[0] + assert row["biz_type"] == "feed_ad_reward_comparison" + assert row["amount"] == 12 # 5+3+4 + assert row["merged_count"] == 3 + assert row["balance_after"] == 12 # 末条到账后余额(本用户从 0 起) + + +def test_coin_transactions_distinct_traces_stay_separate(client) -> None: + """不同 trace(两次比价 / 一次领券)各成一条;不同会话不合并。""" + phone = "13800002005" + token = _login(client, phone) + with SessionLocal() as db: + user = get_user_by_phone(db, phone) + assert user is not None + _seed_coin(db, user.id, 5, "feed_ad_reward_comparison", trace_id="cmpA", ref_id="a1", remark="比价奖励") + _seed_coin(db, user.id, 6, "feed_ad_reward_comparison", trace_id="cmpB", ref_id="b1", remark="比价奖励") + _seed_coin(db, user.id, 6, "feed_ad_reward_comparison", trace_id="cmpB", ref_id="b2", remark="比价奖励") + _seed_coin(db, user.id, 7, "feed_ad_reward_coupon", trace_id="cpnC", ref_id="c1", remark="领券奖励") + db.commit() + items = client.get("/api/v1/wallet/coin-transactions", headers=_auth(token)).json()["items"] + assert len(items) == 3 + assert sorted(i["amount"] for i in items) == [5, 7, 12] + cpn = next(i for i in items if i["biz_type"] == "feed_ad_reward_coupon") + assert cpn["amount"] == 7 and cpn["merged_count"] == 1 + + +def test_coin_transactions_non_session_rows_stay_per_row(client) -> None: + """签到 / 无 trace 的通用信息流各自一行,不被聚合;夹在比价广告中间的签到不影响比价聚合。""" + phone = "13800002006" + token = _login(client, phone) + with SessionLocal() as db: + user = get_user_by_phone(db, phone) + assert user is not None + _seed_coin(db, user.id, 5, "feed_ad_reward_comparison", trace_id="t1", ref_id="e1", remark="比价奖励") + _seed_coin(db, user.id, 30, "signin", remark="每日签到奖励") # 夹在两条比价广告中间 + _seed_coin(db, user.id, 4, "feed_ad_reward_comparison", trace_id="t1", ref_id="e2", remark="比价奖励") + _seed_coin(db, user.id, 8, "feed_ad_reward", trace_id=None, ref_id="w1", remark="信息流广告奖励") + db.commit() + items = client.get("/api/v1/wallet/coin-transactions", headers=_auth(token)).json()["items"] + assert len(items) == 3 + cmp_row = next(i for i in items if i["biz_type"] == "feed_ad_reward_comparison") + assert cmp_row["amount"] == 9 and cmp_row["merged_count"] == 2 + signin_row = next(i for i in items if i["biz_type"] == "signin") + assert signin_row["amount"] == 30 and signin_row["merged_count"] == 1 + feed_row = next(i for i in items if i["biz_type"] == "feed_ad_reward") + assert feed_row["amount"] == 8 and feed_row["merged_count"] == 1 +``` + +- [ ] **Step 2: 跑测试确认失败** + +Run: `pytest tests/test_welfare.py::test_coin_transactions_aggregate_by_trace -q` +Expected: FAIL —— 未聚合，返回 3 条 / `merged_count` 字段缺失（`KeyError` 或 `len(items)==3`）。 + +- [ ] **Step 3: schema 加 `merged_count`** + +`app/schemas/welfare.py` 的 `CoinTransactionOut`（第 23-32 行）在 `created_at` 后加一行： + +```python +class CoinTransactionOut(BaseModel): + model_config = ConfigDict(from_attributes=True) + + id: int + amount: int = Field(..., description="正=入账,负=出账") + balance_after: int + biz_type: str + ref_id: str | None = None + remark: str | None = None + created_at: datetime + merged_count: int = Field(1, description="本行合并的底层流水条数(比价/领券按会话聚合;未合并=1)") +``` + +- [ ] **Step 4: 补 imports + 加 `CoinLedgerRow`** + +`app/repositories/wallet.py` 顶部：把 `from sqlalchemy import func, select, update`（第 15 行）改为： + +```python +from sqlalchemy import String, and_, case, cast, func, literal, select, update +``` + +在 stdlib import 区（约第 7-13 行）加一行： + +```python +from dataclasses import dataclass +``` + +在 `list_coin_transactions` 之前加返回行类型： + +```python +@dataclass(frozen=True) +class CoinLedgerRow: + """`list_coin_transactions` 的返回行。广告类按 trace_id 聚合后的展示行, + **非 ORM 对象**(避免把聚合后的 amount 误写回底层流水)。""" + id: int + amount: int + balance_after: int + biz_type: str + ref_id: str | None + remark: str | None + created_at: datetime + merged_count: int +``` + +- [ ] **Step 5: 重写 `list_coin_transactions`** + +`app/repositories/wallet.py` 第 260-279 行整体替换为： + +```python +def list_coin_transactions( + db: Session, + user_id: int, + *, + limit: int = 20, + cursor: int | None = None, +) -> tuple[list[CoinLedgerRow], int | None]: + """金币流水分页(游标式,按 id 倒序)。 + + 比价/领券两类信息流广告(rewards.FEED_AD_SESSION_BIZ_TYPES)且带 trace_id 的行, + 按 trace_id 聚合成一条(一次比价/领券 = 一行):金额合计、代表行取组内最新一条 + (MAX(id))的余额/时间、merged_count=组内条数。其余每条一行。 + + 分组必须在该用户**全量**行上算真实 rep_id 后再按 rep_id 过滤——**不可**把 CTE 输入 + 裁成 id" +``` + +--- + +## Task 5: 分页防残组回归测试 + +**Files:** +- Test: `tests/test_welfare.py`（追加） + +- [ ] **Step 1: 写回归测试** + +追加： + +```python +def test_coin_transactions_pagination_no_phantom_regroup(client) -> None: + """交错跨游标不产生残组:会话广告成员被其它记录隔开、rep_id 在游标上、成员在游标下时, + 翻到下一页该会话不得以「残组」重复出现(锁死 spec §11 反面优化警示)。""" + phone = "13800002007" + token = _login(client, phone) + with SessionLocal() as db: + user = get_user_by_phone(db, phone) + assert user is not None + _seed_coin(db, user.id, 5, "feed_ad_reward_comparison", trace_id="t1", ref_id="e1", remark="比价奖励") # id=n+1 + _seed_coin(db, user.id, 30, "signin", remark="每日签到奖励") # id=n+2 + _seed_coin(db, user.id, 40, "signin", remark="每日签到奖励") # id=n+3 + _seed_coin(db, user.id, 4, "feed_ad_reward_comparison", trace_id="t1", ref_id="e2", remark="比价奖励") # id=n+4 = t1 的 rep + db.commit() + # 第 1 页(limit=2):按 rep 降序 = [t1(rep=n+4, 合计 9), signin(n+3, 40)] + p1 = client.get("/api/v1/wallet/coin-transactions?limit=2", headers=_auth(token)).json() + assert len(p1["items"]) == 2 + assert p1["items"][0]["biz_type"] == "feed_ad_reward_comparison" + assert p1["items"][0]["amount"] == 9 and p1["items"][0]["merged_count"] == 2 + assert p1["items"][1]["biz_type"] == "signin" and p1["items"][1]["amount"] == 40 + assert p1["next_cursor"] is not None + # 第 2 页:只剩另一条 signin(30);t1 的 rep 在游标上,成员虽在游标下也不得成残组重复 + p2 = client.get( + f"/api/v1/wallet/coin-transactions?limit=2&cursor={p1['next_cursor']}", + headers=_auth(token), + ).json() + assert len(p2["items"]) == 1 + assert p2["items"][0]["biz_type"] == "signin" and p2["items"][0]["amount"] == 30 + assert all(i["biz_type"] != "feed_ad_reward_comparison" for i in p2["items"]) + assert p2["next_cursor"] is None +``` + +- [ ] **Step 2: 跑测试确认通过（锁定不变量）** + +Run: `pytest tests/test_welfare.py::test_coin_transactions_pagination_no_phantom_regroup -q` +Expected: PASS（Task 4 的全量分组实现已正确；若 FAIL 说明 CTE 被错误地按 cursor 裁剪，回到 Task 4 Step 5 修正）。 + +- [ ] **Step 3: lint** + +Run: `ruff check tests/test_welfare.py` +Expected: 无新增 lint。 + +- [ ] **Step 4: 提交** + +```bash +git add tests/test_welfare.py +git commit -m "test(wallet): 锁死金币记录分组分页不产生残组 + +Co-Authored-By: Claude Opus 4.8 (1M context) " +``` + +--- + +## Task 6: 更正 `ad_feed_reward_record.trace_id` 过时注释 + +**Files:** +- Modify: `app/models/ad_feed_reward.py:39-40` + +- [ ] **Step 1: 改注释** + +`app/models/ad_feed_reward.py` 第 39-40 行两行注释替换为： + +```python + # 本次会话 trace_id:比价一直带;领券自 2026-07-15(客户端 a98cab8)起也带;福利/旧客户端 = NULL。 + # 用途:比价记录页按 trace_id 聚合「比价赚 N 金币」;金币记录列表把一次比价/领券的多条广告金币聚合成一条。 + trace_id: Mapped[str | None] = mapped_column(String(64), index=True, nullable=True) +``` + +- [ ] **Step 2: 冒烟 + lint** + +Run: `python -c "import app.models.ad_feed_reward" && ruff check app/models/ad_feed_reward.py` +Expected: 无报错、无 lint。 + +- [ ] **Step 3: 提交** + +```bash +git add app/models/ad_feed_reward.py +git commit -m "docs(ad-feed): 更正 trace_id 注释(领券自 2026-07-15 也带) + +Co-Authored-By: Claude Opus 4.8 (1M context) " +``` + +--- + +## 收尾验证 + +- [ ] **全量测试**：`pytest -q`（对齐 memory「preexisting-test-lint-debt」：先前已知失败数不变、无本改动引入的新失败）。 +- [ ] **lint**：`ruff check .`（无本改动引入的新问题）。 +- [ ] **迁移落 dev**：`python -m alembic upgrade head`（把 dev SQLite 迁到最新，确认 `./run.sh` 正常）。 + +--- + +## Self-Review（写完计划后自查） + +**Spec 覆盖：** +- §4 加列 → Task 1 Step 4 + Task 2。✓ +- §5 写入透传 → Task 1 Step 5 + Task 3。✓ +- §6 聚合查询（CASE 分组键、MAX(id) 代表/游标、biz_type 门控、常量）→ Task 4 Step 4-5 + Task 1 常量。✓ +- §7 `merged_count` 下发 → Task 4 Step 3。✓ +- §8 回填（幂等 + EXISTS 守护 + 通用 SQL）→ Task 2 Step 4。✓ +- §9 注释更正 → Task 6。✓ +- §10 App 端无需改 → 端点未改（Task 4 Step 5 注）。✓ +- §11 反面优化警示（残组）→ Task 4 docstring + Task 5 回归测试。✓ +- §3 只改用户接口、admin 不动 → 全程只碰 `crud_wallet.list_coin_transactions`，未触 `app/admin`。✓ +- §12 测试（分组/隔离/空 trace/分页/回填）→ Task 2/4/5。✓ + +**占位扫描：** 无 TBD / TODO；每个代码步骤均给出完整代码与确切命令。✓ + +**类型一致：** `grant_coins(..., trace_id=...)`（Task 1）↔ `grant_feed_reward` 调用（Task 3）↔ 查询 `rewards.FEED_AD_SESSION_BIZ_TYPES`（Task 4）↔ 迁移硬编码同两值（Task 2，有意快照）一致；`CoinLedgerRow` 字段（Task 4 Step 4）↔ `CoinTransactionOut` 字段（Task 4 Step 3）逐一对应（id/amount/balance_after/biz_type/ref_id/remark/created_at/merged_count）。✓ diff --git a/docs/superpowers/specs/2026-08-07-coin-ledger-aggregate-ad-rewards-design.md b/docs/superpowers/specs/2026-08-07-coin-ledger-aggregate-ad-rewards-design.md new file mode 100644 index 0000000..80de998 --- /dev/null +++ b/docs/superpowers/specs/2026-08-07-coin-ledger-aggregate-ad-rewards-design.md @@ -0,0 +1,190 @@ +# 金币记录：比价/领券看广告金币按会话汇总成一条 设计 + +- 日期：2026-08-07 +- 分支：feat/coin-ledger-aggregate-ad-rewards +- 相关：`app/repositories/wallet.py`（发币/流水）、`app/repositories/ad_feed_reward.py`（信息流发奖）、`app/api/v1/wallet.py`（金币流水接口）、Android `CoinHistoryViewModel`（列表渲染） + +## 1. 背景与问题 + +App「金币变动记录」列表按**每看一次广告一条**展示。一次比价的等候期会连续看多条信息流广告，一次领券流程同理，于是列表里一次比价/领券会刷出一长串「比价奖励 +x」「领券奖励 +x」，淹没其它记录、观感差。 + +需求：**一次比价的所有看广告金币合并成一条、一次领券的合并成一条**，金额取该次会话合计。 + +## 2. 现状（关键事实） + +- 发币唯一入口 `crud_wallet.grant_coins`（`app/repositories/wallet.py:165`）：每条广告写一条 `CoinTransaction`。金币账本 `coin_transaction` 与现金账本物理分离。 +- 信息流发奖 `grant_feed_reward`（`app/repositories/ad_feed_reward.py:71`）按点位场景拆 `biz_type`： + - `feed_ad_reward_comparison`（比价，remark「比价奖励」） + - `feed_ad_reward_coupon`（领券，remark「领券奖励」） + - `feed_ad_reward`（通用/福利/旧端） +- **trace_id 已全链贯通到结算**： + - 比价：客户端结算上报早已带 trace_id。 + - 领券：`CouponForegroundService.reportFeedReward` 自 **2026-07-15**（commit `a98cab8`）起带 `feedScene="coupon" + traceId=sessionTraceId`。 + - 两者的 trace_id 都落到了 `ad_feed_reward_record.trace_id`（`grant_feed_reward` 写入）。 +- **唯一缺口**：`grant_feed_reward` 调 `grant_coins` 时**没把 trace_id 透传下去**，`coin_transaction` 表本身**没有 trace_id 列** → 金币流水层无法按会话分组。 +- 列表接口 `GET /api/v1/coin-transactions`（`app/api/v1/wallet.py:61` → `list_coin_transactions` `app/repositories/wallet.py:260`）：按 `id` 倒序游标分页。 +- 展示文案权在客户端：Android `CoinHistoryViewModel.coinTitle` 按 `bizType` 直显固定标题（比价奖励/领券奖励），后端 `remark` 只作兜底。 + +结论：这是**纯后端**改动，领券不需要额外 Android 改动即可生效。 + +## 3. 目标与范围 + +- 目标：金币记录列表里，同一次比价的多条「比价奖励」合并成一条、同一次领券的多条「领券奖励」合并成一条；金额为该次会话合计。 +- 会话键：`trace_id`（一次比价/一次领券 = 一个 trace_id）。与「比价记录页」现有的 trace_id 聚合口径一致。 +- **范围**：只圈两类 `biz_type` —— `feed_ad_reward_comparison`、`feed_ad_reward_coupon`。激励视频 `reward_video`、引导视频 `guide_video`、通用信息流 `feed_ad_reward`、签到、任务、兑换等**一律不动**，仍每条一行。 +- 聚合位置：**后端**（`/coin-transactions` 直接返回合并后的行）。客户端聚合被否决（见 §11）。 +- **只改 App 用户接口**：聚合仅作用于 `crud_wallet.list_coin_transactions`（`GET /api/v1/wallet/coin-transactions`）。**admin 接口不动**——`app/admin/routers/wallet.py` 走独立的 `queries.list_all_coin_transactions`（跨用户、可按 `biz_type` 筛），审计/客服必须能看到**每一条**广告发币，保持每条一行。两者物理分离，聚合天然不波及 admin，此处显式声明防误改。 +- 历史：**加列 + 一次性回填**，历史记录也合并。 + +## 4. 数据模型改动 + +`coin_transaction` 新增列： + +| 列 | 类型 | 约束 | 说明 | +|---|---|---|---| +| `trace_id` | `String(64)` | nullable, index | 会话键；仅比价/领券信息流发奖时写入，其余为 NULL | + +- 新增 Alembic 迁移（`render_as_batch` 兼容 SQLite）：加列 + 索引 `ix_coin_transaction_trace_id`。 +- 可选复合索引 `(user_id, trace_id)` 辅助分组扫描（视线上量级决定，MVP 可先只加单列索引）。 + +## 5. 写入路径改动 + +- `grant_coins(...)` 增加可选参数 `trace_id: str | None = None`，写入 `CoinTransaction.trace_id`。默认 None → 其余调用方（签到/任务/兑换/激励视频）零改动、保持 NULL。 +- `grant_feed_reward` 调 `grant_coins` 时透传 `trace_id=trace_id`（比价/领券自然有值；welfare/旧端为 None）。 +- 幂等不变：`client_event_id` 仍是幂等键；`grant_coins` 仍不 commit，由 `_commit_record` 同事务提交。 + +## 6. 读取/聚合查询（核心） + +`list_coin_transactions` 改为「分组游标分页」。用 CTE 先分组、再 JOIN 回代表行取展示字段。 + +分组规则： + +``` +group_key = + 若 biz_type ∈ (feed_ad_reward_comparison, feed_ad_reward_coupon) 且 trace_id 非空 + → 'T:' || trace_id # 同一会话所有广告归一组 + 否则 + → 'I:' || id # 每条自成一组(行为等同现状) +``` + +每组取：`rep_id = MAX(id)`、`total = SUM(amount)`、`merged_count = COUNT(*)`；再 JOIN `coin_transaction` 取 `rep_id` 那条的 `balance_after / biz_type / ref_id / remark / created_at / trace_id`。 + +等价 SQL（SQLite / PostgreSQL 通用）： + +```sql +WITH grp AS ( + SELECT + CASE WHEN biz_type IN ('feed_ad_reward_comparison','feed_ad_reward_coupon') + AND trace_id IS NOT NULL + THEN 'T:' || trace_id + ELSE 'I:' || CAST(id AS TEXT) END AS group_key, + MAX(id) AS rep_id, + SUM(amount) AS total_amount, + COUNT(*) AS merged_count + FROM coin_transaction + WHERE user_id = :uid + GROUP BY group_key +) +SELECT ct.id, grp.total_amount AS amount, ct.balance_after, + ct.biz_type, ct.ref_id, ct.remark, ct.trace_id, ct.created_at, + grp.merged_count +FROM grp +JOIN coin_transaction ct ON ct.id = grp.rep_id +WHERE (:cursor IS NULL OR grp.rep_id < :cursor) +ORDER BY grp.rep_id DESC +LIMIT :limit; +``` + +返回行字段口径： + +- `id` = 组内 `MAX(id)`：做列表 key + 下一页游标。组间不重复（每个 id 只属一组），`rep_id DESC` 是全序，游标 `rep_id < cursor` 干净。 +- `amount` = 组内合计（该次会话总金币；这些广告行金额恒正）。 +- `balance_after` / `created_at` / `remark` / `ref_id` = 代表行（最后一条）的值。`balance_after` 即该次会话最后一条广告到账后的余额，正确。 +- `merged_count` = 合并条数（未合并 = 1）。 +- `next_cursor` = 本页最后一行的 `id`（= 其组 rep_id），够 limit 才给，否则 None。 + +非分组类型、`trace_id` 为空的旧广告行 → 各自成组（`'I:'||id`），行为与现状完全一致。 + +> 说明（分组安全性）：分组键按 `biz_type` **门控**——通用 `feed_ad_reward`（福利/旧端）即便偶带 trace_id 也走 `'I:'||id` 保持每条一行，只有比价/领券两类才按 trace_id 合并。trace_id 由后端按会话签发、全局唯一（一个 trace = 一次比价**或**一次领券），故仅按 trace_id 分组不会把两类混并，代表行的 `biz_type` 唯一确定展示标题。 +> +> 实现约定：这两类 biz_type 抽成模块常量 `FEED_AD_SESSION_BIZ_TYPES`，**查询 / 回填 / 测试共用一处**，避免字符串散落三地漂移。 + +## 7. 接口契约 + +`GET /api/v1/coin-transactions` 响应新增 `merged_count`（`CoinTransactionOut` 加字段，Pydantic 默认 1）： + +```jsonc +{ + "items": [ + { "id": 1520, "amount": 12, "balance_after": 3380, + "biz_type": "feed_ad_reward_comparison", "ref_id": "evt_...", + "remark": "比价奖励", "created_at": "2026-08-07T12:03:11", + "merged_count": 5 }, // 本次比价看了 5 条广告，合计 +12 + { "id": 1512, "amount": 30, "balance_after": 3368, + "biz_type": "signin", "remark": "每日签到奖励", + "merged_count": 1 } + ], + "next_cursor": 1490 +} +``` + +`trace_id` 是否在响应里下发：**可选**，MVP 不下发（客户端用不到，标题按 biz_type、金额已合计）。若后续要做「点开看明细」再补。 + +## 8. 历史回填 + +迁移里一次性把历史 `coin_transaction.trace_id` 从 `ad_feed_reward_record` 补齐（相关子查询 UPDATE，SQLite/PG 通用）： + +```sql +UPDATE coin_transaction SET trace_id = ( + SELECT r.trace_id FROM ad_feed_reward_record r + WHERE r.client_event_id = coin_transaction.ref_id) +WHERE biz_type IN ('feed_ad_reward_comparison','feed_ad_reward_coupon') + AND trace_id IS NULL + AND EXISTS (SELECT 1 FROM ad_feed_reward_record r2 + WHERE r2.client_event_id = coin_transaction.ref_id + AND r2.trace_id IS NOT NULL); +``` + +- 关联键：`coin_transaction.ref_id == ad_feed_reward_record.client_event_id`（`grant_feed_reward` 里两者同源，见 `app/repositories/ad_feed_reward.py:221`）。`client_event_id` 有唯一约束 → 相关子查询至多一行，无「子查询多行」风险。 +- 只补两类 biz_type；只在广告行确有 trace_id 时补（`EXISTS` 守护）。 +- **幂等**：`AND trace_id IS NULL` 保证重跑不重复改。 +- **大表守护**：`coin_transaction` 线上若很大，单条全表 UPDATE 在 PG 上是一次性长事务/行锁；必要时按 `id` 区间**分批**（每批数万行）跑。SQLite（dev）无所谓。 +- 2026-07-15 前的领券老行 `ad_feed_reward_record.trace_id` 本就为空 → 回填后仍为空 → 保持每条一行（可接受，会随时间自然淘汰）。 + +## 9. 顺手更正 + +更新 `app/models/ad_feed_reward.py:39` 的 `trace_id` 注释：现「仅 comparison 场景由客户端带上……领券/福利/旧客户端 = NULL」已过时，改为「比价一直带；领券自 2026-07-15 起也带；福利/旧客户端仍 NULL」。 + +## 10. App 端 + +- **核心无需改**：接口返回合并后的行，`CoinHistoryViewModel.coinTitle` 仍按 `bizType` 显示「比价奖励/领券奖励」，金额显示合计值。 +- **可选增强（后续、非本次必须）**：客户端读到 `merged_count > 1` 时把标题渲染成「比价奖励 ×5」或副行「看了 5 条广告」。做不做都不影响本功能生效。 + +## 11. 边界与取舍 + +- **时间交错重排**：一次比价的广告中间夹了签到等其它记录时，合并行锚定在「该会话最后一条广告」的 id 上，夹在中间的签到会排到合并行之后。这是折叠的固有效果，可接受。 +- **余额不连续**：相邻两展示行的 `balance_after` 之间存在被折叠的中间变动，属聚合固有现象；每行 `balance_after` 仍是该点真实运行余额。 +- **迟到广告**：会话已过、同 trace 的迟到广告到账后并入该组、金额变大、合并行上移（因 rep_id 变大）；用户未刷新则短暂 stale。会话结束后广告即时结算，迟到罕见，可接受。 +- **单条会话**：一次会话只看了 1 条广告 → 组内 1 行、`merged_count=1`、展示与现状一致，无需特判。 +- **查询成本**：分组 CTE 每页对该用户**全部** `coin_transaction` 行分组后再分页，O(N)/页，比原「`id < cursor LIMIT n`」的索引 seek 重。金币记录非热路径（偶尔打开、翻几页）；粗估重度用户万级行/年、单页分组数毫秒级（估算，非实测），可接受。 + - **索引口径**：分组按 CASE 表达式（`'T:'||trace_id` / `'I:'||id`）在内存做，`(user_id, trace_id)` 复合索引**并不加速**该 group-by；真正需要的只是「按 user_id 取该用户全部行」，现有 `ix_coin_transaction_user_id` 已够。故本设计**只加 `trace_id` 单列索引**（回填 / 潜在按 trace 查用），不加 `(user_id, trace_id)`；若要 index-only 可选覆盖索引 `(user_id, id, biz_type, trace_id, amount)`，MVP 不加。 + - ⚠️ **反面优化警示**：**不要**把 CTE 输入裁成 `WHERE id < :cursor` 求快。会话行若被其它记录交错、成员跨越游标（部分成员 `id < cursor`、但组的 `rep_id ≥ cursor`），裁剪后会算出一个 rep_id 更小的「残组」→ 与上一页已展示的整组**重复出现**。分组必须在该用户全量行上算出真实 `rep_id` 后，再 `rep_id < cursor` 过滤。 + - **未来读优化（YAGNI，暂不做）**：若量级压力显现，加只读会话聚合投影表 `coin_ad_session_agg(user_id, trace_id, total_coin, cnt, last_txn_id, last_created_at)`，写广告流水时 upsert，列表用「非广告行 UNION 该投影」走索引分页。这是**读优化**（明细账本仍每条一行、审计不动），与 §13 否决的「写时账本聚合」不同。 + +## 12. 测试 + +- **分组**：同 trace 多条比价广告 → 一行（`amount` 合计 / `merged_count=N` / `balance_after` 取最后 / `id=MAX`）；两次不同 trace 的比价 → 两行；领券同 trace → 一行。 +- **隔离**：比价广告序列中间夹一条签到 → 比价折叠成一行、签到独立成行；`reward_video` / `signin` / `exchange_out` → 仍 1:1，`merged_count=1`。 +- **空 trace_id**：`feed_ad_reward`（通用）或 trace_id 为 NULL 的广告行 → 每条一行。 +- **分页**：多组跨页时按 `rep_id` 游标翻页不重不漏；一个大组（跨越 limit 边界的多条底层行）不被拆成两页；`next_cursor` 到底返回 None。 +- **交错跨游标不重复（防残组回归）**：造一个会话，其广告成员 id 被一条非广告行隔开（部分成员在游标下、`rep_id` 在游标上），翻到下一页时该组**不得**再以「残组」重复出现——锁死 §11 的反面优化警示。 +- **回填迁移**：造 `coin_transaction`（比价/领券行，ref_id 指向带 trace_id 的 `ad_feed_reward_record`）+ 无关行，跑迁移后仅两类被正确补 trace_id、无关行不动。 +- **既有测试**：`/coin-transactions` 现有用例（非广告类型 1:1、分页）保持通过。 + +## 13. 不做（YAGNI） + +- 不做写时聚合（会破坏实时到账 / 幂等 / 余额连续性）。 +- 不做客户端聚合（SWR 缓存 + 游标分页下，一个会话可能跨页，跨页分组脆弱）。 +- 不动激励视频 / 引导视频 / 通用信息流 / 签到等其它类型。 +- 不动「比价记录页」——它的「比价赚 N 金币」是独立查询（直接聚合 `ad_feed_reward_record`），不受本改动影响。 +- MVP 不下发 trace_id、不做「点开看明细」。 diff --git a/tests/test_admin_read.py b/tests/test_admin_read.py index 6b86c2e..8630a82 100644 --- a/tests/test_admin_read.py +++ b/tests/test_admin_read.py @@ -906,6 +906,7 @@ def test_comparison_records_show_real_order_status( order_amount_cents=1800, saved_amount_cents=300, shop_name="真实下单店", + trace_id="comparison-ordered-shop", source="compare", client_event_id="admin-comparison-real-order", ), diff --git a/tests/test_compare_record.py b/tests/test_compare_record.py index f16798c..c452981 100644 --- a/tests/test_compare_record.py +++ b/tests/test_compare_record.py @@ -223,10 +223,10 @@ def test_stats_compare_count_and_saved(client) -> None: def test_records_ordered_flag(client) -> None: - """「已下单」店级标记:店名命中该用户 source='compare' 的下单记录才 True。 + """「已下单」按 trace_id 标记:下单上报带上本次比价的 trace_id,精确命中该条记录才 True。 - 覆盖 list_records 只按**本页店名**反查 savings 的写法(原来是把该用户全部下单店名捞回内存 - 再取交集,随下单量线性变慢)——两种写法结果必须一致,故这里按店名逐条断言。 + 覆盖 list_records 只按**本页 trace_id**反查 savings 的写法(不把该用户全部下单 trace_id 捞回 + 内存再取交集)——两种写法结果必须一致,故这里逐条断言。 """ token = _login(client, "13800002010") @@ -236,14 +236,15 @@ def test_records_ordered_flag(client) -> None: other["store_name"] = "没下过单的店" client.post("/api/v1/compare/record", json=other, headers=_auth(token)) - # 下单前:两条都不该带「已下单」 - items = client.get("/api/v1/compare/records", headers=_auth(token)).json()["items"] - assert {it["store_name"]: it["ordered"] for it in items} == { - "海底捞(朝阳店)": False, - "没下过单的店": False, + # 下单前:两条都不该带「已下单」。按 trace_id 断言本测试自己的两条(不受同库其它用例数据干扰)。 + flags = { + it["trace_id"]: it["ordered"] + for it in client.get("/api/v1/compare/records", headers=_auth(token)).json()["items"] } + assert flags["ord-1"] is False + assert flags["ord-2"] is False - # 对海底捞真实下单一笔(order/report 写 source='compare' 的 savings_record) + # 对海底捞那次比价真实下单一笔:上报带上该次比价的 trace_id(order/report 写 source='compare') r = client.post( "/api/v1/order/report", json={ @@ -255,23 +256,28 @@ def test_records_ordered_flag(client) -> None: "paid_amount_cents": 12350, "shop_name": "海底捞(朝阳店)", "original_price_cents": 12850, + "trace_id": "ord-1", }, headers=_auth(token), ) assert r.status_code == 200, r.text - # 下单后:只有同店名那条翻成 True,另一条不受影响 - items = client.get("/api/v1/compare/records", headers=_auth(token)).json()["items"] - assert {it["store_name"]: it["ordered"] for it in items} == { - "海底捞(朝阳店)": True, - "没下过单的店": False, + # 下单后:只有命中 trace_id(ord-1)那条翻成 True,同店的另一次比价(ord-2)不受影响 + flags = { + it["trace_id"]: it["ordered"] + for it in client.get("/api/v1/compare/records", headers=_auth(token)).json()["items"] } + assert flags["ord-1"] is True + assert flags["ord-2"] is False - # 别人的下单不该影响本人标记(_ordered_shop_names 按 user_id 过滤) + # 别人的下单不该影响本人标记(_ordered_trace_ids 按 user_id 过滤) token_b = _login(client, "13800002011") client.post("/api/v1/compare/record", json=_food_payload("ord-b"), headers=_auth(token_b)) - items_b = client.get("/api/v1/compare/records", headers=_auth(token_b)).json()["items"] - assert [it["ordered"] for it in items_b] == [False] + flags_b = { + it["trace_id"]: it["ordered"] + for it in client.get("/api/v1/compare/records", headers=_auth(token_b)).json()["items"] + } + assert flags_b["ord-b"] is False def test_records_list_omits_raw_payload(client) -> None: @@ -316,20 +322,24 @@ def test_records_ordered_filter(client) -> None: q["store_name"] = "没下过单的店" client.post("/api/v1/compare/record", json=q, headers=_auth(token)) - client.post( - "/api/v1/order/report", - json={ - "client_event_id": "evt-ordered-filter", - "platform": "美团", - "platform_package": "com.sankuai.meituan", - "pay_channel": "wechat", - "compared_price_cents": 12350, - "paid_amount_cents": 12350, - "shop_name": "下过单的店", - "original_price_cents": 12850, - }, - headers=_auth(token), - ) + # 对这 3 条「下过单的店」比价分别真实下单(各带自己的 trace_id);2 条「没下过单的店」不下单 + for i in range(3): + r = client.post( + "/api/v1/order/report", + json={ + "client_event_id": f"evt-ordered-filter-{i}", + "platform": "美团", + "platform_package": "com.sankuai.meituan", + "pay_channel": "wechat", + "compared_price_cents": 12350, + "paid_amount_cents": 12350, + "shop_name": "下过单的店", + "original_price_cents": 12850, + "trace_id": f"of-ordered-{i}", + }, + headers=_auth(token), + ) + assert r.status_code == 200, r.text # 不传 ordered:5 条全出(「全部记录」tab 口径不变) assert len(client.get("/api/v1/compare/records", headers=_auth(token)).json()["items"]) == 5 @@ -439,3 +449,84 @@ def test_trace_id_required(client) -> None: token = _login(client, "13800002009") r = client.post("/api/v1/compare/record", json={"business_type": "food"}, headers=_auth(token)) assert r.status_code == 422 + + +def test_records_ordered_by_trace_id(client) -> None: + """「已下单」精确到单次比价:同一家店比价多次,只有真实下单的那一次(下单上报带 trace_id)才算已下单。 + + 下单上报带上本次比价的 trace_id,服务端按 trace_id 对齐 —— 同店其它比价(哪怕成功)不再被一并 + 标「已下单」。这是从"店级"改"单次级"的核心契约。 + """ + token = _login(client, "13800002016") + + # 同一家店(默认 payload 店名都是「海底捞(朝阳店)」)比价两次,不同 trace_id + client.post("/api/v1/compare/record", json=_food_payload("ord-trace-a"), headers=_auth(token)) + client.post("/api/v1/compare/record", json=_food_payload("ord-trace-b"), headers=_auth(token)) + + # 只对其中一次(trace-a)真实下单,上报带上该次比价的 trace_id + r = client.post( + "/api/v1/order/report", + json={ + "client_event_id": "evt-trace-a", + "platform": "美团", + "platform_package": "com.sankuai.meituan", + "pay_channel": "wechat", + "compared_price_cents": 12350, + "paid_amount_cents": 12350, + "shop_name": "海底捞(朝阳店)", + "original_price_cents": 12850, + "trace_id": "ord-trace-a", + }, + headers=_auth(token), + ) + assert r.status_code == 200, r.text + + # ordered=true 只出下单的那条(trace-a);同店没下单的 trace-b 不进来 + page = client.get("/api/v1/compare/records?ordered=true", headers=_auth(token)).json() + assert [it["trace_id"] for it in page["items"]] == ["ord-trace-a"] + + # 全部记录里,只有 trace-a 带 ordered=true,同店的 trace-b 仍是 False + items = client.get("/api/v1/compare/records", headers=_auth(token)).json()["items"] + assert {it["trace_id"]: it["ordered"] for it in items} == { + "ord-trace-a": True, + "ord-trace-b": False, + } + + +def test_records_ordered_ignores_order_without_trace_id(client) -> None: + """无 trace_id 的下单(老客户端/历史)即使店名相同也不进「已下单」—— 锁定「无店名回退」契约。 + + 下单上报不带 trace_id → savings_record.trace_id 为空 → 对齐不上任何比价记录。老逻辑靠店名 + 会把同店比价一并标已下单,这条正是要防它被改回去。 + """ + token = _login(client, "13800002017") + + # 一条比价记录(默认店名「海底捞(朝阳店)」) + client.post("/api/v1/compare/record", json=_food_payload("no-trace-order"), headers=_auth(token)) + + # 同店名、但**不带 trace_id** 的下单(模拟老客户端) + r = client.post( + "/api/v1/order/report", + json={ + "client_event_id": "evt-no-trace", + "platform": "美团", + "platform_package": "com.sankuai.meituan", + "pay_channel": "wechat", + "compared_price_cents": 12350, + "paid_amount_cents": 12350, + "shop_name": "海底捞(朝阳店)", + "original_price_cents": 12850, + }, + headers=_auth(token), + ) + assert r.status_code == 200, r.text + + # 店名虽同,但订单没 trace_id → 该记录不该被标「已下单」,也不进 ordered=true(无店名回退) + flags = { + it["trace_id"]: it["ordered"] + for it in client.get("/api/v1/compare/records", headers=_auth(token)).json()["items"] + } + assert flags["no-trace-order"] is False + assert client.get( + "/api/v1/compare/records?ordered=true", headers=_auth(token) + ).json()["items"] == [] diff --git a/tests/test_welfare.py b/tests/test_welfare.py index bd99334..fc8d50e 100644 --- a/tests/test_welfare.py +++ b/tests/test_welfare.py @@ -4,7 +4,7 @@ """ from __future__ import annotations -import json +from sqlalchemy import select, text from app.core.config import settings from app.core.rewards import ( @@ -17,6 +17,9 @@ from app.core.rewards import ( ) from app.db.session import SessionLocal from app.integrations import pangle +from app.models.ad_feed_reward import AdFeedRewardRecord +from app.models.wallet import CoinTransaction +from app.repositories import ad_feed_reward as crud_feed from app.repositories import wallet as crud_wallet from app.repositories.user import get_user_by_phone @@ -384,3 +387,208 @@ def test_endpoints_require_auth(client) -> None: assert client.get("/api/v1/savings/summary").status_code == 401 assert client.get("/api/v1/savings/battle").status_code == 401 assert client.get("/api/v1/savings/records").status_code == 401 + + +def test_grant_coins_persists_trace_id(client) -> None: + """grant_coins 传 trace_id 落库;不传则为 None。""" + phone = "13800002001" + _login(client, phone) + with SessionLocal() as db: + user = get_user_by_phone(db, phone) + assert user is not None + _, txn1 = crud_wallet.grant_coins( + db, user.id, 5, biz_type="feed_ad_reward_comparison", + ref_id="evt1", remark="比价奖励", trace_id="trace-A", + ) + _, txn2 = crud_wallet.grant_coins( + db, user.id, 30, biz_type="signin", remark="每日签到奖励", + ) + db.commit() + assert txn1.trace_id == "trace-A" + assert txn2.trace_id is None + + +def test_backfill_coin_trace_id_from_ad_record(client) -> None: + """回填:coin_transaction(trace_id 空)按 ref_id==client_event_id 从 ad_feed_reward_record 补 trace_id; + 只补比价/领券两类,无关类型与无匹配的不动。SQL 与迁移 coin_transaction_trace_id 保持同步。""" + phone = "13800002002" + _login(client, phone) + with SessionLocal() as db: + user = get_user_by_phone(db, phone) + assert user is not None + db.add(AdFeedRewardRecord( + client_event_id="evt-cmp", user_id=user.id, reward_date="2026-08-07", + duration_seconds=10, unit_count=1, ecpm_raw="1000", + feed_scene="comparison", trace_id="trace-CMP", coin=5, status="granted", + )) + db.add(AdFeedRewardRecord( + client_event_id="evt-cpn", user_id=user.id, reward_date="2026-08-07", + duration_seconds=10, unit_count=1, ecpm_raw="1000", + feed_scene="coupon", trace_id="trace-CPN", coin=7, status="granted", + )) + # 广告行存在但 trace_id 为空(2026-07-15 前的老比价广告):回填必须跳过、保持 NULL(EXISTS 守护)。 + db.add(AdFeedRewardRecord( + client_event_id="evt-null", user_id=user.id, reward_date="2026-08-07", + duration_seconds=10, unit_count=1, ecpm_raw="1000", + feed_scene="comparison", trace_id=None, coin=5, status="granted", + )) + db.commit() + _, c1 = crud_wallet.grant_coins(db, user.id, 5, biz_type="feed_ad_reward_comparison", ref_id="evt-cmp", remark="比价奖励") + _, c2 = crud_wallet.grant_coins(db, user.id, 7, biz_type="feed_ad_reward_coupon", ref_id="evt-cpn", remark="领券奖励") + _, c3 = crud_wallet.grant_coins(db, user.id, 30, biz_type="signin", remark="每日签到奖励") + _, c4 = crud_wallet.grant_coins(db, user.id, 5, biz_type="feed_ad_reward_comparison", ref_id="evt-null", remark="比价奖励") + db.commit() + assert c1.trace_id is None and c2.trace_id is None + + db.execute(text( + """ + UPDATE coin_transaction SET trace_id = ( + SELECT r.trace_id FROM ad_feed_reward_record r + WHERE r.client_event_id = coin_transaction.ref_id) + WHERE biz_type IN ('feed_ad_reward_comparison', 'feed_ad_reward_coupon') + AND trace_id IS NULL + AND EXISTS (SELECT 1 FROM ad_feed_reward_record r2 + WHERE r2.client_event_id = coin_transaction.ref_id + AND r2.trace_id IS NOT NULL) + """ + )) + db.commit() + db.refresh(c1) + db.refresh(c2) + db.refresh(c3) + db.refresh(c4) + assert c1.trace_id == "trace-CMP" + assert c2.trace_id == "trace-CPN" + assert c3.trace_id is None # 无关类型不动 + assert c4.trace_id is None # 广告行 trace_id 为空 → EXISTS 守护跳过、保持 NULL + + +def test_grant_feed_reward_sets_coin_trace_id(client) -> None: + """grant_feed_reward(comparison) 把 trace_id 透传给 grant_coins,coin_transaction 带上本场 trace_id。""" + phone = "13800002003" + _login(client, phone) + with SessionLocal() as db: + user = get_user_by_phone(db, phone) + assert user is not None + rec = crud_feed.grant_feed_reward( + db, user.id, + client_event_id="evt-fr-1", ecpm="1000", duration_seconds=10, + feed_scene="comparison", trace_id="trace-FR", display_coin=5, + ) + assert rec.status == "granted", rec.status + txn = db.execute( + select(CoinTransaction).where( + CoinTransaction.user_id == user.id, + CoinTransaction.ref_id == "evt-fr-1", + ) + ).scalar_one() + assert txn.biz_type == "feed_ad_reward_comparison" + assert txn.trace_id == "trace-FR" + + +def _seed_coin(db, user_id, amount, biz_type, *, trace_id=None, ref_id=None, remark=None): + crud_wallet.grant_coins( + db, user_id, amount, biz_type=biz_type, ref_id=ref_id, remark=remark, trace_id=trace_id + ) + + +def test_coin_transactions_aggregate_by_trace(client) -> None: + """一次比价的多条广告金币聚合成一条:金额合计、merged_count=条数、balance_after 取最后一条。""" + phone = "13800002004" + token = _login(client, phone) + with SessionLocal() as db: + user = get_user_by_phone(db, phone) + assert user is not None + _seed_coin(db, user.id, 5, "feed_ad_reward_comparison", trace_id="t1", ref_id="e1", remark="比价奖励") + _seed_coin(db, user.id, 3, "feed_ad_reward_comparison", trace_id="t1", ref_id="e2", remark="比价奖励") + _seed_coin(db, user.id, 4, "feed_ad_reward_comparison", trace_id="t1", ref_id="e3", remark="比价奖励") + db.commit() + r = client.get("/api/v1/wallet/coin-transactions", headers=_auth(token)) + assert r.status_code == 200, r.text + items = r.json()["items"] + assert len(items) == 1 + row = items[0] + assert row["biz_type"] == "feed_ad_reward_comparison" + assert row["amount"] == 12 + assert row["merged_count"] == 3 + assert row["balance_after"] == 12 + + +def test_coin_transactions_distinct_traces_stay_separate(client) -> None: + """不同 trace(两次比价 / 一次领券)各成一条;不同会话不合并。""" + phone = "13800002005" + token = _login(client, phone) + with SessionLocal() as db: + user = get_user_by_phone(db, phone) + assert user is not None + _seed_coin(db, user.id, 5, "feed_ad_reward_comparison", trace_id="cmpA", ref_id="a1", remark="比价奖励") + _seed_coin(db, user.id, 6, "feed_ad_reward_comparison", trace_id="cmpB", ref_id="b1", remark="比价奖励") + _seed_coin(db, user.id, 6, "feed_ad_reward_comparison", trace_id="cmpB", ref_id="b2", remark="比价奖励") + _seed_coin(db, user.id, 7, "feed_ad_reward_coupon", trace_id="cpnC", ref_id="c1", remark="领券奖励") + db.commit() + items = client.get("/api/v1/wallet/coin-transactions", headers=_auth(token)).json()["items"] + assert len(items) == 3 + assert sorted(i["amount"] for i in items) == [5, 7, 12] + cpn = next(i for i in items if i["biz_type"] == "feed_ad_reward_coupon") + assert cpn["amount"] == 7 and cpn["merged_count"] == 1 + + +def test_coin_transactions_non_session_rows_stay_per_row(client) -> None: + """签到 / 无 trace 的通用信息流各自一行,不被聚合;夹在比价广告中间的签到不影响比价聚合。""" + phone = "13800002006" + token = _login(client, phone) + with SessionLocal() as db: + user = get_user_by_phone(db, phone) + assert user is not None + _seed_coin(db, user.id, 5, "feed_ad_reward_comparison", trace_id="t1", ref_id="e1", remark="比价奖励") + _seed_coin(db, user.id, 30, "signin", remark="每日签到奖励") + _seed_coin(db, user.id, 4, "feed_ad_reward_comparison", trace_id="t1", ref_id="e2", remark="比价奖励") + _seed_coin(db, user.id, 8, "feed_ad_reward", trace_id=None, ref_id="w1", remark="信息流广告奖励") + db.commit() + items = client.get("/api/v1/wallet/coin-transactions", headers=_auth(token)).json()["items"] + assert len(items) == 3 + cmp_row = next(i for i in items if i["biz_type"] == "feed_ad_reward_comparison") + assert cmp_row["amount"] == 9 and cmp_row["merged_count"] == 2 + signin_row = next(i for i in items if i["biz_type"] == "signin") + assert signin_row["amount"] == 30 and signin_row["merged_count"] == 1 + feed_row = next(i for i in items if i["biz_type"] == "feed_ad_reward") + assert feed_row["amount"] == 8 and feed_row["merged_count"] == 1 + + +def test_coin_transactions_pagination_no_phantom_regroup(client) -> None: + """交错跨游标不产生残组:会话广告成员被其它记录隔开、rep_id 在游标上、成员在游标下时, + 翻到下一页该会话不得以「残组」重复出现(锁死 spec §11 反面优化警示)。""" + phone = "13800002007" + token = _login(client, phone) + with SessionLocal() as db: + user = get_user_by_phone(db, phone) + assert user is not None + _seed_coin(db, user.id, 5, "feed_ad_reward_comparison", trace_id="t1", ref_id="e1", remark="比价奖励") # id=n+1 + _seed_coin(db, user.id, 30, "signin", remark="每日签到奖励") # id=n+2 + _seed_coin(db, user.id, 40, "signin", remark="每日签到奖励") # id=n+3 + _seed_coin(db, user.id, 4, "feed_ad_reward_comparison", trace_id="t1", ref_id="e2", remark="比价奖励") # id=n+4 = t1 的 rep + db.commit() + t1_ids = db.execute( + select(CoinTransaction.id) + .where(CoinTransaction.user_id == user.id, CoinTransaction.trace_id == "t1") + .order_by(CoinTransaction.id) + ).scalars().all() + # 第 1 页(limit=2):按 rep 降序 = [t1(rep=n+4, 合计 9), signin(n+3, 40)] + p1 = client.get("/api/v1/wallet/coin-transactions?limit=2", headers=_auth(token)).json() + assert len(p1["items"]) == 2 + assert p1["items"][0]["biz_type"] == "feed_ad_reward_comparison" + assert p1["items"][0]["amount"] == 9 and p1["items"][0]["merged_count"] == 2 + assert p1["items"][1]["biz_type"] == "signin" and p1["items"][1]["amount"] == 40 + assert p1["next_cursor"] is not None + # 前置条件:游标(第1页末条 rep)必须严格落在 t1 两成员 id 之间,才是真正的「跨游标」场景; + # 否则用例会退化成非跨游标、悄悄测错形状仍通过,失去回归守护意义。 + assert t1_ids[0] < p1["next_cursor"] < t1_ids[-1] + # 第 2 页:只剩另一条 signin(30);t1 的 rep 在游标上,成员虽在游标下也不得成残组重复 + p2 = client.get( + f"/api/v1/wallet/coin-transactions?limit=2&cursor={p1['next_cursor']}", + headers=_auth(token), + ).json() + assert len(p2["items"]) == 1 + assert p2["items"][0]["biz_type"] == "signin" and p2["items"][0]["amount"] == 30 + assert all(i["biz_type"] != "feed_ad_reward_comparison" for i in p2["items"]) + assert p2["next_cursor"] is None