cmshoppe/app/cmhub_models.py

"""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