350 lines
16 KiB
Python
350 lines
16 KiB
Python
"""平台统一计价引擎(积分制 · 参考 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.points_pricing 挂牌(分辨率积分/秒 × 时长;可配含视频参考价);
|
||
未挂牌才走 metadata.pricing 火山成本表 × 毛利 → 积分,并按 usage.total_tokens true-up。
|
||
挂牌任务预留=精确积分、结算认下单快照;token 任务预留仍 × video_reserve_buffer。
|
||
|
||
取整(前后端必须逐字一致,前端镜像在 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")
|
||
# 提炼提示词 / 上传视频提炼:功能挂牌价,不跟 Gemini unit_price(那是普通文本 10 积分/次)。
|
||
VIDEO_DIGEST_POINTS = Decimal("30")
|
||
|
||
|
||
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:
|
||
"""文本(次)/ 图像(张)。优先 metadata.points_pricing 挂牌;否则 unit_price;≤0 回落默认积分。team 传入则乘团队价格系数。"""
|
||
from apps.billing.points_rules import image_points_per_unit, text_points_per_call
|
||
|
||
capability = str(getattr(model_config, "capability", "") or "")
|
||
listed = None
|
||
if capability == "image":
|
||
listed = image_points_per_unit(model_config)
|
||
elif capability in {"text", "vision"}:
|
||
listed = text_points_per_call(model_config)
|
||
listed_from_hang = listed is not None and listed > 0
|
||
if listed_from_hang:
|
||
unit_points = Decimal(listed)
|
||
else:
|
||
unit_points = Decimal(str(getattr(model_config, "unit_price", 0) or 0))
|
||
if unit_points <= 0:
|
||
is_image = 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)
|
||
# 挂牌:image=per_image / text=per_call;未挂牌仍记 flat(unit_price 或默认)
|
||
if listed_from_hang and capability == "image":
|
||
rule = "per_image"
|
||
elif listed_from_hang and capability in {"text", "vision"}:
|
||
rule = "per_call"
|
||
else:
|
||
rule = "flat"
|
||
return Quote(
|
||
points=points,
|
||
base_cost_yuan=base_per_unit * units,
|
||
meta={"rule": rule, "units": units, "unit_points": str(unit_points), "price_multiplier": str(multiplier), "rate": str(get_billing_config().points_per_yuan)},
|
||
)
|
||
|
||
|
||
def quote_video_digest(*, team=None, model_config=None) -> Quote:
|
||
"""视频提炼提示词:优先 metadata.points_pricing.points_per_call 挂牌;
|
||
不吃 unit_price(与旧「固定 30、忽略 unit_price」一致);未挂牌回落固定 30。
|
||
失败退还。team 传入则乘团队价格系数。"""
|
||
from apps.billing.points_rules import points_pricing_of, _int_points
|
||
|
||
multiplier = team_price_multiplier(team)
|
||
listed = None
|
||
if model_config is not None:
|
||
listed = _int_points(points_pricing_of(model_config).get("points_per_call"))
|
||
if listed is not None and listed > 0:
|
||
unit = Decimal(listed)
|
||
rule = "per_call"
|
||
else:
|
||
unit = VIDEO_DIGEST_POINTS
|
||
rule = "video_digest_flat"
|
||
points = apply_team_price(unit, multiplier)
|
||
base_cost = Decimal("0")
|
||
if model_config is not None:
|
||
base_cost = Decimal(str(_pricing_meta(model_config).get("base_cost_yuan") or 0))
|
||
return Quote(
|
||
points=points,
|
||
base_cost_yuan=base_cost,
|
||
meta={
|
||
"rule": rule,
|
||
"units": 1,
|
||
"unit_points": str(unit),
|
||
"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]:
|
||
"""预估视频积分。优先老板挂牌(分辨率×秒);未配置则回落 token×成本×毛利。返回 (tokens占位, Quote)。"""
|
||
from apps.ai.video_pricing import estimate_video_cost, has_video_reference
|
||
from apps.billing.points_rules import quote_points_per_second, video_points_per_second
|
||
|
||
with_ref = has_video_reference(references)
|
||
pps = video_points_per_second(model_config, resolution=resolution, with_ref_video=with_ref)
|
||
multiplier = team_price_multiplier(team)
|
||
if pps is not None:
|
||
list_points = quote_points_per_second(points_per_second=pps, duration=duration)
|
||
points = apply_team_price(list_points, multiplier)
|
||
return 0, Quote(
|
||
points=points,
|
||
base_cost_yuan=Decimal("0"),
|
||
meta={
|
||
"rule": "points_per_second",
|
||
"resolution": resolution,
|
||
"duration": duration,
|
||
"with_ref_video": with_ref,
|
||
"points_per_second": pps,
|
||
"price_multiplier": str(multiplier),
|
||
"rate": str(get_billing_config().points_per_yuan),
|
||
},
|
||
)
|
||
|
||
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=multiplier)
|
||
|
||
|
||
def quote_video_actual(
|
||
model_config,
|
||
*,
|
||
tokens: int,
|
||
with_video_ref: bool,
|
||
resolution: str,
|
||
multiplier: Decimal | None = None,
|
||
duration: int | None = None,
|
||
listed_points_per_second: int | None = None,
|
||
) -> Quote:
|
||
"""结算视频积分。优先老板挂牌秒价(与预估同口径);未挂牌才按 tokens×成本表。
|
||
multiplier / listed_points_per_second 必须用**下单快照**,中途改价不影响在途任务。"""
|
||
from apps.ai.video_pricing import tokens_to_cost
|
||
from apps.billing.points_rules import quote_points_per_second, video_points_per_second
|
||
|
||
multiplier = multiplier if multiplier is not None else Decimal("1")
|
||
pps = listed_points_per_second
|
||
if pps is None:
|
||
pps = video_points_per_second(model_config, resolution=resolution, with_ref_video=with_video_ref)
|
||
if pps is not None and duration is not None and int(duration) > 0:
|
||
list_points = quote_points_per_second(points_per_second=int(pps), duration=int(duration))
|
||
points = apply_team_price(list_points, multiplier)
|
||
return Quote(
|
||
points=points,
|
||
base_cost_yuan=Decimal("0"),
|
||
meta={
|
||
"rule": "points_per_second",
|
||
"resolution": resolution,
|
||
"duration": int(duration),
|
||
"with_ref_video": with_video_ref,
|
||
"points_per_second": int(pps),
|
||
"price_multiplier": str(multiplier),
|
||
"rate": str(get_billing_config().points_per_yuan),
|
||
},
|
||
)
|
||
|
||
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, *, rule: str | None = None) -> Decimal:
|
||
"""视频预留额。挂牌秒价=精确积分(无 buffer);token 估价才 × buffer(真实 tokens 可能略超预估)。"""
|
||
points = Decimal(str(points)).quantize(Decimal("1"), rounding=ROUND_HALF_UP)
|
||
if rule == "points_per_second":
|
||
return max(points, Decimal("1")) if points > 0 else Decimal("0")
|
||
buffer = get_billing_config().video_reserve_buffer
|
||
return (points * buffer).quantize(Decimal("1"), rounding=ROUND_HALF_UP)
|
||
|
||
|
||
def video_quote_payload(quote: Quote) -> dict:
|
||
"""下单时写入 request_payload 的计价快照,结算只认这些字段。"""
|
||
meta = quote.meta or {}
|
||
out = {
|
||
"pricing_rule": meta.get("rule") or "",
|
||
"price_multiplier": meta.get("price_multiplier", "1"),
|
||
"points_per_yuan_snapshot": meta.get("rate", ""),
|
||
}
|
||
if meta.get("rule") == "points_per_second":
|
||
out["listed_points_per_second"] = int(meta.get("points_per_second") or 0)
|
||
out["with_ref_video"] = bool(meta.get("with_ref_video"))
|
||
if meta.get("duration") is not None:
|
||
out["billing_duration"] = int(meta["duration"])
|
||
return out
|
||
|
||
|
||
def settle_video_from_payload(model_config, *, payload: dict, tokens: int = 0) -> Quote:
|
||
"""统一视频结算:挂牌任务按快照秒价×时长扣;token 任务按 usage true-up。
|
||
挂牌不依赖 tokens——用量缺失也能按后台填的积分扣到位。"""
|
||
payload = payload or {}
|
||
multiplier = Decimal(str(payload.get("price_multiplier") or "1"))
|
||
resolution = str(payload.get("resolution") or "720p")
|
||
duration = payload.get("billing_duration")
|
||
if duration in (None, ""):
|
||
duration = payload.get("duration")
|
||
try:
|
||
duration_i = int(duration or 0) or None
|
||
except (TypeError, ValueError):
|
||
duration_i = None
|
||
with_ref = payload.get("with_ref_video")
|
||
if with_ref is None:
|
||
with_ref = any((r or {}).get("type") == "video" for r in (payload.get("references") or []))
|
||
listed = payload.get("listed_points_per_second")
|
||
try:
|
||
listed_i = int(listed) if listed not in (None, "") else None
|
||
except (TypeError, ValueError):
|
||
listed_i = None
|
||
rule = str(payload.get("pricing_rule") or "")
|
||
# 挂牌快照在:即使 usage.tokens=0 也按秒价结算(与预估同数)
|
||
if rule == "points_per_second" or listed_i is not None:
|
||
return quote_video_actual(
|
||
model_config,
|
||
tokens=int(tokens or 0),
|
||
with_video_ref=bool(with_ref),
|
||
resolution=resolution,
|
||
multiplier=multiplier,
|
||
duration=duration_i,
|
||
listed_points_per_second=listed_i,
|
||
)
|
||
if int(tokens or 0) <= 0:
|
||
# 无 usage 且无挂牌:由调用方回落 estimated_cost
|
||
return Quote(points=Decimal("0"), base_cost_yuan=Decimal("0"), meta={"rule": "missing_usage"})
|
||
# 无挂牌快照 → 纯 token 结算(即便模型后来配了挂牌也不改在途口径)
|
||
from apps.ai.video_pricing import tokens_to_cost
|
||
|
||
cost_yuan = tokens_to_cost(
|
||
model_config, int(tokens), with_video_ref=bool(with_ref), resolution=resolution,
|
||
)
|
||
return quote_video_from_cost(cost_yuan, tokens=int(tokens), multiplier=multiplier)
|