Finding duplicate files on a disk using MD5

Question

I started learning Python a month ago (used to write programs in Delphi before). Can you please take a look at my code and give me some tips what is good and what is bad in it?

The aim is to find duplicate files on a disk and saves this information. I'm using md5 hash. Is it right to use global variable like I did?

import os
import time
import hashlib

def md5_sum(path, block_size=256*128, hr=False):
    md5 = hashlib.md5()
    with open(path, 'rb') as f:
        for chunk in iter(lambda: f.read(block_size), b''):
            md5.update(chunk)

    if hr:
        return md5.hexdigest()
    return md5.digest()

def md5_simple(path, chunk):
    if chunk:
        return md5_sum(path, 8192)
    else:
        return hashlib.md5(open(path, 'rb').read()).digest()

def walking(path, chunk=False):
    for dirpath, dirnames, filenames in os.walk(path):
        for filename in filenames:
            global dupl_string
            try:
                if len(found_files) % 500 == 0:
                    print(len(found_files))

                fname = os.path.join(dirpath, filename)
                file_size = os.path.getsize(fname)
                found_files[fname] = file_size

                if chunk:
                    file_hash = md5_sum(fname, 8192)
                else:
                    file_hash = md5_simple(fname, file_size > 1000 * 1024 * 1024)

                if file_hash in hash_files:
                    dupl_string += 'Src: ' + hash_files[file_hash] + ', dupl: ' + fname + '\n'
                else:
                    hash_files[file_hash] = fname

            except OSError:
                print('OSError: ' + fname)


my_dir = "C:"
dupl_string = ''
found_files = dict()
hash_files = dict()

time1 = time.time()

try:
    walking(my_dir)
finally:
    if dupl_string:
        with open('dupl.txt', 'w') as f3:
            f3.write(dupl_string)     

    with open('files.csv', 'w') as f:
        for key, value in sorted(found_files.items(), key=lambda item: item[1], reverse=True):
            f.write("%s, %s\n" % (key, value))

    print("Elapsed: " + str(time.time() - time1) + ' sec')

Answer 1

Formatting

The code formatting is good. The only major gripe I have is that there are some long lines (117 chars). Part of this is due to the high level of indenation, which we'll solve in another way, but even with heavily indented code, lines can be shorter. I use the black code formatter (with a linelength of 79) to format my code, so I don't have to worry about spacing, long lines or inconsistent quote characters any more

Try-except

You should surround the part that can fail (reading the filesize and file_hash) as closely as possible with the try-except clause. If you want to print a debug message and prevent the rest of that iteration to be executed, you can do that by adding a continue after the print

globals

as you question it, globals are not the correct way to pass information here. A different approach would be to have parts that

1. generate the paths to the files to check
2. return the hash
3. look for duplicates in the hashes
4. report the duplicates

Each part can receive the result of the previous step as input

generate the paths

Since python 3.4 there is a more convenient way to handle file paths: pathlib.Path

from pathlib import Path
def walking(path):
    return Path(path).glob("**/*")

This returns a generator that yields Paths to all of the files under path. Since it is so simple, it can be inlined.

hashing

def hashing(paths, chunk=False):
    for path in paths:
        filesize = path.stat().st_size
        if chunk:
            file_hash = md5_sum(fname, 8192)
        else:
            file_hash = md5_simple(fname, file_size > 1000 * 1024 * 1024)
        yield path, file_hash

magic numbers

This previous function has some magic numbers 8192 and 1000 * 1024 * 1024. Better would be to define those a module level constants:

CHUNKSIZE_DEFAULT = 8192
MAX_SIZE_BEFORE_CHUNK = 1000 * 1024 * 1024 # files larger than this need to be hashed in chuncks

def hashing(paths, chunk=False):
    for path in paths:
        filesize = path.stat().st_size
        if chunk or file_size > MAX_SIZE_BEFORE_CHUNK:
            file_hash = md5_sum(fname, CHUNKSIZE_DEFAULT )
        else:
            file_hash = md5_simple(fname)
        yield path, file_hash

I'm not 100% happy with MAX_SIZE_BEFORE_CHUNK as variable name, but can't immediately think of a better on

hoist the OI

If you look at this talk by Brandon Rhodes, it makes sense not to open the file in the method that calculates the hash, but have it accept a filehandle. This is also the approach taken in this code review answer So to reuse the slightly adapted code from this answer

def read_chunks(file_handle, chunk_size=CHUNKSIZE_DEFAULT):
    while True:
        data = file_handle.read(chunk_size)
        if not data:
            break
        yield data


def md5(file_handle, chunk_size=None, hex_representation=False):
    if chunk_size is None:
        hasher = hashlib.md5(file_handle.read())
    else:
        hasher = hashlib.md5()
        for chunk in read_chunks(file_handle, chunk_size):
            hasher.update(chunk)

    return hasher.digest() if not hex_representation else hasher.hexdigest()

This gets called like this:

def hashing(paths, chunk=False, hex_representation=False):
    for path in paths:
        file_size = path.stat().st_size
        with path.open("rb") as file_handle:
            chunk_size = (
                CHUNKSIZE_DEFAULT
                if chunk or file_size > MAX_SIZE_BEFORE_CHUNK
                else None
            )
            file_hash = md5(
                file_handle,
                chunk_size=chunk_size,
                hex_representation=hex_representation,
            )
        yield path, file_hash

looking for duplicates

Instead of keeping a long string with all the duplicates, you can keep a set of the Paths for each file_hash. A collections.defaultdict(set) is the suited container for this. You add each path to the set at the in the dict. At the end of the function, you filter the keys that have more than 1 entry:

def duplicates(hashings):
    duplicates = defaultdict(set)
    for path, file_hash in hashings:
        duplicates[file_hash].add(path)

    return {
        filehash: paths
        for filehash, paths in duplicates.items()
        if len(paths) > 1
    }

report the filesizes

instead of formating the csv-file yourself, you can use the csv module. It even has a writerows. Instead of defining the lambda function to srt by value, you can use operator.itemgetter

import csv
from operator import itemgetter


def report_results(file_handle, filesizes):
    writer = csv.writer(file_handle, delimiter=",", lineterminator="\n")
    sorted_sizes = sorted(filesizes.items(), key=itemgetter(1))
    writer.writerows(sorted_sizes)

This method can also be used to save the file_hashes

reporting the duplicates

def report_duplicates(file_handle, duplicates):
    writer = csv.writer(file_handle, delimiter=",", lineterminator="\n")
    for file_hash, duplicate_files in duplicates.items():
        for file_name in duplicate_files:
            writer.writerow((file_hash, str(file)))

Here I exported a csv file with the hash and filenames of the duplicates. If you want it in another format, you can easily change this method

bringing it together

def main(path, chunk=False):
    files = [file for file in Path(path).glob("**/*") if file.is_file()]

    filesizes = {str(path): path.stat().st_size for path in files}
    with open("test_filesizes.csv", "w") as file_handle:
        report_results(file_handle, filesizes)
    filehashes = dict(hashing(files, chunk=chunk))

    with open("test_hashes.csv", "w") as file_handle:
        report_results(file_handle, filehashes)

    duplicates = find_duplicates(filehashes.items())
    with open("test_duplicates.csv", "w") as file_handle:
        report_duplicates(file_handle, duplicates)

And then we put everything behind a __main__ guard:

if __name__ == "__main__":
    main(<path>)