pkgflow.py

#!/usr/bin/env python3
"""
Directory scanner and installer helper for macOS pkg bundles.

Features:
* Recursively scans for .pkg and .dmg files up to a configurable depth.
* Validates pkg bundles with `installer -pkginfo` before installation.
* Presents an interactive picker (curses-based) for choosing pkgs/DMGs.
* Allows selective DMG extraction with size awareness and sudo handling.
* Installs chosen packages with visible progress.
"""

from __future__ import annotations

import argparse
import os
import plistlib
import shutil
import subprocess
import sys
import tempfile
import xml.etree.ElementTree as ET
from dataclasses import dataclass
from pathlib import Path
from typing import Callable, Dict, Iterable, List, NamedTuple, Optional, Tuple

# Maximum directory depth (root directory counts as depth 0).
MAX_DEPTH = 4

# DMG extraction defaults.
DMG_SIZE_LIMIT_MB = 800
DMG_SIZE_LIMIT_BYTES = DMG_SIZE_LIMIT_MB * 1024 * 1024

# Global context filled during collection.
ROOT_DIR: Path | None = None

# Validation caches to avoid repeated pkginfo calls.
VALID_PACKAGES: set[Path] = set()
INVALID_PACKAGES: set[Path] = set()


@dataclass(frozen=True)
class PackageInfo:
    """Description of a discovered pkg bundle."""

    path: Path


@dataclass(frozen=True)
class DmgInfo:
    """Description of a discovered dmg image."""

    path: Path
    size_bytes: int


def human_readable_size(num_bytes: int) -> str:
    """Convert a byte count to a compact human readable string."""

    units = ["B", "KB", "MB", "GB", "TB"]
    size = float(num_bytes)
    for unit in units:
        if size < 1024 or unit == units[-1]:
            return f"{size:.1f} {unit}"
        size /= 1024
    return f"{size:.1f} {units[-1]}"


def format_relative(path: Path | str, root: Path | None) -> str:
    """
    Return a path relative to ``root`` when possible.

    Falls back to the path's name if it lies outside the root.
    """

    try:
        resolved = Path(path).resolve()
        if root is not None:
            return str(resolved.relative_to(root))
    except (ValueError, RuntimeError, FileNotFoundError):
        pass

    return Path(path).name


def collect_artifacts(
    input_dir: str, max_depth: int = MAX_DEPTH
) -> Tuple[List[PackageInfo], List[DmgInfo], Path]:
    """
    Traverse ``input_dir`` up to ``max_depth`` and gather pkg/dmg files.

    Returns:
        (packages, dmgs, resolved_root_path)
    """

    root = Path(input_dir).expanduser().resolve()
    global ROOT_DIR
    ROOT_DIR = root

    if not root.exists():
        raise FileNotFoundError(f"Input directory does not exist: {root}")
    if not root.is_dir():
        raise NotADirectoryError(f"Input path is not a directory: {root}")

    packages: List[PackageInfo] = []
    dmgs: List[DmgInfo] = []

    stack: List[Tuple[Path, int]] = [(root, 0)]
    while stack:
        current_dir, depth = stack.pop()

        for entry in current_dir.iterdir():
            if entry.is_symlink():
                continue

            if entry.is_file():
                suffix = entry.suffix.lower()
                if suffix == ".pkg":
                    packages.append(PackageInfo(path=entry.resolve()))
                    continue
                if suffix == ".dmg":
                    try:
                        size = entry.stat().st_size
                    except OSError:
                        size = 0
                    dmgs.append(DmgInfo(path=entry.resolve(), size_bytes=size))
                    continue

            if entry.is_dir() and depth < max_depth:
                stack.append((entry, depth + 1))

    packages.sort(key=lambda info: info.path)
    dmgs.sort(key=lambda info: info.path)

    return packages, dmgs, root


def validate_packages(
    packages: Iterable[Path], reset: bool = False
) -> Tuple[List[Path], List[Path]]:
    """
    Validate pkg bundles via ``installer -pkginfo``.

    Args:
        packages: Iterable of pkg paths to validate.
        reset:    When True, clear previous validation state first.

    Returns:
        Tuple of (valid_packages, invalid_packages) as sorted Path lists.
    """

    if reset:
        VALID_PACKAGES.clear()
        INVALID_PACKAGES.clear()

    for pkg in packages:
        pkg_path = Path(pkg).resolve()

        if pkg_path in VALID_PACKAGES or pkg_path in INVALID_PACKAGES:
            continue

        try:
            result = subprocess.run(
                ["/usr/sbin/installer", "-pkginfo", "-pkg", str(pkg_path)],
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                check=False,
            )
        except FileNotFoundError:
            INVALID_PACKAGES.add(pkg_path)
            continue

        if result.returncode == 0:
            VALID_PACKAGES.add(pkg_path)
        else:
            INVALID_PACKAGES.add(pkg_path)

    return sorted(VALID_PACKAGES), sorted(INVALID_PACKAGES)


def ensure_sudo() -> None:
    """
    Authenticate sudo credentials upfront so later commands can reuse them.
    """

    if os.geteuid() == 0:
        return

    if not sys.stdin.isatty() or not sys.stdout.isatty():
        raise PermissionError(
            "Sudo authentication required but no interactive terminal is available."
        )

    try:
        subprocess.run(["sudo", "-v"], check=True)
    except subprocess.CalledProcessError as exc:
        raise PermissionError("Unable to authenticate with sudo.") from exc


def extract_distribution_choices(pkg_path: Path) -> Optional[List[str]]:
    """
    Extract choice IDs from a Distribution package.

    For flat .pkg files, expands to a temp directory first.
    Returns None if not a Distribution package, or list of choice IDs.
    """
    pkg_path = Path(pkg_path).resolve()

    # Create temp base dir, but pkgutil --expand needs target to NOT exist
    base_dir = Path(tempfile.gettempdir()) / "pkgflow" / "expand"
    base_dir.mkdir(parents=True, exist_ok=True)
    expand_dir = base_dir / pkg_path.stem
    if expand_dir.exists():
        shutil.rmtree(expand_dir)

    try:
        # Expand the package
        result = subprocess.run(
            ["/usr/sbin/pkgutil", "--expand", str(pkg_path), str(expand_dir)],
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            check=False,
        )

        if result.returncode != 0:
            return None

        # Look for Distribution file
        dist_file = expand_dir / "Distribution"
        if not dist_file.exists():
            return None

        # Parse the Distribution XML
        try:
            tree = ET.parse(dist_file)
            root = tree.getroot()
        except ET.ParseError:
            return None

        # Extract all choice IDs
        choices: List[str] = []
        for choice in root.iter("choice"):
            choice_id = choice.get("id")
            if choice_id:
                choices.append(choice_id)

        return choices if choices else None

    finally:
        # Clean up expanded package
        shutil.rmtree(expand_dir, ignore_errors=True)


def generate_choice_changes_xml(choices: List[str], output_path: Path) -> bool:
    """
    Generate an applyChoiceChangesXML plist that selects all given choices.

    Returns True on success, False on failure.
    """
    choice_dicts = []
    for choice_id in choices:
        choice_dicts.append({
            "attributeSetting": 1,
            "choiceAttribute": "selected",
            "choiceIdentifier": choice_id,
        })

    try:
        with open(output_path, "wb") as f:
            plistlib.dump(choice_dicts, f)
        return True
    except OSError:
        return False


def is_distribution_package(pkg_path: Path) -> bool:
    """
    Quick check if a package is a Distribution package (has choices).
    """
    choices = extract_distribution_choices(pkg_path)
    return choices is not None and len(choices) > 0


# Cache for distribution package choices to avoid re-expanding
DISTRIBUTION_CHOICES_CACHE: Dict[Path, Optional[List[str]]] = {}


def get_distribution_choices(pkg_path: Path) -> Optional[List[str]]:
    """
    Get distribution choices for a package, using cache if available.
    """
    pkg_path = Path(pkg_path).resolve()
    if pkg_path not in DISTRIBUTION_CHOICES_CACHE:
        DISTRIBUTION_CHOICES_CACHE[pkg_path] = extract_distribution_choices(pkg_path)
    return DISTRIBUTION_CHOICES_CACHE[pkg_path]


def extract_dmg_packages(
    dmg_paths: Iterable[Path],
    progress_cb: Callable[[str], None] | None = None,
    require_sudo: bool = True,
) -> Tuple[Dict[Path, List[Path]], List[str]]:
    """
    Mount DMG images, copy contained pkgs into /tmp, and return discovered pkgs.
    """

    progress = progress_cb or (lambda message: None)

    dmg_path_list = [Path(path).resolve() for path in dmg_paths]
    extracted: Dict[Path, List[Path]] = {}
    errors: List[str] = []

    total_dmgs = len(dmg_path_list)
    progress(f"Preparing to unpack {total_dmgs} DMG file(s)...")

    if total_dmgs == 0:
        return extracted, errors

    if require_sudo:
        try:
            ensure_sudo()
        except PermissionError as exc:
            message = str(exc)
            errors.append(message)
            progress(f"Unable to authenticate with sudo: {message}")
            return extracted, errors

    command_prefix: List[str] = []
    if os.geteuid() != 0:
        command_prefix = ["sudo"]

    base_tmp = Path(tempfile.gettempdir()) / "pkgflow"
    mount_base = base_tmp / "mnt"
    extract_base = base_tmp / "pkgs"
    mount_base.mkdir(parents=True, exist_ok=True)
    extract_base.mkdir(parents=True, exist_ok=True)

    for index, dmg_path in enumerate(dmg_path_list, start=1):
        display_name = format_relative(dmg_path, ROOT_DIR)
        progress(f"[{index}/{total_dmgs}] Preparing {display_name}...")

        token = f"{dmg_path.stem}_{os.getpid()}_{index}"
        mount_dir = mount_base / token
        extract_dir = extract_base / token

        try:
            mount_dir.mkdir(parents=True, exist_ok=True)
            if extract_dir.exists():
                shutil.rmtree(extract_dir)
            extract_dir.mkdir(parents=True, exist_ok=True)
        except OSError as exc:
            message = f"Failed to prepare temporary directories for {display_name}: {exc}"
            errors.append(message)
            progress(f"[{index}/{total_dmgs}] {message}")
            continue

        progress(f"[{index}/{total_dmgs}] Mounting {display_name}...")

        attach_cmd = [
            *command_prefix,
            "/usr/bin/hdiutil",
            "attach",
            "-quiet",
            "-nobrowse",
            "-noautoopen",
            "-noverify",
            "-readonly",
            "-mountpoint",
            str(mount_dir),
            str(dmg_path),
        ]

        attach = subprocess.run(
            attach_cmd,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            check=False,
        )

        if attach.returncode != 0:
            error_msg = attach.stderr.decode(errors="ignore").strip()
            message = f"Failed to mount DMG {display_name}: {error_msg}"
            errors.append(message)
            progress(f"[{index}/{total_dmgs}] {message}")
            shutil.rmtree(mount_dir, ignore_errors=True)
            shutil.rmtree(extract_dir, ignore_errors=True)
            continue

        dmg_pkg_paths: List[Path] = []

        try:
            progress(f"[{index}/{total_dmgs}] Scanning {display_name} for packages...")

            for pkg_source in mount_dir.rglob("*.pkg"):
                dest = extract_dir / pkg_source.name
                progress(
                    f"[{index}/{total_dmgs}] Extracting {pkg_source.name} from {display_name}..."
                )

                try:
                    if dest.exists():
                        if dest.is_dir():
                            shutil.rmtree(dest)
                        else:
                            dest.unlink()

                    if pkg_source.is_dir():
                        shutil.copytree(pkg_source, dest)
                    else:
                        shutil.copy2(pkg_source, dest)

                    dmg_pkg_paths.append(dest.resolve())
                except OSError as exc:
                    message = (
                        f"Failed to copy package {pkg_source.name} from {display_name}: {exc}"
                    )
                    errors.append(message)
                    progress(f"[{index}/{total_dmgs}] {message}")

            dmg_pkg_paths.sort()
            extracted[dmg_path] = dmg_pkg_paths
        finally:
            progress(f"[{index}/{total_dmgs}] Detaching {display_name}...")
            detach = subprocess.run(
                [*command_prefix, "/usr/bin/hdiutil", "detach", "-quiet", str(mount_dir)],
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                check=False,
            )

            if detach.returncode != 0:
                error_msg = detach.stderr.decode(errors="ignore").strip()
                message = f"Failed to detach DMG {display_name}: {error_msg}"
                errors.append(message)
                progress(f"[{index}/{total_dmgs}] {message}")
            else:
                progress(
                    f"[{index}/{total_dmgs}] Completed {display_name} "
                    f"({len(dmg_pkg_paths)} package(s) found)."
                )

            shutil.rmtree(mount_dir, ignore_errors=True)

    progress("DMG unpacking complete.")

    return extracted, errors


def select_packages(
    packages: List[PackageInfo], dmgs: List[DmgInfo], root: Path | None
) -> List[Path]:
    """
    Present an interactive selector to choose pkgs and DMGs for installation.
    """

    pkg_selection: Dict[Path, bool] = {
        pkg.path: pkg.path in VALID_PACKAGES for pkg in packages
    }
    dmg_selection: Dict[Path, bool] = {
        dmg.path: dmg.size_bytes <= DMG_SIZE_LIMIT_BYTES for dmg in dmgs
    }

    extracted_pkg_map: Dict[Path, List[Path]] = {}
    processed_dmgs: set[Path] = set()
    dmg_errors: List[str] = []

    class Row(NamedTuple):
        text: str
        selectable: bool
        key: Tuple[str, Path] | None

    def integrate_dmg_packages(
        target_dmgs: Iterable[DmgInfo],
        verbose_callback: Callable[[str], None] | None = None,
        suspend_ui: Callable[[Callable[[], None]], None] | None = None,
    ) -> None:
        """Extract pkgs from the provided DMGs and merge them into the selection."""

        targets = list(target_dmgs)
        if not targets:
            progress = verbose_callback or (lambda message: None)
            progress("No DMG files selected for unpack.")
            return

        progress = verbose_callback or (lambda message: None)

        if os.geteuid() != 0:
            progress("Requesting sudo credentials...")

            def require_auth() -> None:
                ensure_sudo()

            try:
                if suspend_ui:
                    suspend_ui(require_auth)
                else:
                    require_auth()
            except PermissionError as exc:
                message = str(exc)
                dmg_errors.append(message)
                progress(f"Sudo authentication failed: {message}")
                return

        total_bytes = sum(dmg.size_bytes for dmg in targets)
        progress(
            f"Extracting {len(targets)} DMG file(s) "
            f"({human_readable_size(total_bytes)})..."
        )

        extracted, errors = extract_dmg_packages(
            (dmg.path for dmg in targets), progress, require_sudo=False
        )
        dmg_errors.extend(errors)

        for dmg_path, pkg_paths in extracted.items():
            processed_dmgs.add(dmg_path)
            extracted_pkg_map[dmg_path] = pkg_paths

            if pkg_paths:
                validate_packages(pkg_paths, reset=False)
                for pkg_path in pkg_paths:
                    pkg_selection.setdefault(
                        pkg_path, pkg_path in VALID_PACKAGES and pkg_path not in INVALID_PACKAGES
                    )
            else:
                pkg_selection.setdefault(dmg_path, False)

        for dmg in targets:
            dmg_selection[dmg.path] = False

    def build_rows() -> List[Row]:
        rows: List[Row] = []

        if packages:
            rows.append(Row("[PKG]", False, None))
            for pkg in packages:
                indicator = "[x]" if pkg_selection.get(pkg.path, False) else "[ ]"
                label = format_relative(pkg.path, root)
                if pkg.path in INVALID_PACKAGES:
                    label = f"{label} (invalid)"
                rows.append(Row(f"{indicator} {label}", True, ("pkg", pkg.path)))

        if dmgs:
            if packages:
                rows.append(Row("", False, None))

            rows.append(Row("[DMG]", False, None))
            for dmg in dmgs:
                indicator = "[x]" if dmg_selection.get(dmg.path, False) else "[ ]"
                label = format_relative(dmg.path, root)
                size_label = human_readable_size(dmg.size_bytes)
                notes: List[str] = []
                if dmg.size_bytes > DMG_SIZE_LIMIT_BYTES:
                    notes.append(f">{DMG_SIZE_LIMIT_MB}MB limit")
                if dmg.path in processed_dmgs:
                    notes.append("extracted")
                annotation = f"{label} [{size_label}]"
                if notes:
                    annotation = f"{annotation} ({', '.join(notes)})"
                rows.append(Row(f"{indicator} {annotation}", True, ("dmg", dmg.path)))

                pkg_entries = extracted_pkg_map.get(dmg.path, [])
                if pkg_entries:
                    for pkg_path in pkg_entries:
                        indicator = "[x]" if pkg_selection.get(pkg_path, False) else "[ ]"
                        label = Path(pkg_path).name
                        if pkg_path in INVALID_PACKAGES:
                            label = f"{label} (invalid)"
                        rows.append(Row(f"    {indicator} {label}", True, ("pkg", pkg_path)))
                elif dmg.path in processed_dmgs:
                    rows.append(Row("    (No .pkg files found)", False, None))

        if dmg_errors:
            rows.append(Row("", False, None))
            rows.append(Row("[DMG Errors]", False, None))
            for message in dmg_errors:
                rows.append(Row(f"  {message}", False, None))

        return rows

    # Non-interactive mode (e.g., piped output) defaults to extracting
    # all DMGs that pass the size limit and installing validated pkgs.
    if not sys.stdin.isatty() or not sys.stdout.isatty():
        eligible_dmgs = [dmg for dmg in dmgs if dmg_selection.get(dmg.path, False)]
        if eligible_dmgs:
            integrate_dmg_packages(eligible_dmgs, lambda message: print(message, flush=True))

        ordered_paths: List[Path] = []
        for pkg in packages:
            if pkg_selection.get(pkg.path, False) and pkg.path in VALID_PACKAGES:
                ordered_paths.append(pkg.path)
        for dmg in dmgs:
            for pkg_path in extracted_pkg_map.get(dmg.path, []):
                if pkg_selection.get(pkg_path, False) and pkg_path in VALID_PACKAGES:
                    ordered_paths.append(pkg_path)
        for pkg_path, selected in pkg_selection.items():
            if selected and pkg_path in VALID_PACKAGES and pkg_path not in ordered_paths:
                ordered_paths.append(pkg_path)
        return ordered_paths

    try:
        import curses  # type: ignore
    except ImportError:
        eligible_dmgs = [dmg for dmg in dmgs if dmg_selection.get(dmg.path, False)]
        if eligible_dmgs:
            integrate_dmg_packages(eligible_dmgs, lambda message: print(message, flush=True))

        ordered_paths: List[Path] = []
        for pkg in packages:
            if pkg_selection.get(pkg.path, False) and pkg.path in VALID_PACKAGES:
                ordered_paths.append(pkg.path)
        for dmg in dmgs:
            for pkg_path in extracted_pkg_map.get(dmg.path, []):
                if pkg_selection.get(pkg_path, False) and pkg_path in VALID_PACKAGES:
                    ordered_paths.append(pkg_path)
        for pkg_path, selected in pkg_selection.items():
            if selected and pkg_path in VALID_PACKAGES and pkg_path not in ordered_paths:
                ordered_paths.append(pkg_path)
        return ordered_paths

    index = 0
    offset = 0

    def selector(stdscr) -> None:
        nonlocal index, offset

        curses.curs_set(0)
        stdscr.keypad(True)

        instruction_lines = [
            "Use ↑/↓ to move, space to toggle, Enter to confirm.",
            "Press 'd' to unpack selected DMGs; resize to refresh layout.",
        ]

        status_message = "Ready."
        rows = build_rows()
        list_height = 1
        list_start_row = 0

        def render() -> None:
            nonlocal list_height, list_start_row, index, offset, rows
            stdscr.erase()
            max_y, max_x = stdscr.getmaxyx()

            stdscr.addnstr(0, 0, "Select packages to install:", max_x - 1)
            for line_idx, text in enumerate(instruction_lines, start=1):
                stdscr.addnstr(line_idx, 0, text, max_x - 1)

            status_row = len(instruction_lines) + 1
            stdscr.addnstr(status_row, 0, f"Status: {status_message}"[: max_x - 1], max_x - 1)

            list_start_row = status_row + 2
            list_height = max(1, max_y - list_start_row)

            if rows:
                index_limit = len(rows) - 1
                if index > index_limit:
                    index = index_limit
                if index < 0:
                    index = 0

                if index < offset:
                    offset = index
                elif index >= offset + list_height:
                    offset = max(0, index - list_height + 1)

                visible_end = min(len(rows), offset + list_height)

                for visible_row, row_index in enumerate(range(offset, visible_end), start=0):
                    row = rows[row_index]
                    screen_row = list_start_row + visible_row
                    row_text = row.text[: max_x - 1]
                    if row_index == index:
                        stdscr.attron(curses.A_REVERSE)
                        stdscr.addstr(screen_row, 0, row_text)
                        stdscr.attroff(curses.A_REVERSE)
                    else:
                        stdscr.addstr(screen_row, 0, row_text)
            else:
                stdscr.addnstr(list_start_row, 0, "(No items to display.)", max_x - 1)

            stdscr.refresh()

        def update_status(message: str) -> None:
            nonlocal status_message
            status_message = message
            render()

        def with_ui_suspended(block: Callable[[], None]) -> None:
            curses.def_prog_mode()
            curses.endwin()
            try:
                block()
            finally:
                curses.reset_prog_mode()
                stdscr.refresh()
                curses.curs_set(0)

        while True:
            render()
            key = stdscr.getch()

            if key in (curses.KEY_UP, ord("k")):
                index = (index - 1) % len(rows) if rows else 0
                if rows:
                    update_status(f"Focused: {rows[index].text.strip()}")
                else:
                    update_status("No items available.")
            elif key in (curses.KEY_DOWN, ord("j")):
                index = (index + 1) % len(rows) if rows else 0
                if rows:
                    update_status(f"Focused: {rows[index].text.strip()}")
                else:
                    update_status("No items available.")
            elif key == ord(" "):
                if rows and rows[index].selectable and rows[index].key:
                    item_type, item_path = rows[index].key
                    if item_type == "pkg":
                        current = pkg_selection.get(item_path, False)
                        pkg_selection[item_path] = not current
                        rows = build_rows()
                        status = "Selected" if pkg_selection[item_path] else "Deselected"
                        suffix = " (invalid, will be skipped)" if item_path in INVALID_PACKAGES else ""
                        update_status(f"{status} {format_relative(item_path, root)}{suffix}")
                    elif item_type == "dmg":
                        current = dmg_selection.get(item_path, False)
                        dmg_selection[item_path] = not current
                        rows = build_rows()
                        status = "Selected" if dmg_selection[item_path] else "Deselected"
                        update_status(f"{status} {format_relative(item_path, root)}")
                else:
                    update_status("Row not selectable.")
            elif key in (curses.KEY_ENTER, ord("\n"), ord("\r")):
                update_status("Selection confirmed.")
                break
            elif key in (ord("d"), ord("D")):
                selected_dmgs = [dmg for dmg in dmgs if dmg_selection.get(dmg.path, False)]
                if not selected_dmgs:
                    update_status("No DMG files selected for unpack.")
                    continue

                total_bytes = sum(dmg.size_bytes for dmg in selected_dmgs)
                update_status(
                    f"{len(selected_dmgs)} DMG(s) selected totaling "
                    f"{human_readable_size(total_bytes)}. Press 'y' to confirm."
                )
                confirm_key = stdscr.getch()
                if confirm_key not in (ord("y"), ord("Y")):
                    update_status("DMG unpack cancelled.")
                    continue

                update_status("Unpacking selected DMGs...")
                was_processed = processed_dmgs.copy()
                try:
                    integrate_dmg_packages(selected_dmgs, update_status, with_ui_suspended)
                except Exception as exc:  # pragma: no cover - defensive
                    update_status(f"DMG unpack failed: {exc}")
                    continue

                rows = build_rows()
                index = 0
                offset = 0

                if processed_dmgs != was_processed:
                    update_status("DMG unpack complete. Review results below.")
                else:
                    update_status("DMG unpack attempted. Review status messages above.")
            elif key == curses.KEY_RESIZE:
                update_status("Resized terminal window.")
            else:
                update_status("Unrecognized key. Refer to instructions.")

    try:
        import curses  # type: ignore

        curses.wrapper(selector)
    except Exception:
        # If curses fails, fall back to best-effort non-interactive behaviour.
        eligible_dmgs = [dmg for dmg in dmgs if dmg_selection.get(dmg.path, False)]
        if eligible_dmgs:
            integrate_dmg_packages(eligible_dmgs, lambda message: print(message, flush=True))

    ordered_paths: List[Path] = []
    for pkg in packages:
        if pkg_selection.get(pkg.path, False) and pkg.path in VALID_PACKAGES:
            ordered_paths.append(pkg.path)
    for dmg in dmgs:
        for pkg_path in extracted_pkg_map.get(dmg.path, []):
            if pkg_selection.get(pkg_path, False) and pkg_path in VALID_PACKAGES:
                ordered_paths.append(pkg_path)
    for pkg_path, selected in pkg_selection.items():
        if selected and pkg_path in VALID_PACKAGES and pkg_path not in ordered_paths:
            ordered_paths.append(pkg_path)

    return ordered_paths


def install_packages(packages: List[Path]) -> Tuple[List[Path], List[Path]]:
    """
    Install the provided packages using the macOS installer command.

    For Distribution packages (with choices), automatically generates and applies
    a choice changes XML to select all components.
    """

    installed: List[Path] = []
    failed: List[Path] = []

    command_prefix: List[str] = []
    if os.geteuid() != 0:
        command_prefix = ["sudo"]

    total = len(packages)
    if total == 0:
        return installed, failed

    root = ROOT_DIR

    # Temp directory for choice XML files
    choices_dir = Path(tempfile.gettempdir()) / "pkgflow" / "choices"
    choices_dir.mkdir(parents=True, exist_ok=True)

    print(f"Queued {total} package(s) for installation.")
    print("Pending queue:")
    for idx, pkg_path in enumerate(packages, start=1):
        display = format_relative(pkg_path, root)
        choices = get_distribution_choices(pkg_path)
        dist_marker = " [Distribution]" if choices else ""
        print(f"  {idx}. {display}{dist_marker}")

    for index, pkg_path in enumerate(packages, start=1):
        remaining = total - index
        display = format_relative(pkg_path, root)

        # Check if this is a Distribution package
        choices = get_distribution_choices(pkg_path)
        choice_xml_path: Optional[Path] = None

        if choices:
            print(f"[{index}/{total}] Installing (Distribution, {len(choices)} choices): {display}", flush=True)
            # Generate choice XML
            choice_xml_path = choices_dir / f"{pkg_path.stem}_choices.plist"
            if not generate_choice_changes_xml(choices, choice_xml_path):
                print(f"[{index}/{total}] Warning: Failed to generate choice XML, trying without...", flush=True)
                choice_xml_path = None
        else:
            print(f"[{index}/{total}] Installing: {display}", flush=True)

        # Build installer command
        installer_cmd = [
            *command_prefix,
            "/usr/sbin/installer",
        ]

        if choice_xml_path and choice_xml_path.exists():
            installer_cmd.extend(["-applyChoiceChangesXML", str(choice_xml_path)])

        installer_cmd.extend(["-pkg", str(pkg_path), "-target", "/"])

        result = subprocess.run(
            installer_cmd,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            check=False,
        )

        # Clean up choice XML
        if choice_xml_path and choice_xml_path.exists():
            try:
                choice_xml_path.unlink()
            except OSError:
                pass

        if result.returncode == 0:
            installed.append(pkg_path)
            print(
                f"[{index}/{total}] Installed successfully. Remaining: {remaining}",
                flush=True,
            )
        else:
            failed.append(pkg_path)
            error_msg = result.stderr.decode(errors="ignore").strip()
            if error_msg:
                print(f"installer error for {display}:\n{error_msg}\n")
            print(
                f"[{index}/{total}] Installation failed. Remaining: {remaining}",
                flush=True,
            )

    # Clean up choices directory
    shutil.rmtree(choices_dir, ignore_errors=True)

    return installed, failed


def main() -> None:
    parser = argparse.ArgumentParser(
        description=(
            "Scan an input directory for .pkg/.dmg files, validate them, and "
            "optionally install chosen packages."
        )
    )
    parser.add_argument(
        "input_dir",
        nargs="?",
        default=".",
        help="Directory to scan (defaults to current working directory).",
    )
    parser.add_argument(
        "--max-depth",
        type=int,
        default=MAX_DEPTH,
        help=f"Maximum subdirectory depth to explore (default: {MAX_DEPTH}).",
    )

    args = parser.parse_args()

    packages, dmgs, root = collect_artifacts(args.input_dir, args.max_depth)

    valid_packages, invalid_packages = validate_packages(
        (pkg.path for pkg in packages), reset=True
    )

    print("Valid packages:")
    for pkg_path in valid_packages:
        print(format_relative(pkg_path, root))

    print("\nInvalid packages:")
    for pkg_path in invalid_packages:
        print(format_relative(pkg_path, root))

    print(
        "\nDMG files discovered: "
        f"{len(dmgs)} (limit: {DMG_SIZE_LIMIT_MB} MB per image by default)"
    )

    selected_packages = select_packages(packages, dmgs, root)

    if not selected_packages:
        print("\nNo packages selected for installation.")
        return

    try:
        ensure_sudo()
    except PermissionError as exc:
        print(f"\n{exc}")
        return

    print("\nInstalling selected packages...")
    installed, failed = install_packages(selected_packages)

    print("\nInstalled packages:")
    for pkg_path in installed:
        print(format_relative(pkg_path, root))

    if failed:
        print("\nFailed installations:")
        for pkg_path in failed:
            print(format_relative(pkg_path, root))


if __name__ == "__main__":
    main()