#!/usr/bin/env python3
# Copyright (c) 2026 Petr Balvín <opensource@petrbalvin.org> (https://petrbalvin.org)
# SPDX-License-Identifier: MIT

"""Idempotent system cleanup and optimisation for Fedora, CentOS Stream and openEuler.

Cleans old kernels, DNF caches, systemd journals, temp files, and core dumps.
Every operation checks current state before acting; running the script
repeatedly produces the same result with no errors and no duplicate work.
ostree-based (atomic) systems are refused — rpm-ostree manages those.

Usage:
    system-optimise.py                  # full cleanup
    system-optimise.py --dry-run        # preview without changes
    system-optimise.py --skip-dnf       # skip DNF cleanup
    system-optimise.py --skip-journal   # skip journal cleanup
    system-optimise.py --skip-tmp       # skip temp file cleanup
    system-optimise.py --skip-cores     # skip core dump cleanup
    system-optimise.py --version
"""

from __future__ import annotations

import argparse
import json
import os
import re
import subprocess
import sys
import time
from functools import cmp_to_key
from typing import Any, Callable

BOLD = "\033[1m"
RED = "\033[31m"
GREEN = "\033[32m"
YELLOW = "\033[33m"
DIM = "\033[2m"
RESET = "\033[0m"
VERSION = "1.4.0"

SUPPORTED_OS: dict[str, str] = {
    "fedora": "Fedora",
    "centos": "CentOS Stream",
    "openeuler": "openEuler",
}

# Kernel retention: keep the newest INSTALLONLY_LIMIT kernels plus the running
# one. dnf forbids installonly_limit=1 — a bootable fallback must always remain.
INSTALLONLY_LIMIT = 2

# Temp file retention (matches systemd-tmpfiles defaults on both targets).
TMP_MAX_AGE_DAYS = 10
VARTMP_MAX_AGE_DAYS = 30

# Core dumps younger than this are kept: a crash under active analysis must
# never be swept away by a cleanup run.
COREDUMP_MAX_AGE_DAYS = 7

JOURNAL_VACUUM_SIZE_MB = 500
JOURNAL_WARN_BYTES = 1024 * 1024 * 1024  # warn when the journal stays above 1 GiB
JOURNAL_DIRS = ("/var/log/journal", "/run/log/journal")

# Unit file types audited for leftovers in /etc/systemd/system/.
UNIT_SUFFIXES = (
    ".service",
    ".socket",
    ".timer",
    ".target",
    ".path",
    ".mount",
    ".automount",
    ".slice",
)


# ---------------------------------------------------------------------------
# Progress helpers — write to stderr so stdout stays clean
# ---------------------------------------------------------------------------


def _status(msg: str) -> None:
    """Print a progress message to stderr — stays on the same line until ``_status_done``."""
    print(f"  {msg}...", end="", flush=True, file=sys.stderr)


def _status_done(msg: str = "done") -> None:
    """Complete the current progress line on stderr."""
    print(f" {msg}", file=sys.stderr)


def _info(msg: str) -> None:
    """Print an informational line to stderr."""
    print(f"  {DIM}{msg}{RESET}", file=sys.stderr)


def _warn(msg: str) -> None:
    """Print a warning to stderr."""
    print(f"  {YELLOW}⚠ {msg}{RESET}", file=sys.stderr)


def _ok(msg: str) -> None:
    """Print a success message to stderr."""
    print(f"  {GREEN}✓ {msg}{RESET}", file=sys.stderr)


def _fail(msg: str) -> None:
    """Print a failure message to stderr."""
    print(f"  {RED}✗ {msg}{RESET}", file=sys.stderr)


# ---------------------------------------------------------------------------
# Low-level command runner
# ---------------------------------------------------------------------------


def run(cmd: list[str], timeout: int = 120) -> subprocess.CompletedProcess[str]:
    """Execute a command and return the CompletedProcess.

    LC_ALL=C is forced for reliable English-language output parsing.
    Missing binaries, timeouts, and exec failures are converted into a
    synthetic failed CompletedProcess (rc 127 / 124 / 126) so callers only
    ever need to check returncode — exceptions never escape this function.
    """
    env = {**os.environ, "LANG": "C", "LC_ALL": "C"}
    try:
        return subprocess.run(cmd, capture_output=True, text=True, timeout=timeout, env=env)
    except FileNotFoundError:
        return subprocess.CompletedProcess(cmd, 127, "", f"command not found: {cmd[0]}")
    except subprocess.TimeoutExpired:
        return subprocess.CompletedProcess(cmd, 124, "", f"timed out after {timeout}s")
    except OSError as exc:
        return subprocess.CompletedProcess(cmd, 126, "", f"cannot execute {cmd[0]}: {exc}")


# ---------------------------------------------------------------------------
# OS detection
# ---------------------------------------------------------------------------


def parse_os_release() -> dict[str, str]:
    """Parse /etc/os-release into a dict of KEY=VALUE pairs."""
    result: dict[str, str] = {}
    try:
        with open("/etc/os-release", encoding="utf-8") as fh:
            for line in fh:
                line = line.strip()
                if not line or line.startswith("#"):
                    continue
                if "=" not in line:
                    continue
                key, _, value = line.partition("=")
                value = value.strip().strip('"').strip("'")
                result[key] = value
    except OSError:
        pass
    return result


def detect_os() -> tuple[str, str, str]:
    """Detect the operating system.

    Returns a tuple of (os_id, display_name, version_id).
    Exits with an error message when the OS is not supported.
    """
    release = parse_os_release()
    os_id = release.get("ID", "").lower()
    version_id = release.get("VERSION_ID", "unknown")

    # ostree-based (atomic) systems report a normal ID (e.g. Silverblue is
    # ID=fedora) but have no dnf and manage kernels via rpm-ostree — refuse.
    if os.path.exists("/run/ostree-booted"):
        print(
            f"{RED}{BOLD}Error:{RESET} ostree-based (atomic) system detected "
            "(/run/ostree-booted present).",
            file=sys.stderr,
        )
        print(
            "  Package and kernel management there belongs to rpm-ostree; "
            "this script only supports traditional dnf/rpm systems.",
            file=sys.stderr,
        )
        sys.exit(1)

    if os_id not in SUPPORTED_OS:
        print(
            f"{RED}{BOLD}Error:{RESET} Unsupported operating system: "
            f"'{release.get('ID', 'unknown')}' (detected from /etc/os-release).",
            file=sys.stderr,
        )
        print(
            f"  Supported systems: {', '.join(SUPPORTED_OS.values())}",
            file=sys.stderr,
        )
        sys.exit(1)

    return os_id, SUPPORTED_OS[os_id], version_id


# ---------------------------------------------------------------------------
# Root check
# ---------------------------------------------------------------------------


def check_root() -> None:
    """Exit with a clear message if not running as root (UID 0)."""
    if os.geteuid() != 0:
        print(
            f"{RED}{BOLD}Error:{RESET} This script must be run as root (UID 0).",
            file=sys.stderr,
        )
        print(
            f"  Current UID: {os.geteuid()}. Try: sudo python3 system-optimise.py",
            file=sys.stderr,
        )
        sys.exit(1)


# ---------------------------------------------------------------------------
# Utility helpers
# ---------------------------------------------------------------------------


def _format_size(size_bytes: int) -> str:
    """Format a byte count into a human-readable string (MB or GB)."""
    if size_bytes < 0:
        return "0 B"
    if size_bytes >= 1024 * 1024 * 1024:
        return f"{size_bytes / (1024 * 1024 * 1024):.2f} GB"
    if size_bytes >= 1024 * 1024:
        return f"{size_bytes / (1024 * 1024):.1f} MB"
    if size_bytes >= 1024:
        return f"{size_bytes / 1024:.1f} KB"
    return f"{size_bytes} B"


def _du(path: str) -> int:
    """Return the total size (in bytes) of all files under *path*.

    Returns 0 if the path does not exist.
    """
    if not os.path.exists(path):
        return 0
    total: int = 0
    for dirpath, _dirnames, filenames in os.walk(path):
        for fname in filenames:
            fpath = os.path.join(dirpath, fname)
            try:
                total += os.path.getsize(fpath)
            except OSError:
                pass
    return total


def _last_lines(text: str, count: int = 3) -> str:
    """Return the last *count* non-empty lines of *text*, joined for display."""
    lines = [line.strip() for line in text.strip().split("\n") if line.strip()]
    return " | ".join(lines[-count:]) if lines else "no error output"


# ---------------------------------------------------------------------------
# Section 1 — DNF Cleanup
# ---------------------------------------------------------------------------


def _kernel_pkg_name(os_id: str) -> str:
    """Return the installonly kernel package name for the detected OS."""
    return "kernel" if os_id == "openeuler" else "kernel-core"


def _rpmvercmp(ver_a: str, ver_b: str) -> int:
    """Compare two version-release strings using RPM's rpmvercmp algorithm.

    Segments alternate between digits and letters; numeric segments compare
    numerically and always beat alphabetic ones; "~" sorts before everything
    (including end-of-string). Returns <0, 0 or >0 like strcmp.
    """
    pos_a = pos_b = 0
    while pos_a < len(ver_a) or pos_b < len(ver_b):
        tilde_a = pos_a < len(ver_a) and ver_a[pos_a] == "~"
        tilde_b = pos_b < len(ver_b) and ver_b[pos_b] == "~"
        if tilde_a or tilde_b:
            if tilde_a and tilde_b:
                pos_a += 1
                pos_b += 1
                continue
            return -1 if tilde_a else 1
        # skip separators (anything not alphanumeric)
        while pos_a < len(ver_a) and not ver_a[pos_a].isalnum():
            pos_a += 1
        while pos_b < len(ver_b) and not ver_b[pos_b].isalnum():
            pos_b += 1
        if pos_a >= len(ver_a) or pos_b >= len(ver_b):
            break
        if ver_a[pos_a].isdigit() != ver_b[pos_b].isdigit():
            # numeric segments always beat alphabetic ones
            return 1 if ver_a[pos_a].isdigit() else -1
        start_a, start_b = pos_a, pos_b
        if ver_a[pos_a].isdigit():
            while pos_a < len(ver_a) and ver_a[pos_a].isdigit():
                pos_a += 1
            while pos_b < len(ver_b) and ver_b[pos_b].isdigit():
                pos_b += 1
            seg_a = ver_a[start_a:pos_a].lstrip("0") or "0"
            seg_b = ver_b[start_b:pos_b].lstrip("0") or "0"
            if len(seg_a) != len(seg_b):
                return 1 if len(seg_a) > len(seg_b) else -1
        else:
            while pos_a < len(ver_a) and ver_a[pos_a].isalpha():
                pos_a += 1
            while pos_b < len(ver_b) and ver_b[pos_b].isalpha():
                pos_b += 1
            seg_a = ver_a[start_a:pos_a]
            seg_b = ver_b[start_b:pos_b]
        if seg_a != seg_b:
            return 1 if seg_a > seg_b else -1
    # whichever side still has segments left over is the newer version
    if pos_a < len(ver_a):
        return 1
    if pos_b < len(ver_b):
        return -1
    return 0


def _rpm_evr_cmp(evr_a: str, evr_b: str) -> int:
    """Compare two EVR (epoch:version-release) strings, epoch first."""
    epoch_a, _, rest_a = evr_a.partition(":")
    epoch_b, _, rest_b = evr_b.partition(":")
    if epoch_a.isdigit() and epoch_b.isdigit() and epoch_a != epoch_b:
        return 1 if int(epoch_a) > int(epoch_b) else -1
    return _rpmvercmp(rest_a, rest_b)


def _list_kernel_packages(os_id: str) -> list[str] | None:
    """List installed kernel package EVRs (epoch:version-release.arch), newest first.

    Returns None when the rpm query itself fails — callers must distinguish
    that from an empty list (no kernel packages installed). Sorting uses RPM
    version comparison: rpm -q output order is install order and must never
    be trusted as version order.
    """
    result = run(
        [
            "rpm",
            "-q",
            _kernel_pkg_name(os_id),
            "--qf",
            "%{EPOCHNUM}:%{VERSION}-%{RELEASE}.%{ARCH}\n",
        ],
        timeout=30,
    )
    if result.returncode != 0:
        return None
    evrs = [line.strip() for line in result.stdout.strip().split("\n") if line.strip()]
    evrs.sort(key=cmp_to_key(_rpm_evr_cmp), reverse=True)
    return evrs


def _select_kernels_to_keep(kernels: list[str], running: str) -> tuple[list[str], list[str]]:
    """Split EVRs (newest first) into (kept, would_remove).

    Mirrors dnf's --oldinstallonly behaviour with INSTALLONLY_LIMIT: keep the
    newest INSTALLONLY_LIMIT versions and always keep the running kernel,
    even when it is not among the newest (updated but not yet rebooted).
    """
    kept = list(kernels[:INSTALLONLY_LIMIT])
    running_evr = next((e for e in kernels if e.split(":", 1)[1] == running), None)
    if running_evr is not None and running_evr not in kept:
        kept.append(running_evr)
    would_remove = [e for e in kernels if e not in kept]
    return kept, would_remove


def _dnf_generation() -> int:
    """Return the dnf major version (e.g. 4 or 5); defaults to 4 when undetectable."""
    p = run(["dnf", "--version"], timeout=15)
    if p.returncode == 0 and p.stdout.strip():
        match = re.search(r"(\d+)\.\d+", p.stdout.strip().splitlines()[0])
        if match:
            return int(match.group(1))
    return 4


def _dnf_cache_dirs(dnf_major: int) -> tuple[list[str], list[str]]:
    """Return (active, stale) dnf cache directories for the dnf generation.

    dnf5 caches in /var/cache/libdnf5; dnf4 in /var/cache/dnf (+ legacy yum).
    Stale dirs belong to the other generation — reported, never auto-cleaned.
    """
    if dnf_major >= 5:
        return ["/var/cache/libdnf5"], ["/var/cache/dnf", "/var/cache/yum"]
    return ["/var/cache/dnf", "/var/cache/yum"], []


def _parse_removal_count(text: str) -> int | None:
    """Extract the removed-package count from dnf transaction output.

    Handles dnf4 ("Remove  12 Packages") and dnf5 ("Removing:  12 packages")
    summary wording. Returns 0 for "Nothing to do", None when unparseable.
    """
    match = re.search(r"^Remove\s+(\d+)\s+Packages?\b", text, re.MULTILINE)
    if match is None:
        match = re.search(r"^\s*Removing:\s+(\d+)\s+packages?\b", text, re.MULTILINE)
    if match:
        return int(match.group(1))
    return 0 if "Nothing to do" in text else None


def dnf_cleanup(
    os_id: str, dnf_major: int, *, dry_run: bool, warnings: list[str]
) -> dict[str, Any]:
    """Prune old kernels via dnf's own mechanism, autoremove, and clean the cache.

    Kernel pruning delegates to ``dnf remove --oldinstallonly``, which keeps
    the newest INSTALLONLY_LIMIT versions and never touches the running
    kernel. Returns a dict with keys: autoremoved, autoremoved_count,
    kernels_removed (real mode) / kernels_planned (dry-run), kernels_kept,
    space_saved, cache_cleaned, errors.
    """
    pkg_name = _kernel_pkg_name(os_id)
    result: dict[str, Any] = {
        "autoremoved": False,
        "autoremoved_count": 0,
        "kernels_removed": 0,
        "kernels_planned": 0,
        "kernels_kept": "",
        "space_saved": 0,
        "cache_cleaned": False,
        "errors": [],
    }

    # 1. Autoremove unneeded packages
    _status("Removing unneeded packages (dnf autoremove)")
    if dry_run:
        # --assumeno makes dnf print the transaction and answer "no"; the
        # exit code may then be non-zero, so parse stdout before checking it.
        p = run(["dnf", "autoremove", "--assumeno"], timeout=300)
        count = _parse_removal_count(p.stdout)
        if count is not None:
            result["autoremoved_count"] = count
            _status_done(f"would remove {count} package(s)" if count else "nothing to remove")
        elif p.returncode == 0:
            _status_done("would run (package count unavailable)")
        else:
            _status_done("failed")
            _warn(f"dnf autoremove preview failed: {_last_lines(p.stderr)}")
            result["errors"].append("dnf autoremove preview failed")
    else:
        p = run(["dnf", "autoremove", "-y"], timeout=300)
        if p.returncode == 0:
            count = _parse_removal_count(p.stdout)
            result["autoremoved_count"] = count or 0
            result["autoremoved"] = bool(count)
            _status_done(f"removed {count} package(s)" if count else "nothing to do")
        else:
            _status_done("failed")
            _warn(f"dnf autoremove failed: {_last_lines(p.stderr)}")
            result["errors"].append("dnf autoremove failed")

    # 2. Prune old kernels — dnf decides what to remove, never the script
    kernels = _list_kernel_packages(os_id)
    if kernels is None:
        _warn("Could not query installed kernels (rpm) — skipping kernel cleanup")
        warnings.append("Kernel query via rpm failed — kernel cleanup skipped")
        result["errors"].append("rpm kernel query failed")
        result["kernels_kept"] = "unknown"
    elif not kernels:
        _info("No kernel packages found")
        result["kernels_kept"] = "none"
    else:
        running = os.uname().release
        kept, would_remove = _select_kernels_to_keep(kernels, running)
        result["kernels_kept"] = ", ".join(f"{pkg_name}-{evr}" for evr in kept)
        if not would_remove:
            _info(f"Old kernels: none to remove ({len(kernels)} installed, keeping {len(kept)})")
        elif dry_run:
            _status(f"Pruning old kernels (keeping newest {INSTALLONLY_LIMIT} + running)")
            result["kernels_planned"] = len(would_remove)
            _status_done(f"would remove {len(would_remove)}")
            for evr in would_remove:
                _info(f"would remove {pkg_name}-{evr}")
        else:
            _status(f"Pruning {len(would_remove)} old kernel(s) via dnf (keeping {len(kept)})")
            cmd = ["dnf", "remove", "-y", "--oldinstallonly"]
            if dnf_major >= 5:
                cmd.append(f"--limit={INSTALLONLY_LIMIT}")
            else:
                cmd += ["--setopt", f"installonly_limit={INSTALLONLY_LIMIT}"]
            cmd.append(pkg_name)
            p = run(cmd, timeout=300)
            if p.returncode == 0:
                result["kernels_removed"] = len(would_remove)
                _status_done(f"removed {len(would_remove)} (running: {running})")
            else:
                _status_done("failed")
                _warn(f"Kernel removal failed: {_last_lines(p.stderr)}")
                result["errors"].append("old kernel removal failed")

    # 3. Clean DNF cache — measure only what the active dnf generation cleans
    active_dirs, stale_dirs = _dnf_cache_dirs(dnf_major)
    for stale in stale_dirs:
        stale_size = _du(stale)
        if stale_size > 0:
            msg = f"Stale dnf4 cache at {stale} ({_format_size(stale_size)}) — unused by dnf5"
            _info(msg)
            warnings.append(msg)
    cache_before = sum(_du(d) for d in active_dirs)
    _status("Cleaning DNF cache")
    if dry_run:
        result["space_saved"] += cache_before
        _status_done(f"would free ~{_format_size(cache_before)}")
    else:
        p = run(["dnf", "clean", "all"], timeout=60)
        if p.returncode == 0:
            saved = max(0, cache_before - sum(_du(d) for d in active_dirs))
            result["space_saved"] += saved
            result["cache_cleaned"] = True
            _status_done(_format_size(saved))
        else:
            _status_done("failed")
            _warn(f"dnf clean all failed: {_last_lines(p.stderr)}")
            result["errors"].append("dnf clean all failed")

    return result


# ---------------------------------------------------------------------------
# Section 2 — Journal Cleanup
# ---------------------------------------------------------------------------


def journal_cleanup(*, dry_run: bool, warnings: list[str]) -> dict[str, Any]:
    """Vacuum the systemd journal to JOURNAL_VACUUM_SIZE_MB and report space freed.

    Measures both the persistent (/var/log/journal) and volatile
    (/run/log/journal) locations. Returns a dict with keys: vacuumed,
    space_saved, journal_size_after, persistent_warning, errors.
    """
    result: dict[str, Any] = {
        "vacuumed": False,
        "space_saved": 0,
        "journal_size_after": 0,
        "persistent_warning": False,
        "errors": [],
    }

    def journal_size() -> int:
        return sum(_du(path) for path in JOURNAL_DIRS if os.path.exists(path))

    vacuum_arg = f"--vacuum-size={JOURNAL_VACUUM_SIZE_MB}M"
    size_before = journal_size()
    _status(f"Vacuuming journal ({vacuum_arg})")
    if dry_run:
        estimated = max(0, size_before - JOURNAL_VACUUM_SIZE_MB * 1024 * 1024)
        result["space_saved"] = estimated
        _status_done(f"would free ~{_format_size(estimated)}")
        size_check = size_before - estimated
    else:
        p = run(["journalctl", vacuum_arg], timeout=120)
        if p.returncode == 0:
            result["vacuumed"] = True
            result["journal_size_after"] = journal_size()
            result["space_saved"] = max(0, size_before - result["journal_size_after"])
            _status_done(_format_size(result["space_saved"]))
        else:
            _status_done("failed")
            _warn(f"journalctl vacuum failed: {_last_lines(p.stderr)}")
            result["errors"].append("journal vacuum failed")
        size_check = journal_size()

    # Warn when the journal stays large despite vacuuming (post-vacuum size)
    if size_check > JOURNAL_WARN_BYTES:
        msg = (
            f"Journal still large ({_format_size(size_check)}) after vacuum. "
            "Consider setting SystemMaxUse= in /etc/systemd/journald.conf"
        )
        _warn(msg)
        warnings.append(msg)
        result["persistent_warning"] = True

    return result


# ---------------------------------------------------------------------------
# Section 3 — systemd Audit
# ---------------------------------------------------------------------------


def systemd_audit(*, warnings: list[str]) -> dict[str, Any]:
    """Audit systemd for failed units, disabled unit files, and leftover
    files in /etc/systemd/system/.

    This section is purely informational — no changes are made.

    Returns a dict with keys: failed_units, disabled_count, leftover_files,
    errors.
    """
    result: dict[str, Any] = {
        "failed_units": [],
        "disabled_count": 0,
        "leftover_files": [],
        "errors": [],
    }

    # 1. Failed units
    _status("Checking for failed systemd units")
    p = run(["systemctl", "--failed", "--no-legend"], timeout=30)
    if p.returncode != 0 and not p.stdout.strip():
        _status_done("unavailable")
        msg = f"systemctl --failed unavailable: {_last_lines(p.stderr)}"
        _warn(msg)
        warnings.append(msg)
    else:
        failed_lines = [line.strip() for line in p.stdout.strip().split("\n") if line.strip()]
        result["failed_units"] = failed_lines
        _status_done(f"{len(failed_lines)} failed")
        for line in failed_lines:
            _warn(f"Failed unit: {line}")
        if failed_lines:
            warnings.append(f"{len(failed_lines)} failed systemd unit(s) — see above")

    # 2. Disabled unit files (informational)
    _status("Counting disabled unit files")
    p = run(
        ["systemctl", "list-unit-files", "--state=disabled", "--no-legend"],
        timeout=30,
    )
    disabled_lines = [line.strip() for line in p.stdout.strip().split("\n") if line.strip()]
    result["disabled_count"] = len(disabled_lines)
    _status_done(f"{result['disabled_count']} disabled (informational only)")

    # 3. Leftover unit files: regular files systemd does not know about.
    # A single list-unit-files call yields every unit name systemd
    # recognises — static, disabled and indirect units are legitimate and
    # must NOT be flagged; only names missing entirely are leftovers.
    etc_systemd = "/etc/systemd/system"
    if os.path.isdir(etc_systemd):
        _status("Checking for leftover unit files")
        p = run(["systemctl", "list-unit-files", "--no-legend"], timeout=30)
        if p.returncode != 0:
            _status_done("unavailable")
            msg = f"systemctl list-unit-files unavailable: {_last_lines(p.stderr)}"
            _warn(msg)
            warnings.append(msg)
        else:
            known = {line.split()[0] for line in p.stdout.splitlines() if line.split()}
            leftovers = []
            for entry in sorted(os.listdir(etc_systemd)):
                full_path = os.path.join(etc_systemd, entry)
                if os.path.islink(full_path) or not os.path.isfile(full_path):
                    continue
                if not entry.endswith(UNIT_SUFFIXES):
                    continue
                if entry not in known:
                    leftovers.append(entry)
            result["leftover_files"] = leftovers
            if leftovers:
                _status_done(f"{len(leftovers)} leftover(s)")
                for leftover in leftovers:
                    _warn(f"Leftover unit file: {etc_systemd}/{leftover}")
                warnings.append(f"{len(leftovers)} leftover unit file(s) in {etc_systemd}/")
            else:
                _status_done("none")

    return result


# ---------------------------------------------------------------------------
# Section 4 — Temp File Cleanup
# ---------------------------------------------------------------------------


def _find_old_files(directory: str, days: int) -> list[str] | None:
    """List regular files under *directory* older than *days* days.

    Stays on one filesystem (-xdev) and never matches the start directory
    itself (-mindepth 1). Returns None when the find command fails.
    """
    p = run(
        [
            "find",
            directory,
            "-mindepth",
            "1",
            "-xdev",
            "-type",
            "f",
            "-mtime",
            f"+{days}",
            "-print0",
        ],
        timeout=120,
    )
    if p.returncode != 0:
        return None
    return [path for path in p.stdout.split("\0") if path]


def _paths_size(paths: list[str]) -> int:
    """Return the total size of *paths*, ignoring files that vanished."""
    total = 0
    for path in paths:
        try:
            total += os.path.getsize(path)
        except OSError:
            pass
    return total


def _clean_temp_dir(
    directory: str,
    days: int,
    *,
    dry_run: bool,
    warnings: list[str],
    errors: list[str],
) -> tuple[int, int]:
    """Clean one temp directory; return (files affected, bytes freed).

    Files are counted and measured BEFORE anything is deleted, so real-mode
    reporting reflects what was actually removed. Dry-run deletes nothing and
    reports the exact size of the matching files only.
    """
    old_files = _find_old_files(directory, days)
    if old_files is None:
        _status_done("failed")
        _warn(f"find {directory} failed — cannot list old files")
        warnings.append(f"Temp cleanup of {directory} failed (find error)")
        errors.append(f"find failed in {directory}")
        return 0, 0
    old_size = _paths_size(old_files)
    if dry_run:
        _status_done(f"would remove {len(old_files)} file(s) (~{_format_size(old_size)})")
        return len(old_files), old_size

    for extra_args in (["-type", "f", "-mtime", f"+{days}"], ["-type", "d", "-empty"]):
        p = run(
            ["find", directory, "-mindepth", "1", "-xdev", *extra_args, "-delete"],
            timeout=120,
        )
        if p.returncode != 0:
            _warn(f"find {directory} -delete failed: {_last_lines(p.stderr)}")
            warnings.append(f"Temp cleanup of {directory} hit find errors")
            errors.append(f"find -delete failed in {directory}")

    remaining = _find_old_files(directory, days) or []
    removed = len(old_files) - len(remaining)
    freed = max(0, old_size - _paths_size(remaining))
    _status_done(f"{removed} file(s), freed {_format_size(freed)}")
    return removed, freed


def temp_cleanup(*, dry_run: bool, warnings: list[str]) -> dict[str, Any]:
    """Remove old files from /tmp and /var/tmp.

    /tmp: files older than TMP_MAX_AGE_DAYS. /var/tmp: files older than
    VARTMP_MAX_AGE_DAYS. Returns a dict with keys: tmp_files_removed,
    vartmp_files_removed (real mode) / *_planned (dry-run), space_saved,
    errors.
    """
    result: dict[str, Any] = {
        "tmp_files_removed": 0,
        "vartmp_files_removed": 0,
        "tmp_files_planned": 0,
        "vartmp_files_planned": 0,
        "space_saved": 0,
        "errors": [],
    }

    _status(f"Cleaning /tmp (files older than {TMP_MAX_AGE_DAYS} days)")
    removed, freed = _clean_temp_dir(
        "/tmp", TMP_MAX_AGE_DAYS, dry_run=dry_run, warnings=warnings, errors=result["errors"]
    )
    result["tmp_files_planned" if dry_run else "tmp_files_removed"] = removed
    result["space_saved"] += freed

    if os.path.isdir("/var/tmp"):
        _status(f"Cleaning /var/tmp (files older than {VARTMP_MAX_AGE_DAYS} days)")
        removed, freed = _clean_temp_dir(
            "/var/tmp",
            VARTMP_MAX_AGE_DAYS,
            dry_run=dry_run,
            warnings=warnings,
            errors=result["errors"],
        )
        result["vartmp_files_planned" if dry_run else "vartmp_files_removed"] = removed
        result["space_saved"] += freed
    else:
        _info("/var/tmp: directory not found, skipping")

    return result


# ---------------------------------------------------------------------------
# Section 5 — Core Dump Cleanup
# ---------------------------------------------------------------------------


def coredump_cleanup(*, dry_run: bool, warnings: list[str]) -> dict[str, Any]:
    """Remove core dumps older than COREDUMP_MAX_AGE_DAYS.

    Operates on /var/lib/systemd/coredump/. Recent dumps are always kept —
    a crash under active analysis must never be deleted. Returns a dict with
    keys: cores_removed (real mode) / cores_planned (dry-run),
    cores_kept_recent, space_saved, errors.
    """
    result: dict[str, Any] = {
        "cores_removed": 0,
        "cores_planned": 0,
        "cores_kept_recent": 0,
        "space_saved": 0,
        "errors": [],
    }

    coredump_dir = "/var/lib/systemd/coredump"
    if not os.path.isdir(coredump_dir):
        _info("No coredump directory found, skipping")
        return result

    cutoff = time.time() - COREDUMP_MAX_AGE_DAYS * 86400
    old_dumps: list[tuple[str, float, int]] = []  # (name, mtime, size)
    recent = 0
    try:
        entries = sorted(os.listdir(coredump_dir))
    except OSError as exc:
        _warn(f"Cannot list {coredump_dir}: {exc}")
        warnings.append(f"Core dump cleanup skipped — cannot read {coredump_dir}")
        result["errors"].append("coredump listdir failed")
        return result
    for entry in entries:
        full_path = os.path.join(coredump_dir, entry)
        try:
            stat = os.stat(full_path)
        except OSError:
            continue
        if not os.path.isfile(full_path):
            continue
        if stat.st_mtime < cutoff:
            old_dumps.append((entry, stat.st_mtime, stat.st_size))
        else:
            recent += 1
    result["cores_kept_recent"] = recent

    if not old_dumps:
        _info(f"No core dumps older than {COREDUMP_MAX_AGE_DAYS} days found")
        if recent:
            _info(f"Keeping {recent} recent core dump(s) (< {COREDUMP_MAX_AGE_DAYS} days old)")
        return result

    # List every dump with its date and size before any deletion happens.
    total_size = sum(size for _, _, size in old_dumps)
    heading = "Would remove" if dry_run else "Removing"
    _info(
        f"{heading} {len(old_dumps)} core dump(s) older than "
        f"{COREDUMP_MAX_AGE_DAYS} days ({_format_size(total_size)}):"
    )
    for name, mtime, size in old_dumps:
        date = time.strftime("%Y-%m-%d %H:%M", time.localtime(mtime))
        prefix = "would remove" if dry_run else "remove"
        _info(f"  {prefix}: {name} ({date}, {_format_size(size)})")

    if dry_run:
        result["cores_planned"] = len(old_dumps)
        result["space_saved"] = total_size
        return result

    _status(f"Deleting {len(old_dumps)} core dump(s)")
    removed = 0
    freed = 0
    failed = 0
    for name, _, size in old_dumps:
        try:
            os.unlink(os.path.join(coredump_dir, name))
            removed += 1
            freed += size
        except OSError as exc:
            failed += 1
            _warn(f"Could not remove {name}: {exc}")
    result["cores_removed"] = removed
    result["space_saved"] = max(0, freed)
    if failed:
        msg = f"{failed} core dump(s) could not be removed — see above"
        warnings.append(msg)
        result["errors"].append("coredump removal incomplete")
    _status_done(f"removed {removed}, freed {_format_size(result['space_saved'])}")
    if recent:
        _info(f"Keeping {recent} recent core dump(s) (< {COREDUMP_MAX_AGE_DAYS} days old)")
    return result


# ---------------------------------------------------------------------------
# Section 6 — Package Audit (informational only)
# ---------------------------------------------------------------------------


def package_audit(*, dnf_major: int, warnings: list[str]) -> dict[str, Any]:
    """Audit dnf for packages not present in any repository.

    Informational only — no packages are removed. Uses the ``--extras``
    option form, which works on both dnf4 and dnf5 (the positional
    "dnf list extras" is dnf4-only); on dnf5, JSON output avoids fragile
    text parsing. Returns a dict with keys: extras_count, extras_packages,
    errors.
    """
    result: dict[str, Any] = {
        "extras_count": 0,
        "extras_packages": [],
        "errors": [],
    }

    _status("Checking for packages not in any repo (dnf list --extras)")
    names: list[str] | None = None
    if dnf_major >= 5:
        p = run(["dnf", "list", "--extras", "--json"], timeout=60)
        if p.returncode == 0:
            try:
                data = json.loads(p.stdout)
                names = sorted(
                    f"{pkg['name']}.{pkg['arch']}" if pkg.get("arch") else pkg["name"]
                    for section in data.values()
                    if isinstance(section, list)
                    for pkg in section
                    if isinstance(pkg, dict) and pkg.get("name")
                )
            except (json.JSONDecodeError, AttributeError, TypeError):
                names = None
    else:
        p = run(["dnf", "list", "--extras", "--quiet"], timeout=60)
        if p.returncode == 0:
            headers = (
                "last metadata",
                "extra packages",
                "available packages",
                "installed packages",
            )
            names = [
                line.split()[0]
                for line in p.stdout.strip().split("\n")
                if line.strip() and not line.strip().lower().startswith(headers)
            ]

    if names is None:
        _status_done("unavailable")
        detail = _last_lines(p.stderr) if p.returncode != 0 else "unparseable dnf output"
        msg = f"dnf list --extras unavailable: {detail}"
        _warn(msg)
        warnings.append(msg)
        return result

    result["extras_count"] = len(names)
    result["extras_packages"] = names
    if names:
        _status_done(f"{len(names)} package(s)")
        for name in names[:10]:
            _warn(f"Extra package: {name}")
        if len(names) > 10:
            _warn(f"  ... and {len(names) - 10} more")
        warnings.append(f"{len(names)} installed package(s) not found in any repo")
    else:
        _status_done("none")

    return result


# ---------------------------------------------------------------------------
# Summary
# ---------------------------------------------------------------------------


def print_summary(
    os_display: str,
    version_id: str,
    results: dict[str, Any],
    warnings: list[str],
    elapsed: float,
    *,
    dry_run: bool,
) -> None:
    """Print a final summary of all sections to stderr.

    In dry-run mode everything is reported as "would …" — planned actions
    are never stated as accomplished, and sizes are estimates marked "~".
    """
    print(f"\n{BOLD}── Optimisation Summary ──{RESET}", file=sys.stderr)
    print(f"  OS: {os_display} {version_id}", file=sys.stderr)
    if dry_run:
        print(
            f'  {YELLOW}DRY RUN — nothing was changed; all actions are "would …"{RESET}',
            file=sys.stderr,
        )

    # DNF
    rd = results.get("dnf", {})
    if rd:
        count = rd.get("autoremoved_count") or 0
        if dry_run:
            if count > 0:
                _info(f"DNF autoremove: would remove {count} package(s)")
            else:
                _info("DNF autoremove: nothing to remove")
        elif rd.get("autoremoved"):
            _ok(f"DNF autoremove: removed {count} package(s)")
        else:
            _info("DNF autoremove: nothing to remove")
        kernels = rd.get("kernels_planned", 0) if dry_run else rd.get("kernels_removed", 0)
        if kernels > 0:
            if dry_run:
                _info(f"Old kernels: would remove {kernels}")
            else:
                _ok(f"Old kernels removed: {kernels}")
        else:
            _info("Old kernels: none to remove")
        _info(f"Kernels kept: {rd.get('kernels_kept', '?')}")
        cache_saved = _format_size(rd.get("space_saved", 0))
        if dry_run:
            _info(f"DNF cache: would clean (~{cache_saved})")
        elif rd.get("cache_cleaned"):
            _ok(f"DNF cache: cleaned (freed {cache_saved})")

    # Journal
    rj = results.get("journal", {})
    if rj:
        if dry_run:
            _info(
                f"Journal: would vacuum to {JOURNAL_VACUUM_SIZE_MB}M "
                f"(est. freed ~{_format_size(rj.get('space_saved', 0))})"
            )
        elif rj.get("vacuumed"):
            _ok(f"Journal vacuumed: freed {_format_size(rj.get('space_saved', 0))}")
        if rj.get("persistent_warning"):
            _warn("Journal storage is still large — check SystemMaxUse= in journald.conf")

    # systemd audit (informational in both modes)
    rs = results.get("systemd", {})
    if rs:
        failed = len(rs.get("failed_units", []))
        if failed > 0:
            _warn(f"Failed systemd units: {failed}")
        else:
            _ok("Failed systemd units: none")
        _info(f"Disabled unit files: {rs.get('disabled_count', 0)} (informational)")
        leftovers = len(rs.get("leftover_files", []))
        if leftovers > 0:
            _warn(f"Leftover unit files in /etc/systemd/system/: {leftovers}")
        else:
            _info("Leftover unit files: none")

    # Temp cleanup
    rt = results.get("tmp", {})
    if rt:
        if dry_run:
            tmp_count = rt.get("tmp_files_planned", 0)
            vartmp_count = rt.get("vartmp_files_planned", 0)
        else:
            tmp_count = rt.get("tmp_files_removed", 0)
            vartmp_count = rt.get("vartmp_files_removed", 0)
        if tmp_count > 0 or vartmp_count > 0:
            verb = "would remove" if dry_run else "removed"
            msg = f"Temp files: {verb} {tmp_count} from /tmp, {vartmp_count} from /var/tmp"
            (_info if dry_run else _ok)(msg)
        else:
            _info("Temp files: none to remove")

    # Core dumps
    rc = results.get("cores", {})
    if rc:
        cores = rc.get("cores_planned", 0) if dry_run else rc.get("cores_removed", 0)
        if cores > 0:
            verb = "would remove" if dry_run else "removed"
            msg = f"Core dumps: {verb} {cores} (older than {COREDUMP_MAX_AGE_DAYS} days)"
            (_info if dry_run else _ok)(msg)
        else:
            _info("Core dumps: none to remove")
        if rc.get("cores_kept_recent", 0) > 0:
            _info(f"Core dumps kept (recent): {rc['cores_kept_recent']}")

    # Package audit (informational in both modes)
    rp = results.get("packages", {})
    if rp:
        extras = rp.get("extras_count", 0)
        if extras > 0:
            _warn(f"Packages not in any repo: {extras}")
        else:
            _ok("Package audit: all packages from known repos")

    # Total space (dry-run: consistent estimates across all sections)
    total_space: int = 0
    for section in ["dnf", "journal", "tmp", "cores"]:
        total_space += results.get(section, {}).get("space_saved", 0)
    if dry_run:
        _info(f"Estimated space to free: ~{_format_size(total_space)}")
    elif total_space > 0:
        _ok(f"Total space freed: {_format_size(total_space)}")
    else:
        _info("Total space freed: 0 B")

    if warnings:
        print(file=sys.stderr)
        for warning in warnings:
            _warn(warning)

    print(f"\n{BOLD}{'═' * 60}{RESET}", file=sys.stderr)
    print(f"  {BOLD}Total time: {elapsed:.1f}s{RESET}", file=sys.stderr)
    print(f"{BOLD}{'═' * 60}{RESET}", file=sys.stderr)
    print(file=sys.stderr)


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------


def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
    """Parse command-line arguments."""
    parser = argparse.ArgumentParser(
        description="Idempotent system cleanup and optimisation for Fedora, CentOS Stream and openEuler",
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    parser.add_argument(
        "--dry-run",
        action="store_true",
        help="Preview without making changes",
    )
    parser.add_argument(
        "--skip-dnf",
        action="store_true",
        help="Skip DNF cleanup (autoremove, old kernels, cache)",
    )
    parser.add_argument(
        "--skip-journal",
        action="store_true",
        help="Skip journal vacuum",
    )
    parser.add_argument(
        "--skip-tmp",
        action="store_true",
        help="Skip temp file cleanup",
    )
    parser.add_argument(
        "--skip-cores",
        action="store_true",
        help="Skip core dump cleanup",
    )
    parser.add_argument(
        "--version",
        action="version",
        version=f"%(prog)s {VERSION}",
    )
    return parser.parse_args(argv)


# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------


def main() -> None:
    """Entry point — detect OS, check root, then run enabled sections.

    Every section is isolated: a crashing or failing section is recorded and
    reported, but never aborts the remaining sections or the summary. The
    process exits non-zero when any operation failed.
    """
    start_time = time.monotonic()
    warnings: list[str] = []
    failures: list[str] = []
    results: dict[str, Any] = {}

    args = parse_args()
    check_root()
    os_id, os_display, version_id = detect_os()
    dnf_major = _dnf_generation()

    # Header
    print(f"\n{BOLD}{'═' * 60}{RESET}", file=sys.stderr)
    print(
        f"{BOLD}  System Optimise  v{VERSION} — {os_display} {version_id}{RESET}",
        file=sys.stderr,
    )
    print(f"{BOLD}{'═' * 60}{RESET}", file=sys.stderr)
    if args.dry_run:
        print(
            f"\n  {YELLOW}{BOLD}DRY RUN — no changes will be made{RESET}",
            file=sys.stderr,
        )
    print(file=sys.stderr)

    sections: list[tuple[str, str, Callable[[], dict[str, Any]], str | None]] = [
        (
            "dnf",
            "DNF Cleanup",
            lambda: dnf_cleanup(os_id, dnf_major, dry_run=args.dry_run, warnings=warnings),
            "--skip-dnf" if args.skip_dnf else None,
        ),
        (
            "journal",
            "Journal Cleanup",
            lambda: journal_cleanup(dry_run=args.dry_run, warnings=warnings),
            "--skip-journal" if args.skip_journal else None,
        ),
        (
            "systemd",
            "systemd Audit",
            lambda: systemd_audit(warnings=warnings),
            None,
        ),
        (
            "tmp",
            "Temp File Cleanup",
            lambda: temp_cleanup(dry_run=args.dry_run, warnings=warnings),
            "--skip-tmp" if args.skip_tmp else None,
        ),
        (
            "cores",
            "Core Dump Cleanup",
            lambda: coredump_cleanup(dry_run=args.dry_run, warnings=warnings),
            "--skip-cores" if args.skip_cores else None,
        ),
        (
            "packages",
            "Package Audit",
            lambda: package_audit(dnf_major=dnf_major, warnings=warnings),
            None,
        ),
    ]
    for key, title, section_func, skip_flag in sections:
        print(f"\n{BOLD}── {title} ──{RESET}", file=sys.stderr)
        if skip_flag is not None:
            _info(f"{title}: skipped ({skip_flag})")
            continue
        try:
            results[key] = section_func()
        except Exception as exc:  # isolation: one section must not kill the rest
            _fail(f"{title} crashed: {exc}")
            failures.append(f"{title}: crashed ({exc})")
            continue
        for error in results[key].get("errors", []):
            failures.append(f"{title}: {error}")

    elapsed = time.monotonic() - start_time
    print_summary(os_display, version_id, results, warnings, elapsed, dry_run=args.dry_run)

    if failures:
        print(f"{RED}{BOLD}Failed operations ({len(failures)}):{RESET}", file=sys.stderr)
        for failure in failures:
            _fail(failure)
        print(file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()