Deduplication by hard-linking files in multiple directories

Question

About a year ago, I wrote a Python script that enumerates all the files in a directory, partitions them based on a couple of criteria (file size, a custom byte-signature of the file, hash, ...) and finally creates hardlinks between identical files. The partitioning is intended to minimise the number of full-file comparisons that are performed. There's also a progress reporter using multiple progress bars on the terminal.

I think it's high time this code is viewed from a different perspective, so please provide a thorough review. Feel free to tear it apart completely, I've long forgotten this code existed. Off the bat, I can already see multiple issues with it, but I don't want to influence the review. I'm hoping to find new flaws in my coding style.

As for design decisions that went into this code: It ran on a corpus of multiple hundreds of files of varying sizes, totalling nearly 10TB of data. Files were of different types, some textual, some binary. There were also many files of the exact same size, I can't remember concrete numbers but I believe many groups contained tens of thousands of files, some over 100.000. So that's why there is an additional partitioning on "byte signatures". I chose to extract these signatures from the middle of the file rather than the beginning or the end to account for file headers etc. The data storage solution was HDDs, in RAID 0 configuration. In the end I think it ran in about 30 minutes, but my memory is cloudy.

A couple of practical considerations: I believe third-party requirements are tqdm and filehash only. I'll also put a pyproject.toml for use with Poetry at the end.

The script:

#!/usr/bin/env python3

"""
Deduper

A Python script to find duplicate files and create hard links to dedupe these files.
It achieves this task in 6 steps:
1. List all files in the given directories
2. Partition the files into :class:`FileGroup`s,
   which group together multiple files with the same inode
3. Partition the file groups into :class:`FileBag`s based on the file size of the files
4. Further partition each file bag based on the first bytes of the files
5. Further partition each file bag based on the xxHash checksums of the files
6. (optional) Hard link each file in the equivalent file bags.

NOTE: Multiple devices are currently NOT supported
"""

# pylint: disable=line-too-long

import os
import errno
import filecmp
import sys

from time import time

from tqdm import tqdm

from filehash import FileHash

__docformat__ = 'reStructuredText'

# Amount of first bytes to read. Lower values will lead to more comparisons
# and more hashes to compute. Higher values will lead to more memory consumption
NUM_BYTES = 10

CHUNK_SIZE = 2 ** 12


def readable_bytes(num_bytes):
    """
    Turn a byte count into a human readable byte count, as a string

    :param int num_bytes: The byte count to convert
    :returns: A human-readable number of bytes
    :rtype: str
    """
    # https://stackoverflow.com/a/1094933

    for unit in ['B', 'K', 'M', 'G', 'T', 'P', 'E', 'Z']:
        if abs(num_bytes) < 1024.0:
            return ('{0:.1f}{1}').format(num_bytes, unit)
        num_bytes /= 1024.0
    return '{0:4.0f}Y'.format(num_bytes)


def truncate_or_pad_string(string, max_length):
    """
    Truncate a string in the middle, replacing the middle contents with a single character ellipsis,
    or pad it with whitespace at the end to ensure it fills exactly max_length characters.

    It is recommended to set max_length to an uneven number, so that we can account for the ellipsis
    character. Otherwise, max_length will be decreased by one.

    :param str string: The string to truncate
    :param int max_length: The maximum length of the string
    :returns: The formatted string
    """
    if len(string) <= max_length:
        return string.ljust(max_length)

    max_length = max_length - 1 if max_length % 1 else max_length
    slice_chars = (max_length - 1) // 2
    return '{0}…{1}'.format(string[:slice_chars], string[-slice_chars:])


class IOProgress:
    """
    A class to print progress of multithreaded I/O operations
    """

    def __init__(self, total_items, total_processors, task_description, task_unit, subtask_unit):
        # pylint: disable=too-many-arguments
        super().__init__()
        self.total_items = total_items
        self.total_processors = total_processors
        self.main_pbar = tqdm(total=total_items, desc=task_description,
                              unit=' {0}'.format(task_unit), ncols=150,
                              bar_format='{desc}: {percentage:3.0f}% |{bar}| '
                                         '{n_fmt}/{total_fmt} [Elapsed: {elapsed}s, '
                                         'ETA: {remaining}s, {rate_fmt}]')
        self.worker_pbars = [tqdm(unit=' {0}'.format(subtask_unit), ncols=100, unit_scale=True,
                                  position=i + 2,
                                  bar_format='#' + str(i) +
                                  ': {desc}{percentage:3.0f}% |{bar}| '
                                  '{n_fmt:>6}/{total_fmt:>6} '
                                  '[ETA {remaining:>5}s, {rate_fmt:>9}]', leave=False)
                             for i in range(total_processors)]

    def start_task(self, processor_no, task_name, task_size):
        """
        Notify the progress printer of a newly started task

        :param int processor_no: The index of the processor starting the task
        :param str task_name: The name of the task
        :param int task_size: The size of the task
        """
        pbar = self.worker_pbars[processor_no]
        pbar.n = 0
        pbar.total = task_size
        pbar.start_t = time()
        pbar.set_description(truncate_or_pad_string(task_name, 21))
        pbar.refresh()

    def chunk_processed(self, processor_no, chunk_size):
        """
        Notify the progress printer that a chunk of a certain size has finished processing

        :param int processor_no: The index of the processor starting the task
        :param int chunk_size: The size of the chunk that is finished
        """
        pbar = self.worker_pbars[processor_no]
        pbar.n += chunk_size
        pbar.refresh()

    def task_finished(self):
        """
        Notify the progress printer that a task has finished processing
        """
        self.main_pbar.n += 1
        self.main_pbar.refresh()

    def processor_finished(self, processor_no):
        """
        Notify the progress printer that a process has finished and its progress bar can be removed
        """
        # Remove current progress bar
        self.worker_pbars[processor_no].close()

    def cleanup(self):
        """
        Remove traces of progress bars
        """
        self.main_pbar.close()


class FileGroup:
    """
    A file group groups together multiple directory entries with the same inode.

    :ivar list[str] files: The absolute paths to the files in this file group
    :ivar int inode: The inode for the file group
    :ivar int size: The file size for the file group
    :ivar bytes _byte_signature: The byte signature of the files, lazily generated
    :ivar int _hash: A hash value for the files, lazily generated
    """
    def __init__(self, first_file, inode, size):
        self.files = [first_file]
        self.inode = inode
        self.size = size
        self._byte_signature = None
        self._hash = None

    def add_file(self, new_file):
        """
        Add a file to a file group

        :param str new_file: Absolute path to the new file
        """
        self.files.append(new_file)

    def get_byte_signature(self):
        """
        Get the byte signature for the files. For file groups of the
        same size, this signature will be taken from the same range of
        the files.

        :returns: A NUM_BYTES byte signature of the files
        :rtype bytes:
        """
        if not self._byte_signature:
            self._read_byte_signature()
        return self._byte_signature

    def get_hash(self):
        """
        Get the hash for the files

        :returns: The hash of the files
        :rtype int:
        """
        if not self._hash:
            self._compute_hash()
        return self._hash

    def _read_byte_signature(self):
        """
        Read the byte signature of the files
        """
        with open(self.files[0], 'rb') as handle:
            # Read bytes in the middle of the file
            # Reading from the beginning would be ineffective when there are headers involved
            # Reading from the end would be ineffective when there is padding involved
            handle.seek(self.size // 2)
            self._byte_signature = handle.read(NUM_BYTES)

    def _compute_hash(self):
        """
        Compute the hash of the files
        """
        self._hash = FileHash('sha1').hash_file(self.files[0])


class FileBag:
    """
    A file bag groups together similar/identical file groups

    :ivar list[FileGroup] file_groups: The file groups in the bag
    """
    _processor_no = 0

    def __init__(self):
        self.file_groups = []

    def add(self, file_group):
        """
        Add a file group
        """
        self.file_groups.append(file_group)

    def is_unique(self):
        """
        A file bag is unique if it has only one file group
        """
        return len(self.file_groups) == 1

    def partition_on_bytes(self, io_progress):
        """
        Partition the file bag based on the byte signatures of the file groups

        :returns: A list of new file bags, without unique file bags
        :rtype: list[FileBag]
        """
        old_file_bags = [self]
        new_file_bags = {}
        file_groups_discarded = 0
        for num_bytes in range(1, NUM_BYTES + 1):
            io_progress.start_task(self._processor_no,
                                   str.format('Bag {0} Pass {1}',
                                              readable_bytes(self.file_groups[0].size), num_bytes),
                                   len(self.file_groups))
            io_progress.chunk_processed(self._processor_no, file_groups_discarded)
            for file_bag in old_file_bags:
                if not file_bag.is_unique():
                    for file_group in file_bag.file_groups:
                        byte_sig = file_group.get_byte_signature()[:num_bytes]
                        if byte_sig not in new_file_bags:
                            new_file_bags[byte_sig] = FileBag()
                        new_file_bags[byte_sig].add(file_group)
                        io_progress.chunk_processed(self._processor_no, 1)
                else:
                    file_groups_discarded += 1
                    io_progress.chunk_processed(self._processor_no, 1)
            old_file_bags = list(new_file_bags.values())
            new_file_bags = {}

        io_progress.task_finished()
        return [fb for fb in old_file_bags if not fb.is_unique()]

    def partition_on_hash(self, io_progress):
        """
        Partition the file bag based on the hashes of the file groups

        :returns: A list of new file bags, without unique file bags
        :rtype: list[FileBag]
        """
        io_progress.start_task(self._processor_no,
                               'Bag {0}'.format(readable_bytes(self.file_groups[0].size)),
                               len(self.file_groups))
        new_file_bags = {}
        if not self.is_unique():
            for file_group in self.file_groups:
                hash_ = file_group.get_hash()
                if hash_ not in new_file_bags:
                    new_file_bags[hash_] = FileBag()
                new_file_bags[hash_].add(file_group)
                io_progress.chunk_processed(self._processor_no, 1)

        io_progress.task_finished()
        return [fb for fb in new_file_bags.values() if not fb.is_unique()]

    def link_groups(self):
        """
        Hard link all files from the file groups to each other
        Ask for user confirmation first, optionally perform a byte-by-byte check
        """
        for group in self.file_groups[1:]:
            target = self.file_groups[0].files[0]
            # if self.file_groups[0].size != 0:
            #    resp = input('Will link {0} to {1}, okay? [y/N] '.format(group.files[0], target))
            # else:
            #    print('Linking empty file {0} to {1}'.format(group.files[0], target))
            #    resp = 'y'
            # if resp.lower() == 'y':
            #    link_files(group.files, target)
            # elif assure_files_identical(target, group.files[0]):
            #    resp = input('Are you sure? Files are identical based on byte-by-byte comparison [y/N]')
            #    if resp.lower() == 'y':
            #        link_files(group.files, target)
            # else:
            #    print('Good call! Files are not identical.')
            if assure_files_identical(target, group.files[0]):
                print('Linking {0} to {1}'.format(group.files[0], target))
                link_files(group.files, target)


def assure_files_identical(a_file, another_file):
    """
    Perform byte-by-byte comparison of files
    """
    return filecmp.cmp(a_file, another_file)


def link_files(files_to_link, target):
    """
    Link files

    :param list[str] files_to_link: The files to link
    :param str target: The target to link to
    """
    for file_to_link in files_to_link:
        if not isinstance(file_to_link, str):
            print('Was supplied a bad path: {0}: {1}'.format(type(file_to_link), file_to_link))
        try:
            os.unlink(file_to_link)
        except OSError as ose:
            if ose.errno != errno.ENOENT:
                print('Failed to remove file: {0}'.format(ose))
                continue

        try:
            os.link(target, file_to_link)
        except OSError as ose:
            print('Failed to link: {0}'.format(ose))


def scan_directory(directory, file_groups):
    """
    Scan a directory
    """
    for entry in os.scandir(directory):
        if entry.is_file() and entry.stat().st_size != 0:
            inode = entry.inode()
            if inode in file_groups:
                file_groups[inode].add_file(entry.path)
            else:
                file_groups[inode] = FileGroup(entry.path, inode, entry.stat().st_size)
        elif entry.is_dir():
            scan_directory(entry.path, file_groups)


def dedupe(directories):
    """
    Dedupe the files in the given directories. Note that directories
    should have absolute paths. Directories may be an iterable
    """
    file_groups = {}
    num_file_groups = 0
    for directory in directories:
        print('Scanning {0}'.format(directory))
        scan_directory(directory, file_groups)

    num_file_groups = len(file_groups)
    print(str.format('Found {0} unique file groups, totalling {1} files',
                     num_file_groups, sum(len(fg.files) for fg in file_groups.values())))
    file_bags = {}
    for group in tqdm(file_groups.values(), desc='Partitioning based on size', unit=' groups',
                      ncols=150, bar_format='{desc}: {percentage:3.0f}% |{bar}| '
                                            '{n_fmt}/{total_fmt} [Elapsed: {elapsed}s, '
                                            'ETA: {remaining}s, {rate_fmt}]'):
        if group.size not in file_bags:
            file_bags[group.size] = FileBag()
        file_bags[group.size].add(group)
    # Already remove unique bags
    file_bags = [bag for bag in file_bags.values() if not bag.is_unique()]
    old_num_file_groups = num_file_groups
    num_file_groups = sum(len(b.file_groups) for b in file_bags)
    print(str.format('Partitioned into {0} bags and {1} file groups. {2} eliminated',
                     len(file_bags), num_file_groups, old_num_file_groups - num_file_groups))

    signature_progress = IOProgress(len(file_bags), 1, 'Partitioning based on byte signature',
                                    'bags', 'groups')
    new_file_bags = []
    for file_bag in file_bags:
        new_file_bags.extend(file_bag.partition_on_bytes(signature_progress))
    signature_progress.processor_finished(0)
    signature_progress.cleanup()

    old_num_file_groups = num_file_groups
    num_file_groups = sum(len(b.file_groups) for b in new_file_bags)
    print(str.format('\rPartitioned into {0} bags and {1} file groups. {2} eliminated',
                     len(new_file_bags), num_file_groups, old_num_file_groups - num_file_groups))

    file_bags = new_file_bags
    new_file_bags = []
    hash_progress = IOProgress(len(file_bags), 1, 'Partitioning based on hash',
                               'bags', 'groups')
    for file_bag in file_bags:
        new_file_bags.extend(file_bag.partition_on_hash(hash_progress))
    hash_progress.processor_finished(0)
    hash_progress.cleanup()

    old_num_file_groups = num_file_groups
    num_file_groups = sum(len(b.file_groups) for b in new_file_bags)
    print(str.format('\rPartitioned into {0} bags and {1} file groups. {2} eliminated',
                     len(new_file_bags), num_file_groups, old_num_file_groups - num_file_groups))

    for bag in new_file_bags:
        bag.link_groups()


if __name__ == '__main__':
    dedupe(map(os.path.abspath, sys.argv[1:]))

pyproject.toml:

[tool.poetry]
name = "deduper"
version = "0.1.0"
description = ""
authors = ["ROpdebee"]

[tool.poetry.dependencies]
python = "^3.7"


[build-system]
requires = ["poetry>=0.12"]
build-backend = "poetry.masonry.api"

Have fun!

Answer 1

Just reviewing the FileGroup class.

1. Review

1. The name FileGroup does not convey the intended purpose of the class (namely, that all the files in the group share an inode). A name like InodeFiles would be better.

2. The intention of the class is that all the files share an inode. But the class itself does not check this: it relies on the caller to do so. It is more reliable when a data structure checks its own integrity.

3. Adding a file to a FileGroup needs to be done differently depending on whether it is the first file in the group (in which case you pass its path, inode and size to the constructor), or a subsequent file in the group, in which case you call the add_file method, passing the path only. This causes difficulties at the point where the class is used: you have to check if inode in file_groups: before deciding which approach to use. It would be simpler if adding files to the collection were done in the same way for each file.

4. If you look at the only place where FileGroup is used, it is initialized based on an os.DirEntry object. So it would be simpler for it to collect os.DirEntry objects instead of paths.

5. If you implemented suggestions §3 and §4, then there would not need to be a difference between the first file and subsequent files, and in dedupe you could use collections.defaultdict:

   file_groups = collections.defaultdict(FileGroup)
   

   and in scan_directory you would write:

   if entry.is_file() and entry.stat().st_size != 0:
       file_groups[inode].add(entry)
   

   with no need for the else clause.

6. The expression files[0] appears in many places. I would suggest using a property with a docstring to clarify the intention:

   @property
   def representative(self):
       """A representative file from the collection.""" 
       return self.files[0]
   

7. The method get_hash takes no arguments and always returns the same result. This suggests that it should be a property:

   @property
   def hash(self):
       # etc.
   

8. get_hash uses a cache to avoid recomputing the result:

   def get_hash(self):
       if not self._hash:
           self._compute_hash()
       return self._hash
   

   The cache logic can be simplified using the @cached_property decorator:

   @cached_property
   def hash(self):
       return self._compute_hash()
   

   The @cached_property decorator was new in Python 3.8. In earlier versions you can get a similar effect by composing @property with @functools.lru_cache:

   def cached_property(fn):
       return property(functools.lru_cache(maxsize=None)(fn))
   

   (This implementation is not quite the same as the one in Python 3.8, in that all instances of the class share a single cache, which might not be appropriate for some use cases, but I think it will be fine in your particular case.)

9. Having made a cached property, there's no point having a separate _compute_hash method.

10. Similar remarks apply to the get_byte_signature, which can become the cached property signature, avoiding the need for a separate _read_byte_signature method.

11. The _read_byte_signature method reads and hashes NUM_BYTES bytes starting at self.size // 2. But if self.size is less than 2 * NUM_BYTES, the read will be short. In these cases I would prioritize reading more bytes, so I suggest something like:

    handle.seek(min(self.size // 2, max(self.size - NUM_BYTES, 0)))
    

12. Using the filehash package is overkill since you could implement what you need in a handful of lines using the built-in hashlib module:

    import hashlib
    
    def filehash(algorithm, fileobj):
        """Return hash object corresponding to the contents of a file hashed
        using the specified algorithm.
    
        """
        hashobj = hashlib.new(algorithm)
        while True:
            chunk = fileobj.read(4096)
            if not chunk:
                return hashobj
            hashobj.update(chunk)
    

    (It's always tempting to use third-party packages for the convenience, even if it only replaces a few lines of code, but extra dependencies come with maintenance burdens that should be offset against the benefits.)

13. It is inconsistent that get_hash returns a string (the hexdigest of the hash) while get_byte_signature returns a bytes object. It would make more sense to use the digest instead of the hexdigest, so that both results are bytes objects. This would allow them to be treated the same by other parts of the program.

2. Revised code

(You'll have to make the docstrings conform to your preferred style, but that should be straightforward.)

class InodeFiles:
    """Collection of os.DirEntry objects for files that share an inode."""
    def __init__(self):
        self.files = []
        self.inode = None

    def add(self, entry):
        """Add file (an os.DirEntry object) to collection."""
        if self.inode is None:
            self.inode = entry.inode()
        elif self.inode != entry.inode():
            raise ValueError("Inode does not match.")
        self.files.append(entry)

    @property
    def representative(self):
        """A representative file from the collection, as an os.DirEntry object."""
        return self.files[0]

    @property
    def size(self):
        """Size of the files (in bytes)."""
        return self.representative.stat().st_size

    @cached_property
    def hash(self):
        """SHA1 hash of the files, as a bytes object."""
        with open(self.representative, 'rb') as f:
            return filehash('sha1', f).digest()

    @cached_property
    def signature(self):
        """Signature of the files, as a bytes object."""
        with open(self.representative, 'rb') as f:
            f.seek(min(self.size // 2, max(self.size - NUM_BYTES, 0)))
            return f.read(NUM_BYTES)