homebrew to pacman - Map installed Homebrew formulae to CachyOS/Arch pacman packages.

brew_to_pacman.py

#!/usr/bin/env python3
"""
SPDX-FileCopyrightText: 2026 Sam Bacha <@sambacha>

SPDX-License-Identifier: GPL-2.0-or-later

brew_to_pacman.py -- Map installed Homebrew formulae to CachyOS/Arch pacman packages.


    Capture  : `brew leaves` lists explicitly-installed (non-dependency) formulae.
    Universe : pacman sync databases (CachyOS + Arch core/extra) are downloaded and
               parsed locally, so matching never depends on interactive `pacman`
               state and works identically on any machine.
    Matching : a five-tier pipeline, ordered by confidence, tries each formula
               against progressively less certain strategies until one succeeds:

                 1. OVERRIDE    -- curated, hand-verified brew->pacman renames.
                 2. EXACT       -- byte-identical name exists in the universe.
                 3. NORMALIZED  -- name matches after stripping brew-only
                                   decorations (e.g. 'openssl@3' -> 'openssl').
                 4. REPOLOGY    -- cross-distro *project identity* lookup via
                                   the Repology API; resolves true renames that
                                   no string algorithm can (e.g. 'exa' -> 'eza').
                 5. FUZZY       -- last resort: Levenshtein similarity ratio
                                   against the universe, blocked by first
                                   letter, gated by --fuzzy-threshold.

    Emission : an install script (deterministic/semantic matches only), a
               human review report (fuzzy + unmatched), and a JSON audit
               trail covering every formula regardless of outcome.

Requires: Python 3.12+, standard library only (no pip installs, no
python-Levenshtein). Network access is only needed for the sync-database
download and Repology lookups; both are skippable with --offline.
"""

from __future__ import annotations

import argparse
import functools
import gzip
import io
import json
import logging
import re
import shlex
import subprocess
import sys
import tarfile
import time
import urllib.error
import urllib.parse
import urllib.request
from collections import defaultdict
from dataclasses import dataclass
from enum import Enum, auto
from pathlib import Path
from typing import Iterable, Mapping, Protocol, Sequence

if sys.version_info < (3, 12):
    sys.exit("brew_to_pacman.py requires Python 3.12 or newer.")

try:
    import zstandard
except ImportError:
    sys.exit(
        "brew_to_pacman.py requires the 'zstandard' package to read pacman "
        "sync databases (they are zstd-compressed, not gzip).\n"
        "  pip install zstandard"
    )

try:
    from rich.console import Console
    from rich.logging import RichHandler
    from rich.panel import Panel
    from rich.progress import track
    from rich.table import Table
except ImportError:
    sys.exit(
        "brew_to_pacman.py requires the 'rich' package for CLI output.\n"
        "  pip install rich"
    )

LOG = logging.getLogger("brew2pacman")
CONSOLE = Console()


# ---------------------------------------------------------------------------
# Core types
# ---------------------------------------------------------------------------

class MatchTier(Enum):
    """Provenance/confidence rank of a discovered package match, best first."""

    OVERRIDE = auto()
    EXACT = auto()
    NORMALIZED = auto()
    REPOLOGY = auto()
    FUZZY = auto()
    UNMATCHED = auto()


@dataclass(frozen=True, slots=True)
class PackageMatch:
    """Immutable record of how one brew formula resolved to a pacman package.

    `pacman_name` is None and `confidence` is 0.0 iff `tier` is UNMATCHED.
    """

    brew_name: str
    pacman_name: str | None
    tier: MatchTier
    confidence: float


@dataclass(frozen=True, slots=True)
class Config:
    """Immutable runtime configuration assembled once from CLI arguments."""

    output_dir: Path
    sync_db_urls: tuple[str, ...]
    offline: bool
    fuzzy_threshold: float
    repology_user_agent: str
    override_path: Path | None
    brew_binary: str


class Matcher(Protocol):
    """A pure function: brew formula name -> a match, or None if it can't resolve one."""

    def __call__(self, brew_name: str, /) -> PackageMatch | None: ...


# ---------------------------------------------------------------------------
# Phase 1: Capture installed formulae
# ---------------------------------------------------------------------------

def get_installed_formulae(brew_binary: str = "brew") -> tuple[str, ...]:
    """Return the sorted, deduplicated set of user-installed (leaf) formulae.

    `brew leaves` excludes dependencies pulled in transitively, approximating
    "what the user explicitly asked to install" rather than the full
    dependency closure.
    """
    try:
        completed = subprocess.run(
            [brew_binary, "leaves"],
            check=True,
            capture_output=True,
            text=True,
        )
    except FileNotFoundError as exc:
        raise RuntimeError(
            f"'{brew_binary}' not found on PATH; is Homebrew installed?"
        ) from exc
    except subprocess.CalledProcessError as exc:
        raise RuntimeError(f"'brew leaves' failed: {exc.stderr.strip()}") from exc

    return tuple(sorted({line.strip() for line in completed.stdout.splitlines() if line.strip()}))


# ---------------------------------------------------------------------------
# Phase 2: Build the pacman package-name universe from sync databases
# ---------------------------------------------------------------------------

def fetch_sync_database(url: str, *, timeout: float = 15.0) -> bytes:
    """Download a pacman sync database (a compressed tarball) from a repo mirror."""
    request = urllib.request.Request(url, headers={"User-Agent": "brew2pacman/1.0"})
    with urllib.request.urlopen(request, timeout=timeout) as response:  # noqa: S310
        return response.read()


def decompress_sync_database(raw: bytes) -> bytes:
    """Decompress a pacman sync database's tar payload.

    Modern pacman/CachyOS sync databases (`*.db`) are **zstd**-compressed
    tar archives, despite the `.db` extension making them look like plain
    gzip tarballs -- pacman moved off gzip back in the 5.x series. Opening
    one with `gzip`/`tarfile(mode="r:gz")` raises a misleading
    "not a gzip file" `ImportError`/`OSError`. This decompresses with
    `zstandard` first (the current, expected format) and falls back to
    gzip only for the rare legacy/third-party mirror that still serves it.
    """
    try:
        return zstandard.ZstdDecompressor().stream_reader(io.BytesIO(raw)).read()
    except zstandard.ZstdError:
        return gzip.decompress(raw)


def _extract_desc_field(desc_text: str, field_name: str) -> str | None:
    """Pull a single %FIELD% value out of a pacman `desc` file's text."""
    marker = f"%{field_name}%\n"
    if marker not in desc_text:
        return None
    _, _, remainder = desc_text.partition(marker)
    value, _, _ = remainder.partition("\n")
    return value.strip() or None


def parse_sync_database(raw: bytes) -> frozenset[str]:
    """Extract package names from a pacman sync database tarball.

    Each entry is a directory named `<pkgname>-<pkgver>-<rel>/desc`. We read
    the `%NAME%` field from the `desc` file itself rather than splitting the
    directory name, since versions containing hyphens make naive splitting
    unreliable (e.g. `some-pkg-1.2-rc1-1`).
    """
    decompressed = decompress_sync_database(raw)
    names: set[str] = set()
    with tarfile.open(fileobj=io.BytesIO(decompressed), mode="r:") as archive:
        for member in archive.getmembers():
            if not member.name.endswith("/desc"):
                continue
            extracted = archive.extractfile(member)
            if extracted is None:
                continue
            text = extracted.read().decode("utf-8", errors="replace")
            if (name := _extract_desc_field(text, "NAME")) is not None:
                names.add(name)
    return frozenset(names)


@dataclass(frozen=True, slots=True)
class SyncDbLoadResult:
    """Outcome of loading every configured sync database for one run.

    Kept distinct from a bare `frozenset[str]` so that per-mirror successes
    and failures survive into the terminal/markdown/JSON report instead of
    being silently swallowed after logging.
    """

    names: frozenset[str]
    loaded_urls: tuple[str, ...]
    failed_urls: tuple[tuple[str, str], ...]  # (url, reason)


def load_package_universe(sync_db_urls: tuple[str, ...]) -> SyncDbLoadResult:
    """Download and merge every configured sync database into one name set.

    Each mirror is attempted independently and failures are logged and
    skipped rather than aborting the whole run, so a down or renamed mirror
    degrades coverage instead of crashing the tool -- this is what gives
    --sync-db-url its automatic multi-mirror fallback behavior: list
    several mirrors for redundancy and whichever respond successfully are
    unioned together.
    """
    loaded: list[str] = []
    failed: list[tuple[str, str]] = []
    names: set[str] = set()

    for url in track(sync_db_urls, description="Loading sync databases", console=CONSOLE):
        try:
            names |= parse_sync_database(fetch_sync_database(url))
            loaded.append(url)
        except (urllib.error.URLError, tarfile.TarError, zstandard.ZstdError, OSError) as exc:
            reason = f"{type(exc).__name__}: {exc}"
            LOG.warning("Could not load sync database %s: %s", url, reason)
            failed.append((url, reason))

    return SyncDbLoadResult(names=frozenset(names), loaded_urls=tuple(loaded), failed_urls=tuple(failed))


# ---------------------------------------------------------------------------
# Phase 3: Name normalization
# ---------------------------------------------------------------------------

_VERSION_SUFFIX_RE = re.compile(r"@[\d.]+$")
_KNOWN_NOISE_SUFFIXES: tuple[str, ...] = ("-cli", "-bin")


def normalize_formula_name(name: str) -> str:
    """Strip brew-specific decorations that don't change package identity.

    Deliberately conservative: only removes tokens known to be brew-side
    noise (version pins, a couple of common suffixes) rather than guessing.
    Examples: 'openssl@3' -> 'openssl', 'python@3.12' -> 'python'.
    """
    stripped = _VERSION_SUFFIX_RE.sub("", name)
    for suffix in _KNOWN_NOISE_SUFFIXES:
        if stripped.endswith(suffix) and len(stripped) > len(suffix):
            stripped = stripped[: -len(suffix)]
    return stripped


# ---------------------------------------------------------------------------
# Phase 4: Levenshtein similarity (stdlib-only)
# ---------------------------------------------------------------------------

@functools.lru_cache(maxsize=None)
def levenshtein_distance(a: str, b: str) -> int:
    """Compute the Levenshtein edit distance between two strings.

    Iterative DP, O(len(a) * len(b)) time, O(min(len(a), len(b))) memory.
    Memoized because short package-name pairs recur across many comparisons
    within a single run (blocking still leaves repeated pairs across formulae
    that share a first letter).
    """
    if a == b:
        return 0
    if not a:
        return len(b)
    if not b:
        return len(a)
    if len(a) < len(b):
        a, b = b, a  # keep `b` as the shorter string to minimize row width

    previous_row = list(range(len(b) + 1))
    for i, char_a in enumerate(a, start=1):
        current_row = [i] + [0] * len(b)
        for j, char_b in enumerate(b, start=1):
            insertion = current_row[j - 1] + 1
            deletion = previous_row[j] + 1
            substitution = previous_row[j - 1] + (char_a != char_b)
            current_row[j] = min(insertion, deletion, substitution)
        previous_row = current_row
    return previous_row[-1]


def levenshtein_ratio(a: str, b: str) -> float:
    """Normalize edit distance to a similarity ratio in [0.0, 1.0], 1.0 = identical."""
    if not a and not b:
        return 1.0
    return 1.0 - (levenshtein_distance(a, b) / max(len(a), len(b)))


def _block_by_first_letter(universe: frozenset[str]) -> Mapping[str, tuple[str, ...]]:
    """Group package names by first character, for cheap fuzzy-match blocking.

    Avoids an O(n) scan of the whole universe per formula. As a side effect
    it also rejects same-length, different-letter false positives (e.g.
    'grep' vs 'pgrep' at edit distance 1 but different first letters).
    """
    blocks: dict[str, list[str]] = defaultdict(list)
    for name in universe:
        if name:
            blocks[name[0]].append(name)
    return {letter: tuple(names) for letter, names in blocks.items()}


# ---------------------------------------------------------------------------
# Phase 5: Repology cross-distro project lookup
# ---------------------------------------------------------------------------

_RELEVANT_REPOLOGY_REPOS = frozenset({"arch", "arch_community", "arch_aur", "cachyos"})


def query_repology_project(
    name: str,
    *,
    user_agent: str,
    timeout: float = 10.0,
) -> tuple[Mapping[str, object], ...]:
    """Query Repology's project API for every known cross-distro package entry.

    Repology tracks upstream *project identity* across ~120 repositories,
    resolving renames no string-similarity algorithm can discover (e.g.
    'youtube-dl' -> 'yt-dlp'). Returns an empty tuple on 404 or any network
    failure, so callers can treat "no data" and "no match" uniformly.
    """
    url = f"https://repology.org/api/v1/project/{urllib.parse.quote(name)}"
    request = urllib.request.Request(url, headers={"User-Agent": user_agent})
    try:
        with urllib.request.urlopen(request, timeout=timeout) as response:  # noqa: S310
            payload = json.loads(response.read())
    except urllib.error.HTTPError as exc:
        if exc.code != 404:
            LOG.warning("Repology query failed for %s: HTTP %s", name, exc.code)
        return ()
    except (urllib.error.URLError, json.JSONDecodeError, TimeoutError) as exc:
        LOG.warning("Repology query failed for %s: %s", name, exc)
        return ()
    return tuple(payload) if isinstance(payload, list) else ()


def extract_arch_family_names(
    repology_entries: Sequence[Mapping[str, object]],
) -> frozenset[str]:
    """Filter Repology's per-repo entries down to Arch/CachyOS-family binary names."""
    return frozenset(
        binname
        for entry in repology_entries
        if entry.get("repo") in _RELEVANT_REPOLOGY_REPOS
        and isinstance(binname := entry.get("binname"), str)
    )


# ---------------------------------------------------------------------------
# Matcher factories -- each returns a pure str -> PackageMatch | None closure
# ---------------------------------------------------------------------------

def make_override_matcher(overrides: Mapping[str, str]) -> Matcher:
    """Curated, hand-verified table. Runs first: human-verified truth must
    never be shadowed by an automated guess, however confident.
    """

    def matcher(brew_name: str) -> PackageMatch | None:
        if (pacman_name := overrides.get(brew_name)) is not None:
            return PackageMatch(brew_name, pacman_name, MatchTier.OVERRIDE, 1.0)
        return None

    return matcher


def make_exact_matcher(universe: frozenset[str]) -> Matcher:
    """Byte-identical name exists directly in the package universe."""

    def matcher(brew_name: str) -> PackageMatch | None:
        if brew_name in universe:
            return PackageMatch(brew_name, brew_name, MatchTier.EXACT, 1.0)
        return None

    return matcher


def make_normalized_matcher(universe: frozenset[str]) -> Matcher:
    """Name matches once brew-only decorations (versions, known suffixes) are stripped."""

    def matcher(brew_name: str) -> PackageMatch | None:
        normalized = normalize_formula_name(brew_name)
        if normalized != brew_name and normalized in universe:
            return PackageMatch(brew_name, normalized, MatchTier.NORMALIZED, 1.0)
        return None

    return matcher


def make_repology_matcher(universe: frozenset[str], *, user_agent: str) -> Matcher:
    """Cross-distro project-identity lookup for true renames."""

    @functools.cache
    def cached_candidates(name: str) -> frozenset[str]:
        entries = query_repology_project(name, user_agent=user_agent)
        return extract_arch_family_names(entries)

    def matcher(brew_name: str) -> PackageMatch | None:
        candidates = cached_candidates(brew_name) & universe
        if not candidates:
            return None
        chosen = min(candidates, key=len)  # prefer the shortest as most-canonical
        return PackageMatch(brew_name, chosen, MatchTier.REPOLOGY, 1.0)

    return matcher


def make_fuzzy_matcher(universe: frozenset[str], *, threshold: float) -> Matcher:
    """Last resort: Levenshtein similarity, blocked by first letter, threshold-gated.

    Always tagged with its true confidence so `partition_matches_by_confidence`
    can keep statistical guesses out of the auto-install list.
    """
    blocks = _block_by_first_letter(universe)

    def matcher(brew_name: str) -> PackageMatch | None:
        if not brew_name:
            return None
        candidates = blocks.get(brew_name[0], ())
        scored = ((candidate, levenshtein_ratio(brew_name, candidate)) for candidate in candidates)
        best_candidate, best_ratio = max(scored, key=lambda pair: pair[1], default=(None, 0.0))
        if best_candidate is not None and best_ratio >= threshold:
            return PackageMatch(brew_name, best_candidate, MatchTier.FUZZY, best_ratio)
        return None

    return matcher


def build_pipeline(
    universe: frozenset[str],
    overrides: Mapping[str, str],
    *,
    offline: bool,
    fuzzy_threshold: float,
    user_agent: str,
) -> tuple[Matcher, ...]:
    """Assemble the ordered matcher pipeline: override > exact > normalized > repology > fuzzy.

    Order encodes confidence per the design review: curated human knowledge
    outranks deterministic string matches, which outrank semantic API
    lookups, which outrank statistical guessing. Repology is skipped in
    --offline mode since it requires network access.
    """
    pipeline: list[Matcher] = [
        make_override_matcher(overrides),
        make_exact_matcher(universe),
        make_normalized_matcher(universe),
    ]
    if not offline:
        pipeline.append(make_repology_matcher(universe, user_agent=user_agent))
    pipeline.append(make_fuzzy_matcher(universe, threshold=fuzzy_threshold))
    return tuple(pipeline)


def resolve_formula(brew_name: str, pipeline: Sequence[Matcher]) -> PackageMatch:
    """Run one formula through the pipeline; first matcher to return non-None wins.

    Always yields exactly one PackageMatch, falling through to UNMATCHED if
    nothing in the pipeline resolves it.
    """
    for matcher in pipeline:
        if (result := matcher(brew_name)) is not None:
            return result
    return PackageMatch(brew_name, None, MatchTier.UNMATCHED, 0.0)


def resolve_all(formulae: Iterable[str], pipeline: Sequence[Matcher]) -> tuple[PackageMatch, ...]:
    """Resolve every formula against the pipeline, preserving input order."""
    resolver = functools.partial(resolve_formula, pipeline=pipeline)
    return tuple(map(resolver, formulae))


# ---------------------------------------------------------------------------
# Reporting: partition by confidence, then render artifacts
# ---------------------------------------------------------------------------

_AUTO_INSTALL_TIERS = frozenset(
    {MatchTier.OVERRIDE, MatchTier.EXACT, MatchTier.NORMALIZED, MatchTier.REPOLOGY}
)


def partition_matches_by_confidence(
    matches: Sequence[PackageMatch],
) -> tuple[tuple[PackageMatch, ...], tuple[PackageMatch, ...]]:
    """Split matches into (auto-installable, needs-review).

    Only deterministic and semantic-identity tiers are considered safe to
    auto-install; FUZZY and UNMATCHED always require a human look, keeping
    the "generate, then inspect" safety window meaningful rather than nominal.
    """
    auto_install = tuple(m for m in matches if m.tier in _AUTO_INSTALL_TIERS)
    needs_review = tuple(m for m in matches if m.tier not in _AUTO_INSTALL_TIERS)
    return auto_install, needs_review


def render_install_script(matches: Sequence[PackageMatch]) -> str:
    """Render a shell script installing every auto-approved pacman package.

    Every package name is passed through shlex.quote even though pacman
    names are normally shell-safe; this costs nothing and removes an entire
    class of injection risk if the universe or override table is ever
    hand-edited carelessly.
    """
    package_lines = "\n".join(
        f"  {shlex.quote(match.pacman_name)}"
        for match in matches
        if match.pacman_name is not None
    )
    return (
        "#!/usr/bin/env bash\n"
        "set -euo pipefail\n\n"
        "# Generated by brew_to_pacman.py -- review before running.\n"
        f"# {len(matches)} package(s) matched with high confidence.\n\n"
        "sudo pacman -S --needed \\\n"
        f"{package_lines}\n"
    )


def render_review_report(matches: Sequence[PackageMatch]) -> str:
    """Render a human-readable report of formulae needing manual attention."""
    lines = [
        "# Formulae requiring manual review",
        "# Generated by brew_to_pacman.py",
        "#",
        "# FUZZY matches are statistical guesses (Levenshtein similarity) --",
        "# verify the package before installing.",
        "# UNMATCHED formulae had no candidate above threshold anywhere in the",
        "# pipeline; they may be macOS-only, renamed beyond recognition, or",
        "# available only via the AUR.",
        "",
    ]
    for match in matches:
        if match.tier is MatchTier.FUZZY:
            lines.append(
                f"{match.brew_name} -> {match.pacman_name}  "
                f"(FUZZY, confidence={match.confidence:.2f})"
            )
        else:
            lines.append(f"{match.brew_name} -> ??? (UNMATCHED)")
    return "\n".join(lines) + "\n"


def render_json_report(matches: Sequence[PackageMatch]) -> str:
    """Render the full match set as JSON, for auditing or downstream tooling."""
    payload = [
        {
            "brew_name": m.brew_name,
            "pacman_name": m.pacman_name,
            "tier": m.tier.name,
            "confidence": m.confidence,
        }
        for m in matches
    ]
    return json.dumps(payload, indent=2, sort_keys=True) + "\n"


def render_markdown_report(
    matches: Sequence[PackageMatch],
    sync_result: SyncDbLoadResult,
    config: Config,
    duration_seconds: float,
) -> str:
    """Render a standalone markdown summary of one run, for sharing or archiving."""
    tier_counts: dict[str, int] = defaultdict(int)
    for match in matches:
        tier_counts[match.tier.name] += 1
    tier_rows = "\n".join(f"| {tier.name} | {tier_counts.get(tier.name, 0)} |" for tier in MatchTier)

    auto_install, needs_review = partition_matches_by_confidence(matches)
    fuzzy_rows = (
        "\n".join(
            f"| {m.brew_name} | {m.pacman_name} | {m.confidence:.2f} |"
            for m in needs_review
            if m.tier is MatchTier.FUZZY
        )
        or "| _none_ | | |"
    )
    unmatched_list = (
        "\n".join(f"- {m.brew_name}" for m in needs_review if m.tier is MatchTier.UNMATCHED) or "_none_"
    )
    mirror_lines = "\n".join(f"- ✅ {url}" for url in sync_result.loaded_urls) or "_none loaded_"
    failed_lines = (
        "\n".join(f"- ❌ {url} — {reason}" for url, reason in sync_result.failed_urls) or "_none_"
    )

    return f"""# brew_to_pacman report

- Mode: {"offline" if config.offline else "online"}
- Duration: {duration_seconds:.2f}s
- Fuzzy threshold: {config.fuzzy_threshold}
- Formulae processed: {len(matches)}
- Package universe size: {len(sync_result.names):,}

## Sync database mirrors

**Loaded:**
{mirror_lines}

**Failed:**
{failed_lines}

## Results by tier

| Tier | Count |
|------|-------|
{tier_rows}

## Needs review — fuzzy matches

| Formula | Suggested package | Confidence |
|---------|--------------------|------------|
{fuzzy_rows}

## Unmatched

{unmatched_list}
"""


def render_terminal_summary(
    matches: Sequence[PackageMatch],
    sync_result: SyncDbLoadResult,
    config: Config,
    duration_seconds: float,
) -> None:
    """Print a colored summary panel and tier breakdown table to the terminal."""
    auto_install, needs_review = partition_matches_by_confidence(matches)
    mode = "[yellow]offline[/yellow]" if config.offline else "[green]online[/green]"
    header = (
        f"[bold]{len(matches)}[/bold] formulae processed in [bold]{duration_seconds:.2f}s[/bold] "
        f"({mode} mode, fuzzy threshold {config.fuzzy_threshold})"
    )
    if config.offline:
        sync_line = "sync db: [dim]skipped (offline mode)[/dim]"
    else:
        sync_line = (
            f"sync db: [green]{len(sync_result.loaded_urls)} mirror(s) loaded[/green], "
            f"[red]{len(sync_result.failed_urls)} failed[/red], "
            f"[bold]{len(sync_result.names):,}[/bold] package names available"
        )
    CONSOLE.print(Panel.fit(f"{header}\n{sync_line}", title="brew_to_pacman summary", border_style="blue"))

    tier_style = {
        MatchTier.OVERRIDE: "magenta",
        MatchTier.EXACT: "bold green",
        MatchTier.NORMALIZED: "green",
        MatchTier.REPOLOGY: "cyan",
        MatchTier.FUZZY: "yellow",
        MatchTier.UNMATCHED: "bold red",
    }
    table = Table(title="Results by tier", box=None)
    table.add_column("Tier")
    table.add_column("Count", justify="right")
    for tier in MatchTier:
        count = sum(1 for m in matches if m.tier is tier)
        style = tier_style[tier]
        table.add_row(f"[{style}]{tier.name}[/{style}]", str(count))
    CONSOLE.print(table)

    fuzzy_matches = [m for m in needs_review if m.tier is MatchTier.FUZZY]
    if fuzzy_matches:
        review_table = Table(title="Needs review (fuzzy matches)", box=None)
        review_table.add_column("Formula")
        review_table.add_column("Suggested package")
        review_table.add_column("Confidence", justify="right")
        for m in fuzzy_matches:
            review_table.add_row(m.brew_name, m.pacman_name or "", f"{m.confidence:.2f}")
        CONSOLE.print(review_table)


def write_artifacts(
    output_dir: Path,
    matches: Sequence[PackageMatch],
    sync_result: SyncDbLoadResult,
    config: Config,
    duration_seconds: float,
) -> tuple[Path, Path, Path, Path]:
    """Write the install script, review report, JSON audit trail, and markdown report to disk."""
    output_dir.mkdir(parents=True, exist_ok=True)
    auto_install, needs_review = partition_matches_by_confidence(matches)

    install_path = output_dir / "install-cachyos.sh"
    review_path = output_dir / "review-manual.txt"
    json_path = output_dir / "match-report.json"
    markdown_path = output_dir / "report.md"

    install_path.write_text(render_install_script(auto_install), encoding="utf-8")
    install_path.chmod(0o755)
    review_path.write_text(render_review_report(needs_review), encoding="utf-8")
    json_path.write_text(render_json_report(matches), encoding="utf-8")
    markdown_path.write_text(
        render_markdown_report(matches, sync_result, config, duration_seconds), encoding="utf-8"
    )

    return install_path, review_path, json_path, markdown_path


# ---------------------------------------------------------------------------
# Overrides + defaults
# ---------------------------------------------------------------------------

_DEFAULT_OVERRIDES: Mapping[str, str] = {
    "node": "nodejs",
    "exa": "eza",
    "youtube-dl": "yt-dlp",
    "openssl@3": "openssl",
    "openssl@1.1": "openssl",
    "python@3.11": "python",
    "python@3.12": "python",
    "gnu-sed": "sed",
}

# NOTE: verify these mirror URLs are still current before relying on them --
# mirror layouts and hostnames do change over time. Override with
# --sync-db-url (repeatable) to point at different mirrors entirely.
#
# Each repo below is listed against more than one mirror on purpose: mirrors
# are unioned together, and load_package_universe() tolerates individual
# failures, so if e.g. mirror.cachyos.org is down, the geo-mirror entry
# still supplies coverage for that repo instead of the whole repo silently
# dropping out.
_DEFAULT_SYNC_DB_URLS: tuple[str, ...] = (
    # CachyOS repos (primary migration target)
    "https://mirror.cachyos.org/repo/x86_64/cachyos/cachyos.db",
    "https://mirror.cachyos.org/repo/x86_64/cachyos-core/cachyos-core.db",
    "https://mirror.cachyos.org/repo/x86_64/cachyos-extra/cachyos-extra.db",
    "https://geo-mirror.chaotic.cx/cachyos/cachyos/x86_64/cachyos.db",
    # Arch upstream -- CachyOS is Arch-based and mirrors most package names,
    # so this is a broad-coverage fallback layer, tried second.
    "https://geo.mirror.pkgbuild.com/core/os/x86_64/core.db",
    "https://geo.mirror.pkgbuild.com/extra/os/x86_64/extra.db",
    "https://america.mirror.pkgbuild.com/core/os/x86_64/core.db",
    "https://america.mirror.pkgbuild.com/extra/os/x86_64/extra.db",
)


def load_overrides(path: Path | None) -> Mapping[str, str]:
    """Load the curated override table, merging any user-supplied entries over defaults.

    A user-supplied JSON file (flat string->string object) always wins over
    built-in defaults, so users can correct or extend the table without
    editing this script.
    """
    if path is None:
        return dict(_DEFAULT_OVERRIDES)
    try:
        user_overrides = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exc:
        raise RuntimeError(f"Could not load overrides file {path}: {exc}") from exc
    if not isinstance(user_overrides, dict):
        raise RuntimeError(f"Overrides file {path} must contain a JSON object")
    return {**_DEFAULT_OVERRIDES, **user_overrides}


# ---------------------------------------------------------------------------
# CLI + entry point
# ---------------------------------------------------------------------------

def parse_args(argv: Sequence[str] | None = None) -> Config:
    """Parse command-line arguments into an immutable Config."""
    parser = argparse.ArgumentParser(
        description="Map installed Homebrew formulae to CachyOS/Arch pacman packages.",
    )
    parser.add_argument(
        "--output-dir",
        type=Path,
        default=Path("./brew2pacman-out"),
        help="Directory to write install-cachyos.sh, review-manual.txt, match-report.json",
    )
    parser.add_argument(
        "--sync-db-url",
        dest="sync_db_urls",
        action="append",
        help="Pacman sync database URL to consult (repeatable). "
        "Defaults to CachyOS + Arch core/extra mirrors.",
    )
    parser.add_argument(
        "--offline",
        action="store_true",
        help="Skip network calls (sync-db download, Repology); use only curated overrides.",
    )
    parser.add_argument(
        "--fuzzy-threshold",
        type=float,
        default=0.85,
        help="Minimum Levenshtein similarity ratio (0.0-1.0) to accept a FUZZY match.",
    )
    parser.add_argument(
        "--overrides",
        type=Path,
        default=None,
        help="Path to a JSON file of brew_name->pacman_name overrides, merged over defaults.",
    )
    parser.add_argument(
        "--brew-binary",
        default="brew",
        help="Path to the brew executable (default: 'brew' resolved from PATH).",
    )
    parser.add_argument("-v", "--verbose", action="store_true", help="Enable debug logging.")
    args = parser.parse_args(argv)

    logging.basicConfig(
        level=logging.DEBUG if args.verbose else logging.INFO,
        format="%(message)s",
        handlers=[RichHandler(console=CONSOLE, show_time=False, show_path=False, markup=True)],
    )

    return Config(
        output_dir=args.output_dir,
        sync_db_urls=tuple(args.sync_db_urls or _DEFAULT_SYNC_DB_URLS),
        offline=args.offline,
        fuzzy_threshold=args.fuzzy_threshold,
        repology_user_agent="brew2pacman/1.0 (+https://github.com/example/brew2pacman)",
        override_path=args.overrides,
        brew_binary=args.brew_binary,
    )


def main(argv: Sequence[str] | None = None) -> int:
    """Entry point: capture, resolve, and emit artifacts end-to-end."""
    config = parse_args(argv)

    try:
        formulae = get_installed_formulae(config.brew_binary)
    except RuntimeError as exc:
        LOG.error(str(exc))
        return 1
    LOG.info("Captured %d installed formula(e) from Homebrew.", len(formulae))

    overrides = load_overrides(config.override_path)

    universe: frozenset[str] = frozenset()
    if not config.offline:
        universe = load_package_universe(config.sync_db_urls)
        LOG.info("Loaded %d package name(s) from sync databases.", len(universe))
    if not universe:
        LOG.warning(
            "Package universe is empty (offline mode or download failure); "
            "only curated overrides will resolve."
        )

    pipeline = build_pipeline(
        universe,
        overrides,
        offline=config.offline,
        fuzzy_threshold=config.fuzzy_threshold,
        user_agent=config.repology_user_agent,
    )
    matches = resolve_all(formulae, pipeline)

    install_path, review_path, json_path = write_artifacts(config.output_dir, matches)

    auto_install, needs_review = partition_matches_by_confidence(matches)
    LOG.info(
        "Resolved %d/%d formula(e) with high confidence; %d need manual review.",
        len(auto_install),
        len(matches),
        len(needs_review),
    )
    LOG.info("Wrote: %s, %s, %s", install_path, review_path, json_path)
    return 0


if __name__ == "__main__":
    sys.exit(main())