from rest_framework import status, viewsets from rest_framework.decorators import action from rest_framework.response import Response from django.core.cache import cache from .serializers import ( CatalogsSerializer, CatalogsWithFileUploadSerializer, CategoriesSerializer, SeriesSerializer, ProductsSerializer, ProductGallerySerializer, NewsAndPressReleasesSerializer, NewsGallerySerializer, NewsGalleryCustomSerializer, SettingsSerializer, FileUploadSerializer, FeaturedProductsSerializer, ProductsWithUploadImagesSerializer, ProductGalleryCustomSerializer, NewsAndPressReleasesCustomSerializer, ProductsIdSerializer, ProductsCustomSerializer, BulkProductsIdSerializer, NewArrivalProductsSerializer, ) from .models import ( Catalogs, Categories, Series, Products, ProductGallery, NewsAndPressReleases, NewsGallery, Settings, CustomTwentyPagesPagination, FeaturedProducts, NewArrivalProducts, ) from django_filters.rest_framework import DjangoFilterBackend from rest_framework.filters import SearchFilter from .permissions import IsAdminOrReadOnly, IsShopManagerOrWebAdmin from .filters import ( ProductsFilter, ProductGalleryFilter, NewsGalleryFilter, CatalogsFilter, CategoriesFilter, SeriesFilter, NewsAndPressReleasesFilter, ) import logging, json, pandas as pd, os, uuid, urllib.parse, random from .tasks import upload_products from utils.r2.client import R2 from utils.mixins.archive_mixins import ArchiveViewSetMixin from pathlib import Path from collections import defaultdict from users.models import Users from django.conf import settings from django.db.models import Count from activity_logs.models import ActivityLogs from django.db.models.query import QuerySet BASE_DIR = Path(__file__).resolve().parent.parent logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s" ) # Create your views here. class CatalogsViewset(viewsets.ModelViewSet, ArchiveViewSetMixin): queryset = Catalogs.objects.order_by("-created_at") permission_classes = [IsShopManagerOrWebAdmin] filter_backends = [DjangoFilterBackend, SearchFilter] filterset_class = CatalogsFilter pagination_class = CustomTwentyPagesPagination def get_serializer_class(self): method = self.request.method if method in ["POST", "PUT", "PATCH"]: return CatalogsWithFileUploadSerializer return CatalogsSerializer def create_or_update(self, data, instance=None): if instance: serializer = self.get_serializer(data=data, instance=instance) else: serializer = self.get_serializer(data=data) serializer.is_valid(raise_exception=True) validated_data = serializer.validated_data title = validated_data.get("title") link = validated_data.get("link") image = validated_data.get("image_path", None) r2_path = None if image: r2 = R2(file=image, main_folder_name="catalogs") r2_path = r2.upload_fileobj( sub_folder_path=f"{title.replace(" ", "_")}/", ) # always end it with / if instance: catalog: Catalogs = instance old_image_path = catalog.image_path # Add condition to check if a new image is uploaded during UPDATE. Also remove the old image if a new image is uploaded if r2_path: r2 = R2() r2.delete_object(r2_path=old_image_path) catalog.image_path = r2_path catalog.title = title catalog.link = link else: # For CREATE, r2_path is automatically included as it is required if creating a new catalog catalog = Catalogs(title=title, image_path=r2_path, link=link) catalog.save() read_serializer_data = CatalogsSerializer(catalog, many=False).data return read_serializer_data def create(self, request, *args, **kwargs): data = request.data image = data.get("image_path", None) if image is None: return Response( {"message": "No image provided. Please upload an image."}, status=status.HTTP_400_BAD_REQUEST, ) init_catalog_title = data["title"] existing_catalog_title = Catalogs.objects.filter( title__iexact=init_catalog_title ) if existing_catalog_title.exists(): return Response( { "message": f"Catalog with title '{init_catalog_title}' already exists." }, status=status.HTTP_400_BAD_REQUEST, ) serialized_created_data = self.create_or_update(data=data) ActivityLogs.record_activity( user=request.user, action=f"Created catalog {serialized_created_data['title']}", ) return Response(serialized_created_data, status=status.HTTP_201_CREATED) def update(self, request, *args, **kwargs): """ If image is not a new file upload, no need to be included in the payload. """ data = request.data instance: Catalogs = self.get_object() init_catalog_title = data["title"] existing_catalog_title = Catalogs.objects.filter( title__iexact=init_catalog_title ).exclude(id=instance.id) if existing_catalog_title.exists(): return Response( { "message": f"Catalog with title '{init_catalog_title}' already exists." }, status=status.HTTP_400_BAD_REQUEST, ) serialized_updated_data = self.create_or_update(data=data, instance=instance) ActivityLogs.record_activity( user=request.user, action=f"Created catalog {serialized_updated_data['title']}", ) return Response(serialized_updated_data, status=status.HTTP_200_OK) class CategoriesViewset(viewsets.ModelViewSet, ArchiveViewSetMixin): queryset = Categories.objects.order_by("sub_category_name") serializer_class = CategoriesSerializer permission_classes = [IsShopManagerOrWebAdmin] filter_backends = [DjangoFilterBackend, SearchFilter] filterset_class = CategoriesFilter pagination_class = CustomTwentyPagesPagination @action(detail=False, methods=["GET"], url_path="v1/all-categories") def get_unpaginated_categories_v1(self, request): """ Return a customized response in a format of:\n {\n 'navlinks': [\n {\n 'header': 'CATEGORY',\n 'link': '/products/category',\n 'links': [\n {\n 'name': 'Subcategory',\n 'path': '/products/category?sub_category=urlencoded(Subcategory)',\n }\n ]\n }\n ]\n } """ response = {"navlinks": []} main_categories_label = settings.MAIN_CATEGORIES_LABEL grouped_categories = defaultdict(list) for category_label in main_categories_label: categories = ( Categories.objects.filter( is_active=True, multiple_orderings__has_key=category_label ) .annotate(products_count=Count("category_products")) .order_by(f"multiple_orderings__{category_label}") ) for category in categories: products_count = getattr(category, "products_count", 0) if products_count > 0 and category_label in category.main_category_name: grouped_categories[category_label].append(category) for category_label in main_categories_label: category_object = { "header": category_label.upper(), "link": f"/products/{category_label.lower()}", "links": [], } for category in grouped_categories.get(category_label, []): category_object["links"].append( { "name": category.sub_category_name.title(), "path": f"/products/{category_label.lower()}?sub_category={urllib.parse.quote(category.sub_category_name.title())}", } ) response["navlinks"].append(category_object) return Response(response, status=status.HTTP_200_OK) class SeriesViewset(viewsets.ModelViewSet, ArchiveViewSetMixin): queryset = Series.objects.order_by("name") serializer_class = SeriesSerializer permission_classes = [IsShopManagerOrWebAdmin] filter_backends = [DjangoFilterBackend, SearchFilter] filterset_class = SeriesFilter pagination_class = CustomTwentyPagesPagination class NewArrivalProductsViewset(viewsets.ModelViewSet): queryset = NewArrivalProducts.objects.select_related("product").order_by( "created_at" ) permission_classes = [IsAdminOrReadOnly] pagination_class = CustomTwentyPagesPagination http_method_names = ["get", "post"] def get_serializer_class(self): method = self.request.method path = self.request.path if "bulk-update-status" in path: return BulkProductsIdSerializer if method == "POST": return ProductsIdSerializer return NewArrivalProductsSerializer def update_status(self, product_id): try: product = Products.objects.get(id=product_id) product.is_new_arrival = not product.is_new_arrival product.save() if product.is_new_arrival is False: new_arrival_product = NewArrivalProducts.objects.filter(product=product) if new_arrival_product.exists(): new_arrival_product.delete() else: NewArrivalProducts.objects.create(product=product) serialized_data = ProductsCustomSerializer(product, many=False).data return serialized_data, "OK" except Products.DoesNotExist: return f"Product with id {product_id} does not exist.", "NOT_FOUND" def create(self, request, *args, **kwargs): data = request.data serializer = self.get_serializer(data=data) serializer.is_valid(raise_exception=True) id = serializer.validated_data["id"] product_data, update_status = self.update_status(product_id=id) if update_status == "NOT_FOUND": return Response( {"message": product_data, "product": update_status}, status=status.HTTP_404_NOT_FOUND, ) return Response( { "message": "Product new arrival status updated successfully.", "product": product_data, }, status=status.HTTP_200_OK, ) @action(detail=False, methods=["POST"], url_path="bulk-update-status") def bulk_update_status(self, request): data = request.data serializer = self.get_serializer(data=data) serializer.is_valid(raise_exception=True) ids = serializer.validated_data["ids"] updated_products = [] failed_products = [] for id in ids: product_data, update_status = self.update_status(product_id=id) if update_status == "NOT_FOUND": failed_products.append(product_data) else: updated_products.append(product_data) response_status = status.HTTP_200_OK if len(updated_products) == 0 and len(failed_products) > 0: message = "No products were updated. Errors occured." response_status = status.HTTP_400_BAD_REQUEST elif len(updated_products) > 0 and len(failed_products) == 0: message = "All products were updated successfully." elif len(updated_products) > 0 and len(failed_products) > 0: message = "Some products were updated successfully, while others encountered errors." return Response( { "message": message, "updated_products": updated_products, "failed_products": failed_products, }, status=response_status, ) class ProductsViewset(viewsets.ModelViewSet, ArchiveViewSetMixin): queryset = ( Products.objects.select_related("category", "series") .prefetch_related("product_images") .order_by("category__sub_category_name", "name") ) permission_classes = [IsShopManagerOrWebAdmin] filter_backends = [DjangoFilterBackend] filterset_class = ProductsFilter pagination_class = CustomTwentyPagesPagination http_method_names = ["post", "put", "patch", "get", "delete"] def get_serializer_class(self): method = self.request.method path = self.request.path if "upload-products" in path: return FileUploadSerializer if method in ["POST", "PUT", "PATCH"]: return ProductsWithUploadImagesSerializer return ProductsSerializer def get_serializer_context(self): context = super().get_serializer_context() file_path = f"{BASE_DIR}/caracole_ph_api/app_settings.json" user = getattr(self.request, "user", None) if user and isinstance(user, Users): context["is_authenticated"] = True context["is_price_visible"] = "no" # Initial value if os.path.exists(file_path): with open(file_path, "r") as file: settings_data = json.load(file) all_settings = settings_data["settings"] price_visibility_setting = None for setting in all_settings: if setting["name"] == "Price Visibility": price_visibility_setting = setting break if price_visibility_setting: is_price_visible = price_visibility_setting.get("value", "no") context["is_price_visible"] = is_price_visible return context def update_new_arrival_status(self, product: Products): new_arrival_status = product.is_new_arrival if new_arrival_status is False: new_arrival_product = NewArrivalProducts.objects.filter(product=product) if new_arrival_product.exists(): new_arrival_product.delete() else: NewArrivalProducts.objects.create(product=product) def create_or_update( self, data, lifestyle_images, existing_images_id=None, instance=None ): if instance: serializer = self.get_serializer(data=data, instance=instance) else: serializer = self.get_serializer(data=data) serializer.is_valid(raise_exception=True) validated_data = serializer.validated_data image_path = validated_data.get("image_path", None) edp_number = validated_data["edp_number"] category_id = validated_data["category"] series_id = validated_data.get("series", None) description = validated_data.get("description", None) if description: modified_description = json.dumps({"delta": "", "html": description}) validated_data["description"] = modified_description if series_id: related_series = Series.objects.get(id=series_id) validated_data["series"] = related_series else: validated_data["series"] = None related_category = Categories.objects.get(id=category_id) validated_data["category"] = related_category if image_path: r2 = R2(file=image_path, main_folder_name="products") main_image_r2_path = r2.upload_fileobj(sub_folder_path=f"{edp_number}/") validated_data["image_path"] = main_image_r2_path validated_data.pop("lifestyle_images", None) product_lifestyle_images = [] if instance: if not existing_images_id: existing_images_id = [] else: existing_images_id = [uuid.UUID(u) for u in existing_images_id] ProductGallery.objects.filter(product__id=instance.id).exclude( id__in=existing_images_id ).delete() # Resets the Product Gallery based on updated images for attr, value in validated_data.items(): setattr(instance, attr, value) instance.save() else: instance = Products.objects.create(**validated_data) for image in lifestyle_images: r2 = R2(file=image, main_folder_name="products") lifestyle_image_r2_path = r2.upload_fileobj( sub_folder_path=f"{edp_number}/" ) product_lifestyle_images.append( ProductGallery(image_path=lifestyle_image_r2_path, product=instance) ) product_images = ProductGallery.objects.bulk_create(product_lifestyle_images) self.update_new_arrival_status(product=instance) serialized_product = ProductsSerializer(instance, many=False).data serialized_product["product_images"] = ProductGalleryCustomSerializer( product_images, many=True ).data return serialized_product def create(self, request, *args, **kwargs): """ Category and Series are ids of the objects. Lifestyle Images is a list of Files. Image Path is a File """ data = request.data lifestyle_images = data.getlist("lifestyle_images[]", None) if len(lifestyle_images) < 2: return Response( {"message": "At least 2 lifestyle images are required per product."}, status=status.HTTP_400_BAD_REQUEST, ) name = data.get("name") edp_number = data.get("edp_number") article_number = data.get("article_number") existing_product_name = Products.objects.filter(name__iexact=name) existing_product_edp = Products.objects.filter(edp_number__exact=edp_number) # existing_product_article = Products.objects.filter( # article_number__exact=article_number # ) if existing_product_name.exists(): return Response( {"message": f"Product with name {name.title()} already exists."}, status=status.HTTP_400_BAD_REQUEST, ) if existing_product_edp.exists(): return Response( {"message": f"Product with EDP number {edp_number} already exists."}, status=status.HTTP_400_BAD_REQUEST, ) # if existing_product_article.exists(): # return Response( # { # "message": f"Product with article number {article_number} already exists." # }, # status=status.HTTP_400_BAD_REQUEST, # ) try: serialized_created_data = self.create_or_update( data=data, lifestyle_images=lifestyle_images ) return Response(serialized_created_data, status=status.HTTP_201_CREATED) except Series.DoesNotExist: return Response( {"message": "Selected series does not exists."}, status=status.HTTP_404_NOT_FOUND, ) except Categories.DoesNotExist: return Response( {"message": "Selected category does not exists."}, status=status.HTTP_404_NOT_FOUND, ) def update(self, request, *args, **kwargs): """ Category and Series are ids of the objects. Lifestyle Images is a list of Files. Image Path is a File """ data = request.data instance = self.get_object() lifestyle_images = data.getlist("lifestyle_images[]", None) existing_images_id = data.getlist("existing_images_id[]", None) # error_message = "At least 2 lifestyle images are required per product." # if not existing_images_id and (lifestyle_images and len(lifestyle_images) < 2): # return Response( # {"message": error_message}, # status=status.HTTP_400_BAD_REQUEST, # ) # elif not lifestyle_images and ( # existing_images_id and len(existing_images_id) < 2 # ): # return Response( # {"message": error_message}, # status=status.HTTP_400_BAD_REQUEST, # ) edp_number = data.get("edp_number") article_number = data.get("article_number") existing_product_edp = Products.objects.filter( edp_number__exact=edp_number ).exclude(id=instance.id) # existing_product_article = Products.objects.filter( # article_number__exact=article_number # ).exclude(id=instance.id) if existing_product_edp.exists(): return Response( {"message": f"Product with EDP number {edp_number} already exists."}, status=status.HTTP_400_BAD_REQUEST, ) # if existing_product_article.exists(): # return Response( # { # "message": f"Product with article number {article_number} already exists." # }, # status=status.HTTP_400_BAD_REQUEST, # ) try: serialized_updated_data = self.create_or_update( data=data, lifestyle_images=lifestyle_images, existing_images_id=existing_images_id, instance=instance, ) return Response(serialized_updated_data, status=status.HTTP_200_OK) except Series.DoesNotExist: return Response( {"message": "Selected series does not exists."}, status=status.HTTP_404_NOT_FOUND, ) except Categories.DoesNotExist: return Response( {"message": "Selected category does not exists."}, status=status.HTTP_404_NOT_FOUND, ) @action( detail=False, methods=["GET"], url_path="related-products", ) def get_random_related_products(self, request): """ Returns array of product objects. Maximum of 10 objects. """ products = ( Products.objects.filter(is_active=True) .select_related("category", "series") .prefetch_related("product_images") ) product_ids = list(products.values_list("id", flat=True)) if len(product_ids) == 0: return Response( {"message": "No products were found."}, status=status.HTTP_200_OK ) if len(product_ids) <= 10: serializer = ProductsSerializer(products, many=True) return Response(serializer.data, status=status.HTTP_200_OK) try: random_ids = [] product_length = len(product_ids) - 1 while len(random_ids) < 10: random_number = random.randint(0, product_length) product_id = product_ids[random_number] if product_id not in random_ids: random_ids.append(product_ids[random_number]) related_products = products.filter(id__in=random_ids) serializer = ProductsSerializer(related_products, many=True) return Response( serializer.data, status=status.HTTP_200_OK, ) except Exception as e: return Response( { "message": "An error occurred while retrieving related products. Try again.", "detailed_error": str(e), }, status=status.HTTP_500_INTERNAL_SERVER_ERROR, ) @action( detail=False, methods=["POST"], url_path="upload-products", ) def upload_products(self, request): data = request.data serializer = self.get_serializer(data=data) serializer.is_valid(raise_exception=True) validated_data = serializer.validated_data file = validated_data["file"] if ".csv" in str(file): df = pd.read_csv(file) elif ".xlsx" in str(file): df = pd.read_excel(file) else: return Response( {"message": "Invalid file format. Only CSV and Excel are supported"}, status=status.HTTP_400_BAD_REQUEST, ) df = df.dropna(how="all") file_contents = df.to_dict(orient="records") upload_products.delay(file_contents=file_contents) return Response( {"message": f"File {str(file)} is being uploaded."}, status=status.HTTP_202_ACCEPTED, ) @action( detail=False, methods=["GET"], url_path="upload-status", ) def get_upload_status(self, request): # pragma: no cover upload_errors = cache.get("product_upload_error", None) upload_status = cache.get("upload_status", None) if upload_status == "error": return Response( {"message": f"Product upload failed: {upload_errors}"}, status=status.HTTP_500_INTERNAL_SERVER_ERROR, ) if upload_status == "uploading": return Response( {"message": "Product upload is in progress"}, status=status.HTTP_202_ACCEPTED, ) cache.delete("product_upload_error") cache.delete("upload_status") return Response( {"message": "Product upload is complete"}, status=status.HTTP_200_OK, ) class ProductGalleryViewset(viewsets.ModelViewSet, ArchiveViewSetMixin): queryset = ProductGallery.objects.select_related("product").order_by( "product__name" ) serializer_class = ProductGallerySerializer permission_classes = [IsShopManagerOrWebAdmin] filter_backends = [DjangoFilterBackend] filterset_class = ProductGalleryFilter pagination_class = CustomTwentyPagesPagination class NewsAndPressReleasesViewset(viewsets.ModelViewSet, ArchiveViewSetMixin): queryset = NewsAndPressReleases.objects.order_by("-created_at").prefetch_related( "news_images" ) permission_classes = [IsAdminOrReadOnly] filter_backends = [DjangoFilterBackend, SearchFilter] filterset_class = NewsAndPressReleasesFilter pagination_class = CustomTwentyPagesPagination def get_serializer_class(self): method = self.request.method if method in ["POST", "PUT", "PATCH"]: return NewsAndPressReleasesCustomSerializer return NewsAndPressReleasesSerializer def create_or_update(self, data, news_images=None, instance=None): if instance: serializer = self.get_serializer(instance=instance, data=data) else: serializer = self.get_serializer(data=data) serializer.is_valid(raise_exception=True) validated_data = serializer.validated_data image_path = validated_data.get("image_path", None) title = validated_data["title"] type = validated_data["type"] if image_path: r2 = R2(file=image_path, main_folder_name=type.lower()) r2_path = r2.upload_fileobj(sub_folder_path=f"{title}/") validated_data["image_path"] = r2_path if instance: for attr, value in validated_data.items(): setattr(instance, attr, value) instance.save() else: instance = NewsAndPressReleases.objects.create(**validated_data) news_gallery = [] if news_images: NewsGallery.objects.filter(news__id=instance.id).delete() for image in news_images: r2 = R2(file=image, main_folder_name="news") r2_path = r2.upload_fileobj(sub_folder_path=f"{title}/") news_gallery.append(NewsGallery(image_path=r2_path, news=instance)) news_gallery_images = NewsGallery.objects.bulk_create(news_gallery) serialized_data = NewsAndPressReleasesSerializer(instance, many=False).data if news_gallery: serialized_data["news"] = NewsGalleryCustomSerializer( news_gallery_images, many=True ).data return serialized_data def create(self, request, *args, **kwargs): """ Field 'type' choices: news or press """ data = request.data news_images = data.getlist("news_images[]", None) init_title = data.get("title") type = data.get("type") existing_title_by_type = NewsAndPressReleases.objects.filter( title__iexact=init_title, type=type ) if existing_title_by_type.exists(): return Response( {"message": f"Title {init_title} under {type.upper()} already exists."}, status=status.HTTP_400_BAD_REQUEST, ) serialized_created_data = self.create_or_update( data=data, news_images=news_images ) return Response(serialized_created_data, status=status.HTTP_201_CREATED) def update(self, request, *args, **kwargs): """ Field 'type' choices: news or press """ instance = self.get_object() data = request.data news_images = data.getlist("news_images[]", None) init_title = data.get("title") type = data.get("type") existing_title_by_type = NewsAndPressReleases.objects.filter( title__iexact=init_title, type=type ).exclude(id=instance.id) if existing_title_by_type.exists(): return Response( {"message": f"Title {init_title} under {type.upper()} already exists."}, status=status.HTTP_400_BAD_REQUEST, ) serialized_updated_data = self.create_or_update( data=data, news_images=news_images, instance=instance ) return Response(serialized_updated_data, status=status.HTTP_200_OK) class NewsGalleryViewset(viewsets.ModelViewSet, ArchiveViewSetMixin): queryset = NewsGallery.objects.select_related("news").order_by("-created_at") serializer_class = NewsGallerySerializer permission_classes = [IsAdminOrReadOnly] filter_backends = [DjangoFilterBackend] filterset_class = NewsGalleryFilter pagination_class = CustomTwentyPagesPagination class SettingsViewset(viewsets.ModelViewSet): # pragma: no cover queryset = Settings.objects.order_by("name") serializer_class = SettingsSerializer permission_classes = [IsAdminOrReadOnly] def update_settings_json_file(self, validated_data, init_name=None): name = validated_data.get("name") type = validated_data.get("type") options = validated_data.get("options") value = validated_data.get("value") if init_name: name = init_name settings_json_path = f"{BASE_DIR}/caracole_ph_api/app_settings.json" with open(settings_json_path, "r") as file: settings_data = json.load(file) all_settings = settings_data["settings"] new_settings = [] for setting in all_settings: setting_name = setting["name"] if setting_name != name: new_settings.append(setting) new_settings.append( { "name": validated_data.get("name").title(), "type": type, "options": options, "value": value, } ) new_setting_data = {"settings": new_settings} with open(settings_json_path, "w") as file: json.dump(new_setting_data, file, indent=4) def create(self, request, *args, **kwargs): data = request.data serializer = self.get_serializer(data=data) serializer.is_valid(raise_exception=True) validated_data = serializer.validated_data Settings.objects.create(**validated_data) self.update_settings_json_file(validated_data=validated_data) return super().create(request, *args, **kwargs) def update(self, request, *args, **kwargs): instance: Settings = self.get_object() init_name = instance.name data = request.data serializer = self.get_serializer(instance=instance, data=data) serializer.is_valid(raise_exception=True) validated_data = serializer.validated_data for attr, value in validated_data.items(): setattr(instance, attr, value) instance.save() self.update_settings_json_file( validated_data=validated_data, init_name=init_name ) return super().update(request, *args, **kwargs) class FeaturedProductsViewset(viewsets.ModelViewSet): queryset = FeaturedProducts.objects.order_by("-created_at").select_related( "product", "model" ) serializer_class = FeaturedProductsSerializer permission_classes = [IsAdminOrReadOnly]