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 from rest_framework.permissions import IsAuthenticated, AllowAny from rest_framework.authentication import SessionAuthentication, BasicAuthentication from rest_framework.decorators import api_view, permission_classes, authentication_classes, action 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.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 from .services.tawhiri import TawhiriClient from drf_spectacular.utils import extend_schema from .permissions import ReadOnlyOrAuthenticated, IsOwner 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(ViewSet): permission_classes = [IsAuthenticated] 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) 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, 'deleted_at__isnull': True } 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) return Response(PredictionSerializer(queryset, many=True).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): authentication_classes = [SessionAuthentication, BasicAuthentication] permission_classes = [IsAuthenticated] @staticmethod def get(request, format=None): return JsonResponse({'isAuthenticated': True}) class WhoAmIView(APIView): authentication_classes = [SessionAuthentication, BasicAuthentication] permission_classes = [IsAuthenticated] @staticmethod def get(request, format=None): return JsonResponse({'username': request.user.username}) @extend_schema(methods=["GET"], description="Get CSRF token") @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([]) @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): authentication_classes = [SessionAuthentication, BasicAuthentication] 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 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']