384 lines
14 KiB
Python
384 lines
14 KiB
Python
# combo - content management system
|
|
# Copyright (C) 2017-2023 Entr'ouvert
|
|
#
|
|
# This program is free software: you can redistribute it and/or modify it
|
|
# under the terms of the GNU Affero General Public License as published
|
|
# by the Free Software Foundation, either version 3 of the License, or
|
|
# (at your option) any later version.
|
|
#
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU Affero General Public License for more details.
|
|
#
|
|
# You should have received a copy of the GNU Affero General Public License
|
|
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
|
|
|
import io
|
|
import json
|
|
import tarfile
|
|
|
|
from django.contrib.auth.models import Group
|
|
from django.contrib.contenttypes.models import ContentType
|
|
from django.core.files.base import ContentFile
|
|
from django.http import Http404
|
|
from django.shortcuts import get_object_or_404, redirect
|
|
from django.urls import reverse
|
|
from django.utils.translation import gettext_lazy as _
|
|
from rest_framework import permissions
|
|
from rest_framework.generics import GenericAPIView
|
|
from rest_framework.response import Response
|
|
|
|
from combo.apps.export_import.models import Application, ApplicationAsyncJob, ApplicationElement
|
|
from combo.apps.wcs.utils import WCSError
|
|
from combo.data.models import Page, PageSnapshot
|
|
from combo.utils.api import APIErrorBadRequest
|
|
from combo.utils.misc import is_portal_agent
|
|
|
|
klasses = {klass.application_component_type: klass for klass in [Page]}
|
|
klasses['roles'] = Group
|
|
|
|
|
|
class Index(GenericAPIView):
|
|
permission_classes = (permissions.IsAdminUser,)
|
|
|
|
def get(self, request, *args, **kwargs):
|
|
if is_portal_agent():
|
|
response = [
|
|
{
|
|
'id': 'portal-agent-pages',
|
|
'text': _('Pages (agent portal)'),
|
|
'singular': _('Page (agent portal)'),
|
|
},
|
|
]
|
|
else:
|
|
response = [
|
|
{'id': 'pages', 'text': _('Pages'), 'singular': _('Page')},
|
|
]
|
|
|
|
response[0]['urls'] = {
|
|
'list': request.build_absolute_uri(
|
|
reverse(
|
|
'api-export-import-components-list',
|
|
kwargs={'component_type': Page.application_component_type},
|
|
)
|
|
),
|
|
}
|
|
response.append(
|
|
{
|
|
'id': 'roles',
|
|
'text': _('Roles'),
|
|
'singular': _('Role'),
|
|
'urls': {
|
|
'list': request.build_absolute_uri(
|
|
reverse(
|
|
'api-export-import-components-list',
|
|
kwargs={'component_type': 'roles'},
|
|
)
|
|
),
|
|
},
|
|
'minor': True,
|
|
}
|
|
)
|
|
|
|
return Response({'data': response})
|
|
|
|
|
|
index = Index.as_view()
|
|
|
|
|
|
def get_component_bundle_entry(request, component, order):
|
|
if isinstance(component, Group):
|
|
return {
|
|
'id': component.role.slug if hasattr(component, 'role') else component.id,
|
|
'text': component.name,
|
|
'type': 'roles',
|
|
'urls': {},
|
|
# include uuid in object reference, this is not used for applification API but is useful
|
|
# for authentic creating its role summary page.
|
|
'uuid': component.role.uuid if hasattr(component, 'role') else None,
|
|
}
|
|
return {
|
|
'id': str(component.uuid),
|
|
'text': component.title,
|
|
'indent': getattr(component, 'level', 0),
|
|
'type': 'portal-agent-pages' if is_portal_agent() else 'pages',
|
|
'order': order,
|
|
'urls': {
|
|
'export': request.build_absolute_uri(
|
|
reverse(
|
|
'api-export-import-component-export',
|
|
kwargs={'uuid': str(component.uuid), 'component_type': Page.application_component_type},
|
|
)
|
|
),
|
|
'dependencies': request.build_absolute_uri(
|
|
reverse(
|
|
'api-export-import-component-dependencies',
|
|
kwargs={'uuid': str(component.uuid), 'component_type': Page.application_component_type},
|
|
)
|
|
),
|
|
'redirect': request.build_absolute_uri(
|
|
reverse(
|
|
'api-export-import-component-redirect',
|
|
kwargs={'uuid': str(component.uuid), 'component_type': Page.application_component_type},
|
|
)
|
|
),
|
|
},
|
|
}
|
|
|
|
|
|
class ListComponents(GenericAPIView):
|
|
permission_classes = (permissions.IsAdminUser,)
|
|
|
|
def get(self, request, *args, **kwargs):
|
|
klass = klasses[kwargs['component_type']]
|
|
if klass == Page:
|
|
components = Page.get_as_reordered_flat_hierarchy(Page.objects.all())
|
|
elif klass == Group:
|
|
components = Group.objects.order_by('name')
|
|
else:
|
|
raise Http404
|
|
response = [get_component_bundle_entry(request, x, i) for i, x in enumerate(components)]
|
|
return Response({'data': response})
|
|
|
|
|
|
list_components = ListComponents.as_view()
|
|
|
|
|
|
class ExportComponent(GenericAPIView):
|
|
permission_classes = (permissions.IsAdminUser,)
|
|
|
|
def get(self, request, uuid, *args, **kwargs):
|
|
serialisation = get_object_or_404(Page, uuid=uuid).get_serialized_page()
|
|
return Response({'data': serialisation})
|
|
|
|
|
|
export_component = ExportComponent.as_view()
|
|
|
|
|
|
class ComponentDependencies(GenericAPIView):
|
|
permission_classes = (permissions.IsAdminUser,)
|
|
|
|
def get(self, request, uuid, *args, **kwargs):
|
|
klass = klasses[kwargs['component_type']]
|
|
component = get_object_or_404(klass, uuid=uuid)
|
|
|
|
def dependency_dict(component):
|
|
if isinstance(component, dict):
|
|
return component
|
|
return get_component_bundle_entry(request, component, 0)
|
|
|
|
try:
|
|
dependencies = [dependency_dict(x) for x in component.get_dependencies() if x]
|
|
except WCSError as e:
|
|
return Response({'err': 1, 'err_desc': str(e)}, status=400)
|
|
return Response({'err': 0, 'data': dependencies})
|
|
|
|
|
|
component_dependencies = ComponentDependencies.as_view()
|
|
|
|
|
|
def component_redirect(request, component_type, uuid):
|
|
klass = klasses[component_type]
|
|
page = get_object_or_404(klass, uuid=uuid)
|
|
if klass == Page:
|
|
url = reverse('combo-manager-page-view', kwargs={'pk': page.pk})
|
|
if (
|
|
'compare' in request.GET
|
|
and request.GET.get('application')
|
|
and request.GET.get('version1')
|
|
and request.GET.get('version2')
|
|
):
|
|
url = '%s?version1=%s&version2=%s&application=%s' % (
|
|
reverse('combo-manager-page-history-compare', args=[page.pk]),
|
|
request.GET['version1'],
|
|
request.GET['version2'],
|
|
request.GET['application'],
|
|
)
|
|
return redirect(url)
|
|
raise Http404
|
|
|
|
|
|
class BundleCheck(GenericAPIView):
|
|
permission_classes = (permissions.IsAdminUser,)
|
|
|
|
def put(self, request, *args, **kwargs):
|
|
tar_io = io.BytesIO(request.read())
|
|
page_type = 'portal-agent-pages' if is_portal_agent() else 'pages'
|
|
try:
|
|
with tarfile.open(fileobj=tar_io) as tar:
|
|
try:
|
|
manifest = json.loads(tar.extractfile('manifest.json').read().decode())
|
|
except KeyError:
|
|
raise APIErrorBadRequest(_('Invalid tar file, missing manifest'))
|
|
application_slug = manifest.get('slug')
|
|
application_version = manifest.get('version_number')
|
|
if not application_slug or not application_version:
|
|
return Response({'data': {}})
|
|
|
|
differences = []
|
|
unknown_elements = []
|
|
no_history_elements = []
|
|
legacy_elements = []
|
|
content_type = ContentType.objects.get_for_model(Page)
|
|
for element in manifest.get('elements'):
|
|
if element.get('type') != page_type:
|
|
continue
|
|
try:
|
|
page = Page.objects.get(uuid=element['slug'])
|
|
except Page.DoesNotExist:
|
|
unknown_elements.append(
|
|
{
|
|
'type': element['type'],
|
|
'slug': element['slug'],
|
|
}
|
|
)
|
|
continue
|
|
elements_qs = ApplicationElement.objects.filter(
|
|
application__slug=application_slug,
|
|
content_type=content_type,
|
|
object_id=page.pk,
|
|
)
|
|
if not elements_qs.exists():
|
|
# object exists, but not linked to the application
|
|
legacy_elements.append(
|
|
{
|
|
'type': element['type'],
|
|
'slug': element['slug'],
|
|
# information needed here, Relation objects may not exist yet in hobo
|
|
'text': page.title,
|
|
'url': request.build_absolute_uri(
|
|
reverse(
|
|
'api-export-import-component-redirect',
|
|
kwargs={
|
|
'uuid': str(page.uuid),
|
|
'component_type': page.application_component_type,
|
|
},
|
|
)
|
|
),
|
|
}
|
|
)
|
|
continue
|
|
snapshot_for_app = (
|
|
PageSnapshot.objects.filter(
|
|
page=page,
|
|
application_slug=application_slug,
|
|
application_version=application_version,
|
|
)
|
|
.order_by('timestamp')
|
|
.last()
|
|
)
|
|
if not snapshot_for_app:
|
|
# no snapshot for this bundle
|
|
no_history_elements.append(
|
|
{
|
|
'type': element['type'],
|
|
'slug': element['slug'],
|
|
}
|
|
)
|
|
continue
|
|
last_snapshot = PageSnapshot.objects.filter(page=page).latest('timestamp')
|
|
if snapshot_for_app.pk != last_snapshot.pk:
|
|
differences.append(
|
|
{
|
|
'type': element['type'],
|
|
'slug': element['slug'],
|
|
'url': '%s%s?version1=%s&version2=%s'
|
|
% (
|
|
request.build_absolute_uri('/')[:-1],
|
|
reverse('combo-manager-page-history-compare', args=[page.pk]),
|
|
snapshot_for_app.pk,
|
|
last_snapshot.pk,
|
|
),
|
|
}
|
|
)
|
|
except tarfile.TarError:
|
|
raise APIErrorBadRequest(_('Invalid tar file'))
|
|
|
|
return Response(
|
|
{
|
|
'data': {
|
|
'differences': differences,
|
|
'unknown_elements': unknown_elements,
|
|
'no_history_elements': no_history_elements,
|
|
'legacy_elements': legacy_elements,
|
|
}
|
|
}
|
|
)
|
|
|
|
|
|
bundle_check = BundleCheck.as_view()
|
|
|
|
|
|
class BundleImport(GenericAPIView):
|
|
permission_classes = (permissions.IsAdminUser,)
|
|
action = 'import_bundle'
|
|
|
|
def put(self, request, *args, **kwargs):
|
|
tar_io = io.BytesIO(request.read())
|
|
try:
|
|
with tarfile.open(fileobj=tar_io) as tar:
|
|
try:
|
|
manifest = json.loads(tar.extractfile('manifest.json').read().decode())
|
|
except KeyError:
|
|
raise APIErrorBadRequest(_('Invalid tar file, missing manifest'))
|
|
application_slug = manifest.get('slug')
|
|
except tarfile.TarError:
|
|
raise APIErrorBadRequest(_('Invalid tar file'))
|
|
job = ApplicationAsyncJob(
|
|
action=self.action,
|
|
)
|
|
job.bundle.save('%s.tar' % application_slug, content=ContentFile(tar_io.getvalue()))
|
|
job.save()
|
|
job.run(spool=True)
|
|
return Response({'err': 0, 'url': job.get_api_status_url(request)})
|
|
|
|
|
|
bundle_import = BundleImport.as_view()
|
|
|
|
|
|
class BundleDeclare(BundleImport):
|
|
action = 'declare_bundle'
|
|
|
|
|
|
bundle_declare = BundleDeclare.as_view()
|
|
|
|
|
|
class BundleUnlink(GenericAPIView):
|
|
permission_classes = (permissions.IsAdminUser,)
|
|
|
|
def post(self, request, *args, **kwargs):
|
|
if request.POST.get('application'):
|
|
try:
|
|
application = Application.objects.get(slug=request.POST['application'])
|
|
except Application.DoesNotExist:
|
|
pass
|
|
else:
|
|
application.delete()
|
|
|
|
return Response({'err': 0})
|
|
|
|
|
|
bundle_unlink = BundleUnlink.as_view()
|
|
|
|
|
|
class JobStatus(GenericAPIView):
|
|
permission_classes = (permissions.IsAuthenticated,)
|
|
|
|
def get(self, request, *args, **kwargs):
|
|
job = get_object_or_404(ApplicationAsyncJob, uuid=kwargs['job_uuid'])
|
|
return Response(
|
|
{
|
|
'err': 0,
|
|
'data': {
|
|
'status': job.status,
|
|
'creation_time': job.creation_timestamp,
|
|
'completion_time': job.completion_timestamp,
|
|
'completion_status': job.get_completion_status(),
|
|
},
|
|
}
|
|
)
|
|
|
|
|
|
job_status = JobStatus.as_view()
|