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",
}
# 分辨率 -> 高度
_RESOLUTION_HEIGHT: dict[str, int] = {"480p": 480, "720p": 720, "1080p": 1080}


def resolve_video_dimensions(resolution: str, ratio: str) -> tuple[int, int]:
    """把 (resolution, ratio) 解析为 (width, height)。"""
    key = str(resolution or "").strip()
    key_l = key.lower()
    res_key = _RESOLUTION_ALIASES.get(key_l) or _RESOLUTION_ALIASES.get(key) or "720p"
    h = _RESOLUTION_HEIGHT.get(res_key, 720)
    r = str(ratio or "").strip().lower()
    if r == "16:9":
        w = h * 16 // 9
    elif r == "1:1":
        w = h
    else:
        w = h * 9 // 16
    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(height: int) -> str:
    """从像素高度推断 resolution key。"""
    if height >= 1000:
        return "1080p"
    if height >= 650:
        return "720p"
    return "480p"


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 元）。

    公式:
        tokens = duration * width * height * fps / 1024
        video_cost = tokens / 1_000_000 * model_token_price
        total = round((video_cost + fixed_cost) * 1.3, 2)
    若传入 actual_tokens 则用它替代计算值。
    """
    prefix = _match_model_prefix(model)
    res_key = _infer_resolution_key(int(height or 720))
    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))
        w = max(1, int(width or 1))
        h = max(1, int(height or 1))
        tokens = dur * w * h * int(fps or VIRAL_VIDEO_FPS) / 1024.0

    video_cost = tokens / 1_000_000.0 * float(price)
    total = (video_cost + VIRAL_VIDEO_FIXED_COST) * VIRAL_VIDEO_PROFIT_MULTIPLIER
    return round(float(total), 2)


# ============ 场景定义 ============
# 每个场景: base_points(基础积分), unit(计费单位), name(显示名称)
# 说明：仅保留需要扣点的场景；免费场景不要写入此字典。

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 积分",
    },
}

# 免费用户积分消耗上浮系数（仅对 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）
        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

    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)