feat(admin): add admin backend endpoints and service layer
Why: - Dashboard needs system health, Qdrant management, chunk preview, questions, and cost estimation endpoints Changes: - Add admin router with 11 endpoints (health, Qdrant CRUD, chunk preview, questions management, cost estimation) - Add delete_experiment to SQLite storage - Mount admin router and dashboard static files at /app Impact: - New /admin/* API routes available - Dashboard served at /app/ via StaticFiles
This commit is contained in:
275
src/admin/service.py
Normal file
275
src/admin/service.py
Normal file
@@ -0,0 +1,275 @@
|
||||
"""Admin service — health checks, Qdrant management, chunk preview, questions CRUD, cost estimation."""
|
||||
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from qdrant_client.models import VectorParams, Distance
|
||||
|
||||
from src.core.dependencies import get_qdrant_client, get_openai_client
|
||||
from src.core.config import settings
|
||||
from src.core.models import StrategyName
|
||||
from src.storage import qdrant as qdrant_store
|
||||
from src.storage import sqlite as db
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Project root (two levels up from src/admin/)
|
||||
PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent
|
||||
QUESTIONS_DIR = PROJECT_ROOT / "files"
|
||||
|
||||
|
||||
# ── Health ──────────────────────────────────────────────────
|
||||
|
||||
def get_health() -> dict[str, Any]:
|
||||
"""Check server, Qdrant, and SQLite status."""
|
||||
result: dict[str, Any] = {"status": "ok"}
|
||||
|
||||
# Check Qdrant
|
||||
try:
|
||||
client = get_qdrant_client()
|
||||
collections = client.get_collections()
|
||||
result["qdrant_connected"] = True
|
||||
result["qdrant_collections"] = len(collections.collections)
|
||||
except Exception as exc:
|
||||
result["qdrant_connected"] = False
|
||||
result["qdrant_error"] = str(exc)
|
||||
logger.warning("Qdrant health check failed: %s", exc)
|
||||
|
||||
# Check SQLite
|
||||
try:
|
||||
conn = db._connect()
|
||||
conn.execute("SELECT 1")
|
||||
conn.close()
|
||||
result["sqlite_ok"] = True
|
||||
except Exception as exc:
|
||||
result["sqlite_ok"] = False
|
||||
result["sqlite_error"] = str(exc)
|
||||
logger.warning("SQLite health check failed: %s", exc)
|
||||
|
||||
# Check OpenAI
|
||||
try:
|
||||
client = get_openai_client()
|
||||
# Just check the client exists; don't make a real API call
|
||||
result["openai_configured"] = bool(settings.openai_api_key)
|
||||
except Exception:
|
||||
result["openai_configured"] = False
|
||||
|
||||
return result
|
||||
|
||||
|
||||
# ── Qdrant Collections ─────────────────────────────────────
|
||||
|
||||
def list_qdrant_collections() -> dict[str, Any]:
|
||||
"""List all Qdrant collections with their point counts."""
|
||||
client = get_qdrant_client()
|
||||
collections_data = client.get_collections().collections
|
||||
|
||||
result = []
|
||||
for col in collections_data:
|
||||
try:
|
||||
info = client.get_collection(collection_name=col.name)
|
||||
result.append({
|
||||
"name": col.name,
|
||||
"points_count": info.points_count or 0,
|
||||
})
|
||||
except Exception as exc:
|
||||
result.append({
|
||||
"name": col.name,
|
||||
"points_count": None,
|
||||
"error": str(exc),
|
||||
})
|
||||
|
||||
return {"collections": result}
|
||||
|
||||
|
||||
def create_qdrant_collection(collection_name: str) -> dict[str, Any]:
|
||||
"""Create a new Qdrant collection."""
|
||||
client = get_qdrant_client()
|
||||
|
||||
existing = [c.name for c in client.get_collections().collections]
|
||||
if collection_name in existing:
|
||||
return {"created": False, "message": f"Collection '{collection_name}' already exists"}
|
||||
|
||||
client.create_collection(
|
||||
collection_name=collection_name,
|
||||
vectors_config=VectorParams(
|
||||
size=qdrant_store.VECTOR_DIMENSION,
|
||||
distance=Distance.COSINE,
|
||||
),
|
||||
)
|
||||
logger.info("Created Qdrant collection: %s", collection_name)
|
||||
return {"created": True, "collection": collection_name}
|
||||
|
||||
|
||||
def delete_qdrant_collection(collection_name: str) -> dict[str, Any]:
|
||||
"""Delete a Qdrant collection entirely."""
|
||||
client = get_qdrant_client()
|
||||
client.delete_collection(collection_name=collection_name)
|
||||
logger.info("Deleted Qdrant collection: %s", collection_name)
|
||||
return {"deleted": True, "collection": collection_name}
|
||||
|
||||
|
||||
def wipe_qdrant_collection_points(collection_name: str) -> dict[str, Any]:
|
||||
"""Delete all points in a collection but keep the collection."""
|
||||
client = get_qdrant_client()
|
||||
from qdrant_client.models import PointIdsList
|
||||
|
||||
info = client.get_collection(collection_name=collection_name)
|
||||
count = info.points_count or 0
|
||||
|
||||
if count == 0:
|
||||
return {"deleted": 0, "collection": collection_name}
|
||||
|
||||
client.delete(
|
||||
collection_name=collection_name,
|
||||
points_selector=PointIdsList(points=list(range(count))),
|
||||
)
|
||||
logger.info("Wiped %d points from %s", count, collection_name)
|
||||
return {"deleted": count, "collection": collection_name}
|
||||
|
||||
|
||||
# ── Chunk Preview ───────────────────────────────────────────
|
||||
|
||||
def preview_chunks(doc_id: str, strategy: str | None = None) -> dict[str, Any]:
|
||||
"""Preview chunks for a document. Uses Qdrant scroll to fetch chunks with payload."""
|
||||
client = get_qdrant_client()
|
||||
|
||||
from qdrant_client.models import Filter, FieldCondition, MatchValue
|
||||
|
||||
# Get document info from SQLite
|
||||
doc = db.get_document(doc_id)
|
||||
if doc is None:
|
||||
return {"error": f"Document not found: {doc_id}"}
|
||||
|
||||
doc_name = doc.get("filename", "")
|
||||
|
||||
# Determine which collections to search
|
||||
strategies_to_search = []
|
||||
if strategy:
|
||||
strategies_to_search = [strategy]
|
||||
else:
|
||||
strategies_to_search = [s.value for s in StrategyName]
|
||||
|
||||
results = {}
|
||||
for strat_name in strategies_to_search:
|
||||
col_name = f"{strat_name}_collection"
|
||||
|
||||
try:
|
||||
existing = [c.name for c in client.get_collections().collections]
|
||||
if col_name not in existing:
|
||||
results[strat_name] = {"chunks": [], "count": 0}
|
||||
continue
|
||||
|
||||
scroll_filter = Filter(
|
||||
must=[FieldCondition(key="document_name", match=MatchValue(value=doc_name))]
|
||||
)
|
||||
points, _ = client.scroll(
|
||||
collection_name=col_name,
|
||||
scroll_filter=scroll_filter,
|
||||
limit=10000,
|
||||
with_payload=True,
|
||||
with_vectors=False,
|
||||
)
|
||||
|
||||
chunks = []
|
||||
for p in points:
|
||||
payload = p.payload or {}
|
||||
chunks.append({
|
||||
"chunk_id": payload.get("chunk_id", str(p.id)),
|
||||
"chunk_index": payload.get("chunk_index"),
|
||||
"text": (payload.get("text") or "")[:500],
|
||||
"token_count": payload.get("token_count"),
|
||||
"character_count": payload.get("character_count"),
|
||||
"parent_id": payload.get("parent_id"),
|
||||
})
|
||||
|
||||
# Sort by chunk_index
|
||||
chunks.sort(key=lambda c: c.get("chunk_index") or 0)
|
||||
results[strat_name] = {"chunks": chunks, "count": len(chunks)}
|
||||
|
||||
except Exception as exc:
|
||||
results[strat_name] = {"error": str(exc), "chunks": [], "count": 0}
|
||||
|
||||
return {"document_id": doc_id, "filename": doc_name, "strategies": results}
|
||||
|
||||
|
||||
# ── Questions Dataset ───────────────────────────────────────
|
||||
|
||||
def list_question_files() -> dict[str, Any]:
|
||||
"""List JSON files in the files/ directory."""
|
||||
QUESTIONS_DIR.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
files = []
|
||||
for f in sorted(QUESTIONS_DIR.glob("*.json")):
|
||||
try:
|
||||
with open(f, encoding="utf-8") as fh:
|
||||
data = json.load(fh)
|
||||
count = len(data.get("questions", []))
|
||||
except Exception:
|
||||
count = -1
|
||||
files.append({
|
||||
"id": f.name,
|
||||
"name": f.name,
|
||||
"questions_count": count,
|
||||
"size_bytes": f.stat().st_size,
|
||||
})
|
||||
|
||||
return {"files": files}
|
||||
|
||||
|
||||
def upload_questions(filename: str, content: bytes) -> dict[str, Any]:
|
||||
"""Save a questions JSON file to the files/ directory."""
|
||||
QUESTIONS_DIR.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# Validate JSON
|
||||
try:
|
||||
data = json.loads(content.decode("utf-8"))
|
||||
except (json.JSONDecodeError, UnicodeDecodeError) as exc:
|
||||
return {"error": f"Invalid JSON: {exc}"}
|
||||
|
||||
if "questions" not in data and not isinstance(data, list):
|
||||
return {"error": "Invalid format: must have a 'questions' key or be a list"}
|
||||
|
||||
# Ensure filename ends with .json
|
||||
if not filename.endswith(".json"):
|
||||
filename = filename + ".json"
|
||||
|
||||
dest = QUESTIONS_DIR / filename
|
||||
dest.write_bytes(content)
|
||||
logger.info("Uploaded questions file: %s", dest)
|
||||
|
||||
return {"uploaded": True, "id": filename, "questions_count": len(data.get("questions", []) if isinstance(data, dict) else data)}
|
||||
|
||||
|
||||
def get_questions(file_id: str) -> dict[str, Any]:
|
||||
"""Read and return the content of a questions file."""
|
||||
path = QUESTIONS_DIR / file_id
|
||||
if not path.exists():
|
||||
return {"error": f"File not found: {file_id}"}
|
||||
|
||||
with open(path, encoding="utf-8") as f:
|
||||
data = json.load(f)
|
||||
|
||||
return {"id": file_id, "data": data}
|
||||
|
||||
|
||||
def delete_questions(file_id: str) -> dict[str, Any]:
|
||||
"""Delete a questions JSON file."""
|
||||
path = QUESTIONS_DIR / file_id
|
||||
if not path.exists():
|
||||
return {"error": f"File not found: {file_id}"}
|
||||
|
||||
path.unlink()
|
||||
logger.info("Deleted questions file: %s", path)
|
||||
return {"deleted": True, "id": file_id}
|
||||
|
||||
|
||||
# ── Cost Estimation ─────────────────────────────────────────
|
||||
|
||||
def estimate_cost(num_questions: int, num_strategies: int) -> dict[str, Any]:
|
||||
"""Estimate benchmark cost. Delegates to benchmark service."""
|
||||
from src.benchmarking.benchmark_service import estimate_cost as bench_estimate
|
||||
return bench_estimate(num_questions, num_strategies)
|
||||
Reference in New Issue
Block a user