import logging import os import tempfile from collections.abc import Generator from dataclasses import dataclass from typing import Optional from django.conf import settings from django.contrib import messages from django.contrib.auth.decorators import login_required from django.core.paginator import Paginator from django.db import IntegrityError from django.db import transaction from django.shortcuts import get_object_or_404 from django.shortcuts import redirect from django.shortcuts import render from . import services from .forms import ResourceCreateForm from .forms import ResourceUpdateMetadataForm from .forms import ResourceUpdatePDFsForm from .forms import ResourceUpdateThumbnailsForm from .models import PDFPageSnapshot from .models import PDFResource from .models import Resource from .models import ResourceCategory from .models import ResourceSubcategory from .s3 import get_presigned_obj_url from .s3 import upload_files_to_s3 from .s3 import upload_snapshotted_pages_to_s3 from .s3 import upload_to_s3 logger = logging.getLogger(__name__) # I want to create a dataclass here to hold the resource information to pass to the view @dataclass class ResourceInfo: id: int name: str description: str card_description: str main_resource_category_name: str main_resource_category_colour_css_class: str main_resource_badge_foreground_colour: str subcategories: str | None age_range: str | None pdf_filenames: list[str] pdf_urls: list[Optional[str]] snapshot_urls: dict[str, list[str]] thumbnail_filenames: list[str] thumbnail_urls: list[Optional[str]] feature_slot: int created: str updated: str @login_required def create_featured(request): return render(request, "resources/create_featured_resource.html") def _extract_metadata_from_resource(resource_obj) -> ResourceInfo | None: """ This function extracts the resource information from the model object and returns it as a ResourceInfo object. :param resource_obj: An instance of the resource model from which metadata is extracted. :return: ResourceInfo object containing the extracted metadata, or None if an error occurs. """ # Fetch and create a list of filenames for all PDF resources linked to the given resource object. pdf_resource_filenames = [ x.file_name for x in PDFResource.objects.filter(resource=resource_obj).all() ] # Retrieve all PDFResource objects associated with the resource_obj. pdf_resources = PDFResource.objects.filter(resource=resource_obj).all() # Initialize a dictionary to hold mappings of PDF filenames to their associated page snapshot filenames. snapshot_dict = {} # Iterate through each PDF resource to build the snapshot_dict mapping. for p in pdf_resources: # For each PDF resource, populate the dictionary with its filename as the key # and a list of its associated snapshot filenames as the value. snapshot_dict[p.file_name] = [ x.file_name for x in PDFPageSnapshot.objects.filter(pdf_file=p).all() ] # Initialize a dictionary to hold the URLs for the snapshots. snapshot_url_dict = {} # Generate the URLs for each snapshot based on the snapshot_dict. for k, v in snapshot_dict.items(): snapshot_url_dict[k] = [ get_presigned_obj_url( settings.AWS_STORAGE_BUCKET_NAME, # Specify the bucket name for S3 storage. f"snapshotted_pages/{f}", # Build the path to the snapshot image in S3. ) for f in v # Iterate through the filenames in the snapshot list. ] # Generate pre-signed URLs for the PDF files collected. pdf_urls = [ get_presigned_obj_url(settings.AWS_STORAGE_BUCKET_NAME, f"pdfuploads/{f}") for f in pdf_resource_filenames # Iterate through the PDF filenames. ] # Generate pre-signed URLs for the thumbnail images associated with the resource. thumbnail_urls = [ get_presigned_obj_url(settings.AWS_STORAGE_BUCKET_NAME, f"thumbnails/{f}") for f in resource_obj.thumbnail_filenames # Iterate through the thumbnail filenames. ] try: # Safely retrieve the subcategory name from resource_obj; handle the case where subcategories might be None. arc_name = resource_obj.subcategories.name if resource_obj.subcategories else None # Create and return a ResourceInfo object, populating it with various properties from resource_obj. return ResourceInfo( id=resource_obj.id, # Resource ID name=resource_obj.name, # Resource name description=resource_obj.description, # Resource description card_description=resource_obj.card_description, # Description for card visuals main_resource_category_name=resource_obj.main_resource_category.name, # Main category name main_resource_category_colour_css_class=resource_obj.main_resource_category.colour_css_class, # CSS class for category color main_resource_badge_foreground_colour=resource_obj.main_resource_category.badge_foreground_colour, # Badge color for category subcategories=arc_name, # Subcategory if it exists age_range=resource_obj.age_range, # Target age range for the resource pdf_filenames=pdf_resource_filenames, # List of PDF filenames pdf_urls=pdf_urls, # List of URLs for PDFs snapshot_urls=snapshot_url_dict, # Dictionary of snapshot filenames mapped to their URLs thumbnail_filenames=resource_obj.thumbnail_filenames, # List of thumbnail filenames thumbnail_urls=thumbnail_urls, # List of URLs for thumbnails feature_slot=resource_obj.feature_slot, # Feature slot information created=resource_obj.created_at, # Creation timestamp updated=resource_obj.updated_at, # Last updated timestamp ) except Exception: # Log an exception if there was an error during the extraction process. logging.exception("Error extracting resource information: ") return None # Return None in case of an error. @login_required def index(request): resource_objs = Resource.objects.all() categories = ResourceCategory.objects.all() category = request.GET.get("category", "all") resource_list = [_extract_metadata_from_resource(r) for r in resource_objs] for r in resource_list: try: cart_items = request.user.shoppingcart.items.all() if r.name in [r.resource.name for r in cart_items]: r.in_cart = True except: pass # Create a separate queryset for Featured resources featured_resources = [r for r in resource_list if r.feature_slot] featured_resources = sorted(featured_resources, key=lambda resource: resource.feature_slot) if category != "all": resource_list = [r for r in resource_list if r.main_resource_category_name == category] paginator = Paginator(resource_list, 20) page_number = request.GET.get("page") page_obj = paginator.get_page(page_number) context = { "page_obj": page_obj, "categories": categories, "featured_resources": featured_resources, "selected_category": category, } return render(request, "resources/resource_list.html", context) def create_metadata(pdf_files) -> Generator[tuple[services.PDFMetadata, str], None, None]: """ Generates PDF metadata and snapshot images for a list of PDF files. This function takes a list of PDF file objects, creates temporary files for each one, and then uses the `services.get_pdf_metadata_from_path` and `services.export_pages_as_images` functions to extract metadata and snapshot images for each PDF. The function yields a tuple containing the PDF metadata and a list of snapshot image paths for each PDF file. Args: pdf_files (list[django.core.files.uploadedfile.InMemoryUploadedFile]): A list of PDF file objects. Yields: tuple[servies.PDFMetadata, list[str]]: A tuple containing the PDF metadata and a list of snapshot image paths for each PDF file. """ with tempfile.TemporaryDirectory() as temp_dir: for pdf_file in pdf_files: file_path = os.path.join(temp_dir, pdf_file.name) with open(file_path, "wb") as temp_file: for chunk in pdf_file.chunks(): temp_file.write(chunk) metadata = services.get_pdf_metadata_from_path(file_path) snapshot_images = services.export_pages_as_images(file_path) yield metadata, snapshot_images @login_required def create_resource(request): if request.method == "POST": form = ResourceCreateForm(request.POST, request.FILES) if form.is_valid(): pdf_files = form.cleaned_data["pdf_files"] thumbnail_files = form.cleaned_data["thumbnail_files"] name = form.cleaned_data["name"] description = form.cleaned_data["description"] card_description = form.cleaned_data["card_description"] resource_type = form.cleaned_data["resource_type"] age_range = form.cleaned_data["age_range"] curriculum = form.cleaned_data["curriculum"] main_resource_category = form.cleaned_data["main_resource_category"] subcategories = form.cleaned_data["subcategories"] feature_slot = form.cleaned_data["feature_slot"] # We use get here because we know these categories exist subcategories_objs = [ResourceSubcategory.objects.get(name=x) for x in subcategories] try: with transaction.atomic(): resource = Resource( name=name, description=description, card_description=card_description, resource_type=resource_type, age_range=age_range, curriculum=curriculum, main_resource_category=main_resource_category, feature_slot=feature_slot, ) resource.save() resource.subcategories.set(subcategories_objs) metadata_generator = create_metadata(pdf_files) snapshotted_pages = [] for metadata, snapshot_images in metadata_generator: pdf_resource = PDFResource.objects.create( resource=resource, file_name=os.path.basename(metadata.file_name), file_size=metadata.file_size, ) for snapshot_image in snapshot_images: PDFPageSnapshot.objects.create( name="test", file_name=os.path.basename(snapshot_image), pdf_file=pdf_resource, ) snapshotted_pages.append(snapshot_images) resource.thumbnail_filenames = [f.name for f in thumbnail_files] resource.save() # Reset the file pointers for pdf_files for pdf_file in pdf_files: pdf_file.seek(0) if not upload_to_s3(pdf_files, thumbnail_files, snapshotted_pages): raise Exception("Error uploading files to S3") return redirect("resources:resource_detail", resource_id=resource.id) except IntegrityError: slot = form.cleaned_data["feature_slot"] messages.add_message( request, messages.ERROR, f"Feature slot {slot} is already " "in use. Quit this form and remove from existing " "resource.", ) except Exception: logger.exception("Error creating resource") form.add_error(None, "An error occurred while creating the resource.") else: # extract form errors errors = {} for field in form: if field.errors: errors[field.name] = field.errors # add non-field errors if form.non_field_errors(): errors["non_field_errors"] = form.non_field_errors() # render form with errors return render( request, "resources/resource_create.html", {"form": form, "errors": errors}, ) else: form = ResourceCreateForm() return render(request, "resources/resource_create.html", {"form": form}) @login_required def resource_detail(request, resource_id): """ This function returns the resource detail page. """ resource_obj = get_object_or_404(Resource, pk=resource_id) resource_metadata = _extract_metadata_from_resource(resource_obj) try: price = resource_obj.price_obj.first().price / 1000 except AttributeError: price = 0 resource = { "id": resource_obj.id, "name": resource_obj.name, "description": resource_obj.description, "card_description": resource_obj.card_description, "resource_type": resource_obj.resource_type.name, "main_resource_category": resource_obj.main_resource_category.name, "main_resource_category_colour_css_class": resource_obj.main_resource_category.colour_css_class, "additional_resource_category": ( resource_obj.subcategories.name if resource_obj.subcategories else None ), "age_range": resource_obj.age_range, "curriculum": resource_obj.curriculum, "price": price, "pdf_filenames": resource_metadata.pdf_filenames, "pdf_urls": resource_metadata.pdf_urls, "thumbnails": list( zip( resource_metadata.thumbnail_urls, resource_metadata.thumbnail_filenames, strict=False, ), ), "thumbnail_filenames": resource_metadata.thumbnail_filenames, "thumbnail_urls": resource_metadata.thumbnail_urls, "snapshot_urls": resource_metadata.snapshot_urls, "created": resource_metadata.created, "updated": resource_metadata.updated, } try: cart_items = request.user.shoppingcart.items.all() if resource["name"] in [r.resource.name for r in cart_items]: resource.update(in_cart=True) except: pass return render(request, "resources/resource_detail.html", {"resource": resource}) @login_required() def update_resource_thumbnails(request, pk): resource = get_object_or_404(Resource, pk=pk) if request.method == "POST": form = ResourceUpdateThumbnailsForm(request.POST, request.FILES) if form.is_valid(): thumbnail_files = form.cleaned_data["thumbnail_files"] resource.thumbnail_filenames = [f.name for f in thumbnail_files] upload_files_to_s3(thumbnail_files, "thumbnails") resource.save() return redirect("resources:resource_detail", resource_id=resource.id) else: form = ResourceUpdateThumbnailsForm(resource=pk) return render(request, "resources/update_thumbnails.html", {"form": form, "resource": resource}) @login_required def hx_download_button(request): """ This is an HTMX view that is called when the user clicks the download button. :param: :return: """ pdf = request.GET.get("rn") res = Resource.objects.get(pdf_filename=pdf) return render( request, "resources/hx_download_button.html", {"pdf_url": _extract_metadata_from_resource(res).pdf_url}, ) @login_required def update_resource_metadata(request, pk): # Change resource_id to pk resource = get_object_or_404(Resource, pk=pk) if request.method == "POST": form = ResourceUpdateMetadataForm(request.POST, instance=resource) if form.is_valid(): form.save() return redirect( "resources:resource_detail", resource_id=resource.pk, ) # Use pk instead of resource_id else: form = ResourceUpdateMetadataForm(instance=resource) return render( request, "resources/resource_metadata_update.html", {"form": form, "resource": resource}, ) @login_required() def add_resource_pdfs(request, pk): """ Adds PDF files to a resource in the system. This view handles the process of adding PDF files to an existing resource. It allows the user to upload one or more PDF files, which are then processed and associated with the resource. The view creates PDFResource and PDFPageSnapshot objects to represent the uploaded PDFs and their page snapshots, and uploads the files to S3 storage. Args: request (django.http.request.HttpRequest): The HTTP request object. pk (int): The primary key of the resource to which the PDFs will be added. Returns: django.http.response.HttpResponse: A redirect to the resource detail page upon successful PDF upload. """ resource = get_object_or_404(Resource, pk=pk) if request.method == "POST": form = ResourceUpdatePDFsForm(resource.get_absolute_url(), request.POST, request.FILES) if form.is_valid(): pdf_files = form.cleaned_data["pdf_files"] metadata_generator = create_metadata(pdf_files) snapshotted_pages = [] for metadata, snapshot_images in metadata_generator: # TODO replace or add? This needs to be decided here pdf_resource = PDFResource.objects.create( resource=resource, file_name=os.path.basename(metadata.file_name), file_size=metadata.file_size, ) for snapshot_image in snapshot_images: PDFPageSnapshot.objects.create( name="test", file_name=os.path.basename(snapshot_image), pdf_file=pdf_resource, ) snapshotted_pages.append(snapshot_images) # Reset the file pointers for pdf_files for pdf_file in pdf_files: pdf_file.seek(0) upload_files_to_s3(pdf_files, "pdfuploads") if not upload_snapshotted_pages_to_s3(snapshotted_pages): raise Exception("Error uploading snapshotted pages to S3") return redirect("resources:resource_detail", resource_id=resource.id) else: form = ResourceUpdatePDFsForm(resource.get_absolute_url(), resource=pk) return render(request, "resources/update_pdfs.html", {"form": form, "resource": resource})