Make Applications a real app manager, and clean up storage without the racket
Claude-Session: https://claude.ai/code/session_01Ms2FbjQy31TVf3CEvQhGM8
This commit is contained in:
Gabriel Brown
+583
@@ -0,0 +1,583 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
"""What is installed, what Panama offers to install, and what a flatpak may do.
|
||||
|
||||
Three boundaries, deliberately separate:
|
||||
|
||||
flatpaks / permissions read-only questions about what is already here.
|
||||
catalog the same optional-application catalog `panama apps`
|
||||
and ./install read, parsed by the same rules.
|
||||
install / uninstall mutations, each one validated against the catalog or
|
||||
against what flatpak reports as installed.
|
||||
|
||||
Deliberately absent: removing dnf packages. A settings page that uninstalls
|
||||
system packages is one mis-click away from removing the compositor it is drawn
|
||||
by, and dnf's dependency resolution will happily take half the desktop with it.
|
||||
The Applications page says "installed by the system package manager" and names
|
||||
the command instead. There is no dnf removal path anywhere in this file, and the
|
||||
contract checks for one.
|
||||
|
||||
The catalog is a closed surface: `install` refuses any entry that is not in
|
||||
setup/packages/extras, so this helper can never become a way to install an
|
||||
arbitrary package by passing a different string.
|
||||
|
||||
panama-applications flatpaks
|
||||
panama-applications permissions APP_ID
|
||||
panama-applications uninstall APP_ID
|
||||
panama-applications unused-runtimes
|
||||
panama-applications clean-unused
|
||||
panama-applications catalog
|
||||
panama-applications install CATEGORY ENTRY_ID
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import configparser
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
# Flathub ids, dnf package names, and category file names. Everything that
|
||||
# reaches a command line is matched against one of these first.
|
||||
APP_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,254}$")
|
||||
PACKAGE_NAME = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._+-]{0,127}$")
|
||||
CATEGORY_NAME = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,63}$")
|
||||
|
||||
# flatpak prints sizes for people, not for programs: there is no machine-readable
|
||||
# column for the installed size in any released flatpak. The display string is
|
||||
# kept as-is for the page, and the parsed number is only ever used for the
|
||||
# storage breakdown, which says out loud that it is flatpak's own rounded figure.
|
||||
SIZE_UNITS = {
|
||||
"b": 1, "byte": 1, "bytes": 1,
|
||||
"kb": 10**3, "mb": 10**6, "gb": 10**9, "tb": 10**12,
|
||||
"kib": 1024, "mib": 1024**2, "gib": 1024**3, "tib": 1024**4,
|
||||
}
|
||||
SIZE = re.compile(r"^\s*([0-9]+(?:[.,][0-9]+)?)\s*([A-Za-z]+)\s*$")
|
||||
|
||||
# The remote every catalog flatpak comes from. install-packages adds it during
|
||||
# setup; naming it here keeps a machine with several remotes from resolving an
|
||||
# id against whichever one happens to be first.
|
||||
FLATHUB = "flathub"
|
||||
|
||||
|
||||
class BoundaryError(RuntimeError):
|
||||
"""A user-visible validation or command failure."""
|
||||
|
||||
|
||||
def run(command: list[str], timeout: float = 30.0) -> subprocess.CompletedProcess:
|
||||
try:
|
||||
return subprocess.run(command, capture_output=True, text=True,
|
||||
timeout=timeout, check=False)
|
||||
except subprocess.TimeoutExpired as error:
|
||||
raise BoundaryError(f"{command[0]} did not answer in time.") from error
|
||||
except OSError as error:
|
||||
raise BoundaryError(f"{command[0]} is not available.") from error
|
||||
|
||||
|
||||
def require_flatpak() -> None:
|
||||
if not shutil.which("flatpak"):
|
||||
raise BoundaryError("Flatpak is not installed on this machine.")
|
||||
|
||||
|
||||
def parse_size(text: str) -> int | None:
|
||||
"""flatpak's own size string as bytes, or nothing when it cannot be read.
|
||||
|
||||
Nothing is not zero. A size this cannot parse is reported as unmeasured so
|
||||
the storage breakdown can say so, rather than quietly shrinking the
|
||||
applications segment and inflating the remainder.
|
||||
"""
|
||||
match = SIZE.match(text or "")
|
||||
if not match:
|
||||
return None
|
||||
number, unit = match.groups()
|
||||
factor = SIZE_UNITS.get(unit.lower())
|
||||
if factor is None:
|
||||
return None
|
||||
try:
|
||||
return int(float(number.replace(",", ".")) * factor)
|
||||
except ValueError:
|
||||
return None
|
||||
|
||||
|
||||
def flatpak_rows(scope: list[str], columns: list[str]) -> list[list[str]]:
|
||||
"""Tab-separated `flatpak list` output as rows, or nothing when it fails."""
|
||||
result = run(["flatpak", "list", *scope, "--columns=" + ",".join(columns)])
|
||||
if result.returncode != 0:
|
||||
return []
|
||||
rows = []
|
||||
for line in result.stdout.splitlines():
|
||||
if not line.strip():
|
||||
continue
|
||||
fields = line.split("\t")
|
||||
# A column flatpak could not fill comes back empty rather than missing,
|
||||
# but a future flatpak that drops one should not throw an IndexError
|
||||
# across the whole page.
|
||||
fields += [""] * (len(columns) - len(fields))
|
||||
rows.append(fields[:len(columns)])
|
||||
return rows
|
||||
|
||||
|
||||
def flatpaks() -> list[dict]:
|
||||
require_flatpak()
|
||||
entries = []
|
||||
for app_id, name, size, origin in flatpak_rows(
|
||||
["--app"], ["application", "name", "size", "origin"]):
|
||||
entries.append({
|
||||
"id": app_id,
|
||||
"name": name or app_id,
|
||||
"size": size,
|
||||
"sizeBytes": parse_size(size),
|
||||
"origin": origin,
|
||||
})
|
||||
entries.sort(key=lambda entry: (entry["name"].casefold(), entry["id"]))
|
||||
return entries
|
||||
|
||||
|
||||
# ── Permissions ──────────────────────────────────────────────────────────────
|
||||
#
|
||||
# A flatpak's sandbox is described by a handful of keys whose values are
|
||||
# semicolon-separated tokens. The buckets below are the questions people
|
||||
# actually ask -- can it see my camera, can it read my files -- and everything
|
||||
# that does not land in a bucket still gets a line. A permission summary that
|
||||
# silently omits what it did not recognise is worse than no summary: it reads as
|
||||
# "this app asks for nothing else".
|
||||
|
||||
CONTEXT_KEYS = ("shared", "sockets", "devices", "filesystems", "features", "persistent")
|
||||
HOST_FILESYSTEMS = {"host", "host-os", "host-etc", "/"}
|
||||
HOME_FILESYSTEMS = {"home", "~", "~/"}
|
||||
DEVICE_LABELS = {
|
||||
"all": "every device",
|
||||
"dri": "graphics acceleration",
|
||||
"kvm": "virtual machines",
|
||||
"shm": "shared memory",
|
||||
"input": "input devices",
|
||||
"usb": "USB devices",
|
||||
}
|
||||
SOCKET_LABELS = {
|
||||
"wayland": "Wayland",
|
||||
"x11": "X11",
|
||||
"fallback-x11": "X11 when Wayland is unavailable",
|
||||
"session-bus": "the whole session bus",
|
||||
"system-bus": "the whole system bus",
|
||||
"ssh-auth": "your SSH agent",
|
||||
"gpg-agent": "your GPG agent",
|
||||
"cups": "printing",
|
||||
"pcsc": "smart cards",
|
||||
"inherit-wayland-socket": "an inherited Wayland socket",
|
||||
}
|
||||
|
||||
|
||||
def permission_sections(text: str) -> dict[str, dict[str, str]]:
|
||||
parser = configparser.RawConfigParser(strict=False)
|
||||
# Keys are lower-case already in the [Context] section, but bus names are
|
||||
# not, and lower-casing org.gnome.Software would make the raw payload lie.
|
||||
parser.optionxform = str
|
||||
try:
|
||||
parser.read_string(text)
|
||||
except configparser.Error as error:
|
||||
raise BoundaryError("That application's permissions could not be read.") from error
|
||||
return {section: dict(parser.items(section)) for section in parser.sections()}
|
||||
|
||||
|
||||
def tokens(value: str) -> list[str]:
|
||||
return [item.strip() for item in (value or "").split(";") if item.strip()]
|
||||
|
||||
|
||||
def base_path(entry: str) -> str:
|
||||
"""`xdg-download:create` names the path `xdg-download`."""
|
||||
return entry.split(":", 1)[0]
|
||||
|
||||
|
||||
def join_some(items: list[str], limit: int = 4) -> str:
|
||||
if len(items) <= limit:
|
||||
return ", ".join(items)
|
||||
return ", ".join(items[:limit]) + f" and {len(items) - limit} more"
|
||||
|
||||
|
||||
def permission_summary(sections: dict[str, dict[str, str]]) -> list[str]:
|
||||
context = sections.get("Context", {})
|
||||
shared = tokens(context.get("shared", ""))
|
||||
sockets = tokens(context.get("sockets", ""))
|
||||
devices = tokens(context.get("devices", ""))
|
||||
filesystems = tokens(context.get("filesystems", ""))
|
||||
|
||||
summary: list[str] = []
|
||||
host = [item for item in filesystems if base_path(item) in HOST_FILESYSTEMS]
|
||||
home = [item for item in filesystems if base_path(item) in HOME_FILESYSTEMS]
|
||||
|
||||
# Camera. Flatpak has no camera token: a webcam is reached through
|
||||
# `devices=all`, so that is what this reports, and it says why.
|
||||
if "all" in devices:
|
||||
summary.append("Camera — full device access reaches webcams")
|
||||
# Microphone. Likewise, audio is one permission in both directions.
|
||||
if "pulseaudio" in sockets:
|
||||
summary.append("Microphone — audio access records as well as plays")
|
||||
if host:
|
||||
summary.append("Full file system access")
|
||||
if home:
|
||||
summary.append("Home folder")
|
||||
if "network" in shared:
|
||||
summary.append("Network")
|
||||
if devices:
|
||||
summary.append("Devices: " + join_some(
|
||||
[DEVICE_LABELS.get(item, item) for item in devices]))
|
||||
|
||||
# Everything the buckets did not claim, said plainly rather than dropped.
|
||||
rest_shared = [item for item in shared if item != "network"]
|
||||
if rest_shared:
|
||||
summary.append("Also shares: " + join_some(rest_shared))
|
||||
rest_sockets = [item for item in sockets if item != "pulseaudio"]
|
||||
if rest_sockets:
|
||||
# Sorted by the words shown rather than by flatpak's order, so
|
||||
# "X11 when Wayland is unavailable" follows "X11" instead of leading it.
|
||||
summary.append("Talks to " + join_some(
|
||||
sorted(SOCKET_LABELS.get(item, item) for item in rest_sockets)))
|
||||
rest_files = [item for item in filesystems if item not in host and item not in home]
|
||||
if rest_files:
|
||||
summary.append("Other locations: " + join_some(rest_files))
|
||||
features = tokens(context.get("features", ""))
|
||||
if features:
|
||||
summary.append("Sandbox features: " + join_some(features))
|
||||
persistent = tokens(context.get("persistent", ""))
|
||||
if persistent:
|
||||
summary.append("Keeps files in " + join_some(persistent))
|
||||
|
||||
for key, value in context.items():
|
||||
if key in CONTEXT_KEYS:
|
||||
continue
|
||||
items = tokens(value)
|
||||
summary.append(f"Also requests {key}: " + (join_some(items) if items else str(value)))
|
||||
|
||||
for name, entries in sections.items():
|
||||
if name == "Context" or not entries:
|
||||
continue
|
||||
count = len(entries)
|
||||
thing = "setting" if count == 1 else "settings"
|
||||
summary.append(f"{name}: {count} {thing}")
|
||||
|
||||
if not summary:
|
||||
summary.append("Nothing beyond the sandbox defaults")
|
||||
return summary
|
||||
|
||||
|
||||
def permissions(app_id: str) -> dict:
|
||||
require_flatpak()
|
||||
if not APP_ID.fullmatch(app_id or ""):
|
||||
raise BoundaryError("That is not an application id.")
|
||||
result = run(["flatpak", "info", "--show-permissions", app_id])
|
||||
if result.returncode != 0:
|
||||
raise BoundaryError("That application is not installed.")
|
||||
sections = permission_sections(result.stdout)
|
||||
return {"id": app_id, "summary": permission_summary(sections), "raw": sections}
|
||||
|
||||
|
||||
# ── Mutations ────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def installed_app_ids() -> set[str]:
|
||||
return {row[0] for row in flatpak_rows(["--app"], ["application"]) if row[0]}
|
||||
|
||||
|
||||
def uninstall(app_id: str) -> list[dict]:
|
||||
require_flatpak()
|
||||
if not APP_ID.fullmatch(app_id or ""):
|
||||
raise BoundaryError("That is not an application id.")
|
||||
# --app so a mistyped id can never resolve to a runtime, and the id is
|
||||
# checked against what is actually installed rather than trusting the page.
|
||||
if app_id not in installed_app_ids():
|
||||
raise BoundaryError("That application is not installed.")
|
||||
result = run(["flatpak", "uninstall", "--app", "--noninteractive", app_id], timeout=300)
|
||||
if result.returncode != 0:
|
||||
raise BoundaryError(_refusal(result, "That application could not be removed."))
|
||||
return flatpaks()
|
||||
|
||||
|
||||
def unused_runtimes() -> list[dict]:
|
||||
"""Runtimes and extensions no installed application asks for.
|
||||
|
||||
This is an estimate, and the page says so: the removal itself is
|
||||
`flatpak uninstall --unused`, which recomputes the list with flatpak's own
|
||||
dependency graph. Showing a size before asking needs a number now, and
|
||||
flatpak offers no way to ask what --unused would do without doing it.
|
||||
"""
|
||||
require_flatpak()
|
||||
apps = flatpak_rows(["--app"], ["application", "runtime"])
|
||||
runtimes = flatpak_rows(["--runtime"], ["application", "ref", "branch", "size", "runtime"])
|
||||
|
||||
# Seeded with the apps, so an application's own extensions -- OBS's plugins
|
||||
# are installed as runtimes named com.obsproject.Studio.Plugin.* -- count as
|
||||
# used rather than as sixteen orphans.
|
||||
used: set[str] = {row[0] for row in apps if row[0]}
|
||||
used |= {row[1].split("/")[0] for row in apps if row[1]}
|
||||
used_refs: set[str] = {row[1] for row in apps if row[1]}
|
||||
|
||||
# An extension of something used is used, and a runtime's own runtime is
|
||||
# used. Repeated until nothing new appears, because the chain can be two or
|
||||
# three long (app -> Platform -> Platform.Locale).
|
||||
changed = True
|
||||
while changed:
|
||||
changed = False
|
||||
for app_id, ref, _branch, _size, runtime in runtimes:
|
||||
if not app_id or app_id in used:
|
||||
continue
|
||||
parent = next((name for name in used
|
||||
if app_id.startswith(name + ".")), None)
|
||||
if parent is not None or ref in used_refs:
|
||||
used.add(app_id)
|
||||
if runtime:
|
||||
used.add(runtime.split("/")[0])
|
||||
used_refs.add(runtime)
|
||||
changed = True
|
||||
|
||||
entries = []
|
||||
for app_id, ref, branch, size, _runtime in runtimes:
|
||||
if not app_id or app_id in used:
|
||||
continue
|
||||
entries.append({
|
||||
"id": app_id,
|
||||
"ref": ref,
|
||||
"branch": branch,
|
||||
"size": size,
|
||||
"sizeBytes": parse_size(size),
|
||||
})
|
||||
entries.sort(key=lambda entry: entry["sizeBytes"] or 0, reverse=True)
|
||||
return entries
|
||||
|
||||
|
||||
def clean_unused() -> list[dict]:
|
||||
require_flatpak()
|
||||
result = run(["flatpak", "uninstall", "--unused", "--noninteractive"], timeout=600)
|
||||
if result.returncode != 0:
|
||||
raise BoundaryError(_refusal(result, "The unused runtimes could not be removed."))
|
||||
return flatpaks()
|
||||
|
||||
|
||||
# ── The catalog ──────────────────────────────────────────────────────────────
|
||||
#
|
||||
# The rules below are setup/lib/extras-catalog's, restated in Python because a
|
||||
# QML page cannot source a bash library. They are pinned by a contract that
|
||||
# reads both, so a change to one that is not made to the other fails loudly
|
||||
# rather than producing a menu that installs something else.
|
||||
#
|
||||
# * everything from the first `#` onward is stripped, inline comments included
|
||||
# * blank and whitespace-only lines are skipped, after stripping
|
||||
# * a line beginning with whitespace continues the entry above it and is
|
||||
# installed with it, never listed on its own
|
||||
# * `target | label`, split on the first `|`, both sides trimmed
|
||||
# * with no label, a `flatpak:` id becomes its last dotted component and a dnf
|
||||
# package keeps its own name
|
||||
# * installed means `flatpak info <id>` for a flatpak and `rpm -q <name>`
|
||||
# otherwise -- here, membership in one listing of each rather than a process
|
||||
# per entry, which is the same question asked cheaply
|
||||
|
||||
|
||||
def extras_directory() -> Path:
|
||||
override = os.environ.get("PANAMA_EXTRAS_DIR")
|
||||
if override:
|
||||
return Path(override)
|
||||
return Path(__file__).resolve().parents[4] / "setup" / "packages" / "extras"
|
||||
|
||||
|
||||
def category_files() -> list[Path]:
|
||||
directory = extras_directory()
|
||||
if not directory.is_dir():
|
||||
raise BoundaryError("The application catalog is not available.")
|
||||
return sorted(path for path in directory.iterdir()
|
||||
if path.is_file() and CATEGORY_NAME.fullmatch(path.name))
|
||||
|
||||
|
||||
def category_file(name: str) -> Path:
|
||||
if not CATEGORY_NAME.fullmatch(name or ""):
|
||||
raise BoundaryError("That is not an application category.")
|
||||
for path in category_files():
|
||||
if path.name == name:
|
||||
return path
|
||||
raise BoundaryError("That is not an application category.")
|
||||
|
||||
|
||||
def catalog_lines(path: Path) -> list[str]:
|
||||
try:
|
||||
text = path.read_text(encoding="utf-8", errors="replace")
|
||||
except OSError as error:
|
||||
raise BoundaryError("The application catalog could not be read.") from error
|
||||
return [line.split("#", 1)[0] for line in text.splitlines()]
|
||||
|
||||
|
||||
def split_entry(line: str) -> tuple[str, str]:
|
||||
target, _, label = line.partition("|")
|
||||
target = target.strip()
|
||||
label = label.strip()
|
||||
if not label:
|
||||
label = target[len("flatpak:"):] if target.startswith("flatpak:") else target
|
||||
if target.startswith("flatpak:"):
|
||||
label = label.rsplit(".", 1)[-1]
|
||||
return target, label
|
||||
|
||||
|
||||
def catalog_entries(path: Path) -> list[tuple[str, str]]:
|
||||
entries = []
|
||||
for line in catalog_lines(path):
|
||||
if not line.strip() or line[:1].isspace():
|
||||
continue
|
||||
target, label = split_entry(line)
|
||||
if target:
|
||||
entries.append((target, label))
|
||||
return entries
|
||||
|
||||
|
||||
def catalog_targets(path: Path, wanted: str) -> list[str]:
|
||||
"""The entry itself, then the indented lines beneath it."""
|
||||
targets: list[str] = []
|
||||
seen = False
|
||||
for line in catalog_lines(path):
|
||||
if not line.strip():
|
||||
continue
|
||||
if line[:1].isspace():
|
||||
if seen:
|
||||
targets.append(line.split("|", 1)[0].strip())
|
||||
continue
|
||||
target, _ = split_entry(line)
|
||||
if target == wanted:
|
||||
seen = True
|
||||
targets.append(target)
|
||||
elif seen:
|
||||
break
|
||||
return [target for target in targets if target]
|
||||
|
||||
|
||||
def installed_sets() -> tuple[set[str], set[str]]:
|
||||
"""Every installed flatpak ref name, and every installed rpm name.
|
||||
|
||||
One listing each rather than `flatpak info` and `rpm -q` per entry: the
|
||||
catalog runs to a hundred entries and the page opens with it.
|
||||
"""
|
||||
flatpak_ids: set[str] = set()
|
||||
if shutil.which("flatpak"):
|
||||
flatpak_ids = {row[0] for row in flatpak_rows([], ["application"]) if row[0]}
|
||||
packages: set[str] = set()
|
||||
if shutil.which("rpm"):
|
||||
result = run(["rpm", "-qa", "--qf", "%{NAME}\\n"], timeout=60)
|
||||
if result.returncode == 0:
|
||||
packages = {line.strip() for line in result.stdout.splitlines() if line.strip()}
|
||||
return flatpak_ids, packages
|
||||
|
||||
|
||||
def entry_payload(target: str, label: str,
|
||||
flatpak_ids: set[str], packages: set[str]) -> dict:
|
||||
is_flatpak = target.startswith("flatpak:")
|
||||
ref = target[len("flatpak:"):] if is_flatpak else target
|
||||
return {
|
||||
# The id is the catalog line's target, verbatim, because that is what
|
||||
# `install` matches and what extras-catalog's own lookup takes.
|
||||
"id": target,
|
||||
"ref": ref,
|
||||
"label": label,
|
||||
"kind": "flatpak" if is_flatpak else "dnf",
|
||||
"installed": ref in (flatpak_ids if is_flatpak else packages),
|
||||
}
|
||||
|
||||
|
||||
def catalog() -> dict:
|
||||
flatpak_ids, packages = installed_sets()
|
||||
categories = []
|
||||
for path in category_files():
|
||||
entries = [entry_payload(target, label, flatpak_ids, packages)
|
||||
for target, label in catalog_entries(path)]
|
||||
categories.append({
|
||||
"name": path.name,
|
||||
# "gpu-compute" is a file name; "GPU compute" is a heading. The
|
||||
# page needs the second and the helper needs the first.
|
||||
"label": path.name.replace("-", " ").capitalize(),
|
||||
"entries": entries,
|
||||
})
|
||||
return {"categories": categories}
|
||||
|
||||
|
||||
def install(category: str, entry_id: str) -> dict:
|
||||
"""Install one catalog entry, and nothing that is not a catalog entry."""
|
||||
path = category_file(category)
|
||||
known = {target for target, _ in catalog_entries(path)}
|
||||
if entry_id not in known:
|
||||
raise BoundaryError("That application is not in the catalog.")
|
||||
|
||||
flatpak_targets: list[str] = []
|
||||
dnf_targets: list[str] = []
|
||||
for target in catalog_targets(path, entry_id):
|
||||
if target.startswith("flatpak:"):
|
||||
identifier = target[len("flatpak:"):]
|
||||
if not APP_ID.fullmatch(identifier):
|
||||
raise BoundaryError("That catalog entry names something unusable.")
|
||||
flatpak_targets.append(identifier)
|
||||
else:
|
||||
if not PACKAGE_NAME.fullmatch(target):
|
||||
raise BoundaryError("That catalog entry names something unusable.")
|
||||
dnf_targets.append(target)
|
||||
|
||||
if not flatpak_targets and not dnf_targets:
|
||||
raise BoundaryError("That catalog entry installs nothing.")
|
||||
|
||||
if flatpak_targets:
|
||||
require_flatpak()
|
||||
result = run(["flatpak", "install", "--noninteractive", FLATHUB, *flatpak_targets],
|
||||
timeout=1800)
|
||||
if result.returncode != 0:
|
||||
raise BoundaryError(_refusal(result, "That application could not be installed."))
|
||||
if dnf_targets:
|
||||
if not shutil.which("dnf"):
|
||||
raise BoundaryError("dnf is not available on this machine.")
|
||||
# pkexec rather than sudo: the desktop already runs a polkit agent, and
|
||||
# a settings page has no terminal to type a password into.
|
||||
result = run(["pkexec", "dnf", "install", "-y", *dnf_targets], timeout=1800)
|
||||
if result.returncode != 0:
|
||||
raise BoundaryError(_refusal(result, "That package could not be installed."))
|
||||
return catalog()
|
||||
|
||||
|
||||
def _refusal(result: subprocess.CompletedProcess, fallback: str) -> str:
|
||||
text = (result.stderr or result.stdout or "").strip().splitlines()
|
||||
if not text:
|
||||
return fallback
|
||||
last = text[-1].strip()
|
||||
lowered = last.lower()
|
||||
if "not authorized" in lowered or "dismissed" in lowered:
|
||||
return "That change was not authorized."
|
||||
return last[:200] or fallback
|
||||
|
||||
|
||||
def emit(payload) -> None:
|
||||
print(json.dumps(payload, separators=(",", ":")))
|
||||
|
||||
|
||||
def main(arguments: list[str]) -> int:
|
||||
try:
|
||||
if arguments == ["flatpaks"]:
|
||||
emit(flatpaks())
|
||||
elif arguments == ["catalog"]:
|
||||
emit(catalog())
|
||||
elif arguments == ["unused-runtimes"]:
|
||||
emit(unused_runtimes())
|
||||
elif arguments == ["clean-unused"]:
|
||||
emit(clean_unused())
|
||||
elif len(arguments) == 2 and arguments[0] == "permissions":
|
||||
emit(permissions(arguments[1]))
|
||||
elif len(arguments) == 2 and arguments[0] == "uninstall":
|
||||
emit(uninstall(arguments[1]))
|
||||
elif len(arguments) == 3 and arguments[0] == "install":
|
||||
emit(install(arguments[1], arguments[2]))
|
||||
else:
|
||||
raise BoundaryError(
|
||||
"Usage: panama-applications flatpaks | permissions APP_ID | "
|
||||
"uninstall APP_ID | unused-runtimes | clean-unused | catalog | "
|
||||
"install CATEGORY ENTRY_ID")
|
||||
except BoundaryError as error:
|
||||
print(str(error), file=sys.stderr)
|
||||
return 2
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main(sys.argv[1:]))
|
||||
@@ -57,6 +57,15 @@ ROLE_TARGETS = {
|
||||
}
|
||||
DESKTOP_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._+-]*\.desktop$")
|
||||
EXEC_CMD = re.compile(r"hl\.exec_cmd\(\s*(\"(?:\\.|[^\"\\])*\")\s*\)")
|
||||
MIME_TYPE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9!#$&^_.+-]{0,126}"
|
||||
r"/[A-Za-z0-9][A-Za-z0-9!#$&^_.+-]{0,126}$")
|
||||
|
||||
# A single-type override is a scalpel, not a browser: someone looking for "the
|
||||
# thing that opens .heic" wants a short answer, and a list of six hundred types
|
||||
# is not one. Anything past this is reported as truncated so the page can say
|
||||
# "narrow the search" rather than pretending these are all of them.
|
||||
TYPE_SEARCH_LIMIT = 20
|
||||
TYPE_CANDIDATE_LIMIT = 12
|
||||
|
||||
|
||||
class BoundaryError(RuntimeError):
|
||||
@@ -224,6 +233,182 @@ def set_default(role: str, desktop_id: str) -> None:
|
||||
run(["xdg-mime", "default", desktop_id, setting])
|
||||
|
||||
|
||||
# ── One file type at a time ──────────────────────────────────────────────────
|
||||
#
|
||||
# The roles above govern families, which is right nearly always and wrong
|
||||
# exactly when a family is too broad: SVG belongs in an editor while the rest of
|
||||
# the images belong in a viewer, and setting the whole "Images" role to the
|
||||
# editor is not what anyone wanted. These two verbs are the escape hatch.
|
||||
#
|
||||
# Searching is over the type name and its file extensions, and says so on the
|
||||
# page. Matching human descriptions would mean reading the whole shared-mime-info
|
||||
# database -- some thousands of small XML files -- to answer a keystroke.
|
||||
|
||||
|
||||
def mime_globs() -> dict[str, list[str]]:
|
||||
"""Extension patterns per type, from shared-mime-info's globs2.
|
||||
|
||||
Absent on a machine without shared-mime-info, which is not fatal: the search
|
||||
falls back to matching the type name, and every type still resolves.
|
||||
"""
|
||||
globs: dict[str, list[str]] = {}
|
||||
for root in xdg_data_roots():
|
||||
path = root / "mime" / "globs2"
|
||||
if not path.is_file():
|
||||
continue
|
||||
try:
|
||||
lines = path.read_text(encoding="utf-8", errors="replace").splitlines()
|
||||
except OSError:
|
||||
continue
|
||||
for line in lines:
|
||||
if line.startswith("#"):
|
||||
continue
|
||||
parts = line.split(":")
|
||||
# weight:type:glob, with optional trailing flags.
|
||||
if len(parts) < 3:
|
||||
continue
|
||||
mime, pattern = parts[1], parts[2]
|
||||
if not MIME_TYPE.fullmatch(mime):
|
||||
continue
|
||||
patterns = globs.setdefault(mime, [])
|
||||
if pattern not in patterns:
|
||||
patterns.append(pattern)
|
||||
return globs
|
||||
|
||||
|
||||
def mime_registrations() -> dict[str, list[str]]:
|
||||
"""Which installed applications declare which types.
|
||||
|
||||
Read from the desktop files themselves rather than mimeinfo.cache, because
|
||||
the cache is regenerated by update-desktop-database and is stale on exactly
|
||||
the machine where an application was just installed.
|
||||
"""
|
||||
registrations: dict[str, list[str]] = {}
|
||||
for desktop_id, path in discovered_desktop_files().items():
|
||||
try:
|
||||
values = parse_desktop_entry(path)
|
||||
except BoundaryError:
|
||||
continue
|
||||
if values.get("NoDisplay", "false").lower() == "true":
|
||||
continue
|
||||
for mime in values.get("MimeType", "").split(";"):
|
||||
mime = mime.strip()
|
||||
if not MIME_TYPE.fullmatch(mime):
|
||||
continue
|
||||
registered = registrations.setdefault(mime, [])
|
||||
if desktop_id not in registered:
|
||||
registered.append(desktop_id)
|
||||
return registrations
|
||||
|
||||
|
||||
def mime_label(mime: str) -> str:
|
||||
"""shared-mime-info's own description, or "" when it has none.
|
||||
|
||||
Read only for the handful of types a search actually returns.
|
||||
"""
|
||||
media, _, subtype = mime.partition("/")
|
||||
if not media or not subtype:
|
||||
return ""
|
||||
for root in xdg_data_roots():
|
||||
path = root / "mime" / media / f"{subtype}.xml"
|
||||
if not path.is_file():
|
||||
continue
|
||||
try:
|
||||
text = path.read_text(encoding="utf-8", errors="replace")
|
||||
except OSError:
|
||||
continue
|
||||
# The untranslated <comment> comes first; the xml:lang ones follow.
|
||||
found = re.search(r"<comment>([^<]*)</comment>", text)
|
||||
if found:
|
||||
return found.group(1).strip()
|
||||
return ""
|
||||
|
||||
|
||||
def application_names() -> dict[str, str]:
|
||||
names: dict[str, str] = {}
|
||||
for desktop_id, path in discovered_desktop_files().items():
|
||||
try:
|
||||
values = parse_desktop_entry(path)
|
||||
except BoundaryError:
|
||||
continue
|
||||
names[desktop_id] = values.get("Name", desktop_id[:-len(".desktop")])
|
||||
return names
|
||||
|
||||
|
||||
def current_handler(mime: str) -> str:
|
||||
completed = subprocess.run(["xdg-mime", "query", "default", mime],
|
||||
check=False, capture_output=True, text=True)
|
||||
if completed.returncode != 0:
|
||||
return ""
|
||||
output = completed.stdout.strip().splitlines()
|
||||
return output[0] if output else ""
|
||||
|
||||
|
||||
def search_types(query: str) -> dict[str, object]:
|
||||
needle = (query or "").strip().lower().lstrip(".")
|
||||
if len(needle) < 2:
|
||||
return {"query": query, "types": [], "truncated": False}
|
||||
|
||||
globs = mime_globs()
|
||||
registrations = mime_registrations()
|
||||
names = application_names()
|
||||
known = sorted(set(globs) | set(registrations))
|
||||
|
||||
def score(mime: str) -> tuple[int, str]:
|
||||
patterns = globs.get(mime, [])
|
||||
extensions = [pattern[2:].lower() for pattern in patterns
|
||||
if pattern.startswith("*.")]
|
||||
if needle in extensions:
|
||||
return (0, mime)
|
||||
if mime.lower() == needle or mime.lower().split("/")[-1] == needle:
|
||||
return (1, mime)
|
||||
return (2, mime)
|
||||
|
||||
matched = []
|
||||
for mime in known:
|
||||
patterns = globs.get(mime, [])
|
||||
haystack = " ".join([mime.lower(),
|
||||
*(pattern.lower() for pattern in patterns)])
|
||||
if needle in haystack:
|
||||
matched.append(mime)
|
||||
matched.sort(key=score)
|
||||
|
||||
truncated = len(matched) > TYPE_SEARCH_LIMIT
|
||||
types = []
|
||||
for mime in matched[:TYPE_SEARCH_LIMIT]:
|
||||
handler = current_handler(mime)
|
||||
candidates = list(registrations.get(mime, []))
|
||||
# The current handler belongs in the list even when it never declared
|
||||
# the type -- an override put it there, and a picker that cannot show
|
||||
# the answer it is displaying is a picker nobody trusts.
|
||||
if handler and handler not in candidates:
|
||||
candidates.insert(0, handler)
|
||||
candidates = candidates[:TYPE_CANDIDATE_LIMIT]
|
||||
types.append({
|
||||
"mime": mime,
|
||||
"label": mime_label(mime),
|
||||
"extensions": [pattern[1:] for pattern in globs.get(mime, [])
|
||||
if pattern.startswith("*.")][:6],
|
||||
"handler": handler,
|
||||
"handlerName": names.get(handler, ""),
|
||||
"candidates": [{"id": desktop_id, "name": names.get(desktop_id, desktop_id)}
|
||||
for desktop_id in candidates],
|
||||
})
|
||||
return {"query": query, "types": types, "truncated": truncated}
|
||||
|
||||
|
||||
def set_type(mime: str, desktop_id: str) -> None:
|
||||
"""Point one type at one application, leaving its family alone."""
|
||||
if not MIME_TYPE.fullmatch(mime or ""):
|
||||
raise BoundaryError("That is not a file type.")
|
||||
globs = mime_globs()
|
||||
registrations = mime_registrations()
|
||||
if mime not in globs and mime not in registrations:
|
||||
raise BoundaryError("This system does not know that file type.")
|
||||
require_desktop_id(desktop_id, discovered=discovered_desktop_ids())
|
||||
run(["xdg-mime", "default", desktop_id, mime])
|
||||
|
||||
|
||||
# What this desktop opens a file with, when nobody has said otherwise.
|
||||
#
|
||||
# Applications register themselves for every type they can technically read, so
|
||||
@@ -442,10 +627,15 @@ def main(arguments: list[str]) -> int:
|
||||
remove_autostart(arguments[1])
|
||||
elif len(arguments) == 2 and arguments[0] == "add-autostart":
|
||||
add_autostart(arguments[1])
|
||||
elif len(arguments) == 2 and arguments[0] == "search-types":
|
||||
print(json.dumps(search_types(arguments[1]), separators=(",", ":")))
|
||||
elif len(arguments) == 3 and arguments[0] == "set-type":
|
||||
set_type(arguments[1], arguments[2])
|
||||
else:
|
||||
raise BoundaryError(
|
||||
"Usage: panama-default-apps snapshot | seed | set-default ROLE DESKTOP_ID | "
|
||||
"set-autostart DESKTOP_ID true|false | add-autostart DESKTOP_ID"
|
||||
"set-autostart DESKTOP_ID true|false | add-autostart DESKTOP_ID | "
|
||||
"remove-autostart DESKTOP_ID | search-types QUERY | set-type MIME DESKTOP_ID"
|
||||
)
|
||||
except BoundaryError as error:
|
||||
print(str(error), file=sys.stderr)
|
||||
|
||||
@@ -10,11 +10,22 @@ Two boundaries, deliberately separate:
|
||||
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
|
||||
@@ -47,6 +58,22 @@ SCAN_TARGETS = [
|
||||
("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."""
|
||||
@@ -329,6 +356,295 @@ def scan() -> dict:
|
||||
}
|
||||
|
||||
|
||||
# ── 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.
|
||||
|
||||
@@ -354,6 +670,12 @@ def main(arguments: list[str]) -> int:
|
||||
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"
|
||||
@@ -361,7 +683,8 @@ def main(arguments: list[str]) -> int:
|
||||
run(["udisksctl", action, flag, arguments[1]], timeout=30.0)
|
||||
else:
|
||||
raise BoundaryError(
|
||||
"Usage: panama-disks snapshot | scan | unmount DEVICE | eject DEVICE")
|
||||
"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
|
||||
|
||||
@@ -308,11 +308,21 @@ def delete(config: str, number: str) -> None:
|
||||
raise BoundaryError(_refusal(result, "That snapshot could not be removed."))
|
||||
|
||||
|
||||
# The three horizons the page can edit, and the largest number it will accept
|
||||
# for any of them. 999 hourly snapshots is not a retention policy, it is a typo
|
||||
# that fills a drive; the page offers a dropdown and this is its ceiling. The
|
||||
# monthly and yearly limits are left exactly as snapper has them -- nothing here
|
||||
# writes them, so a config with longer horizons keeps them.
|
||||
RETENTION_LIMIT = 50
|
||||
|
||||
|
||||
def set_retention(config: str, hourly: str, daily: str, weekly: str) -> None:
|
||||
values = []
|
||||
for label, value in (("HOURLY", hourly), ("DAILY", daily), ("WEEKLY", weekly)):
|
||||
if not str(value).isdigit() or int(value) > 999:
|
||||
if not str(value).isdigit():
|
||||
raise BoundaryError("Keep counts must be whole numbers.")
|
||||
if int(value) > RETENTION_LIMIT:
|
||||
raise BoundaryError(f"Keep counts go up to {RETENTION_LIMIT}.")
|
||||
values.append(f"TIMELINE_LIMIT_{label}={int(value)}")
|
||||
result = run(["snapper", "-c", require_config(config), "set-config", *values])
|
||||
if result.returncode != 0:
|
||||
|
||||
Reference in New Issue
Block a user