import requests import time import json from rest_framework import status, generics, permissions from rest_framework.response import Response from rest_framework.views import APIView from rest_framework.viewsets import ModelViewSet, ViewSet, GenericViewSet from rest_framework.exceptions import APIException from rest_framework.permissions import IsAuthenticated, AllowAny from rest_framework.authentication import SessionAuthentication, BasicAuthentication, TokenAuthentication from rest_framework.decorators import api_view, permission_classes, authentication_classes, action from rest_framework.authtoken.models import Token from django.utils import timezone from django.views.decorators.csrf import csrf_exempt from django.utils.decorators import method_decorator from django.http import JsonResponse from django.contrib.auth import authenticate, login, logout, get_user_model from django.middleware.csrf import get_token from django.core.exceptions import ValidationError from django.utils.dateparse import parse_datetime from .models import Prediction, User, Satellite, SavedPoint, SavedRateProfile, PreditctionTemplate, TelemetryPacket from .serializers import PredictionSerializer, TelemetryPacketSerializer, PredictionRequestSerializer, PredictionListSerializer, PredictionDetailSerializer, SavedPointSerializer, SavedRateProfileSerializer, PreditctionTemplateSerializer, UserSerializer, ChangePasswordSerializer, DeleteAccountSerializer from .services.tawhiri import TawhiriClient from drf_spectacular.utils import extend_schema from .permissions import ReadOnlyOrAuthenticated, IsOwner from .custom_pagination import CustomLimitOffsetPagination from datetime import datetime User = get_user_model() def get_prediction_from_tawhiri(params): base_url = "https://fly.stratonautica.ru/api/v2" response = requests.get(base_url, params=params) if response.status_code == 200: return response.json() # получаем результат предсказания else: raise Exception( f"Tawhiri error: {response.status_code} {response.text}") class PredictionViewSet(GenericViewSet): permission_classes = [IsAuthenticated] pagination_class = CustomLimitOffsetPagination def list(self, request): queryset = Prediction.objects.filter(user=request.user) return Response(PredictionSerializer(queryset, many=True).data) def create(self, request): serializer = PredictionRequestSerializer(data=request.data) if not serializer.is_valid(): return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) validated_data = serializer.validated_data try: prediction_result = TawhiriClient.get_prediction(validated_data) except requests.RequestException as e: return Response({"error": f"Tawhiri error: {str(e)}"}, status=status.HTTP_502_BAD_GATEWAY) # prediction = Prediction.objects.create( # result=prediction_result, user=request.user, request=request.data, validated_data=validated_data) prediction = serializer.save( user=request.user, result=prediction_result, request=request.data ) return Response({ "id": prediction.id, "created_at": prediction.created_at, "result": prediction_result }, status=status.HTTP_201_CREATED) @action(detail=False, methods=['get']) def list_user(self, request): user = request.user satellite_id = request.query_params.get('satellite_id') created_from = request.query_params.get('created_from') created_till = request.query_params.get('created_till') filters = { 'user': user, } if created_from: filters['created_at__gte'] = parse_datetime(created_from) if created_till: filters['created_at__lte'] = parse_datetime(created_till) if satellite_id: if not user.satellites.filter(id=satellite_id).exists(): return Response({'detail': 'Access denied'}, status=403) filters['satellite_id'] = satellite_id queryset = Prediction.objects.filter(**filters) queryset = self.filter_queryset(queryset) page = self.paginate_queryset(queryset) if page is not None: serializer = PredictionSerializer(page, many=True) return self.get_paginated_response(serializer.data) serializer = PredictionSerializer(queryset, many=True) return Response(serializer.data) @action(detail=False, methods=["get"]) def history(self, request): queryset = Prediction.objects.filter(user=request.user) return Response(PredictionListSerializer(queryset, many=True).data) @action(detail=True, methods=["get"]) def detail(self, request, pk=None): prediction = Prediction.objects.filter( user=request.user, pk=pk).first() if not prediction: return Response({'detail': 'Not found'}, status=404) return Response(PredictionDetailSerializer(prediction).data) @action(detail=True, methods=["delete"]) def delete(self, request, pk=None): prediction = Prediction.objects.filter( user=request.user, pk=pk).first() if not prediction: return Response({'detail': 'Not found'}, status=404) prediction.delete() return Response(status=204) class TelemetryListCreateView(generics.ListCreateAPIView): serializer_class = TelemetryPacketSerializer permission_classes = [permissions.AllowAny] def get_queryset(self): qs = TelemetryPacket.objects.filter(satellite_id=self.kwargs["pk"]) from_ts = self.request.query_params.get("from") till_ts = self.request.query_params.get("till") if from_ts: qs = qs.filter(timestamp__gte=int(from_ts)) if till_ts: qs = qs.filter(timestamp__lte=int(till_ts)) return qs.order_by("-timestamp") def post(self, request, pk): serializer = TelemetryPacketSerializer(data=request.data) if not serializer.is_valid(): return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) validated_data = serializer.validated_data TelemetryPacket.objects.create(timestamp=time.time(), satellite=Satellite.objects.get(id=pk), lat=validated_data["lat"], lon=validated_data["lon"], alt=validated_data["alt"], payload=validated_data['payload'], ) return Response(serializer.errors, status=status.HTTP_201_CREATED) class SessionView(APIView): permission_classes = [IsAuthenticated] @staticmethod def get(request, format=None): return JsonResponse({'isAuthenticated': True}) class WhoAmIView(APIView): permission_classes = [IsAuthenticated] @staticmethod def get(request, format=None): return JsonResponse({'username': request.user.username}) @extend_schema(methods=["GET"], description="Get CSRF token") @csrf_exempt @api_view(["GET"]) @permission_classes([AllowAny]) def get_csrf(request): response = JsonResponse({'detail': 'CSRF cookie set'}) response['X-CSRFToken'] = get_token(request) return response @extend_schema(methods=["POST"], description="Login user") @csrf_exempt @api_view(["POST"]) @authentication_classes([BasicAuthentication]) @permission_classes([AllowAny]) def login_view(request): data = json.loads(request.body) username = data.get('username') password = data.get('password') if username is None or password is None: return JsonResponse({'detail': 'Please provide username and password.'}, status=400) user = authenticate(username=username, password=password) if user is None: return JsonResponse({'detail': 'Invalid credentials.'}, status=400) login(request, user) return JsonResponse({'detail': 'Successfully logged in.'}) @extend_schema(methods=["POST"], description="Logout user") @api_view(["POST"]) @permission_classes([AllowAny]) def logout_view(request): if not request.user.is_authenticated: return JsonResponse({'detail': 'You\'re not logged in.'}, status=400) logout(request) return JsonResponse({'detail': 'Successfully logged out.'}) class SavedPointViewset(ModelViewSet): permission_classes = [IsOwner] serializer_class = SavedPointSerializer pagination_class = None def get_queryset(self): return SavedPoint.objects.filter(user=self.request.user) def perform_create(self, serializer): serializer.save(user=self.request.user) class PreditctionTemplateViewset(ModelViewSet): permission_classes = [IsOwner] serializer_class = PreditctionTemplateSerializer pagination_class = None def get_queryset(self): return PreditctionTemplate.objects.filter(user=self.request.user) def perform_create(self, serializer): serializer.save(user=self.request.user) class UserProfileView(APIView): permission_classes = [IsAuthenticated] def get(self, request): serializer = UserSerializer(request.user) return Response(serializer.data) def patch(self, request): user = request.user serializer = UserSerializer(user, data=request.data, partial=True) if not serializer.is_valid(): return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) serializer.save() return Response(serializer.data) class ChangePasswordView(APIView): permission_classes = [IsAuthenticated] def post(self, request): user = request.user serializer = ChangePasswordSerializer(data=request.data) if not serializer.is_valid(): return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) if not user.check_password(serializer.validated_data['old_password']): return Response({'detail': 'Old password is incorrect'}, status=status.HTTP_400_BAD_REQUEST) user.set_password(serializer.validated_data['new_password']) user.save() return Response({'detail': 'Password changed successfully'}) class DeleteAccountView(APIView): permission_classes = [IsAuthenticated] def delete(self, request): user = request.user serializer = DeleteAccountSerializer(data=request.data) if not serializer.is_valid(): return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) if not user.check_password(serializer.validated_data['password']): return Response({'detail': 'Incorrect password'}, status=status.HTTP_400_BAD_REQUEST) Prediction.objects.filter(user=user).delete() SavedPoint.objects.filter(user=user).delete() PreditctionTemplate.objects.filter(user=user).delete() user.delete() return Response({'detail': 'Account deleted successfully'}) class DeleteUserDataView(APIView): permission_classes = [IsAuthenticated] def delete(self, request): user = request.user Prediction.objects.filter(user=user).delete() SavedPoint.objects.filter(user=user).delete() PreditctionTemplate.objects.filter(user=user).delete() return Response({'detail': 'All user data deleted successfully'}) class TokenManagementView(APIView): permission_classes = [IsAuthenticated] def get(self, request): token, created = Token.objects.get_or_create(user=request.user) return Response({"token": token.key}) def post(self, request): Token.objects.filter(user=request.user).delete() token = Token.objects.create(user=request.user) return Response({"token": token.key}) # class PredictionCreateView(APIView): # permission_classes = [IsAuthenticated] # class TelemetryPacket(models.Model): # id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) # satellite = models.ForeignKey(Satellite, on_delete=models.CASCADE, related_name="telemetry") # timestamp = models.BigIntegerField() # unix time # lat = models.FloatField() # lon = models.FloatField() # alt = models.FloatField() # payload = models.JSONField(null=True, blank=True) # created_at = models.DateTimeField(auto_now_add=True) # fields = ['id', 'timestamp', 'lat', 'lon', 'alt', 'payload']