added management command

CHANGELOG.txt

@@ -4,6 +4,8 @@ CHANGELOG
 1.3.3 (unreleased)
 ------------------
 
+* Added management command ``filer_check`` to check the integrity of the
+  database against the file system, and vice versa.
 * Enabled django-mptt 0.9
 * Converted QueryDict to dict before manipulating in admin
 * Hide 'Save as new' button in file admin

docs/management_commands.rst

@@ -12,3 +12,33 @@ command.
 To generate them, use::
 
     ./manage.py generate_thumbnails
+
+
+Filesystem Checks
+-----------------
+
+**django-filer** offers a few commands to check the integrity of the database against the files
+stored on disk. By invoking::
+
+    ./manage.py filer_check --missing
+
+all files which are referenced by the database, but missing on disk are reported.
+
+Invoking::
+
+    ./manage.py filer_check --delete-missing
+
+deletes those file references from the database.
+
+Invoking::
+
+    ./manage.py filer_check --orphanes
+
+lists all files found on disk belonging to the configured storage engine, but which
+are not referenced by the database.
+
+Invoking::
+
+    ./manage.py filer_check --delete-orphanes
+
+deletes those orphaned files from disk.

filer/management/commands/filer_check.py

@@ -0,0 +1,116 @@
+# -*- coding: utf-8 -*-
+from __future__ import unicode_literals
+
+import os
+
+from django.core.files.storage import DefaultStorage
+from django.core.management.base import BaseCommand, CommandError
+from django.utils.module_loading import import_string
+from django.utils.six.moves import input
+from django.utils.translation import ugettext_lazy as _
+from filer.settings import DEFAULT_FILER_STORAGES
+
+
+class Command(BaseCommand):
+    help = _("Look for orphaned files in media folders.")
+    storage = DefaultStorage()
+    prefix = DEFAULT_FILER_STORAGES['public']['main']['UPLOAD_TO_PREFIX']
+
+    def add_arguments(self, parser):
+        parser.add_argument(
+            '--orphans',
+            action='store_true',
+            dest='orphans',
+            default=False,
+            help=_("Walk through the media folders and look for orphaned files."),
+        )
+        parser.add_argument(
+            '--delete-orphans',
+            action='store_true',
+            dest='delete_orphans',
+            default=False,
+            help=_("Delete orphaned files from their media folders."),
+        )
+        parser.add_argument(
+            '--missing',
+            action='store_true',
+            dest='missing',
+            default=False,
+            help=_("Verify media folders and report about missing files."),
+        )
+        parser.add_argument(
+            '--delete-missing',
+            action='store_true',
+            dest='delete_missing',
+            default=False,
+            help=_("Delete references in database if files are missing in media folder."),
+        )
+        parser.add_argument(
+            '--noinput',
+            '--no-input',
+            action='store_false',
+            dest='interactive',
+            default=True,
+            help="Do NOT prompt the user for input of any kind."
+        )
+
+    def handle(self, *args, **options):
+        if options['missing']:
+            self.verify_references(options)
+        if options['delete_missing']:
+            if options['interactive']:
+                msg = _("\nThis will delete entries from your database. Are you sure you want to do this?\n\n"
+                        "Type 'yes' to continue, or 'no' to cancel: ")
+                if input(msg) != 'yes':
+                    raise CommandError("Aborted: Delete missing file entries from database.")
+            self.verify_references(options)
+
+        if options['orphans']:
+            self.verify_storages(options)
+        if options['delete_orphans']:
+            if options['interactive']:
+                msg = _("\nThis will delete orphaned files from your storage. Are you sure you want to do this?\n\n"
+                        "Type 'yes' to continue, or 'no' to cancel: ")
+                if input(msg) != 'yes':
+                    raise CommandError("Aborted: Delete orphaned files from storage.")
+            self.verify_storages(options)
+
+    def verify_references(self, options):
+        from filer.models.filemodels import File
+
+        for file in File.objects.all():
+            if not file.file.storage.exists(file.file.name):
+                if options['delete_missing']:
+                    file.delete()
+                    msg = _("Delete missing file reference '{}/{}' from database.")
+                else:
+                    msg = _("Referenced file '{}/{}' is missing in media folder.")
+                if options['verbosity'] > 2:
+                    self.stdout.write(msg.format(str(file.folder), str(file)))
+                elif options['verbosity']:
+                    self.stdout.write(os.path.join(str(file.folder), str(file)))
+
+    def verify_storages(self, options):
+        from filer.models.filemodels import File
+
+        def walk(prefix):
+            child_dirs, files = storage.listdir(prefix)
+            for filename in files:
+                relfilename = os.path.join(prefix, filename)
+                if not File.objects.filter(file=relfilename).exists():
+                    if options['delete_orphans']:
+                        storage.delete(relfilename)
+                        msg = _("Deleted orphanded file '{}'")
+                    else:
+                        msg = _("Found orphanded file '{}'")
+                    if options['verbosity'] > 2:
+                        self.stdout.write(msg.format(relfilename))
+                    elif options['verbosity']:
+                        self.stdout.write(relfilename)
+
+            for child in child_dirs:
+                walk(os.path.join(prefix, child))
+
+        filer_public = DEFAULT_FILER_STORAGES['public']['main']
+        storage = import_string(filer_public['ENGINE'])()
+        walk(filer_public['UPLOAD_TO_PREFIX'])

filer/tests/__init__.py

@@ -9,3 +9,4 @@
 from .server_backends import *
 from .tools import *
 from .utils import *
+from .filer_check import *

filer/tests/filer_check.py

@@ -0,0 +1,52 @@
+#-*- coding: utf-8 -*-
+from __future__ import absolute_import
+
+import os, shutil
+
+from django.core.management import call_command
+from django.utils.six import StringIO
+from django.utils.module_loading import import_string
+
+from filer.models.filemodels import File
+from filer.settings import DEFAULT_FILER_STORAGES
+from .server_backends import BaseServerBackendTestCase
+
+
+class FilerCheckTestCase(BaseServerBackendTestCase):
+    def test_delete_missing(self):
+        out = StringIO()
+        self.assertTrue(os.path.exists(self.filer_file.file.path))
+        file_pk = self.filer_file.id
+        call_command('filer_check', stdout=out, missing=True)
+        self.assertEqual('', out.getvalue())
+
+        os.remove(self.filer_file.file.path)
+        call_command('filer_check', stdout=out, missing=True)
+        self.assertEqual("None/testimage.jpg\n", out.getvalue())
+        self.assertIsInstance(File.objects.get(id=file_pk), File)
+
+        call_command('filer_check', delete_missing=True, interactive=False, verbosity=0)
+        with self.assertRaises(File.DoesNotExist):
+            File.objects.get(id=file_pk)
+
+    def test_delete_orphanes(self):
+        out = StringIO()
+        self.assertTrue(os.path.exists(self.filer_file.file.path))
+        call_command('filer_check', stdout=out, orphanes=True)
+        self.assertEqual('', out.getvalue())
+
+        # add an orphan file to our storage
+        storage = import_string(DEFAULT_FILER_STORAGES['public']['main']['ENGINE'])()
+        filer_public = os.path.join(storage.base_location, DEFAULT_FILER_STORAGES['public']['main']['UPLOAD_TO_PREFIX'])
+        if os.path.isdir(filer_public):
+            shutil.rmtree(filer_public)
+        os.mkdir(filer_public)
+        orphan_file = os.path.join(filer_public, 'hello.txt')
+        with open(orphan_file, 'w') as fh:
+            fh.write("I don't belong here!")
+        call_command('filer_check', stdout=out, orphans=True)
+        self.assertEqual("filer_public/hello.txt\n", out.getvalue())
+        self.assertTrue(os.path.exists(orphan_file))
+
+        call_command('filer_check', delete_orphans=True, interactive=False, verbosity=0)
+        self.assertFalse(os.path.exists(orphan_file))