487 lines
17 KiB
Python
487 lines
17 KiB
Python
from datetime import timedelta
|
|
from django.db import transaction
|
|
from django.shortcuts import get_object_or_404
|
|
from django.utils import timezone
|
|
from rest_framework import status, generics
|
|
from rest_framework.views import APIView
|
|
from django.conf import settings
|
|
from accounts.models import IAmPrincipalLocation
|
|
from accounts.permission import IsOwnerOrReadOnly
|
|
from goodtimes import constants
|
|
from django.db.models import Q
|
|
from goodtimes import services
|
|
from goodtimes.utils import ApiResponse, CapacityError
|
|
from rest_framework.permissions import IsAuthenticated
|
|
from rest_framework_simplejwt.authentication import JWTAuthentication
|
|
from manage_events.api.serializers import (
|
|
EventMasterSearchSerializer,
|
|
EventMasterSerializer,
|
|
CreateEventSerializer,
|
|
CreateVenueSerializer,
|
|
EventCategorySerializer,
|
|
EventDetailSerializer,
|
|
IAmPrincipalLocationSerializer,
|
|
PrincipalPreferenceSerializer,
|
|
VenueSerializer,
|
|
)
|
|
from manage_events.models import (
|
|
EventInteractionType,
|
|
EventMaster,
|
|
Event,
|
|
EventCategory,
|
|
EventPrincipalInteraction,
|
|
Favorites,
|
|
PrincipalPreference,
|
|
Venue,
|
|
)
|
|
import requests
|
|
|
|
from manage_events.utils import filter_events_by_location
|
|
|
|
|
|
class CreateEventApi(APIView):
|
|
authentication_classes = [JWTAuthentication]
|
|
permission_classes = [IsAuthenticated]
|
|
|
|
def post(self, request):
|
|
serializer = CreateEventSerializer(data=request.data)
|
|
serializer.is_valid(raise_exception=True)
|
|
|
|
serializer.save(created_by=self.request.user)
|
|
|
|
# Add additional logic for handling other relationships (e.g., Venue)
|
|
return ApiResponse.success(
|
|
status=status.HTTP_201_CREATED,
|
|
message=constants.SUCCESS,
|
|
data=serializer.data,
|
|
)
|
|
|
|
|
|
class CreateVenueApi(APIView):
|
|
authentication_classes = [JWTAuthentication]
|
|
permission_classes = [IsAuthenticated]
|
|
|
|
def post(self, request):
|
|
serializer = VenueSerializer(data=request.data, context={"request": request})
|
|
serializer.is_valid(raise_exception=True)
|
|
|
|
serializer.save(created_by=self.request.user, active=True)
|
|
|
|
# Add additional logic for handling other relationships (e.g., Venue)
|
|
return ApiResponse.success(
|
|
status=status.HTTP_201_CREATED,
|
|
message=constants.SUCCESS,
|
|
data=serializer.data,
|
|
)
|
|
|
|
|
|
class EventsAPIView(APIView):
|
|
authentication_classes = [JWTAuthentication]
|
|
permission_classes = [IsAuthenticated]
|
|
|
|
def get(self, request, *args, **kwargs):
|
|
filter = request.query_params.get("filter", None)
|
|
category_id = request.query_params.get("category_id", None)
|
|
params = ["expensive", "cheap", "preference", "today", "tomorrow", "category"]
|
|
if filter not in params:
|
|
return ApiResponse.error(
|
|
status=status.HTTP_400_BAD_REQUEST,
|
|
message=constants.FAILURE,
|
|
errors="No filter found",
|
|
)
|
|
|
|
try:
|
|
if filter == "today":
|
|
events = services.EventFilterService.filter_events_for_today()
|
|
elif filter == "tomorrow":
|
|
events = services.EventFilterService.filter_events_for_tomorrow()
|
|
elif filter == "category" and category_id is not None:
|
|
events = services.EventFilterService.filter_events_by_category(
|
|
int(category_id)
|
|
)
|
|
else:
|
|
events = services.EventFilterService.filter_events(
|
|
filter, principal=request.user
|
|
)
|
|
serializer = EventDetailSerializer(
|
|
events, context={"request": request}, many=True
|
|
)
|
|
return ApiResponse.success(
|
|
status=status.HTTP_200_OK,
|
|
message=constants.SUCCESS,
|
|
data=serializer.data,
|
|
)
|
|
except Exception as e:
|
|
return ApiResponse.error(
|
|
status=status.HTTP_400_BAD_REQUEST,
|
|
message=constants.FAILURE,
|
|
errors=str(e),
|
|
)
|
|
|
|
|
|
class PrinciaplPreferenceEventsAPIView(APIView):
|
|
authentication_classes = [JWTAuthentication]
|
|
permission_classes = [IsAuthenticated]
|
|
|
|
def get(self, request, *args, **kwargs):
|
|
today = timezone.now().date()
|
|
try:
|
|
preferences = PrincipalPreference.objects.get(principal=request.user)
|
|
preferred_categories_ids = preferences.preferred_categories.values_list(
|
|
"id", flat=True
|
|
)
|
|
# Filter events based on user preferences and that are upcoming or ongoing
|
|
events = Event.objects.filter(
|
|
category__in=preferred_categories_ids, end_date__gte=today
|
|
).distinct()
|
|
|
|
if not events.exists():
|
|
# If no events found based on preferences, get future events
|
|
events = Event.objects.filter(start_date__gt=today)
|
|
serializer = EventDetailSerializer(events, many=True)
|
|
return ApiResponse.success(
|
|
status=status.HTTP_200_OK,
|
|
message=constants.SUCCESS,
|
|
data=serializer.data,
|
|
)
|
|
|
|
except PrincipalPreference.DoesNotExist:
|
|
# If the user has no preferences, default to future events
|
|
events = Event.objects.filter(start_date__gt=today)
|
|
|
|
|
|
class EventDetailAPIView(APIView):
|
|
authentication_classes = [JWTAuthentication]
|
|
permission_classes = [IsAuthenticated]
|
|
|
|
def get(self, request, pk):
|
|
try:
|
|
event = Event.objects.get(pk=pk)
|
|
interaction = services.InteractionCalculator(event)
|
|
interactions = interaction.calculate()
|
|
|
|
# Serialize your event data here
|
|
event_data = EventDetailSerializer(event, context={"request": request}).data
|
|
event_data["interactions"] = interactions
|
|
|
|
return ApiResponse.success(
|
|
status=status.HTTP_200_OK,
|
|
message=constants.SUCCESS,
|
|
data=event_data,
|
|
)
|
|
except Event.DoesNotExist:
|
|
return ApiResponse.error(
|
|
status=status.HTTP_404_NOT_FOUND,
|
|
message=constants.FAILURE,
|
|
errors=constants.RECORD_NOT_FOUND,
|
|
)
|
|
except CapacityError as e:
|
|
return ApiResponse.error(
|
|
status=status.HTTP_405_METHOD_NOT_ALLOWED,
|
|
message=constants.FAILURE,
|
|
errors=str(e),
|
|
)
|
|
|
|
|
|
class VenueListView(generics.ListAPIView):
|
|
serializer_class = VenueSerializer
|
|
permission_classes = [IsAuthenticated]
|
|
|
|
def get_queryset(self):
|
|
# Ensures that a user sees only their venues
|
|
return Venue.objects.filter(created_by=self.request.user)
|
|
|
|
def list(self, request, *args, **kwargs):
|
|
queryset = self.get_queryset()
|
|
serializer = self.get_serializer(queryset, many=True)
|
|
# Customizing the response format
|
|
return ApiResponse.success(
|
|
status=status.HTTP_200_OK,
|
|
message=constants.SUCCESS,
|
|
data=serializer.data,
|
|
)
|
|
|
|
|
|
class VenueDetailView(generics.RetrieveUpdateDestroyAPIView):
|
|
queryset = Venue.objects.all()
|
|
serializer_class = VenueSerializer
|
|
permission_classes = [IsAuthenticated, IsOwnerOrReadOnly]
|
|
|
|
def get_queryset(self):
|
|
# This ensures a user can only access their own venues
|
|
return self.queryset.filter(created_by=self.request.user)
|
|
|
|
|
|
class GeocodeAPIView(APIView):
|
|
authentication_classes = []
|
|
permission_classes = []
|
|
"""
|
|
API View to fetch latitude and longitude for a given address using the Google Maps Geocoding API.
|
|
"""
|
|
|
|
def get(self, request, *args, **kwargs):
|
|
# Extract 'address' from query parameters
|
|
address = request.query_params.get("address")
|
|
print("address: ", address)
|
|
if not address:
|
|
return ApiResponse.error(
|
|
{"error": "Address parameter is missing."},
|
|
status=status.HTTP_400_BAD_REQUEST,
|
|
)
|
|
|
|
# Call the get_lat_long method
|
|
lat_long = self.get_lat_long(address)
|
|
if lat_long:
|
|
return ApiResponse.success(
|
|
{"latitude": lat_long[0], "longitude": lat_long[1]}
|
|
)
|
|
else:
|
|
return ApiResponse.error(
|
|
{"error": "Failed to fetch latitude and longitude."},
|
|
status=status.HTTP_404_NOT_FOUND,
|
|
)
|
|
|
|
def get_lat_long(self, address):
|
|
"""
|
|
Fetches latitude and longitude for a given address using the Google Maps Geocoding API.
|
|
"""
|
|
url = "https://maps.googleapis.com/maps/api/geocode/json?"
|
|
params = {
|
|
"address": address,
|
|
# "key": settings.GOOGLE_MAPS_API_KEY, # Replace with your actual API key
|
|
"key": "AIzaSyCQv-Cfzkh3cXerrui55oId7CDHhuIImhc", # Replace with your actual API key
|
|
}
|
|
|
|
response = requests.get(url, params=params)
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
if data["status"] == "OK":
|
|
location = data["results"][0]["geometry"]["location"]
|
|
latitude = location["lat"]
|
|
longitude = location["lng"]
|
|
return latitude, longitude
|
|
else:
|
|
print(f"Geocoding failed: {data['status']}")
|
|
return None
|
|
else:
|
|
print(f"API request failed with status code: {response.status_code}")
|
|
return None
|
|
|
|
|
|
class EventCategoryListAPIView(generics.ListAPIView):
|
|
authentication_classes = [JWTAuthentication]
|
|
permission_classes = [IsAuthenticated]
|
|
queryset = EventCategory.objects.all()
|
|
serializer_class = EventCategorySerializer
|
|
|
|
|
|
class IAmPrincipalLocationAPIView(APIView):
|
|
authentication_classes = [JWTAuthentication]
|
|
permission_classes = [IsAuthenticated]
|
|
|
|
def get(self, request, *args, **kwargs):
|
|
# Assuming there's a UserLocation model that stores user locations,
|
|
# and it has a ForeignKey to the user model.
|
|
try:
|
|
user_location = IAmPrincipalLocation.objects.filter(
|
|
principal=request.user
|
|
).last()
|
|
print("user_location: ", user_location)
|
|
serializer = IAmPrincipalLocationSerializer(user_location)
|
|
return ApiResponse.success(
|
|
status=status.HTTP_200_OK,
|
|
message=constants.SUCCESS,
|
|
data=serializer.data,
|
|
)
|
|
except IAmPrincipalLocation.DoesNotExist:
|
|
return ApiResponse.error(
|
|
status=status.HTTP_404_NOT_FOUND,
|
|
message="Location not found for the user.",
|
|
errors="No location data available for the current user.",
|
|
)
|
|
|
|
def post(self, request, *args, **kwargs):
|
|
serializer = IAmPrincipalLocationSerializer(
|
|
data=request.data, context={"request": request}
|
|
)
|
|
print("serializer: ", serializer)
|
|
if serializer.is_valid():
|
|
serializer.save()
|
|
return ApiResponse.success(
|
|
status=status.HTTP_200_OK,
|
|
message=constants.SUCCESS,
|
|
data=serializer.data,
|
|
)
|
|
return ApiResponse.error(
|
|
status=status.HTTP_400_BAD_REQUEST,
|
|
message=constants.FAILURE,
|
|
errors=serializer.errors,
|
|
)
|
|
|
|
|
|
class PrincipalPreferenceView(APIView):
|
|
authentication_classes = [JWTAuthentication]
|
|
permission_classes = [IsAuthenticated]
|
|
|
|
def post(self, request, *args, **kwargs):
|
|
serializer = PrincipalPreferenceSerializer(
|
|
data=request.data, context={"request": request}
|
|
)
|
|
if serializer.is_valid():
|
|
serializer.save()
|
|
return ApiResponse.success(
|
|
message=constants.SUCCESS,
|
|
data=serializer.data,
|
|
status=status.HTTP_201_CREATED,
|
|
)
|
|
return ApiResponse.error(
|
|
message=constants.FAILURE,
|
|
errors=serializer.errors,
|
|
status=status.HTTP_400_BAD_REQUEST,
|
|
)
|
|
|
|
|
|
class PrincipalPreferenceDetailView(generics.RetrieveAPIView):
|
|
authentication_classes = [JWTAuthentication]
|
|
permission_classes = [IsAuthenticated]
|
|
serializer_class = PrincipalPreferenceSerializer
|
|
|
|
def get_object(self):
|
|
# Attempt to retrieve the authenticated user's preferences
|
|
user = self.request.user
|
|
obj, created = PrincipalPreference.objects.get_or_create(principal=user)
|
|
return obj
|
|
|
|
|
|
class EventMasterSearchAPIView(APIView):
|
|
authentication_classes = [JWTAuthentication]
|
|
permission_classes = [IsAuthenticated]
|
|
|
|
def post(self, request, *args, **kwargs):
|
|
serializer = EventMasterSearchSerializer(data=request.data)
|
|
if serializer.is_valid():
|
|
title = serializer.validated_data.get("title")
|
|
# Search for existing brands
|
|
existing_event_master = EventMaster.objects.filter(title__icontains=title)
|
|
if existing_event_master.exists():
|
|
# Return existing brands
|
|
return ApiResponse.success(
|
|
data=EventMasterSerializer(existing_event_master, many=True).data,
|
|
message=constants.SUCCESS,
|
|
status=status.HTTP_200_OK,
|
|
)
|
|
else:
|
|
# No brand found, return an empty list
|
|
return ApiResponse.success(
|
|
data=[],
|
|
message="No related EventMaster Found",
|
|
status=status.HTTP_200_OK,
|
|
)
|
|
return ApiResponse.error(
|
|
message=constants.FAILURE,
|
|
errors=serializer.errors,
|
|
status=status.HTTP_400_BAD_REQUEST,
|
|
)
|
|
|
|
|
|
class ToggleFavoriteView(APIView):
|
|
authentication_classes = [JWTAuthentication]
|
|
permission_classes = [IsAuthenticated]
|
|
|
|
def post(self, request, event_id, format=None):
|
|
principal = request.user
|
|
event = get_object_or_404(Event, pk=event_id)
|
|
|
|
favorite, created = Favorites.objects.get_or_create(
|
|
principal=principal, event=event
|
|
)
|
|
if not created:
|
|
favorite.delete()
|
|
return ApiResponse.success(
|
|
data="removed",
|
|
message=constants.SUCCESS,
|
|
status=status.HTTP_200_OK,
|
|
)
|
|
|
|
return ApiResponse.success(
|
|
data="added",
|
|
message=constants.SUCCESS,
|
|
status=status.HTTP_201_CREATED,
|
|
)
|
|
|
|
|
|
class EventStatusUpdateAPIView(APIView):
|
|
authentication_classes = [JWTAuthentication]
|
|
permission_classes = [IsAuthenticated]
|
|
|
|
def post(self, request, event_id):
|
|
principal = (
|
|
request.user
|
|
) # Assuming you're using some form of user authentication
|
|
event = get_object_or_404(Event, pk=event_id)
|
|
|
|
status_requested = request.data.get("status") # 'going' or 'interested'
|
|
if status_requested not in EventInteractionType.values:
|
|
return ApiResponse.error(
|
|
message=constants.FAILURE,
|
|
errors="Invalid status provided.",
|
|
status=status.HTTP_400_BAD_REQUEST,
|
|
)
|
|
|
|
interaction, created = EventPrincipalInteraction.objects.update_or_create(
|
|
principal=principal,
|
|
event=event,
|
|
defaults={"status": status_requested},
|
|
)
|
|
|
|
if created:
|
|
return ApiResponse.success(
|
|
data=status_requested,
|
|
message=f"Marked as {status_requested}.",
|
|
status=status.HTTP_201_CREATED,
|
|
)
|
|
else:
|
|
return ApiResponse.success(
|
|
data=status_requested,
|
|
message=f"Updated status to {status_requested}.",
|
|
status=status.HTTP_200_OK,
|
|
)
|
|
|
|
|
|
class EventFilterByLocationAPIView(APIView):
|
|
authentication_classes = [JWTAuthentication]
|
|
permission_classes = [IsAuthenticated]
|
|
|
|
def get(self, request, *args, **kwargs):
|
|
principal = request.user
|
|
|
|
# Fetching user location from IAmPrincipalLocation
|
|
try:
|
|
user_location = IAmPrincipalLocation.objects.filter(principal=principal).last()
|
|
user_lat = user_location.latitude
|
|
user_lon = user_location.longitude
|
|
except IAmPrincipalLocation.DoesNotExist:
|
|
return ApiResponse.error(
|
|
errors=constants.FAILURE,
|
|
message="User location not set.",
|
|
status=status.HTTP_400_BAD_REQUEST,
|
|
)
|
|
|
|
# Now filter events based on user location
|
|
events = filter_events_by_location(user_lat, user_lon)
|
|
print("events: ", events)
|
|
# Assuming you only want current and future events
|
|
today = timezone.now().date()
|
|
events = events.filter(Q(end_date__gte=today) | Q(start_date__gte=today))
|
|
|
|
# Serialize and return the filtered events
|
|
serializer = EventDetailSerializer(
|
|
events, context={"request": request}, many=True
|
|
)
|
|
return ApiResponse.success(
|
|
data=serializer.data,
|
|
message=constants.SUCCESS,
|
|
status=status.HTTP_200_OK,
|
|
)
|