208 lines
8.0 KiB
Python
208 lines
8.0 KiB
Python
"""老板挂牌积分规则(不谈供应商成本)。
|
|
|
|
存在 ModelConfig.metadata["points_pricing"]:
|
|
· image → mode=per_image, points_per_image
|
|
· text → mode=per_call, points_per_call
|
|
· video → mode=per_second, tiers=[{resolution, points_per_second, points_per_second_with_ref?}]
|
|
|
|
能力(可选分辨率/时长)在 metadata["capabilities"]。
|
|
未配置 points_pricing 时由 pricing.quote_* 回落旧口径(unit_price / token 价表)。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from decimal import Decimal, ROUND_HALF_UP
|
|
from typing import Any
|
|
|
|
|
|
def _dict(value: Any) -> dict:
|
|
return value if isinstance(value, dict) else {}
|
|
|
|
|
|
def _int_points(value: Any) -> int | None:
|
|
if value is None or value is False:
|
|
return None
|
|
try:
|
|
n = int(Decimal(str(value)).quantize(Decimal("1"), rounding=ROUND_HALF_UP))
|
|
except Exception: # noqa: BLE001
|
|
return None
|
|
return n if n >= 0 else None
|
|
|
|
|
|
def points_pricing_of(model_config) -> dict:
|
|
return _dict(_dict(getattr(model_config, "metadata", None)).get("points_pricing"))
|
|
|
|
|
|
def capabilities_of(model_config) -> dict:
|
|
return _dict(_dict(getattr(model_config, "metadata", None)).get("capabilities"))
|
|
|
|
|
|
def has_points_pricing(model_config) -> bool:
|
|
pricing = points_pricing_of(model_config)
|
|
mode = str(pricing.get("mode") or "").strip()
|
|
if mode == "per_image":
|
|
return _int_points(pricing.get("points_per_image")) is not None
|
|
if mode == "per_call":
|
|
return _int_points(pricing.get("points_per_call")) is not None
|
|
if mode == "per_second":
|
|
tiers = pricing.get("tiers")
|
|
return isinstance(tiers, list) and any(isinstance(t, dict) for t in tiers)
|
|
# 兼容:只填了数字字段
|
|
return (
|
|
_int_points(pricing.get("points_per_image")) is not None
|
|
or _int_points(pricing.get("points_per_call")) is not None
|
|
)
|
|
|
|
|
|
def image_points_per_unit(model_config) -> int | None:
|
|
pricing = points_pricing_of(model_config)
|
|
n = _int_points(pricing.get("points_per_image"))
|
|
if n is not None:
|
|
return n
|
|
# 回落 unit_price(积分/张)
|
|
return _int_points(getattr(model_config, "unit_price", None))
|
|
|
|
|
|
def text_points_per_call(model_config) -> int | None:
|
|
pricing = points_pricing_of(model_config)
|
|
n = _int_points(pricing.get("points_per_call"))
|
|
if n is not None:
|
|
return n
|
|
return _int_points(getattr(model_config, "unit_price", None))
|
|
|
|
|
|
def video_tier(model_config, *, resolution: str, with_ref_video: bool = False) -> dict | None:
|
|
pricing = points_pricing_of(model_config)
|
|
tiers = pricing.get("tiers")
|
|
if not isinstance(tiers, list):
|
|
return None
|
|
resolution = str(resolution or "").strip().lower()
|
|
hit = None
|
|
for raw in tiers:
|
|
if not isinstance(raw, dict):
|
|
continue
|
|
if str(raw.get("resolution") or "").strip().lower() == resolution:
|
|
hit = raw
|
|
break
|
|
return hit
|
|
|
|
|
|
def video_points_per_second(model_config, *, resolution: str, with_ref_video: bool = False) -> int | None:
|
|
tier = video_tier(model_config, resolution=resolution, with_ref_video=with_ref_video)
|
|
if not tier:
|
|
return None
|
|
if with_ref_video:
|
|
n = _int_points(tier.get("points_per_second_with_ref"))
|
|
if n is not None:
|
|
return n
|
|
return _int_points(tier.get("points_per_second"))
|
|
|
|
|
|
def quote_points_per_second(*, points_per_second: int, duration: int) -> Decimal:
|
|
seconds = max(int(duration or 0), 0)
|
|
pts = max(int(points_per_second) * seconds, 1) if seconds > 0 else 0
|
|
return Decimal(pts)
|
|
|
|
|
|
def quote_points_units(*, points_per_unit: int, units: int) -> Decimal:
|
|
count = max(int(units or 0), 0)
|
|
if count <= 0:
|
|
return Decimal("0")
|
|
return Decimal(max(int(points_per_unit) * count, 1))
|
|
|
|
|
|
def normalize_points_pricing_for_save(capability: str, metadata: dict | None) -> dict:
|
|
"""整理 metadata:补齐 points_pricing.mode,并让 unit_price 与挂牌积分对齐(兼容旧读法)。"""
|
|
root = dict(metadata or {})
|
|
pricing = dict(_dict(root.get("points_pricing")))
|
|
caps = dict(_dict(root.get("capabilities")))
|
|
capability = str(capability or "")
|
|
|
|
if capability == "image":
|
|
pts = _int_points(pricing.get("points_per_image"))
|
|
if pts is not None:
|
|
pricing["mode"] = "per_image"
|
|
pricing["points_per_image"] = pts
|
|
root["points_pricing"] = pricing
|
|
return root
|
|
|
|
if capability in {"text", "vision"}:
|
|
pts = _int_points(pricing.get("points_per_call"))
|
|
if pts is not None:
|
|
pricing["mode"] = "per_call"
|
|
pricing["points_per_call"] = pts
|
|
root["points_pricing"] = pricing
|
|
return root
|
|
|
|
if capability == "video":
|
|
tiers_in = pricing.get("tiers") if isinstance(pricing.get("tiers"), list) else []
|
|
tiers: list[dict] = []
|
|
resolutions: list[str] = []
|
|
for raw in tiers_in:
|
|
if not isinstance(raw, dict):
|
|
continue
|
|
res = str(raw.get("resolution") or "").strip()
|
|
pps = _int_points(raw.get("points_per_second"))
|
|
if not res or pps is None:
|
|
continue
|
|
row = {"resolution": res, "points_per_second": pps}
|
|
with_ref = _int_points(raw.get("points_per_second_with_ref"))
|
|
if with_ref is not None:
|
|
row["points_per_second_with_ref"] = with_ref
|
|
tiers.append(row)
|
|
if res not in resolutions:
|
|
resolutions.append(res)
|
|
if tiers:
|
|
pricing["mode"] = "per_second"
|
|
pricing["tiers"] = tiers
|
|
# 能力分辨率与价表对齐,前端下拉只出有价的档
|
|
if not caps.get("resolutions"):
|
|
caps["resolutions"] = resolutions
|
|
else:
|
|
# 保留已配能力,但确保价表分辨率都在里面
|
|
existing = [str(x) for x in (caps.get("resolutions") or [])]
|
|
for res in resolutions:
|
|
if res not in existing:
|
|
existing.append(res)
|
|
caps["resolutions"] = existing
|
|
root["points_pricing"] = pricing
|
|
root["capabilities"] = caps
|
|
return root
|
|
|
|
root["points_pricing"] = pricing
|
|
return root
|
|
|
|
|
|
def points_pricing_errors(capability: str, metadata: Any) -> tuple[str, ...]:
|
|
"""可选校验:填了 points_pricing 就要完整。"""
|
|
root = _dict(metadata)
|
|
pricing = _dict(root.get("points_pricing"))
|
|
if not pricing:
|
|
return ()
|
|
errors: list[str] = []
|
|
mode = str(pricing.get("mode") or "").strip()
|
|
if capability == "image":
|
|
if _int_points(pricing.get("points_per_image")) is None and mode in {"", "per_image"}:
|
|
if "points_per_image" in pricing:
|
|
errors.append("points_pricing.points_per_image 必须是 ≥0 的整数积分")
|
|
elif capability in {"text", "vision"}:
|
|
if "points_per_call" in pricing and _int_points(pricing.get("points_per_call")) is None:
|
|
errors.append("points_pricing.points_per_call 必须是 ≥0 的整数积分")
|
|
elif capability == "video":
|
|
tiers = pricing.get("tiers")
|
|
if tiers is None:
|
|
return tuple(errors)
|
|
if not isinstance(tiers, list) or not tiers:
|
|
errors.append("视频 points_pricing.tiers 至少配置一档分辨率积分")
|
|
else:
|
|
for i, raw in enumerate(tiers):
|
|
if not isinstance(raw, dict):
|
|
errors.append(f"tiers[{i}] 必须是对象")
|
|
continue
|
|
if not str(raw.get("resolution") or "").strip():
|
|
errors.append(f"tiers[{i}].resolution 不能为空")
|
|
if _int_points(raw.get("points_per_second")) is None:
|
|
errors.append(f"tiers[{i}].points_per_second 必须是 ≥0 的整数积分")
|
|
if "points_per_second_with_ref" in raw and _int_points(raw.get("points_per_second_with_ref")) is None:
|
|
errors.append(f"tiers[{i}].points_per_second_with_ref 必须是 ≥0 的整数积分")
|
|
return tuple(errors)
|