Create import directory

bookwyrm/views/imports/import_data.py

@@ -0,0 +1,153 @@
+""" import books from another app """
+from io import TextIOWrapper
+import math
+
+from django.contrib.auth.decorators import login_required
+from django.core.exceptions import PermissionDenied
+from django.core.paginator import Paginator
+from django.http import HttpResponseBadRequest
+from django.shortcuts import get_object_or_404, redirect
+from django.template.response import TemplateResponse
+from django.utils.decorators import method_decorator
+from django.utils.translation import gettext_lazy as _
+from django.views import View
+
+from bookwyrm import forms, models
+from bookwyrm.importers import (
+    Importer,
+    LibrarythingImporter,
+    GoodreadsImporter,
+    StorygraphImporter,
+)
+from bookwyrm.settings import PAGE_LENGTH
+
+# pylint: disable= no-self-use
+@method_decorator(login_required, name="dispatch")
+class Import(View):
+    """import view"""
+
+    def get(self, request):
+        """load import page"""
+        return TemplateResponse(
+            request,
+            "import/import.html",
+            {
+                "import_form": forms.ImportForm(),
+                "jobs": models.ImportJob.objects.filter(user=request.user).order_by(
+                    "-created_date"
+                ),
+            },
+        )
+
+    def post(self, request):
+        """ingest a goodreads csv"""
+        form = forms.ImportForm(request.POST, request.FILES)
+        if form.is_valid():
+            include_reviews = request.POST.get("include_reviews") == "on"
+            privacy = request.POST.get("privacy")
+            source = request.POST.get("source")
+
+            importer = None
+            if source == "LibraryThing":
+                importer = LibrarythingImporter()
+            elif source == "Storygraph":
+                importer = StorygraphImporter()
+            else:
+                # Default : Goodreads
+                importer = GoodreadsImporter()
+
+            try:
+                job = importer.create_job(
+                    request.user,
+                    TextIOWrapper(
+                        request.FILES["csv_file"], encoding=importer.encoding
+                    ),
+                    include_reviews,
+                    privacy,
+                )
+            except (UnicodeDecodeError, ValueError, KeyError):
+                return HttpResponseBadRequest(_("Not a valid csv file"))
+
+            importer.start_import(job)
+
+            return redirect(f"/import/{job.id}")
+        return HttpResponseBadRequest()
+
+
+@method_decorator(login_required, name="dispatch")
+class ImportStatus(View):
+    """status of an existing import"""
+
+    def get(self, request, job_id):
+        """status of an import job"""
+        job = get_object_or_404(models.ImportJob, id=job_id)
+        if job.user != request.user:
+            raise PermissionDenied()
+
+        items = job.items.order_by("index")
+        pending_items = items.filter(fail_reason__isnull=True, book__isnull=True)
+        item_count = items.count() or 1
+
+        paginated = Paginator(items, PAGE_LENGTH)
+        page = paginated.get_page(request.GET.get("page"))
+        data = {
+            "job": job,
+            "items": page,
+            "manual_review_count": items.filter(
+                fail_reason__isnull=False, book_guess__isnull=False, book__isnull=True
+            ).count(),
+            "fail_count": items.filter(
+                fail_reason__isnull=False, book_guess__isnull=True
+            ).count(),
+            "page_range": paginated.get_elided_page_range(
+                page.number, on_each_side=2, on_ends=1
+            ),
+            "complete": not pending_items.exists(),
+            "percent": math.floor(  # pylint: disable=c-extension-no-member
+                (item_count - pending_items.count()) / item_count * 100
+            ),
+        }
+
+        return TemplateResponse(request, "import/import_status.html", data)
+
+
+@method_decorator(login_required, name="dispatch")
+class ImportTroubleshoot(View):
+    """problems items in an existing import"""
+
+    def get(self, request, job_id):
+        """status of an import job"""
+        job = get_object_or_404(models.ImportJob, id=job_id)
+        if job.user != request.user:
+            raise PermissionDenied()
+
+        items = job.items.order_by("index").filter(
+            fail_reason__isnull=False, book_guess__isnull=False
+        )
+
+        paginated = Paginator(items, PAGE_LENGTH)
+        page = paginated.get_page(request.GET.get("page"))
+        data = {
+            "job": job,
+            "items": page,
+            "page_range": paginated.get_elided_page_range(
+                page.number, on_each_side=2, on_ends=1
+            ),
+            "complete": True,
+        }
+
+        return TemplateResponse(request, "import/troubleshoot.html", data)
+
+    def post(self, request, job_id):
+        """retry lines from an import"""
+        job = get_object_or_404(models.ImportJob, id=job_id)
+        items = job.items.filter(fail_reason__isnull=False)
+
+        importer = Importer()
+        job = importer.create_retry_job(
+            request.user,
+            job,
+            items,
+        )
+        importer.start_import(job)
+        return redirect(f"/import/{job.id}")