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

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

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

发金币复用 `wallet.grant_coins`(biz_type='ad_reward', ref_id=trans_id),与发奖记录同事务,
保证"记一笔 + 加金币"原子化。
"""
from __future__ import annotations

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

from app.core.rewards import AD_REWARD_COIN, DAILY_AD_REWARD_LIMIT, cn_today
from app.repositories import wallet as crud_wallet
from app.models.ad_reward import AdRewardRecord
from app.models.user import User


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 _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",
        )
    ).scalar_one()


def grant_ad_reward(
    db: Session,
    user_id: int,
    trans_id: str,
    *,
    coin: int = AD_REWARD_COIN,
    reward_name: str | None = None,
    raw: str | None = None,
) -> AdRewardRecord:
    """看广告发奖(幂等 + 每日限额)。返回发奖记录(status=granted/capped)。

    coin 为本次发放金币(由调用方按穿山甲回调 reward_amount 解析,见
    rewards.resolve_ad_reward_coin);默认 AD_REWARD_COIN 供 test-grant / 缺省场景用。
    user_id 不存在抛 UnknownUserError(不建账户,防伪造 user_id 刷出脏数据)。
    """
    # #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()

    # #3 每日上限:超了记一条 capped(不发金币),让审计能看到"今天到顶了"
    if _granted_today(db, user_id, today) >= DAILY_AD_REWARD_LIMIT:
        rec = AdRewardRecord(
            trans_id=trans_id, user_id=user_id, coin=0, status="capped",
            reward_date=today, reward_name=reward_name, raw=raw,
        )
        return _commit_record(db, rec, trans_id)

    # 发金币 + 记一笔,同事务
    crud_wallet.grant_coins(
        db, user_id, coin,
        biz_type="ad_reward", 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,
    )
    return _commit_record(db, rec, trans_id)


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 today_status(db: Session, user_id: int) -> tuple[int, int, int]:
    """客户端查"今日看广告发奖"进度:返回 (今日已发次数, 每日上限, 单次金币)。"""
    used = _granted_today(db, user_id, cn_today().isoformat())
    return used, DAILY_AD_REWARD_LIMIT, AD_REWARD_COIN