File size: 7,180 Bytes
674fb4e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 | from datetime import datetime, timezone
import logging
from fastapi import APIRouter, Depends, HTTPException, BackgroundTasks, Request, UploadFile, File, Form, Query, Response
from typing import List, Dict, Any, Optional
from ...workers.celery_worker import celery_app
logger = logging.getLogger(__name__)
from ...core.neo4j_store import Neo4jStore
from ...retrieval.agent import AgentRetrievalSystem
from ...ingestion.pipeline import IngestionPipeline
from ...config import settings
from ...api.models import *
from ...api.auth import get_current_user, User
import redis
from ..dependencies import get_graph_store, get_retrieval_agent, get_ingestion_pipeline, get_redis_client
router = APIRouter()
from ...core.storage import get_storage
storage = get_storage()
@router.get("/api/system/health", response_model=SystemHealthResponse, tags=["System"])
async def health_check(request: Request, response: Response):
"""System health check"""
neo4j_connected = False
redis_connected = False
workers_active = 0
gds_version = None
try:
# Check Neo4j
await request.app.state.graph_store.execute_query("RETURN 1")
neo4j_connected = True
# Check GDS
try:
gds_res = await request.app.state.graph_store.execute_query("RETURN gds.version() as version")
if gds_res:
gds_version = gds_res[0]["version"]
except Exception:
pass
except Exception as e:
logger.error(f"Neo4j health check failed: {e}")
try:
# Check Redis
if hasattr(request.app.state, 'redis_client'):
await request.app.state.redis_client.ping()
redis_connected = True
else:
redis_connected = True # If redis isn't configured, ignore it
except Exception as e:
logger.error(f"Redis health check failed: {e}")
try:
# Check Celery workers
inspect = celery_app.control.inspect()
active = inspect.active()
if active:
workers_active = len(active)
except Exception as e:
logger.warning(f"Celery health check failed: {e}")
overall_status = "healthy" if (neo4j_connected and redis_connected) else "degraded"
if overall_status == "degraded":
response.status_code = 503
return SystemHealthResponse(
status=overall_status,
version=settings.app_version,
neo4j_connected=neo4j_connected,
redis_connected=redis_connected,
workers_active=workers_active,
gds_version=gds_version,
timestamp=datetime.now(timezone.utc).replace(tzinfo=None)
)
@router.get("/api/system/stats", response_model=SystemStatsResponse, tags=["System"])
async def get_system_stats(request: Request, current_user: User = Depends(get_current_user)):
"""Get system statistics"""
tenant_id = current_user.tenant_id
params = {"tenant_id": tenant_id} if tenant_id else {}
# Count documents
doc_query = "MATCH (d:Document {tenant_id: $tenant_id}) RETURN count(d) as count" if tenant_id else "MATCH (d:Document) RETURN count(d) as count"
doc_result = await request.app.state.graph_store.execute_query(doc_query, params)
documents_count = doc_result[0]["count"] if doc_result else 0
# Count entities
entity_query = "MATCH (e:Entity {tenant_id: $tenant_id}) RETURN count(e) as count" if tenant_id else "MATCH (e:Entity) RETURN count(e) as count"
entity_result = await request.app.state.graph_store.execute_query(entity_query, params)
entities_count = entity_result[0]["count"] if entity_result else 0
# Count relationships
rel_query = "MATCH ()-[r {tenant_id: $tenant_id}]->() RETURN count(r) as count" if tenant_id else "MATCH ()-[r]->() RETURN count(r) as count"
rel_result = await request.app.state.graph_store.execute_query(rel_query, params)
relationships_count = rel_result[0]["count"] if rel_result else 0
# Count chunks
chunk_query = "MATCH (c:Chunk {tenant_id: $tenant_id}) RETURN count(c) as count" if tenant_id else "MATCH (c:Chunk) RETURN count(c) as count"
chunk_result = await request.app.state.graph_store.execute_query(chunk_query, params)
chunks_count = chunk_result[0]["count"] if chunk_result else 0
ontology = request.app.state.ingestion_pipeline.get_ontology()
ontology_version = ontology.version if ontology else "none"
return SystemStatsResponse(
documents_count=documents_count,
entities_count=entities_count,
relationships_count=relationships_count,
chunks_count=chunks_count,
ontology_version=ontology_version
)
@router.get("/api/system/my-stats", tags=["System"])
async def get_my_stats(request: Request, current_user: User = Depends(get_current_user)):
"""Get activity stats for the currently authenticated user."""
from fastapi.responses import JSONResponse
username = current_user.username
conv_q = """
MATCH (u:User {username: $username})-[:HAS_CONVERSATION]->(c:Conversation)
RETURN count(DISTINCT c) as conversation_count
"""
msg_q = """
MATCH (u:User {username: $username})-[:HAS_CONVERSATION]->(c:Conversation)-[:HAS_MESSAGE]->(m)
WHERE m.role = 'user'
RETURN count(m) as message_count, max(m.created_at) as last_active
"""
try:
conv_rows = await request.app.state.graph_store.execute_query(conv_q, {"username": username})
msg_rows = await request.app.state.graph_store.execute_query(msg_q, {"username": username})
conversation_count = conv_rows[0]["conversation_count"] if conv_rows else 0
message_count = msg_rows[0]["message_count"] if msg_rows else 0
last_active = msg_rows[0]["last_active"] if msg_rows else None
if hasattr(last_active, "isoformat"):
last_active = last_active.isoformat()
except Exception:
conversation_count = 0
message_count = 0
last_active = None
return JSONResponse({
"username": username,
"conversation_count": conversation_count,
"message_count": message_count,
"last_active": last_active,
})
@router.get("/api/system/formats", response_model=SupportedFormatsResponse, tags=["System"])
async def get_supported_formats(request: Request):
"""List supported ingestion file formats"""
return SupportedFormatsResponse(
formats=settings.allowed_file_types,
descriptions={
".pdf": "PDF documents (LlamaParse or pypdf)",
".txt": "Plain text files",
".md": "Markdown files",
".docx": "Microsoft Word documents",
".csv": "CSV spreadsheets (rows → entity facts)",
".xlsx": "Excel spreadsheets (all sheets processed)",
".pptx": "PowerPoint presentations (slides + notes)",
".json": "JSON data files (nested structures flattened)",
}
)
# ── Graph Export Endpoint ─────────────────────────────────────────────────────
|