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
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
|
# SPDX-FileCopyrightText: 2026 Dennis Fink <me+coding@dennisfink.me>
#
# SPDX-License-Identifier: BSD-3-Clause
"""Discover files and group duplicate candidates by metadata and identity."""
import sys
from collections import defaultdict
from collections.abc import Iterator
from pathlib import Path
from tqdm import tqdm
from .cli import debug, error, is_debug_enabled, is_quiet_enabled
from .types import FileIdentity, FileInfo, FilesByCandidateGroup
def should_follow(path: Path, follow_symlinks: bool = False) -> bool:
"""Return whether a filesystem entry should be followed.
Broken symbolic links are skipped and reported when symbolic-link following
is enabled.
:param path: Filesystem path to inspect.
:param follow_symlinks: Whether symbolic links should be followed.
:returns: ``True`` if the entry should be processed, otherwise ``False``.
"""
if not path.is_symlink():
return True
if not follow_symlinks:
return False
if not path.exists():
error(f"{path.absolute()} is a broken symbolic link. Skipping.")
return False
return True
def already_visited(path: Path, visited_paths: set[FileIdentity]) -> bool:
"""Return whether a filesystem path has already been visited.
The path is identified by its device and inode so symbolic links to an
already visited directory are recognized as the same filesystem object.
:param path: Filesystem path to inspect.
:param visited_paths: Filesystem identities that have already been visited.
:returns: ``True`` if the path was already visited, otherwise ``False``.
"""
stat = path.stat()
identity = (stat.st_dev, stat.st_ino)
if identity in visited_paths:
return True
visited_paths.add(identity)
return False
def iter_files_recursively(
paths: Iterator[Path],
*,
follow_symlinks: bool = False,
visited_paths: set[FileIdentity],
) -> Iterator[FileInfo]:
"""Yield regular files and their metadata below the supplied paths.
Metadata required by later stages is collected when each regular file is
discovered and carried forward with the path so candidate grouping and
hashing do not need to inspect the file again.
:param paths: File and directory paths to inspect.
:param follow_symlinks: Whether symbolic links to files and directories should
be followed.
:param visited_paths: Filesystem identities of directories already traversed.
:yields: Regular files discovered while scanning the supplied paths, together
with the metadata needed by later stages.
"""
for path in paths:
try:
if not should_follow(path, follow_symlinks):
continue
if path.is_dir():
if follow_symlinks and already_visited(path, visited_paths):
debug(
f"Skipping already visited directory: {path.absolute()}",
err=True,
)
continue
try:
yield from iter_files_recursively(
path.iterdir(),
follow_symlinks=follow_symlinks,
visited_paths=visited_paths,
)
except PermissionError:
debug(
f"Skipping inaccessible directory: {path.absolute()}", err=True
)
continue
elif path.is_file():
path_stat = path.stat()
yield FileInfo(
path=path,
size=path_stat.st_size,
identity=(path_stat.st_dev, path_stat.st_ino),
)
else:
error(f"{path} is not a file or directory. Skipping!")
except FileNotFoundError:
debug(f"Skipping disappeared path: {path.absolute()}", err=True)
continue
except PermissionError:
debug(f"Skipping inaccessible path: {path.absolute()}", err=True)
continue
def iter_files_from_paths(
paths: Iterator[Path], *, follow_symlinks: bool = False
) -> Iterator[FileInfo]:
"""Yield regular files and their metadata below the supplied paths.
Directories are traversed recursively in depth-first order. Permission
errors are ignored so scanning can continue with the remaining paths. When
symbolic links are followed, directories already visited through another
path are skipped to avoid recursion cycles.
:param paths: File and directory paths to inspect.
:param follow_symlinks: Whether symbolic links to files and directories
should be followed.
:yields: Regular files discovered while scanning the supplied paths, together
with the metadata needed by later stages.
"""
visited_paths: set[FileIdentity] = set()
yield from iter_files_recursively(
paths, follow_symlinks=follow_symlinks, visited_paths=visited_paths
)
def group_file_candidates(
files: Iterator[FileInfo],
*,
minsize: int | None = None,
maxsize: int | None = None,
include_empty: bool = True,
include_file_extension: bool = True,
include_hardlinks: bool = False,
) -> FilesByCandidateGroup:
"""Group files into candidate sets for duplicate detection.
Files are grouped by size and, by default, case-insensitive file extension.
Within each candidate group, files are grouped by filesystem identity so
hard links can share a single hash computation. Groups that cannot contain
duplicates are removed before hashing.
:param files: Files and metadata collected during discovery.
:param minsize: Minimum file size in bytes, or ``None`` for no lower limit.
:param maxsize: Maximum file size in bytes, or ``None`` for no upper limit.
:param include_empty: Whether zero-byte files should be included.
:param include_file_extension: Whether file extensions should be included
in the candidate-group key in addition to file size.
:param include_hardlinks: Whether different paths referring to the same inode
should count as separate duplicate candidates.
:returns: Files grouped into duplicate-candidate sets and then by filesystem
identity. Every returned candidate set can contain at least one duplicate.
"""
files_by_candidate_group = defaultdict(lambda: defaultdict(set))
accepted = 0
filtered_by_size = 0
for file in tqdm(
files,
desc="Scanning files",
unit="file",
dynamic_ncols=True,
disable=is_quiet_enabled() or not sys.stderr.isatty(),
):
if (
(not include_empty and file.size == 0)
or (minsize is not None and file.size < minsize)
or (maxsize is not None and file.size > maxsize)
):
filtered_by_size += 1
continue
group_key = (
(file.size, file.path.suffix.casefold())
if include_file_extension
else file.size
)
files_by_candidate_group[group_key][file.identity].add(file)
accepted += 1
candidate_groups: FilesByCandidateGroup = {}
candidate_paths = 0
for group_key, files_by_identity in files_by_candidate_group.items():
candidate_count = (
sum(len(files) for files in files_by_identity.values())
if include_hardlinks
else len(files_by_identity)
)
if candidate_count < 2:
continue
candidate_groups[group_key] = files_by_identity
candidate_paths += sum(len(files) for files in files_by_identity.values())
debug(
"File scan summary:",
f"{accepted} accepted, {filtered_by_size} filtered by size; "
f"{candidate_paths} paths remain in {len(candidate_groups)} candidate groups.",
err=True,
)
if is_debug_enabled():
hardlinked_identity_count = 0
hardlinked_path_count = 0
for files_by_identity in candidate_groups.values():
for files in files_by_identity.values():
if len(files) < 2:
continue
hardlinked_identity_count += 1
hardlinked_path_count += len(files)
debug(
"Hardlink summary:",
f"{hardlinked_path_count} paths across "
f"{hardlinked_identity_count} multiply referenced identities "
"in candidate groups.",
err=True,
)
return candidate_groups
|
