from django.contrib.auth.decorators import login_required
from django.contrib.auth.mixins import LoginRequiredMixin
from django.http import HttpResponse
from django.shortcuts import redirect, get_object_or_404, render
from django.urls import reverse
from django.utils.translation import gettext_lazy as _
from django.views.generic import ListView, DetailView, FormView, UpdateView
from ferrosoft.apps.dataimport.filter import LogEntryFilter
from ferrosoft.apps.dataimport.forms import (
ImportJobForm,
SettingsForm,
)
from ferrosoft.apps.dataimport.models import (
MappingCollection,
ImportJob,
Settings,
ImportStatus,
ImportJobTemplate,
LogEntry,
LogHighlight,
)
from ferrosoft.apps.date_time import now_utc
from ferrosoft.apps.ferrobase.models import LocalUser
from ferrosoft.apps.ferrobase.views import PlatformActivityMixin
from ferrosoft.apps.ferrobase.views.generic import FilterView
from ferrosoft.report import generate_report
[docs]
class IndexView(LoginRequiredMixin, PlatformActivityMixin, ListView):
model = ImportJob
paginate_by = 10
template_name = "dataimport/index.html"
activity_title = _("Recent Imports")
[docs]
def get_queryset(self):
return ImportJob.objects.order_by("-creation_time")
[docs]
@login_required
def collection_docs(request, id):
collection = get_object_or_404(MappingCollection, pk=id)
blob = generate_report("dataimport/gendoc.html", {"collection": collection})
return HttpResponse(blob, content_type="application/pdf")
[docs]
class JobFromTemplateBaseView(ListView):
model = ImportJobTemplate
paginate_by = 10
template_name = "dataimport/job/from-template.html"
[docs]
def get_context_data(self, **kwargs):
context = super().get_context_data(**kwargs)
context.update(
{
"job_url": reverse("dataimport:job_create"),
}
)
return context
[docs]
class JobFromTemplateView(
LoginRequiredMixin, PlatformActivityMixin, JobFromTemplateBaseView
):
activity_title = _("New Import")
[docs]
class JobCreateView(LoginRequiredMixin, PlatformActivityMixin, FormView):
activity_title = _("New Import")
template_name = "dataimport/job/create.html"
form_class = ImportJobForm
_TEMPLATE_QUERY_KEY = "template"
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.collection = None
self.job_template = None
[docs]
def get(self, *args, **kwargs):
template_id = self.request.GET.get(self._TEMPLATE_QUERY_KEY, None)
self.job_template = (
None
if template_id is None
else ImportJobTemplate.objects.filter(pk=template_id).first()
)
if self.job_template is not None:
self.collection = self.job_template.collection
else:
self.initial["field_separator"] = ImportJob.DEFAULT_FIELD_SEPARATOR
self.initial["decimal_separator"] = ImportJob.DEFAULT_DECIMAL_SEPARATOR
collection_id = self.request.GET.get("collection", None)
if collection_id is not None:
self.collection = get_object_or_404(MappingCollection, pk=collection_id)
self.initial["collection"] = self.collection
return super().get(*args, **kwargs)
[docs]
def get_context_data(self, **kwargs):
return super().get_context_data(**kwargs) | {
"collection": self.collection,
}
[docs]
class JobDetail(LoginRequiredMixin, PlatformActivityMixin, DetailView):
activity_title = _("Import Report")
model = ImportJob
template_name = "dataimport/job/detail.html"
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.job_status = None
[docs]
def get(self, request, *args, **kwargs):
self.object = self.get_object()
self.job_status = ImportStatus(self.object.status)
# Displays old logs from database.
enable_old_logs = self.job_status in [
ImportStatus.FINISHED,
ImportStatus.ERRORED,
]
if enable_old_logs:
return redirect("dataimport:job_logs", id=self.object.pk)
context = self.get_context_data(object=self.object)
response = self.render_to_response(context)
# If the user navigates back using the browser controls, prevent cached page
# being loaded and data import started again even though already finished.
response.headers["Cache-Control"] = "no-cache, no-store, must-revalidate"
return response
[docs]
def get_context_data(self, **kwargs):
# Enable receiving log messages from websocket for live updates.
enable_live_log = self.job_status in [ImportStatus.NEW, ImportStatus.PROCESSING]
# Whether to kick off the import (status change from NEW to PROCESSING).
enable_kick_off = self.job_status == ImportStatus.NEW
return super().get_context_data(**kwargs) | {
"job": self.object,
"enable_live_log": enable_live_log,
"enable_kick_off": enable_kick_off,
"error_review_url": "%s?highlight=%s"
% (
reverse("dataimport:job_logs", args=(self.object.pk,)),
LogHighlight.DANGER.value,
),
"data_layer_json": {
"websockets": {
"ui-events": "/ws/dataimport/import-job/%s/" % str(self.object.pk),
}
},
}
[docs]
class JobLogList(FilterView):
activity_title = _("Import Log")
filterset_class = LogEntryFilter
model = LogEntry
paginate_by = 100
ordering = "-log_time"
template_name = "dataimport/job/log-list.html"
[docs]
def get_queryset(self):
return super().get_queryset().filter(import_job__id=self.kwargs["id"])
[docs]
@login_required
def job_status(request, pk):
job = get_object_or_404(ImportJob, pk=pk)
return render(
request,
"dataimport/job/status-messages.html",
{
"alive_time": now_utc(),
"enable_live_log": True,
"line_count": job.line_count,
"job": job,
"show_waiting": not job.is_done(),
"show_error": job.is_status(ImportStatus.ERRORED),
"show_success": job.is_status(ImportStatus.FINISHED),
},
)
[docs]
class SettingsIndex(LoginRequiredMixin, PlatformActivityMixin, UpdateView):
activity_title = _("Settings")
form_class = SettingsForm
model = Settings
template_name = "dataimport/settings/index.html"
[docs]
def get_object(self, **kwargs):
return Settings.get_instance()