gentoo-legion-python/install/utils.py

"""Common utilities for Gentoo installation."""

import subprocess
import sys
from dataclasses import dataclass
from pathlib import Path
from typing import NoReturn


# --- Colors ---

class Color:
    RED = "\033[0;31m"
    GREEN = "\033[0;32m"
    YELLOW = "\033[0;33m"
    BOLD = "\033[1m"
    NC = "\033[0m"  # No color


def info(msg: str) -> None:
    print(f"{Color.YELLOW}{msg}{Color.NC}")


def success(msg: str) -> None:
    print(f"{Color.GREEN}{msg}{Color.NC}")


def error(msg: str) -> None:
    print(f"{Color.RED}{msg}{Color.NC}", file=sys.stderr)


def warn(msg: str) -> None:
    print(f"{Color.YELLOW}WARNING: {msg}{Color.NC}", file=sys.stderr)


def fatal(msg: str) -> NoReturn:
    error(f"FATAL: {msg}")
    sys.exit(1)


def prompt(msg: str) -> str:
    return input(f"{Color.YELLOW}{msg}{Color.NC}")


def confirm(msg: str, require: str = "yes") -> bool:
    response = prompt(f"{msg} (type '{require}' to confirm): ")
    return response == require


# --- Command Execution ---

@dataclass
class RunResult:
    returncode: int
    stdout: str
    stderr: str

    @property
    def ok(self) -> bool:
        return self.returncode == 0


def run(
    *args: str,
    check: bool = True,
    capture: bool = False,
    quiet: bool = False,
) -> RunResult:
    """Run a command with proper error handling.

    Args:
        *args: Command and arguments
        check: Raise exception on non-zero exit
        capture: Capture stdout/stderr (otherwise inherit terminal)
        quiet: Suppress command echo
    """
    if not quiet:
        info(f"$ {' '.join(args)}")

    result = subprocess.run(
        args,
        capture_output=capture,
        text=True,
    )

    run_result = RunResult(
        returncode=result.returncode,
        stdout=result.stdout if capture else "",
        stderr=result.stderr if capture else "",
    )

    if check and not run_result.ok:
        error(f"Command failed with exit code {result.returncode}")
        if capture and result.stderr:
            error(result.stderr)
        raise subprocess.CalledProcessError(result.returncode, args)

    return run_result


def run_quiet(*args: str, check: bool = True) -> RunResult:
    """Run a command silently, capturing output."""
    return run(*args, check=check, capture=True, quiet=True)


# --- System Checks ---

def check_root() -> None:
    """Ensure running as root."""
    import os
    if os.geteuid() != 0:
        fatal("This script must be run as root")


def check_uefi() -> None:
    """Ensure booted in UEFI mode."""
    if not Path("/sys/firmware/efi/efivars").is_dir():
        fatal("UEFI boot mode not detected. This script requires UEFI.")
    success("UEFI boot mode confirmed.")


def is_mounted(path: Path) -> bool:
    """Check if a path is a mount point."""
    result = run_quiet("mountpoint", "-q", str(path), check=False)
    return result.ok


def is_block_device(path: Path) -> bool:
    """Check if path is a block device."""
    return path.is_block_device()


# --- Portage Helpers ---

def dispatch_config() -> bool:
    """Auto-merge pending portage config changes.

    Returns True if changes were dispatched, False if none found.
    """
    portage_dir = Path("/etc/portage")

    # Check for pending changes
    has_cfg_files = bool(list(portage_dir.glob("._cfg*")))
    has_autounmask = any(
        (portage_dir / subdir / "zz-autounmask").exists()
        for subdir in ["package.use", "package.accept_keywords", "package.license"]
    )

    if not has_cfg_files and not has_autounmask:
        return False

    info("Auto-merging pending portage config changes...")

    # Try dispatch-conf first, fall back to etc-update
    result = run(
        "bash", "-c", "yes u | dispatch-conf",
        check=False, capture=True, quiet=True
    )
    if not result.ok:
        run(
            "etc-update", "--automode", "-5",
            check=False, capture=True, quiet=True
        )

    return True


def emerge(
    *packages: str,
    ask: bool = True,
    verbose: bool = True,
    extra_args: list[str] | None = None,
    max_retries: int = 3,
) -> None:
    """Run emerge with automatic handling of USE/keyword/license changes.

    Args:
        *packages: Package atoms to emerge
        ask: Prompt for confirmation (--ask)
        verbose: Show detailed output (--verbose)
        extra_args: Additional emerge arguments (e.g., ["--update", "--deep"])
        max_retries: Max attempts after auto-dispatching config changes
    """
    args = ["emerge", "--autounmask-write", "--autounmask-continue"]
    if ask:
        args.append("--ask")
    if verbose:
        args.append("--verbose")
    if extra_args:
        args.extend(extra_args)
    args.extend(packages)

    for attempt in range(1, max_retries + 1):
        if max_retries > 1:
            info(f"Emerge attempt {attempt}/{max_retries}...")

        result = run(*args, check=False)

        if result.ok:
            return

        # Check if we can auto-dispatch and retry
        if dispatch_config():
            continue

        # No config changes to dispatch, actual failure
        fatal(f"Emerge failed: {' '.join(packages)}")

    fatal(f"Emerge failed after {max_retries} attempts")