138 lines
5.8 KiB
Python
138 lines
5.8 KiB
Python
from rest_framework import status
|
|
from adrf.views import APIView
|
|
from rest_framework.permissions import IsAuthenticated
|
|
from django.utils import timezone
|
|
from django.utils.decorators import method_decorator
|
|
from django.views.decorators.cache import never_cache
|
|
from datetime import timedelta
|
|
|
|
from utils.response_codes import (
|
|
ResponseCode,
|
|
create_standardized_response,
|
|
create_standardized_error_response
|
|
)
|
|
from ..models import LoginRecord
|
|
from ..serializers.user_serializers import LoginRecordSerializer
|
|
from drf_yasg.utils import swagger_auto_schema
|
|
from drf_yasg import openapi
|
|
from chunyu_project.common_schemas import success_response, error_response, unauthorized_response, not_found_response
|
|
|
|
|
|
@method_decorator(never_cache, name='dispatch')
|
|
class LoginRecordListAPIView(APIView):
|
|
permission_classes = [IsAuthenticated]
|
|
|
|
@swagger_auto_schema(
|
|
tags=['登录记录'],
|
|
operation_summary='获取登录记录列表',
|
|
operation_description='获取当前用户的登录记录列表，支持按状态、时间范围筛选和分页',
|
|
manual_parameters=[
|
|
openapi.Parameter('page', openapi.IN_QUERY, description='页码', type=openapi.TYPE_INTEGER),
|
|
openapi.Parameter('page_size', openapi.IN_QUERY, description='每页数量', type=openapi.TYPE_INTEGER),
|
|
openapi.Parameter('status', openapi.IN_QUERY, description='登录状态筛选', type=openapi.TYPE_STRING),
|
|
openapi.Parameter('time_range', openapi.IN_QUERY, description='时间范围(today/yesterday/7days/30days)', type=openapi.TYPE_STRING),
|
|
openapi.Parameter('start_date', openapi.IN_QUERY, description='开始日期(YYYY-MM-DD)', type=openapi.TYPE_STRING),
|
|
openapi.Parameter('end_date', openapi.IN_QUERY, description='结束日期(YYYY-MM-DD)', type=openapi.TYPE_STRING),
|
|
],
|
|
responses={200: success_response, 401: unauthorized_response},
|
|
)
|
|
async def get(self, request):
|
|
user = request.user
|
|
page = int(request.query_params.get('page', 1))
|
|
page_size = int(request.query_params.get('page_size', 10))
|
|
filter_status = request.query_params.get('status', None)
|
|
time_range = request.query_params.get('time_range', None)
|
|
start_date = request.query_params.get('start_date', None)
|
|
end_date = request.query_params.get('end_date', None)
|
|
|
|
records = LoginRecord.objects.filter(user=user)
|
|
|
|
if filter_status:
|
|
records = records.filter(status=filter_status)
|
|
|
|
now = timezone.now()
|
|
if time_range == 'today':
|
|
today_start = now.replace(hour=0, minute=0, second=0, microsecond=0)
|
|
records = records.filter(login_time__gte=today_start)
|
|
elif time_range == 'yesterday':
|
|
today_start = now.replace(hour=0, minute=0, second=0, microsecond=0)
|
|
yesterday_start = today_start - timedelta(days=1)
|
|
records = records.filter(login_time__gte=yesterday_start, login_time__lt=today_start)
|
|
elif time_range == '7days':
|
|
days_7 = now - timedelta(days=7)
|
|
records = records.filter(login_time__gte=days_7)
|
|
elif time_range == '30days':
|
|
days_30 = now - timedelta(days=30)
|
|
records = records.filter(login_time__gte=days_30)
|
|
elif start_date and end_date:
|
|
records = records.filter(login_time__date__gte=start_date, login_time__date__lte=end_date)
|
|
|
|
total = await records.acount()
|
|
start = (page - 1) * page_size
|
|
end = start + page_size
|
|
|
|
items = [item async for item in records[start:end]]
|
|
serializer = LoginRecordSerializer(items, many=True)
|
|
|
|
return create_standardized_response(
|
|
data={
|
|
'records': await serializer.adata,
|
|
'total': total,
|
|
'page': page,
|
|
'page_size': page_size,
|
|
'total_pages': (total + page_size - 1) // page_size,
|
|
},
|
|
code=ResponseCode.SUCCESS,
|
|
status_code=status.HTTP_200_OK
|
|
)
|
|
|
|
|
|
@method_decorator(never_cache, name='dispatch')
|
|
class LoginRecordDeleteAPIView(APIView):
|
|
permission_classes = [IsAuthenticated]
|
|
|
|
@swagger_auto_schema(
|
|
tags=['登录记录'],
|
|
operation_summary='删除登录记录',
|
|
operation_description='删除指定的登录记录',
|
|
manual_parameters=[
|
|
openapi.Parameter('pk', openapi.IN_PATH, description='记录ID', type=openapi.TYPE_INTEGER),
|
|
],
|
|
responses={200: success_response, 401: unauthorized_response, 404: not_found_response},
|
|
)
|
|
async def delete(self, request, pk):
|
|
user = request.user
|
|
try:
|
|
record = await LoginRecord.objects.aget(pk=pk, user=user)
|
|
await record.adelete()
|
|
return create_standardized_response(
|
|
code=ResponseCode.SUCCESS,
|
|
status_code=status.HTTP_200_OK
|
|
)
|
|
except LoginRecord.DoesNotExist:
|
|
return create_standardized_error_response(
|
|
message='记录不存在',
|
|
code=ResponseCode.NOT_FOUND,
|
|
status_code=status.HTTP_404_NOT_FOUND
|
|
)
|
|
|
|
|
|
@method_decorator(never_cache, name='dispatch')
|
|
class LoginRecordClearAPIView(APIView):
|
|
permission_classes = [IsAuthenticated]
|
|
|
|
@swagger_auto_schema(
|
|
tags=['登录记录'],
|
|
operation_summary='清空登录记录',
|
|
operation_description='清空当前用户的所有登录记录',
|
|
responses={200: success_response, 401: unauthorized_response},
|
|
)
|
|
async def delete(self, request):
|
|
user = request.user
|
|
deleted_count, _ = await LoginRecord.objects.filter(user=user).adelete()
|
|
return create_standardized_response(
|
|
data={'deleted_count': deleted_count},
|
|
code=ResponseCode.SUCCESS,
|
|
status_code=status.HTTP_200_OK
|
|
)
|