696 lines
28 KiB
Python
Executable File
696 lines
28 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
|
|
"""Read storage layout, usage, and drive health for Panama's Storage page.
|
|
|
|
Two boundaries, deliberately separate:
|
|
|
|
snapshot topology, usage, health, removable media. Cheap -- lsblk and a
|
|
single udisks call -- so the page can open with it.
|
|
scan what is actually filling the drive. Expensive: measuring a folder
|
|
means walking it, and this machine has a 1.2 TiB Steam library.
|
|
The page asks for this on demand and remembers the answer.
|
|
|
|
breakdown what the used space is made of, as four segments and a remainder.
|
|
Same walk as `scan`, so it costs the same and is asked for on
|
|
demand.
|
|
cleanables what could be freed, itemized and sized. Reading only.
|
|
clean ID frees exactly one of them, named explicitly.
|
|
|
|
unmount PATH / eject PATH removable media only, by explicit request.
|
|
|
|
Deliberately absent: partitioning and formatting. A settings pane is the wrong
|
|
place to hand someone a way to erase a disk in two clicks; GNOME Disks is one
|
|
button away on the page for that.
|
|
|
|
Also deliberately absent: anything that runs on its own. Nothing here is
|
|
pre-selected, nothing is measured in order to nag about it, and `clean` refuses
|
|
every id it was not handed. A storage page that cleans things you did not ask it
|
|
to clean is a cleaner, and cleaners are how people lose files.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
|
|
# Bounded so a pathological tree cannot hang the page. A folder that cannot be
|
|
# measured in this long is reported as unmeasured rather than silently omitted,
|
|
# because a missing row reads as "this folder is small".
|
|
SCAN_TIMEOUT_SECONDS = 90
|
|
|
|
# Where "what is filling my drive" actually gets answered. Ordered so the most
|
|
# likely culprits are measured first; the walk stops when the budget runs out.
|
|
SCAN_TARGETS = [
|
|
("Steam library", "~/.local/share/Steam"),
|
|
("Documents", "~/Documents"),
|
|
("Downloads", "~/Downloads"),
|
|
("Videos", "~/Videos"),
|
|
("Pictures", "~/Pictures"),
|
|
("Music", "~/Music"),
|
|
("Flatpak applications", "~/.var/app"),
|
|
("Caches", "~/.cache"),
|
|
("Trash", "~/.local/share/Trash"),
|
|
]
|
|
|
|
# The cache folder is its own segment in the breakdown and its own cleanable, so
|
|
# it is named once here rather than spelled out in three places.
|
|
CACHE_TARGET = "~/.cache"
|
|
TRASH_TARGET = "~/.local/share/Trash"
|
|
|
|
# Where flatpak keeps what it installed. Measured rather than summed from
|
|
# `flatpak list --columns=size`: those are per-ref installed sizes, and ostree
|
|
# hard-links every object shared between refs, so adding them up on this machine
|
|
# reports about twice what the drive actually holds.
|
|
FLATPAK_ROOTS = ["/var/lib/flatpak", "~/.local/share/flatpak"]
|
|
|
|
# dnf keeps downloaded rpms under <repo>/packages and its metadata beside them.
|
|
# Only the packages are offered: dropping the metadata costs a re-download on
|
|
# the next install and frees comparatively little.
|
|
DNF_CACHE_ROOTS = ["/var/cache/libdnf5", "/var/cache/dnf"]
|
|
|
|
|
|
class BoundaryError(RuntimeError):
|
|
"""A user-visible validation or command failure."""
|
|
|
|
|
|
def run(command: list[str], timeout: float = 15.0) -> str:
|
|
try:
|
|
completed = subprocess.run(command, check=False, capture_output=True,
|
|
text=True, timeout=timeout)
|
|
except (OSError, subprocess.TimeoutExpired) as error:
|
|
raise BoundaryError(f"{command[0]} did not answer") from error
|
|
if completed.returncode != 0:
|
|
raise BoundaryError(completed.stderr.strip() or f"{command[0]} failed")
|
|
return completed.stdout
|
|
|
|
|
|
def lsblk() -> list[dict]:
|
|
"""The block device tree.
|
|
|
|
PANAMA_DISKS_LSBLK substitutes a recorded tree for the real one. It exists
|
|
so the refusal rules below can be tested against a removable drive that is
|
|
not plugged in -- the alternative is a test that runs unmount against
|
|
whatever is actually mounted, which is not a test anyone should write.
|
|
"""
|
|
fixture = os.environ.get("PANAMA_DISKS_LSBLK")
|
|
if fixture:
|
|
try:
|
|
return json.loads(Path(fixture).read_text(encoding="utf-8"))["blockdevices"]
|
|
except (OSError, json.JSONDecodeError, KeyError) as error:
|
|
raise BoundaryError("could not read the recorded block device layout") from error
|
|
fields = ("NAME,PATH,SIZE,TYPE,FSTYPE,MOUNTPOINTS,MODEL,SERIAL,RM,HOTPLUG,"
|
|
"ROTA,FSSIZE,FSUSED,FSAVAIL")
|
|
try:
|
|
return json.loads(run(["lsblk", "-J", "-b", "-o", fields]))["blockdevices"]
|
|
except (json.JSONDecodeError, KeyError) as error:
|
|
raise BoundaryError("could not read the block device layout") from error
|
|
|
|
|
|
def udisks_objects() -> dict:
|
|
"""Every udisks object in one call, or nothing if udisks is not running.
|
|
|
|
Health is a nice-to-have: a machine without udisks still gets its layout and
|
|
usage, which is most of the page. Failing the whole snapshot because a
|
|
temperature is unavailable would be the wrong trade.
|
|
"""
|
|
if not shutil.which("busctl"):
|
|
return {}
|
|
try:
|
|
raw = run(["busctl", "--system", "--json=short", "call",
|
|
"org.freedesktop.UDisks2", "/org/freedesktop/UDisks2",
|
|
"org.freedesktop.DBus.ObjectManager", "GetManagedObjects"])
|
|
payload = json.loads(raw)
|
|
except (BoundaryError, json.JSONDecodeError):
|
|
return {}
|
|
# busctl wraps the reply as {"type": "...", "data": [ {path: {iface: {prop: {"type","data"}}}} ]}
|
|
data = payload.get("data") or []
|
|
return data[0] if data else {}
|
|
|
|
|
|
def unwrap(value):
|
|
"""busctl encodes every value as {"type": t, "data": v}."""
|
|
if isinstance(value, dict) and "data" in value and "type" in value:
|
|
return value["data"]
|
|
return value
|
|
|
|
|
|
def drive_health(objects: dict) -> dict[str, dict]:
|
|
"""Health keyed by drive serial, which is the only stable join to lsblk."""
|
|
health: dict[str, dict] = {}
|
|
for path, interfaces in objects.items():
|
|
if "/drives/" not in path:
|
|
continue
|
|
drive = {key: unwrap(value)
|
|
for key, value in (interfaces.get("org.freedesktop.UDisks2.Drive") or {}).items()}
|
|
serial = str(drive.get("Serial") or "")
|
|
if not serial:
|
|
continue
|
|
|
|
entry = {
|
|
"ejectable": bool(drive.get("Ejectable")),
|
|
"removable": bool(drive.get("Removable")),
|
|
"temperatureC": None,
|
|
"powerOnHours": None,
|
|
"healthy": None,
|
|
"warnings": [],
|
|
"selfTest": "",
|
|
}
|
|
|
|
nvme = {key: unwrap(value) for key, value in
|
|
(interfaces.get("org.freedesktop.UDisks2.NVMe.Controller") or {}).items()}
|
|
ata = {key: unwrap(value) for key, value in
|
|
(interfaces.get("org.freedesktop.UDisks2.Drive.Ata") or {}).items()}
|
|
|
|
if nvme:
|
|
kelvin = nvme.get("SmartTemperature")
|
|
if isinstance(kelvin, (int, float)) and kelvin > 0:
|
|
entry["temperatureC"] = round(kelvin - 273.15, 1)
|
|
hours = nvme.get("SmartPowerOnHours")
|
|
if isinstance(hours, (int, float)):
|
|
entry["powerOnHours"] = int(hours)
|
|
warnings = nvme.get("SmartCriticalWarning") or []
|
|
entry["warnings"] = [str(item) for item in warnings]
|
|
entry["healthy"] = not entry["warnings"]
|
|
entry["selfTest"] = str(nvme.get("SmartSelftestStatus") or "")
|
|
elif ata:
|
|
kelvin = ata.get("SmartTemperature")
|
|
if isinstance(kelvin, (int, float)) and kelvin > 0:
|
|
entry["temperatureC"] = round(kelvin - 273.15, 1)
|
|
seconds = ata.get("SmartPowerOnSeconds")
|
|
if isinstance(seconds, (int, float)) and seconds > 0:
|
|
entry["powerOnHours"] = int(seconds // 3600)
|
|
failing = ata.get("SmartFailing")
|
|
if isinstance(failing, bool):
|
|
entry["healthy"] = not failing
|
|
entry["warnings"] = ["failing"] if failing else []
|
|
entry["selfTest"] = str(ata.get("SmartSelftestStatus") or "")
|
|
|
|
health[serial] = entry
|
|
return health
|
|
|
|
|
|
def walk(node: dict, depth: int = 0):
|
|
yield node, depth
|
|
for child in node.get("children") or []:
|
|
yield from walk(child, depth + 1)
|
|
|
|
|
|
def mountpoints(node: dict) -> list[str]:
|
|
return [point for point in (node.get("mountpoints") or []) if point and point != "[SWAP]"]
|
|
|
|
|
|
def snapshot() -> dict:
|
|
tree = lsblk()
|
|
health = drive_health(udisks_objects())
|
|
|
|
drives = []
|
|
swap = []
|
|
filesystems: dict[str, dict] = {}
|
|
|
|
for root in tree:
|
|
if root.get("type") != "disk":
|
|
continue
|
|
|
|
# zram is compressed swap in RAM. It is a block device and it is not
|
|
# storage; listing it as a drive would be actively misleading about how
|
|
# much of this machine is disk.
|
|
if str(root.get("name", "")).startswith("zram"):
|
|
swap.append({
|
|
"name": root.get("name"),
|
|
"sizeBytes": root.get("size") or 0,
|
|
"kind": "zram",
|
|
})
|
|
continue
|
|
|
|
serial = str(root.get("serial") or "")
|
|
drive_health_entry = health.get(serial, {})
|
|
partitions = []
|
|
encrypted = False
|
|
|
|
for node, depth in walk(root):
|
|
if depth == 0:
|
|
continue
|
|
if node.get("fstype") == "crypto_LUKS":
|
|
encrypted = True
|
|
if node.get("type") in ("part", "crypt"):
|
|
partitions.append({
|
|
"name": node.get("name"),
|
|
"path": node.get("path"),
|
|
"sizeBytes": node.get("size") or 0,
|
|
"fstype": node.get("fstype") or "",
|
|
"type": node.get("type"),
|
|
"mountpoints": mountpoints(node),
|
|
})
|
|
for point in mountpoints(node):
|
|
if node.get("fssize") is None:
|
|
continue
|
|
# Keyed by device: btrfs subvolumes mounted at / and /home are
|
|
# ONE filesystem with one pool of free space. Reporting them as
|
|
# two independent bars, which is what df does, doubles the free
|
|
# space on screen.
|
|
key = str(node.get("path"))
|
|
entry = filesystems.setdefault(key, {
|
|
"device": node.get("path"),
|
|
"fstype": node.get("fstype") or "",
|
|
"sizeBytes": int(node.get("fssize") or 0),
|
|
"usedBytes": int(node.get("fsused") or 0),
|
|
"availBytes": int(node.get("fsavail") or 0),
|
|
"mountpoints": [],
|
|
"encrypted": node.get("type") == "crypt",
|
|
})
|
|
entry["mountpoints"].append(point)
|
|
entry["mountpoints"].sort(key=lambda mount: (mount != "/", mount))
|
|
|
|
drives.append({
|
|
"name": root.get("name"),
|
|
"path": root.get("path"),
|
|
"model": (root.get("model") or "").strip() or str(root.get("name")),
|
|
"serial": serial,
|
|
"sizeBytes": root.get("size") or 0,
|
|
"rotational": bool(root.get("rota")),
|
|
"removable": bool(root.get("rm")) or bool(root.get("hotplug"))
|
|
or bool(drive_health_entry.get("removable")),
|
|
"ejectable": bool(drive_health_entry.get("ejectable")),
|
|
"encrypted": encrypted,
|
|
"partitions": partitions,
|
|
"temperatureC": drive_health_entry.get("temperatureC"),
|
|
"powerOnHours": drive_health_entry.get("powerOnHours"),
|
|
"healthy": drive_health_entry.get("healthy"),
|
|
"warnings": drive_health_entry.get("warnings", []),
|
|
"selfTest": drive_health_entry.get("selfTest", ""),
|
|
})
|
|
|
|
ordered = sorted(filesystems.values(),
|
|
key=lambda entry: (0 if "/" in entry["mountpoints"] else 1,
|
|
entry["mountpoints"][0] if entry["mountpoints"] else ""))
|
|
return {"drives": drives, "filesystems": ordered, "swap": swap}
|
|
|
|
|
|
def measure(path: Path, budget: float) -> tuple[int | None, float]:
|
|
"""Bytes used by a folder, and what is left of the time budget."""
|
|
if not path.is_dir():
|
|
return None, budget
|
|
started = time.monotonic()
|
|
try:
|
|
completed = subprocess.run(["du", "-sxb", str(path)], check=False,
|
|
capture_output=True, text=True, timeout=budget)
|
|
except (OSError, subprocess.TimeoutExpired):
|
|
# A folder that ran out of time has consumed the whole budget by
|
|
# definition; the caller stops rather than starting another walk.
|
|
return None, 0.0
|
|
left = max(budget - (time.monotonic() - started), 0.0)
|
|
if completed.returncode != 0:
|
|
return None, left
|
|
match = re.match(r"^(\d+)", completed.stdout)
|
|
return (int(match.group(1)) if match else None), left
|
|
|
|
|
|
def container_reclaimable() -> dict | None:
|
|
"""Container images nothing is using. Absent when podman is not installed."""
|
|
if not shutil.which("podman"):
|
|
return None
|
|
try:
|
|
raw = run(["podman", "system", "df", "--format", "json"], timeout=20.0)
|
|
entries = json.loads(raw)
|
|
except (BoundaryError, json.JSONDecodeError):
|
|
return None
|
|
for entry in entries if isinstance(entries, list) else []:
|
|
if not str(entry.get("Type", "")).lower().startswith("image"):
|
|
continue
|
|
# Raw* are integers; Size and Reclaimable are display strings like
|
|
# "10.8GB" and "5.31GB (49%)". Reading the display strings is how this
|
|
# first crashed, so only the raw fields are trusted, and an old podman
|
|
# that lacks them reports nothing rather than a wrong number.
|
|
total = entry.get("RawSize")
|
|
reclaimable = entry.get("RawReclaimable")
|
|
if not isinstance(total, int) or not isinstance(reclaimable, int):
|
|
return None
|
|
return {"totalBytes": total, "reclaimableBytes": reclaimable}
|
|
return None
|
|
|
|
|
|
def scan() -> dict:
|
|
folders = []
|
|
remaining = float(SCAN_TIMEOUT_SECONDS)
|
|
truncated = False
|
|
for label, target in SCAN_TARGETS:
|
|
path = Path(os.path.expanduser(target))
|
|
if remaining <= 1.0:
|
|
truncated = True
|
|
break
|
|
size, remaining = measure(path, remaining)
|
|
if size is None:
|
|
continue
|
|
folders.append({"label": label, "path": str(path), "bytes": size})
|
|
folders.sort(key=lambda item: item["bytes"], reverse=True)
|
|
return {
|
|
"folders": folders,
|
|
"truncated": truncated,
|
|
"containers": container_reclaimable(),
|
|
}
|
|
|
|
|
|
# ── What the used space is made of ───────────────────────────────────────────
|
|
#
|
|
# Four measured segments and one remainder. The remainder is what is left of the
|
|
# filesystem's used bytes after the four are subtracted, and it is labelled
|
|
# "System & everything else" on the page for exactly that reason: it is not a
|
|
# measurement of the system, it is everything this did not measure.
|
|
#
|
|
# The arithmetic rule, which a contract pins: the measured segments never sum to
|
|
# more than the filesystem reports as used, and nothing is scaled to make a bar
|
|
# look tidy. When a measurement does overshoot -- possible when /home lives on a
|
|
# different filesystem from the flatpak installation -- the remainder is zero and
|
|
# `exceedsUsed` says so rather than inventing a number.
|
|
|
|
|
|
def measure_all(paths: list[str], budget: float) -> tuple[int, float, bool]:
|
|
"""Bytes across several paths, and whether every one of them was measured."""
|
|
total = 0
|
|
complete = True
|
|
for target in paths:
|
|
if budget <= 1.0:
|
|
return total, budget, False
|
|
size, budget = measure(Path(os.path.expanduser(target)), budget)
|
|
if size is None:
|
|
# A path that does not exist contributes nothing and is not a gap;
|
|
# one that timed out is, and the budget is gone either way.
|
|
if Path(os.path.expanduser(target)).is_dir():
|
|
complete = False
|
|
continue
|
|
total += size
|
|
return total, budget, complete
|
|
|
|
|
|
def backing_device(path: str) -> str:
|
|
"""The block device behind a path, with any btrfs subvolume stripped.
|
|
|
|
st_dev is not the question being asked. btrfs hands every subvolume its own
|
|
device number, so / and /home compare as different filesystems by that test
|
|
even though they are one pool with one free-space total -- which is the
|
|
exact confusion the filesystems list upstairs already exists to avoid. The
|
|
first version of the breakdown left the system-wide flatpak installation out
|
|
of the applications segment for that reason, and reported 4 kB of apps on a
|
|
machine with twenty gigabytes of them.
|
|
"""
|
|
if not shutil.which("findmnt"):
|
|
try:
|
|
return str(os.stat(path).st_dev)
|
|
except OSError:
|
|
return ""
|
|
try:
|
|
source = run(["findmnt", "-n", "-o", "SOURCE", "--target", path], timeout=10.0)
|
|
except BoundaryError:
|
|
return ""
|
|
return source.strip().split("[", 1)[0]
|
|
|
|
|
|
def same_filesystem(first: str, second: str) -> bool:
|
|
left = backing_device(first)
|
|
return left != "" and left == backing_device(second)
|
|
|
|
|
|
def breakdown() -> dict:
|
|
home = os.path.expanduser("~")
|
|
try:
|
|
usage = shutil.disk_usage(home)
|
|
except OSError as error:
|
|
raise BoundaryError("The filesystem holding your home folder could not be read.") from error
|
|
|
|
remaining = float(SCAN_TIMEOUT_SECONDS)
|
|
caches, remaining, caches_complete = measure_all([CACHE_TARGET], remaining)
|
|
|
|
home_targets = [target for _, target in SCAN_TARGETS if target != CACHE_TARGET]
|
|
home_bytes, remaining, home_complete = measure_all(home_targets, remaining)
|
|
|
|
# Only the installations that live on the same filesystem as home, because
|
|
# adding bytes from another drive into this drive's bar is a lie about this
|
|
# drive.
|
|
flatpak_paths = [target for target in FLATPAK_ROOTS
|
|
if Path(os.path.expanduser(target)).is_dir()
|
|
and same_filesystem(os.path.expanduser(target), home)]
|
|
applications, remaining, applications_complete = measure_all(flatpak_paths, remaining)
|
|
|
|
used = int(usage.used)
|
|
accounted = home_bytes + applications + caches
|
|
system = max(0, used - accounted)
|
|
|
|
return {
|
|
"segments": {
|
|
"home": home_bytes,
|
|
"applications": applications,
|
|
"caches": caches,
|
|
"system": system,
|
|
"free": int(usage.free),
|
|
},
|
|
"totalBytes": int(usage.total),
|
|
"usedBytes": used,
|
|
"freeBytes": int(usage.free),
|
|
# The measured segments are floors when this is false: something took
|
|
# longer than the budget and was left out rather than guessed at.
|
|
"complete": home_complete and caches_complete and applications_complete,
|
|
"exceedsUsed": accounted > used,
|
|
"path": home,
|
|
}
|
|
|
|
|
|
# ── Cleaning up, honestly ────────────────────────────────────────────────────
|
|
#
|
|
# Every row is itemized, sized in real bytes, and inert until its own id is
|
|
# passed to `clean`. There is no "clean everything" verb and there is no
|
|
# recommendation: the page shows what each one costs you -- caches are rebuilt,
|
|
# first launches get slower -- and lets it be somebody's decision.
|
|
|
|
|
|
def applications_helper() -> str:
|
|
return os.environ.get("PANAMA_APPLICATIONS_HELPER") or str(
|
|
Path(__file__).resolve().parent / "panama-applications")
|
|
|
|
|
|
def unused_runtime_bytes() -> int:
|
|
"""What the flatpak helper reports as unused, in bytes.
|
|
|
|
Asked of the applications helper rather than reimplemented, so the number
|
|
shown here and the thing `clean` removes can never come from two different
|
|
ideas of "unused".
|
|
"""
|
|
if not shutil.which("flatpak"):
|
|
return 0
|
|
try:
|
|
raw = run([applications_helper(), "unused-runtimes"], timeout=60.0)
|
|
entries = json.loads(raw)
|
|
except (BoundaryError, json.JSONDecodeError):
|
|
return 0
|
|
return sum(int(entry.get("sizeBytes") or 0)
|
|
for entry in entries if isinstance(entry, dict))
|
|
|
|
|
|
def dnf_package_cache_paths() -> list[str]:
|
|
paths = []
|
|
for root in DNF_CACHE_ROOTS:
|
|
directory = Path(root)
|
|
if not directory.is_dir():
|
|
continue
|
|
try:
|
|
paths.extend(str(child / "packages") for child in directory.iterdir()
|
|
if (child / "packages").is_dir())
|
|
except OSError:
|
|
continue
|
|
return paths
|
|
|
|
|
|
def cleanables() -> list[dict]:
|
|
remaining = float(SCAN_TIMEOUT_SECONDS)
|
|
cache_bytes, remaining, _ = measure_all([CACHE_TARGET], remaining)
|
|
trash_bytes, remaining, _ = measure_all([TRASH_TARGET], remaining)
|
|
dnf_bytes, remaining, _ = measure_all(dnf_package_cache_paths(), remaining)
|
|
|
|
return [
|
|
{
|
|
"id": "cache",
|
|
"label": "Application caches",
|
|
"detail": "~/.cache · rebuilt as apps run · first launches get slower once",
|
|
"bytes": cache_bytes,
|
|
"privileged": False,
|
|
},
|
|
{
|
|
"id": "trash",
|
|
"label": "Trash",
|
|
"detail": "Files you deleted · emptying is permanent",
|
|
"bytes": trash_bytes,
|
|
"privileged": False,
|
|
},
|
|
{
|
|
"id": "flatpak-unused",
|
|
"label": "Unused Flatpak runtimes",
|
|
"detail": "Runtimes no installed app asks for · flatpak decides the final list",
|
|
"bytes": unused_runtime_bytes(),
|
|
"privileged": False,
|
|
},
|
|
{
|
|
"id": "dnf-cache",
|
|
"label": "Package download cache",
|
|
"detail": "Downloaded packages · the system will ask for your password",
|
|
"bytes": dnf_bytes,
|
|
"privileged": True,
|
|
},
|
|
]
|
|
|
|
|
|
def guarded_cache_directory() -> Path:
|
|
"""~/.cache, or a refusal.
|
|
|
|
This function is the whole reason the cache row is safe to press. It refuses
|
|
a symlinked ~/.cache and refuses anything that resolves outside the home
|
|
directory, so XDG_CACHE_HOME pointing somewhere alarming, or a ~/.cache
|
|
someone linked to /, cannot turn one click into a deleted system.
|
|
"""
|
|
home = Path(os.path.expanduser("~")).resolve(strict=False)
|
|
target = Path(os.path.expanduser(CACHE_TARGET))
|
|
if target.is_symlink():
|
|
raise BoundaryError("The cache folder is a link, so it will not be emptied.")
|
|
if not target.is_dir():
|
|
raise BoundaryError("There is no cache folder to empty.")
|
|
resolved = target.resolve(strict=True)
|
|
if resolved == home or home not in resolved.parents:
|
|
raise BoundaryError("The cache folder is not inside your home folder.")
|
|
return resolved
|
|
|
|
|
|
def empty_cache() -> None:
|
|
"""Delete what is inside ~/.cache, never following a link out of it.
|
|
|
|
A cache file an application still has open cannot be removed, and that is
|
|
the normal case rather than a failure -- so a partial pass succeeds, and the
|
|
freshly measured size the caller gets back is what says how much is left.
|
|
Only a pass that removed nothing at all is reported as a failure.
|
|
"""
|
|
directory = guarded_cache_directory()
|
|
removed = 0
|
|
failures = 0
|
|
with os.scandir(directory) as entries:
|
|
for entry in entries:
|
|
try:
|
|
# is_symlink first: a symlinked directory must be unlinked, not
|
|
# walked, or this deletes whatever it points at.
|
|
if entry.is_symlink() or not entry.is_dir(follow_symlinks=False):
|
|
os.unlink(entry.path)
|
|
else:
|
|
# rmtree lstats as it goes and refuses to descend a symlink.
|
|
shutil.rmtree(entry.path, ignore_errors=False)
|
|
removed += 1
|
|
except OSError:
|
|
failures += 1
|
|
if failures and not removed:
|
|
raise BoundaryError("The cache is in use and nothing could be removed.")
|
|
|
|
|
|
def empty_trash() -> None:
|
|
if not shutil.which("gio"):
|
|
raise BoundaryError("gio is not available, so the trash cannot be emptied.")
|
|
# gio rather than removing ~/.local/share/Trash by hand: the trash is a
|
|
# freedesktop structure with per-file metadata and mount-point trash
|
|
# directories elsewhere, and gio empties all of it correctly.
|
|
run(["gio", "trash", "--empty"], timeout=300.0)
|
|
|
|
|
|
def clean_flatpak_unused() -> None:
|
|
if not shutil.which("flatpak"):
|
|
raise BoundaryError("Flatpak is not installed on this machine.")
|
|
run([applications_helper(), "clean-unused"], timeout=600.0)
|
|
|
|
|
|
def clean_dnf_cache() -> None:
|
|
if not shutil.which("dnf"):
|
|
raise BoundaryError("dnf is not available on this machine.")
|
|
# `clean packages` and never `clean all`: this drops the downloaded rpms,
|
|
# which is what was measured and what takes the space. Dropping the metadata
|
|
# as well would free little and make the next install slow for no reason.
|
|
#
|
|
# This is the only dnf invocation in Panama's settings surface, and it
|
|
# removes downloads. Nothing here removes an installed package.
|
|
try:
|
|
run(["pkexec", "dnf", "clean", "packages"], timeout=300.0)
|
|
except BoundaryError as error:
|
|
detail = str(error).lower()
|
|
if "dismissed" in detail or "not authorized" in detail:
|
|
raise BoundaryError("That change was not authorized.") from error
|
|
raise
|
|
|
|
|
|
CLEANERS = {
|
|
"cache": empty_cache,
|
|
"trash": empty_trash,
|
|
"flatpak-unused": clean_flatpak_unused,
|
|
"dnf-cache": clean_dnf_cache,
|
|
}
|
|
|
|
|
|
def clean(identifier: str) -> list[dict]:
|
|
"""Free exactly one named thing, and refuse everything else.
|
|
|
|
One id per call, no list, no "all". The caller has to name what it wants
|
|
removed, which is what keeps a mis-wired button from emptying four things.
|
|
"""
|
|
cleaner = CLEANERS.get(identifier or "")
|
|
if cleaner is None:
|
|
raise BoundaryError("There is nothing by that name to clean up.")
|
|
cleaner()
|
|
return cleanables()
|
|
|
|
|
|
def removable_device(path: str) -> dict:
|
|
"""Resolve a device path, refusing anything that is not removable.
|
|
|
|
Unmounting the root filesystem is not a feature. The check is on the device
|
|
rather than the button because the caller is a UI that can be wrong.
|
|
"""
|
|
if not re.fullmatch(r"/dev/[A-Za-z0-9/_-]+", path or ""):
|
|
raise BoundaryError("That is not a device path.")
|
|
for root in lsblk():
|
|
for node, _ in walk(root):
|
|
if node.get("path") != path:
|
|
continue
|
|
top = root
|
|
if not (bool(top.get("rm")) or bool(top.get("hotplug"))):
|
|
raise BoundaryError("That drive is not removable.")
|
|
return node
|
|
raise BoundaryError("No such device.")
|
|
|
|
|
|
def main(arguments: list[str]) -> int:
|
|
try:
|
|
if arguments == ["snapshot"]:
|
|
print(json.dumps(snapshot(), separators=(",", ":")))
|
|
elif arguments == ["scan"]:
|
|
print(json.dumps(scan(), separators=(",", ":")))
|
|
elif arguments == ["breakdown"]:
|
|
print(json.dumps(breakdown(), separators=(",", ":")))
|
|
elif arguments == ["cleanables"]:
|
|
print(json.dumps(cleanables(), separators=(",", ":")))
|
|
elif len(arguments) == 2 and arguments[0] == "clean":
|
|
print(json.dumps(clean(arguments[1]), separators=(",", ":")))
|
|
elif len(arguments) == 2 and arguments[0] in ("unmount", "eject"):
|
|
removable_device(arguments[1])
|
|
action = "unmount" if arguments[0] == "unmount" else "power-off"
|
|
flag = "-b" if arguments[0] == "unmount" else "-b"
|
|
run(["udisksctl", action, flag, arguments[1]], timeout=30.0)
|
|
else:
|
|
raise BoundaryError(
|
|
"Usage: panama-disks snapshot | scan | breakdown | cleanables | "
|
|
"clean ID | unmount DEVICE | eject DEVICE")
|
|
except BoundaryError as error:
|
|
print(str(error), file=sys.stderr)
|
|
return 2
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main(sys.argv[1:]))
|