240 lines
12 KiB
Python
240 lines
12 KiB
Python
import json
|
|
from pathlib import Path, PurePosixPath
|
|
from tempfile import SpooledTemporaryFile
|
|
from zipfile import ZIP_DEFLATED, BadZipFile, ZipFile
|
|
|
|
from django.conf import settings
|
|
from django.contrib.contenttypes.models import ContentType
|
|
from django.db import transaction
|
|
from django.utils.text import slugify
|
|
from tenancy.models import Tenant, TenantGroup
|
|
|
|
from .models import Document, DocumentAssignment, DocumentAttachment, DocumentCategory
|
|
|
|
|
|
ARCHIVE_FORMAT = "netbox-documentation"
|
|
ARCHIVE_VERSION = 1
|
|
MAX_ENTRIES = 5000
|
|
|
|
|
|
class ArchiveFailure(ValueError):
|
|
pass
|
|
|
|
|
|
def documents_for_categories(categories, include_subfolders=True, user=None):
|
|
"""Return distinct permitted documents contained in the selected category trees."""
|
|
category_ids = set(categories.values_list("pk", flat=True))
|
|
frontier = set(category_ids)
|
|
while include_subfolders and frontier:
|
|
children = set(DocumentCategory.objects.filter(parent_id__in=frontier).values_list("pk", flat=True))
|
|
frontier = children - category_ids
|
|
category_ids.update(children)
|
|
documents = Document.objects.filter(category_id__in=category_ids).distinct().order_by("title")
|
|
if user is not None:
|
|
documents = documents.restrict(user, "view")
|
|
return documents
|
|
|
|
|
|
def _category_path(category):
|
|
path, seen = [], set()
|
|
while category and category.pk not in seen:
|
|
path.append(category.name)
|
|
seen.add(category.pk)
|
|
category = category.parent
|
|
return list(reversed(path))
|
|
|
|
|
|
def _category_tenancy_path(category):
|
|
path, seen = [], set()
|
|
while category and category.pk not in seen:
|
|
path.append({
|
|
"tenant": category.tenant.slug if category.tenant else None,
|
|
"tenant_group": category.tenant_group.slug if category.tenant_group else None,
|
|
})
|
|
seen.add(category.pk)
|
|
category = category.parent
|
|
return list(reversed(path))
|
|
|
|
|
|
def export_documents(documents):
|
|
stream = SpooledTemporaryFile(max_size=10 * 1024 * 1024, mode="w+b")
|
|
manifest = {"format": ARCHIVE_FORMAT, "version": ARCHIVE_VERSION, "documents": []}
|
|
with ZipFile(stream, "w", compression=ZIP_DEFLATED, compresslevel=6) as archive:
|
|
for document in documents.prefetch_related("assignments__assigned_object_type", "attachments").select_related(
|
|
"category", "tenant", "tenant_group", "category__tenant", "category__tenant_group",
|
|
):
|
|
root = f"documents/{document.pk}"
|
|
extension = "html" if document.body_format == "html" else "md"
|
|
body_path = f"{root}/content.{extension}"
|
|
archive.writestr(body_path, document.body.encode("utf-8"))
|
|
attachments = []
|
|
for attachment in document.attachments.all():
|
|
safe_name = Path(attachment.original_name).name or f"attachment-{attachment.pk}"
|
|
member = f"{root}/attachments/{attachment.pk}-{safe_name}"
|
|
try:
|
|
with attachment.file.open("rb") as source:
|
|
archive.writestr(member, source.read())
|
|
except (FileNotFoundError, OSError):
|
|
continue
|
|
attachments.append({
|
|
"path": member, "original_name": safe_name,
|
|
"content_type": attachment.content_type, "size": attachment.size,
|
|
"source_url": attachment.file.url,
|
|
})
|
|
assignments = [{
|
|
"app_label": item.assigned_object_type.app_label,
|
|
"model": item.assigned_object_type.model,
|
|
"object_id": item.assigned_object_id,
|
|
"note": item.note,
|
|
} for item in document.assignments.all()]
|
|
manifest["documents"].append({
|
|
"title": document.title, "slug": document.slug, "summary": document.summary,
|
|
"body_format": document.body_format, "is_published": document.is_published,
|
|
"category_path": _category_path(document.category), "body_path": body_path,
|
|
"category_tenancy": _category_tenancy_path(document.category),
|
|
"tenant": document.tenant.slug if document.tenant else None,
|
|
"tenant_group": document.tenant_group.slug if document.tenant_group else None,
|
|
"assignments": assignments, "attachments": attachments,
|
|
})
|
|
archive.writestr("manifest.json", json.dumps(manifest, ensure_ascii=False, indent=2).encode("utf-8"))
|
|
stream.seek(0)
|
|
return stream
|
|
|
|
|
|
def _validate_archive(archive):
|
|
infos = archive.infolist()
|
|
if len(infos) > MAX_ENTRIES:
|
|
raise ArchiveFailure(f"Das Archiv enthält mehr als {MAX_ENTRIES} Dateien.")
|
|
limit = settings.PLUGINS_CONFIG.get("netbox_documentation", {}).get("max_archive_size_mb", 250) * 1024 * 1024
|
|
if sum(info.file_size for info in infos) > limit:
|
|
raise ArchiveFailure("Die entpackte Gesamtgröße überschreitet das konfigurierte Limit.")
|
|
for info in infos:
|
|
path = PurePosixPath(info.filename)
|
|
if path.is_absolute() or ".." in path.parts:
|
|
raise ArchiveFailure("Das Archiv enthält einen unsicheren Dateipfad.")
|
|
if info.compress_size and info.file_size / info.compress_size > 200:
|
|
raise ArchiveFailure("Das Archiv enthält eine verdächtig stark komprimierte Datei.")
|
|
|
|
|
|
def _read_json(archive, name):
|
|
try:
|
|
return json.loads(archive.read(name).decode("utf-8"))
|
|
except (KeyError, UnicodeDecodeError, json.JSONDecodeError) as exc:
|
|
raise ArchiveFailure("Das Archiv enthält kein gültiges manifest.json.") from exc
|
|
|
|
|
|
def _category_from_path(names, tenancy_path=None):
|
|
parent = None
|
|
tenancy_path = tenancy_path or []
|
|
for index, name in enumerate(names):
|
|
name = str(name).strip()[:100]
|
|
if not name:
|
|
continue
|
|
slug = slugify(name)[:100] or "ordner"
|
|
category = DocumentCategory.objects.filter(parent=parent, slug=slug).first()
|
|
if not category:
|
|
category = DocumentCategory.objects.create(name=name, slug=slug, parent=parent)
|
|
tenancy = tenancy_path[index] if index < len(tenancy_path) and isinstance(tenancy_path[index], dict) else {}
|
|
if tenancy:
|
|
category.tenant_group = TenantGroup.objects.filter(slug=tenancy.get("tenant_group")).first()
|
|
category.tenant = Tenant.objects.filter(slug=tenancy.get("tenant")).first()
|
|
category.save(update_fields=("tenant_group", "tenant", "last_updated"))
|
|
parent = category
|
|
return parent
|
|
|
|
|
|
def _unique_slug(value):
|
|
base = (slugify(value) or "dokumentation")[:180]
|
|
candidate, number = base, 2
|
|
while Document.objects.filter(slug=candidate).exists():
|
|
candidate = f"{base[:190-len(str(number))]}-{number}"
|
|
number += 1
|
|
return candidate
|
|
|
|
|
|
@transaction.atomic
|
|
def import_archive(upload, update_existing=False):
|
|
try:
|
|
archive = ZipFile(upload)
|
|
except BadZipFile as exc:
|
|
raise ArchiveFailure("Die hochgeladene Datei ist kein gültiges ZIP-Archiv.") from exc
|
|
with archive:
|
|
_validate_archive(archive)
|
|
manifest = _read_json(archive, "manifest.json")
|
|
if manifest.get("format") != ARCHIVE_FORMAT or manifest.get("version") != ARCHIVE_VERSION:
|
|
raise ArchiveFailure("Archivformat oder Version wird nicht unterstützt.")
|
|
records = manifest.get("documents")
|
|
if not isinstance(records, list):
|
|
raise ArchiveFailure("Die Dokumentliste im Archiv ist ungültig.")
|
|
allowed = set(settings.PLUGINS_CONFIG.get("netbox_documentation", {}).get("allowed_object_types", []))
|
|
created, updated, skipped_assignments = 0, 0, 0
|
|
for record in records:
|
|
if not isinstance(record, dict) or not record.get("title") or not record.get("body_path"):
|
|
raise ArchiveFailure("Das Archiv enthält einen unvollständigen Dokumenteintrag.")
|
|
try:
|
|
body = archive.read(record["body_path"]).decode("utf-8")
|
|
except (KeyError, UnicodeDecodeError) as exc:
|
|
raise ArchiveFailure("Ein Dokumentinhalt fehlt oder ist nicht UTF-8-kodiert.") from exc
|
|
source_slug = str(record.get("slug") or record["title"])
|
|
document = Document.objects.filter(slug=source_slug).first() if update_existing else None
|
|
category = _category_from_path(
|
|
record.get("category_path") or [], record.get("category_tenancy") or [],
|
|
)
|
|
if document:
|
|
updated += 1
|
|
else:
|
|
document = Document(slug=_unique_slug(source_slug))
|
|
created += 1
|
|
document.title = str(record["title"])[:200]
|
|
document.summary = str(record.get("summary") or "")[:500]
|
|
document.body = body
|
|
document.body_format = record.get("body_format") if record.get("body_format") in {"html", "markdown"} else "html"
|
|
document.is_published = bool(record.get("is_published", True))
|
|
document.category = category
|
|
document.tenant_group = TenantGroup.objects.filter(slug=record.get("tenant_group")).first()
|
|
document.tenant = Tenant.objects.filter(slug=record.get("tenant")).first()
|
|
document.save()
|
|
for assignment in record.get("assignments") or []:
|
|
label = f"{assignment.get('app_label')}.{assignment.get('model')}"
|
|
if label not in allowed:
|
|
skipped_assignments += 1
|
|
continue
|
|
content_type = ContentType.objects.filter(app_label=assignment.get("app_label"), model=assignment.get("model")).first()
|
|
model = content_type.model_class() if content_type else None
|
|
if not model or not model.objects.filter(pk=assignment.get("object_id")).exists():
|
|
skipped_assignments += 1
|
|
continue
|
|
DocumentAssignment.objects.get_or_create(
|
|
document=document, assigned_object_type=content_type,
|
|
assigned_object_id=assignment["object_id"],
|
|
defaults={"note": str(assignment.get("note") or "")[:200]},
|
|
)
|
|
for item in record.get("attachments") or []:
|
|
member = item.get("path")
|
|
if not member:
|
|
continue
|
|
try:
|
|
content = archive.read(member)
|
|
except KeyError as exc:
|
|
raise ArchiveFailure(f"Anhang {member} fehlt im Archiv.") from exc
|
|
from django.core.files.base import ContentFile
|
|
original_name = Path(str(item.get("original_name") or "attachment")).name[:255]
|
|
allowed_extensions = {".docx", ".xlsx", ".xlsm", ".pdf", ".md", ".txt", ".jpg", ".jpeg", ".png", ".gif", ".webp"}
|
|
if Path(original_name).suffix.lower() not in allowed_extensions:
|
|
raise ArchiveFailure(f"Der Anhang {original_name} verwendet einen nicht erlaubten Dateityp.")
|
|
existing = document.attachments.filter(original_name=original_name, size=len(content)).first()
|
|
if existing:
|
|
if item.get("source_url"):
|
|
document.body = document.body.replace(str(item["source_url"]), existing.file.url)
|
|
continue
|
|
attachment = DocumentAttachment(
|
|
document=document, original_name=original_name,
|
|
content_type=str(item.get("content_type") or "")[:100], size=len(content),
|
|
)
|
|
attachment.file.save(original_name, ContentFile(content), save=False)
|
|
attachment.save()
|
|
if item.get("source_url"):
|
|
document.body = document.body.replace(str(item["source_url"]), attachment.file.url)
|
|
document.save(update_fields=("body", "last_updated"))
|
|
return {"created": created, "updated": updated, "skipped_assignments": skipped_assignments}
|