Upload backend/services/dashboard_metrics.py with huggingface_hub

backend/services/dashboard_metrics.py

@@ -0,0 +1,229 @@
+"""
+Real dashboard metrics — single source of truth for cards and System Status drawer.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import re
+from pathlib import Path
+from typing import Any, Dict, List, Optional, Tuple
+
+_ENV_PROBE_VARS = [
+    "CRYPTOCOMPARE_API_KEY",
+    "COINMARKETCAP_KEY_1",
+    "COINMARKETCAP_KEY_2",
+    "ETHERSCAN_KEY_1",
+    "ETHERSCAN_KEY_2",
+    "BSCSCAN_API_KEY",
+    "TRONSCAN_API_KEY",
+    "NEWSAPI_KEY",
+    "HF_TOKEN",
+    "HUGGINGFACE_TOKEN",
+    "COINGECKO_API_KEY",
+]
+
+_PLACEHOLDER = re.compile(r"^\$\{[A-Z0-9_]+\}$")
+
+
+def _workspace_root() -> Path:
+    here = Path(__file__).resolve()
+    for parent in (here, *here.parents):
+        if (parent / "hf_unified_server.py").exists():
+            return parent
+    return here.parents[2]
+
+
+def _api_keys_path() -> Path:
+    return _workspace_root() / "config" / "api_keys.json"
+
+
+def _env_configured() -> Tuple[List[str], List[str]]:
+    configured: List[str] = []
+    missing: List[str] = []
+    for name in _ENV_PROBE_VARS:
+        val = (os.getenv(name) or "").strip()
+        if val and val.lower() not in ("null", "none", "cg-xxx", "cg-xxxxxxxxxxxxxxxxxxxxxxxx"):
+            if name not in configured:
+                configured.append(name)
+        else:
+            missing.append(name)
+    # HF_TOKEN / HUGGINGFACE_TOKEN count as one slot
+    if "HF_TOKEN" in configured and "HUGGINGFACE_TOKEN" in configured:
+        configured.remove("HUGGINGFACE_TOKEN")
+    if "HF_TOKEN" in missing and "HUGGINGFACE_TOKEN" in missing:
+        missing = [m for m in missing if m != "HUGGINGFACE_TOKEN"]
+    return configured, missing
+
+
+def _file_backed_keys() -> Dict[str, Any]:
+    """Count inline keys stored in config/api_keys.json (not env placeholders)."""
+    path = _api_keys_path()
+    if not path.exists():
+        return {"inline_key_count": 0, "providers": [], "by_provider": {}}
+
+    try:
+        cfg = json.loads(path.read_text(encoding="utf-8", errors="replace"))
+    except Exception:
+        return {"inline_key_count": 0, "providers": [], "by_provider": {}}
+
+    by_provider: Dict[str, int] = {}
+
+    def walk(obj: Any, provider: str = "") -> None:
+        if isinstance(obj, dict):
+            name = obj.get("description") or provider
+            if "keys" in obj and isinstance(obj["keys"], list):
+                valid = [k for k in obj["keys"] if k and not _PLACEHOLDER.match(str(k).strip())]
+                if valid:
+                    key = name or provider or "unknown"
+                    by_provider[key] = by_provider.get(key, 0) + len(valid)
+            if "key" in obj:
+                k = str(obj["key"]).strip()
+                if k and not _PLACEHOLDER.match(k):
+                    key = name or provider or "unknown"
+                    by_provider[key] = by_provider.get(key, 0) + 1
+            for k, v in obj.items():
+                if k in ("keys", "key", "env_aliases", "endpoints", "description"):
+                    continue
+                walk(v, k if isinstance(k, str) else provider)
+        elif isinstance(obj, list):
+            for item in obj:
+                walk(item, provider)
+
+    for section, content in cfg.items():
+        if section in ("description", "last_updated", "environment_variables", "aggregated_sources"):
+            continue
+        walk(content, section)
+
+    total = sum(by_provider.values())
+    return {
+        "inline_key_count": total,
+        "providers": list(by_provider.keys()),
+        "by_provider": by_provider,
+    }
+
+
+def get_api_keys_status() -> Dict[str, Any]:
+    configured_env, missing_env = _env_configured()
+    file_info = _file_backed_keys()
+    inline = file_info["inline_key_count"]
+    env_count = len(configured_env)
+    # Effective: env secrets + file keys (env takes precedence for same provider)
+    effective = env_count + inline
+    probe_total = len(_ENV_PROBE_VARS) - (1 if "HUGGINGFACE_TOKEN" in _ENV_PROBE_VARS else 0)
+
+    return {
+        "config_exists": _api_keys_path().exists(),
+        "env_configured": configured_env,
+        "env_missing": missing_env,
+        "env_configured_count": env_count,
+        "env_total_probed": probe_total,
+        "file_backed_keys": inline,
+        "file_providers": file_info["providers"],
+        "by_provider": file_info["by_provider"],
+        "configured_keys": effective,
+        "total_key_refs": probe_total + inline,
+        "display": f"{env_count} secrets + {inline} file keys",
+    }
+
+
+def get_rotation_summary() -> Dict[str, Any]:
+    try:
+        from provider_manager import get_provider_manager
+
+        manager = get_provider_manager()
+        manager.ensure_unified_sync()
+        pools: List[Dict[str, Any]] = []
+        total_providers = 0
+        available_providers = 0
+        for pool_id, pool in manager.pools.items():
+            stats = pool.get_stats()
+            avail = int(stats.get("available_providers") or 0)
+            tot = int(stats.get("total_providers") or 0)
+            total_providers += tot
+            available_providers += avail
+            pools.append({
+                "pool_id": pool_id,
+                "name": pool_id.replace("_", " ").title(),
+                "available_providers": avail,
+                "total_providers": tot,
+                "rotation_strategy": stats.get("rotation_strategy"),
+                "healthy": avail > 0,
+            })
+        healthy_pools = sum(1 for p in pools if p["healthy"])
+        return {
+            "success": True,
+            "pools_total": len(pools),
+            "pools_healthy": healthy_pools,
+            "providers_total": total_providers,
+            "providers_available": available_providers,
+            "pools": pools,
+            "env_keys": get_api_keys_status(),
+        }
+    except Exception as exc:
+        return {
+            "success": False,
+            "error": str(exc),
+            "pools_total": 0,
+            "pools_healthy": 0,
+            "providers_total": 0,
+            "providers_available": 0,
+            "pools": [],
+        }
+
+
+def get_registry_summary() -> Tuple[Dict[str, Any], List[Dict[str, Any]]]:
+    """Load category counts from unified registry cache if available."""
+    try:
+        import hf_unified_server as server
+        return server._summarize_resources()
+    except Exception:
+        return {"total": 0}, []
+
+
+async def get_models_combined() -> Dict[str, Any]:
+    v2_loaded = v2_total = 0
+    v4_loaded = v4_total = 0
+    hf_mode = "unknown"
+    v4_hf_mode = None
+    transformers = False
+    try:
+        from backend.routers.hf_models_summary import build_models_summary
+
+        ms = await build_models_summary()
+        if ms.get("success"):
+            sm = ms.get("summary") or {}
+            v2_loaded = int(sm.get("loaded_models") or 0)
+            v2_total = int(sm.get("total_models") or 0)
+            hf_mode = sm.get("hf_mode") or hf_mode
+            transformers = bool(sm.get("transformers_available"))
+            c4 = ms.get("complement_v4") or {}
+            v4_loaded = int(c4.get("loaded") or 0)
+            v4_total = int(c4.get("total") or 0)
+            v4_hf_mode = c4.get("hf_mode")
+    except Exception:
+        try:
+            from ai_models import _registry, HF_MODE, TRANSFORMERS_AVAILABLE, MODEL_SPECS
+
+            v2_loaded = len(getattr(_registry, "_pipelines", {}))
+            v2_total = len(MODEL_SPECS)
+            hf_mode = HF_MODE
+            transformers = TRANSFORMERS_AVAILABLE
+        except Exception:
+            pass
+
+    combined_loaded = v2_loaded + v4_loaded
+    combined_total = v2_total + v4_total
+    return {
+        "models_loaded_v2": v2_loaded,
+        "models_total_v2": v2_total,
+        "models_loaded_v4": v4_loaded,
+        "models_total_v4": v4_total,
+        "models_loaded": combined_loaded,
+        "models_total": combined_total,
+        "hf_mode": hf_mode,
+        "v4_hf_mode": v4_hf_mode,
+        "transformers_available": transformers,
+        "health": combined_loaded > 0 and hf_mode not in ("off", "unknown"),
+    }