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)，单位：
# - billing_mode=token: 元/百万tokens（输出）
# - billing_mode=per_second: 元/秒（视频时长）
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,
    ("seedance-2.0", "4k", False): 80.0,
    ("seedance-2.0-fast", "480p", False): 28.0,
    ("seedance-2.0-fast", "720p", False): 28.0,
    ("seedance-2.0-mini", "480p", False): 9.2,
    ("seedance-2.0-mini", "720p", False): 9.2,
    ("wan-3.0", "480p", False): 0.3,
    ("wan-3.0", "720p", False): 0.6,
    ("wan-3.0", "1080p", False): 1.2,
}

# 固定成本（元）：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, "4k": 2160}
_RESOLUTION_ALIASES["4k"] = "4k"
_RESOLUTION_ALIASES["2160p"] = "4k"
_RESOLUTION_ALIASES["uhd"] = "4k"


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)


# ── 爆款视频多模型元数据 (#2159) ──────────────────────────────────────
VIRAL_VIDEO_MODEL_CONFIG: dict[str, dict] = {
    "seedance-2.5": {
        "key": "seedance-2.5",
        "display_name": "Seedance 2.5 — 最新最强",
        "model_id": "doubao-seedance-2-5-260628",
        "provider": "doubao",
        "supports_audio": True,
        "supported_resolutions": ["480p", "720p", "1080p"],
        "max_duration": 30,
        "billing_mode": "token",
        "is_default": True,
    },
    "seedance-2.0": {
        "key": "seedance-2.0",
        "display_name": "Seedance 2.0 — 正式首选",
        "model_id": "doubao-seedance-2-0-260128",
        "provider": "doubao",
        "supports_audio": True,
        "supported_resolutions": ["480p", "720p", "1080p", "4k"],
        "max_duration": 15,
        "billing_mode": "token",
        "is_default": False,
    },
    "seedance-2.0-fast": {
        "key": "seedance-2.0-fast",
        "display_name": "Seedance 2.0 Fast — 快速低成本",
        "model_id": "doubao-seedance-2-0-fast-260128",
        "provider": "doubao",
        "supports_audio": True,
        "supported_resolutions": ["480p", "720p"],
        "max_duration": 15,
        "billing_mode": "token",
        "is_default": False,
    },
    "seedance-2.0-mini": {
        "key": "seedance-2.0-mini",
        "display_name": "Seedance 2.0 Mini — 低成本测试",
        "model_id": "doubao-seedance-2-0-mini-260615",
        "provider": "doubao",
        "supports_audio": True,
        "supported_resolutions": ["480p", "720p"],
        "max_duration": 15,
        "billing_mode": "token",
        "is_default": False,
    },
    "wan-3.0": {
        "key": "wan-3.0",
        "display_name": "Wan 3.0 — 通义万相（阿里云）",
        "model_id": "wan3.0-video",
        "provider": "dashscope",
        "supports_audio": True,
        "supported_resolutions": ["480p", "720p", "1080p"],
        "max_duration": 30,
        "billing_mode": "per_second",
        "is_default": False,
    },
}


def get_viral_video_model_config(model_key: str | None) -> dict:
    """获取模型配置，未知 key 回落到默认 seedance-2.5。"""
    key = (model_key or "").strip().lower()
    if key and key in VIRAL_VIDEO_MODEL_CONFIG:
        return VIRAL_VIDEO_MODEL_CONFIG[key]
    return VIRAL_VIDEO_MODEL_CONFIG["seedance-2.5"]


def list_viral_video_models(
    include_placeholder: bool = False,
    dashscope_available: bool = False,
) -> list[dict]:
    """返回前端可用的模型列表（供 GET /api/v1/viral-video/models 端点用）。"""
    out: list[dict] = []
    for _k, cfg in VIRAL_VIDEO_MODEL_CONFIG.items():
        if cfg.get("_placeholder") and not include_placeholder:
            continue
        if cfg.get("provider") == "dashscope" and not dashscope_available:
            continue
        out.append(
            {
                "key": cfg["key"],
                "display_name": cfg["display_name"],
                "supports_audio": bool(cfg.get("supports_audio", True)),
                "supported_resolutions": list(cfg.get("supported_resolutions", ["720p"])),
                "max_duration": int(cfg.get("max_duration", 15)),
                "billing_mode": cfg.get("billing_mode", "token"),
                "is_default": bool(cfg.get("is_default", False)),
            }
        )
    return out


def _match_model_prefix(model: str | None) -> str:
    """匹配 model key（支持全部内部别名，未知回落到 seedance-2.5）。

    按 key 长度从长到短匹配，避免 "seedance-2.0-fast" 被 "seedance-2.0" 前缀命中。
    """
    mm = (model or "").strip().lower()
    for k in sorted(VIRAL_VIDEO_MODEL_CONFIG.keys(), key=len, reverse=True):
        if mm == k or mm.startswith(k):
            return k
    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 >= 1900:
        return "4k"
    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)
    cfg = get_viral_video_model_config(prefix)
    res_key = _infer_resolution_key(w, h)
    billing = cfg.get("billing_mode", "token")
    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)
    dur = max(1, int(duration_seconds or 15))
    if billing == "per_second":
        tokens = 0.0
        video_cost = dur * float(price)
        billing_unit = "second"
    else:
        if actual_tokens is not None and actual_tokens > 0:
            tokens = float(actual_tokens)
        else:
            tokens = dur * w * h * effective_fps / 1024.0
        video_cost = tokens / 1_000_000.0 * float(price)
        billing_unit = "token"

    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),
        "model_key": prefix,
        "billing_mode": billing,
        "billing_unit": billing_unit,
        "width": int(w),
        "height": int(h),
        "fps": int(effective_fps),
        "duration": dur,
    }
    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)