Storae`optimizer
📝 Python
Only for people who have serious balls. A combination of rule-based logic and AI to do a precision sweep if you really can't find big directories anymore.
Python
#!/usr/bin/env python3
# retoor <retoor@molodetz.nl>
from __future__ import annotations
import argparse
import asyncio
import getpass
import hashlib
import heapq
import itertools
import json
import logging
import os
import shutil
import statistics
import sys
import threading
import time
import urllib.error
import urllib.request
from collections import Counter
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path
from typing import Any
DEFAULT_GATEWAY_ENDPOINT = "https://devplace.net/openai/v1/chat/completions"
DEFAULT_GATEWAY_MODEL = "molodetz"
DEFAULT_REQUEST_TIMEOUT_SECONDS = 180.0
DEFAULT_TEMPERATURE = 0.4
DEFAULT_MAX_OUTPUT_TOKENS = 16000
DEFAULT_DIRECTORY_TIMEOUT_SECONDS = 45.0
DEFAULT_MAX_DIRECTORIES_PER_TOP_LEVEL = 200_000
DEFAULT_MAX_WORKERS = min(16, max(4, (os.cpu_count() or 4) * 2))
DEFAULT_TOP_DIRECTORIES_SHOWN = 20
DEFAULT_TOP_FILES_SHOWN = 25
DEFAULT_TOP_EXTENSIONS_SHOWN = 15
DEFAULT_LARGE_FILE_THRESHOLD_MB = 100.0
DEFAULT_STALE_DAYS = 180
DEFAULT_MIN_DUPLICATE_SIZE_MB = 1.0
DEFAULT_MAX_DUPLICATE_GROUPS_HASHED = 300
DEFAULT_RECLAIMABLE_RECENT_ACTIVITY_DAYS = 98
TOP_DIRECTORY_CANDIDATES_PER_TOP_LEVEL = 100
LARGE_FILE_HEAP_MINIMUM_CAPACITY = 500
LARGE_FILE_HEAP_CAPACITY_MULTIPLIER = 20
DUPLICATE_GROUP_MAX_MEMBERS_TRACKED = 64
DUPLICATE_HASH_CHUNK_BYTES = 1024 * 1024
DUPLICATE_GROUP_PATHS_SHOWN_IN_PAYLOAD = 8
RECLAIMABLE_SUBTREE_ENTRY_CAP = 300_000
RECLAIMABLE_CANDIDATES_SHOWN = 20
STALE_FILES_SHOWN = 20
INACCESSIBLE_EXAMPLES_TRACKED = 30
INACCESSIBLE_EXAMPLES_SHOWN_IN_PAYLOAD = 15
PROTECTED_DATABASE_FILES_SHOWN = 15
BYTE_UNITS = ("B", "KiB", "MiB", "GiB", "TiB", "PiB")
HARD_EXCLUDED_DIRECTORY_NAMES = frozenset({"lost+found"})
HARD_EXCLUDED_ABSOLUTE_PATHS = frozenset({
Path("/proc"), Path("/sys"), Path("/dev"), Path("/run"), Path("/var/run"),
})
DATABASE_FILE_EXTENSIONS = frozenset({
".db", ".db3", ".sqlite", ".sqlite3", ".sqlitedb", ".s3db", ".sl3",
".db-wal", ".db-shm", ".db-journal", ".sqlite-wal", ".sqlite-shm", ".sqlite-journal",
".mdb", ".accdb", ".fdb", ".gdb", ".ndf", ".myd", ".myi", ".ibd", ".frm",
".rdb", ".dbf", ".mdf", ".ldf", ".dsk", ".kdbx", ".realm", ".cdb",
})
RECLAIMABLE_DIRECTORY_NAMES: dict[str, str] = {
"__pycache__": "python bytecode cache",
".pytest_cache": "pytest cache",
".mypy_cache": "mypy cache",
".ruff_cache": "ruff cache",
".tox": "python tox environments",
"venv": "python virtual environment",
".venv": "python virtual environment",
"virtualenv": "python virtual environment",
"site-packages": "python installed packages",
"node_modules": "node.js dependency tree",
".pnpm-store": "pnpm package store",
".npm": "npm cache",
".yarn": "yarn cache",
"bower_components": "bower dependency tree",
"target": "build output (cargo/maven/gradle)",
"build": "build output",
"dist": "build output",
"out": "build output",
"cmake-build-debug": "build output",
"cmake-build-release": "build output",
".gradle": "gradle cache",
".m2": "maven cache",
".cargo": "cargo registry cache",
".rustup": "rust toolchains",
".nuget": "nuget cache",
".gem": "ruby gem cache",
"vendor": "vendored dependencies",
".terraform": "terraform provider cache",
"coverage": "test coverage artifacts",
"htmlcov": "test coverage artifacts",
".cache": "generic application cache",
"cache": "generic application cache",
"cache2": "browser cache",
"gpucache": "browser gpu cache",
".docker": "docker data",
".vscode-server": "vs code remote server data",
"deriveddata": "xcode derived data",
"trash": "trash / recycle bin",
".local-share-trash": "trash / recycle bin",
}
VIRTUAL_FILESYSTEM_TYPES = frozenset({
"proc", "sysfs", "devtmpfs", "devpts", "tmpfs", "cgroup", "cgroup2",
"pstore", "bpf", "tracefs", "debugfs", "securityfs", "mqueue",
"hugetlbfs", "fusectl", "configfs", "autofs", "binfmt_misc", "efivarfs",
"overlay", "squashfs", "ramfs", "rpc_pipefs",
})
LOGGER = logging.getLogger("storageanalysis")
class AnalysisError(Exception):
pass
@dataclass(frozen=True)
class ScanConfig:
root: Path
cross_filesystems: bool
directory_timeout_seconds: float
max_directories_per_top_level: int
max_workers: int
min_duplicate_size_bytes: int
max_duplicate_groups_hashed: int
large_file_threshold_bytes: int
stale_days: int
top_directories_shown: int
top_files_shown: int
top_extensions_shown: int
reclaimable_recent_activity_days: int
extra_excluded_names: frozenset[str]
verbose: bool
@dataclass(frozen=True)
class GatewaySettings:
api_key: str | None
model: str
endpoint: str
temperature: float
max_output_tokens: int
request_timeout_seconds: float
@dataclass(frozen=True)
class AnalysisConfig:
scan_config: ScanConfig
gateway_settings: GatewaySettings
output_path: Path | None
use_llm: bool
verbose: bool
@dataclass
class DirFrame:
path: Path
depth: int
apparent_bytes: int = 0
disk_bytes: int = 0
file_count: int = 0
dir_count: int = 1
inaccessible_count: int = 0
last_modified: float = 0.0
@dataclass(frozen=True)
class ReclaimableCandidate:
path: Path
category: str
apparent_bytes: int
disk_bytes: int
file_count: int
last_modified: float
@dataclass(frozen=True)
class LargeFileRecord:
path: Path
size_bytes: int
mtime: float
atime: float
is_database: bool
@dataclass(frozen=True)
class DuplicateGroup:
size_bytes: int
sha256: str
paths: tuple[Path, ...]
@property
def reclaimable_bytes(self) -> int:
return self.size_bytes * (len(self.paths) - 1)
@dataclass
class TopLevelScanResult:
name: str
path: Path
apparent_bytes: int
disk_bytes: int
file_count: int
dir_count: int
inaccessible_count: int
timed_out: bool
capped: bool
elapsed_seconds: float
top_directories: list[DirFrame] = field(default_factory=list)
@dataclass(frozen=True)
class FilesystemFact:
mount_point: str
filesystem_type: str
total_bytes: int
used_bytes: int
free_bytes: int
@property
def used_percent(self) -> float:
return (self.used_bytes / self.total_bytes * 100) if self.total_bytes else 0.0
@dataclass
class AnalysisFacts:
generated_at: datetime
root_label: str
cross_filesystems: bool
running_as_user: str
running_as_root: bool
total_apparent_bytes: int
total_disk_bytes: int
total_files: int
total_directories: int
inaccessible_count: int
inaccessible_examples: list[tuple[Path, str]]
symlinks_skipped: int
boundary_skipped_mounts: list[Path]
timed_out_entries: list[str]
capped_entries: list[str]
hardlink_bytes_deduplicated: int
hardlink_files_deduplicated: int
median_top_level_disk_bytes: float
mean_top_level_disk_bytes: float
top_directories: list[DirFrame]
top_files: list[LargeFileRecord]
top_extensions: list[tuple[str, int, int]]
reclaimable_recent_activity_days: int
reclaimable_by_category: list[tuple[str, int, int, int]]
top_reclaimable_candidates: list[ReclaimableCandidate]
reclaimable_total_disk_bytes: int
reclaimable_recent_total_disk_bytes: int
reclaimable_recent_location_count: int
reclaimable_recent_examples: list[ReclaimableCandidate]
duplicate_groups: list[DuplicateGroup]
duplicate_total_reclaimable_bytes: int
duplicate_groups_skipped: int
duplicate_files_hashed: int
min_duplicate_size_bytes: int
stale_large_files: list[LargeFileRecord]
stale_days: int
large_file_threshold_bytes: int
protected_database_files: list[LargeFileRecord]
filesystems: list[FilesystemFact]
root_filesystem: FilesystemFact | None
scan_coverage_percent: float | None
def load_dotenv_if_present(*candidate_paths: Path) -> None:
for candidate in candidate_paths:
if not candidate.is_file():
continue
LOGGER.debug(f"loading environment overrides from {candidate}")
try:
with candidate.open("r", encoding="utf-8", errors="replace") as handle:
for raw_line in handle:
line = raw_line.strip()
if not line or line.startswith("#") or "=" not in line:
continue
key, _, value = line.partition("=")
key = key.strip()
value = value.strip().strip('"').strip("'")
if key and key not in os.environ:
os.environ[key] = value
except OSError as error:
LOGGER.warning(f"could not read env file {candidate}: {error!r}")
def format_bytes(num_bytes: int) -> str:
value = float(max(0, num_bytes))
for unit in BYTE_UNITS:
if value < 1024.0:
return f"{value:.0f} {unit}" if unit == "B" else f"{value:.2f} {unit}"
value /= 1024.0
return f"{value:.2f} PiB"
def format_age_days(timestamp: float, now: datetime) -> str:
delta = now - datetime.fromtimestamp(timestamp)
days = delta.days
if days < 1:
return "today"
if days == 1:
return "1 day ago"
return f"{days} days ago"
def classify_extension(filename: str) -> str:
suffix = Path(filename).suffix.lower()
return suffix if suffix else "(no extension)"
def is_database_file(filename: str) -> bool:
return Path(filename).suffix.lower() in DATABASE_FILE_EXTENSIONS
def reclaimable_candidate_reason(
candidate: ReclaimableCandidate, recent_activity_days: int, now: datetime
) -> str:
if candidate.last_modified <= 0:
activity_desc = "no file activity observed inside it (empty or entirely unreadable)"
else:
activity_desc = f"most recent file inside it was modified {format_age_days(candidate.last_modified, now)}"
cutoff = now.timestamp() - recent_activity_days * 86400
if candidate.last_modified >= cutoff:
verdict = f"EXCLUDED from recommendations (within the {recent_activity_days}-day recent-activity window)"
else:
verdict = f"RECOMMENDED for cleanup review (outside the {recent_activity_days}-day recent-activity window)"
return (
f"directory name '{candidate.path.name}' exact-matched the known category '{candidate.category}'; "
f"{activity_desc}; verdict: {verdict}"
)
def stale_file_reason(record: LargeFileRecord, threshold_bytes: int, stale_days: int, now: datetime) -> str:
return (
f"size {format_bytes(record.size_bytes)} >= large-file-threshold {format_bytes(threshold_bytes)}; "
f"modified {format_age_days(record.mtime, now)} and accessed {format_age_days(record.atime, now)}, "
f"both older than the {stale_days}-day stale threshold; not a database file, so it is eligible to "
"be listed as an archival/cleanup candidate"
)
def duplicate_group_reason(group: DuplicateGroup) -> str:
return (
f"{len(group.paths)} file(s) share identical size {format_bytes(group.size_bytes)} AND an identical "
f"full-file SHA-256 hash ({group.sha256[:12]}...); this is a byte-for-byte confirmed duplicate, not "
f"just a same-size coincidence; reclaimable = (copies - 1) x size = {format_bytes(group.reclaimable_bytes)}"
)
def database_file_reason(record: LargeFileRecord) -> str:
extension = record.path.suffix.lower()
return (
f"extension '{extension}' is present in the protected database-extension list -> permanently "
"excluded from duplicate detection and stale-file candidacy, and never eligible for a deletion/move "
"recommendation regardless of size, age, or redundancy"
)
class ScanState:
def __init__(self, config: ScanConfig) -> None:
self.config = config
self.lock = threading.Lock()
self.seen_inodes: set[tuple[int, int]] = set()
self.hardlink_bytes_deduplicated = 0
self.hardlink_files_deduplicated = 0
self.extension_apparent_bytes: Counter[str] = Counter()
self.extension_file_counts: Counter[str] = Counter()
self._large_file_heap: list[tuple[int, int, LargeFileRecord]] = []
self._large_file_counter = itertools.count()
self.large_file_heap_capacity = max(
LARGE_FILE_HEAP_MINIMUM_CAPACITY,
config.top_files_shown * LARGE_FILE_HEAP_CAPACITY_MULTIPLIER,
)
self.duplicate_size_buckets: dict[int, list[Path]] = {}
self.duplicate_candidates_truncated = 0
self.reclaimable_candidates: list[ReclaimableCandidate] = []
self.reclaimable_subtrees_truncated: list[Path] = []
self.inaccessible_paths: list[tuple[Path, str]] = []
self.inaccessible_count = 0
self.boundary_skipped_mounts: set[Path] = set()
self.symlinks_skipped = 0
def record_inaccessible(self, path: Path, error: OSError) -> None:
with self.lock:
self.inaccessible_count += 1
if len(self.inaccessible_paths) < INACCESSIBLE_EXAMPLES_TRACKED:
self.inaccessible_paths.append((path, repr(error)))
LOGGER.debug(f"inaccessible path={path} reason={error!r}")
def record_symlink_skip(self) -> None:
with self.lock:
self.symlinks_skipped += 1
def record_boundary_skip(self, path: Path) -> None:
with self.lock:
self.boundary_skipped_mounts.add(path)
LOGGER.info(f"mount boundary skip: {path} (different filesystem, use --cross-filesystems to include)")
def record_reclaimable(self, candidate: ReclaimableCandidate) -> None:
with self.lock:
self.reclaimable_candidates.append(candidate)
LOGGER.info(
f"reclaimable candidate: path={candidate.path} category={candidate.category} "
f"disk-bytes={candidate.disk_bytes} files={candidate.file_count}"
)
def record_reclaimable_truncated(self, path: Path) -> None:
with self.lock:
self.reclaimable_subtrees_truncated.append(path)
LOGGER.warning(f"reclaimable subtree sizing truncated (deadline/cap reached): {path}")
def record_file(self, entry: os.DirEntry, frame: DirFrame) -> None:
try:
stat_result = entry.stat(follow_symlinks=False)
except OSError as error:
frame.inaccessible_count += 1
self.record_inaccessible(Path(entry.path), error)
return
size_bytes = stat_result.st_size
actual_disk_bytes = stat_result.st_blocks * 512
path = Path(entry.path)
counted = True
if stat_result.st_nlink > 1:
inode_key = (stat_result.st_dev, stat_result.st_ino)
with self.lock:
if inode_key in self.seen_inodes:
counted = False
self.hardlink_bytes_deduplicated += actual_disk_bytes
self.hardlink_files_deduplicated += 1
else:
self.seen_inodes.add(inode_key)
frame.file_count += 1
if counted:
frame.apparent_bytes += size_bytes
frame.disk_bytes += actual_disk_bytes
extension = classify_extension(entry.name)
with self.lock:
self.extension_apparent_bytes[extension] += size_bytes
self.extension_file_counts[extension] += 1
if counted:
is_database = is_database_file(entry.name)
self._consider_large_file(path, size_bytes, stat_result.st_mtime, stat_result.st_atime, is_database)
if size_bytes >= self.config.min_duplicate_size_bytes and not is_database:
self._record_duplicate_candidate(size_bytes, path)
def _consider_large_file(
self, path: Path, size_bytes: int, mtime: float, atime: float, is_database: bool
) -> None:
record = LargeFileRecord(path=path, size_bytes=size_bytes, mtime=mtime, atime=atime, is_database=is_database)
with self.lock:
heapq.heappush(self._large_file_heap, (size_bytes, next(self._large_file_counter), record))
if len(self._large_file_heap) > self.large_file_heap_capacity:
heapq.heappop(self._large_file_heap)
def _record_duplicate_candidate(self, size_bytes: int, path: Path) -> None:
with self.lock:
bucket = self.duplicate_size_buckets.setdefault(size_bytes, [])
if len(bucket) < DUPLICATE_GROUP_MAX_MEMBERS_TRACKED:
bucket.append(path)
else:
self.duplicate_candidates_truncated += 1
def largest_files(self, count: int) -> list[LargeFileRecord]:
with self.lock:
ordered = sorted(self._large_file_heap, key=lambda item: item[0], reverse=True)
return [record for _, _, record in ordered[:count]]
def stale_large_files(
self, threshold_bytes: int, stale_days: int, count: int, now: datetime
) -> list[LargeFileRecord]:
cutoff = now.timestamp() - stale_days * 86400
with self.lock:
ordered = sorted(self._large_file_heap, key=lambda item: item[0], reverse=True)
candidates = [
record for _, _, record in ordered
if record.size_bytes >= threshold_bytes and record.atime < cutoff and record.mtime < cutoff
and not record.is_database
]
return candidates[:count]
def database_files(self, count: int) -> list[LargeFileRecord]:
with self.lock:
ordered = sorted(self._large_file_heap, key=lambda item: item[0], reverse=True)
candidates = [record for _, _, record in ordered if record.is_database]
return candidates[:count]
def duplicate_group_candidates(self, max_groups: int) -> tuple[list[tuple[int, list[Path]]], int]:
with self.lock:
buckets = [(size, list(paths)) for size, paths in self.duplicate_size_buckets.items() if len(paths) >= 2]
buckets.sort(key=lambda item: item[0] * (len(item[1]) - 1), reverse=True)
selected = buckets[:max_groups]
skipped = len(buckets) - len(selected)
return selected, skipped
def is_hard_excluded(entry_path: Path, lowered_name: str, config: ScanConfig) -> bool:
if lowered_name in HARD_EXCLUDED_DIRECTORY_NAMES:
return True
if lowered_name in config.extra_excluded_names:
return True
return entry_path in HARD_EXCLUDED_ABSOLUTE_PATHS
def sum_subtree_fast(top: Path, config: ScanConfig, state: ScanState, deadline: float, boundary_dev: int) -> DirFrame:
frame = DirFrame(path=top, depth=-1)
stack: list[Path] = [top]
processed = 0
while stack:
if time.monotonic() > deadline or processed >= RECLAIMABLE_SUBTREE_ENTRY_CAP:
state.record_reclaimable_truncated(top)
break
current = stack.pop()
try:
with os.scandir(current) as scanner:
entries = list(scanner)
except OSError:
frame.inaccessible_count += 1
continue
for entry in entries:
processed += 1
try:
if entry.is_symlink():
continue
if entry.is_dir(follow_symlinks=False):
entry_path = Path(entry.path)
lowered = entry.name.lower()
if is_hard_excluded(entry_path, lowered, config):
continue
try:
entry_dev = entry.stat(follow_symlinks=False).st_dev
except OSError:
frame.inaccessible_count += 1
continue
if entry_dev != boundary_dev and not config.cross_filesystems:
continue
frame.dir_count += 1
stack.append(entry_path)
elif entry.is_file(follow_symlinks=False):
try:
stat_result = entry.stat(follow_symlinks=False)
except OSError:
frame.inaccessible_count += 1
continue
size_bytes = stat_result.st_size
actual_disk_bytes = stat_result.st_blocks * 512
counted = True
if stat_result.st_nlink > 1:
inode_key = (stat_result.st_dev, stat_result.st_ino)
with state.lock:
if inode_key in state.seen_inodes:
counted = False
else:
state.seen_inodes.add(inode_key)
frame.file_count += 1
frame.last_modified = max(frame.last_modified, stat_result.st_mtime)
if counted:
frame.apparent_bytes += size_bytes
frame.disk_bytes += actual_disk_bytes
except OSError:
frame.inaccessible_count += 1
return frame
def discover_top_level_directories(config: ScanConfig) -> list[Path]:
directories: list[Path] = []
try:
with os.scandir(config.root) as scanner:
entries = list(scanner)
except OSError as error:
raise AnalysisError(f"cannot list root directory {config.root}: {error!r}") from error
try:
root_dev = os.stat(config.root, follow_symlinks=False).st_dev
except OSError as error:
raise AnalysisError(f"cannot stat root directory {config.root}: {error!r}") from error
entries.sort(key=lambda entry: entry.name.lower())
for entry in entries:
try:
if entry.is_symlink():
continue
if not entry.is_dir(follow_symlinks=False):
continue
entry_path = Path(entry.path)
lowered = entry.name.lower()
if is_hard_excluded(entry_path, lowered, config):
LOGGER.debug(f"root-scan skip dir={entry.name} reason=hard-excluded")
continue
entry_dev = entry.stat(follow_symlinks=False).st_dev
if entry_dev != root_dev and not config.cross_filesystems:
LOGGER.info(f"mount boundary skip at root: {entry.path} (different filesystem, use --cross-filesystems to include)")
continue
except OSError as error:
LOGGER.debug(f"root-scan skip entry={entry.path} reason={error!r}")
continue
directories.append(entry_path)
LOGGER.info(f"root-scan candidate directory discovered: {entry.name}")
return directories
def scan_top_level_directory(top_path: Path, config: ScanConfig, state: ScanState) -> TopLevelScanResult:
name = top_path.name
started = time.monotonic()
deadline = started + config.directory_timeout_seconds
try:
top_dev = os.stat(top_path, follow_symlinks=False).st_dev
except OSError as error:
LOGGER.warning(f"[{name}] cannot stat top-level directory, skipping: {error!r}")
return TopLevelScanResult(
name=name, path=top_path, apparent_bytes=0, disk_bytes=0, file_count=0,
dir_count=0, inaccessible_count=0, timed_out=False, capped=False, elapsed_seconds=0.0,
)
frames: dict[Path, DirFrame] = {}
parent_of: dict[Path, Path] = {}
discovery_order: list[Path] = []
stack: list[tuple[Path, int]] = [(top_path, 0)]
timed_out = False
capped = False
while stack:
if time.monotonic() > deadline:
timed_out = True
LOGGER.warning(f"[{name}] directory deadline reached, reporting partial results")
break
if len(frames) >= config.max_directories_per_top_level:
capped = True
LOGGER.warning(f"[{name}] directory cap reached ({config.max_directories_per_top_level}), reporting partial results")
break
current, depth = stack.pop()
frame = DirFrame(path=current, depth=depth)
frames[current] = frame
discovery_order.append(current)
try:
with os.scandir(current) as scanner:
entries = list(scanner)
except OSError as error:
frame.inaccessible_count += 1
state.record_inaccessible(current, error)
continue
for entry in entries:
try:
if entry.is_symlink():
state.record_symlink_skip()
continue
if entry.is_dir(follow_symlinks=False):
entry_path = Path(entry.path)
lowered = entry.name.lower()
if is_hard_excluded(entry_path, lowered, config):
continue
try:
entry_dev = entry.stat(follow_symlinks=False).st_dev
except OSError as stat_error:
frame.inaccessible_count += 1
state.record_inaccessible(entry_path, stat_error)
continue
if entry_dev != top_dev and not config.cross_filesystems:
state.record_boundary_skip(entry_path)
continue
category = RECLAIMABLE_DIRECTORY_NAMES.get(lowered)
if category is not None:
aggregate = sum_subtree_fast(entry_path, config, state, deadline, entry_dev)
frame.apparent_bytes += aggregate.apparent_bytes
frame.disk_bytes += aggregate.disk_bytes
frame.file_count += aggregate.file_count
frame.dir_count += aggregate.dir_count
frame.inaccessible_count += aggregate.inaccessible_count
state.record_reclaimable(ReclaimableCandidate(
path=entry_path, category=category,
apparent_bytes=aggregate.apparent_bytes, disk_bytes=aggregate.disk_bytes,
file_count=aggregate.file_count, last_modified=aggregate.last_modified,
))
continue
parent_of[entry_path] = current
stack.append((entry_path, depth + 1))
elif entry.is_file(follow_symlinks=False):
state.record_file(entry, frame)
except OSError as error:
frame.inaccessible_count += 1
state.record_inaccessible(Path(entry.path), error)
# discovery_order records each directory before its children are pushed, so its reverse is
# a valid postorder: every child's totals are already folded in by the time its parent is visited.
for node in reversed(discovery_order):
parent = parent_of.get(node)
if parent is None:
continue
parent_frame = frames.get(parent)
if parent_frame is None:
continue
child_frame = frames[node]
parent_frame.apparent_bytes += child_frame.apparent_bytes
parent_frame.disk_bytes += child_frame.disk_bytes
parent_frame.file_count += child_frame.file_count
parent_frame.dir_count += child_frame.dir_count
parent_frame.inaccessible_count += child_frame.inaccessible_count
root_frame = frames.get(top_path)
elapsed = time.monotonic() - started
top_directory_candidates = heapq.nlargest(
TOP_DIRECTORY_CANDIDATES_PER_TOP_LEVEL, frames.values(), key=lambda candidate: candidate.disk_bytes
)
LOGGER.info(
f"[{name}] scan finished elapsed={elapsed:.2f}s "
f"disk-bytes={root_frame.disk_bytes if root_frame else 0} "
f"files={root_frame.file_count if root_frame else 0} "
f"timed-out={timed_out} capped={capped}"
)
return TopLevelScanResult(
name=name, path=top_path,
apparent_bytes=root_frame.apparent_bytes if root_frame else 0,
disk_bytes=root_frame.disk_bytes if root_frame else 0,
file_count=root_frame.file_count if root_frame else 0,
dir_count=root_frame.dir_count if root_frame else 0,
inaccessible_count=root_frame.inaccessible_count if root_frame else 0,
timed_out=timed_out, capped=capped, elapsed_seconds=elapsed,
top_directories=top_directory_candidates,
)
def scan_root_level_files(config: ScanConfig, state: ScanState) -> TopLevelScanResult:
started = time.monotonic()
frame = DirFrame(path=config.root, depth=0)
try:
with os.scandir(config.root) as scanner:
entries = list(scanner)
except OSError as error:
LOGGER.warning(f"cannot list root directory for loose files: {error!r}")
entries = []
for entry in entries:
try:
if entry.is_symlink():
state.record_symlink_skip()
continue
if entry.is_file(follow_symlinks=False):
state.record_file(entry, frame)
except OSError as error:
state.record_inaccessible(Path(entry.path), error)
elapsed = time.monotonic() - started
return TopLevelScanResult(
name="(loose files directly under scan root)", path=config.root,
apparent_bytes=frame.apparent_bytes, disk_bytes=frame.disk_bytes,
file_count=frame.file_count, dir_count=0, inaccessible_count=frame.inaccessible_count,
timed_out=False, capped=False, elapsed_seconds=elapsed,
)
def log_scan_criteria(config: ScanConfig) -> None:
LOGGER.info(f"scan root: {config.root}")
LOGGER.info(
"criteria: this tool is strictly read-only, it never deletes, moves, renames, or "
"modifies any file or directory; it only measures and reports"
)
LOGGER.info(
"criteria: every regular file's true on-disk footprint is measured via st_blocks*512 "
"(actual allocated blocks), not just st_size (apparent size), because sparse files and "
"filesystem block rounding make these differ; both totals are reported"
)
LOGGER.info(
"criteria: files with more than one hard link are counted exactly once, keyed by "
"(device, inode), across the entire scan; without this, hard-linked backups or "
"deduplicated trees would inflate totals with bytes that are not actually duplicated on disk"
)
LOGGER.info(
f"criteria: scanning stays within the filesystem of {config.root} by default (mount "
f"boundaries are respected via st_dev); pass --cross-filesystems to also descend into "
"separately mounted drives and network shares"
)
LOGGER.info(
f"criteria: {len(RECLAIMABLE_DIRECTORY_NAMES)} well-known generated/cache/vendor directory "
"names (node_modules, __pycache__, venv, build/dist/target, package manager caches, browser "
"caches, trash, ...) are sized as a single aggregate and not descended into file-by-file, "
"both for speed and because their individual files are rarely relevant to a cleanup decision"
)
LOGGER.info(
f"criteria: each top-level entry under the root gets its own {config.directory_timeout_seconds:.1f}s "
f"soft deadline and a {config.max_directories_per_top_level} directory cap, checked cooperatively "
f"so one huge subtree cannot stall the others; all top-level entries scan concurrently across up "
f"to {config.max_workers} worker threads"
)
LOGGER.info(
"criteria: duplicate files are found in two passes: files of identical size are grouped in "
"memory first (cheap), then the largest-potential-reclaim groups are fully hashed with SHA-256 "
"to confirm true byte-for-byte duplicates before being reported"
)
LOGGER.info(
f"criteria: a large file is one at or above {format_bytes(config.large_file_threshold_bytes)}; "
f"it is additionally flagged as stale when both its last-modified and last-accessed time are "
f"older than {config.stale_days} day(s)"
)
LOGGER.info(
"criteria: directories are never skipped for being hidden (dotfiles/dotdirs) since caches, "
"trash, and application data live there and are frequently the largest reclaim opportunity; "
"only a small set of pseudo-filesystems (/proc, /sys, /dev, /run) and lost+found are always skipped"
)
LOGGER.info(
f"criteria: files of any known database format ({len(DATABASE_FILE_EXTENSIONS)} extensions "
"covering SQLite, MySQL/MariaDB, Access, Firebird, Redis, DBF, ...) are never recommended for "
"deletion, never included in duplicate-file detection, and never listed as a stale-file cleanup "
"candidate; they are only ever reported as factual, protected context"
)
LOGGER.info(
f"criteria: a reclaimable-category location is only recommended for cleanup when it was not "
f"modified within the last {config.reclaimable_recent_activity_days} day(s) "
f"({config.reclaimable_recent_activity_days // 7} week(s)); locations touched more recently are "
"presumed to be in active use and reported separately, excluded from recommendations"
)
LOGGER.info(
"criteria: every location this run could not read due to insufficient permissions is listed "
"explicitly with advice to skip it; this run does not guess at the contents of unreadable paths"
)
async def run_scan(config: ScanConfig) -> tuple[list[TopLevelScanResult], ScanState]:
if not config.root.is_dir():
raise AnalysisError(f"root is not a directory: {config.root}")
top_level_dirs = discover_top_level_directories(config)
state = ScanState(config)
if not top_level_dirs:
LOGGER.warning(f"no accessible subdirectories found under {config.root}, only loose files will be measured")
LOGGER.info(f"scan starting: {len(top_level_dirs)} top-level entr(y/ies) under {config.root}, workers={config.max_workers}")
started = time.monotonic()
loop = asyncio.get_running_loop()
with ThreadPoolExecutor(max_workers=config.max_workers) as executor:
results = await asyncio.gather(*(
loop.run_in_executor(executor, scan_top_level_directory, top_dir, config, state)
for top_dir in top_level_dirs
))
LOGGER.info(f"scan finished in {time.monotonic() - started:.2f}s across {len(results)} top-level entr(y/ies)")
return list(results), state
def hash_file(path: Path) -> str | None:
hasher = hashlib.sha256()
try:
with path.open("rb") as handle:
while True:
chunk = handle.read(DUPLICATE_HASH_CHUNK_BYTES)
if not chunk:
break
hasher.update(chunk)
except OSError as error:
LOGGER.debug(f"duplicate-hash skip path={path} reason={error!r}")
return None
return hasher.hexdigest()
async def find_confirmed_duplicates(
state: ScanState, config: ScanConfig
) -> tuple[list[DuplicateGroup], int, int]:
buckets, skipped_groups = state.duplicate_group_candidates(config.max_duplicate_groups_hashed)
if not buckets:
return [], skipped_groups, 0
LOGGER.info(f"hashing {len(buckets)} same-size group(s) to confirm true duplicates ({skipped_groups} lower-priority group(s) skipped)")
loop = asyncio.get_running_loop()
confirmed: list[DuplicateGroup] = []
hashed_files = 0
with ThreadPoolExecutor(max_workers=config.max_workers) as executor:
for size_bytes, paths in buckets:
hashes = await asyncio.gather(*(loop.run_in_executor(executor, hash_file, path) for path in paths))
hashed_files += len(paths)
by_hash: dict[str, list[Path]] = {}
for path, digest in zip(paths, hashes):
if digest is None:
continue
by_hash.setdefault(digest, []).append(path)
for digest, group_paths in by_hash.items():
if len(group_paths) >= 2:
confirmed.append(DuplicateGroup(size_bytes=size_bytes, sha256=digest, paths=tuple(group_paths)))
confirmed.sort(key=lambda group: group.reclaimable_bytes, reverse=True)
LOGGER.info(f"duplicate confirmation finished: {len(confirmed)} confirmed group(s), {hashed_files} file(s) hashed")
return confirmed, skipped_groups, hashed_files
def discover_real_filesystems() -> list[FilesystemFact]:
mounts_path = Path("/proc/mounts")
facts: list[FilesystemFact] = []
if not mounts_path.is_file():
return facts
seen_mount_points: set[str] = set()
try:
text = mounts_path.read_text(encoding="utf-8", errors="replace")
except OSError as error:
LOGGER.warning(f"could not read /proc/mounts: {error!r}")
return facts
for line in text.splitlines():
fields = line.split()
if len(fields) < 3:
continue
mount_point, filesystem_type = fields[1], fields[2]
if filesystem_type in VIRTUAL_FILESYSTEM_TYPES:
continue
if mount_point in seen_mount_points:
continue
try:
usage = shutil.disk_usage(mount_point)
except OSError as error:
LOGGER.debug(f"filesystem overview: cannot stat mount {mount_point}: {error!r}")
continue
seen_mount_points.add(mount_point)
facts.append(FilesystemFact(
mount_point=mount_point, filesystem_type=filesystem_type,
total_bytes=usage.total, used_bytes=usage.used, free_bytes=usage.free,
))
facts.sort(key=lambda fact: fact.used_bytes, reverse=True)
return facts
def build_facts(
results: list[TopLevelScanResult],
state: ScanState,
config: ScanConfig,
duplicate_groups: list[DuplicateGroup],
duplicate_groups_skipped: int,
duplicate_files_hashed: int,
filesystems: list[FilesystemFact],
) -> AnalysisFacts:
generated_at = datetime.now()
total_apparent = sum(result.apparent_bytes for result in results)
total_disk = sum(result.disk_bytes for result in results)
total_files = sum(result.file_count for result in results)
total_dirs = sum(result.dir_count for result in results)
timed_out = sorted(result.name for result in results if result.timed_out)
capped = sorted(result.name for result in results if result.capped)
disk_bytes_by_result = [result.disk_bytes for result in results if result.dir_count > 0]
median_top_level = statistics.median(disk_bytes_by_result) if disk_bytes_by_result else 0.0
mean_top_level = statistics.mean(disk_bytes_by_result) if disk_bytes_by_result else 0.0
all_directory_frames: list[DirFrame] = []
for result in results:
all_directory_frames.extend(result.top_directories)
top_directories = heapq.nlargest(
config.top_directories_shown, all_directory_frames, key=lambda candidate: candidate.disk_bytes
)
top_files = state.largest_files(config.top_files_shown)
top_extensions = [
(extension, size, state.extension_file_counts[extension])
for extension, size in state.extension_apparent_bytes.most_common(config.top_extensions_shown)
]
recent_activity_cutoff = generated_at.timestamp() - config.reclaimable_recent_activity_days * 86400
stale_candidates = [c for c in state.reclaimable_candidates if c.last_modified < recent_activity_cutoff]
recent_candidates = [c for c in state.reclaimable_candidates if c.last_modified >= recent_activity_cutoff]
reclaimable_grouped: dict[str, list[ReclaimableCandidate]] = {}
for candidate in stale_candidates:
reclaimable_grouped.setdefault(candidate.category, []).append(candidate)
reclaimable_by_category = sorted(
(
(category, sum(c.disk_bytes for c in items), sum(c.file_count for c in items), len(items))
for category, items in reclaimable_grouped.items()
),
key=lambda item: item[1], reverse=True,
)
reclaimable_total = sum(item[1] for item in reclaimable_by_category)
top_reclaimable_candidates = sorted(
stale_candidates, key=lambda candidate: candidate.disk_bytes, reverse=True
)[:RECLAIMABLE_CANDIDATES_SHOWN]
reclaimable_recent_total = sum(c.disk_bytes for c in recent_candidates)
reclaimable_recent_examples = sorted(
recent_candidates, key=lambda candidate: candidate.disk_bytes, reverse=True
)[:RECLAIMABLE_CANDIDATES_SHOWN]
duplicate_total_reclaimable = sum(group.reclaimable_bytes for group in duplicate_groups)
stale_files = state.stale_large_files(
config.large_file_threshold_bytes, config.stale_days, STALE_FILES_SHOWN, generated_at
)
protected_database_files = state.database_files(PROTECTED_DATABASE_FILES_SHOWN)
root_filesystem: FilesystemFact | None = None
scan_coverage_percent: float | None = None
try:
usage = shutil.disk_usage(config.root)
root_filesystem = FilesystemFact(
mount_point=str(config.root), filesystem_type="(scan root filesystem)",
total_bytes=usage.total, used_bytes=usage.used, free_bytes=usage.free,
)
if usage.used > 0:
scan_coverage_percent = min(100.0, total_disk / usage.used * 100)
except OSError as error:
LOGGER.warning(f"could not read disk usage for scan root: {error!r}")
running_as_root = os.geteuid() == 0
try:
running_as_user = getpass.getuser()
except (OSError, KeyError):
running_as_user = f"uid={os.geteuid()}"
return AnalysisFacts(
generated_at=generated_at,
root_label=str(config.root),
cross_filesystems=config.cross_filesystems,
running_as_user=running_as_user,
running_as_root=running_as_root,
total_apparent_bytes=total_apparent,
total_disk_bytes=total_disk,
total_files=total_files,
total_directories=total_dirs,
inaccessible_count=state.inaccessible_count,
inaccessible_examples=list(state.inaccessible_paths),
symlinks_skipped=state.symlinks_skipped,
boundary_skipped_mounts=sorted(state.boundary_skipped_mounts),
timed_out_entries=timed_out,
capped_entries=capped,
hardlink_bytes_deduplicated=state.hardlink_bytes_deduplicated,
hardlink_files_deduplicated=state.hardlink_files_deduplicated,
median_top_level_disk_bytes=median_top_level,
mean_top_level_disk_bytes=mean_top_level,
top_directories=top_directories,
top_files=top_files,
top_extensions=top_extensions,
reclaimable_recent_activity_days=config.reclaimable_recent_activity_days,
reclaimable_by_category=reclaimable_by_category,
top_reclaimable_candidates=top_reclaimable_candidates,
reclaimable_total_disk_bytes=reclaimable_total,
reclaimable_recent_total_disk_bytes=reclaimable_recent_total,
reclaimable_recent_location_count=len(recent_candidates),
reclaimable_recent_examples=reclaimable_recent_examples,
duplicate_groups=duplicate_groups,
duplicate_total_reclaimable_bytes=duplicate_total_reclaimable,
duplicate_groups_skipped=duplicate_groups_skipped,
duplicate_files_hashed=duplicate_files_hashed,
min_duplicate_size_bytes=config.min_duplicate_size_bytes,
stale_large_files=stale_files,
stale_days=config.stale_days,
large_file_threshold_bytes=config.large_file_threshold_bytes,
protected_database_files=protected_database_files,
filesystems=filesystems,
root_filesystem=root_filesystem,
scan_coverage_percent=scan_coverage_percent,
)
def facts_to_prompt_payload(facts: AnalysisFacts) -> dict[str, Any]:
def age(record: LargeFileRecord) -> str:
return format_age_days(record.mtime, facts.generated_at)
return {
"generated_at": facts.generated_at.isoformat(),
"root_label": facts.root_label,
"cross_filesystems": facts.cross_filesystems,
"running_as_user": facts.running_as_user,
"running_as_root": facts.running_as_root,
"totals": {
"total_apparent_bytes": facts.total_apparent_bytes,
"total_apparent_human": format_bytes(facts.total_apparent_bytes),
"total_disk_bytes": facts.total_disk_bytes,
"total_disk_human": format_bytes(facts.total_disk_bytes),
"total_files": facts.total_files,
"total_directories": facts.total_directories,
"symlinks_skipped": facts.symlinks_skipped,
"boundary_skipped_mounts": [str(path) for path in facts.boundary_skipped_mounts[:10]],
"timed_out_entries": facts.timed_out_entries,
"capped_entries": facts.capped_entries,
"hardlink_bytes_deduplicated_human": format_bytes(facts.hardlink_bytes_deduplicated),
"hardlink_files_deduplicated": facts.hardlink_files_deduplicated,
"median_top_level_disk_bytes_human": format_bytes(int(facts.median_top_level_disk_bytes)),
"mean_top_level_disk_bytes_human": format_bytes(int(facts.mean_top_level_disk_bytes)),
"scan_coverage_percent_of_filesystem_used": (
round(facts.scan_coverage_percent, 1) if facts.scan_coverage_percent is not None else None
),
},
"inaccessible_locations": {
"count": facts.inaccessible_count,
"advice": "skip these locations (or re-run as a user/root with read access to them)",
"examples": [
{"path": str(path), "reason": reason, "advice": "skip this location"}
for path, reason in facts.inaccessible_examples[:INACCESSIBLE_EXAMPLES_SHOWN_IN_PAYLOAD]
],
},
"top_directories": [
{"path": str(frame.path), "disk_bytes_human": format_bytes(frame.disk_bytes), "files": frame.file_count}
for frame in facts.top_directories
],
"top_files": [
{
"path": str(record.path), "size_human": format_bytes(record.size_bytes),
"last_modified": age(record),
"is_database": record.is_database,
}
for record in facts.top_files
],
"top_extensions_by_size": [
{"extension": extension, "total_size_human": format_bytes(size), "file_count": count}
for extension, size, count in facts.top_extensions
],
"protected_database_files": {
"note": (
"These are database files (any format). Never recommend deleting, moving, truncating, "
"or otherwise modifying them; mention size/location as factual context only, and suggest "
"database-level maintenance (VACUUM, archiving, purging old rows) if relevant."
),
"files": [
{
"path": str(record.path), "size_human": format_bytes(record.size_bytes),
"last_modified": age(record), "reason": database_file_reason(record),
}
for record in facts.protected_database_files
],
},
"reclaimable_space": {
"recent_activity_window_days": facts.reclaimable_recent_activity_days,
"recommended_for_cleanup": {
"note": "these locations were NOT modified within the recent-activity window; safe to recommend for review",
"total_disk_bytes_human": format_bytes(facts.reclaimable_total_disk_bytes),
"by_category": [
{
"category": category, "disk_bytes_human": format_bytes(disk_bytes),
"file_count": file_count, "location_count": location_count,
}
for category, disk_bytes, file_count, location_count in facts.reclaimable_by_category
],
"top_locations": [
{
"path": str(candidate.path), "category": candidate.category,
"disk_bytes_human": format_bytes(candidate.disk_bytes), "file_count": candidate.file_count,
"reason": reclaimable_candidate_reason(
candidate, facts.reclaimable_recent_activity_days, facts.generated_at
),
}
for candidate in facts.top_reclaimable_candidates
],
},
"excluded_recently_active": {
"note": (
"these locations WERE modified within the recent-activity window and are presumed to be "
"in active use; do not recommend cleanup for them"
),
"location_count": facts.reclaimable_recent_location_count,
"total_disk_bytes_human": format_bytes(facts.reclaimable_recent_total_disk_bytes),
"examples": [
{
"path": str(candidate.path), "category": candidate.category,
"disk_bytes_human": format_bytes(candidate.disk_bytes),
"reason": reclaimable_candidate_reason(
candidate, facts.reclaimable_recent_activity_days, facts.generated_at
),
}
for candidate in facts.reclaimable_recent_examples
],
},
},
"duplicate_files": {
"note": "database files are never included here, regardless of duplication",
"min_duplicate_size_human": format_bytes(facts.min_duplicate_size_bytes),
"confirmed_group_count": len(facts.duplicate_groups),
"groups_skipped_due_to_limit": facts.duplicate_groups_skipped,
"files_hashed_for_confirmation": facts.duplicate_files_hashed,
"total_reclaimable_bytes_human": format_bytes(facts.duplicate_total_reclaimable_bytes),
"top_groups": [
{
"size_human": format_bytes(group.size_bytes), "copies": len(group.paths),
"reclaimable_human": format_bytes(group.reclaimable_bytes),
"paths": [str(path) for path in group.paths[:DUPLICATE_GROUP_PATHS_SHOWN_IN_PAYLOAD]],
"paths_omitted": max(0, len(group.paths) - DUPLICATE_GROUP_PATHS_SHOWN_IN_PAYLOAD),
"reason": duplicate_group_reason(group),
}
for group in facts.duplicate_groups[:20]
],
},
"stale_large_files": {
"note": "database files are never included here, regardless of staleness; see protected_database_files instead",
"stale_after_days": facts.stale_days,
"large_file_threshold_human": format_bytes(facts.large_file_threshold_bytes),
"files": [
{
"path": str(record.path), "size_human": format_bytes(record.size_bytes),
"last_modified": age(record),
"reason": stale_file_reason(record, facts.large_file_threshold_bytes, facts.stale_days, facts.generated_at),
}
for record in facts.stale_large_files
],
},
"filesystem_overview": {
"root_filesystem": (
{
"mount_point": facts.root_filesystem.mount_point,
"total_human": format_bytes(facts.root_filesystem.total_bytes),
"used_human": format_bytes(facts.root_filesystem.used_bytes),
"free_human": format_bytes(facts.root_filesystem.free_bytes),
"used_percent": round(facts.root_filesystem.used_percent, 1),
}
if facts.root_filesystem is not None else None
),
"all_mounted_filesystems": [
{
"mount_point": fact.mount_point, "type": fact.filesystem_type,
"total_human": format_bytes(fact.total_bytes), "used_human": format_bytes(fact.used_bytes),
"used_percent": round(fact.used_percent, 1),
}
for fact in facts.filesystems
],
},
}
REPORT_SYSTEM_PROMPT = """\
You write critical, evidence-based disk-usage sweep reports from structured JSON facts \
produced by a read-only filesystem scanner. You are professional, scientific, and concise. \
Never use emoticons, conversational filler, or self-referential AI language. Base every \
claim strictly on the JSON facts supplied by the user; never invent paths, byte counts, or \
categories not present in the data. Be balanced and genuinely critical: state real space \
consumers and real reclaim opportunities, supported by concrete figures quoted from the \
facts. Do not soften findings to be polite, and do not manufacture urgency or reassurance \
the data does not support.
This tool is strictly read-only: it never deletes, moves, renames, or modifies any file. \
Every recommendation you write must be framed as something for the reader to manually \
review and decide on, never as an instruction to blindly execute a destructive command. \
When you mention a cleanup candidate (a cache, build artifact, duplicate file, or old \
large file), state what it is, how large it is, and why it is typically safe to review \
for removal, but stop short of telling the reader to run rm -rf or similar without \
first inspecting the contents themselves. This report may be consumed by automated \
cleanup tooling, so every rule below is a hard constraint, not a stylistic preference.
HARD RULE — databases: never recommend deleting, moving, truncating, compacting, \
archiving-by-removal, or otherwise modifying any file listed under "protected_database_files" \
or flagged "is_database": true anywhere in the JSON, regardless of its size, age, or apparent \
redundancy, and regardless of database format (SQLite, MySQL/MariaDB, PostgreSQL, Access, \
Firebird, Redis RDB, DBF, or any other). You may state their path and size as neutral factual \
context. If you want to note a large database as a space consumer, phrase it as "review with \
the owning application (VACUUM/archiving/purging), not as a file to remove."
HARD RULE — recent activity: the JSON separates reclaimable locations into \
"recommended_for_cleanup" (untouched within the recent-activity window) and \
"excluded_recently_active" (modified within that window, presumed in active use). Only ever \
name specific locations from "recommended_for_cleanup" as cleanup candidates. You may mention \
the excluded_recently_active total as a factual aggregate, but never name or recommend \
removing an individual location from it.
HARD RULE — permissions: for every entry in "inaccessible_locations.examples", state plainly \
that the current user could not read it and advise the reader to skip that location (or re-run \
with a user/root that has access if they specifically need it included); never guess at what it \
might contain.
EVALUABILITY: every item under reclaimable_space, protected_database_files, duplicate_files, and \
stale_large_files carries a "reason" field explaining exactly why it was classified the way it \
was. When you name a specific location or file as a cleanup candidate (or explain why one was \
excluded), paraphrase or quote its "reason" field so the claim is traceable back to a concrete \
rule and measurement, not just an assertion. Do not write your own separate "rules" or "methodology" \
section — a "Decision Rules Applied" appendix listing every rule and this run's exact threshold \
values is appended programmatically after your narrative; do not duplicate it.
Structure the report as Markdown with exactly these sections, in this order:
1. Start with one bold-lead-in paragraph summarizing the overall picture.
2. "### 1. Overall Disk Picture"
3. "### 2. Largest Space Consumers"
4. "### 3. Reclaimable Space Candidates"
5. "### 4. Structural Patterns and Risks"
6. "### 5. Broader Implications"
7. "### Potential Actions to Consider" as a numbered list
8. A final paragraph starting with "**Bottom Line**:"
Each of sections 2 through 5 must end with a short italicized conclusion sentence starting \
with "**Conclusion**:" or "**Key observation**:" that states the critical takeaway for that \
section, grounded in the numbers just discussed. Do not include a top-level Markdown title, \
do not repeat the raw JSON, and do not use emoji.
"""
def build_user_prompt(payload: dict[str, Any]) -> str:
return (
"Analyze the following read-only disk-usage sweep dataset and produce the report "
"described in the system prompt. Every number you cite must come from this JSON.\n\n"
"```json\n" + json.dumps(payload, indent=2, sort_keys=False) + "\n```\n"
)
def call_gateway_chat(settings: GatewaySettings, system_prompt: str, user_prompt: str) -> str:
if not settings.api_key:
raise AnalysisError("DEVPLACE_API_KEY is not set")
payload = {
"model": settings.model,
"messages": [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt},
],
"temperature": settings.temperature,
"max_tokens": settings.max_output_tokens,
}
body = json.dumps(payload).encode("utf-8")
request = urllib.request.Request(
settings.endpoint,
data=body,
method="POST",
headers={
"Authorization": f"Bearer {settings.api_key}",
"Content-Type": "application/json",
},
)
try:
with urllib.request.urlopen(request, timeout=settings.request_timeout_seconds) as response:
raw_body = response.read()
gateway_headers = dict(response.headers)
except urllib.error.HTTPError as error:
detail = error.read().decode("utf-8", errors="replace")
raise AnalysisError(f"DevPlace gateway HTTP error {error.code}: {detail}") from error
except urllib.error.URLError as error:
raise AnalysisError(f"DevPlace gateway request failed: {error.reason!r}") from error
except TimeoutError as error:
raise AnalysisError(f"DevPlace gateway request timed out after {settings.request_timeout_seconds}s") from error
cost = gateway_headers.get("X-Gateway-Cost-USD")
total_tokens = gateway_headers.get("X-Gateway-Total-Tokens")
backend = gateway_headers.get("X-Gateway-Backend")
if cost is not None or total_tokens is not None:
LOGGER.info(f"gateway usage: backend={backend} total-tokens={total_tokens} cost-usd={cost}")
try:
parsed = json.loads(raw_body)
choice = parsed["choices"][0]
content = choice["message"]["content"]
except (json.JSONDecodeError, KeyError, IndexError, TypeError) as error:
raise AnalysisError(f"unexpected gateway response shape: {error!r}") from error
finish_reason = choice.get("finish_reason")
if finish_reason == "length":
usage = parsed.get("usage", {})
raise AnalysisError(
f"gateway response was truncated (finish_reason=length, usage={usage}); "
"raise --max-output-tokens or shorten the prompt"
)
if finish_reason not in (None, "stop"):
LOGGER.warning(f"gateway finish_reason was {finish_reason!r}, response may be incomplete")
return content.strip()
async def generate_llm_report(settings: GatewaySettings, payload: dict[str, Any]) -> str:
loop = asyncio.get_running_loop()
user_prompt = build_user_prompt(payload)
LOGGER.info(f"requesting narrative analysis from DevPlace gateway model={settings.model}")
started = time.monotonic()
with ThreadPoolExecutor(max_workers=1) as executor:
content = await loop.run_in_executor(
executor, call_gateway_chat, settings, REPORT_SYSTEM_PROMPT, user_prompt
)
LOGGER.info(f"gateway response received in {time.monotonic() - started:.2f}s ({len(content)} chars)")
return content
def build_fallback_report(facts: AnalysisFacts) -> str:
lines: list[str] = []
lines.append(
f"**{format_bytes(facts.total_disk_bytes)} of actual disk usage** were measured across "
f"**{facts.total_files} file(s)** in **{facts.total_directories} director(y/ies)** under "
f"`{facts.root_label}`. This is a deterministic, template-based analysis: DevPlace gateway "
"narrative generation was not used for this report. This tool is read-only and modified nothing."
)
lines.append("")
lines.append("### 1. Overall Disk Picture")
role_label = "root (superuser)" if facts.running_as_root else "a regular user"
lines.append(f"- Running as: `{facts.running_as_user}` ({role_label}).")
lines.append(
f"- Apparent size: {format_bytes(facts.total_apparent_bytes)}; actual on-disk size: "
f"{format_bytes(facts.total_disk_bytes)} (block-rounding and sparse files explain the difference)."
)
if facts.root_filesystem is not None:
lines.append(
f"- Scan root filesystem: {format_bytes(facts.root_filesystem.used_bytes)} used of "
f"{format_bytes(facts.root_filesystem.total_bytes)} total "
f"({facts.root_filesystem.used_percent:.1f}% full)."
)
if facts.scan_coverage_percent is not None:
lines.append(
f"- This scan directly measured {facts.scan_coverage_percent:.1f}% of the bytes the "
"filesystem reports as used; the remainder is outside the scan root, on other mounted "
"filesystems, or inaccessible."
)
if facts.hardlink_files_deduplicated:
lines.append(
f"- {facts.hardlink_files_deduplicated} hard-linked file(s) "
f"({format_bytes(facts.hardlink_bytes_deduplicated)}) were counted only once, avoiding "
"double counting."
)
if facts.inaccessible_count:
lines.append(
f"- {facts.inaccessible_count} location(s) could not be read by `{facts.running_as_user}` "
"(permission denied or transient errors); their contents are not reflected in totals above. "
"Skip these locations, or re-run as a user/root with read access if they must be included:"
)
for path, _ in facts.inaccessible_examples[:INACCESSIBLE_EXAMPLES_SHOWN_IN_PAYLOAD]:
lines.append(f" - `{path}` — skip this location")
omitted_inaccessible = facts.inaccessible_count - min(
len(facts.inaccessible_examples), INACCESSIBLE_EXAMPLES_SHOWN_IN_PAYLOAD
)
if omitted_inaccessible > 0:
lines.append(f" - ... {omitted_inaccessible} more inaccessible location(s) omitted ...")
if facts.timed_out_entries or facts.capped_entries:
lines.append(
f"- {len(facts.timed_out_entries)} top-level entr(y/ies) hit the scan deadline and "
f"{len(facts.capped_entries)} hit the directory cap; true size for those is at least as "
"high as reported."
)
lines.append(
"**Conclusion**: " + (
"The measured on-disk footprint accounts for the large majority of what the filesystem "
"reports as used, giving high confidence in the figures below."
if (facts.scan_coverage_percent or 0) >= 80 else
"A meaningful share of used disk space falls outside this scan's direct coverage; treat "
"the figures below as a lower bound rather than the full picture."
)
)
lines.append("")
lines.append("### 2. Largest Space Consumers")
if facts.top_directories:
top_dirs = "; ".join(
f"{frame.path} ({format_bytes(frame.disk_bytes)})" for frame in facts.top_directories[:8]
)
lines.append(f"- Heaviest directories: {top_dirs}.")
if facts.top_files:
top_files = "; ".join(
f"{record.path} ({format_bytes(record.size_bytes)}{', database file — protected' if record.is_database else ''})"
for record in facts.top_files[:8]
)
lines.append(f"- Heaviest individual files: {top_files}.")
if facts.top_extensions:
top_ext = ", ".join(
f"{extension} ({format_bytes(size)}, {count} file(s))"
for extension, size, count in facts.top_extensions[:10]
)
lines.append(f"- Space by file type: {top_ext}.")
lines.append(
"**Conclusion**: " + (
"Disk usage is concentrated in a small number of directories and file types rather than "
"spread evenly, which narrows where cleanup effort will have the most impact."
if facts.top_directories else
"No directories exceeded the reclaimable/tracking thresholds for this run."
)
)
lines.append("")
lines.append("### 3. Reclaimable Space Candidates")
if facts.reclaimable_by_category:
category_lines = "; ".join(
f"{category} ({format_bytes(disk_bytes)} across {location_count} location(s))"
for category, disk_bytes, _, location_count in facts.reclaimable_by_category[:10]
)
lines.append(
f"- Known generated/cache/vendor directories not modified in the last "
f"{facts.reclaimable_recent_activity_days} day(s) total "
f"{format_bytes(facts.reclaimable_total_disk_bytes)} and are recommended for review: "
f"{category_lines}."
)
else:
lines.append("- No well-known generated/cache/vendor directories were detected as safe to recommend.")
if facts.top_reclaimable_candidates:
lines.append("- Evidence for the top recommended candidates (evaluate each against its stated reason):")
for candidate in facts.top_reclaimable_candidates[:5]:
reason = reclaimable_candidate_reason(candidate, facts.reclaimable_recent_activity_days, facts.generated_at)
lines.append(f" - `{candidate.path}` ({format_bytes(candidate.disk_bytes)}): {reason}")
if facts.reclaimable_recent_location_count:
lines.append(
f"- {facts.reclaimable_recent_location_count} additional reclaimable-category location(s) "
f"totaling {format_bytes(facts.reclaimable_recent_total_disk_bytes)} were modified within the "
f"last {facts.reclaimable_recent_activity_days} day(s) and are presumed in active use; they are "
"excluded from cleanup recommendations. Evidence for the largest excluded location(s):"
)
for candidate in facts.reclaimable_recent_examples[:3]:
reason = reclaimable_candidate_reason(candidate, facts.reclaimable_recent_activity_days, facts.generated_at)
lines.append(f" - `{candidate.path}` ({format_bytes(candidate.disk_bytes)}): {reason}")
if facts.duplicate_groups:
lines.append(
f"- {len(facts.duplicate_groups)} confirmed duplicate-file group(s) "
f"({facts.duplicate_files_hashed} file(s) hash-verified, database files never included) could "
f"reclaim {format_bytes(facts.duplicate_total_reclaimable_bytes)} if redundant copies were removed."
)
for group in facts.duplicate_groups[:3]:
lines.append(f" - {duplicate_group_reason(group)}")
if facts.duplicate_groups_skipped:
lines.append(
f"- {facts.duplicate_groups_skipped} additional same-size group(s) were not "
"hash-verified due to the group limit for this run."
)
if facts.stale_large_files:
stale_names = ", ".join(
f"{record.path} ({format_bytes(record.size_bytes)}, unused {facts.stale_days}+ days)"
for record in facts.stale_large_files[:6]
)
lines.append(f"- Large non-database files untouched for {facts.stale_days}+ days: {stale_names}.")
top_stale = facts.stale_large_files[0]
lines.append(
f" - Evidence for `{top_stale.path}`: "
f"{stale_file_reason(top_stale, facts.large_file_threshold_bytes, facts.stale_days, facts.generated_at)}"
)
if facts.protected_database_files:
db_total = sum(record.size_bytes for record in facts.protected_database_files)
db_names = ", ".join(
f"{record.path} ({format_bytes(record.size_bytes)})" for record in facts.protected_database_files[:6]
)
lines.append(
f"- {len(facts.protected_database_files)} large database file(s) totaling {format_bytes(db_total)} "
f"were found and are excluded from every cleanup recommendation below: {db_names}. Use "
"database-level maintenance (VACUUM, archiving, purging old rows) instead of file deletion. "
f"Evidence: {database_file_reason(facts.protected_database_files[0])}"
)
lines.append(
"**Conclusion**: " + (
f"Reviewing generated/cache directories, confirmed duplicates, and stale large files "
f"together represents a real, non-destructive-to-decide reclaim opportunity of roughly "
f"{format_bytes(facts.reclaimable_total_disk_bytes + facts.duplicate_total_reclaimable_bytes)}."
if (facts.reclaimable_total_disk_bytes or facts.duplicate_total_reclaimable_bytes) else
"No significant reclaimable-space signals were detected in this dataset."
)
)
lines.append("")
lines.append("### 4. Structural Patterns and Risks")
scattered_categories = [item for item in facts.reclaimable_by_category if item[3] >= 3]
if scattered_categories:
scattered_names = ", ".join(f"{category} ({count}x)" for category, _, _, count in scattered_categories[:8])
lines.append(f"- Categories repeated across many locations (possible sprawl): {scattered_names}.")
if facts.boundary_skipped_mounts:
lines.append(
f"- {len(facts.boundary_skipped_mounts)} separately mounted filesystem(s) under the scan "
"root were not descended into; pass --cross-filesystems to include them."
)
if facts.symlinks_skipped:
lines.append(f"- {facts.symlinks_skipped} symbolic link(s) were encountered and skipped (never followed).")
lines.append(
"**Key observation**: " + (
"Repeated generated-directory categories across many locations point to per-project "
"tooling artifacts accumulating system-wide rather than a single large offender."
if scattered_categories else
"No strong sprawl signal was detected; reclaimable space is concentrated rather than scattered."
)
)
lines.append("")
lines.append("### 5. Broader Implications")
if facts.root_filesystem is not None:
lines.append(
f"- At {facts.root_filesystem.used_percent:.1f}% full "
f"({format_bytes(facts.root_filesystem.free_bytes)} free), the urgency of reclaiming space "
"depends on how close this is to the filesystem's practical operating threshold."
)
risk_signals = []
if facts.reclaimable_total_disk_bytes:
risk_signals.append("generated/cache sprawl")
if facts.duplicate_total_reclaimable_bytes:
risk_signals.append("duplicate data")
if facts.inaccessible_count:
risk_signals.append("incomplete coverage due to permissions")
lines.append(
"**Conclusion**: " + (
f"The dataset shows identifiable, low-risk reclaim opportunities ({', '.join(risk_signals)}) "
"that can reduce disk pressure without touching personal data."
if risk_signals else
"No material risk signals were detected in this dataset."
)
)
lines.append("")
lines.append("### Potential Actions to Consider")
suggestion_number = 1
if facts.reclaimable_by_category:
lines.append(
f"{suggestion_number}. Review the {len(facts.reclaimable_by_category)} reclaimable "
f"categor(y/ies) above (total {format_bytes(facts.reclaimable_total_disk_bytes)}); most "
"package/build caches regenerate automatically and are typically safe to clear after "
"confirming nothing is mid-build."
)
suggestion_number += 1
if facts.duplicate_groups:
lines.append(
f"{suggestion_number}. Inspect the confirmed duplicate-file groups and manually remove "
"redundant copies you no longer need, keeping at least one copy of each."
)
suggestion_number += 1
if facts.stale_large_files:
lines.append(
f"{suggestion_number}. Review the stale large non-database files list for archival candidates "
"(move to cold storage) rather than immediate deletion."
)
suggestion_number += 1
if facts.protected_database_files:
lines.append(
f"{suggestion_number}. Do not delete or move any of the "
f"{len(facts.protected_database_files)} database file(s) listed above; if they need to shrink, "
"use the owning application's own maintenance (VACUUM, archiving, purging old rows)."
)
suggestion_number += 1
if facts.reclaimable_recent_location_count:
lines.append(
f"{suggestion_number}. Leave the {facts.reclaimable_recent_location_count} recently-active "
f"reclaimable location(s) alone for now; revisit them in a future sweep once they have been "
f"untouched for {facts.reclaimable_recent_activity_days}+ days."
)
suggestion_number += 1
if facts.inaccessible_count:
lines.append(
f"{suggestion_number}. Skip the {facts.inaccessible_count} inaccessible location(s) listed "
f"above, or re-run as a user/root with read access to them for a more complete picture."
)
suggestion_number += 1
if facts.boundary_skipped_mounts:
lines.append(
f"{suggestion_number}. Re-run with --cross-filesystems if the "
f"{len(facts.boundary_skipped_mounts)} skipped mount(s) should also be swept."
)
suggestion_number += 1
lines.append(
f"{suggestion_number}. Re-run this sweep periodically to track whether reclaimable categories "
"are growing faster than they are being cleaned up."
)
lines.append("")
lines.append(
"**Bottom Line**: This is a deterministic summary generated without an LLM narrative pass; "
"figures above are exact given the input data and this scan's coverage, but the prose is "
"templated rather than reasoned. Re-run with LLM generation enabled and a valid DEVPLACE_API_KEY "
"for a fuller critical narrative. No files were modified in producing this report."
)
return "\n".join(lines)
def render_rules_appendix(facts: AnalysisFacts) -> str:
lines: list[str] = []
lines.append("### Decision Rules Applied")
lines.append(
"This section is generated deterministically by code, not by the narrative model above, so every "
"finding in this report can be independently checked against it. Each finding's `reason` field in "
"the underlying data cites the specific rule that produced it."
)
lines.append("")
lines.append(
"1. **Read-only, always.** This tool never deletes, moves, renames, or modifies any file or "
"directory. Every action in this run was a read/stat/hash operation only."
)
lines.append(
"2. **True on-disk size.** Bytes are counted via `st_blocks * 512` (actual allocated blocks), not "
f"just `st_size` (apparent size); this run measured {format_bytes(facts.total_disk_bytes)} actual "
f"vs {format_bytes(facts.total_apparent_bytes)} apparent."
)
lines.append(
"3. **Hard-link deduplication.** A file with more than one hard link is counted exactly once, keyed "
f"by (device, inode), across the whole scan. This run deduplicated "
f"{facts.hardlink_files_deduplicated} file(s) / {format_bytes(facts.hardlink_bytes_deduplicated)}."
)
lines.append(
f"4. **Filesystem boundary.** Subdirectories on a different device (`st_dev`) than the scan root are "
f"not descended into unless `--cross-filesystems` is passed (this run: "
f"{'enabled' if facts.cross_filesystems else 'disabled'}). "
f"{len(facts.boundary_skipped_mounts)} mount(s) were skipped this run."
)
lines.append(
f"5. **Reclaimable-category match.** A directory is sized as a single aggregate 'reclaimable "
f"candidate' ONLY if its name (case-insensitive, exact match) is one of "
f"{len(RECLAIMABLE_DIRECTORY_NAMES)} known generated/cache/vendor directory names (e.g. "
"`node_modules`, `__pycache__`, `venv`, `build`, `.cache`, `trash`); matching a name stops further "
"descent and the entire subtree's size is attributed to that category. No content inspection or "
"heuristic guessing is used — only the directory name."
)
lines.append(
f"6. **Recent-activity exclusion.** A reclaimable candidate is only recommended for cleanup if no "
f"file inside it was modified within the last {facts.reclaimable_recent_activity_days} day(s) "
f"(~{facts.reclaimable_recent_activity_days // 7} week(s)); this is checked against the single most "
f"recent file modification time found anywhere inside that directory. This run excluded "
f"{facts.reclaimable_recent_location_count} location(s) totaling "
f"{format_bytes(facts.reclaimable_recent_total_disk_bytes)} for being touched more recently — see "
"`reclaimable_space.excluded_recently_active` for the list."
)
lines.append(
f"7. **Database protection (hard rule).** Any file whose extension is one of "
f"{len(DATABASE_FILE_EXTENSIONS)} known database-format extensions (SQLite, MySQL/MariaDB, "
"PostgreSQL/Access-adjacent, Firebird, Redis RDB, DBF, and their WAL/SHM/journal siblings) is "
"excluded from duplicate detection and stale-file candidacy at the code level, and is never "
f"eligible for a removal/move recommendation regardless of size, age, or redundancy. "
f"{len(facts.protected_database_files)} such file(s) were found this run."
)
lines.append(
f"8. **Duplicate confirmation.** Files are grouped by exact byte size first (only sizes "
f">= {format_bytes(facts.min_duplicate_size_bytes)} are tracked); the highest-potential-reclaim "
"same-size groups are then fully SHA-256 hashed, and only groups where >=2 files share an identical "
f"hash are reported as confirmed duplicates. This run hashed {facts.duplicate_files_hashed} file(s) "
f"and confirmed {len(facts.duplicate_groups)} duplicate group(s); "
f"{facts.duplicate_groups_skipped} lower-priority same-size group(s) were not hash-verified due to "
"this run's group limit (so more duplicates may exist beyond what is reported)."
)
lines.append(
f"9. **Stale large file.** A non-database file is flagged stale only if its size is at or above "
f"{format_bytes(facts.large_file_threshold_bytes)} AND both its last-modified and last-accessed "
f"time are older than {facts.stale_days} day(s)."
)
lines.append(
f"10. **Permission handling.** Any path this run could not read (permission denied or a transient "
"error) is listed under inaccessible locations with explicit advice to skip it; this tool never "
f"estimates or guesses at the contents of a path it could not read. This run found "
f"{facts.inaccessible_count} such location(s), run as `{facts.running_as_user}` "
f"({'root' if facts.running_as_root else 'a regular user'})."
)
return "\n".join(lines)
def render_markdown_report(facts: AnalysisFacts, narrative: str) -> str:
role_label = "root" if facts.running_as_root else "regular user"
header = (
f"# Disk Usage Sweep Report\n\n"
f"Generated: {facts.generated_at:%Y-%m-%d %H:%M:%S} \n"
f"Scan root: {facts.root_label} \n"
f"Running as: {facts.running_as_user} ({role_label}) \n"
f"Cross-filesystem scan: {'yes' if facts.cross_filesystems else 'no'} \n\n"
"---\n\n"
)
return header + narrative.strip() + "\n\n---\n\n" + render_rules_appendix(facts) + "\n"
async def generate_report(config: AnalysisConfig) -> str:
results, state = await run_scan(config.scan_config)
results.append(scan_root_level_files(config.scan_config, state))
duplicate_groups, duplicate_groups_skipped, duplicate_files_hashed = await find_confirmed_duplicates(
state, config.scan_config
)
filesystems = discover_real_filesystems()
analysis_facts = build_facts(
results=results, state=state, config=config.scan_config,
duplicate_groups=duplicate_groups, duplicate_groups_skipped=duplicate_groups_skipped,
duplicate_files_hashed=duplicate_files_hashed, filesystems=filesystems,
)
narrative: str | None = None
if config.use_llm:
payload = facts_to_prompt_payload(analysis_facts)
try:
narrative = await generate_llm_report(config.gateway_settings, payload)
except AnalysisError as error:
LOGGER.error(f"LLM narrative generation failed, falling back to deterministic report: {error}")
if narrative is None:
narrative = build_fallback_report(analysis_facts)
return render_markdown_report(analysis_facts, narrative)
def build_arg_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
description=(
"Perform a read-only, high-accuracy disk-usage sweep and generate a critical, "
"reasoning-based narrative report: largest space consumers, reclaimable caches/build "
"artifacts, confirmed duplicate files, and stale large files. Never deletes anything, "
"never recommends removing database files, and never recommends cleanup of recently-active "
"locations. Works as any user, including root; pass '/' as root for a whole-disk sweep."
),
)
parser.add_argument(
"root", nargs="?", default=Path.cwd(), type=Path,
help="root directory to sweep (default: current working directory)",
)
parser.add_argument(
"--cross-filesystems", action="store_true",
help="also descend into separately mounted filesystems and network shares under root",
)
parser.add_argument(
"--output", type=Path, default=None,
help="write the Markdown report to this path instead of stdout",
)
parser.add_argument(
"--model", default=os.environ.get("DEVPLACE_MODEL", DEFAULT_GATEWAY_MODEL),
help="DevPlace gateway model id for narrative generation (default: %(default)s)",
)
parser.add_argument(
"--api-key", default=None,
help="DevPlace gateway API key (default: DEVPLACE_API_KEY environment variable)",
)
parser.add_argument(
"--no-llm", action="store_true",
help="skip DevPlace gateway narrative generation and emit a deterministic template report",
)
parser.add_argument(
"--exclude", action="append", default=[], metavar="NAME",
help="additional directory name to always skip (repeatable, case-insensitive)",
)
parser.add_argument("--dir-timeout", type=float, default=DEFAULT_DIRECTORY_TIMEOUT_SECONDS)
parser.add_argument("--max-directories", type=int, default=DEFAULT_MAX_DIRECTORIES_PER_TOP_LEVEL)
parser.add_argument("--workers", type=int, default=DEFAULT_MAX_WORKERS)
parser.add_argument("--top-dirs", type=int, default=DEFAULT_TOP_DIRECTORIES_SHOWN)
parser.add_argument("--top-files", type=int, default=DEFAULT_TOP_FILES_SHOWN)
parser.add_argument("--top-extensions", type=int, default=DEFAULT_TOP_EXTENSIONS_SHOWN)
parser.add_argument("--large-file-threshold-mb", type=float, default=DEFAULT_LARGE_FILE_THRESHOLD_MB)
parser.add_argument("--stale-days", type=int, default=DEFAULT_STALE_DAYS)
parser.add_argument(
"--reclaimable-recent-days", type=int, default=DEFAULT_RECLAIMABLE_RECENT_ACTIVITY_DAYS,
help=(
"reclaimable-category locations modified within this many days are treated as actively used "
"and excluded from cleanup recommendations (default: %(default)s, i.e. 14 weeks)"
),
)
parser.add_argument("--min-duplicate-size-mb", type=float, default=DEFAULT_MIN_DUPLICATE_SIZE_MB)
parser.add_argument("--max-duplicate-groups", type=int, default=DEFAULT_MAX_DUPLICATE_GROUPS_HASHED)
parser.add_argument("--temperature", type=float, default=DEFAULT_TEMPERATURE)
parser.add_argument("--max-output-tokens", type=int, default=DEFAULT_MAX_OUTPUT_TOKENS)
parser.add_argument("--request-timeout", type=float, default=DEFAULT_REQUEST_TIMEOUT_SECONDS)
parser.add_argument("--verbose", action="store_true")
return parser
def validate_args(args: argparse.Namespace) -> str | None:
if not Path(args.root).exists():
return f"root does not exist: {args.root}"
if args.dir_timeout <= 0:
return "--dir-timeout must be greater than 0"
if args.max_directories < 1:
return "--max-directories must be at least 1"
if args.workers < 1:
return "--workers must be at least 1"
if args.top_dirs < 1:
return "--top-dirs must be at least 1"
if args.top_files < 1:
return "--top-files must be at least 1"
if args.top_extensions < 1:
return "--top-extensions must be at least 1"
if args.large_file_threshold_mb <= 0:
return "--large-file-threshold-mb must be greater than 0"
if args.stale_days < 0:
return "--stale-days cannot be negative"
if args.reclaimable_recent_days < 0:
return "--reclaimable-recent-days cannot be negative"
if args.min_duplicate_size_mb <= 0:
return "--min-duplicate-size-mb must be greater than 0"
if args.max_duplicate_groups < 0:
return "--max-duplicate-groups cannot be negative"
if args.request_timeout <= 0:
return "--request-timeout must be greater than 0"
if args.max_output_tokens < 1:
return "--max-output-tokens must be at least 1"
return None
class ElapsedFormatter(logging.Formatter):
def __init__(self, start_time: float) -> None:
super().__init__()
self.start_time = start_time
def format(self, record: logging.LogRecord) -> str:
elapsed = time.monotonic() - self.start_time
return f"[{elapsed:7.3f}s] {record.levelname:<7} {record.getMessage()}"
def configure_logging(verbose: bool) -> None:
start_time = time.monotonic()
LOGGER.setLevel(logging.DEBUG if verbose else logging.INFO)
handler = logging.StreamHandler(sys.stderr)
handler.setFormatter(ElapsedFormatter(start_time))
LOGGER.handlers.clear()
LOGGER.addHandler(handler)
LOGGER.propagate = False
def write_report(report: str, output_path: Path | None) -> None:
if output_path is None:
print(report)
return
try:
output_path.parent.mkdir(parents=True, exist_ok=True)
output_path.write_text(report, encoding="utf-8")
except OSError as error:
raise AnalysisError(f"could not write report to {output_path}: {error!r}") from error
LOGGER.info(f"report written to {output_path}")
def main() -> int:
args = build_arg_parser().parse_args()
configure_logging(args.verbose)
validation_error = validate_args(args)
if validation_error is not None:
print(f"error: {validation_error}", file=sys.stderr)
return 2
load_dotenv_if_present(Path.cwd() / ".env", Path(__file__).resolve().parent / ".env")
api_key = args.api_key or os.environ.get("DEVPLACE_API_KEY")
scan_config = ScanConfig(
root=Path(args.root).resolve(),
cross_filesystems=args.cross_filesystems,
directory_timeout_seconds=args.dir_timeout,
max_directories_per_top_level=args.max_directories,
max_workers=args.workers,
min_duplicate_size_bytes=int(args.min_duplicate_size_mb * 1024 * 1024),
max_duplicate_groups_hashed=args.max_duplicate_groups,
large_file_threshold_bytes=int(args.large_file_threshold_mb * 1024 * 1024),
stale_days=args.stale_days,
top_directories_shown=args.top_dirs,
top_files_shown=args.top_files,
top_extensions_shown=args.top_extensions,
reclaimable_recent_activity_days=args.reclaimable_recent_days,
extra_excluded_names=frozenset(name.lower() for name in args.exclude),
verbose=args.verbose,
)
gateway_settings = GatewaySettings(
api_key=api_key,
model=args.model,
endpoint=DEFAULT_GATEWAY_ENDPOINT,
temperature=args.temperature,
max_output_tokens=args.max_output_tokens,
request_timeout_seconds=args.request_timeout,
)
config = AnalysisConfig(
scan_config=scan_config,
gateway_settings=gateway_settings,
output_path=args.output.resolve() if args.output else None,
use_llm=not args.no_llm and bool(api_key),
verbose=args.verbose,
)
if args.no_llm:
LOGGER.info("LLM narrative generation disabled via --no-llm")
elif not api_key:
LOGGER.warning("no DevPlace gateway API key available, falling back to deterministic report")
log_scan_criteria(scan_config)
try:
report = asyncio.run(generate_report(config))
except AnalysisError as error:
LOGGER.error(f"analysis failed: {error}")
return 1
except KeyboardInterrupt:
LOGGER.warning("interrupted by user")
return 130
except Exception as error:
LOGGER.error(f"unexpected top-level error, aborting cleanly: {error!r}")
return 1
try:
write_report(report, config.output_path)
except AnalysisError as error:
LOGGER.error(str(error))
return 1
return 0
if __name__ == "__main__":
sys.exit(main())
Comments
No comments yet. Start the discussion.