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

"""信息流广告奖励 CRUD。

点位 2:每展示满 10 秒累计一份奖励,视频完成后一次性入账。当前一期由客户端在
完成回调后上报;后续若 SDK/S2S 能提供更强确认信号,可继续复用本表的幂等键。
"""
from __future__ import annotations

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

from app.core import rewards
from app.core.rewards import cn_today
from app.models.ad_feed_reward import AdFeedRewardRecord
from app.repositories import wallet as crud_wallet


FEED_REWARD_UNIT_SECONDS = 10
# 单个 feed 事件的时长上限(秒):一期 duration_seconds 由客户端上报,伪造超长时长会刷份数
# (每 10 秒 1 份)。真实单条信息流视频远小于此;取 120s=12 份封顶,挡刷量、不影响正规单。
# 与 rewards.AD_ECPM_MAX_FEN(eCPM 钳顶)合起来,把单事件可铸金币锁进有限区间。
FEED_MAX_DURATION_SECONDS = 120


def _find_by_event(db: Session, client_event_id: str) -> AdFeedRewardRecord | None:
    return db.execute(
        select(AdFeedRewardRecord).where(
            AdFeedRewardRecord.client_event_id == client_event_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(AdFeedRewardRecord)
        .where(
            AdFeedRewardRecord.user_id == user_id,
            AdFeedRewardRecord.reward_date == reward_date,
            AdFeedRewardRecord.status == "granted",
        )
    ).scalar_one()


def _unit_reward_total(db: Session, user_id: int, ecpm: str, unit_count: int) -> int:
    """按每个 10 秒单位逐份计算奖励,LT 使用**账号累计**奖励份序号(不按天重置)。"""
    if unit_count <= 0:
        return 0
    existing_units = db.execute(
        select(func.coalesce(func.sum(AdFeedRewardRecord.unit_count), 0))
        .where(
            AdFeedRewardRecord.user_id == user_id,
            AdFeedRewardRecord.status == "granted",
        )
    ).scalar_one()
    total = 0
    for offset in range(1, unit_count + 1):
        total += rewards.calculate_ad_reward_coin(ecpm, int(existing_units) + offset)
    return total


def grant_feed_reward(
    db: Session,
    user_id: int,
    *,
    client_event_id: str,
    ecpm: str,
    duration_seconds: int,
    ad_session_id: str | None = None,
    adn: str | None = None,
    slot_id: str | None = None,
    app_env: str | None = None,
    our_code_id: str | None = None,
    aborted: bool = False,
) -> AdFeedRewardRecord:
    """比价/领券一整场信息流(轮播多条)结束后结算奖励。client_event_id 幂等,同号重试不重复发。

    发奖规则:**比价全程不关广告才发**,金额按整场**总观看时长**折份(每 10 秒 1 份)。
    - aborted=True(用户中途 ✕ 关闭):整场不发,记 status='closed_early' 留痕(原因可查)。
    - 总时长不足 10 秒(unit_count==0):记 status='too_short' 不发。
    - 命中当日条数上限:记 status='capped' 不发。
    duration_seconds 是整场累计秒数。一期 eCPM/时长均由客户端上报,故服务端两道硬闸防刷:时长钳到
    FEED_MAX_DURATION_SECONDS 限单场份数,eCPM 在 rewards.calculate_ad_reward_coin 内钳到
    AD_ECPM_MAX_FEN 限单份金额;叠加每日 get_ad_daily_limit 条数上限,把单用户日产出锁进有限区间。
    """
    existing = _find_by_event(db, client_event_id)
    if existing is not None:
        return existing

    today = cn_today().isoformat()
    # 客户端上报时长先钳到 FEED_MAX_DURATION_SECONDS,防伪造超长时长刷份数(见常量注释)。
    safe_duration = max(0, min(duration_seconds, FEED_MAX_DURATION_SECONDS))
    unit_count = safe_duration // FEED_REWARD_UNIT_SECONDS

    # 用户中途关闭广告:整场不发(全程不关才发),留一条 closed_early 记录原因。优先级最高。
    if aborted:
        rec = AdFeedRewardRecord(
            client_event_id=client_event_id,
            user_id=user_id,
            reward_date=today,
            duration_seconds=safe_duration,
            unit_count=unit_count,
            ad_session_id=ad_session_id,
            ecpm_raw=ecpm,
            adn=adn,
            slot_id=slot_id,
            app_env=app_env,
            our_code_id=our_code_id,
            coin=0,
            status="closed_early",
        )
        return _commit_record(db, rec, client_event_id)

    if _granted_today(db, user_id, today) >= rewards.get_ad_daily_limit(db):
        rec = AdFeedRewardRecord(
            client_event_id=client_event_id,
            user_id=user_id,
            reward_date=today,
            duration_seconds=safe_duration,
            unit_count=unit_count,
            ad_session_id=ad_session_id,
            ecpm_raw=ecpm,
            adn=adn,
            slot_id=slot_id,
            app_env=app_env,
            our_code_id=our_code_id,
            coin=0,
            status="capped",
        )
        return _commit_record(db, rec, client_event_id)

    # 整场总时长不足 10 秒,凑不满一份 → 不发,记 too_short 留痕。
    if unit_count == 0:
        rec = AdFeedRewardRecord(
            client_event_id=client_event_id,
            user_id=user_id,
            reward_date=today,
            duration_seconds=safe_duration,
            unit_count=0,
            ad_session_id=ad_session_id,
            ecpm_raw=ecpm,
            adn=adn,
            slot_id=slot_id,
            app_env=app_env,
            our_code_id=our_code_id,
            coin=0,
            status="too_short",
        )
        return _commit_record(db, rec, client_event_id)

    coin = _unit_reward_total(db, user_id, ecpm, unit_count)
    if coin > 0:
        crud_wallet.grant_coins(
            db, user_id, coin,
            biz_type="feed_ad_reward", ref_id=client_event_id,
            remark=f"信息流广告奖励 {unit_count}份",
        )
    rec = AdFeedRewardRecord(
        client_event_id=client_event_id,
        user_id=user_id,
        reward_date=today,
        duration_seconds=safe_duration,
        unit_count=unit_count,
        ad_session_id=ad_session_id,
        ecpm_raw=ecpm,
        adn=adn,
        slot_id=slot_id,
        app_env=app_env,
        our_code_id=our_code_id,
        coin=coin,
        status="granted",
    )
    return _commit_record(db, rec, client_event_id)


def _commit_record(db: Session, rec: AdFeedRewardRecord, client_event_id: str) -> AdFeedRewardRecord:
    db.add(rec)
    try:
        db.commit()
    except IntegrityError:
        db.rollback()
        existing = _find_by_event(db, client_event_id)
        if existing is not None:
            return existing
        raise
    db.refresh(rec)
    return rec