Find duplicate files using Python

Question

Most Python "duplicate file finder" scripts I found do a brute-force of calculating the hashes of all files under a directory. So, I wrote my own -- hopefully faster -- script to kind of do things more intelligently.

Basically, it first searches for files of exact same size, then it compares only N bytes at the head and tail of the files, and finally it compares the files' full hashes.

My #1 concern of course would be correctness, followed closely by maintainability.

__author__ = 'pepoluan'

import os
import hashlib
from glob import glob
from itertools import chain


# Global variables
class G:
    OutFormat = 'list'  # Possible values: 'list', 'csv'
    OutFile = None
    StartPaths = [
        'D:\\DATA_2',
        'D:\\DATA_1',
        'D:\\'
    ]
    PartialCheckSize = 8192
    FullFileHash = True

    MinSize = 16 * 1024 * 1024
    ProgPeriod = 1000
    FullBlockSize = 1024 * 1024
    Quiet = False

    HashFunc = hashlib.md5


def get_walker_generator(at_path):
    return (
        chain.from_iterable(
            glob(
                os.path.join(
                    x[0].replace('[', '[[]').replace(']', '[]]'),
                    '*.*'
                )
            ) for x in os.walk(at_path)
        )
    )


def dict_filter_by_len(rawdict, minlen=2):
    assert isinstance(rawdict, dict)
    return {k: v for k, v in rawdict.items() if len(v) >= minlen}


def qprint(*args, **kwargs):
    if not G.Quiet:
        print(*args, **kwargs)


def get_dupes_by_size(path_list):
    qprint('===== Recursively stat-ing {0}'.format(path_list))
    processed = set()
    size_dict = {}
    for statpath in path_list:
        c = 0
        uniq_in_path = 0
        qprint('{0}...'.format(statpath), end='')
        for fname in get_walker_generator(statpath):
            try:
                if c >= G.ProgPeriod:
                    print('.', end='', flush=True)
                    c = 0
                if fname not in processed:
                    c += 1
                    uniq_in_path += 1
                    fstat = os.stat(fname)
                    fsize = fstat.st_size
                    flist = size_dict.get(fsize, set())
                    flist.add(fname)
                    size_dict[fsize] = flist
                    processed.add(fname)
            except:
                print('\nException on ', fname)
                raise
        qprint(uniq_in_path)
    qprint('\nTotal files: ', len(processed))
    dupe_sizes = {(None, sz): list(fset) for sz, fset in size_dict.items() if sz >= G.MinSize and len(fset) > 1}
    qprint('Dupes: ', len(dupe_sizes))
    return dupe_sizes


def refine_dupes_by_partial_hash(dupes_dict, partial_check_size=G.PartialCheckSize, hashfunc=G.HashFunc):
    assert isinstance(dupes_dict, dict)
    qprint('===== Checking hash of first and last {0} bytes ====='.format(partial_check_size))
    qprint('Processing...', end='', flush=True)
    size_and_hashes = {}
    for selector, flist in dupes_dict.items():
        fsize = selector[-1]
        for fname in flist:
            with open(fname, 'rb') as fin:
                hash_front = hashfunc(fin.read(partial_check_size)).hexdigest()
                seek_targ = fsize - G.PartialCheckSize - 1
                if seek_targ > 0:
                    fin.seek(seek_targ)
                    hash_rear = hashfunc(fin.read(partial_check_size)).hexdigest()
                else:
                    hash_rear = hash_front
            # "size" at rear, so a simple print will still result in a nicely-aligned table
            selector = (hash_front, hash_rear, fsize)
            flist = size_and_hashes.get(selector, [])
            flist.append(fname)
            size_and_hashes[selector] = flist
            qprint('.', end='', flush=True)
    dupe_exact = dict_filter_by_len(size_and_hashes)
    qprint('\nDupes: ', len(dupe_exact))
    return dupe_exact


def refine_dupes_by_full_hash(dupes_dict, block_size=G.FullBlockSize, hashfunc=G.HashFunc):
    assert isinstance(dupes_dict, dict)
    qprint('===== Checking full hashes of Dupes')
    qprint('Processing...', end='', flush=True)
    fullhashes = {}
    for selector, flist in dupes_dict.items():
        sz = selector[-1]  # Save size so we can still inform the user of the size
        for fname in flist:
            hasher = hashfunc()
            with open(fname, 'rb') as fin:
                while True:
                    buf = fin.read(block_size)
                    if not buf: break
                    hasher.update(buf)
            # "size" at rear, so a simple print will still result in a nicely-aligned table
            slct = (hasher.hexdigest(), sz)
            flist = fullhashes.get(slct, [])
            flist.append(fname)
            fullhashes[slct] = flist
            qprint('.', end='', flush=True)
    dupe_exact = dict_filter_by_len(fullhashes)
    qprint('\nDupes: ', len(dupe_exact))
    return dupe_exact


def output_results(dupes_dict, out_format=G.OutFormat, out_file=G.OutFile):
    assert isinstance(dupes_dict, dict)
    kiys = [k for k in dupes_dict]
    kiys.sort(key=lambda x: x[-1])

    if out_file is not None:
        qprint('Writing result in "{0}" format to file: {1} ...'.format(out_format, out_file), end='')
    else:
        qprint()

    if out_format == 'list':
        for kiy in kiys:
            flist = dupes_dict[kiy]
            print('-- {0}:'.format(kiy), file=out_file)
            flist.sort()
            for fname in flist:
                print('   {0}'.format(fname), file=out_file)
    elif out_format == 'csv':
        print('"Ord","Selector","FullPath"', file=out_file)
        order = 1
        for kiy in kiys:
            flist = dupes_dict[kiy]
            flist.sort()
            for fname in flist:
                print('"{0}","{1}","{2}"'.format(order, kiy, fname), file=out_file)
                order += 1

    if out_file is not None:
        qprint('done.')


def _main():
    dupes = get_dupes_by_size(G.StartPaths)
    dupes = refine_dupes_by_partial_hash(dupes)
    if G.FullFileHash:
        dupes = refine_dupes_by_full_hash(dupes)
    output_results(dupes, out_format=G.OutFormat, out_file=G.OutFile)


if __name__ == '__main__':
    _main()

Answer 1

• Don't Repeat Yourself

  The refine_dupes_by_partial_hash is almost identical to refine_dupes_by_full_hash; the only difference is what to be done with a file. Factor the difference out into a hashing_strategy callable:

  def refine_dupes(dupes_dict, hashing_strategy, block_size, hashfunc):           
      hashes = {}
      for selector, flist in dupes_dict.items():                          
          fsize = selector[-1]
          for name in flist:
              hasher = hashfunc()                                         
              with open(fname, 'rb') as fin:                              
                  result = hashing_strategy(fin, strategy, blocksize, hasher)
                  flist = fullhashes.get(result, [])                      
                  flist.append(fname)                                     
                  hashes[result] = flist                                  
      dupe_exact = dict_filter_by_len(hashes)                             
      return dupe_exact
  

  In fact, I'd go one step further and make hashing_strategy a class to incapsulate block_size and hashfunc.

• Use builtins where possible:

  for order, fname in enumerate(flist, 1):
  

  instead of manually incrementing order

• Use sys.argv instead of hardcoding paths.

Answer 2

To improve maintainability the Python style guide is key: PEP0008

In particular, your global variables should all be uppercase to signify that they're constants. PascalCase is for objects and makes it look like that's what these variables are.

Also you don't need class G. Python's scoping means that if you just declare those variables outside of any class or function they'll be global for this file. And if you import this file as a module you can just reference them with module_name.CONSTANT.

Lastly docstrings are a must for maintainability. You don't explain what your functions do at all, making it harder for others to use or change them. Python has another PEP about docstrings that gives a lot of info about how to make good ones.