Panama/config/dot/quickshell/scripts/panama-containers

#!/usr/bin/env python3

"""Rootless containers, grouped by the project that defines them.

Every container on this machine is created by podman-compose and labelled with
the project it belongs to, so the grouping is read from the labels rather than
invented. Acting on a group is then done with plain `podman` over the labelled
set -- never `podman-compose down`, which would REMOVE the containers. Nothing
here creates, recreates, or removes a container: the compose file is the source
of truth for what exists, and it belongs to the repository, not to this tool.

Rootless throughout, so nothing here needs privilege.

The one exception to "does not touch the compose file" is `bind-local`, which
exists because a development database published on every interface is worth
closing and the fix is a single token. It prepends a loopback bind address and
leaves the rest of the line byte-for-byte -- variables, quoting and style
intact -- then re-parses to confirm only that value moved. Anything it cannot
read unambiguously it refuses rather than guesses.

    panama-containers snapshot
    panama-containers start NAME | stop NAME | restart NAME
    panama-containers project-start NAME | project-stop NAME | project-restart NAME
    panama-containers prune-images | prune-volumes
    panama-containers bind-local PROJECT SERVICE
"""

from __future__ import annotations

import json
import os
import re
import subprocess
import sys
from pathlib import Path

import yaml

# A container or project name as podman and compose accept them. Deliberately
# strict: these values reach an argv, and nothing legitimate needs more.
NAME = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.-]{0,127}$")

# Project names that say nothing, because podman-compose defaults to the
# directory holding the compose file. "docker" is a directory, not a project.
ANONYMOUS = {"docker", "compose", "containers", "container", "db", "dev", "local", "src"}

# Substituted before a port mapping is split, because ${POSTGRES_PORT:-5432}
# contains a colon and would otherwise be torn in half.
INTERPOLATION = re.compile(r"\$\{[^}]*\}")

# Host addresses that mean "every interface", and so mean reachable.
EVERY_INTERFACE = {"", "0.0.0.0", "::", "[::]", "*"}

LOOPBACK = "127.0.0.1"


class BoundaryError(RuntimeError):
    """A user-visible validation or podman failure."""


def podman_binary() -> str:
    """The podman to call.

    Overridable so that stopping a container and removing an image can be
    tested without stopping a real container or removing a real image. The
    containers on this machine are a working development database; a test suite
    has no business touching them.
    """
    return os.environ.get("PANAMA_CONTAINERS_PODMAN", "podman")


def run(command: list[str], timeout: float = 60.0) -> subprocess.CompletedProcess:
    try:
        return subprocess.run(command, capture_output=True, text=True, timeout=timeout)
    except FileNotFoundError as error:
        raise BoundaryError("podman is not installed.") from error
    except subprocess.TimeoutExpired as error:
        raise BoundaryError("podman did not respond.") from error


def podman_json(arguments: list[str], timeout: float = 60.0) -> list | dict:
    result = run([podman_binary(), *arguments], timeout=timeout)
    if result.returncode != 0:
        detail = (result.stderr or "").strip().splitlines()
        raise BoundaryError(detail[-1] if detail else "podman could not be read.")
    try:
        return json.loads(result.stdout or "[]")
    except json.JSONDecodeError as error:
        raise BoundaryError("podman returned something unreadable.") from error


def podman_do(arguments: list[str], failure: str, timeout: float = 120.0) -> None:
    result = run([podman_binary(), *arguments], timeout=timeout)
    if result.returncode != 0:
        detail = (result.stderr or "").strip().splitlines()
        raise BoundaryError(detail[-1] if detail else failure)


def require(pattern: re.Pattern[str], value: str, message: str) -> str:
    if not pattern.match(value or ""):
        raise BoundaryError(message)
    return value


# ── reading ──────────────────────────────────────────────────────────────────


def split_mapping(mapping: str) -> list[str]:
    """The parts of a compose port mapping, with interpolations kept whole."""
    masked: list[str] = []
    placeholder = "\x00{}\x00"
    def keep(match: re.Match[str]) -> str:
        masked.append(match.group(0))
        return placeholder.format(len(masked) - 1)
    stand_in = INTERPOLATION.sub(keep, mapping)
    return [
        re.sub(r"\x00(\d+)\x00", lambda m: masked[int(m.group(1))], part)
        for part in stand_in.split(":")
    ]


def health_of(status: str) -> str:
    """The health podman prints inside the status line, when it prints one."""
    match = re.search(r"\((healthy|unhealthy|starting)\)", status or "")
    return match.group(1) if match else ""


def containers() -> list[dict]:
    raw = podman_json(["ps", "-a", "--format", "json"])
    result: list[dict] = []
    for entry in raw if isinstance(raw, list) else []:
        labels = entry.get("Labels") or {}
        names = entry.get("Names") or []
        status = str(entry.get("Status") or "")
        ports = [
            {
                "hostIp": str(port.get("host_ip") or ""),
                "hostPort": int(port.get("host_port") or 0),
                "containerPort": int(port.get("container_port") or 0),
                "protocol": str(port.get("protocol") or "tcp"),
                "range": int(port.get("range") or 1),
            }
            for port in (entry.get("Ports") or [])
        ]
        result.append({
            "id": str(entry.get("Id") or "")[:12],
            "name": names[0] if names else str(entry.get("Id") or "")[:12],
            "image": str(entry.get("Image") or ""),
            "state": str(entry.get("State") or ""),
            "status": status,
            "health": health_of(status),
            "exitCode": int(entry.get("ExitCode") or 0),
            "startedAt": int(entry.get("StartedAt") or 0),
            "restarts": int(entry.get("Restarts") or 0),
            "project": str(labels.get("com.docker.compose.project") or ""),
            "service": str(labels.get("com.docker.compose.service") or ""),
            "configFile": str(labels.get("com.docker.compose.project.config_files") or ""),
            "workingDir": str(labels.get("com.docker.compose.project.working_dir") or ""),
            # Published ports whose host address is every interface.
            "ports": ports,
        })
    return result


def resolve_config(path: str, working_dir: str) -> str:
    """The compose file a label points at, normalised.

    podman-compose records the path as it was invoked, which is how a perfectly
    valid label ends up containing "/scripts/../compose.yml".
    """
    if not path:
        return ""
    first = path.split(",")[0].strip()
    if not first:
        return ""
    candidate = Path(first)
    if not candidate.is_absolute() and working_dir:
        candidate = Path(working_dir) / candidate
    try:
        return str(candidate.resolve(strict=False))
    except OSError:
        return str(candidate)


def repository_name(config_file: str) -> str:
    """The repository a compose file lives in, for naming a project sensibly."""
    if not config_file:
        return ""
    result = run(
        ["git", "-C", str(Path(config_file).parent), "rev-parse", "--show-toplevel"],
        timeout=10.0,
    )
    if result.returncode != 0:
        return ""
    return Path(result.stdout.strip()).name if result.stdout.strip() else ""


def display_name(project: str, config_file: str) -> str:
    """A project name worth showing.

    podman-compose names a project after the directory holding its compose
    file, so a database stack can end up called "docker". Where the name says
    nothing, the repository it lives in says more.
    """
    if project.lower() not in ANONYMOUS:
        return project
    return repository_name(config_file) or project


def projects_of(entries: list[dict]) -> list[dict]:
    """Containers grouped by the compose project that declares them."""
    grouped: dict[str, dict] = {}
    for container in entries:
        key = container["project"]
        if not key:
            continue
        group = grouped.get(key)
        if group is None:
            config_file = resolve_config(container["configFile"], container["workingDir"])
            group = grouped[key] = {
                "name": key,
                "title": display_name(key, config_file),
                "configFile": config_file,
                "workingDir": container["workingDir"],
                "containers": [],
            }
        group["containers"].append(container)

    result = []
    for group in grouped.values():
        group["containers"].sort(key=lambda c: (c["state"] != "running", c["name"]))
        group["running"] = sum(1 for c in group["containers"] if c["state"] == "running")
        group["total"] = len(group["containers"])
        result.append(group)

    # Whatever is running comes first; a project nobody is using can wait.
    result.sort(key=lambda g: (-g["running"], g["title"].lower()))
    return result


def exposures(entries: list[dict]) -> list[dict]:
    """Published ports any machine on the network can reach.

    A container is only reachable if it is running AND publishes on an address
    that is not loopback. A stopped container publishes nothing, whatever its
    compose file says -- so it is reported as something that WILL expose, not
    something that does.
    """
    found: list[dict] = []
    for container in entries:
        for port in container["ports"]:
            if port["hostIp"] not in EVERY_INTERFACE:
                continue
            if port["hostPort"] <= 0:
                continue
            found.append({
                "container": container["name"],
                "project": container["project"],
                "service": container["service"],
                "configFile": resolve_config(container["configFile"], container["workingDir"]),
                "image": container["image"],
                "hostPort": port["hostPort"],
                "containerPort": port["containerPort"],
                "protocol": port["protocol"],
                "running": container["state"] == "running",
            })
    found.sort(key=lambda e: (not e["running"], e["hostPort"]))
    return found


def disk() -> dict:
    """What the container store costs, and what of it nothing references."""
    usage = podman_json(["system", "df", "--format", "json"])
    totals = {
        str(row.get("Type") or ""): row
        for row in (usage if isinstance(usage, list) else [])
    }

    def raw(kind: str, field: str) -> int:
        return int((totals.get(kind) or {}).get(field) or 0)

    images = podman_json(["images", "--format", "json"])
    unused = []
    for image in images if isinstance(images, list) else []:
        if int(image.get("Containers") or 0) > 0:
            continue
        tags = image.get("Names") or image.get("RepoTags") or []
        unused.append({
            "id": str(image.get("Id") or "")[:12],
            "name": tags[0] if tags else "<untagged>",
            "size": int(image.get("Size") or 0),
        })
    unused.sort(key=lambda i: -i["size"])

    # Podman's own answer to "does anything reference this volume", rather
    # than MountCount, which is a runtime lock counter: it reads zero for a
    # volume that a running container has mounted this second, and using it
    # here would offer to delete a live database.
    volumes = podman_json(["volume", "ls", "--filter", "dangling=true", "--format", "json"])
    idle = [
        {"name": str(volume.get("Name") or "")}
        for volume in (volumes if isinstance(volumes, list) else [])
        if volume.get("Name")
    ]
    idle.sort(key=lambda v: v["name"])

    return {
        "imagesSize": raw("Images", "RawSize"),
        "imagesReclaimable": raw("Images", "RawReclaimable"),
        "containersSize": raw("Containers", "RawSize"),
        "volumesSize": raw("Local Volumes", "RawSize"),
        "volumesReclaimable": raw("Local Volumes", "RawReclaimable"),
        "unusedImages": unused,
        "unusedVolumes": idle,
    }


def snapshot() -> dict:
    entries = containers()
    grouped = projects_of(entries)
    return {
        "available": True,
        "projects": grouped,
        "loose": [c for c in entries if not c["project"]],
        "running": sum(1 for c in entries if c["state"] == "running"),
        "total": len(entries),
        "exposed": exposures(entries),
        "disk": disk(),
        "error": "",
    }


def unavailable(message: str) -> dict:
    return {
        "available": False, "projects": [], "loose": [], "running": 0, "total": 0,
        "exposed": [], "disk": {
            "imagesSize": 0, "imagesReclaimable": 0, "containersSize": 0,
            "volumesSize": 0, "volumesReclaimable": 0,
            "unusedImages": [], "unusedVolumes": [],
        },
        "error": message,
    }


# ── acting ───────────────────────────────────────────────────────────────────


def find_container(name: str) -> dict:
    for container in containers():
        if container["name"] == name or container["id"] == name:
            return container
    raise BoundaryError(f"There is no container called {name}.")


def find_project(name: str) -> dict:
    for project in projects_of(containers()):
        if project["name"] == name:
            return project
    raise BoundaryError(f"There is no project called {name}.")


def act_on_container(verb: str, name: str) -> None:
    container = find_container(name)
    if verb == "start" and container["state"] == "running":
        raise BoundaryError(f"{container['name']} is already running.")
    if verb == "stop" and container["state"] != "running":
        raise BoundaryError(f"{container['name']} is not running.")
    podman_do([verb, container["name"]], f"{container['name']} could not be {verb}ed.")


def act_on_project(verb: str, name: str) -> None:
    """The whole stack, one container at a time, with plain podman.

    Deliberately not `podman-compose down`: that removes containers, and this
    tool does not remove what the compose file created.
    """
    project = find_project(name)
    wanted = "running" if verb == "stop" else "not running"
    targets = [
        container["name"] for container in project["containers"]
        if (container["state"] == "running") == (wanted == "running")
    ] if verb != "restart" else [
        container["name"] for container in project["containers"]
        if container["state"] == "running"
    ]
    if not targets:
        raise BoundaryError(f"Nothing in {project['title']} needs to be {verb}ed.")

    failures: list[str] = []
    for target in targets:
        result = run([podman_binary(), verb, target], timeout=120.0)
        if result.returncode != 0:
            failures.append(target)
    if failures:
        raise BoundaryError(f"Could not {verb} {', '.join(failures)}.")


def prune_images() -> None:
    """Remove images nothing references.

    Scoped to exactly what the snapshot showed as unused, by id, so that an
    image which gained a container between the panel rendering and the button
    being pressed is not swept up by a blanket prune.
    """
    unused = disk()["unusedImages"]
    if not unused:
        raise BoundaryError("Every image is in use.")
    failures = []
    for image in unused:
        result = run([podman_binary(), "rmi", image["id"]], timeout=120.0)
        if result.returncode != 0:
            failures.append(image["name"])
    if failures:
        raise BoundaryError(f"Could not remove {len(failures)} image(s): {', '.join(failures[:3])}.")


def prune_volumes() -> None:
    idle = disk()["unusedVolumes"]
    if not idle:
        raise BoundaryError("Every volume is in use.")
    failures = []
    for volume in idle:
        result = run([podman_binary(), "volume", "rm", volume["name"]], timeout=120.0)
        if result.returncode != 0:
            failures.append(volume["name"])
    if failures:
        raise BoundaryError(f"Could not remove {len(failures)} volume(s): {', '.join(failures[:3])}.")


# ── the compose edit ─────────────────────────────────────────────────────────


def rewrite_mapping(mapping: str) -> str:
    """A published port bound to loopback, with everything else left alone.

    Only the address is added. The host port keeps whatever form it had --
    literal, ${VAR}, or ${VAR:-default} -- because rewriting it to the number
    podman happens to report today would silently delete the variable that lets
    the port be configured at all.
    """
    parts = split_mapping(mapping.strip())
    if len(parts) == 3:
        raise BoundaryError(f"{mapping} already names an address.")
    if len(parts) != 2:
        raise BoundaryError(f"{mapping} is not a mapping this can read.")
    return f"{LOOPBACK}:{parts[0]}:{parts[1]}"


def compose_ports(document: dict, service: str) -> list[str]:
    services = document.get("services")
    if not isinstance(services, dict) or service not in services:
        raise BoundaryError(f"{service} is not in that compose file.")
    definition = services.get(service)
    if not isinstance(definition, dict):
        raise BoundaryError(f"{service} is not readable in that compose file.")
    ports = definition.get("ports")
    if ports is None:
        raise BoundaryError(f"{service} does not publish any ports.")
    if not isinstance(ports, list) or not all(isinstance(p, str) for p in ports):
        raise BoundaryError(f"The ports of {service} are not in a form this can edit.")
    return ports


def bind_local(project_name: str, service: str) -> None:
    """Bind a service's published ports to loopback, in place.

    Read to understand, edit as text so the file keeps its comments, quoting
    and layout, then read back to confirm that exactly the intended values
    moved and nothing else did.
    """
    project = find_project(project_name)
    path = Path(project["configFile"])
    if not project["configFile"] or not path.is_file():
        raise BoundaryError("The compose file for that project could not be found.")

    original = path.read_text(encoding="utf-8")
    try:
        document = yaml.safe_load(original)
    except yaml.YAMLError as error:
        raise BoundaryError("That compose file could not be parsed.") from error
    if not isinstance(document, dict):
        raise BoundaryError("That compose file is not a mapping.")

    current = compose_ports(document, service)
    wanted = [rewrite_mapping(mapping) for mapping in current]

    # Replace each mapping where it is written, not the line it sits on, so
    # flow style, block style and inline comments all survive untouched. The
    # mapping text is searched for within the service's own span only.
    updated = original
    for before, after in zip(current, wanted):
        needle = before.strip()
        occurrences = updated.count(needle)
        if occurrences == 0:
            raise BoundaryError(f"Could not find {needle} in the compose file.")
        if occurrences > 1:
            raise BoundaryError(
                f"{needle} appears {occurrences} times in that file; "
                "it is not clear which one belongs to this service.")
        updated = updated.replace(needle, after)

    if updated == original:
        raise BoundaryError("That compose file already binds these ports to loopback.")

    path.write_text(updated, encoding="utf-8")

    # Read back. A change that was not applied, or that broke the document, is
    # worse than no change at all.
    try:
        reread = yaml.safe_load(path.read_text(encoding="utf-8"))
    except yaml.YAMLError as error:
        path.write_text(original, encoding="utf-8")
        raise BoundaryError("The edit would have broken that compose file; it was undone.") from error

    if not isinstance(reread, dict) or compose_ports(reread, service) != wanted:
        path.write_text(original, encoding="utf-8")
        raise BoundaryError("The edit did not take effect; it was undone.")

    # Everything except this service's ports must be identical.
    before_document = yaml.safe_load(original)
    before_document["services"][service]["ports"] = wanted
    if before_document != reread:
        path.write_text(original, encoding="utf-8")
        raise BoundaryError("The edit changed more than those ports; it was undone.")


def main(arguments: list[str]) -> int:
    try:
        if arguments == ["snapshot"]:
            print(json.dumps(snapshot(), separators=(",", ":")))
            return 0

        if len(arguments) == 2 and arguments[0] in ("start", "stop", "restart"):
            act_on_container(arguments[0], require(NAME, arguments[1], "That is not a container name."))
        elif len(arguments) == 2 and arguments[0] in ("project-start", "project-stop", "project-restart"):
            verb = arguments[0].split("-", 1)[1]
            act_on_project(verb, require(NAME, arguments[1], "That is not a project name."))
        elif arguments == ["prune-images"]:
            prune_images()
        elif arguments == ["prune-volumes"]:
            prune_volumes()
        elif len(arguments) == 3 and arguments[0] == "bind-local":
            bind_local(
                require(NAME, arguments[1], "That is not a project name."),
                require(NAME, arguments[2], "That is not a service name."),
            )
        else:
            raise BoundaryError(
                "Usage: panama-containers snapshot | start NAME | stop NAME | restart NAME | "
                "project-start NAME | project-stop NAME | project-restart NAME | "
                "prune-images | prune-volumes | bind-local PROJECT SERVICE")
    except BoundaryError as error:
        try:
            state = snapshot()
        except BoundaryError:
            state = unavailable("")
        state["error"] = str(error)
        print(json.dumps(state, separators=(",", ":")))
        return 0

    print(json.dumps(snapshot(), separators=(",", ":")))
    return 0


if __name__ == "__main__":
    raise SystemExit(main(sys.argv[1:]))