Add management command to delete orphaned netbox attachments

netbox_attachments/management/cleanup.py

@@ -0,0 +1,84 @@
+# -*- coding: utf-8 -*-
+
+import os
+import re
+import time
+
+import six
+from django.conf import settings
+from django.core.validators import EMPTY_VALUES
+
+from netbox_attachments.management.utils import get_file_fields
+from netbox_attachments.utils import ATTACHMENT_MEDIA_ROOT
+
+
+def get_used_media():
+    """
+    Get media which are still used in models
+    """
+
+    media = set()
+
+    for field in get_file_fields():
+        is_null = {
+            "%s__isnull" % field.name: True,
+        }
+        is_empty = {
+            "%s" % field.name: "",
+        }
+
+        storage = field.storage
+
+        for value in (
+            field.model._base_manager.values_list(field.name, flat=True)
+            .exclude(**is_empty)
+            .exclude(**is_null)
+        ):
+            if value not in EMPTY_VALUES:
+                media.add(storage.path(value))
+
+    return media
+
+
+def get_all_media(exclude=None, minimum_file_age=None):
+    """
+    Get all media from MEDIA_ROOT/ATTACHMENT_PATH
+    """
+
+    if not exclude:
+        exclude = []
+
+    media = set()
+    initial_time = time.time()
+
+    for root, dirs, files in os.walk(six.text_type(ATTACHMENT_MEDIA_ROOT)):
+        for name in files:
+            path = os.path.abspath(os.path.join(root, name))
+            relpath = os.path.relpath(path, settings.MEDIA_ROOT)
+
+            if minimum_file_age:
+                file_age = initial_time - os.path.getmtime(path)
+                if file_age < minimum_file_age:
+                    continue
+
+            for e in exclude:
+                if re.match(r"^%s$" % re.escape(e).replace("\\*", ".*"), relpath):
+                    break
+            else:
+                media.add(path)
+
+    return media
+
+
+def get_unused_media(exclude=None, minimum_file_age=None):
+    """
+    Get media which are not used in models
+    """
+
+    if not exclude:
+        exclude = []
+
+    all_media = get_all_media(exclude, minimum_file_age)
+    used_media = get_used_media()
+
+    return all_media - used_media

netbox_attachments/management/commands/remove_orphaned_netbox_attachments.py

@@ -0,0 +1,114 @@
+# -*- coding: utf-8 -*-
+
+import os
+
+import six.moves
+from django.conf import settings
+from django.core.management.base import BaseCommand
+
+from netbox_attachments.management.cleanup import (
+    get_unused_media,
+    ATTACHMENT_MEDIA_ROOT,
+)
+
+
+class Command(BaseCommand):
+    help = "Clean unused media files which have no reference in models"
+
+    # verbosity
+    # 0 means silent
+    # 1 means normal output (default).
+    # 2 means verbose output
+
+    verbosity = 1
+
+    def add_arguments(self, parser):
+        parser.add_argument(
+            "--noinput",
+            "--no-input",
+            dest="interactive",
+            action="store_false",
+            default=True,
+            help="Do not ask confirmation",
+        )
+
+        parser.add_argument(
+            "-e",
+            "--exclude",
+            dest="exclude",
+            action="append",
+            default=[],
+            help="Exclude files by mask (only * is supported), can use multiple --exclude",
+        )
+
+        parser.add_argument(
+            "--minimum-file-age",
+            dest="minimum_file_age",
+            default=60,
+            type=int,
+            help="Skip files younger this age (sec) - default 60 sec",
+        )
+
+        parser.add_argument(
+            "-n",
+            "--dry-run",
+            dest="dry_run",
+            action="store_true",
+            default=False,
+            help="Dry run without any affect on your data",
+        )
+
+    def info(self, message):
+        if self.verbosity > 0:
+            self.stdout.write(message)
+
+    def debug(self, message):
+        if self.verbosity > 1:
+            self.stdout.write(message)
+
+    def _show_files_to_delete(self, unused_media):
+        self.debug("Files to remove:")
+
+        for f in unused_media:
+            self.debug(f)
+
+        self.info("Total files will be removed: {}".format(len(unused_media)))
+
+    def handle(self, *args, **options):
+        if "verbosity" in options:
+            self.verbosity = options["verbosity"]
+
+        self.info("Scanned folder: {}".format(ATTACHMENT_MEDIA_ROOT))
+
+        unused_media = get_unused_media(
+            exclude=options.get("exclude"),
+            minimum_file_age=options.get("minimum_file_age"),
+        )
+
+        if not unused_media:
+            self.info("Nothing to delete. Exit")
+            return
+
+        if options.get("dry_run"):
+            self._show_files_to_delete(unused_media)
+            self.info("Dry run. Exit.")
+            return
+
+        if options.get("interactive"):
+            self._show_files_to_delete(unused_media)
+
+            # ask user
+
+            question = "Are you sure you want to remove {} unused files? (y/N)".format(
+                len(unused_media)
+            )
+
+            if six.moves.input(question).upper() != "Y":
+                self.info("Interrupted by user. Exit.")
+                return
+
+        for f in unused_media:
+            self.debug("Remove %s" % f)
+            os.remove(os.path.join(settings.MEDIA_ROOT, f))
+
+        self.info("Done. Total files removed: {}".format(len(unused_media)))

netbox_attachments/management/utils.py

@@ -0,0 +1,25 @@
+# -*- coding: utf-8 -*-
+
+from django.apps import apps
+from django.db import models
+from netbox_attachments import NetBoxAttachmentsConfig
+
+
+def get_file_fields():
+    """
+    Get all fields which are inherited from FileField in NetBoxAttachment plugin models
+    """
+
+    # get NetBox Attachment models
+    attachment_app_config = apps.get_app_config(NetBoxAttachmentsConfig.name)
+    attachment_models = attachment_app_config.get_models()
+    # get fields
+
+    fields = []
+
+    for model in attachment_models:
+        for field in model._meta.get_fields():
+            if isinstance(field, models.FileField):
+                fields.append(field)
+
+    return fields

netbox_attachments/utils.py

@@ -1,17 +1,22 @@
+import os
 from pathlib import Path
 
+from django.conf import settings
+
+ATTACHMENT_PATH = "netbox-attachments/"
+ATTACHMENT_MEDIA_ROOT = os.path.join(settings.MEDIA_ROOT, ATTACHMENT_PATH)
+
 
 def attachment_upload(instance, filename):
     """
     Return a path for uploading file attchments.
     """
-    path = "netbox-attachments/"
 
     if instance.name != filename:
         # Rename the file to the provided name, if any. Attempt to preserve the file extension.
         extension = "".join(Path(filename).suffixes)
         filename = "".join([instance.name, extension])
 
     return "{}{}_{}_{}".format(
-        path, instance.object_type.name, instance.object_id, filename
+        ATTACHMENT_PATH, instance.object_type.name, instance.object_id, filename
     )

netbox_attachments/version.py

@@ -1 +1 @@
-__version__ = "5.1.3"
+__version__ = "5.1.4-b1"