feat(suite): confirm AI writing cost before request

app/cmhub_models.py

@@ -2,6 +2,11 @@
 
 from __future__ import annotations
 
+import copy
+from decimal import Decimal, InvalidOperation
+import threading
+import time
+
 
 TIER_DEFAULT = "default"
 TIER_HIGH_QUALITY = "high_quality"
@@ -20,6 +25,154 @@ TIER_DESCRIPTIONS = {
 }
 
 
+# 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"}: