shaguabijia-app-server/app/repositories/ad_reward.py

"""看激励视频发奖 CRUD。

发奖三道闸(仿提现的资金安全思路):
  1. 验签不过 → API 层直接拒,不进这里。
  2. trans_id 唯一 → 同一交易号二次回调不重复发(穿山甲会重试)。
  3. 当日发奖次数 ≥ 上限 → 记一条 status='capped' 但不发金币。

普通福利页激励视频按 eCPM 公式发金币;缺 eCPM 时记 status='ecpm_missing' 但不发金币。
签到膨胀等其他激励视频场景复用本表记录 S2S 幂等,实际发币由各自业务仓储完成。
"""
from __future__ import annotations

from datetime import datetime, timezone

from sqlalchemy import func, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session

from app.core import rewards
from app.core.ad_cooldown import compute_cooldown
from app.core.rewards import DAILY_AD_WATCH_SECONDS_LIMIT, cn_today
from app.models.ad_reward import AdRewardRecord
from app.models.user import User
from app.repositories import ad_ecpm as crud_ecpm
from app.repositories import wallet as crud_wallet
from app.repositories.ad_watch import watched_seconds_today


class UnknownUserError(Exception):
    """回调里的 user_id 不存在(可能是伪造)。"""


def _find_by_trans(db: Session, trans_id: str) -> AdRewardRecord | None:
    return db.execute(
        select(AdRewardRecord).where(AdRewardRecord.trans_id == trans_id)
    ).scalar_one_or_none()


def find_by_trans(db: Session, trans_id: str) -> AdRewardRecord | None:
    """按 S2S 交易号查询处理记录,供路由在分场景前做幂等短路。"""
    return _find_by_trans(db, trans_id)


def _granted_today(db: Session, user_id: int, reward_date: str) -> int:
    return db.execute(
        select(func.count())
        .select_from(AdRewardRecord)
        .where(
            AdRewardRecord.user_id == user_id,
            AdRewardRecord.reward_date == reward_date,
            AdRewardRecord.status == "granted",
            AdRewardRecord.reward_scene == "reward_video",
        )
    ).scalar_one()


def _granted_cumulative(db: Session, user_id: int) -> int:
    """账号累计已发奖的看视频次数(不按天重置)——LT 因子(因子2)用它定"第 N 次广告"。
    与 _granted_today 区别仅在去掉 reward_date 过滤;每日次数上限/冷却仍按当日统计。"""
    return db.execute(
        select(func.count())
        .select_from(AdRewardRecord)
        .where(
            AdRewardRecord.user_id == user_id,
            AdRewardRecord.status == "granted",
            AdRewardRecord.reward_scene == "reward_video",
        )
    ).scalar_one()


def grant_ad_reward(
    db: Session,
    user_id: int,
    trans_id: str,
    *,
    ecpm: str | None = None,
    ad_session_id: str | None = None,
    reward_scene: str = "reward_video",
    reward_name: str | None = None,
    raw: str | None = None,
) -> AdRewardRecord:
    """福利页激励视频发奖(幂等 + 每日限额 + eCPM 公式)。"""
    # #2 幂等:同 trans_id 已处理过 → 原样返回,不重复发
    existing = _find_by_trans(db, trans_id)
    if existing is not None:
        return existing

    if db.get(User, user_id) is None:
        raise UnknownUserError

    today = cn_today().isoformat()

    # 按 ad_session_id 匹配客户端 eCPM 上报:既用于缺 eCPM 时回退取值,也把「来源」
    # (我们的应用 app_env + 我们配置的代码位 our_code_id)回填到发奖记录,供广告收益报表聚合。
    # S2S 回调本身不带这俩;查不到(未上报 eCPM)则留空。
    ecpm_rec = (
        crud_ecpm.find_by_session(db, user_id=user_id, ad_session_id=ad_session_id)
        if ad_session_id else None
    )
    src_app_env = ecpm_rec.app_env if ecpm_rec is not None else None
    src_code_id = ecpm_rec.our_code_id if ecpm_rec is not None else None

    # #3 每日上限:当前产品只保留发奖次数上限(默认 500 次)。旧的观看时长闸保留字段,
    # 但 DAILY_AD_WATCH_SECONDS_LIMIT=0 时视为停用,不能命中 capped。
    over_time = (
        DAILY_AD_WATCH_SECONDS_LIMIT > 0
        and watched_seconds_today(db, user_id, today=today) >= DAILY_AD_WATCH_SECONDS_LIMIT
    )
    over_count = _granted_today(db, user_id, today) >= rewards.get_ad_daily_limit(db)
    if over_time or over_count:
        rec = AdRewardRecord(
            trans_id=trans_id, user_id=user_id, coin=0, status="capped",
            reward_date=today, reward_name=reward_name, raw=raw,
            reward_scene=reward_scene, ad_session_id=ad_session_id, ecpm_raw=ecpm,
            app_env=src_app_env, our_code_id=src_code_id,
        )
        return _commit_record(db, rec, trans_id)

    ecpm_raw = ecpm or (ecpm_rec.ecpm_raw if ecpm_rec is not None else None)

    if not ecpm_raw:
        rec = AdRewardRecord(
            trans_id=trans_id, user_id=user_id, coin=0, status="ecpm_missing",
            reward_date=today, reward_name=reward_name, raw=raw,
            reward_scene=reward_scene, ad_session_id=ad_session_id, ecpm_raw=None,
            app_env=src_app_env, our_code_id=src_code_id,
        )
        return _commit_record(db, rec, trans_id)

    # LT 因子(因子2)按"账号累计第 N 次看广告"递减(2.0→1.0),不再按天重置;
    # 每日次数上限/冷却仍按当日统计(over_count 用 _granted_today)。
    coin = rewards.calculate_ad_reward_coin(ecpm_raw, _granted_cumulative(db, user_id) + 1)

    # 发金币 + 记一笔,同事务
    crud_wallet.grant_coins(
        db, user_id, coin,
        biz_type="reward_video", ref_id=trans_id, remark="看视频奖励金币",
    )
    rec = AdRewardRecord(
        trans_id=trans_id, user_id=user_id, coin=coin, status="granted",
        reward_date=today, reward_name=reward_name, raw=raw,
        reward_scene=reward_scene, ad_session_id=ad_session_id, ecpm_raw=ecpm_raw,
        app_env=src_app_env, our_code_id=src_code_id,
    )
    return _commit_record(db, rec, trans_id)


def record_reward_noshow(
    db: Session,
    user_id: int,
    *,
    ad_session_id: str,
    ecpm: str | None = None,
    adn: str | None = None,
    slot_id: str | None = None,
    app_env: str | None = None,
    our_code_id: str | None = None,
    reward_scene: str = "reward_video",
) -> AdRewardRecord:
    """客户端上报「激励视频展示了但用户提前关/跳过、未触发发奖」,落一条 coin=0 status='closed_early'
    记录,供广告收益报表把「不发金币的原因」也呈现出来(只留痕,不发币)。

    幂等键 trans_id = 'noreward:{ad_session_id}'(每次展示唯一)。若同一 ad_session_id 已有 granted
    记录(S2S 已发奖,正常路径),说明用户其实看完了 → 跳过不写、原样返回那条,避免与正常发奖重复。
    app_env/our_code_id 由客户端直接带上(它本就持有);查不到 user 抛 UnknownUserError。
    """
    if db.get(User, user_id) is None:
        raise UnknownUserError
    # 同一次展示已正常发奖 → 不再记 closed_early(防与 S2S granted 重复)
    granted = db.execute(
        select(AdRewardRecord)
        .where(
            AdRewardRecord.user_id == user_id,
            AdRewardRecord.ad_session_id == ad_session_id,
            AdRewardRecord.status == "granted",
        )
        .limit(1)
    ).scalar_one_or_none()
    if granted is not None:
        return granted

    trans_id = f"noreward:{ad_session_id}"
    existing = _find_by_trans(db, trans_id)
    if existing is not None:
        return existing

    rec = AdRewardRecord(
        trans_id=trans_id, user_id=user_id, coin=0, status="closed_early",
        reward_date=cn_today().isoformat(), reward_name=None, raw=None,
        reward_scene=reward_scene, ad_session_id=ad_session_id, ecpm_raw=ecpm,
        app_env=app_env, our_code_id=our_code_id,
    )
    return _commit_record(db, rec, trans_id)


def record_external_reward(
    db: Session,
    user_id: int,
    trans_id: str,
    *,
    coin: int,
    reward_scene: str,
    ad_session_id: str | None = None,
    ecpm: str | None = None,
    reward_name: str | None = None,
    raw: str | None = None,
    status: str = "granted",
    commit: bool = True,
) -> AdRewardRecord:
    """记录非普通看视频场景的 S2S 回调幂等,发币由调用方业务仓储完成。"""
    existing = _find_by_trans(db, trans_id)
    if existing is not None:
        return existing
    if db.get(User, user_id) is None:
        raise UnknownUserError
    rec = AdRewardRecord(
        trans_id=trans_id,
        user_id=user_id,
        coin=coin,
        status=status,
        reward_date=cn_today().isoformat(),
        reward_name=reward_name,
        raw=raw,
        reward_scene=reward_scene,
        ad_session_id=ad_session_id,
        ecpm_raw=ecpm,
    )
    db.add(rec)
    if commit:
        try:
            db.commit()
        except IntegrityError:
            db.rollback()
            existing = _find_by_trans(db, trans_id)
            if existing is not None:
                return existing
            raise
        db.refresh(rec)
    else:
        db.flush()
    return rec


def _commit_record(db: Session, rec: AdRewardRecord, trans_id: str) -> AdRewardRecord:
    """提交发奖记录;并发下同 trans_id 撞唯一约束时回滚并返回已存在的那条(幂等兜底)。"""
    db.add(rec)
    try:
        db.commit()
    except IntegrityError:
        db.rollback()
        existing = _find_by_trans(db, trans_id)
        if existing is not None:
            return existing
        raise
    db.refresh(rec)
    return rec


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",
                AdRewardRecord.reward_scene == "reward_video",
            )
            .order_by(AdRewardRecord.created_at.desc())
        ).scalars()
    )


def today_status(
    db: Session, user_id: int
) -> tuple[int, int, int, int, datetime | None, int, int]:
    """客户端查"今日看广告发奖"进度。

    返回 (今日已发次数, 每日次数上限, 单次金币, 本轮已看次数, 本轮冷却结束时间(UTC),
          今日已观看总秒数, 每日观看总时长上限(秒))。
    次数上限/单次金币/每轮次数/冷却秒 均从 app_config 读(运营后台可改);次数维度的"本轮已看
    几次 + 冷却到几点"委托 [app.core.ad_cooldown.compute_cooldown](纯函数);观看时长字段保留给
    旧客户端兼容,当前 DAILY_AD_WATCH_SECONDS_LIMIT=0 表示不启用时长闸。
    """
    today = cn_today().isoformat()
    granted_desc = _granted_times_today_desc(db, user_id, today)
    state = compute_cooldown(
        granted_desc,
        datetime.now(timezone.utc),
        round_size=rewards.get_ad_round_count(db),
        cooldown_seconds=rewards.get_ad_cooldown_sec(db),
    )
    return (
        len(granted_desc),
        rewards.get_ad_daily_limit(db),
        0,
        state.round_count,
        state.cooldown_until,
        watched_seconds_today(db, user_id, today=today),
        DAILY_AD_WATCH_SECONDS_LIMIT,
    )