1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
|
# SPDX-FileCopyrightText: 2026 Dennis Fink <me+coding@dennisfink.me>
#
# SPDX-License-Identifier: BSD-3-Clause
"""Hash candidate files and group paths with identical content."""
import concurrent.futures
import hashlib
import sys
from collections import defaultdict
from collections.abc import Callable
from functools import partial
from pathlib import Path
from typing import Any
import xxhash
from tqdm import tqdm
from duplicate_finder.cli import debug, is_debug_enabled, is_quiet_enabled
from .types import FileInfo, FilesByCandidateGroup, FilesByHash, FileSet
def compute_hash(path: Path, digest: str | Callable[[], Any]) -> str:
"""Compute a content hash using the supplied digest.
The file is opened in binary mode and streamed through
:func:`hashlib.file_digest`. For extendable-output algorithms whose
``hexdigest`` method requires an explicit output length, a 32-byte digest
is requested.
:param path: Path to the file whose contents should be hashed.
:param digest: Hash algorithm name or callable that creates a hash object.
:returns: The hexadecimal digest of the file contents.
"""
with path.open("rb") as f:
digested = hashlib.file_digest(f, digest)
try:
return digested.hexdigest()
except TypeError:
return digested.hexdigest(length=32)
def select_representative_files(
files_by_candidate_group: FilesByCandidateGroup, *, include_hardlinks: bool
) -> dict[FileInfo, FileSet]:
"""Select one file to hash for each filesystem identity.
Candidate groups have already been reduced to groups that can contain
duplicates. Each filesystem identity contributes one representative file.
When hard links are included, that representative stands for every scanned
path belonging to the identity; otherwise it represents only itself.
:param files_by_candidate_group: Files grouped into duplicate-candidate sets
and then by filesystem identity.
:param include_hardlinks: Whether different paths referring to the same inode
should be treated as duplicate files.
:returns: Representative files mapped to the paths they represent.
"""
representative_files: dict[FileInfo, FileSet] = {}
for files_by_identity in files_by_candidate_group.values():
for files in files_by_identity.values():
representative = next(iter(files))
representative_files[representative] = (
files if include_hardlinks else {representative}
)
return representative_files
def select_hash_function(name: str) -> Callable[[Path], str]:
"""Return the file-hashing callable for a named digest algorithm.
:param name: Name of an xxHash variant or an algorithm available through
:mod:`hashlib`.
:returns: A callable that accepts a file path and returns its hexadecimal digest.
:raises KeyError: If ``name`` is not a supported hash algorithm.
"""
hash_funcs = {
"xxh32": partial(compute_hash, digest=xxhash.xxh32),
"xxh64": partial(compute_hash, digest=xxhash.xxh64),
"xxh128": partial(compute_hash, digest=xxhash.xxh128),
}
for digest in hashlib.algorithms_available:
hash_funcs[digest] = partial(compute_hash, digest=digest)
return hash_funcs[name]
def group_files_by_hash(
files_by_candidate_group: FilesByCandidateGroup,
*,
include_hardlinks: bool = False,
hash_name: str = "xxh128",
jobs: int = 0,
) -> FilesByHash:
"""Group duplicate candidates by content hash.
Only one representative of each filesystem identity is hashed. Hash
computation is parallelized using :class:`ThreadPoolExecutor`, and hash
groups containing only one represented path are discarded before returning.
:param files_by_candidate_group: Files grouped into duplicate-candidate sets
and then by filesystem identity. Candidate sets are normally based on
file size and extension, or file size alone when extension grouping is
disabled.
:param include_hardlinks: Whether multiple paths referring to the same inode
should be included as separate duplicate paths.
:param hash_name: Name of the hash algorithm to use. This may be an xxHash
variant or any algorithm exposed by :mod:`hashlib`.
:param jobs: Maximum number of worker threads to use for hash computation.
If set to ``0``, the default chosen by :class:`ThreadPoolExecutor` is used.
:returns: A mapping from digest strings to sets of files sharing the same
content hash. Singleton hash groups are omitted.
"""
representative_files = select_representative_files(
files_by_candidate_group, include_hardlinks=include_hardlinks
)
if is_debug_enabled():
represented_paths = sum(len(files) for files in representative_files.values())
debug(
"Hash candidate selection:",
f"{len(representative_files)} representative files for "
f"{represented_paths} paths.",
err=True,
)
if not representative_files:
return {}
hash_function = select_hash_function(hash_name)
max_workers = None if jobs == 0 else jobs
hash_groups: defaultdict[str, FileSet] = defaultdict(set)
with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
future_to_file = {
executor.submit(hash_function, file.path): file
for file in representative_files
}
for future in tqdm(
concurrent.futures.as_completed(future_to_file),
total=len(future_to_file),
desc="Hashing files",
unit="file",
dynamic_ncols=True,
disable=is_quiet_enabled() or not sys.stderr.isatty(),
):
file = future_to_file[future]
try:
hashsum = future.result()
except FileNotFoundError:
debug(f"Skipping disappeared file: {file.path.absolute()}", err=True)
continue
except PermissionError:
debug(f"Skipping inaccessible file: {file.path.absolute()}", err=True)
continue
hash_groups[hashsum].update(representative_files[file])
duplicate_groups = {
hashsum: files for hashsum, files in hash_groups.items() if len(files) > 1
}
debug(
"Hashing complete:",
f"{len(representative_files)} files produced "
f"{len(duplicate_groups)} duplicate hash groups.",
err=True,
)
return duplicate_groups
|
