Python 查找大文件：用 os.walk 和 heapq 列出前 N 名

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

"""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())

python3 largest_files.py --help
python3 largest_files.py sample --top 2 --min-mib 2
printf 'exit=%s\n' "$?"

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

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

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"

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