feat(ad): eCPM 上报接口 + 冷却策略解耦纯函数 + savings JSONB 跨库修复

- 新增 POST /api/v1/ad/ecpm-report:激励视频展示后客户端上报本次 eCPM,落 ad_ecpm_record 做内部收益统计(model/repository/schema/alembic 迁移 + 接口文档)
- 看广告冷却策略抽到 app/core/ad_cooldown.py 纯函数;ad_reward.today_status 只取数据,换策略只改这一处
- savings.dishes 改 JSON().with_variant(JSONB,"postgresql"):SQLite 无 visit_JSONB 会让 create_all 编译崩,修复后测试套件恢复

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

alembic/versions/ad_ecpm_record.py

@@ -0,0 +1,47 @@
+"""ad_ecpm_record table (广告展示 eCPM 上报记录,内部收益统计/对账)
+
+Revision ID: a1b2c3d4e5f6
+Revises: f01db5d77dac
+Create Date: 2026-05-31 11:30:00.000000
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = 'a1b2c3d4e5f6'
+down_revision: Union[str, Sequence[str], None] = 'f01db5d77dac'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        'ad_ecpm_record',
+        sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
+        sa.Column('user_id', sa.Integer(), nullable=False),
+        sa.Column('ad_type', sa.String(length=32), nullable=False),
+        sa.Column('adn', sa.String(length=32), nullable=True),
+        sa.Column('slot_id', sa.String(length=64), nullable=True),
+        sa.Column('ecpm_raw', sa.String(length=32), nullable=False),
+        sa.Column('report_date', sa.String(length=10), nullable=False),
+        sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False),
+        sa.ForeignKeyConstraint(['user_id'], ['user.id'], ),
+        sa.PrimaryKeyConstraint('id'),
+    )
+    with op.batch_alter_table('ad_ecpm_record', schema=None) as batch_op:
+        batch_op.create_index(batch_op.f('ix_ad_ecpm_record_user_id'), ['user_id'], unique=False)
+        batch_op.create_index(batch_op.f('ix_ad_ecpm_record_report_date'), ['report_date'], unique=False)
+        batch_op.create_index(batch_op.f('ix_ad_ecpm_record_created_at'), ['created_at'], unique=False)
+
+
+def downgrade() -> None:
+    with op.batch_alter_table('ad_ecpm_record', schema=None) as batch_op:
+        batch_op.drop_index(batch_op.f('ix_ad_ecpm_record_created_at'))
+        batch_op.drop_index(batch_op.f('ix_ad_ecpm_record_report_date'))
+        batch_op.drop_index(batch_op.f('ix_ad_ecpm_record_user_id'))
+
+    op.drop_table('ad_ecpm_record')

app/api/v1/ad.py

@@ -19,8 +19,15 @@ from app.core import rewards
 from app.core.config import settings
 from app.integrations import pangle
 from app.core.ratelimit import rate_limit
+from app.repositories import ad_ecpm as crud_ecpm
 from app.repositories import ad_reward as crud_ad
-from app.schemas.ad import AdRewardStatusOut, PangleCallbackOut, TestGrantOut
+from app.schemas.ad import (
+    AdRewardStatusOut,
+    EcpmReportIn,
+    EcpmReportOut,
+    PangleCallbackOut,
+    TestGrantOut,
+)
 
 logger = logging.getLogger("shagua.ad")
 
@@ -95,6 +102,30 @@ def reward_status(user: CurrentUser, db: DbSession) -> AdRewardStatusOut:
     )
 
 
+@router.post(
+    "/ecpm-report",
+    response_model=EcpmReportOut,
+    summary="上报本次广告展示的 eCPM(内部收益统计)",
+    dependencies=[Depends(rate_limit(120, 60, "ad-ecpm-report"))],
+)
+def ecpm_report(payload: EcpmReportIn, user: CurrentUser, db: DbSession) -> EcpmReportOut:
+    """客户端在广告展示后(onAdShow 读 getShowEcpm)上报 eCPM,落库做内部收益统计/对账。
+
+    Bearer 鉴权,user_id 取自 JWT(不信 body)。best-effort:落库即 ok,客户端 fire-and-forget,
+    丢一两条不影响业务(穿山甲后台报表是结算权威)。eCPM 与发奖(S2S)是两条独立流,不逐条关联。
+    """
+    crud_ecpm.create_ecpm_record(
+        db, user.id,
+        ad_type=payload.ad_type, ecpm_raw=payload.ecpm,
+        adn=payload.adn, slot_id=payload.slot_id,
+    )
+    logger.info(
+        "ad ecpm report user_id=%d type=%s ecpm=%s adn=%s slot=%s",
+        user.id, payload.ad_type, payload.ecpm, payload.adn, payload.slot_id,
+    )
+    return EcpmReportOut(ok=True)
+
+
 @router.post(
     "/test-grant",
     response_model=TestGrantOut,

app/core/ad_cooldown.py

@@ -0,0 +1,54 @@
+"""看激励视频冷却策略 —— 与发奖记录查询解耦的纯计算。
+
+当前策略:**每 N 次一轮,看满一轮后强制冷却若干秒**(N / 秒数 取自 [rewards] 常量)。
+[repositories.ad_reward.today_status] 只负责取数据(今日 granted 的 created_at 列表),
+把"本轮已看几次 + 冷却到几点"的策略判断委托到这里。
+
+⚠️ 这是临时策略,后续要调。换策略(间隔式 / 每日配额式 / 指数退避 …)**只改本文件**,
+repository 不碰——这就是把它独立出来的目的。保持 [compute_cooldown] 签名稳定即可。
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime, timedelta, timezone
+
+from app.core.rewards import VIDEO_ROUND_COOLDOWN_SECONDS, VIDEO_ROUND_REQUIRED_COUNT
+
+
+@dataclass(frozen=True)
+class CooldownState:
+    """冷却策略的输出。"""
+
+    round_count: int  # 本轮已看次数 0..N-1(展示用)
+    cooldown_until: datetime | None  # 本轮冷却结束时间(UTC);None = 不在冷却
+
+
+def compute_cooldown(
+    granted_times_desc: list[datetime],
+    now: datetime,
+    *,
+    round_size: int = VIDEO_ROUND_REQUIRED_COUNT,
+    cooldown_seconds: int = VIDEO_ROUND_COOLDOWN_SECONDS,
+) -> CooldownState:
+    """按"每 round_size 次一轮、看满一轮后冷却 cooldown_seconds 秒"算本轮进度 + 冷却结束时间。
+
+    :param granted_times_desc: 今日 status=granted 记录的 created_at,**按时间倒序**(最新在前)。
+    :param now: 当前时间(UTC,带 tzinfo),用于判断冷却是否已过。
+    :param round_size / cooldown_seconds: 策略参数,默认取 rewards 常量,可注入便于测试/调参。
+
+    纯函数,不碰 DB。冷却派生算法:把今日 granted 倒序,跳过当前未完成轮的 round_count 条,
+    下一条即"上一轮最后一次"的时间,+ cooldown_seconds 仍 > now 则在冷却中。
+    SQLite 上 created_at 可能是 naive,按 UTC 解读再比较。
+    """
+    used = len(granted_times_desc)
+    round_count = used % round_size
+    cooldown_until: datetime | None = None
+    if used >= round_size:
+        # round_count 必 < round_size <= used,索引合法
+        last_round_end = granted_times_desc[round_count]
+        if last_round_end.tzinfo is None:
+            last_round_end = last_round_end.replace(tzinfo=timezone.utc)
+        cd_end = last_round_end + timedelta(seconds=cooldown_seconds)
+        if cd_end > now:
+            cooldown_until = cd_end
+    return CooldownState(round_count=round_count, cooldown_until=cooldown_until)

app/models/__init__.py

@@ -1,4 +1,5 @@
 """所有 ORM model 必须在这里 import 一次,Alembic / metadata 才能扫到。"""
+from app.models.ad_ecpm import AdEcpmRecord  # noqa: F401
 from app.models.ad_reward import AdRewardRecord  # noqa: F401
 from app.models.feedback import Feedback  # noqa: F401
 from app.models.savings import SavingsRecord  # noqa: F401

app/models/ad_ecpm.py

@@ -0,0 +1,49 @@
+"""广告展示 eCPM 上报记录(内部收益统计/对账)。
+
+每条 = 客户端一次广告展示(`onAdShow`)后读到的 eCPM 信息。和发奖记录
+[ad_reward.AdRewardRecord] 是**两条独立的数据流**:
+- 发奖走穿山甲 S2S 回调(后端 → 有 trans_id、无 ecpm);
+- eCPM 走客户端上报(客户端 → 有 ecpm、无 trans_id)。
+两者没有公共键,无法逐条一一对应,所以本表用于**按用户/按天聚合收益**口径的对账,
+不做"这条发奖 = 这条 ecpm"的精确关联。穿山甲后台报表才是结算权威,本表是细粒度补充。
+
+⚠️ `ecpm_raw` 原样存客户端上报的字符串——eCPM 单位(分 / 元)截至 2026-05-31 尚未最终确认,
+确认后再加一列解析好的数值;在此之前对账按"待定单位"处理。
+"""
+from __future__ import annotations
+
+from datetime import datetime
+
+from sqlalchemy import DateTime, ForeignKey, Integer, String, func
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.db.base import Base
+
+
+class AdEcpmRecord(Base):
+    __tablename__ = "ad_ecpm_record"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    user_id: Mapped[int] = mapped_column(
+        Integer, ForeignKey("user.id"), index=True, nullable=False
+    )
+    # 广告类型:reward_video(激励视频) / draw(Draw 信息流) 等;不强行统一代码位,各类型各自上报
+    ad_type: Mapped[str] = mapped_column(String(32), nullable=False)
+    # 实际投放的 ADN(穿山甲 getShowEcpm().getSdkName(),如 pangle / gdt)
+    adn: Mapped[str | None] = mapped_column(String(32), nullable=True)
+    # 实际展示用的代码位(底层 mediation rit,非客户端配置位)
+    slot_id: Mapped[str | None] = mapped_column(String(64), nullable=True)
+    # 客户端上报的 eCPM 原始字符串(单位待确认,原样存)
+    ecpm_raw: Mapped[str] = mapped_column(String(32), nullable=False)
+    # 北京时间日期串 'YYYY-MM-DD',按它等值做"按天聚合"(不在 SQL 里做跨时区 date 比较)
+    report_date: Mapped[str] = mapped_column(String(10), index=True, nullable=False)
+
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now(), index=True, nullable=False
+    )
+
+    def __repr__(self) -> str:  # pragma: no cover
+        return (
+            f"<AdEcpmRecord user_id={self.user_id} {self.ad_type} "
+            f"ecpm={self.ecpm_raw} adn={self.adn}>"
+        )

app/models/savings.py

@@ -8,7 +8,7 @@ from __future__ import annotations
 
 from datetime import datetime
 
-from sqlalchemy import DateTime, ForeignKey, Integer, String, func
+from sqlalchemy import JSON, DateTime, ForeignKey, Integer, String, func
 from sqlalchemy.dialects.postgresql import JSONB
 from sqlalchemy.orm import Mapped, mapped_column
 
@@ -29,8 +29,12 @@ class SavingsRecord(Base):
     title: Mapped[str | None] = mapped_column(String(128), nullable=True)
     # 店铺名(订单明细卡标题,如「窑鸡王（王府井店）」)
     shop_name: Mapped[str | None] = mapped_column(String(128), nullable=True)
-    # 菜品名列表(JSONB),PG 上可建 GIN 索引,前 2 道直接展示,其余收进「还有 N 道菜」展开。
-    dishes: Mapped[list[str]] = mapped_column(JSONB, nullable=False, default=list)
+    # 菜品名列表。PG 上用 JSONB(可建 GIN 索引),SQLite(本地/测试)退化为通用 JSON——
+    # 否则 SQLite 无 visit_JSONB,Base.metadata.create_all 编译报错(测试套件 setup 全崩)。
+    # 前 2 道直接展示,其余收进「还有 N 道菜」展开。
+    dishes: Mapped[list[str]] = mapped_column(
+        JSON().with_variant(JSONB(), "postgresql"), nullable=False, default=list
+    )
     # 来源:demo(演示) / compare(真实比价上报)
     source: Mapped[str] = mapped_column(String(16), nullable=False, default="compare")
 

app/repositories/ad_ecpm.py

@@ -0,0 +1,49 @@
+"""广告 eCPM 上报 CRUD(内部收益统计/对账)。
+
+客户端在广告展示后(onAdShow)读到 eCPM,经鉴权接口上报,这里落库。鉴权接口已确保
+user 存在(JWT),故不做 UnknownUser 校验。best-effort 上报:丢一两条不影响业务,
+穿山甲后台报表是结算权威兜底。
+"""
+from __future__ import annotations
+
+from sqlalchemy import func, select
+from sqlalchemy.orm import Session
+
+from app.core.rewards import cn_today
+from app.models.ad_ecpm import AdEcpmRecord
+
+
+def create_ecpm_record(
+    db: Session,
+    user_id: int,
+    *,
+    ad_type: str,
+    ecpm_raw: str,
+    adn: str | None = None,
+    slot_id: str | None = None,
+) -> AdEcpmRecord:
+    """落一条 eCPM 上报记录。report_date 用北京时间当天,供按天聚合。"""
+    rec = AdEcpmRecord(
+        user_id=user_id,
+        ad_type=ad_type,
+        adn=adn,
+        slot_id=slot_id,
+        ecpm_raw=ecpm_raw,
+        report_date=cn_today().isoformat(),
+    )
+    db.add(rec)
+    db.commit()
+    db.refresh(rec)
+    return rec
+
+
+def count_today(db: Session, user_id: int) -> int:
+    """该用户今日(北京时间)上报的 eCPM 条数,排查/对账辅助用。"""
+    return db.execute(
+        select(func.count())
+        .select_from(AdEcpmRecord)
+        .where(
+            AdEcpmRecord.user_id == user_id,
+            AdEcpmRecord.report_date == cn_today().isoformat(),
+        )
+    ).scalar_one()

app/repositories/ad_reward.py

@@ -10,17 +10,16 @@
 """
 from __future__ import annotations
 
-from datetime import datetime, timedelta, timezone
+from datetime import datetime, timezone
 
 from sqlalchemy import func, select
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
 
+from app.core.ad_cooldown import compute_cooldown
 from app.core.rewards import (
     AD_REWARD_COIN,
     DAILY_AD_REWARD_LIMIT,
-    VIDEO_ROUND_COOLDOWN_SECONDS,
-    VIDEO_ROUND_REQUIRED_COUNT,
     cn_today,
 )
 from app.repositories import wallet as crud_wallet
@@ -110,26 +109,19 @@ def _commit_record(db: Session, rec: AdRewardRecord, trans_id: str) -> AdRewardR
     return rec
 
 
-def _last_completed_round_end_at(
-    db: Session, user_id: int, reward_date: str, round_count: int
-) -> datetime | None:
-    """当日 granted 记录中**最近一个已完成轮**末尾那次的 created_at。
-
-    思路:把当日 granted 按时间倒序排,跳过当前未完成轮的 round_count 条,下一条
-    即"上一轮最后一次"。round_count==0 且 used>=N 时跳 0 条直接取最近一条。
-    used<N(还没完成第一轮)调用方应直接判 None,不进这里。
-    """
-    return db.execute(
-        select(AdRewardRecord.created_at)
-        .where(
-            AdRewardRecord.user_id == user_id,
-            AdRewardRecord.reward_date == reward_date,
-            AdRewardRecord.status == "granted",
-        )
-        .order_by(AdRewardRecord.created_at.desc())
-        .offset(round_count)
-        .limit(1)
-    ).scalar_one_or_none()
+def _granted_times_today_desc(db: Session, user_id: int, reward_date: str) -> list[datetime]:
+    """当日 status=granted 记录的 created_at,按时间倒序(最新在前)——冷却策略的输入数据。"""
+    return list(
+        db.execute(
+            select(AdRewardRecord.created_at)
+            .where(
+                AdRewardRecord.user_id == user_id,
+                AdRewardRecord.reward_date == reward_date,
+                AdRewardRecord.status == "granted",
+            )
+            .order_by(AdRewardRecord.created_at.desc())
+        ).scalars()
+    )
 
 
 def today_status(
@@ -138,21 +130,16 @@ def today_status(
     """客户端查"今日看广告发奖"进度。
 
     返回 (今日已发次数, 每日上限, 单次金币, 本轮已看次数, 本轮冷却结束时间(UTC))。
-    - round_count = used % VIDEO_ROUND_REQUIRED_COUNT,展示用(0..N-1)
-    - cooldown_until 计算:取最近一个已完成轮末尾的 created_at + 10 min;若仍 > now 则返回,
-      否则返回 None。冷却 = UX 约束(客户端 CTA 倒计时不可点),后端发奖逻辑不受影响。
+    本函数只**取数据**(今日 granted 的 created_at 倒序),把"本轮已看几次 + 冷却到几点"的
+    **策略判断**委托给 [app.core.ad_cooldown.compute_cooldown](纯函数)——换冷却策略只动那个模块。
     """
     today = cn_today().isoformat()
-    used = _granted_today(db, user_id, today)
-    round_count = used % VIDEO_ROUND_REQUIRED_COUNT
-    cooldown_until: datetime | None = None
-    if used >= VIDEO_ROUND_REQUIRED_COUNT:
-        last_end = _last_completed_round_end_at(db, user_id, today, round_count)
-        if last_end is not None:
-            # SQLAlchemy 在 SQLite 上拿到的 created_at 可能是 naive,统一按 UTC 解读再比较
-            if last_end.tzinfo is None:
-                last_end = last_end.replace(tzinfo=timezone.utc)
-            cd_end = last_end + timedelta(seconds=VIDEO_ROUND_COOLDOWN_SECONDS)
-            if cd_end > datetime.now(timezone.utc):
-                cooldown_until = cd_end
-    return used, DAILY_AD_REWARD_LIMIT, AD_REWARD_COIN, round_count, cooldown_until
+    granted_desc = _granted_times_today_desc(db, user_id, today)
+    state = compute_cooldown(granted_desc, datetime.now(timezone.utc))
+    return (
+        len(granted_desc),
+        DAILY_AD_REWARD_LIMIT,
+        AD_REWARD_COIN,
+        state.round_count,
+        state.cooldown_until,
+    )

app/schemas/ad.py

@@ -39,6 +39,24 @@ class AdRewardStatusOut(BaseModel):
     )
 
 
+class EcpmReportIn(BaseModel):
+    """客户端上报一次广告展示的 eCPM(内部收益统计/对账)。
+
+    user_id 不在 body 里——由 JWT 取(Bearer),防伪造。ecpm 原样上报字符串(单位待确认)。
+    """
+
+    ad_type: str = Field(..., description="广告类型:reward_video(激励视频) / draw(Draw 信息流) 等")
+    ecpm: str = Field(..., description="穿山甲 getShowEcpm().getEcpm() 原始字符串,单位待确认,原样上报")
+    adn: str | None = Field(None, description="实际投放 ADN(getSdkName),如 pangle")
+    slot_id: str | None = Field(None, description="实际展示代码位(底层 mediation rit)")
+
+
+class EcpmReportOut(BaseModel):
+    """eCPM 上报结果。best-effort,落库即 ok。"""
+
+    ok: bool = True
+
+
 class TestGrantOut(BaseModel):
     """[仅本地联调]模拟发奖结果。带上今日进度,客户端可直接据此刷新展示。"""
 

docs/api/README.md

@@ -52,6 +52,7 @@
 | 32 | `GET /api/v1/ad/pangle-callback` | 验签 | [详情](./ad-pangle-callback.md) |
 | 33 | `GET /api/v1/ad/reward-status` | Bearer | [详情](./ad-reward-status.md) |
 | 34 | `POST /api/v1/ad/test-grant` | Bearer | [详情](./ad-test-grant.md) |
+| 35 | `POST /api/v1/ad/ecpm-report` | Bearer | [详情](./ad-ecpm-report.md) |
 | **用户资料**（前缀 `/api/v1/user`） |||
 | 35 | `PATCH /api/v1/user/profile` | Bearer | [详情](./user-profile.md) |
 | 36 | `POST /api/v1/user/avatar` | Bearer | [详情](./user-avatar.md) |

docs/api/ad-ecpm-report.md

@@ -0,0 +1,36 @@
+# POST /api/v1/ad/ecpm-report — 上报本次广告展示的 eCPM（内部收益统计）
+
+> 所属:Ad 组（前缀 `/api/v1/ad`） | 鉴权:Bearer | [← 返回 API 索引](./README.md)
+
+## 入参
+请求体:`EcpmReportIn`
+
+| 字段 | 类型 | 必填 | 说明 |
+|---|---|---|---|
+| `ad_type` | str | 是 | 广告类型:`reward_video`(激励视频) / `draw`(Draw 信息流) 等 |
+| `ecpm` | str | 是 | 穿山甲 `getShowEcpm().getEcpm()` 原始字符串,**单位待确认(分/元)**,原样上报 |
+| `adn` | str\|null | 否 | 实际投放 ADN(`getSdkName`),如 `pangle` |
+| `slot_id` | str\|null | 否 | 实际展示代码位(底层 mediation rit,非客户端配置位) |
+
+`user_id` 不在 body 里——由 JWT 取(Bearer),防伪造。
+
+## 出参
+响应 `200`:`EcpmReportOut`
+
+| 字段 | 类型 | 说明 |
+|---|---|---|
+| `ok` | bool | 落库即 `true` |
+
+## 说明
+客户端在广告**展示后**(`onAdShow` 读 `getShowEcpm()`)调用,把本次展示的 eCPM 落库做**内部收益统计/对账**。
+
+- **best-effort**:客户端 fire-and-forget,丢一两条不影响业务;穿山甲后台报表是结算权威兜底。
+- 落 `ad_ecpm_record` 表,`report_date` 用北京时间当天,供「按用户/按天聚合」对账。
+- **与发奖是两条独立流**:发奖走 [ad-pangle-callback](./ad-pangle-callback.md)(穿山甲 S2S,有 `trans_id`、无 ecpm),本接口客户端上报(有 ecpm、无 `trans_id`)。两者无公共键,**不逐条一一对应**,只做按用户/按天聚合口径。
+- eCPM 是**每千次展示预估**,单条展示预估收益 ≈ `ecpm / 1000`,且为客户端预估口径,非最终结算。
+- ⚠️ eCPM **单位(分/元)截至 2026-05-31 未最终确认**,故 `ecpm_raw` 原样存字符串;确认后再加一列解析好的数值。
+- 限流:同 IP 120 次/分钟。
+
+## 相关
+- [ad-pangle-callback](./ad-pangle-callback.md) — 穿山甲 S2S 发奖回调
+- [ad-reward-status](./ad-reward-status.md) — 今日看广告发奖进度