shaguabijia-app-server/app/core/rewards.py

"""福利 / 激励相关的业务常量与换算规则。

集中放在这里,业务代码引用,避免数值散落各处。这些是**产品规则**(签到奖励、
任务奖励、金币汇率)而非部署配置,所以不走环境变量;要调整直接改这里。
"""
from __future__ import annotations

from datetime import date, datetime, timedelta, timezone

# 业务时区:签到的"今天"按北京时间算,不能用 UTC。
# 否则 UTC+8 的凌晨 0~8 点会被算成 UTC 的前一天,导致签到日期错位。
CN_TZ = timezone(timedelta(hours=8))


def cn_today() -> date:
    """当前北京时间的日期,签到去重 / 连续判断都以此为准。"""
    return datetime.now(CN_TZ).date()


# ===== 签到:7 天循环,断签重置回第 1 天 =====
# 第 1→7 天的金币奖励,签到逻辑用 cycle_day(1..7)索引。
SIGNIN_REWARDS: tuple[int, ...] = (10, 20, 30, 50, 80, 120, 200)
SIGNIN_CYCLE_LEN: int = len(SIGNIN_REWARDS)


def signin_reward(cycle_day: int) -> int:
    """cycle_day 取值 1..SIGNIN_CYCLE_LEN。"""
    return SIGNIN_REWARDS[cycle_day - 1]


# ===== 金币 → 现金 汇率 =====
# 10000 金币 = 1 元 = 100 分。
COIN_PER_YUAN: int = 10000
CENTS_PER_YUAN: int = 100
# 1 分对应多少金币 = 100。兑换金币数必须是它的整数倍,否则会出现不足 1 分的零头。
COIN_PER_CENT: int = COIN_PER_YUAN // CENTS_PER_YUAN
# 单次兑换最少 1 元(避免大量 1 分级碎兑)
MIN_EXCHANGE_COIN: int = COIN_PER_YUAN


def coins_to_cents(coin_amount: int) -> int:
    """金币换算成现金(分)。调用前应已校验 coin_amount 是 COIN_PER_CENT 的整数倍。"""
    return coin_amount // COIN_PER_CENT


# ===== 提现(现金 → 微信零钱)=====
# 单次提现额度(分)。下限 0.1 元是微信商家转账的最低额,低于它(如 0.01 测试档)真转账会被拒。
WITHDRAW_MIN_CENTS: int = 10
WITHDRAW_MAX_CENTS: int = 5_000_000  # 5 万元


# ===== 一次性任务(领一次,user_task 去重)=====
TASK_ENABLE_NOTIFICATION = "enable_notification"

# task_key -> 奖励金币
# 打开消息提醒: 1000 金币(=¥0.1, 客户端原型展示口径; 量级与签到/里程碑相称)。
# 注意: 已不再 = 兑换下限(MIN_EXCHANGE_COIN=10000), test_exchange_flow 改走 grant_coins 直接供款。
TASK_REWARDS: dict[str, int] = {
    TASK_ENABLE_NOTIFICATION: 1000,
}


# ===== 比价战绩里程碑(累计成功比价 N 次,逐档解锁领金币)=====
# 第 1→6 次的金币奖励(1-based:第 N 次比价解锁第 N 档)。值沿用客户端原型档位。
# 数据源是 comparison_record 里 status='success' 的条数;每档领一次,
# comparison_milestone_claim 去重(仿一次性任务)。要调档位/金额直接改这里。
RECORD_MILESTONES: tuple[int, ...] = (120, 180, 300, 500, 800, 1200)
RECORD_MILESTONE_COUNT: int = len(RECORD_MILESTONES)


def record_milestone_reward(milestone: int) -> int:
    """第 milestone 档(1..RECORD_MILESTONE_COUNT)的金币。越界抛 IndexError。"""
    return RECORD_MILESTONES[milestone - 1]


# ===== 看激励视频发金币(穿山甲 S2S 服务端回调发奖)=====
# 看完一个激励视频发的金币(666 金币 ≈¥0.0666,汇率 10000 金币=1 元)。
# 作用:① 回调缺/坏 reward_amount 时的回退值;② 客户端进度接口展示的"单次预告金币";
#       ③ test-grant 本地联调的发奖额。
# 真实发放以穿山甲回调带回的 reward_amount 为准(见 resolve_ad_reward_coin),后台应把
# 代码位"奖励数量"配成与本值一致(=666),保证"广告内展示 / 进度预告 / 实际到账"三者一致。
AD_REWARD_COIN: int = 666
# 单次发奖金币上限:夹紧穿山甲回调里异常的 reward_amount(如后台多打一个 0),防刷爆余额。
MAX_AD_REWARD_COIN: int = 1000
# 每用户每日发奖次数上限——现作为"看广告时长上限"的**次数兜底**(防前端少报观看时长绕过时长闸)。
# 主闸是 DAILY_AD_WATCH_SECONDS_LIMIT(50 分钟);次数兜底要 ≥ 时长闸内的"合理最大次数",
# 否则会比主闸先掐、误伤看大量短广告的正常用户:50 分钟里若广告各 ~15s,理论可看 ~200 次,
# 故取 200(≈时长闸的次数等价),正常用户先撞 50 分钟时长闸,仅前端时长全报 0 等异常时本闸才兜住。
# 不要为"更严"而下调到 120 之类(会在 ~30 分钟就掐短广告用户);要更严应在客户端如实上报时长。
DAILY_AD_REWARD_LIMIT: int = 200

# ===== 看激励视频每日总时长上限(防刷主闸)=====
# 用户每天最多看 50 分钟激励视频,超了不发奖 / 不给看。时长由前端 onAdClose 上报真实观看
# 秒数累计(POST /api/v1/ad/watch-report → ad_watch_log 表 SUM)。前端不可信,故配合上面
# DAILY_AD_REWARD_LIMIT 次数兜底一起防刷。要调上限直接改这里。
DAILY_AD_WATCH_SECONDS_LIMIT: int = 50 * 60  # 3000 秒 = 50 分钟
# 单次上报观看时长的合理上限(夹紧异常值):激励视频一般 15~60 秒,超 120 秒按 120 记,防刷大值。
MAX_SINGLE_WATCH_SECONDS: int = 120
# 一轮看 N 次激励视频,看完一轮强制 10 分钟冷却(reward-status 派生 cooldown_until 给客户端,
# 客户端任务行 CTA 在冷却期间显示"MM:SS"且不可点)。冷却是 UX 约束,后端发奖只看 daily_limit,
# 冷却期间穿山甲回调仍照常发奖,不另设拒发分支。
VIDEO_ROUND_REQUIRED_COUNT: int = 3
VIDEO_ROUND_COOLDOWN_SECONDS: int = 600


def resolve_ad_reward_coin(db, reward_amount: str | int | None) -> int:  # noqa: ANN001
    """把穿山甲回调的 reward_amount(后台代码位配的"奖励数量")解析成本次发放金币。

    缺失 / 非数字 / ≤0 → 回退配置单次金币;超过配置单次上限 → 夹紧(均从 app_config 读)。
    注:reward_amount 不参与穿山甲 sign(只签 trans_id),但伪造回调需先伪造合法 sign
    (要 SecurityKey),故能过验签的 reward_amount 即视为可信,这里只防后台配错。
    """
    default_coin = get_ad_reward_coin(db)
    try:
        coin = int(reward_amount)  # type: ignore[arg-type]
    except (TypeError, ValueError):
        return default_coin
    if coin <= 0:
        return default_coin
    return min(coin, get_ad_max_coin(db))


# ===== 配置读取(运营后台可覆盖;app_config 表空时返回上面的常量默认 = 现状不变)=====
# 业务处用这些 getter(传 db)取值,而非直接用常量,这样 admin 改 app_config 即生效。
# 函数内延迟 import,避免 rewards ← app_config ← config_schema ← rewards 的模块级循环。

def _cfg(db, key: str):  # noqa: ANN001
    from app.repositories import app_config
    return app_config.get_value(db, key)


def get_signin_rewards(db) -> list[int]:  # noqa: ANN001
    return list(_cfg(db, "signin_rewards"))


def get_min_exchange_coin(db) -> int:  # noqa: ANN001
    return int(_cfg(db, "min_exchange_coin"))


def get_withdraw_min_cents(db) -> int:  # noqa: ANN001
    return int(_cfg(db, "withdraw_min_cents"))


def get_withdraw_max_cents(db) -> int:  # noqa: ANN001
    return int(_cfg(db, "withdraw_max_cents"))


def get_task_rewards(db) -> dict[str, int]:  # noqa: ANN001
    return dict(_cfg(db, "task_rewards"))


def get_record_milestones(db) -> list[int]:  # noqa: ANN001
    return list(_cfg(db, "record_milestones"))


def get_ad_reward_coin(db) -> int:  # noqa: ANN001
    return int(_cfg(db, "ad_reward_coin"))


def get_ad_daily_limit(db) -> int:  # noqa: ANN001
    return int(_cfg(db, "ad_daily_limit"))


def get_ad_max_coin(db) -> int:  # noqa: ANN001
    return int(_cfg(db, "ad_max_coin"))


def get_ad_round_count(db) -> int:  # noqa: ANN001
    return int(_cfg(db, "ad_round_count"))


def get_ad_cooldown_sec(db) -> int:  # noqa: ANN001
    return int(_cfg(db, "ad_cooldown_sec"))