sbt-idp/cope2n-api/fwd_api/utils/health.py

47 lines
1.8 KiB
Python

from django.db.models import Count
from django.utils import timezone
from datetime import timedelta
from ..models import SubscriptionRequest
def get_latest_requests(limit=50):
requests = SubscriptionRequest.objects.filter(is_test_request=False).order_by("-created_at")[:limit]
requests_dict = []
for request in requests:
requests_dict.append({
"request_id": request.request_id,
"pages": request.pages,
"doc_type": request.doc_type,
# "predict_result": request.predict_result,
"created_at": request.created_at,
"updated_at": request.updated_at,
"preprocessing_time": request.preprocessing_time,
"ai_inference_time": request.ai_inference_time,
"cpu_percent": request.cpu_percent,
"gpu_percent": request.gpu_percent,
"used_memory": request.used_memory,
"total_memory": request.total_memory,
"gpu_stats": request.gpu_stats,
"is_reviewed": request.is_reviewed,
# "is_bad_image_quality": request.is_bad_image_quality,
})
return requests_dict
def count_requests_by_date(days_limit=5):
today = timezone.now().date()
start_date = today - timedelta(days=days_limit)
requests_by_date = SubscriptionRequest.objects.filter(is_test_request=False).filter(created_at__gte=start_date).values('created_at__date').annotate(count=Count('id')).values('created_at__date', 'count').order_by('created_at__date')
count_dict = []
for rbd in requests_by_date:
count_dict.append({
"date": rbd["created_at__date"],
"count": rbd["count"],
})
return count_dict
def get_health_report():
return {
"latest_requests": get_latest_requests(),
"count_by_date": count_requests_by_date(),
}