Find Large Files with Python: Rank a Directory with os.walk and heapq
Logs, downloads, and build artifacts keep growing, and you want to know which individual files are largest without reading their contents. This tutorial builds a read-only Python CLI that lists the top N regular files, filters by minimum size, skips symbolic links inside the directory, and reports an incomplete scan as a failure.
It targets beginner-to-intermediate readers who can run terminal commands. Validate a small fixture first, then inspect your logs or Hugo output directory. If writes are already failing, use the Linux disk space and inode troubleshooting guide to identify the affected filesystem first.
1. Method: inspect metadata and keep the top N
- Traverse with
os.walk(), setonerrorto record directory listing failures, and usefollowlinks=Falseto avoid descending into directory symlinks. - Read each file entry with
lstat(), which inspects the link itself. Accept only regular files withstat.S_ISREG(). File symlinks, broken links, and FIFOs do not enter the ranking. - Maintain at most N candidates with
heapq. Once full, replace the smallest item only when a larger record arrives, then sort the final candidates in descending order.
The metric is logical size, st_size, rather than reclaimable disk space. One MiB is 1,048,576 bytes; the report also includes exact byte counts. Equal sizes use descending relative-path string order, making the selection of the top N deterministic.
2. Prepare the environment and sample data
Use Linux, Python 3.9 or newer, and bash. No third-party packages are required. The examples were tested on Linux with Python 3.11.4; use an existing interpreter without installing pip packages.
- Create a separate demo directory in one bash session. Choose another name when repeating the exercise to avoid overwriting your files:
python3 --version
mkdir largest-files-demo
cd largest-files-demo
python3 - <<'PY_DATA'
from pathlib import Path
root = Path("sample")
(root / "nested").mkdir(parents=True)
for name, mib in [("small.bin", 1), ("nested/big.bin", 3), ("space name.bin", 2)]:
with (root / name).open("wb") as stream:
for _ in range(mib):
stream.write(b"x" * (1024 * 1024))
(root / "big-link").symlink_to("nested/big.bin")
(root / "dir-link").symlink_to("nested", target_is_directory=True)
(root / "broken-link").symlink_to("missing.bin")
PY_DATA
- Save the next section as
largest_files.pyin the current directory, or download the complete script. The fixture contains three regular files totaling 6 MiB and three symlinks for checking exclusions.
3. Complete implementation: rank logical sizes
"""List the largest regular files without reading their contents. Python 3.9+."""
import argparse
import heapq
import json
import os
from pathlib import Path
import stat
import sys
def main(argv=None):
parser = argparse.ArgumentParser(description=__doc__, allow_abbrev=False)
parser.add_argument("directory", type=Path)
parser.add_argument("--top", type=int, default=10, help="results to keep, 1-1000")
parser.add_argument("--min-mib", type=int, default=0, help="minimum size in whole MiB")
args = parser.parse_args(argv)
if not 1 <= args.top <= 1000 or args.min_mib < 0:
parser.error("top must be 1-1000 and min-mib must be nonnegative")
try:
root = args.directory.resolve(strict=True)
if not stat.S_ISDIR(root.stat().st_mode):
parser.error("directory must refer to a directory")
except (OSError, RuntimeError) as exc:
parser.error(str(exc))
errors = scanned = matched = 0
largest = []
minimum = args.min_mib * 1024 * 1024
def report_error(exc):
nonlocal errors
errors += 1
print(f"ERROR: {ascii(str(exc))}", file=sys.stderr)
for folder, _dirs, names in os.walk(root, onerror=report_error, followlinks=False):
for name in names:
path = Path(folder) / name
try:
info = path.lstat()
except OSError as exc:
report_error(exc)
continue
if not stat.S_ISREG(info.st_mode):
continue
scanned += 1
if info.st_size < minimum:
continue
matched += 1
item = (info.st_size, str(path.relative_to(root)))
if len(largest) < args.top:
heapq.heappush(largest, item)
elif item > largest[0]:
heapq.heapreplace(largest, item)
print("MiB\tbytes\tpath_json")
for size, relative in sorted(largest, reverse=True):
print(f"{size / (1024 * 1024):.2f}\t{size}\t{json.dumps(relative)}")
print(
f"scanned={scanned} matched={matched} shown={len(largest)} errors={errors}",
file=sys.stderr,
)
return 1 if errors else 0
if __name__ == "__main__":
raise SystemExit(main())
The starting directory is converted to its real path with resolve(strict=True), so an explicitly supplied starting path may itself be a directory symlink. Links inside that directory are not followed. The script reads directory entries and metadata without opening file contents, deleting, or renaming anything.
scanned counts regular-file entries whose metadata was obtained successfully, matched counts entries meeting the size threshold, and shown counts output rows. Errors go to stderr and the report goes to stdout. Paths are JSON-encoded strings, so tabs and newlines in filenames cannot break the one-row-per-line TSV format.
Exit codes are 0 when no scan errors were recorded, 1 when scan errors mean the report may be incomplete, and 2 for invalid arguments or starting directories. No matching files is a successful result containing only the header.
4. Run: list the two largest files of at least 2 MiB
python3 largest_files.py --help
python3 largest_files.py sample --top 2 --min-mib 2
printf 'exit=%s\n' "$?"
Expected content follows. The summary comes from stderr and may appear in a different order relative to stdout in the terminal:
MiB bytes path_json
3.00 3145728 "nested/big.bin"
2.00 2097152 "space name.bin"
scanned=3 matched=2 shown=2 errors=0
exit=0
big-link does not add another 3 MiB, dir-link does not cause nested to be scanned twice, and the broken link is not treated as a read failure. --min-mib accepts nonnegative integers, and --top is limited to 1–1000.
For your own project, replace sample with its directory and quote paths containing spaces. In a Hugo repository, run python3 largest_files.py public --top 20 --min-mib 1 from its root, provided the script is also saved there.
5. Validate: assert the ranking, then test permission errors
- Save the report and summary separately, then check sizes, order, filtering, and symlink exclusions:
python3 largest_files.py sample --top 2 --min-mib 2 > report.tsv 2> scan.log
python3 - <<'PY_CHECK'
import csv
import json
from pathlib import Path
with Path("report.tsv").open(newline="", encoding="utf-8") as stream:
rows = list(csv.DictReader(stream, delimiter="\t", quoting=csv.QUOTE_NONE))
assert [int(row["bytes"]) for row in rows] == [3145728, 2097152]
assert [json.loads(row["path_json"]) for row in rows] == [
"nested/big.bin", "space name.bin"
]
assert Path("scan.log").read_text().strip() == "scanned=3 matched=2 shown=2 errors=0"
print("PASS: ranking, threshold, and symlink exclusions")
PY_CHECK
Expect PASS: ranking, threshold, and symlink exclusions. If you only get a header, check the threshold: --min-mib 100 matches nothing in this fixture.
- As an ordinary user, test a child directory that cannot be listed and restore its permissions afterward:
mkdir sample/locked
chmod 000 sample/locked
if python3 largest_files.py sample > partial.tsv 2> partial.log; then
scan_status=0
else
scan_status=$?
fi
chmod 700 sample/locked
cat partial.log
printf 'exit=%s\n' "$scan_status"
An ordinary user should see Permission denied, errors=1, and exit=1; accessible files are still listed. Root or an account with permission-bypass capabilities may still read the directory, so this will not reproduce the failure there. Do not switch to root for this test. If interrupted, run chmod 700 sample/locked to restore access.
- In automation, check the exit code and store reports outside the scanned directory:
if python3 largest_files.py sample --top 20 --min-mib 2 > latest.tsv 2> latest.log; then
printf 'Complete report: latest.tsv\n'
else
scan_status=$?
cat latest.log >&2
printf 'Incomplete or invalid scan, exit=%s\n' "$scan_status" >&2
fi
The reports are in the parent of sample, preventing them from entering their own ranking. On failure, latest.tsv provides partial clues rather than a complete inventory; a production task should alert or return a nonzero status in the failure branch. For scheduled interpreter and working-directory issues, see Python PATH and working-directory troubleshooting in cron.
6. Troubleshooting and limits
| Observation | Cause and action |
|---|---|
| Empty report with exit code 0 | Lower --min-mib, inspect scanned and matched, and confirm the starting directory. |
Permission denied |
Check the runtime account and access to the specific directory. Keep the error log and treat the report as partial. |
No such file or directory |
A file may have moved or disappeared during traversal. Pause the relevant writer and retry. |
Ranking differs from du |
This script uses logical sizes; sparse files, hard links, and directory storage use different accounting. |
| Logs dominate every report | Confirm the retention policy, then configure size-based rotation and backup counts for future logs. |
The os.stat_result documentation distinguishes st_size from allocated st_blocks: a sparse file’s logical size can greatly exceed its allocated storage. This script counts paths, so multiple hard links may appear separately. Summing the ranking does not measure unique data.
Real mounted subdirectories are also traversed; there is no filesystem-boundary filter equivalent to du -x. Choose a specific directory rather than scanning / as a routine task.
The candidate heap holds only --top records, but os.walk() still stores directory entries and traversal state, so total memory is not constant. Very wide or deep trees add overhead. This is a sequence of observations rather than a snapshot, and ordinary path traversal is not a security boundary when links or directories are replaced concurrently. Use it on directories you manage with a stable structure.
7. Summary
List large files within a specific directory, then combine the results with filesystem statistics before deciding what to clean up. This script reads metadata, keeps a fixed number of ranking candidates, and exposes scan errors through stderr and exit codes. Store reports outside the scan and check completeness before choosing log rotation or archival actions.
- 原文作者:春江暮客
- 原文链接:https://www.bobobk.com/en/python-largest-files-heapq.html
- 版权声明:本作品采用 知识共享署名-非商业性使用-禁止演绎 4.0 国际许可协议 进行许可,非商业转载请注明出处(作者,原文链接),商业转载请联系作者获得授权。
相关文章
- Python Too Many Open Files: Find File Handle Leaks on Linux
- Stream gzip Logs with Python and Detect Truncation and CRC Errors
- Fix Delayed Python Logs with -u, flush, and Bash pipefail
- Troubleshoot Linux Scheduled Jobs: Python Paths, Working Directories, and Environment
- Rotate Python Logs by Size with RotatingFileHandler