1747 lines
No EOL
65 KiB
Python
1747 lines
No EOL
65 KiB
Python
from collections import defaultdict
|
|
import mimetypes
|
|
import os
|
|
|
|
from django.contrib import messages
|
|
from django.contrib.auth.mixins import LoginRequiredMixin
|
|
from django.core.exceptions import PermissionDenied
|
|
from django.db import transaction
|
|
from django.http import FileResponse, Http404, JsonResponse, HttpResponse
|
|
from django.db.models.functions import Lower
|
|
from django.shortcuts import get_object_or_404, redirect, render
|
|
from django.urls import reverse, reverse_lazy
|
|
from django.utils import timezone
|
|
from django.utils.encoding import smart_str
|
|
from django.utils.http import url_has_allowed_host_and_scheme
|
|
from django.utils.translation import gettext_lazy as _
|
|
from django.views import View
|
|
from django.views.generic import DetailView, ListView
|
|
|
|
from .forms import (
|
|
AccessLinkForm,
|
|
AttachmentForm,
|
|
DocumentFolderAccessLinkForm,
|
|
DocumentFolderForm,
|
|
DocumentFolderShareForm,
|
|
DocumentShareForm,
|
|
DocumentValidationForm,
|
|
DocumentVersionForm,
|
|
ManagedDocumentForm,
|
|
)
|
|
from .models import (
|
|
DocumentAccessLink,
|
|
DocumentAttachment,
|
|
DocumentFolder,
|
|
DocumentFolderAccessLink,
|
|
DocumentTag,
|
|
DocumentType,
|
|
DocumentValidation,
|
|
DocumentVersion,
|
|
ManagedDocument,
|
|
)
|
|
from .permissions import (
|
|
filter_documents_for_user,
|
|
filter_folders_for_user,
|
|
get_user_visible_folder_ids,
|
|
user_can_delete_document,
|
|
user_is_internal_manager,
|
|
user_has_document_permission,
|
|
user_is_documents_admin,
|
|
)
|
|
|
|
|
|
class DocumentsAppAccessMixin:
|
|
"""
|
|
Mixin pour vérifier que l'utilisateur a accès à l'app 'documents'
|
|
dans ses accessible_views avant d'accéder aux vues de gestion documentaire.
|
|
"""
|
|
def dispatch(self, request, *args, **kwargs):
|
|
if not request.user.is_authenticated:
|
|
return self.handle_no_permission()
|
|
|
|
config = getattr(request.user, 'config', None)
|
|
|
|
# Les superusers ont toujours accès
|
|
if request.user.is_superuser:
|
|
return super().dispatch(request, *args, **kwargs)
|
|
|
|
if not config:
|
|
raise PermissionDenied(_("You don't have access to the document management module."))
|
|
|
|
# Les utilisateurs avec l'accès explicite à l'app documents
|
|
if config.can_access_view('documents'):
|
|
return super().dispatch(request, *args, **kwargs)
|
|
|
|
# Les utilisateurs internes (managers, controllers, etc.) ont accès
|
|
# Le filtrage des dossiers/documents est assuré par filter_folders_for_user
|
|
if config.is_intern:
|
|
return super().dispatch(request, *args, **kwargs)
|
|
|
|
raise PermissionDenied(_("You don't have access to the document management module."))
|
|
|
|
|
|
def get_folder_navigation_context(user, current_folder=None):
|
|
breadcrumbs = get_folder_breadcrumb(current_folder)
|
|
return {
|
|
"root_folders": get_root_folders(user),
|
|
"folder_tree": build_folder_tree(user),
|
|
"active_folder_slug": current_folder.slug if current_folder else None,
|
|
"active_trail_slugs": [folder.slug for folder in breadcrumbs],
|
|
"breadcrumbs": breadcrumbs,
|
|
}
|
|
|
|
|
|
def build_folder_tree(user):
|
|
visible_ids = get_user_visible_folder_ids(user)
|
|
if not visible_ids:
|
|
return []
|
|
folders = (
|
|
DocumentFolder.objects.prefetch_related("parent_folders")
|
|
.filter(pk__in=visible_ids)
|
|
.order_by("name")
|
|
.distinct()
|
|
)
|
|
children_map = defaultdict(list)
|
|
root_nodes = []
|
|
|
|
for folder in folders:
|
|
parents = list(folder.parent_folders.all())
|
|
if parents:
|
|
for parent in parents:
|
|
children_map[parent.pk].append(folder)
|
|
else:
|
|
root_nodes.append(folder)
|
|
|
|
def sort_children(children):
|
|
return sorted(children, key=lambda child: child.name.lower())
|
|
|
|
def build_nodes(nodes):
|
|
tree = []
|
|
for node in sort_children(nodes):
|
|
tree.append(
|
|
{
|
|
"folder": node,
|
|
"children": build_nodes(children_map[node.pk]),
|
|
}
|
|
)
|
|
return tree
|
|
|
|
return build_nodes(root_nodes)
|
|
|
|
|
|
def get_descendant_folder_ids(folder, visible_ids=None):
|
|
if folder is None:
|
|
return set()
|
|
|
|
descendant_ids = set()
|
|
current_batch = [folder.pk]
|
|
descendant_ids.add(folder.pk)
|
|
|
|
# Optimisation: charger les enfants par batch au lieu d'un par un
|
|
while current_batch:
|
|
# Charger tous les enfants du batch actuel en une seule requête
|
|
child_qs = DocumentFolder.objects.filter(parent_folders__pk__in=current_batch)
|
|
if visible_ids is not None:
|
|
child_qs = child_qs.filter(pk__in=visible_ids)
|
|
|
|
child_ids = list(child_qs.values_list("pk", flat=True).distinct())
|
|
|
|
# Filtrer les enfants non encore visités
|
|
next_batch = []
|
|
for child_id in child_ids:
|
|
if child_id not in descendant_ids:
|
|
descendant_ids.add(child_id)
|
|
next_batch.append(child_id)
|
|
|
|
current_batch = next_batch
|
|
|
|
return descendant_ids
|
|
|
|
|
|
def get_folder_breadcrumb(folder):
|
|
if folder is None:
|
|
return []
|
|
|
|
breadcrumb = []
|
|
visited = set()
|
|
current = folder
|
|
|
|
while current and current.pk not in visited:
|
|
breadcrumb.append(current)
|
|
visited.add(current.pk)
|
|
parents = list(current.parent_folders.all())
|
|
if not parents:
|
|
current = None
|
|
continue
|
|
parents.sort(key=lambda parent: parent.name.lower())
|
|
current = parents[0]
|
|
|
|
return list(reversed(breadcrumb))
|
|
|
|
|
|
def get_navigation_folder_for_document(document, requested_slug=None):
|
|
if requested_slug:
|
|
folder = (
|
|
DocumentFolder.objects.prefetch_related("parent_folders")
|
|
.filter(slug=requested_slug)
|
|
.first()
|
|
)
|
|
if folder:
|
|
return folder
|
|
|
|
primary_folder = document.get_primary_folder()
|
|
if primary_folder:
|
|
return (
|
|
DocumentFolder.objects.prefetch_related("parent_folders")
|
|
.filter(pk=primary_folder.pk)
|
|
.first()
|
|
)
|
|
|
|
return None
|
|
|
|
|
|
def get_root_folders(user):
|
|
visible_ids = get_user_visible_folder_ids(user)
|
|
if not visible_ids:
|
|
return DocumentFolder.objects.none()
|
|
return (
|
|
DocumentFolder.objects.filter(
|
|
pk__in=visible_ids, parent_folders__isnull=True
|
|
)
|
|
.order_by("name")
|
|
.distinct()
|
|
)
|
|
|
|
def ensure_folder_editable(folder, user):
|
|
if folder.is_predefined:
|
|
raise PermissionDenied
|
|
if user_is_documents_admin(user):
|
|
return
|
|
if folder.created_by_id and folder.created_by_id == getattr(user, "pk", None):
|
|
return
|
|
raise PermissionDenied
|
|
|
|
|
|
def ensure_document_deletable(document, user):
|
|
"""Vérifie si l'utilisateur peut supprimer le document."""
|
|
if not user_can_delete_document(user, document):
|
|
raise PermissionDenied
|
|
|
|
|
|
def ensure_folder_shareable(folder, user):
|
|
if user_is_documents_admin(user):
|
|
return
|
|
if folder.created_by_id and folder.created_by_id == getattr(user, "pk", None):
|
|
return
|
|
raise PermissionDenied
|
|
|
|
|
|
def get_safe_next(request, candidate):
|
|
if not candidate:
|
|
return None
|
|
if url_has_allowed_host_and_scheme(
|
|
candidate,
|
|
allowed_hosts={request.get_host()},
|
|
require_https=request.is_secure(),
|
|
):
|
|
return candidate
|
|
return None
|
|
|
|
|
|
class DocumentListView(DocumentsAppAccessMixin, LoginRequiredMixin, ListView):
|
|
model = ManagedDocument
|
|
paginate_by = 50
|
|
template_name = "documents/document_list.html"
|
|
context_object_name = "documents"
|
|
|
|
def get_queryset(self):
|
|
queryset = super().get_queryset()
|
|
queryset = filter_documents_for_user(queryset, self.request.user)
|
|
queryset = queryset.select_related("current_version").prefetch_related(
|
|
"tags", "types", "attachments", "folders"
|
|
)
|
|
self.search_term = (self.request.GET.get("search") or "").strip()
|
|
tag = self.request.GET.get("tag")
|
|
doc_type = self.request.GET.get("type")
|
|
|
|
folder_slug = self.request.GET.get("folder")
|
|
self.current_folder = None
|
|
self.descendant_folder_ids = set()
|
|
self.visible_folder_ids = get_user_visible_folder_ids(self.request.user)
|
|
|
|
if folder_slug:
|
|
candidate = (
|
|
DocumentFolder.objects.prefetch_related(
|
|
"parent_folders", "child_folders"
|
|
)
|
|
.filter(slug=folder_slug)
|
|
.first()
|
|
)
|
|
if candidate and candidate.pk in self.visible_folder_ids:
|
|
self.current_folder = candidate
|
|
|
|
if not self.current_folder and folder_slug:
|
|
raise Http404
|
|
|
|
if self.current_folder:
|
|
self.descendant_folder_ids = get_descendant_folder_ids(
|
|
self.current_folder, visible_ids=self.visible_folder_ids
|
|
)
|
|
if self.search_term:
|
|
queryset = queryset.filter(
|
|
folders__pk__in=self.descendant_folder_ids
|
|
)
|
|
else:
|
|
queryset = queryset.filter(folders=self.current_folder)
|
|
|
|
if self.search_term:
|
|
queryset = queryset.filter(title__icontains=self.search_term)
|
|
if tag:
|
|
queryset = queryset.filter(tags__slug=tag)
|
|
if doc_type:
|
|
queryset = queryset.filter(types__code=doc_type)
|
|
|
|
requested_sort = self.request.GET.get("sort") or "name"
|
|
if requested_sort == "date":
|
|
requested_sort = "-updated_at"
|
|
|
|
allowed_sorts = {"name", "-name", "updated_at", "-updated_at"}
|
|
if requested_sort not in allowed_sorts:
|
|
requested_sort = "name"
|
|
|
|
self.sort_option = requested_sort
|
|
|
|
if requested_sort in {"updated_at", "-updated_at"}:
|
|
order = "updated_at" if requested_sort == "updated_at" else "-updated_at"
|
|
queryset = queryset.order_by(order, "pk")
|
|
else:
|
|
queryset = queryset.annotate(_title_lower=Lower("title"))
|
|
if requested_sort == "-name":
|
|
queryset = queryset.order_by("-_title_lower", "pk")
|
|
else:
|
|
queryset = queryset.order_by("_title_lower", "pk")
|
|
|
|
return queryset.distinct()
|
|
|
|
def get_context_data(self, **kwargs):
|
|
context = super().get_context_data(**kwargs)
|
|
context["tags"] = DocumentTag.objects.order_by("name_fr")
|
|
context["types"] = DocumentType.objects.order_by("label")
|
|
context.update(
|
|
get_folder_navigation_context(
|
|
self.request.user, getattr(self, "current_folder", None)
|
|
)
|
|
)
|
|
context.setdefault("breadcrumbs", get_folder_breadcrumb(getattr(self, "current_folder", None)))
|
|
|
|
folder_slug = self.request.GET.get("folder")
|
|
current_folder = getattr(self, "current_folder", None)
|
|
if current_folder is None and folder_slug:
|
|
current_folder = (
|
|
DocumentFolder.objects.prefetch_related("parent_folders")
|
|
.filter(slug=folder_slug)
|
|
.first()
|
|
)
|
|
self.current_folder = current_folder
|
|
context["current_folder"] = current_folder
|
|
context["is_documents_admin"] = user_is_documents_admin(self.request.user)
|
|
|
|
search_term = getattr(self, "search_term", "") or ""
|
|
descendant_folder_ids = set(getattr(self, "descendant_folder_ids", set()))
|
|
|
|
sort_option = getattr(self, "sort_option", "name")
|
|
|
|
visible_ids = getattr(self, "visible_folder_ids", set())
|
|
|
|
if search_term:
|
|
if current_folder:
|
|
descendant_folder_ids.discard(getattr(current_folder, "pk", None))
|
|
if descendant_folder_ids:
|
|
visible_folders_qs = (
|
|
DocumentFolder.objects.prefetch_related("parent_folders")
|
|
.filter(pk__in=descendant_folder_ids)
|
|
.filter(name__icontains=search_term)
|
|
.distinct()
|
|
)
|
|
else:
|
|
visible_folders_qs = DocumentFolder.objects.none()
|
|
else:
|
|
visible_folders_qs = (
|
|
DocumentFolder.objects.prefetch_related("parent_folders")
|
|
.filter(name__icontains=search_term)
|
|
.distinct()
|
|
)
|
|
elif current_folder:
|
|
visible_folders_qs = (
|
|
DocumentFolder.objects.prefetch_related("parent_folders")
|
|
.filter(parent_folders=current_folder)
|
|
.distinct()
|
|
)
|
|
else:
|
|
visible_folders_qs = (
|
|
DocumentFolder.objects.prefetch_related("parent_folders")
|
|
.filter(parent_folders__isnull=True)
|
|
.distinct()
|
|
)
|
|
|
|
if visible_ids:
|
|
visible_folders_qs = visible_folders_qs.filter(pk__in=visible_ids)
|
|
else:
|
|
visible_folders_qs = DocumentFolder.objects.none()
|
|
|
|
visible_folders = sorted(
|
|
visible_folders_qs,
|
|
key=lambda folder: folder.name.casefold(),
|
|
)
|
|
context["visible_folders"] = visible_folders
|
|
context["subfolders"] = visible_folders
|
|
|
|
breadcrumbs = get_folder_breadcrumb(current_folder)
|
|
context["breadcrumbs"] = breadcrumbs
|
|
|
|
active_folder_slug = current_folder.slug if current_folder else None
|
|
context["active_folder_slug"] = active_folder_slug
|
|
context["active_trail_slugs"] = [folder.slug for folder in breadcrumbs]
|
|
|
|
documents_page = context["documents"]
|
|
documents_list = list(getattr(documents_page, "object_list", documents_page))
|
|
listing_items = []
|
|
|
|
include_folders = getattr(context.get("page_obj"), "number", 1) == 1
|
|
if include_folders:
|
|
for folder in visible_folders:
|
|
listing_items.append(
|
|
{
|
|
"type": "folder",
|
|
"object": folder,
|
|
"updated_at": folder.updated_at,
|
|
"updated_by": "/",
|
|
"name_key": folder.name.casefold(),
|
|
}
|
|
)
|
|
|
|
for document in documents_list:
|
|
listing_items.append(
|
|
{
|
|
"type": "document",
|
|
"object": document,
|
|
"can_delete": user_can_delete_document(self.request.user, document),
|
|
"updated_at": document.updated_at,
|
|
"updated_by": document.latest_version.uploaded_by.get_short_name() if document.latest_version and document.latest_version.uploaded_by else "/",
|
|
"name_key": document.title.casefold(),
|
|
}
|
|
)
|
|
|
|
def updated_timestamp(item):
|
|
updated_at = item.get("updated_at")
|
|
if updated_at is None:
|
|
return float("-inf")
|
|
if timezone.is_naive(updated_at):
|
|
updated_at = timezone.make_aware(
|
|
updated_at, timezone.get_current_timezone()
|
|
)
|
|
return updated_at.timestamp()
|
|
|
|
if sort_option == "-updated_at":
|
|
listing_items.sort(key=updated_timestamp, reverse=True)
|
|
elif sort_option == "updated_at":
|
|
listing_items.sort(key=updated_timestamp)
|
|
elif sort_option == "-name":
|
|
listing_items.sort(key=lambda item: item.get("name_key", ""), reverse=True)
|
|
else:
|
|
listing_items.sort(key=lambda item: item.get("name_key", ""))
|
|
|
|
listing_items.sort(key=lambda item: 0 if item["type"] == "folder" else 1)
|
|
context["listing_items"] = listing_items
|
|
|
|
query_params = self.request.GET.copy()
|
|
query_params.pop("page", None)
|
|
context["current_querystring"] = query_params.urlencode()
|
|
|
|
context["current_sort"] = sort_option
|
|
|
|
def build_sort_query(value):
|
|
params = self.request.GET.copy()
|
|
params.pop("page", None)
|
|
params["sort"] = value
|
|
return params.urlencode()
|
|
|
|
def make_sort_context(asc_value, desc_value, default_direction="asc"):
|
|
if sort_option == asc_value:
|
|
next_value = desc_value
|
|
is_active = True
|
|
direction = "asc"
|
|
elif sort_option == desc_value:
|
|
next_value = asc_value
|
|
is_active = True
|
|
direction = "desc"
|
|
else:
|
|
is_active = False
|
|
direction = None
|
|
next_value = asc_value if default_direction == "asc" else desc_value
|
|
|
|
return {
|
|
"is_active": is_active,
|
|
"direction": direction,
|
|
"next_query": build_sort_query(next_value),
|
|
"asc_query": build_sort_query(asc_value),
|
|
"desc_query": build_sort_query(desc_value),
|
|
}
|
|
|
|
context["title_sort"] = make_sort_context("name", "-name", default_direction="asc")
|
|
context["updated_sort"] = make_sort_context(
|
|
"updated_at", "-updated_at", default_direction="desc"
|
|
)
|
|
|
|
return context
|
|
|
|
|
|
class DocumentCreateView(DocumentsAppAccessMixin, LoginRequiredMixin, View):
|
|
template_name = "documents/document_form.html"
|
|
|
|
def get_current_folder(self):
|
|
"""Récupère le folder courant depuis les paramètres GET."""
|
|
folder_slug = self.request.GET.get("folder")
|
|
if folder_slug:
|
|
try:
|
|
return DocumentFolder.objects.get(slug=folder_slug)
|
|
except DocumentFolder.DoesNotExist:
|
|
pass
|
|
return None
|
|
|
|
def get(self, request):
|
|
current_folder = self.get_current_folder()
|
|
form = ManagedDocumentForm(user=request.user, initial_folder=current_folder)
|
|
return render(
|
|
request,
|
|
self.template_name,
|
|
{
|
|
"form": form,
|
|
**get_folder_navigation_context(request.user),
|
|
},
|
|
)
|
|
|
|
def post(self, request):
|
|
current_folder = self.get_current_folder()
|
|
form = ManagedDocumentForm(request.POST, request.FILES, user=request.user, initial_folder=current_folder)
|
|
if form.is_valid():
|
|
document = form.save(user=request.user)
|
|
messages.success(request, _("Document created successfully."))
|
|
return redirect(document.get_absolute_url())
|
|
return render(
|
|
request,
|
|
self.template_name,
|
|
{
|
|
"form": form,
|
|
**get_folder_navigation_context(request.user),
|
|
},
|
|
)
|
|
|
|
|
|
class DocumentFolderCreateView(LoginRequiredMixin, View):
|
|
template_name = "documents/documentfolder_form.html"
|
|
|
|
def get_initial(self):
|
|
initial = {}
|
|
parent_slug = self.request.GET.get("parent")
|
|
if parent_slug:
|
|
parent = (
|
|
DocumentFolder.objects.prefetch_related("parent_folders")
|
|
.filter(slug=parent_slug)
|
|
.first()
|
|
)
|
|
if parent:
|
|
initial["parent_folders"] = [parent.pk]
|
|
self._initial_parent = parent
|
|
return initial
|
|
|
|
def get_next_url(self):
|
|
return self.request.POST.get("next") or self.request.GET.get("next")
|
|
|
|
def get_safe_next_url(self):
|
|
return get_safe_next(self.request, self.get_next_url())
|
|
|
|
def get_success_url(self, folder):
|
|
next_url = self.get_safe_next_url()
|
|
if next_url:
|
|
return next_url
|
|
if folder:
|
|
return f"{reverse('documents:list')}?folder={folder.slug}"
|
|
return reverse("documents:list")
|
|
|
|
def get(self, request):
|
|
initial = self.get_initial()
|
|
parent = getattr(self, "_initial_parent", None)
|
|
form = DocumentFolderForm(initial=initial, user=request.user)
|
|
context = {
|
|
"form": form,
|
|
"folder": None,
|
|
"title": _("Create folder"),
|
|
"next": self.get_safe_next_url() or "",
|
|
"is_documents_admin": user_is_documents_admin(request.user),
|
|
**get_folder_navigation_context(request.user, parent),
|
|
}
|
|
return render(request, self.template_name, context)
|
|
|
|
def post(self, request):
|
|
form = DocumentFolderForm(request.POST, user=request.user)
|
|
if form.is_valid():
|
|
folder = form.save(commit=False)
|
|
if "created_by" in form.fields:
|
|
folder.created_by = form.cleaned_data.get("created_by")
|
|
if not folder.created_by:
|
|
folder.created_by = request.user
|
|
folder.save()
|
|
form.save_m2m()
|
|
messages.success(request, _("Folder created successfully."))
|
|
return redirect(self.get_success_url(folder))
|
|
|
|
parent = None
|
|
parent_ids = request.POST.getlist("parent_folders")
|
|
if parent_ids:
|
|
parent = (
|
|
DocumentFolder.objects.prefetch_related("parent_folders")
|
|
.filter(pk__in=parent_ids)
|
|
.first()
|
|
)
|
|
context = {
|
|
"form": form,
|
|
"folder": None,
|
|
"title": _("Create folder"),
|
|
"next": self.get_safe_next_url() or "",
|
|
"is_documents_admin": user_is_documents_admin(request.user),
|
|
**get_folder_navigation_context(request.user, parent),
|
|
}
|
|
return render(request, self.template_name, context)
|
|
|
|
|
|
class DocumentFolderUpdateView(DocumentsAppAccessMixin, LoginRequiredMixin, View):
|
|
template_name = "documents/documentfolder_form.html"
|
|
|
|
def get_next_url(self):
|
|
return self.request.POST.get("next") or self.request.GET.get("next")
|
|
|
|
def get_safe_next_url(self):
|
|
return get_safe_next(self.request, self.get_next_url())
|
|
|
|
def get_success_url(self, folder):
|
|
next_url = self.get_safe_next_url()
|
|
if next_url:
|
|
return next_url
|
|
if folder:
|
|
return f"{reverse('documents:list')}?folder={folder.slug}"
|
|
return reverse("documents:list")
|
|
|
|
def get_folder(self, pk):
|
|
return get_object_or_404(
|
|
DocumentFolder.objects.prefetch_related("parent_folders"), pk=pk
|
|
)
|
|
|
|
def get(self, request, pk):
|
|
folder = self.get_folder(pk)
|
|
ensure_folder_editable(folder, request.user)
|
|
form = DocumentFolderForm(instance=folder, user=request.user)
|
|
context = {
|
|
"form": form,
|
|
"folder": folder,
|
|
"title": _("Edit folder"),
|
|
"next": self.get_safe_next_url() or "",
|
|
"is_documents_admin": user_is_documents_admin(request.user),
|
|
**get_folder_navigation_context(request.user, folder),
|
|
}
|
|
return render(request, self.template_name, context)
|
|
|
|
def post(self, request, pk):
|
|
folder = self.get_folder(pk)
|
|
ensure_folder_editable(folder, request.user)
|
|
form = DocumentFolderForm(request.POST, instance=folder, user=request.user)
|
|
if form.is_valid():
|
|
updated_folder = form.save(commit=False)
|
|
if "created_by" in form.fields:
|
|
updated_folder.created_by = form.cleaned_data.get("created_by")
|
|
updated_folder.save()
|
|
form.save_m2m()
|
|
messages.success(request, _("Folder updated successfully."))
|
|
return redirect(self.get_success_url(updated_folder))
|
|
|
|
context = {
|
|
"form": form,
|
|
"folder": folder,
|
|
"title": _("Edit folder"),
|
|
"next": self.get_safe_next_url() or "",
|
|
"is_documents_admin": user_is_documents_admin(request.user),
|
|
**get_folder_navigation_context(request.user, folder),
|
|
}
|
|
return render(request, self.template_name, context)
|
|
|
|
|
|
class DocumentFolderDeleteView(LoginRequiredMixin, View):
|
|
template_name = "documents/documentfolder_confirm_delete.html"
|
|
|
|
def get_next_url(self):
|
|
return self.request.POST.get("next") or self.request.GET.get("next")
|
|
|
|
def get_safe_next_url(self):
|
|
return get_safe_next(self.request, self.get_next_url())
|
|
|
|
def get_success_url(self, folder):
|
|
next_url = self.get_safe_next_url()
|
|
if next_url:
|
|
return next_url
|
|
parent = folder.parent_folders.order_by("name").first() if folder else None
|
|
if parent:
|
|
return f"{reverse('documents:list')}?folder={parent.slug}"
|
|
return reverse("documents:list")
|
|
|
|
def get_folder(self, pk):
|
|
return get_object_or_404(
|
|
DocumentFolder.objects.prefetch_related("parent_folders"), pk=pk
|
|
)
|
|
|
|
def get(self, request, pk):
|
|
folder = self.get_folder(pk)
|
|
ensure_folder_editable(folder, request.user)
|
|
context = {
|
|
"folder": folder,
|
|
"next": self.get_safe_next_url() or "",
|
|
"is_documents_admin": user_is_documents_admin(request.user),
|
|
**get_folder_navigation_context(request.user, folder),
|
|
}
|
|
return render(request, self.template_name, context)
|
|
|
|
def post(self, request, pk):
|
|
folder = self.get_folder(pk)
|
|
ensure_folder_editable(folder, request.user)
|
|
success_url = self.get_success_url(folder)
|
|
folder.delete()
|
|
messages.success(request, _("Folder deleted successfully."))
|
|
return redirect(success_url)
|
|
|
|
|
|
class DocumentDeleteView(DocumentsAppAccessMixin, LoginRequiredMixin, View):
|
|
template_name = "documents/document_confirm_delete.html"
|
|
|
|
def get_next_url(self):
|
|
return self.request.POST.get("next") or self.request.GET.get("next")
|
|
|
|
def get_safe_next_url(self):
|
|
return get_safe_next(self.request, self.get_next_url())
|
|
|
|
def get_success_url(self):
|
|
next_url = self.get_safe_next_url()
|
|
if next_url:
|
|
return next_url
|
|
return reverse("documents:list")
|
|
|
|
def get_document(self, pk):
|
|
return get_object_or_404(ManagedDocument.objects.all(), pk=pk)
|
|
|
|
def get(self, request, pk):
|
|
import os
|
|
document = self.get_document(pk)
|
|
ensure_document_deletable(document, request.user)
|
|
|
|
is_dxf = False
|
|
overlay_doc = None
|
|
imported_poles_count = 0
|
|
latest_version = document.latest_version
|
|
if latest_version and latest_version.file and latest_version.file.name.lower().endswith('.dxf'):
|
|
is_dxf = True
|
|
dxf_filename = os.path.basename(latest_version.file.name)
|
|
overlay_doc = ManagedDocument.objects.filter(
|
|
unique_params__dxf_source_id=document.id,
|
|
is_archived=False
|
|
).first()
|
|
|
|
from assets.models import TrafficLightPole
|
|
imported_poles_count = TrafficLightPole.objects.filter(
|
|
note__contains=dxf_filename
|
|
).count()
|
|
|
|
context = {
|
|
"document": document,
|
|
"next": self.get_safe_next_url() or "",
|
|
"is_documents_admin": user_is_documents_admin(request.user),
|
|
"is_dxf": is_dxf,
|
|
"has_overlay": overlay_doc is not None,
|
|
"imported_poles_count": imported_poles_count,
|
|
**get_folder_navigation_context(request.user),
|
|
}
|
|
return render(request, self.template_name, context)
|
|
|
|
def post(self, request, pk):
|
|
import os
|
|
from django.db import transaction
|
|
document = self.get_document(pk)
|
|
ensure_document_deletable(document, request.user)
|
|
success_url = self.get_success_url()
|
|
|
|
delete_overlay = request.POST.get('delete_overlay') == 'on'
|
|
delete_vectors = request.POST.get('delete_vectors') == 'on'
|
|
|
|
latest_version = document.latest_version
|
|
dxf_filename = os.path.basename(latest_version.file.name) if latest_version and latest_version.file else ""
|
|
|
|
with transaction.atomic():
|
|
if delete_overlay:
|
|
overlay_docs = ManagedDocument.objects.filter(
|
|
unique_params__dxf_source_id=document.id
|
|
)
|
|
for o_doc in overlay_docs:
|
|
o_doc.delete()
|
|
|
|
if delete_vectors and dxf_filename:
|
|
from assets.models import TrafficLightPole
|
|
poles_to_delete = TrafficLightPole.objects.filter(
|
|
note__contains=dxf_filename
|
|
)
|
|
poles_to_delete.delete()
|
|
|
|
# Clean up and delete linked TrafficLightIntersectionDocument shell objects and files
|
|
from assets.models import TrafficLightIntersectionDocument
|
|
from django.conf import settings
|
|
|
|
# Clean up layout geojson and overlay png files on disk directly if it's a DXF
|
|
if dxf_filename and dxf_filename.lower().endswith('.dxf'):
|
|
# 1. layout geojson
|
|
geojson_fn = dxf_filename.lower().replace('.dxf', '.layout.geojson')
|
|
geojson_path = os.path.join(settings.PRIVATE_MEDIA_ROOT, 'dxf_layouts', geojson_fn)
|
|
if os.path.exists(geojson_path):
|
|
try:
|
|
os.remove(geojson_path)
|
|
except Exception:
|
|
pass
|
|
# 2. overlay png
|
|
overlay_fn = dxf_filename.lower().replace('.dxf', '.overlay.png')
|
|
overlay_path = os.path.join(settings.MEDIA_ROOT, 'dxf_layouts', overlay_fn)
|
|
if os.path.exists(overlay_path):
|
|
try:
|
|
os.remove(overlay_path)
|
|
except Exception:
|
|
pass
|
|
|
|
intersection_docs = TrafficLightIntersectionDocument.objects.filter(managed_document=document)
|
|
for idx_doc in intersection_docs:
|
|
idx_doc.delete()
|
|
|
|
document.delete()
|
|
|
|
messages.success(request, _("Document deleted successfully."))
|
|
return redirect(success_url)
|
|
|
|
|
|
class DocumentFolderShareView(DocumentsAppAccessMixin, LoginRequiredMixin, View):
|
|
template_name = "documents/documentfolder_share.html"
|
|
|
|
def get_folder(self, pk):
|
|
return get_object_or_404(
|
|
DocumentFolder.objects.prefetch_related(
|
|
"shares__user", "access_links", "parent_folders"
|
|
),
|
|
pk=pk,
|
|
)
|
|
|
|
def get(self, request, pk):
|
|
folder = self.get_folder(pk)
|
|
ensure_folder_shareable(folder, request.user)
|
|
context = {
|
|
"folder": folder,
|
|
"share_form": DocumentFolderShareForm(folder=folder),
|
|
"access_link_form": DocumentFolderAccessLinkForm(),
|
|
"shares": folder.shares.select_related("user").order_by(
|
|
"user__username"
|
|
),
|
|
"access_links": folder.access_links.order_by("-created_at"),
|
|
"is_documents_admin": user_is_documents_admin(request.user),
|
|
**get_folder_navigation_context(request.user, folder),
|
|
}
|
|
return render(request, self.template_name, context)
|
|
|
|
|
|
class ShareFolderView(LoginRequiredMixin, View):
|
|
def post(self, request, pk):
|
|
folder = get_object_or_404(DocumentFolder, pk=pk)
|
|
ensure_folder_shareable(folder, request.user)
|
|
form = DocumentFolderShareForm(request.POST, folder=folder)
|
|
if form.is_valid():
|
|
form.save(creator=request.user, folder=folder)
|
|
messages.success(request, _("Folder shared successfully."))
|
|
else:
|
|
messages.error(request, _("Unable to share the folder."))
|
|
return redirect(reverse("documents:folder_share", args=[folder.pk]))
|
|
|
|
|
|
class DeleteFolderShareView(LoginRequiredMixin, View):
|
|
def post(self, request, pk, share_id):
|
|
folder = get_object_or_404(DocumentFolder, pk=pk)
|
|
ensure_folder_shareable(folder, request.user)
|
|
share = get_object_or_404(folder.shares, pk=share_id)
|
|
share.delete()
|
|
messages.success(request, _("Folder share removed."))
|
|
return redirect(reverse("documents:folder_share", args=[folder.pk]))
|
|
|
|
|
|
class CreateFolderAccessLinkView(LoginRequiredMixin, View):
|
|
def post(self, request, pk):
|
|
folder = get_object_or_404(DocumentFolder, pk=pk)
|
|
ensure_folder_shareable(folder, request.user)
|
|
form = DocumentFolderAccessLinkForm(request.POST)
|
|
if form.is_valid():
|
|
form.save(user=request.user, folder=folder)
|
|
messages.success(request, _("Shareable link created."))
|
|
else:
|
|
messages.error(request, _("Unable to create the access link."))
|
|
return redirect(reverse("documents:folder_share", args=[folder.pk]))
|
|
|
|
|
|
class DeleteFolderAccessLinkView(LoginRequiredMixin, View):
|
|
def post(self, request, pk, link_id):
|
|
folder = get_object_or_404(DocumentFolder, pk=pk)
|
|
ensure_folder_shareable(folder, request.user)
|
|
link = get_object_or_404(folder.access_links, pk=link_id)
|
|
link.delete()
|
|
messages.success(request, _("Access link revoked."))
|
|
return redirect(reverse("documents:folder_share", args=[folder.pk]))
|
|
|
|
|
|
class DocumentDetailView(DocumentsAppAccessMixin, LoginRequiredMixin, DetailView):
|
|
model = ManagedDocument
|
|
template_name = "documents/document_detail.html"
|
|
context_object_name = "document"
|
|
|
|
def get_queryset(self):
|
|
queryset = super().get_queryset()
|
|
queryset = filter_documents_for_user(queryset, self.request.user)
|
|
return queryset.select_related("current_version").prefetch_related(
|
|
"folders",
|
|
"folders__parent_folders",
|
|
"tags",
|
|
"shares__user",
|
|
"access_links",
|
|
"attachments",
|
|
)
|
|
|
|
def get_context_data(self, **kwargs):
|
|
context = super().get_context_data(**kwargs)
|
|
document = context["document"]
|
|
context["version_form"] = DocumentVersionForm()
|
|
context["share_form"] = DocumentShareForm(document=document)
|
|
context["attachment_form"] = AttachmentForm(document=document)
|
|
context["access_link_form"] = AccessLinkForm()
|
|
context["validation_forms"] = {
|
|
validation.id: DocumentValidationForm(instance=validation)
|
|
for version in document.versions.all()
|
|
for validation in version.validations.all()
|
|
}
|
|
context["root_folders"] = get_root_folders(self.request.user)
|
|
context["folder_tree"] = build_folder_tree(self.request.user)
|
|
|
|
requested_folder_slug = self.request.GET.get("folder")
|
|
navigation_folder = get_navigation_folder_for_document(
|
|
document, requested_slug=requested_folder_slug
|
|
)
|
|
|
|
# Check if the document has a current version file ending in .dxf and is attached to a TrafficLightIntersection
|
|
is_dxf = False
|
|
is_ifc = False
|
|
intersection = None
|
|
document_filename = ""
|
|
latest_version = document.latest_version
|
|
if latest_version and latest_version.file:
|
|
is_dxf = latest_version.file.name.lower().endswith('.dxf')
|
|
is_ifc = latest_version.file.name.lower().endswith('.ifc')
|
|
document_filename = os.path.basename(latest_version.file.name)
|
|
if is_dxf:
|
|
from django.contrib.contenttypes.models import ContentType
|
|
from assets.models import TrafficLightIntersection
|
|
from documents.models import DocumentAttachment
|
|
|
|
ct_tli = ContentType.objects.get_for_model(TrafficLightIntersection)
|
|
attachment = DocumentAttachment.objects.filter(
|
|
document=document,
|
|
content_type=ct_tli
|
|
).first()
|
|
if attachment:
|
|
intersection = attachment.content_object
|
|
|
|
context["is_dxf"] = is_dxf
|
|
context["is_ifc"] = is_ifc
|
|
context["intersection"] = intersection
|
|
context["document_filename"] = document_filename
|
|
|
|
active_folder_slug = navigation_folder.slug if navigation_folder else None
|
|
context["active_folder_slug"] = active_folder_slug
|
|
|
|
breadcrumbs = get_folder_breadcrumb(navigation_folder)
|
|
context["breadcrumbs"] = breadcrumbs
|
|
context["active_trail_slugs"] = [folder.slug for folder in breadcrumbs]
|
|
context["document_breadcrumb"] = breadcrumbs + ([document] if document else [])
|
|
|
|
preview_mime_type = None
|
|
is_preview_image = False
|
|
is_preview_pdf = False
|
|
is_preview_text = False
|
|
is_preview_tabular = False
|
|
preview_text = ""
|
|
preview_rows = []
|
|
preview_truncated = False
|
|
latest_version = document.latest_version
|
|
|
|
if latest_version and getattr(latest_version, "file", None):
|
|
fname = latest_version.file.name
|
|
mime_type, _ = mimetypes.guess_type(fname)
|
|
ext = os.path.splitext(fname)[1].lower()
|
|
if mime_type and mime_type.startswith("image/"):
|
|
preview_mime_type = mime_type
|
|
is_preview_image = True
|
|
elif mime_type == "application/pdf":
|
|
preview_mime_type = mime_type
|
|
is_preview_pdf = True
|
|
elif ext in (".txt", ".log"):
|
|
is_preview_text = True
|
|
try:
|
|
with latest_version.file.open("rb") as f:
|
|
raw = f.read(50_001)
|
|
preview_truncated = len(raw) > 50_000
|
|
preview_text = raw[:50_000].decode("utf-8-sig", errors="replace")
|
|
except Exception:
|
|
pass
|
|
elif ext == ".csv" or mime_type == "text/csv":
|
|
is_preview_tabular = True
|
|
try:
|
|
import csv as csv_mod
|
|
import io as io_mod
|
|
with latest_version.file.open("rb") as f:
|
|
raw = f.read()
|
|
try:
|
|
text_data = raw.decode("utf-8-sig")
|
|
except UnicodeDecodeError:
|
|
text_data = raw.decode("latin-1")
|
|
try:
|
|
dialect = csv_mod.Sniffer().sniff(text_data[:4096])
|
|
except csv_mod.Error:
|
|
dialect = csv_mod.excel
|
|
reader = csv_mod.reader(io_mod.StringIO(text_data), dialect=dialect)
|
|
rows_tmp = [row for _, row in zip(range(201), reader)]
|
|
preview_truncated = len(rows_tmp) > 200
|
|
preview_rows = rows_tmp[:200]
|
|
except Exception:
|
|
pass
|
|
elif ext in (".xlsx", ".xls"):
|
|
is_preview_tabular = True
|
|
try:
|
|
import openpyxl
|
|
from io import BytesIO
|
|
with latest_version.file.open("rb") as f:
|
|
raw = f.read()
|
|
wb = openpyxl.load_workbook(BytesIO(raw), read_only=True, data_only=True)
|
|
ws = wb.active
|
|
rows_tmp = []
|
|
for i, row in enumerate(ws.iter_rows(values_only=True)):
|
|
if i >= 201:
|
|
break
|
|
rows_tmp.append([str(c) if c is not None else "" for c in row])
|
|
wb.close()
|
|
preview_truncated = len(rows_tmp) > 200
|
|
preview_rows = rows_tmp[:200]
|
|
except Exception:
|
|
pass
|
|
|
|
context["preview_mime_type"] = preview_mime_type
|
|
context["is_preview_image"] = is_preview_image
|
|
context["is_preview_pdf"] = is_preview_pdf
|
|
context["is_preview_text"] = is_preview_text
|
|
context["is_preview_tabular"] = is_preview_tabular
|
|
context["preview_text"] = preview_text
|
|
context["preview_rows"] = preview_rows
|
|
context["preview_truncated"] = preview_truncated
|
|
|
|
is_admin = user_is_documents_admin(self.request.user)
|
|
context["is_documents_admin"] = is_admin
|
|
context["can_edit_document"] = is_admin or user_has_document_permission(
|
|
self.request.user,
|
|
document,
|
|
required_permission=ManagedDocument.PERMISSION_EDIT,
|
|
)
|
|
context["can_delete_document"] = user_can_delete_document(self.request.user, document)
|
|
|
|
# Available folders/tags for add forms (exclude already associated ones)
|
|
existing_folder_ids = list(document.folders.values_list("pk", flat=True))
|
|
accessible_folders = filter_folders_for_user(
|
|
DocumentFolder.objects.all(),
|
|
self.request.user,
|
|
required_permission=ManagedDocument.PERMISSION_VIEW,
|
|
).exclude(pk__in=existing_folder_ids).order_by("name")
|
|
context["available_folders"] = accessible_folders
|
|
|
|
existing_tag_ids = list(document.tags.values_list("pk", flat=True))
|
|
context["available_tags"] = DocumentTag.objects.exclude(
|
|
pk__in=existing_tag_ids
|
|
).order_by("name_fr")
|
|
|
|
return context
|
|
|
|
|
|
class VersionPreviewDataView(LoginRequiredMixin, View):
|
|
"""Returns JSON preview data for a document version (text, CSV, or Excel)."""
|
|
|
|
def get(self, request, pk, version_id):
|
|
document = get_object_or_404(ManagedDocument, pk=pk)
|
|
if not (
|
|
user_is_documents_admin(request.user)
|
|
or user_has_document_permission(
|
|
request.user, document, ManagedDocument.PERMISSION_VIEW
|
|
)
|
|
):
|
|
raise PermissionDenied
|
|
|
|
version = get_object_or_404(DocumentVersion, pk=version_id, document=document)
|
|
if not getattr(version, "file", None):
|
|
return JsonResponse({"type": "unsupported"})
|
|
|
|
fname = version.file.name
|
|
ext = os.path.splitext(fname)[1].lower()
|
|
try:
|
|
if ext in (".txt", ".log"):
|
|
with version.file.open("rb") as f:
|
|
raw = f.read(50_001)
|
|
truncated = len(raw) > 50_000
|
|
content = raw[:50_000].decode("utf-8-sig", errors="replace")
|
|
return JsonResponse({"type": "text", "content": content, "truncated": truncated})
|
|
|
|
elif ext == ".csv":
|
|
import csv as csv_mod
|
|
import io as io_mod
|
|
with version.file.open("rb") as f:
|
|
raw = f.read()
|
|
try:
|
|
text_data = raw.decode("utf-8-sig")
|
|
except UnicodeDecodeError:
|
|
text_data = raw.decode("latin-1")
|
|
try:
|
|
dialect = csv_mod.Sniffer().sniff(text_data[:4096])
|
|
except csv_mod.Error:
|
|
dialect = csv_mod.excel
|
|
reader = csv_mod.reader(io_mod.StringIO(text_data), dialect=dialect)
|
|
rows_tmp = [row for _, row in zip(range(201), reader)]
|
|
truncated = len(rows_tmp) > 200
|
|
return JsonResponse({"type": "tabular", "rows": rows_tmp[:200], "truncated": truncated})
|
|
|
|
elif ext in (".xlsx", ".xls"):
|
|
import openpyxl
|
|
from io import BytesIO
|
|
with version.file.open("rb") as f:
|
|
raw = f.read()
|
|
wb = openpyxl.load_workbook(BytesIO(raw), read_only=True, data_only=True)
|
|
ws = wb.active
|
|
rows_tmp = []
|
|
for i, row in enumerate(ws.iter_rows(values_only=True)):
|
|
if i >= 201:
|
|
break
|
|
rows_tmp.append([str(c) if c is not None else "" for c in row])
|
|
wb.close()
|
|
truncated = len(rows_tmp) > 200
|
|
return JsonResponse({"type": "tabular", "rows": rows_tmp[:200], "truncated": truncated})
|
|
|
|
except Exception as exc:
|
|
return JsonResponse({"type": "error", "message": str(exc)})
|
|
|
|
return JsonResponse({"type": "unsupported"})
|
|
|
|
|
|
class AddVersionView(LoginRequiredMixin, View):
|
|
def post(self, request, pk):
|
|
document = get_object_or_404(ManagedDocument, pk=pk)
|
|
form = DocumentVersionForm(request.POST, request.FILES)
|
|
if form.is_valid():
|
|
form.save(user=request.user, document=document)
|
|
messages.success(request, _("New version uploaded."))
|
|
else:
|
|
messages.error(request, _("Unable to upload the new version."))
|
|
return redirect(document.get_absolute_url())
|
|
|
|
|
|
class ShareDocumentView(LoginRequiredMixin, View):
|
|
def post(self, request, pk):
|
|
document = get_object_or_404(ManagedDocument, pk=pk)
|
|
form = DocumentShareForm(request.POST, document=document)
|
|
if form.is_valid():
|
|
form.save(creator=request.user, document=document)
|
|
messages.success(request, _("Share created."))
|
|
else:
|
|
messages.error(request, _("Unable to create share."))
|
|
return redirect(document.get_absolute_url())
|
|
|
|
|
|
class DeleteShareView(LoginRequiredMixin, View):
|
|
def post(self, request, pk, share_id):
|
|
document = get_object_or_404(ManagedDocument, pk=pk)
|
|
share = get_object_or_404(document.shares, pk=share_id)
|
|
share.delete()
|
|
messages.success(request, _("Share removed."))
|
|
return redirect(document.get_absolute_url())
|
|
|
|
|
|
class ValidateVersionView(LoginRequiredMixin, View):
|
|
def post(self, request, pk, validation_id):
|
|
document = get_object_or_404(ManagedDocument, pk=pk)
|
|
validation = get_object_or_404(DocumentValidation, pk=validation_id)
|
|
form = DocumentValidationForm(request.POST, instance=validation)
|
|
if form.is_valid():
|
|
form.save(user=request.user)
|
|
messages.success(request, _("Validation updated."))
|
|
else:
|
|
messages.error(request, _("Unable to update validation."))
|
|
return redirect(document.get_absolute_url())
|
|
|
|
|
|
class AttachDocumentView(LoginRequiredMixin, View):
|
|
def post(self, request, pk):
|
|
document = get_object_or_404(ManagedDocument, pk=pk)
|
|
form = AttachmentForm(request.POST, document=document)
|
|
if form.is_valid():
|
|
form.save(user=request.user, document=document)
|
|
messages.success(request, _("Document linked successfully."))
|
|
else:
|
|
messages.error(request, _("Unable to link the document."))
|
|
return redirect(document.get_absolute_url())
|
|
|
|
|
|
class DeleteVersionView(LoginRequiredMixin, View):
|
|
def post(self, request, pk, version_id):
|
|
document = get_object_or_404(ManagedDocument.objects.all(), pk=pk)
|
|
ensure_document_deletable(document, request.user)
|
|
version = get_object_or_404(document.versions, pk=version_id)
|
|
|
|
with transaction.atomic():
|
|
if document.current_version_id == version.pk:
|
|
next_version = (
|
|
document.versions
|
|
.exclude(pk=version.pk)
|
|
.order_by('-version_number')
|
|
.first()
|
|
)
|
|
document.current_version = next_version
|
|
document.save(update_fields=['current_version'])
|
|
version.delete()
|
|
|
|
messages.success(request, _("Version deleted successfully."))
|
|
next_url = get_safe_next(request, request.POST.get('next'))
|
|
if next_url:
|
|
return redirect(next_url)
|
|
return redirect(document.get_absolute_url())
|
|
|
|
|
|
class DetachDocumentView(LoginRequiredMixin, View):
|
|
def post(self, request, pk, attachment_id):
|
|
document = get_object_or_404(ManagedDocument, pk=pk)
|
|
attachment = get_object_or_404(document.attachments, pk=attachment_id)
|
|
attachment.delete()
|
|
messages.success(request, _("Link removed."))
|
|
return redirect(document.get_absolute_url())
|
|
|
|
|
|
class CreateAccessLinkView(LoginRequiredMixin, View):
|
|
def post(self, request, pk):
|
|
document = get_object_or_404(ManagedDocument, pk=pk)
|
|
form = AccessLinkForm(request.POST)
|
|
if form.is_valid():
|
|
form.save(user=request.user, document=document)
|
|
messages.success(request, _("Shareable link created."))
|
|
else:
|
|
messages.error(request, _("Unable to create the access link."))
|
|
return redirect(document.get_absolute_url())
|
|
|
|
|
|
class RevokeAccessLinkView(LoginRequiredMixin, View):
|
|
def post(self, request, pk, link_id):
|
|
document = get_object_or_404(ManagedDocument, pk=pk)
|
|
link = get_object_or_404(document.access_links, pk=link_id)
|
|
link.delete()
|
|
messages.success(request, _("Access link revoked."))
|
|
return redirect(document.get_absolute_url())
|
|
|
|
|
|
class AddFolderToDocumentView(LoginRequiredMixin, View):
|
|
def post(self, request, pk):
|
|
document = get_object_or_404(
|
|
filter_documents_for_user(
|
|
ManagedDocument.objects.all(), request.user
|
|
),
|
|
pk=pk,
|
|
)
|
|
if not (
|
|
user_is_documents_admin(request.user)
|
|
or user_has_document_permission(
|
|
request.user, document, required_permission=ManagedDocument.PERMISSION_EDIT
|
|
)
|
|
):
|
|
from django.core.exceptions import PermissionDenied
|
|
raise PermissionDenied
|
|
folder_id = request.POST.get("folder_id")
|
|
folder = get_object_or_404(DocumentFolder, pk=folder_id)
|
|
document.folders.add(folder)
|
|
messages.success(request, _("Folder added."))
|
|
return redirect(document.get_absolute_url())
|
|
|
|
|
|
class RemoveFolderFromDocumentView(LoginRequiredMixin, View):
|
|
def post(self, request, pk, folder_id):
|
|
document = get_object_or_404(
|
|
filter_documents_for_user(
|
|
ManagedDocument.objects.all(), request.user
|
|
),
|
|
pk=pk,
|
|
)
|
|
if not (
|
|
user_is_documents_admin(request.user)
|
|
or user_has_document_permission(
|
|
request.user, document, required_permission=ManagedDocument.PERMISSION_EDIT
|
|
)
|
|
):
|
|
from django.core.exceptions import PermissionDenied
|
|
raise PermissionDenied
|
|
folder = get_object_or_404(DocumentFolder, pk=folder_id)
|
|
document.folders.remove(folder)
|
|
messages.success(request, _("Folder removed."))
|
|
return redirect(document.get_absolute_url())
|
|
|
|
|
|
class AddTagToDocumentView(LoginRequiredMixin, View):
|
|
def post(self, request, pk):
|
|
document = get_object_or_404(
|
|
filter_documents_for_user(
|
|
ManagedDocument.objects.all(), request.user
|
|
),
|
|
pk=pk,
|
|
)
|
|
if not (
|
|
user_is_documents_admin(request.user)
|
|
or user_has_document_permission(
|
|
request.user, document, required_permission=ManagedDocument.PERMISSION_EDIT
|
|
)
|
|
):
|
|
from django.core.exceptions import PermissionDenied
|
|
raise PermissionDenied
|
|
tag_id = request.POST.get("tag_id")
|
|
tag = get_object_or_404(DocumentTag, pk=tag_id)
|
|
document.tags.add(tag)
|
|
messages.success(request, _("Tag added."))
|
|
return redirect(document.get_absolute_url())
|
|
|
|
|
|
class RemoveTagFromDocumentView(LoginRequiredMixin, View):
|
|
def post(self, request, pk, tag_id):
|
|
document = get_object_or_404(
|
|
filter_documents_for_user(
|
|
ManagedDocument.objects.all(), request.user
|
|
),
|
|
pk=pk,
|
|
)
|
|
if not (
|
|
user_is_documents_admin(request.user)
|
|
or user_has_document_permission(
|
|
request.user, document, required_permission=ManagedDocument.PERMISSION_EDIT
|
|
)
|
|
):
|
|
from django.core.exceptions import PermissionDenied
|
|
raise PermissionDenied
|
|
tag = get_object_or_404(DocumentTag, pk=tag_id)
|
|
document.tags.remove(tag)
|
|
messages.success(request, _("Tag removed."))
|
|
return redirect(document.get_absolute_url())
|
|
|
|
|
|
class PublicDocumentView(View):
|
|
template_name = "documents/document_public.html"
|
|
|
|
def get(self, request, token):
|
|
link = get_object_or_404(DocumentAccessLink, token=token)
|
|
if link.is_expired:
|
|
raise Http404
|
|
document = link.document
|
|
return render(
|
|
request,
|
|
self.template_name,
|
|
{
|
|
"document": document,
|
|
"link": link,
|
|
"can_download": link.permission in {
|
|
ManagedDocument.PERMISSION_VIEW,
|
|
ManagedDocument.PERMISSION_COMMENT,
|
|
ManagedDocument.PERMISSION_EDIT,
|
|
ManagedDocument.PERMISSION_APPROVE,
|
|
},
|
|
},
|
|
)
|
|
|
|
|
|
class PublicDocumentDownloadView(View):
|
|
def get(self, request, token):
|
|
link = get_object_or_404(
|
|
DocumentAccessLink.objects.select_related("document", "document__current_version"),
|
|
token=token,
|
|
)
|
|
|
|
if link.is_expired:
|
|
raise Http404
|
|
|
|
if link.permission not in {
|
|
ManagedDocument.PERMISSION_VIEW,
|
|
ManagedDocument.PERMISSION_COMMENT,
|
|
ManagedDocument.PERMISSION_EDIT,
|
|
ManagedDocument.PERMISSION_APPROVE,
|
|
}:
|
|
raise PermissionDenied
|
|
|
|
document = link.document
|
|
version = document.latest_version
|
|
if version is None or not version.file:
|
|
raise Http404
|
|
|
|
file_field = version.file
|
|
if not file_field.name or not file_field.storage.exists(file_field.name):
|
|
raise Http404
|
|
|
|
response = FileResponse(file_field.open("rb"))
|
|
filename = smart_str(os.path.basename(file_field.name))
|
|
response["Content-Disposition"] = f'attachment; filename="{filename}"'
|
|
return response
|
|
|
|
|
|
class PublicFolderView(View):
|
|
template_name = "documents/documentfolder_public.html"
|
|
|
|
def get(self, request, token):
|
|
link = get_object_or_404(DocumentFolderAccessLink, token=token)
|
|
if link.is_expired:
|
|
raise Http404
|
|
|
|
folder = link.folder
|
|
documents = (
|
|
ManagedDocument.objects.filter(folders=folder)
|
|
.select_related("current_version")
|
|
.prefetch_related("tags", "types")
|
|
.order_by("title")
|
|
)
|
|
subfolders = (
|
|
DocumentFolder.objects.prefetch_related("parent_folders")
|
|
.filter(parent_folders=folder)
|
|
.order_by("name")
|
|
)
|
|
|
|
return render(
|
|
request,
|
|
self.template_name,
|
|
{
|
|
"folder": folder,
|
|
"link": link,
|
|
"documents": documents,
|
|
"subfolders": subfolders,
|
|
},
|
|
)
|
|
|
|
|
|
class SearchObjectsView(LoginRequiredMixin, View):
|
|
def get(self, request, pk):
|
|
from django.contrib.contenttypes.models import ContentType
|
|
from common.models import Thematic
|
|
from django.db.models import Q
|
|
|
|
document = get_object_or_404(ManagedDocument.objects.all(), pk=pk)
|
|
if not (
|
|
request.user.is_superuser
|
|
or user_has_document_permission(request.user, document, required_permission=ManagedDocument.PERMISSION_VIEW)
|
|
):
|
|
raise PermissionDenied
|
|
|
|
content_type_id = request.GET.get("content_type")
|
|
q = request.GET.get("q", "").strip()
|
|
|
|
if not content_type_id:
|
|
return JsonResponse({"results": []})
|
|
|
|
try:
|
|
content_type = ContentType.objects.get(pk=content_type_id)
|
|
model_class = content_type.model_class()
|
|
except (ContentType.DoesNotExist, ValueError):
|
|
return JsonResponse({"results": []})
|
|
|
|
if not model_class:
|
|
return JsonResponse({"results": []})
|
|
|
|
# Security: check if app_label and model are allowed
|
|
allowed_non_assets = {
|
|
("projects", "project"),
|
|
("interventions", "intervention"),
|
|
("observations", "observation"),
|
|
("controls", "control"),
|
|
}
|
|
is_allowed = (
|
|
content_type.app_label == "assets"
|
|
or (content_type.app_label, content_type.model) in allowed_non_assets
|
|
)
|
|
if not is_allowed:
|
|
return JsonResponse({"results": []})
|
|
|
|
queryset = model_class.objects.all()
|
|
|
|
# If the model is an asset (app_label is "assets"), we can optionally restrict to the document's thematics
|
|
if content_type.app_label == "assets":
|
|
thematics = Thematic.objects.filter(document_folders__documents=document).distinct()
|
|
if thematics.exists():
|
|
queryset = queryset.filter(category__thematic__in=thematics)
|
|
|
|
if q:
|
|
# Build filters dynamically based on existing fields
|
|
fields = {f.name for f in model_class._meta.get_fields()}
|
|
query_filter = Q()
|
|
if "code" in fields:
|
|
query_filter |= Q(code__icontains=q)
|
|
if "name" in fields:
|
|
query_filter |= Q(name__icontains=q)
|
|
if "name_fr" in fields:
|
|
query_filter |= Q(name_fr__icontains=q)
|
|
if "name_nl" in fields:
|
|
query_filter |= Q(name_nl__icontains=q)
|
|
if "title" in fields:
|
|
query_filter |= Q(title__icontains=q)
|
|
|
|
queryset = queryset.filter(query_filter)
|
|
|
|
queryset = queryset[:30]
|
|
|
|
results = []
|
|
for obj in queryset:
|
|
results.append({
|
|
"id": obj.pk,
|
|
"text": str(obj)
|
|
})
|
|
|
|
return JsonResponse({"results": results})
|
|
|
|
|
|
from pdf2image import convert_from_path, pdfinfo_from_path
|
|
from io import BytesIO
|
|
from django.core.cache import cache
|
|
import json
|
|
|
|
class DocumentPageImageView(DocumentsAppAccessMixin, LoginRequiredMixin, View):
|
|
def get(self, request, pk, page_num):
|
|
document = get_object_or_404(ManagedDocument.objects.all(), pk=pk)
|
|
if not (
|
|
request.user.is_superuser
|
|
or user_has_document_permission(request.user, document, required_permission=ManagedDocument.PERMISSION_VIEW)
|
|
):
|
|
raise PermissionDenied
|
|
|
|
latest_version = document.latest_version
|
|
if not latest_version or not latest_version.file:
|
|
raise Http404(_("No file found for this document."))
|
|
|
|
file_path = latest_version.file.path
|
|
if not os.path.exists(file_path):
|
|
raise Http404(_("File not found on disk."))
|
|
|
|
fname = latest_version.file.name
|
|
ext = os.path.splitext(fname)[1].lower()
|
|
|
|
# If it is already an image format, serve it directly
|
|
if ext in ['.png', '.jpg', '.jpeg', '.gif', '.bmp', '.svg', '.webp']:
|
|
response = FileResponse(latest_version.file.open("rb"))
|
|
response["Content-Disposition"] = f'inline; filename="{smart_str(os.path.basename(fname))}"'
|
|
response['X-Frame-Options'] = 'ALLOWALL'
|
|
return response
|
|
|
|
if ext in ['.dxf', '.dwg']:
|
|
preview_path = file_path + ".preview.png"
|
|
if not os.path.exists(preview_path):
|
|
try:
|
|
from .utils import render_cad_file_to_image
|
|
render_cad_file_to_image(file_path, preview_path)
|
|
except Exception as e:
|
|
raise Http404(f"Error rendering CAD file: {str(e)}")
|
|
|
|
if os.path.exists(preview_path):
|
|
response = FileResponse(open(preview_path, "rb"))
|
|
response["Content-Type"] = "image/png"
|
|
response["Content-Disposition"] = f'inline; filename="{smart_str(os.path.basename(fname))}.preview.png"'
|
|
response['X-Frame-Options'] = 'ALLOWALL'
|
|
return response
|
|
else:
|
|
raise Http404(_("Could not generate CAD preview image."))
|
|
|
|
if ext != '.pdf':
|
|
raise Http404(_("Unsupported file format for georeferencing image preview."))
|
|
|
|
# Cache key based on file path, page number and modification time
|
|
try:
|
|
mtime = os.path.getmtime(file_path)
|
|
except Exception:
|
|
mtime = 0
|
|
cache_key = f"doc_page_img_{document.pk}_{page_num}_{mtime}"
|
|
cached_img = cache.get(cache_key)
|
|
if cached_img:
|
|
return HttpResponse(cached_img, content_type="image/png")
|
|
|
|
try:
|
|
info = pdfinfo_from_path(file_path)
|
|
max_pages = info.get("Pages", 0)
|
|
if page_num < 1 or page_num > max_pages:
|
|
raise Http404(_("Page number out of range."))
|
|
|
|
images = convert_from_path(file_path, first_page=page_num, last_page=page_num, dpi=144)
|
|
if not images:
|
|
raise Http404(_("Page number out of range or empty PDF."))
|
|
|
|
image = images[0]
|
|
img_buffer = BytesIO()
|
|
image.save(img_buffer, format="PNG")
|
|
img_data = img_buffer.getvalue()
|
|
|
|
# Cache the image data for 1 hour
|
|
cache.set(cache_key, img_data, 3600)
|
|
|
|
return HttpResponse(img_data, content_type="image/png")
|
|
except Http404:
|
|
raise
|
|
except Exception as e:
|
|
raise Http404(f"Error rendering PDF page: {str(e)}")
|
|
|
|
|
|
def _get_associated_asset_coords(document):
|
|
"""
|
|
Traverse direct attachments or folder attachments (including recursively parent folders)
|
|
to find geolocated coordinates of an associated asset.
|
|
Returns a list [longitude, latitude] or None.
|
|
"""
|
|
def extract_coords(obj):
|
|
if not obj:
|
|
return None
|
|
# 1. Check for explicit lon/lat attributes
|
|
if hasattr(obj, "lon") and hasattr(obj, "lat") and obj.lon is not None and obj.lat is not None:
|
|
return [obj.lon, obj.lat]
|
|
# 2. Check for geom centroid, transforming to EPSG:4326 if needed
|
|
if hasattr(obj, "geom") and obj.geom:
|
|
try:
|
|
centroid = obj.geom.centroid
|
|
if centroid.srid != 4326:
|
|
centroid = centroid.clone()
|
|
centroid.transform(4326)
|
|
return [centroid.x, centroid.y]
|
|
except Exception:
|
|
pass
|
|
return None
|
|
|
|
# 1. Direct attachments
|
|
for attachment in document.attachments.all():
|
|
coords = extract_coords(attachment.content_object)
|
|
if coords:
|
|
return coords
|
|
|
|
# 2. Folder attachments (recursively checking folder hierarchy)
|
|
visited_folders = set()
|
|
folders_to_check = list(document.folders.all())
|
|
while folders_to_check:
|
|
folder = folders_to_check.pop(0)
|
|
if folder in visited_folders:
|
|
continue
|
|
visited_folders.add(folder)
|
|
|
|
# Check this folder's attachments
|
|
for attachment in folder.attachments.all():
|
|
coords = extract_coords(attachment.content_object)
|
|
if coords:
|
|
return coords
|
|
|
|
# Queue parent folders
|
|
for parent in folder.parent_folders.all():
|
|
if parent not in visited_folders:
|
|
folders_to_check.append(parent)
|
|
|
|
return None
|
|
|
|
|
|
class DocumentGeoreferenceView(DocumentsAppAccessMixin, LoginRequiredMixin, DetailView):
|
|
model = ManagedDocument
|
|
template_name = "documents/document_georeference.html"
|
|
context_object_name = "document"
|
|
|
|
def get_queryset(self):
|
|
queryset = super().get_queryset()
|
|
queryset = filter_documents_for_user(queryset, self.request.user)
|
|
return queryset
|
|
|
|
def get_context_data(self, **kwargs):
|
|
context = super().get_context_data(**kwargs)
|
|
document = self.object
|
|
|
|
if not (
|
|
self.request.user.is_superuser
|
|
or user_has_document_permission(self.request.user, document, required_permission=ManagedDocument.PERMISSION_EDIT)
|
|
):
|
|
raise PermissionDenied
|
|
|
|
mode = self.request.GET.get("mode", "overlay")
|
|
if mode not in ["anchor", "overlay"]:
|
|
mode = "overlay"
|
|
|
|
context["mode"] = mode
|
|
context["associated_coords"] = _get_associated_asset_coords(document)
|
|
return context
|
|
|
|
|
|
class DocumentGeoreferenceSaveView(DocumentsAppAccessMixin, LoginRequiredMixin, View):
|
|
def post(self, request, pk):
|
|
document = get_object_or_404(ManagedDocument.objects.all(), pk=pk)
|
|
if not (
|
|
request.user.is_superuser
|
|
or user_has_document_permission(request.user, document, required_permission=ManagedDocument.PERMISSION_EDIT)
|
|
):
|
|
raise PermissionDenied
|
|
|
|
try:
|
|
data = json.loads(request.body)
|
|
coords = data.get("coordinates")
|
|
page = data.get("page", 1)
|
|
|
|
if coords is not None:
|
|
if not isinstance(coords, list) or len(coords) != 4:
|
|
return JsonResponse({"status": "error", "message": "Invalid coordinates format. Expected 4 points."}, status=400)
|
|
for pt in coords:
|
|
if not isinstance(pt, list) or len(pt) != 2:
|
|
return JsonResponse({"status": "error", "message": "Each coordinate must be [lng, lat]."}, status=400)
|
|
|
|
document.georeference_coords = coords
|
|
document.georeference_page = page
|
|
document.save()
|
|
return JsonResponse({"status": "success", "message": "Georeference saved successfully."})
|
|
else:
|
|
return JsonResponse({"status": "error", "message": "No coordinates provided."}, status=400)
|
|
except Exception as e:
|
|
return JsonResponse({"status": "error", "message": str(e)}, status=400)
|
|
|
|
|
|
class DocumentMapView(DocumentsAppAccessMixin, LoginRequiredMixin, DetailView):
|
|
model = ManagedDocument
|
|
template_name = "documents/document_map_view.html"
|
|
context_object_name = "document"
|
|
|
|
def get_queryset(self):
|
|
queryset = super().get_queryset()
|
|
queryset = filter_documents_for_user(queryset, self.request.user)
|
|
return queryset
|
|
|
|
def get_context_data(self, **kwargs):
|
|
context = super().get_context_data(**kwargs)
|
|
document = self.object
|
|
if not (
|
|
self.request.user.is_superuser
|
|
or user_has_document_permission(self.request.user, document, required_permission=ManagedDocument.PERMISSION_VIEW)
|
|
):
|
|
raise PermissionDenied
|
|
|
|
if not document.georeference_coords:
|
|
raise Http404(_("This document is not georeferenced yet."))
|
|
|
|
return context |