import ast import pandas as pd from django.contrib.gis import geos from rest_framework import status from rest_framework.decorators import action from rest_framework.generics import CreateAPIView, RetrieveAPIView from rest_framework.parsers import MultiPartParser from rest_framework.permissions import IsAuthenticated, IsAdminUser, AllowAny from rest_framework.response import Response from rest_framework.views import APIView from apps.station import models as station_model from apps.station.background_tasks.upload_station_csv_tasks import ( upload_station_csv_data, ) from apps.station.filters.station_filters import ( StationListFilter, StationMediaListFilter, ) from apps.station.serializers import station_serializers from apps.station.services import station_services, station_list_services from base.paginator import CustomPagination from base.permissions import AdminOrPostOnlyPermission, IsAdminToAllowCRUD from base.views import BaseModelViewSet from station_ev import settings class StationCSVUploadView(CreateAPIView): """ This is the StationCSVUploadView. It handles the CSV file upload for station data. Attributes: parser_classes (tuple): The parsers used for request data (MultiPartParser for handling file uploads). permission_classes (tuple): Permissions required to access this view (IsAuthenticated for authenticated users). Methods: post(request, *args, **kwargs): Handles the processing of uploaded CSV data. """ parser_classes = (MultiPartParser,) permission_classes = (IsAuthenticated,) serializer_class = station_serializers.StationCSVUploadSerializers def post(self, request, *args, **kwargs): """ Handles the processing of uploaded CSV data. Args: request (HttpRequest): The HTTP request containing the uploaded CSV file. *args: Variable-length argument list. **kwargs: Variable-length keyword argument list. Returns: Response: The response message indicating whether the stations were uploaded successfully. """ file = request.FILES.get("file") if not file: return Response( {"message": "File not provided"}, status=status.HTTP_400_BAD_REQUEST ) if not file.name.endswith(".csv"): return Response( {"message": "Invalid file format"}, status=status.HTTP_400_BAD_REQUEST ) # Parse the CSV file and create data for the model df = pd.read_csv(file) columns = list(df.columns) # Required column names for csv file required_columns = [ "scarp_id", "name", "address", "description", "title_description", "meta_description", "phone", "is_active", "is_24_hours_open", "is_parking_free", "cost_description", "latitude", "longitude", "amenities", "connectors", "photos", ] # Check if the required columns are present in the CSV file if all(col in columns for col in required_columns): # Change date datatype stating to original datatype df["connectors"] = df["connectors"].apply(ast.literal_eval) df["amenities"] = df["amenities"].apply(ast.literal_eval) df["photos"] = df["photos"].apply(ast.literal_eval) df.fillna(value="", inplace=True) # Get all rows as list from DataFrame json_data = [] for _, row in df.iterrows(): row_data = dict(row) row_data["coordinates"] = { "latitude": row_data["latitude"], "longitude": row_data["longitude"], } row_data["parking_cost_description"] = ( row_data["cost_description"] or None ) json_data.append(row_data) # Serialize and save the data upload_station_csv_data(json_data, request.user.id) return Response( {"message": "Stations uploaded successfully"}, status=status.HTTP_200_OK, ) else: return Response( {"message": f"File must have {', '.join(required_columns)} columns"}, status=status.HTTP_400_BAD_REQUEST, ) class StationModelViewSet(BaseModelViewSet): """ This is the StationModelViewSet class responsible for managing station data. It provides endpoints for creating, retrieving, updating, and deleting station records. Attributes: permission_classes (tuple): Specifies the permissions required to access this view. It's set to IsAuthenticated, allowing only authenticated users to interact with station records. serializer_class (station_serializers.StationMasterSerializer): The serializer class used to serialize and deserialize station data. filter_backends (list): A list of filter backends to use for filtering station data. queryset (QuerySet): The queryset of station records that this viewset operates on. filterset_class (StationListFilter): The filterset class used to filter station data. search_fields (list): A list of fields to search for when filtering. http_method_names (list): A list of the HTTP methods that this viewset supports. Methods: get(request, *args, **kwargs): Retrieves station data based on the request parameters. delete(request, *args, **kwargs): Deletes a specific station record. list(request, *args, **kwargs): Retrieves a list of station data based on the request parameters. partial_update(request, *args, **kwargs): Handles partial updates (not allowed, returns an error). destroy(request, *args, **kwargs): Handles the destruction of station records (not allowed, returns an error). delete(request, pk=None): Deletes a specific station record based on the provided primary key. station_update(request, pk=None): Updates a specific station record based on the provided primary key. """ permission_classes = (IsAdminToAllowCRUD,) serializer_class = station_serializers.StationMasterSerializer queryset = station_model.StationMaster.objects.all() filterset_class = StationListFilter search_fields = ["name", "address"] http_method_names = ["get", "delete", "patch"] def get_queryset(self): """ Get the queryset of station records based on geographic coordinates. This method filters station records based on a bounding box defined by top-right and bottom-left lat/lng coordinates. Returns: QuerySet: The filtered queryset of station records. """ queryset = self.queryset.annotate_is_rejected_station().filter( is_rejected=False ) # Get only verified stations if self.request.method not in ["DELETE"]: queryset = self.queryset.filter(is_verified=True) # Check latlng available in request top_right_latlng = self.request.GET.get("top_right_latlng") bottom_left_latlng = self.request.GET.get("bottom_left_latlng") if not (top_right_latlng and bottom_left_latlng): return queryset # Create bounding_box for get station between it try: top_right_lat, top_right_lng = map(float, top_right_latlng.split(",")) bottom_left_lat, bottom_left_lng = map(float, bottom_left_latlng.split(",")) bounding_box = geos.Polygon.from_bbox( (bottom_left_lng, bottom_left_lat, top_right_lng, top_right_lat) ) return queryset.filter(coordinates__coveredby=bounding_box) except ValueError: return queryset def list(self, request, *args, **kwargs): """ Retrieve a paginated list of station data based on request parameters. Args: request (HttpRequest): The HTTP request object. *args: Variable-length argument list. **kwargs: Arbitrary keyword arguments. Returns: Response: A paginated response containing a list of serialized station data. """ fields = [ "id", "name", "description", "address", "coordinates", "connectors", "open_time", "close_time", "is_24_hours_open", "bookmark", "rating", "assets", "slug", ] serializer = station_list_services.get_paginated_response(self, fields=fields) return Response(serializer.data) def partial_update(self, request, *args, **kwargs): """ Partially update a station record based on the provided data. Args: request (HttpRequest): The HTTP request object. *args: Variable-length argument list. **kwargs: Arbitrary keyword arguments. Returns: Response: The serialized data of the updated station record. """ station = self.get_object() serializer = station_serializers.StationCreateUpdateMasterSerializer( station, data=request.data, partial=True ) serializer.is_valid(raise_exception=True) serializer.save() return Response(serializer.data, status=status.HTTP_200_OK) @action( detail=False, methods=["get"], url_path="admin", permission_classes=[IsAdminUser], ) def get_admin_station_list(self, request, pk=None): """ Retrieve a paginated list of station data for admin purposes. Args: request (HttpRequest): The HTTP request object. pk: current object id, default null Returns: Response: A paginated response containing a list of serialized station data. """ fields = [ "id", "name", "address", "all_network_operators", "rating", "assets", "slug", ] serializer = station_list_services.get_paginated_response(self, fields=fields) return Response(serializer.data) @action( detail=False, methods=["get"], url_path="map-pins", permission_classes=[AllowAny], ) def get_map_pins(self, request, pk=None): """ Retrieve a paginated list of station data for map pins. Args: request (HttpRequest): The HTTP request object. pk: current object id, default null Returns: Response: A paginated response containing a list of serialized station data. """ fields = [ "id", "name", "address", "coordinates", "slug", ] serializer = station_list_services.get_paginated_response(self, fields=fields) return Response(serializer.data) class StationRetrieveAPIView(RetrieveAPIView): """ A view for retrieving details of a specific StationMaster object. This view allows retrieval of StationMaster objects by their slug field. It does not require authentication and can be accessed by any user (AllowAny permission). """ permission_classes = (AllowAny,) serializer_class = station_serializers.StationMasterSerializer queryset = station_model.StationMaster.objects.all() lookup_field = "slug" class StationCreateUpdateModelViewSet(BaseModelViewSet): """ This is the StationCreateUpdateModelViewSet class responsible for handling CSV file uploads for station data. It provides endpoints for uploading and updating station data from CSV files. Attributes: permission_classes (tuple): Specifies the permissions required to access this view. It's set to IsAuthenticated, allowing only authenticated users to use this view. """ permission_classes = (IsAuthenticated,) serializer_class = station_serializers.StationCreateUpdateMasterSerializer queryset = station_model.StationMaster.objects.all() http_method_names = ["post"] class StationVerificationModelViewSet(BaseModelViewSet): """ This view handles the verification and update of station data from CSV file uploads. This view is responsible for managing the uploading and updating of station data from CSV files, specifically for the purpose of station verification. Attributes: permission_classes (tuple): Specifies the permissions required to access this view. It's set to (IsAuthenticated, IsAdminUser), allowing only authenticated and admin users to use this view. serializer_class (StationVerificationSerializer): The serializer used to validate and process station data. queryset (QuerySet): The initial queryset of station records that this viewset operates on. http_method_names (list): A list of the HTTP methods that this viewset supports, including 'GET' and 'PATCH'. """ permission_classes = (IsAuthenticated, IsAdminUser) serializer_class = station_serializers.StationVerificationSerializer queryset = station_model.StationMaster.objects.all() http_method_names = ["get", "patch"] def get_queryset(self): """ Get the queryset of unverified station records. This method filters station records based on the 'is_verified' field, returning only those that are not yet verified (is_verified=False). Returns: QuerySet: The filtered queryset of unverified station records. """ return ( self.queryset.filter(is_verified=False) .annotate_is_rejected_station() .filter(is_rejected=False) ) def partial_update(self, request, *args, **kwargs): """ Partially update a station record based on the provided data. Args: request (HttpRequest): The HTTP request object. *args: Variable-length argument list. **kwargs: Arbitrary keyword arguments. Returns: Response: The serialized data of the updated station record. """ station = self.get_object() serializer = self.get_serializer(station, data=request.data, partial=True) serializer.is_valid(raise_exception=True) serializer.save() station_model.StationMedia.objects.filter(station=station).update( is_verified=True ) return Response(serializer.data, status=status.HTTP_200_OK) class StationMediaViewSet(BaseModelViewSet): """ This is the StationMediaViewSet class responsible for managing station media assets, such as images or files. It provides endpoints for uploading and deleting media assets associated with station records. Attributes: permission_classes (tuple): Specifies the permissions required to access this view. It's set to IsAuthenticated, allowing only authenticated users to use this view. parser_classes (tuple): The parsers used for request data, which includes MultiPartParser for handling file uploads. serializer_class (UploadStationMediaSerializer): The serializer class used for deserializing and handling media asset data. queryset (QuerySet): The queryset for accessing station media records. http_method_names (list): Specifies the supported HTTP methods for this view, including 'post' and 'delete'. Methods: create(request, *args, **kwargs): Handles the uploading of station media assets and associates them with a specific station. delete(request, *args, **kwargs): Handles the removal of station media assets. """ permission_classes = (IsAuthenticated, AdminOrPostOnlyPermission) serializer_class = station_serializers.UploadStationMediaSerializer queryset = station_model.StationMedia.objects.all() filterset_class = StationMediaListFilter http_method_names = ["get", "post", "patch", "delete"] def get_queryset(self): """ Get the queryset of objects filtered by the associated station. This method retrieves the value of 'station_id' from the URL kwargs and filters the queryset based on the associated station. Returns: QuerySet: The filtered queryset of objects. """ station = self.kwargs.get("station_id") return self.queryset.filter(station=station) def create(self, request, *args, **kwargs): """ Handles the uploading of station media assets and associates them with a specific station. Args: request (HttpRequest): The HTTP request object containing media assets and associated data. *args: Variable-length argument list. **kwargs: Arbitrary keyword arguments. Returns: Response: A response indicating the success or failure of the media asset upload operation. This method processes the uploaded media assets, associates them with a specified station, and handles the creation of station media records. It verifies the provided station ID, asset data, and the associated user. """ station_id = self.kwargs.get("station_id") station = station_services.get_station_by_id(station_id) if not station: return Response( {"message": "Invalid station ID"}, status=status.HTTP_400_BAD_REQUEST, ) assets = [] for index, file in enumerate(request.FILES): assets.append( { "asset": request.FILES.get(f"asset[{index}]"), "asset_type": request.data.get(f"asset_type[{index}]"), } ) if not assets: return Response( {"message": "Assets are required"}, status=status.HTTP_400_BAD_REQUEST, ) # Per user can upload only 5 images if not request.user.is_superuser: station_media = station_model.StationMedia.objects.filter( user=request.user, station=station ) if (station_media.count() + len(assets)) > 5: return Response( {"message": "You can upload up-to 5 images"}, status=status.HTTP_400_BAD_REQUEST, ) serializer = self.get_serializer( data=assets, many=True, context={"request": request} ) if serializer.is_valid(): serializer.save( station=station, user=request.user, is_verified=settings.ALLOW_AUTO_VERIFY_MEDIA, ) return Response( serializer.data, status=status.HTTP_201_CREATED, ) else: return Response( serializer.errors, status=status.HTTP_400_BAD_REQUEST, ) class StationConnectorViewSet(BaseModelViewSet): """ A viewset for handling StationConnectors. This viewset allows CRUD operations (Create, Read, Update, Delete) for StationConnectors. It requires authentication and admin user permissions for access. Attributes: permission_classes (tuple): Permissions required to access this view. serializer_class: Serializer class for handling StationConnectors. queryset: The queryset containing all StationConnectors. http_method_names (list): The allowed HTTP methods for this viewset. """ permission_classes = (IsAuthenticated, IsAdminUser) serializer_class = station_serializers.StationConnectorSerializer queryset = station_model.StationConnectors.objects.all() http_method_names = ["post", "delete", "patch"] class StationPhotoAPIView(APIView): """ API view to get a combined list of station media and rating media assets with pagination. Attributes: permission_classes (tuple): Permissions required to access this view (IsAuthenticated for authenticated users). pagination_class (CustomPagination): The custom pagination class for paginating the combined assets. serializer_class (CombinedAssetSerializer): The serializer used for serializing the combined assets. http_method_names (list): The allowed HTTP methods, in this case, only GET is allowed. Methods: get(request, *args, **kwargs): Retrieve the combined assets for a specific station based on the station ID. """ permission_classes = (AllowAny,) pagination_class = CustomPagination serializer_class = station_serializers.CombinedStationAssetSerializer http_method_names = ["get"] def get(self, request, *args, **kwargs): """ Retrieve the combined assets for a specific station based on the station ID. Args: request (HttpRequest): The HTTP request object. *args: Variable-length argument list. **kwargs: Arbitrary keyword arguments. Returns: Response: A paginated response containing the combined assets. """ # Fetch combined asset and asset_type station_id = kwargs.get("station_id") # Get station media based on station id station_media_assets = station_model.StationMedia.objects.filter( station__id=station_id, is_verified=True, ).values("asset", "asset_type", "id") # Get Station rating media based on station id station_rating_media_assets = station_model.StationRatingMedia.objects.filter( rating_station__station__id=station_id, is_verified=True, ).values("asset", "asset_type", "id") combined_assets = station_media_assets.union(station_rating_media_assets) # Iterate over the results and update the 'asset' field to contain the full URL for asset_info in combined_assets: asset_info["asset"] = f"{settings.MEDIA_URL}{asset_info['asset']}" pagination = CustomPagination() result_page = pagination.paginate_queryset(combined_assets, request) serializers = self.serializer_class( result_page, many=True, context={"request": request} ) return pagination.get_paginated_response(serializers.data) class UnverifiedStationMediaViewSet(BaseModelViewSet): """ This viewset provides endpoints to manage unverified station media. Attributes: permission_classes (tuple): Specifies the permissions required to access this view. serializer_class: The serializer class used for handling station media data. queryset (QuerySet): The queryset for accessing unverified station media records. filterset_class: The filterset class used for filtering station media records. http_method_names (list): Specifies the supported HTTP methods for this view, including 'get', 'delete', and 'patch'. """ permission_classes = (IsAuthenticated, IsAdminUser) serializer_class = station_serializers.StationMediaSerializer queryset = station_model.StationMedia.objects.all() filterset_class = StationMediaListFilter http_method_names = ["get", "delete", "patch"] @action(detail=False, methods=["delete"], url_path="delete") def delete_selected_media(self, request, *args, **kwargs): """ Delete selected unverified media. Args: request (HttpRequest): The HTTP request object. *args: Variable-length argument list. **kwargs: Arbitrary keyword arguments. Returns: Response: A response indicating the success or failure of the media deletion operation. """ media_ids = request.data.get("media_ids", []) try: # Delete selected media deleted_count, _ = station_model.StationMedia.objects.filter( id__in=media_ids ).delete() return Response( {"message": f"{deleted_count} media deleted successfully"}, status=status.HTTP_204_NO_CONTENT, ) except Exception as e: return Response( {"error": str(e)}, status=status.HTTP_500_INTERNAL_SERVER_ERROR ) @action(detail=False, methods=["patch"], url_path="verified") def verify_selected_media(self, request, *args, **kwargs): """ Verify selected unverified media. Args: request (HttpRequest): The HTTP request object. *args: Variable-length argument list. **kwargs: Arbitrary keyword arguments. Returns: Response: A response indicating the success or failure of the media verification operation. """ media_ids = request.data.get("media_ids", []) try: # Update is_verified for selected media updated_count = station_model.StationMedia.objects.filter( id__in=media_ids ).update(is_verified=True) return Response( {"message": f"{updated_count} media verified successfully"}, status=status.HTTP_200_OK, ) except Exception as e: return Response( {"error": str(e)}, status=status.HTTP_500_INTERNAL_SERVER_ERROR )