import time from django.conf import settings from rest_framework.exceptions import PermissionDenied from rest_framework.permissions import AllowAny, IsAuthenticated from rest_framework.response import Response from rest_framework.views import APIView from . import stats from .system_info import get_os_info def _worker_counts(): """Active/queued background jobs (downloads and e621 match scans).""" from apps.library.models import DownloadTask, MatchTask active = DownloadTask.objects.filter( status=DownloadTask.STATUS_DOWNLOADING ).count() + MatchTask.objects.filter(status=MatchTask.STATUS_RUNNING).count() queued = DownloadTask.objects.filter( status=DownloadTask.STATUS_PENDING ).count() + MatchTask.objects.filter(status=MatchTask.STATUS_PENDING).count() return {"active": active, "queued": queued} class StatusView(APIView): """Runtime status used by the SPA shell (storage line, status pill, footer).""" permission_classes = [AllowAny] def get(self, request): started = getattr(request, "start_time", time.perf_counter()) payload = { "app": "J621", "env": settings.APP_ENV, "git_hash": settings.GIT_COMMIT_HASH, "version": settings.APP_VERSION, "os": get_os_info(), "watched_folder": settings.WATCHED_FOLDER, "e621_time_ms": None, # e621 latency is measured client-side "workers": _worker_counts(), } payload["server_time_ms"] = round((time.perf_counter() - started) * 1000, 1) return Response(payload) class StatsView(APIView): """Staff-only runtime statistics for the dashboard.""" permission_classes = [IsAuthenticated] def get(self, request): user = request.user if not ( user.is_staff or user.is_superuser or user.role == user.ROLE_STAFF ): raise PermissionDenied("Staff only.") from apps.library.tools import storage_info return Response( { "system": stats.system_stats(), "gpus": stats.gpu_stats(), "disk": storage_info(), "jobs": stats.jobs_stats(), "log": stats.log_tail(), } )