yingqing/core/backend/apps/billing/pricing.py

"""平台统一计价引擎(积分制 · 参考 jimeng-clone 成本加成模型)。

单位约定:
  · 用户侧(余额/价格/流水/限额)一律**积分**,1 积分 = 1/points_per_yuan 元(默认 ¥0.1);
  · 平台成本(Quote.base_cost_yuan / AITask.base_cost)一律**人民币**——供应商结算是 ¥,
    两本账各自诚实,毛利 = actual_cost/points_per_yuan − base_cost。

规则来源:
  · flat 类型(文本/图像):ModelConfig.unit_price,语义已重定义为「积分/次(张)」(rescale 迁移 ×10);
    平台成本配在 metadata.pricing.base_cost_yuan(未配=0,毛利报表按未知处理)。
  · 配音:metadata.pricing = {"mode":"per_chars","chars_per_unit":500,"points_per_unit":10,
    "min_units":1,"base_cost_yuan_per_unit":...},按字符数阶梯。
  · 视频:metadata.pricing 是火山**成本价表**(元/百万tokens,见 apps/ai/video_pricing.py),
    用户价 = ¥成本 × video_margin_multiplier → 积分。按真实 usage.total_tokens 结算。

取整(前后端必须逐字一致,前端镜像在 frontend/src/components/free-create/constants.ts):
  ¥成本先 quantize 到 0.01(video_pricing.calculate_cost 现状)→ ×毛利 ×汇率 → ROUND_HALF_UP
  到整数积分,最低 1 积分。
"""
import math
from dataclasses import dataclass, field
from decimal import Decimal, ROUND_HALF_UP

from django.core.cache import cache

from .models import BillingConfig

_CONFIG_CACHE_KEY = "billing_config_v1"
_CONFIG_TTL = 60

# flat 类型 unit_price 未配置(≤0)时的兜底价(积分/次):按能力对齐公示价目
# (图像 20 积分/张、其余 10 积分/次)——否则 unit_price=0 的图像模型(火山 Seedream)
# 实扣会比消费页公示价低一半(review 确认的三方不一致)。
FLAT_FALLBACK_POINTS = Decimal("10")
FLAT_FALLBACK_POINTS_IMAGE = Decimal("20")


def get_billing_config() -> BillingConfig:
    """单例读取(60s 缓存;PATCH 端点写后主动失效)。无行则建默认行(幂等,迁移已 seed)。"""
    config = cache.get(_CONFIG_CACHE_KEY)
    if config is None:
        config = BillingConfig.objects.order_by("created_at").first()
        if config is None:
            config = BillingConfig.objects.create()
            # create() 返回的实例属性是模型默认值的 Python float,Decimal×float 直接 TypeError
            # 且会被缓存 60s(结算侧炸 = 成片被误判失败退费)。回读一次拿 DB 转换后的 Decimal。
            config.refresh_from_db()
        cache.set(_CONFIG_CACHE_KEY, config, _CONFIG_TTL)
    return config


def invalidate_billing_config_cache() -> None:
    cache.delete(_CONFIG_CACHE_KEY)


@dataclass(frozen=True)
class Quote:
    points: Decimal  # 用户应扣积分(整数,最低 1)
    base_cost_yuan: Decimal  # 平台成本 ¥(未知配 0)
    meta: dict = field(default_factory=dict)  # 计价快照(rule/units/tokens/margin/rate...)


def team_price_multiplier(team) -> Decimal:
    """团队价格系数(差异化调价):默认 1。非法/缺失一律回落 1,绝不因配置脏数据把计价打挂。"""
    raw = getattr(team, "price_multiplier", None) if team is not None else None
    try:
        value = Decimal(str(raw)) if raw is not None else Decimal("1")
    except Exception:  # noqa: BLE001
        return Decimal("1")
    if not value.is_finite() or value <= 0:
        return Decimal("1")
    return value


def apply_team_price(points: Decimal, multiplier: Decimal) -> Decimal:
    """标准挂牌积分价 × 团队系数 → HALF_UP 取整,最低 1(0 价不强收)。
    这是差异化调价的唯一落点:与前端预估两步取整逐字对齐(先算挂牌价取整,再乘系数取整)。"""
    if points <= 0:
        return points
    if multiplier == Decimal("1"):
        return points
    scaled = (points * multiplier).quantize(Decimal("1"), rounding=ROUND_HALF_UP)
    return max(scaled, Decimal("1"))


def yuan_to_points(yuan: Decimal, *, rate: Decimal | None = None) -> Decimal:
    """¥ × points_per_yuan → ROUND_HALF_UP 整数积分,最低 1(0 元除外)。"""
    if rate is None:
        rate = get_billing_config().points_per_yuan
    if yuan <= 0:
        return Decimal("0")
    points = (Decimal(str(yuan)) * rate).quantize(Decimal("1"), rounding=ROUND_HALF_UP)
    return max(points, Decimal("1"))


def _pricing_meta(model_config) -> dict:
    return ((getattr(model_config, "metadata", None) or {}).get("pricing")) or {}


def quote_flat(model_config, *, units: int = 1, team=None) -> Quote:
    """文本(次)/ 图像(张)。unit_price = 积分/单位;≤0 回落 10 积分。team 传入则乘团队价格系数。"""
    unit_points = Decimal(str(getattr(model_config, "unit_price", 0) or 0))
    if unit_points <= 0:
        is_image = str(getattr(model_config, "capability", "")) == "image"
        unit_points = FLAT_FALLBACK_POINTS_IMAGE if is_image else FLAT_FALLBACK_POINTS
    unit_points = unit_points.quantize(Decimal("1"), rounding=ROUND_HALF_UP)
    base_per_unit = Decimal(str(_pricing_meta(model_config).get("base_cost_yuan") or 0))
    multiplier = team_price_multiplier(team)
    points = apply_team_price(max(unit_points * units, Decimal("1")), multiplier)
    return Quote(
        points=points,
        base_cost_yuan=base_per_unit * units,
        meta={"rule": "flat", "units": units, "unit_points": str(unit_points), "price_multiplier": str(multiplier), "rate": str(get_billing_config().points_per_yuan)},
    )


def quote_voiceover(model_config, *, char_count: int, team=None) -> Quote:
    """配音字符阶梯:ceil(chars / chars_per_unit) × points_per_unit,最低 min_units 档。team 传入则乘系数。"""
    pricing = _pricing_meta(model_config)
    chars_per_unit = int(pricing.get("chars_per_unit") or 500)
    points_per_unit = Decimal(str(pricing.get("points_per_unit") or 10))
    min_units = int(pricing.get("min_units") or 1)
    base_per_unit = Decimal(str(pricing.get("base_cost_yuan_per_unit") or 0))
    units = max(math.ceil(max(char_count, 0) / chars_per_unit), min_units)
    multiplier = team_price_multiplier(team)
    points = apply_team_price(max((points_per_unit * units).quantize(Decimal("1"), rounding=ROUND_HALF_UP), Decimal("1")), multiplier)
    return Quote(
        points=points,
        base_cost_yuan=base_per_unit * units,
        meta={"rule": "per_chars", "char_count": char_count, "units": units, "chars_per_unit": chars_per_unit, "price_multiplier": str(multiplier), "rate": str(get_billing_config().points_per_yuan)},
    )


def quote_video_from_cost(cost_yuan: Decimal, *, tokens: int = 0, multiplier: Decimal | None = None) -> Quote:
    """视频通用:¥成本(video_pricing 已 quantize 0.01)× 毛利 → 挂牌积分 → ×团队系数。
    multiplier 由调用方决定来源:估价=团队当前系数;**按实结算=下单时的快照系数**
    (学 jimeng:中途改价不影响在途任务,否则预留 buffer 可能被中途涨价击穿)。"""
    config = get_billing_config()
    user_yuan = Decimal(str(cost_yuan)) * config.video_margin_multiplier
    list_points = yuan_to_points(user_yuan, rate=config.points_per_yuan)
    multiplier = multiplier if multiplier is not None else Decimal("1")
    points = apply_team_price(list_points, multiplier)
    return Quote(
        points=points,
        base_cost_yuan=Decimal(str(cost_yuan)),
        meta={
            "rule": "video_tokens",
            "tokens": tokens,
            "margin": str(config.video_margin_multiplier),
            "rate": str(config.points_per_yuan),
            "price_multiplier": str(multiplier),
        },
    )


def quote_video_estimate(model_config, *, aspect_ratio: str, resolution: str, duration: int, references: list, team=None) -> tuple[int, Quote]:
    """预估:apps/ai/video_pricing 的 ¥成本 × 毛利 → 积分 → ×团队当前系数。返回 (tokens, Quote)。"""
    from apps.ai.video_pricing import estimate_video_cost

    tokens, cost_yuan = estimate_video_cost(
        model_config, aspect_ratio=aspect_ratio, resolution=resolution, duration=duration, references=references
    )
    return tokens, quote_video_from_cost(cost_yuan, tokens=tokens, multiplier=team_price_multiplier(team))


def quote_video_actual(model_config, *, tokens: int, with_video_ref: bool, resolution: str, multiplier: Decimal | None = None) -> Quote:
    """按真实 usage.total_tokens 结算(与预估同一张成本表+同一毛利)。
    multiplier 必须传**下单时的快照系数**(request_payload.price_multiplier),不读团队当前值。"""
    from apps.ai.video_pricing import tokens_to_cost

    cost_yuan = tokens_to_cost(model_config, tokens, with_video_ref=with_video_ref, resolution=resolution)
    return quote_video_from_cost(cost_yuan, tokens=tokens, multiplier=multiplier)


def video_reserve_amount(points: Decimal) -> Decimal:
    """视频预留额 = 预估积分 × buffer(真实 tokens 可能略超预估;ledger 禁超预留扣费)。"""
    buffer = get_billing_config().video_reserve_buffer
    return (points * buffer).quantize(Decimal("1"), rounding=ROUND_HALF_UP)