331 lines
11 KiB
Python
331 lines
11 KiB
Python
"""User-facing helpers for cmhub hosted model aliases."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import copy
|
||
from decimal import Decimal, InvalidOperation
|
||
import threading
|
||
import time
|
||
|
||
|
||
TIER_DEFAULT = "default"
|
||
TIER_HIGH_QUALITY = "high_quality"
|
||
TIER_ECONOMICAL = "economical"
|
||
|
||
TIER_LABELS = {
|
||
TIER_DEFAULT: "默认档",
|
||
TIER_HIGH_QUALITY: "高质量档",
|
||
TIER_ECONOMICAL: "省点档",
|
||
}
|
||
|
||
TIER_DESCRIPTIONS = {
|
||
TIER_DEFAULT: "日常批量，质量、速度和成本平衡",
|
||
TIER_HIGH_QUALITY: "重点商品或失败重试，质量优先，成本可能更高",
|
||
TIER_ECONOMICAL: "低价值 SKU 或初稿，省点优先，上线前需人工复核",
|
||
}
|
||
|
||
|
||
# The catalog is deliberately process-local. It only caches public model metadata,
|
||
# never the API key or a price estimate chosen by a user.
|
||
MODEL_CATALOG_CACHE_TTL_SECONDS = 300
|
||
_MODEL_CATALOG_CACHE = {}
|
||
_MODEL_CATALOG_CACHE_LOCK = threading.RLock()
|
||
_CATALOG_ALL_ALIASES_KEY = "*"
|
||
_PRICE_CONDITION_KEYS = (
|
||
"resolution",
|
||
"name",
|
||
"quality",
|
||
"size",
|
||
"ratio",
|
||
"aspect_ratio",
|
||
"tier",
|
||
"condition",
|
||
"conditions",
|
||
"min_images",
|
||
"max_images",
|
||
"image_count",
|
||
)
|
||
|
||
|
||
def cache_model_catalog(base_url, models, *, now=None):
|
||
"""Cache one fetched catalog in memory for all aliases it contains."""
|
||
|
||
normalized_base_url = _catalog_base_url(base_url)
|
||
if not normalized_base_url:
|
||
return
|
||
copied_models = [copy.deepcopy(model) for model in list(models or []) if isinstance(model, dict)]
|
||
fetched_at = float(time.monotonic() if now is None else now)
|
||
entry = (fetched_at, copied_models)
|
||
aliases = {
|
||
str(model.get("alias") or "").strip()
|
||
for model in copied_models
|
||
if str(model.get("alias") or "").strip()
|
||
}
|
||
with _MODEL_CATALOG_CACHE_LOCK:
|
||
_MODEL_CATALOG_CACHE[(normalized_base_url, _CATALOG_ALL_ALIASES_KEY)] = entry
|
||
for alias in aliases:
|
||
_MODEL_CATALOG_CACHE[(normalized_base_url, alias)] = entry
|
||
|
||
|
||
def cached_model_catalog(base_url, alias="", *, max_age_seconds=None, now=None):
|
||
"""Return a fresh catalog copy for a normalized gateway/alias pair, if available."""
|
||
|
||
normalized_base_url = _catalog_base_url(base_url)
|
||
normalized_alias = str(alias or "").strip()
|
||
if not normalized_base_url:
|
||
return None
|
||
max_age = MODEL_CATALOG_CACHE_TTL_SECONDS if max_age_seconds is None else max_age_seconds
|
||
try:
|
||
max_age = float(max_age)
|
||
except (TypeError, ValueError):
|
||
return None
|
||
current = float(time.monotonic() if now is None else now)
|
||
keys = [(normalized_base_url, normalized_alias)] if normalized_alias else []
|
||
keys.append((normalized_base_url, _CATALOG_ALL_ALIASES_KEY))
|
||
with _MODEL_CATALOG_CACHE_LOCK:
|
||
for key in keys:
|
||
entry = _MODEL_CATALOG_CACHE.get(key)
|
||
if entry is None:
|
||
continue
|
||
fetched_at, models = entry
|
||
if current - fetched_at > max_age:
|
||
continue
|
||
return [copy.deepcopy(model) for model in models]
|
||
return None
|
||
|
||
|
||
def clear_model_catalog_cache():
|
||
"""Clear the transient catalog cache. Intended for tests and gateway changes."""
|
||
|
||
with _MODEL_CATALOG_CACHE_LOCK:
|
||
_MODEL_CATALOG_CACHE.clear()
|
||
|
||
|
||
def unambiguous_points_cost(models, alias, operation_type, *, requires_image=None):
|
||
"""Return the unique unconditional price for one configured alias, otherwise ``None``."""
|
||
|
||
normalized_alias = str(alias or "").strip()
|
||
normalized_operation = str(operation_type or "").strip().lower()
|
||
matches = [
|
||
model
|
||
for model in list(models or [])
|
||
if isinstance(model, dict) and str(model.get("alias") or "").strip() == normalized_alias
|
||
]
|
||
if len(matches) != 1:
|
||
return None
|
||
model = matches[0]
|
||
if str(model.get("operation_type") or "").strip().lower() != normalized_operation:
|
||
return None
|
||
if str(model.get("pricing_status") or "").strip().lower() != "priced":
|
||
return None
|
||
if requires_image is True and model.get("requires_image") is not True:
|
||
return None
|
||
if requires_image is False and model.get("requires_image") is not False:
|
||
return None
|
||
prices = model.get("prices")
|
||
if not isinstance(prices, list) or len(prices) != 1 or not isinstance(prices[0], dict):
|
||
return None
|
||
price = prices[0]
|
||
if any(_has_value(price.get(key)) for key in _PRICE_CONDITION_KEYS):
|
||
return None
|
||
return _points_cost_value(price)
|
||
|
||
|
||
def format_points_cost(value):
|
||
"""Format a catalog points value without inventing precision."""
|
||
|
||
if value is None:
|
||
return ""
|
||
decimal_value = _points_cost_decimal(value)
|
||
if decimal_value is None:
|
||
return ""
|
||
text = format(decimal_value.normalize(), "f")
|
||
if "." in text:
|
||
text = text.rstrip("0").rstrip(".")
|
||
return text or "0"
|
||
|
||
|
||
def _catalog_base_url(base_url):
|
||
return str(base_url or "").strip().rstrip("/").lower()
|
||
|
||
|
||
def _has_value(value):
|
||
return value not in (None, "", [], {}, ())
|
||
|
||
|
||
def _points_cost_value(price):
|
||
if not isinstance(price, dict):
|
||
return None
|
||
if "points_cost" not in price:
|
||
return None
|
||
return _points_cost_decimal(price.get("points_cost"))
|
||
|
||
|
||
def _points_cost_decimal(value):
|
||
if isinstance(value, bool) or value is None:
|
||
return None
|
||
try:
|
||
decimal_value = Decimal(str(value))
|
||
except (InvalidOperation, ValueError):
|
||
return None
|
||
if not decimal_value.is_finite() or decimal_value < 0:
|
||
return None
|
||
return decimal_value
|
||
|
||
|
||
def normalize_tier(value):
|
||
text = str(value or "").strip().lower().replace("-", "_").replace(" ", "_")
|
||
if text in {"high", "quality", "high_quality", "premium", "pro", "sol"}:
|
||
return TIER_HIGH_QUALITY
|
||
if text in {"low", "cheap", "economy", "economical", "saving", "lite", "luna"}:
|
||
return TIER_ECONOMICAL
|
||
if text in {"default", "standard", "balanced", "balance", "terra"}:
|
||
return TIER_DEFAULT
|
||
return ""
|
||
|
||
|
||
def model_tier(model):
|
||
model = model if isinstance(model, dict) else {}
|
||
for key in ("tier", "model_tier", "quality_tier", "cmhub_tier", "recommended_tier"):
|
||
tier = normalize_tier(model.get(key))
|
||
if tier:
|
||
return tier
|
||
haystack = " ".join(_model_text_values(model)).lower()
|
||
if any(token in haystack for token in ("省点", "低成本", "高频", "high-volume", "luna", "economy", "cheap", "lite")):
|
||
return TIER_ECONOMICAL
|
||
if any(token in haystack for token in ("高质量", "旗舰", "重点", "失败重试", "premium", "high-quality", "sol")):
|
||
return TIER_HIGH_QUALITY
|
||
if any(token in haystack for token in ("平衡", "默认", "日常", "standard", "balanced", "terra")):
|
||
return TIER_DEFAULT
|
||
alias = str(model.get("alias") or "").lower()
|
||
if alias.endswith("-hd") or alias.endswith("_hd") or alias.endswith(".hd") or alias == "hd":
|
||
return TIER_HIGH_QUALITY
|
||
return TIER_DEFAULT
|
||
|
||
|
||
def tier_label(tier):
|
||
return TIER_LABELS.get(normalize_tier(tier) or tier, TIER_LABELS[TIER_DEFAULT])
|
||
|
||
|
||
def tier_description(tier):
|
||
return TIER_DESCRIPTIONS.get(normalize_tier(tier) or tier, TIER_DESCRIPTIONS[TIER_DEFAULT])
|
||
|
||
|
||
def model_display_name(model):
|
||
model = model if isinstance(model, dict) else {}
|
||
for key in ("display_name", "name", "label"):
|
||
value = str(model.get(key) or "").strip()
|
||
if value:
|
||
return value
|
||
return str(model.get("alias") or "").strip()
|
||
|
||
|
||
def price_text(prices):
|
||
if not isinstance(prices, list):
|
||
return ""
|
||
parts = []
|
||
for item in prices[:3]:
|
||
if not isinstance(item, dict):
|
||
continue
|
||
cost = item.get("points_cost")
|
||
if cost is None:
|
||
cost = item.get("cost")
|
||
if cost is None:
|
||
continue
|
||
resolution = item.get("resolution") or item.get("name") or ""
|
||
parts.append(f"{resolution}:{cost}点" if resolution else f"{cost}点")
|
||
return "/".join(parts)
|
||
|
||
|
||
def alias_label(model):
|
||
model = model if isinstance(model, dict) else {}
|
||
alias = str(model.get("alias") or "").strip()
|
||
display_name = model_display_name(model)
|
||
tier = model_tier(model)
|
||
parts = [tier_label(tier)]
|
||
if display_name and display_name != alias:
|
||
parts.append(display_name)
|
||
elif alias:
|
||
parts.append(alias)
|
||
prices = price_text(model.get("prices"))
|
||
if prices:
|
||
parts.append(prices)
|
||
if model.get("requires_image"):
|
||
parts.append("需参考图")
|
||
return " · ".join(parts)
|
||
|
||
|
||
def alias_tooltip(model):
|
||
model = model if isinstance(model, dict) else {}
|
||
alias = str(model.get("alias") or "").strip()
|
||
tier = model_tier(model)
|
||
usage = _recommended_for_text(model)
|
||
lines = [f"{tier_label(tier)}：{usage or tier_description(tier)}"]
|
||
if alias:
|
||
lines.append(f"cmhub 别名：{alias}")
|
||
prices = price_text(model.get("prices"))
|
||
if prices:
|
||
lines.append(f"扣点：{prices}")
|
||
return "\n".join(lines)
|
||
|
||
|
||
def configured_alias_summary(alias, *, model=None):
|
||
alias = str(alias or "").strip()
|
||
if model is None:
|
||
model = {"alias": alias}
|
||
else:
|
||
model = dict(model)
|
||
model.setdefault("alias", alias)
|
||
tier = model_tier(model)
|
||
if alias:
|
||
return f"cmhub 托管{tier_label(tier)}，生图别名 {alias}，扣点以返回结果为准"
|
||
return "cmhub 托管默认档，未配置生图别名；请先到设置选择生图别名"
|
||
|
||
|
||
def _recommended_for_text(model):
|
||
for key in ("recommended_for", "usage", "description"):
|
||
value = model.get(key)
|
||
if isinstance(value, (list, tuple)):
|
||
text = "、".join(str(item).strip() for item in value if str(item).strip())
|
||
else:
|
||
text = str(value or "").strip()
|
||
if text:
|
||
return text
|
||
return ""
|
||
|
||
|
||
def _model_text_values(model):
|
||
for key in (
|
||
"alias",
|
||
"display_name",
|
||
"name",
|
||
"label",
|
||
"tier",
|
||
"model_tier",
|
||
"quality_tier",
|
||
"recommended_tier",
|
||
"recommended_for",
|
||
"description",
|
||
):
|
||
value = model.get(key)
|
||
if isinstance(value, (list, tuple, set)):
|
||
for item in value:
|
||
text = str(item or "").strip()
|
||
if text:
|
||
yield text
|
||
elif isinstance(value, dict):
|
||
for item in value.values():
|
||
text = str(item or "").strip()
|
||
if text:
|
||
yield text
|
||
else:
|
||
text = str(value or "").strip()
|
||
if text:
|
||
yield text
|
||
tags = model.get("tags")
|
||
if isinstance(tags, (list, tuple, set)):
|
||
for tag in tags:
|
||
text = str(tag or "").strip()
|
||
if text:
|
||
yield text
|