findDuplicateFiles

#!/usr/bin/env python3
#
# findDuplicateFiles: "Scan directory subtrees for duplicate files."
# 2014-09-12: Written by Steven J. DeRose.
#
import sys
import os
import re
import hashlib

from alogging import ALogger
lg = ALogger()

__metadata__ = {
    "title"        : "findDuplicateFiles",
    "description"  : "Scan directory subtrees for duplicate files.",
    "rightsHolder" : "Steven J. DeRose",
    "creator"      : "http://viaf.org/viaf/50334488",
    "type"         : "http://purl.org/dc/dcmitype/Software",
    "language"     : "Python 3.7",
    "created"      : "2016-02-06",
    "modified"     : "2021-04-03",
    "publisher"    : "http://github.com/sderose",
    "license"      : "https://creativecommons.org/licenses/by-sa/3.0/"
}
__version__ = __metadata__["modified"]

descr = """
=Usage=

findDuplicateFiles [options] [dirs]

Search all of the directories lists (recursively), for any filenames that
occur more than once. The filenames can be matched exactly, or in various
"fuzzy"/loose ways. Whitespace in names is always normalized and stripped.
However, files that may be the same but have entirely different names,
are not detected.


=Related Commands=

renameFiles, diffDirs, find.


=Known bugs and limitations=

No special Unicode normalizations except whitespace.


=History=

* 2014-09-12: Written by Steven J. DeRose.
* 2019-10-29: Lint.
* 2020-08-11: New layout.


=To do=

* Add options to handle copy/backup, hidden, etc. --exclude/--include.


=Rights=

Copyright 2014-09-12 by Steven J. DeRose. This work is licensed under a
Creative Commons Attribution-Share-alike 3.0 unported license.
For further information on this license, see
[https://creativecommons.org/licenses/by-sa/3.0].

For the most recent version, see [http://www.derose.net/steve/utilities]
or [https://github.com/sderose].

=Options=
    """

###############################################################################
#
extensionMap = {
    r"html": "htm",
    r"jpeg": "jpg",
    r"text": "txt",
}

dateExpr = r"(\d\d\d\d[-.]\d\d[-.]\d\d|\d\d/\d\d/\d\d(\d\d)?)"
timeExpr = r"\d\d[-:]\d\d[-:]\d\d(\.\d+)(Z|[ESMP][SD]T]|GMT|[-+]\d+)?"
dateTimeExpr = dateExpr + r"[- T@:.]" + timeExpr
cdte = re.compile(dateTimeExpr)

def normalizeFilename(
    name:str,
    case:bool       = True,
    space:bool      = True,
    special:bool    = True,
    dates:bool      = True,
    serials:bool    = True,
    extensions:bool = True,
    ) -> str:
    """Normalize a filename so we don't worry about details.
    cf diffDirs.py.
    """
    if (case):
        name = name.lower()
    if (space):
        name = re.sub(r"\s+", " ", name, re.UNICODE).strip()
    if (special):
        name = re.sub(r"^((backup|copy|alias)( \d+)? of )+", "", name, re.I)
        name = re.sub(r" (backup|copy|alias)( \d+)?$", "", name, re.I)
        name = re.sub(r".(bak|bck)$", "", name, re.I)
        name = name.strip("#~")
    if (dates):
        name = re.sub(cdte, "", name).strip()
    if (serials):
        name = re.sub(r"\s*\d+(\.\w+)$", r"\\1", name)
    if (extensions):
        for pair in extensionMap:
            if (name.endswith("."+pair[0])):
                oldLen = len(pair[0])
                name = name[0:-oldLen] + pair[1]
                break
    return(name)

def loadDirInfo(dirName:str, fileDict:dict=None, recursive:bool=True) -> dict:
    """Load info on all the files from the given directory, into a dict keyed by path,
    with checksums as values.
    """
    if (fileDict is None): fileDict = {}
    if (not os.path.isdir(dirName)):
        lg.error("Not a directory: %s." % (dirName))
        return(fileDict)

    for f in (os.listdir(dirName)):
        if (f[0]=="." and not args.hidden): continue
        path = os.path.join(dirName,f)
        if (os.path.isdir(path)):
            if (recursive): fileDict.update(loadDirInfo(path, fileDict))
        else:
            csum = getHash(path) if (args.checksum) else "?"
            fileDict[path] = (
                os.path.getmtime(path), os.path.getsize(path), csum)
    return(fileDict)

def getHash(path):
    hasher = hashlib.sha256()
    with open(path, "rb") as ifh:
        for x in ifh.read(1024):
            hasher.update(x)
    csum = hasher.hexdigest()
    return csum


###############################################################################
# Main
#
if __name__ == "__main__":
    import argparse

    def processOptions():
        try:
            from BlockFormatter import BlockFormatter
            parser = argparse.ArgumentParser(
                description=descr, formatter_class=BlockFormatter)
        except ImportError:
            parser = argparse.ArgumentParser(description=descr)

        parser.add_argument(
            "--checksum", action="store_true",
            help="Calculate a checksum and compare.")
        parser.add_argument(
            "--color", action="store_true",
            help="Colorize the output.")
        parser.add_argument(
            "--nocolor", action="store_false", dest="color",
            help="Turn off colorizing.")
        parser.add_argument(
            "--hidden", action="store_true",
            help='Include files/dirs whose names start with ".".')
        parser.add_argument(
            "--iencoding", "--input-encoding", type=str, metavar="E",
            help="Assume this character set for input files.")
        parser.add_argument(
            "--ignoreCase", "--ignore-case", "-i", action="store_true",
            help="Disregard case distinctions.")
        parser.add_argument(
            "--minDups", type=int, default=2,
            help="Only report sets with more than this many duplicates.")
        parser.add_argument(
            "--normalize", action="store_true",
            help="Ignore copy 4 of, Backup of backup of, ~, #, .bak. etc.")
        parser.add_argument(
            "--ndates", action="store_true",
            help="Ignore date and time fields in names.")
        parser.add_argument(
            "--nextensions", action="store_true",
            help="Ignore differences like .jpeg/.jpg, .html/.htm.")
        parser.add_argument(
            "--nserials", action="store_true",
            help="Ignore trailing numbers before an extension.")
        parser.add_argument(
            "--oencoding", "--output-encoding", type=str, metavar="E",
            help="Assume this character set for output files.")
        parser.add_argument(
            "--quiet", "-q", action="store_true",
            help="Suppress most messages.")
        parser.add_argument(
            "--unicode", action="store_const", dest="iencoding",
            const="utf8", help="Assume utf-8 for input files.")
        parser.add_argument(
            "--verbose", "-v", action="count", default=0,
            help="Add more messages (repeatable).")
        parser.add_argument(
            "--version", action="version", version="Version of "+__version__,
            help="Display version information, then exit.")

        parser.add_argument(
            "files", type=str,
            nargs=argparse.REMAINDER,
            help="Path(s) to input file(s)")

        args0 = parser.parse_args()
        if (args0.color is None):
            args0.color = ("CLI_COLOR" in os.environ and sys.stderr.isatty())
        lg.setColors(args0.color)
        lg.setVerbose(args0.verbose)
        return args0

    ###########################################################################
    #
    args = processOptions()

    totalRecords = 0
    totalFiles = 0

    fileLists = {}  # path -> checksum
    fileCounts = []

    for dnum, diry in enumerate(args.files):
        fileDict0 = loadDirInfo(args.files[dnum], diry)
        fileLists.update(fileDict0)
        fileCounts.append(len(fileDict0))

    lg.vMsg(0, "Done loading. File counts by directory tree:")
    for dnum in (range(0, len(args.files))):
        lg.vMsg(0, "    %8d %s" % (fileCounts[dnum], args.files[dnum]))


    union = {}
    ndups = 0
    for dnum in (range(len(args.files))):
        for k, y in fileLists[dnum].items():
            base = os.path.basename(k)
            base = normalizeFilename(
                base,
                case       = args.ignoreCase,
                space      = True,
                special    = args.normalize,
                dates      = args.ndates,
                serials    = args.nserials,
                extensions = args.nextensions
            )

            if (base not in union):
                union[base] = [ y ]
            else:
                ndups += 1
                base[union].append(y)

    lg.vMsg(0, "Done scanning. Total duplicates: %d." % (ndups))

    for k in union.keys():
        dupSet = union[k]
        if (len(dupSet) > args.minDups):
            print("%s\n    %s\n" % (k, "\n    ".join(dupSet)))

    if (not args.quiet):
        lg.vMsg(0,"Done.")

    sys.exit(0)