xiaoxia-saas/packages/domain/points_rules.py

"""积分消耗规则配置 (#1895)

v1.6.1: 按产品决策，智能混剪/AI数字人/AI配音/抖音解析/改写/标题/封面 全部免费，
仅保留声音克隆合成（voice_clone_synth）的扣点逻辑；声音克隆训练保持免费。
爆款视频（viral_video）走动态定价，见本文件 VIRAL_VIDEO_MODEL_PRICES + calculate_viral_video_credits。
"""

from __future__ import annotations

import math

# ============ 爆款视频动态定价 (#2151) ============
# key = (model_id, resolution, has_video_input)，单位：元/百万token
VIRAL_VIDEO_MODEL_PRICES: dict[tuple[str, str, bool], float] = {
    ("seedance-2.5", "480p", False): 70.0,
    ("seedance-2.5", "720p", False): 70.0,
    ("seedance-2.5", "1080p", False): 77.0,
    ("seedance-2.5", "480p", True): 42.0,
    ("seedance-2.5", "720p", True): 42.0,
    ("seedance-2.5", "1080p", True): 46.0,
    ("seedance-2.0", "480p", False): 46.0,
    ("seedance-2.0", "720p", False): 46.0,
    ("seedance-2.0", "1080p", False): 51.0,
}

# 固定成本（元）：VLM 分析 + LLM 文案 + TTS + OSS + 服务器
VIRAL_VIDEO_FIXED_COST = 0.15
# 利润系数
VIRAL_VIDEO_PROFIT_MULTIPLIER = 1.3
# Seedance 输出帧率
VIRAL_VIDEO_FPS = 24

# 分辨率别名映射 -> 标准 key
_RESOLUTION_ALIASES: dict[str, str] = {
    "480p": "480p",
    "普清": "480p",
    "default": "480p",
    "low": "480p",
    "sd": "480p",
    "720p": "720p",
    "高清": "720p",
    "medium": "720p",
    "hd": "720p",
    "1080p": "1080p",
    "超清": "1080p",
    "high": "1080p",
    "ultra": "1080p",
    "全能": "1080p",
    "fhd": "1080p",
}
# 分辨率 -> 短边像素数（p 值代表短边，不是 height）
_RESOLUTION_SHORT_SIDE: dict[str, int] = {"480p": 480, "720p": 720, "1080p": 1080}


def resolve_video_dimensions(resolution: str, ratio: str) -> tuple[int, int]:
    """把 (resolution, ratio) 解析为 (width, height)。

    resolution 数字代表短边像素数（480p/720p/1080p 等）：
    - 横屏 16:9：短边是 height，width = short * 16/9
    - 竖屏 9:16：短边是 width，height = short * 16/9
    - 方屏 1:1：width = height = short
    """
    key = str(resolution or "").strip()
    key_l = key.lower()
    res_key = _RESOLUTION_ALIASES.get(key_l) or _RESOLUTION_ALIASES.get(key) or "720p"
    short = _RESOLUTION_SHORT_SIDE.get(res_key, 720)
    r = str(ratio or "").strip().lower()
    if r == "16:9":
        # 横屏：短边是 height，width 向上取整并对齐偶数
        w = math.ceil(short * 16 / 9)
        h = short
    elif r == "1:1":
        w, h = short, short
    else:
        # 9:16 竖屏（默认）：短边是 width，height 向上取整并对齐偶数
        w = short
        h = math.ceil(short * 16 / 9)
    # 对齐到偶数（视频编码要求）
    w = w + (w % 2)
    h = h + (h % 2)
    return int(w), int(h)


def _match_model_prefix(model: str) -> str:
    """匹配 model 前缀。"""
    m = (model or "").strip().lower()
    for prefix in ("seedance-2.5", "seedance-2.0"):
        if m.startswith(prefix):
            return prefix
    return "seedance-2.5"


def _infer_resolution_key(width: int, height: int) -> str:
    """从实际 (width, height) 用短边推断 resolution key。"""
    short = min(int(width or 720), int(height or 720))
    if short >= 1000:
        return "1080p"
    if short >= 650:
        return "720p"
    return "480p"


def calculate_viral_video_credits_with_breakdown(
    duration_seconds: int,
    width: int,
    height: int,
    model: str = "seedance-2.5",
    has_video_input: bool = False,
    actual_tokens: int | None = None,
    fps: int = VIRAL_VIDEO_FPS,
) -> tuple[float, dict]:
    """计算爆款视频所需积分（1 积分 = 1 元），并返回计费公式明细。

    公式:
        tokens = duration * width * height * fps / 1024
        video_cost = tokens / 1_000_000 * model_token_price
        total = round((video_cost + fixed_cost) * profit_multiplier, 2)
    若传入 actual_tokens 则用它替代计算值。

    Returns:
        (credits, breakdown) 二元组：
        - credits: 四舍五入保留两位小数的最终积分
        - breakdown: dict，包含 tokens / video_cost / fixed_cost / profit_multiplier /
          model_price / width / height / fps 字段，便于前端展示计费明细。
    """
    w = max(1, int(width or 1))
    h = max(1, int(height or 1))
    effective_fps = int(fps or VIRAL_VIDEO_FPS)

    prefix = _match_model_prefix(model)
    res_key = _infer_resolution_key(w, h)
    key = (prefix, res_key, bool(has_video_input))
    price = VIRAL_VIDEO_MODEL_PRICES.get(key)
    if price is None:
        price = VIRAL_VIDEO_MODEL_PRICES.get(("seedance-2.5", res_key, False), 70.0)
    if actual_tokens is not None and actual_tokens > 0:
        tokens = float(actual_tokens)
    else:
        dur = max(1, int(duration_seconds or 15))
        tokens = dur * w * h * effective_fps / 1024.0

    video_cost = tokens / 1_000_000.0 * float(price)
    total = (video_cost + VIRAL_VIDEO_FIXED_COST) * VIRAL_VIDEO_PROFIT_MULTIPLIER
    credits = round(float(total), 2)
    breakdown = {
        "tokens": float(tokens),
        "video_cost": float(video_cost),
        "fixed_cost": float(VIRAL_VIDEO_FIXED_COST),
        "profit_multiplier": float(VIRAL_VIDEO_PROFIT_MULTIPLIER),
        "model_price": float(price),
        "width": int(w),
        "height": int(h),
        "fps": int(effective_fps),
    }
    return credits, breakdown


def calculate_viral_video_credits(
    duration_seconds: int,
    width: int,
    height: int,
    model: str = "seedance-2.5",
    has_video_input: bool = False,
    actual_tokens: int | None = None,
    fps: int = VIRAL_VIDEO_FPS,
) -> float:
    """计算爆款视频所需积分（1 积分 = 1 元），仅返回积分值（向后兼容包装器）。

    内部调用 calculate_viral_video_credits_with_breakdown，仅返回 credits 部分，
    保持旧调用方签名与返回值类型不变。

    公式:
        tokens = duration * width * height * fps / 1024
        video_cost = tokens / 1_000_000 * model_token_price
        total = round((video_cost + fixed_cost) * profit_multiplier, 2)
    若传入 actual_tokens 则用它替代计算值。
    """
    credits, _ = calculate_viral_video_credits_with_breakdown(
        duration_seconds=duration_seconds,
        width=width,
        height=height,
        model=model,
        has_video_input=has_video_input,
        actual_tokens=actual_tokens,
        fps=fps,
    )
    return credits


# ============ 场景定义 ============
# 每个场景: base_points(基础积分), unit(计费单位), name(显示名称), dynamic(是否动态定价)
# 说明：爆款视频（viral_video）走动态定价（预扣→结算多退少补），因此不使用 @points_gate
# 装饰器，base_points=0，dynamic=True；前端展示场景列表时仍可看到。

POINTS_SCENES: dict[str, dict] = {
    "voice_clone_train": {
        "base_points": 0,
        "unit": "次",
        "name": "声音克隆训练",
        "description": "声音克隆训练免费（每用户限 1 个声音）",
    },
    "voice_clone_synth": {
        "base_points": 1,
        "unit": "分钟",
        "name": "声音克隆合成",
        "description": "克隆音色合成每分钟消耗 1 积分",
    },
    "viral_video": {
        "base_points": 0,
        "unit": "次",
        "name": "爆款视频",
        "dynamic": True,
        "description": "爆款视频动态定价（按视频时长/分辨率/模型计算，预扣→结算多退少补）",
    },
}

# 免费用户积分消耗上浮系数（仅对 voice_clone_synth 生效）
FREE_USER_MULTIPLIER = 1.15

# ============ 积分包定义 ============

POINTS_PACKAGES: dict[str, dict] = {
    "starter_pack": {"name": "体验包", "points": 100, "price_cents": 990},
    "basic_pack": {"name": "基础包", "points": 500, "price_cents": 3900},
    "pro_pack": {"name": "专业包", "points": 2000, "price_cents": 12900},
}

# ============ 会员定价 ============

MEMBERSHIP_PRICES: dict[str, dict] = {
    "monthly": {"name": "月卡", "price_cents": 1990, "duration_days": 30},
    "quarterly": {"name": "季卡", "price_cents": 3990, "duration_days": 90},
    "yearly": {"name": "年卡", "price_cents": 15900, "duration_days": 365},
}

# 会员积分折扣（付费会员按此系数打折）
MEMBER_DISCOUNT: dict[str, float] = {
    "monthly": 0.9,
    "quarterly": 0.87,
    "yearly": 0.8,
}


def calculate_points_cost(
    scene_key: str,
    is_member: bool,
    quantity: int = 1,
    duration_minutes: float = 0,
    member_type: str | None = None,
) -> float:
    """计算指定场景的积分消耗。

    Args:
        scene_key: 场景标识（当前支持 voice_clone_train/voice_clone_synth/viral_video；
                   viral_video 为动态定价场景，此处返回 0，由业务侧调用
                   calculate_viral_video_credits 手动计算）
        is_member: 是否付费会员
        quantity: 数量（按次计费场景）
        duration_minutes: 时长分钟数（按时长计费场景）
        member_type: 会员类型 (monthly/quarterly/yearly)，用于折扣

    Returns:
        实际消耗积分（float；已含免费用户 ×1.15 上浮或会员折扣）；免费/动态/已下线场景统一返回 0。
    """
    scene = POINTS_SCENES.get(scene_key)
    if not scene:
        # 已下线/未注册的场景统一返回 0（免费），保持向后兼容
        return 0.0

    # 动态定价场景（如 viral_video）由业务侧手动计算，这里统一返回 0
    if scene.get("dynamic"):
        return 0.0

    base = scene["base_points"]
    if base == 0:
        return 0.0

    unit = scene["unit"]
    if unit == "分钟":
        total_base = base * max(1, math.ceil(duration_minutes))
    elif unit in ("次", "张"):
        total_base = base * quantity
    else:
        total_base = base

    if is_member and member_type and member_type in MEMBER_DISCOUNT:
        total_base = max(1, math.floor(total_base * MEMBER_DISCOUNT[member_type]))
    elif not is_member:
        total_base = math.ceil(total_base * FREE_USER_MULTIPLIER)

    return float(total_base)