Source code for shuup.addons.admin_module.views.upload

import os
import shutil
import tempfile
import traceback
import zipfile

from django import forms
from django.http.response import HttpResponseRedirect
from django.utils.translation import gettext_lazy as _
from django.views.generic import FormView

from shuup.addons.installer import PackageInstaller
from shuup.admin.toolbar import PostActionButton, Toolbar
from shuup.admin.utils.urls import manipulate_query_string
from shuup.utils.django_compat import reverse
from shuup.utils.excs import Problem
from shuup.utils.iterables import first



[docs]
class AddonUploadForm(forms.Form):
    file = forms.FileField(
        label=_("Addon file (ZIP)"),
        help_text=_("Only upload the addon files you trust."),
    )




[docs]
class AddonUploadView(FormView):
    form_class = AddonUploadForm
    template_name = "shuup/admin/addons/upload.jinja"
    title = "Upload Addon"


[docs]
    def form_valid(self, form):
        file = form.cleaned_data["file"]
        if not file.name.lower().endswith(".whl"):
            raise Problem(_("Only wheel files (`.whl`) are supported."))
        # TODO: Maybe verify the file before saving?
        tmp_dir = tempfile.mkdtemp(prefix="shuup")
        tmp_token = os.path.basename(tmp_dir)
        filename = os.path.basename(file.name)
        with open(os.path.join(tmp_dir, filename), "wb") as outf:
            shutil.copyfileobj(file, outf)
        return HttpResponseRedirect(
            manipulate_query_string(
                reverse("shuup_admin:addon.upload_confirm"),
                file=filename,
                token=tmp_token,
            )
        )



[docs]
    def get_context_data(self, **kwargs):
        context = super().get_context_data(**kwargs)
        context["toolbar"] = Toolbar(
            [
                PostActionButton(
                    icon="fa fa-upload",
                    form_id="upload_form",
                    text=_("Upload"),
                    extra_css_class="btn-success",
                )
            ],
            view=self,
        )
        return context





[docs]
class AddonUploadConfirmView(FormView):
    form_class = forms.Form
    template_name = "shuup/admin/addons/upload_confirm.jinja"
    title = "Upload Addon"


[docs]
    def get_addon_path(self):
        # get filename from GET since this is a view we get redirected in
        filename = os.path.basename(self.request.GET.get("file"))
        tmp_token = self.request.GET.get("token")
        path = os.path.join(tempfile.gettempdir(), tmp_token, filename)
        if not os.path.isfile(path):
            raise ValueError("Error! File not found.")
        if hasattr(os, "geteuid") and os.stat(path).st_uid != os.geteuid():
            raise ValueError("Error! The file is not owned by the current user.")
        return path



[docs]
    def get_context_data(self, **kwargs):
        context = super().get_context_data(**kwargs)

        with zipfile.ZipFile(self.get_addon_path()) as zf:
            context["filenames"] = sorted(zf.namelist())
            pkg_info_path = first(filename for filename in context["filenames"] if filename.endswith("PKG-INFO"))
            if pkg_info_path:
                context["pkg_info"] = zf.read(pkg_info_path).decode("UTF-8", "replace")

        context["toolbar"] = Toolbar(
            [
                PostActionButton(
                    icon="fa fa-download",
                    form_id="install_form",
                    text=_("Install Addon"),
                    extra_css_class="btn-success",
                )
            ],
            view=self,
        )
        return context



[docs]
    def form_valid(self, form):
        installer = PackageInstaller()
        self.template_name = "shuup/admin/addons/upload_complete.jinja"
        context = {}
        try:
            addon_path = self.get_addon_path()
            installer.install_package(addon_path)
        except Exception:
            context["error"] = traceback.format_exc()
            context["success"] = False
        else:
            context["success"] = True
            try:  # Try cleaning up behind ourselves
                os.unlink(self.get_addon_path())
            except Exception:
                pass
        context["log_content"] = installer.get_log()
        return self.render_to_response(context)