267 lines
9.1 KiB
Python
267 lines
9.1 KiB
Python
# chrono - content management system
|
|
# Copyright (C) 2016-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.http import Http404
|
|
from django.shortcuts import get_object_or_404, redirect
|
|
from django.urls import reverse
|
|
from rest_framework import permissions
|
|
from rest_framework.generics import GenericAPIView
|
|
from rest_framework.response import Response
|
|
|
|
from chrono.agendas.models import Agenda, Category, EventsType, Resource, UnavailabilityCalendar
|
|
from chrono.apps.export_import.models import Application, ApplicationElement
|
|
from chrono.manager.utils import import_site
|
|
|
|
klasses = {
|
|
klass.application_component_type: klass
|
|
for klass in [Agenda, Category, EventsType, Resource, UnavailabilityCalendar]
|
|
}
|
|
klasses_translation = {
|
|
'agendas_categories': 'categories',
|
|
}
|
|
klasses_translation_reverse = {v: k for k, v in klasses_translation.items()}
|
|
|
|
|
|
class Index(GenericAPIView):
|
|
permission_classes = (permissions.IsAuthenticatedOrReadOnly,)
|
|
|
|
def get(self, request, *args, **kwargs):
|
|
data = []
|
|
for klass in klasses.values():
|
|
component_type = {
|
|
'id': klass.application_component_type,
|
|
'text': klass.application_label_plural,
|
|
'singular': klass.application_label_singular,
|
|
'urls': {
|
|
'list': request.build_absolute_uri(
|
|
reverse(
|
|
'api-export-import-components-list',
|
|
kwargs={'component_type': klass.application_component_type},
|
|
)
|
|
),
|
|
},
|
|
}
|
|
if klass not in [Agenda]:
|
|
component_type['minor'] = True
|
|
data.append(component_type)
|
|
|
|
return Response({'data': data})
|
|
|
|
|
|
index = Index.as_view()
|
|
|
|
|
|
def get_component_bundle_entry(request, component):
|
|
return {
|
|
'id': str(component.slug),
|
|
'text': component.label,
|
|
'type': component.application_component_type,
|
|
'urls': {
|
|
'export': request.build_absolute_uri(
|
|
reverse(
|
|
'api-export-import-component-export',
|
|
kwargs={
|
|
'component_type': component.application_component_type,
|
|
'slug': str(component.slug),
|
|
},
|
|
)
|
|
),
|
|
'dependencies': request.build_absolute_uri(
|
|
reverse(
|
|
'api-export-import-component-dependencies',
|
|
kwargs={
|
|
'component_type': component.application_component_type,
|
|
'slug': str(component.slug),
|
|
},
|
|
)
|
|
),
|
|
'redirect': request.build_absolute_uri(
|
|
reverse(
|
|
'api-export-import-component-redirect',
|
|
kwargs={
|
|
'component_type': component.application_component_type,
|
|
'slug': str(component.slug),
|
|
},
|
|
)
|
|
),
|
|
},
|
|
}
|
|
|
|
|
|
class ListComponents(GenericAPIView):
|
|
permission_classes = (permissions.IsAuthenticatedOrReadOnly,)
|
|
|
|
def get(self, request, *args, **kwargs):
|
|
klass = klasses[kwargs['component_type']]
|
|
response = [get_component_bundle_entry(request, x) for x in klass.objects.order_by('slug')]
|
|
return Response({'data': response})
|
|
|
|
|
|
list_components = ListComponents.as_view()
|
|
|
|
|
|
class ExportComponent(GenericAPIView):
|
|
permission_classes = (permissions.IsAuthenticatedOrReadOnly,)
|
|
|
|
def get(self, request, slug, *args, **kwargs):
|
|
klass = klasses[kwargs['component_type']]
|
|
serialisation = klass.objects.get(slug=slug).export_json()
|
|
return Response({'data': serialisation})
|
|
|
|
|
|
export_component = ExportComponent.as_view()
|
|
|
|
|
|
class ComponentDependencies(GenericAPIView):
|
|
permission_classes = (permissions.IsAuthenticatedOrReadOnly,)
|
|
|
|
def get(self, request, slug, *args, **kwargs):
|
|
klass = klasses[kwargs['component_type']]
|
|
component = klass.objects.get(slug=slug)
|
|
|
|
def dependency_dict(element):
|
|
if isinstance(element, Group):
|
|
return {
|
|
'id': element.role.slug if hasattr(element, 'role') else element.id,
|
|
'text': element.name,
|
|
'type': 'roles',
|
|
'urls': {},
|
|
}
|
|
return get_component_bundle_entry(request, element)
|
|
|
|
dependencies = [dependency_dict(x) for x in component.get_dependencies() if x]
|
|
return Response({'err': 0, 'data': dependencies})
|
|
|
|
|
|
component_dependencies = ComponentDependencies.as_view()
|
|
|
|
|
|
def component_redirect(request, component_type, slug):
|
|
klass = klasses[component_type]
|
|
component = get_object_or_404(klass, slug=slug)
|
|
if klass == Agenda:
|
|
return redirect(reverse('chrono-manager-agenda-view', kwargs={'pk': component.pk}))
|
|
raise Http404
|
|
|
|
|
|
class BundleCheck(GenericAPIView):
|
|
permission_classes = (permissions.IsAuthenticatedOrReadOnly,)
|
|
|
|
def put(self, request, *args, **kwargs):
|
|
return Response({'err': 0, 'data': {}})
|
|
|
|
|
|
bundle_check = BundleCheck.as_view()
|
|
|
|
|
|
class BundleImport(GenericAPIView):
|
|
permission_classes = (permissions.IsAuthenticatedOrReadOnly,)
|
|
install = True
|
|
|
|
def put(self, request, *args, **kwargs):
|
|
tar_io = io.BytesIO(request.read())
|
|
components = {}
|
|
with tarfile.open(fileobj=tar_io) as tar:
|
|
manifest = json.loads(tar.extractfile('manifest.json').read().decode())
|
|
self.application = Application.update_or_create_from_manifest(
|
|
manifest,
|
|
tar,
|
|
editable=not self.install,
|
|
)
|
|
|
|
for element in manifest.get('elements'):
|
|
component_type = element['type']
|
|
if component_type not in klasses or element['type'] == 'roles':
|
|
continue
|
|
component_type = klasses_translation.get(component_type, component_type)
|
|
if component_type not in components:
|
|
components[component_type] = []
|
|
component_content = (
|
|
tar.extractfile('%s/%s' % (element['type'], element['slug'])).read().decode()
|
|
)
|
|
components[component_type].append(json.loads(component_content).get('data'))
|
|
# init cache of application elements, from manifest
|
|
self.application_elements = set()
|
|
# import agendas
|
|
self.do_something(components)
|
|
# create application elements
|
|
self.link_objects(components)
|
|
# remove obsolete application elements
|
|
self.unlink_obsolete_objects()
|
|
return Response({'err': 0})
|
|
|
|
def do_something(self, components):
|
|
if components:
|
|
import_site(components)
|
|
|
|
def link_objects(self, components):
|
|
for component_type, component_list in components.items():
|
|
component_type = klasses_translation_reverse.get(component_type, component_type)
|
|
klass = klasses[component_type]
|
|
for component in component_list:
|
|
try:
|
|
existing_component = klass.objects.get(slug=component['slug'])
|
|
except klass.DoesNotExist:
|
|
pass
|
|
else:
|
|
element = ApplicationElement.update_or_create_for_object(
|
|
self.application, existing_component
|
|
)
|
|
self.application_elements.add(element.content_object)
|
|
|
|
def unlink_obsolete_objects(self):
|
|
known_elements = ApplicationElement.objects.filter(application=self.application)
|
|
for element in known_elements:
|
|
if element.content_object not in self.application_elements:
|
|
element.delete()
|
|
|
|
|
|
bundle_import = BundleImport.as_view()
|
|
|
|
|
|
class BundleDeclare(BundleImport):
|
|
install = False
|
|
|
|
def do_something(self, agendas):
|
|
# no installation on declare
|
|
pass
|
|
|
|
|
|
bundle_declare = BundleDeclare.as_view()
|
|
|
|
|
|
class BundleUnlink(GenericAPIView):
|
|
permission_classes = (permissions.IsAuthenticatedOrReadOnly,)
|
|
|
|
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()
|