copies or symlinks static files from different locations to the settings staticroot set instance variables based on an options dict perform the bulk of the work of collectstatic split off from handle to facilitate testing prefix the relative path if the source storage contains it storage backends may define a postprocess method add a blank line before the traceback otherwise it s too easy to miss the relevant part of the error message destination files existence not checked play it safe and warn small log helper delete the given relative path using the destination storage backend delete broken symlinks check if the target file should be deleted if it already exists when was the target file modified last time the storage doesn t support getmodifiedtime or failed when was the source file modified last time the full path of the target file if it s link mode and the path isn t a link i e the previous collectstatic wasn t with link or if it s nonlink mode and the path is a link i e the previous collectstatic was with link the old linksfiles must be deleted so it s not safe to skip unmodified files in remote storages skipping is only based on the modified times since symlinks aren t relevant avoid subsecond precision see 14665 19540 then delete the existing file if really needed attempt to link path skip this file if it was already copied earlier delete the target file if needed or break the full path of the source file finally link the file attempt to copy path with storage skip this file if it was already copied earlier delete the target file if needed or break the full path of the source file finally start copying copies or symlinks static files from different locations to the settings static_root set instance variables based on an options dict perform the bulk of the work of collectstatic split off from handle to facilitate testing prefix the relative path if the source storage contains it storage backends may define a post_process method add a blank line before the traceback otherwise it s too easy to miss the relevant part of the error message destination files existence not checked play it safe and warn small log helper delete the given relative path using the destination storage backend delete broken symlinks check if the target file should be deleted if it already exists when was the target file modified last time the storage doesn t support get_modified_time or failed when was the source file modified last time the full path of the target file if it s link mode and the path isn t a link i e the previous collectstatic wasn t with link or if it s non link mode and the path is a link i e the previous collectstatic was with link the old links files must be deleted so it s not safe to skip unmodified files in remote storages skipping is only based on the modified times since symlinks aren t relevant avoid sub second precision see 14665 19540 then delete the existing file if really needed attempt to link path skip this file if it was already copied earlier delete the target file if needed or break the full path of the source file finally link the file attempt to copy path with storage skip this file if it was already copied earlier delete the target file if needed or break the full path of the source file finally start copying
import os

from django.apps import apps
from django.contrib.staticfiles.finders import get_finders
from django.contrib.staticfiles.storage import staticfiles_storage
from django.core.checks import Tags
from django.core.files.storage import FileSystemStorage
from django.core.management.base import BaseCommand, CommandError
from django.core.management.color import no_style
from django.utils.functional import cached_property


class Command(BaseCommand):
    

    help = "Collect static files in a single location."
    requires_system_checks = [Tags.staticfiles]

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.copied_files = []
        self.symlinked_files = []
        self.unmodified_files = []
        self.post_processed_files = []
        self.storage = staticfiles_storage
        self.style = no_style()

    @cached_property
    def local(self):
        try:
            self.storage.path("")
        except NotImplementedError:
            return False
        return True

    def add_arguments(self, parser):
        parser.add_argument(
            "--noinput",
            "--no-input",
            action="store_false",
            dest="interactive",
            help="Do NOT prompt the user for input of any kind.",
        )
        parser.add_argument(
            "--no-post-process",
            action="store_false",
            dest="post_process",
            help="Do NOT post process collected files.",
        )
        parser.add_argument(
            "-i",
            "--ignore",
            action="append",
            default=[],
            dest="ignore_patterns",
            metavar="PATTERN",
            help="Ignore files or directories matching this glob-style "
            "pattern. Use multiple times to ignore more.",
        )
        parser.add_argument(
            "-n",
            "--dry-run",
            action="store_true",
            help="Do everything except modify the filesystem.",
        )
        parser.add_argument(
            "-c",
            "--clear",
            action="store_true",
            help="Clear the existing files using the storage "
            "before trying to copy or link the original file.",
        )
        parser.add_argument(
            "-l",
            "--link",
            action="store_true",
            help="Create a symbolic link to each file instead of copying.",
        )
        parser.add_argument(
            "--no-default-ignore",
            action="store_false",
            dest="use_default_ignore_patterns",
            help=(
                "Don't ignore the common private glob-style patterns (defaults to "
                "'CVS', '.*' and '*~')."
            ),
        )

    def set_options(self, **options):
        
        self.interactive = options["interactive"]
        self.verbosity = options["verbosity"]
        self.symlink = options["link"]
        self.clear = options["clear"]
        self.dry_run = options["dry_run"]
        ignore_patterns = options["ignore_patterns"]
        if options["use_default_ignore_patterns"]:
            ignore_patterns += apps.get_app_config("staticfiles").ignore_patterns
        self.ignore_patterns = list({os.path.normpath(p) for p in ignore_patterns})
        self.post_process = options["post_process"]

    def collect(self):
        
        if self.symlink and not self.local:
            raise CommandError("Can't symlink to a remote destination.")

        if self.clear:
            self.clear_dir("")

        if self.symlink:
            handler = self.link_file
        else:
            handler = self.copy_file

        found_files = {}
        for finder in get_finders():
            for path, storage in finder.list(self.ignore_patterns):
                
                if getattr(storage, "prefix", None):
                    prefixed_path = os.path.join(storage.prefix, path)
                else:
                    prefixed_path = path

                if prefixed_path not in found_files:
                    found_files[prefixed_path] = (storage, path)
                    handler(path, prefixed_path, storage)
                else:
                    self.log(
                        "Found another file with the destination path '%s'. It "
                        "will be ignored since only the first encountered file "
                        "is collected. If this is not what you want, make sure "
                        "every static file has a unique path." % prefixed_path,
                        level=1,
                    )

        
        if self.post_process and hasattr(self.storage, "post_process"):
            processor = self.storage.post_process(found_files, dry_run=self.dry_run)
            for original_path, processed_path, processed in processor:
                if isinstance(processed, Exception):
                    self.stderr.write("Post-processing '%s' failed!" % original_path)
                    
                    
                    self.stderr.write()
                    raise processed
                if processed:
                    self.log(
                        "Post-processed '%s' as '%s'" % (original_path, processed_path),
                        level=2,
                    )
                    self.post_processed_files.append(original_path)
                else:
                    self.log("Skipped post-processing '%s'" % original_path)

        return {
            "modified": self.copied_files + self.symlinked_files,
            "unmodified": self.unmodified_files,
            "post_processed": self.post_processed_files,
        }

    def handle(self, **options):
        self.set_options(**options)
        message = ["\n"]
        if self.dry_run:
            message.append(
                "You have activated the --dry-run option so no files will be "
                "modified.\n\n"
            )

        message.append(
            "You have requested to collect static files at the destination\n"
            "location as specified in your settings"
        )

        if self.is_local_storage() and self.storage.location:
            destination_path = self.storage.location
            message.append(":\n\n    %s\n\n" % destination_path)
            should_warn_user = self.storage.exists(destination_path) and any(
                self.storage.listdir(destination_path)
            )
        else:
            destination_path = None
            message.append(".\n\n")
            
            should_warn_user = True

        if self.interactive and should_warn_user:
            if self.clear:
                message.append("This will DELETE ALL FILES in this location!\n")
            else:
                message.append("This will overwrite existing files!\n")

            message.append(
                "Are you sure you want to do this?\n\n"
                "Type 'yes' to continue, or 'no' to cancel: "
            )
            if input("".join(message)) != "yes":
                raise CommandError("Collecting static files cancelled.")

        collected = self.collect()

        if self.verbosity >= 1:
            modified_count = len(collected["modified"])
            unmodified_count = len(collected["unmodified"])
            post_processed_count = len(collected["post_processed"])
            return (
                "\n%(modified_count)s %(identifier)s %(action)s"
                "%(destination)s%(unmodified)s%(post_processed)s."
            ) % {
                "modified_count": modified_count,
                "identifier": "static file" + ("" if modified_count == 1 else "s"),
                "action": "symlinked" if self.symlink else "copied",
                "destination": (
                    " to '%s'" % destination_path if destination_path else ""
                ),
                "unmodified": (
                    ", %s unmodified" % unmodified_count
                    if collected["unmodified"]
                    else ""
                ),
                "post_processed": (
                    collected["post_processed"]
                    and ", %s post-processed" % post_processed_count
                    or ""
                ),
            }

    def log(self, msg, level=2):
        
        if self.verbosity >= level:
            self.stdout.write(msg)

    def is_local_storage(self):
        return isinstance(self.storage, FileSystemStorage)

    def clear_dir(self, path):
        
        if not self.storage.exists(path):
            return

        dirs, files = self.storage.listdir(path)
        for f in files:
            fpath = os.path.join(path, f)
            if self.dry_run:
                self.log("Pretending to delete '%s'" % fpath, level=1)
            else:
                self.log("Deleting '%s'" % fpath, level=1)
                try:
                    full_path = self.storage.path(fpath)
                except NotImplementedError:
                    self.storage.delete(fpath)
                else:
                    if not os.path.exists(full_path) and os.path.lexists(full_path):
                        
                        os.unlink(full_path)
                    else:
                        self.storage.delete(fpath)
        for d in dirs:
            self.clear_dir(os.path.join(path, d))

    def delete_file(self, path, prefixed_path, source_storage):
        
        if self.storage.exists(prefixed_path):
            try:
                
                target_last_modified = self.storage.get_modified_time(prefixed_path)
            except (OSError, NotImplementedError, AttributeError):
                
                pass
            else:
                try:
                    
                    source_last_modified = source_storage.get_modified_time(path)
                except (OSError, NotImplementedError, AttributeError):
                    pass
                else:
                    
                    if self.local:
                        full_path = self.storage.path(prefixed_path)
                        
                        
                        
                        
                        
                        
                        can_skip_unmodified_files = not (
                            self.symlink ^ os.path.islink(full_path)
                        )
                    else:
                        
                        
                        can_skip_unmodified_files = True
                    
                    file_is_unmodified = target_last_modified.replace(
                        microsecond=0
                    ) >= source_last_modified.replace(microsecond=0)
                    if file_is_unmodified and can_skip_unmodified_files:
                        if prefixed_path not in self.unmodified_files:
                            self.unmodified_files.append(prefixed_path)
                        self.log("Skipping '%s' (not modified)" % path)
                        return False
            
            if self.dry_run:
                self.log("Pretending to delete '%s'" % path)
            else:
                self.log("Deleting '%s'" % path)
                self.storage.delete(prefixed_path)
        return True

    def link_file(self, path, prefixed_path, source_storage):
        
        
        if prefixed_path in self.symlinked_files:
            return self.log("Skipping '%s' (already linked earlier)" % path)
        
        if not self.delete_file(path, prefixed_path, source_storage):
            return
        
        source_path = source_storage.path(path)
        
        if self.dry_run:
            self.log("Pretending to link '%s'" % source_path, level=1)
        else:
            self.log("Linking '%s'" % source_path, level=2)
            full_path = self.storage.path(prefixed_path)
            os.makedirs(os.path.dirname(full_path), exist_ok=True)
            try:
                if os.path.lexists(full_path):
                    os.unlink(full_path)
                os.symlink(source_path, full_path)
            except NotImplementedError:
                import platform

                raise CommandError(
                    "Symlinking is not supported in this "
                    "platform (%s)." % platform.platform()
                )
            except OSError as e:
                raise CommandError(e)
        if prefixed_path not in self.symlinked_files:
            self.symlinked_files.append(prefixed_path)

    def copy_file(self, path, prefixed_path, source_storage):
        
        
        if prefixed_path in self.copied_files:
            return self.log("Skipping '%s' (already copied earlier)" % path)
        
        if not self.delete_file(path, prefixed_path, source_storage):
            return
        
        source_path = source_storage.path(path)
        
        if self.dry_run:
            self.log("Pretending to copy '%s'" % source_path, level=1)
        else:
            self.log("Copying '%s'" % source_path, level=2)
            with source_storage.open(path) as source_file:
                self.storage.save(prefixed_path, source_file)
        self.copied_files.append(prefixed_path)