"""看激励视频发奖 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, )