Upload api/resources_endpoint.py with huggingface_hub

api/resources_endpoint.py

@@ -1,31 +1,86 @@
 """
-Resources Endpoint - API router for resource statistics
+Resources Endpoint - API router for resource statistics (patched for real registry data).
 """
-from fastapi import APIRouter
-from typing import Dict, Any
 from datetime import datetime
-import logging
+from typing import Any, Dict
 
-logger = logging.getLogger(__name__)
+from fastapi import APIRouter
 
 router = APIRouter(prefix="/api/resources", tags=["resources"])
 
+
+def _registry_stats() -> Dict[str, Any]:
+    try:
+        import hf_unified_server as server
+
+        summary, categories = server._summarize_resources()
+        return {
+            "total": summary.get("total", 0),
+            "active": summary.get("free", 0) + summary.get("premium", 0),
+            "free": summary.get("free", 0),
+            "premium": summary.get("premium", 0),
+            "categories": categories,
+            "registry_loaded": bool(getattr(server, "_RESOURCES_CACHE", None)),
+        }
+    except Exception:
+        try:
+            from unified_resource_loader import get_loader
+
+            loader = get_loader()
+            stats = loader.get_stats()
+            categories = [
+                {"name": cat, "count": count}
+                for cat, count in stats.get("categories", {}).items()
+            ]
+            return {
+                "total": stats.get("total_resources", 0),
+                "active": stats.get("total_resources", 0),
+                "free": stats.get("free_resources", 0),
+                "premium": stats.get("auth_required", 0),
+                "categories": categories,
+                "registry_loaded": loader.loaded,
+            }
+        except Exception:
+            return {"total": 0, "active": 0, "categories": [], "registry_loaded": False}
+
+
 @router.get("/stats")
 async def resources_stats() -> Dict[str, Any]:
-    """Get resource statistics"""
-    return {
-        "total": 0,
-        "active": 0,
-        "categories": [],
-        "timestamp": datetime.utcnow().isoformat() + "Z"
-    }
+    """Get resource statistics from unified registry."""
+    data = _registry_stats()
+    data["timestamp"] = datetime.utcnow().isoformat() + "Z"
+    data["success"] = True
+    return data
+
 
 @router.get("/list")
 async def resources_list() -> Dict[str, Any]:
-    """Get list of all resources"""
-    return {
-        "resources": [],
-        "total": 0,
-        "timestamp": datetime.utcnow().isoformat() + "Z"
-    }
+    """Get list of all resources."""
+    try:
+        from unified_resource_loader import get_loader
 
+        loader = get_loader()
+        resources = [
+            {
+                "id": r.id,
+                "name": r.name,
+                "category": r.category,
+                "base_url": r.base_url,
+                "requires_auth": r.requires_auth(),
+            }
+            for r in loader.resources.values()
+        ]
+        return {
+            "success": True,
+            "resources": resources,
+            "total": len(resources),
+            "timestamp": datetime.utcnow().isoformat() + "Z",
+        }
+    except Exception as exc:
+        return {
+            "success": False,
+            "resources": [],
+            "total": 0,
+            "error": str(exc),
+            "timestamp": datetime.utcnow().isoformat() + "Z",
+        }