import logging import requests from rest_framework.views import APIView from rest_framework.generics import ListAPIView from rest_framework.response import Response from rest_framework.permissions import IsAuthenticated, AllowAny from rest_framework.request import Request from rest_framework.response import Response from rest_framework import status from django.views.decorators.csrf import ensure_csrf_cookie from django.utils.decorators import method_decorator from django.contrib.auth import logout from django.utils import timezone from labs.api.serializers.user import ( MeSerializer, UpdateUserProfileSerializer, LogUserChatSerializer, UserLabWorkshopSerializer, UserGroupSerializer, ) from labs.models import UserChat, UserLabWorkshop from labs.user_events import UserEvent, log_user_event logger = logging.getLogger(__name__) @method_decorator(ensure_csrf_cookie, name="dispatch") class MeEndpoint(APIView): permission_classes = (IsAuthenticated,) def get(self, request: Request) -> Response: serializer = MeSerializer(request.user) return Response(serializer.data) class LogoutEndpoint(APIView): permission_classes = (AllowAny,) def post(self, request: Request) -> Response: logout(request) return Response({"message": "Logged out successfully"}) class UserProfileEndpoint(APIView): permission_classes = (IsAuthenticated,) def post(self, request: Request) -> Response: serializer = UpdateUserProfileSerializer( request.user, data=request.data, partial=True, ) serializer.is_valid(raise_exception=True) serializer.save() return Response(MeSerializer(request.user).data) class UserGroupListEndpoint(ListAPIView): permission_classes = (IsAuthenticated,) serializer_class = UserGroupSerializer def get_queryset(self): return self.request.user.groups.all() def _get_chat_title(chat_id): chat_api_url = f"https://tidb.ai/api/v1/chats/{chat_id}" response = requests.get( chat_api_url, headers={ "accept": "application/json", }, ) if response.status_code != 200: return "" chat = response.json().get("chat") if not chat: return "" title = chat.get("title") return title or "" class LogUserChatEndpoint(APIView): permission_classes = (AllowAny,) def post(self, request: Request) -> Response: serializer = LogUserChatSerializer(data=request.data) serializer.is_valid(raise_exception=True) if request.user and request.user.is_authenticated: try: user_chat = UserChat.objects.get( user=request.user, chat_id=serializer.validated_data.get("chat_id"), ) except UserChat.DoesNotExist: chat_title = _get_chat_title(serializer.validated_data.get("chat_id")) UserChat.objects.create( user=request.user, chat_id=serializer.validated_data.get("chat_id"), chat_title=chat_title[:2046], ) log_user_event( user=request.user, event_name=UserEvent.EventName.ASK_AI, event_body={ "chat_id": serializer.validated_data.get("chat_id"), "chat_title": chat_title[:100], }, ) return Response({"message": "Chat logged."}, status=status.HTTP_201_CREATED) class UserWorkshopListEndpoint(APIView): permission_classes = (IsAuthenticated,) def get(self, request: Request) -> Response: qs = UserLabWorkshop.objects.select_related("workshop").filter( user=request.user ) now = timezone.now() happening_workshops_qs = qs.filter( valid_from__lt=now, valid_until__gte=now, ) upcoming_workshops_qs = qs.filter( valid_from__gte=now, ) return Response( { "happening_workshops": UserLabWorkshopSerializer( happening_workshops_qs, many=True ).data, "upcoming_workshops": UserLabWorkshopSerializer( upcoming_workshops_qs, many=True ).data, } )