add sjaur an simple TUI aur client

2026-09-22 20:20:37 +02:00
parent ed15eda0ca
commit 92e5b6e363
14 changed files with 1875 additions and 0 deletions
@@ -0,0 +1,5 @@
 """sjaur — fast, fuzzy TUI search for the Arch User Repository."""
 __version__ = "0.1.0"
 __all__ = ["__version__"]
@@ -0,0 +1,6 @@
 """Allow running the application via ``python -m sjaur``."""
 from .cli import main
 if __name__ == "__main__":
    raise SystemExit(main())
@@ -0,0 +1,244 @@
 """Download, validate, and cache the AUR metadata dump.
 Everything the application stores lives in one directory (by default
 ``~/.cache/sjaur``):
    packages-meta-ext-v1.json           decompressed metadata (the cache)
    packages-meta-ext-v1.json.gz.tmp    download in progress
    packages-meta-ext-v1.json.tmp       decompressed data, being validated
 Cache freshness is decided from the file's modification time: a dump younger
 than :data:`~sjaur.config.CACHE_MAX_AGE` is used as-is, anything older is
 refreshed. The visible cache file is only ever replaced atomically
 (``os.replace``) after a fresh download has been decompressed *and* validated,
 so a failed download can never destroy a working cache — the application
 falls back to the stale dump instead. Temporary files are always cleaned up.
 """
 from __future__ import annotations
 import gzip
 import http.client
 import json
 import os
 import shutil
 import socket
 import urllib.error
 import urllib.request
 from dataclasses import dataclass
 from datetime import datetime, timedelta, timezone
 from pathlib import Path
 from typing import Any, Callable
 from .config import CACHE_MAX_AGE, DUMP_URL, METADATA_FILENAME
 from .errors import CacheCorruptError, DownloadError, SjaurError
 USER_AGENT = "sjaur (AUR metadata client)"
 DOWNLOAD_TIMEOUT_SECONDS = 30.0
 CHUNK_SIZE = 64 * 1024
 COPY_BUFFER_SIZE = 1024 * 1024
 #: Called periodically during downloads with (bytes_done, bytes_total).
 #: ``bytes_total`` is ``-1`` when the server does not announce a size.
 ProgressCallback = Callable[[int, int], None]
 def download_file(
    url: str,
    dest: Path,
    progress: ProgressCallback | None = None,
    timeout: float = DOWNLOAD_TIMEOUT_SECONDS,
 ) -> None:
    """Download ``url`` to ``dest``, following redirects.
    Raises :class:`DownloadError` with a human-readable message on any
    network, HTTP, or I/O problem. The destination is only written to, never
    moved into place; callers are responsible for atomic replacement.
    """
    request = urllib.request.Request(url, headers={"User-Agent": USER_AGENT})
    try:
        with urllib.request.urlopen(request, timeout=timeout) as response:
            # file:// responses carry no status attribute; enforce HTTP codes only.
            status = getattr(response, "status", None)
            if isinstance(status, int) and status != 200:
                raise DownloadError(f"Server returned HTTP {status}.")
            total = -1
            announced = response.headers.get("Content-Length")
            if announced is not None and announced.isdigit():
                total = int(announced)
            done = 0
            with dest.open("wb") as fh:
                while True:
                    chunk = response.read(CHUNK_SIZE)
                    if not chunk:
                        break
                    fh.write(chunk)
                    done += len(chunk)
                    if progress is not None:
                        progress(done, total)
    except urllib.error.HTTPError as error:
        raise DownloadError(f"Server returned HTTP {error.code} {error.reason}.") from error
    except urllib.error.URLError as error:
        raise DownloadError(f"Could not connect to {url}: {error.reason}") from error
    except (TimeoutError, socket.timeout) as error:
        raise DownloadError("Connection timed out.") from error
    except http.client.HTTPException as error:
        raise DownloadError(f"Connection interrupted: {error}") from error
    except OSError as error:
        raise DownloadError(f"Network or disk error: {error.strerror or error}") from error
    if dest.stat().st_size == 0:
        raise DownloadError("Downloaded file is empty.")
@dataclass(frozen=True)
 class MetadataBundle:
    """A loaded metadata snapshot plus information about where it came from."""
    data: list[Any]
    #: One of "cache" (fresh cached dump), "downloaded", or "stale-cache"
    #: (download failed, fell back to the existing dump).
    source: str
    cache_age: timedelta | None
    warning: str | None = None
 def humanize_delta(delta: timedelta) -> str:
    """Render a timedelta as a short human-readable age, e.g. ``3.4 hours``."""
    seconds = max(0, int(delta.total_seconds()))
    if seconds < 120:
        return f"{seconds} seconds"
    minutes = seconds // 60
    if minutes < 120:
        return f"{minutes} minutes"
    hours = minutes / 60
    if hours < 48:
        return f"{hours:.1f} hours"
    return f"{hours / 24:.1f} days"
 class CacheManager:
    """Owns the metadata cache directory and the refresh workflow."""
    def __init__(self, cache_dir: Path, url: str = DUMP_URL) -> None:
        self.cache_dir = cache_dir
        self.url = url
        self.metadata_path = cache_dir / METADATA_FILENAME
    # -- state ----------------------------------------------------------------
    def cache_age(self) -> timedelta | None:
        """Age of the cached dump according to its mtime, or ``None``."""
        try:
            mtime = self.metadata_path.stat().st_mtime
        except OSError:
            return None
        modified = datetime.fromtimestamp(mtime, tz=timezone.utc)
        return datetime.now(timezone.utc) - modified
    def is_fresh(self, max_age: timedelta = CACHE_MAX_AGE) -> bool:
        age = self.cache_age()
        return age is not None and age < max_age
    # -- main entry point -----------------------------------------------------
    def ensure_fresh(
        self,
        *,
        force: bool = False,
        progress: ProgressCallback | None = None,
    ) -> MetadataBundle:
        """Return usable metadata, refreshing the cache when necessary.
        - No cache, ``force=True``, or an expired cache: download a new dump,
          validate it, and atomically replace the old file.
        - A fresh cache is used directly; if it turns out to be corrupt, a
          re-download is attempted.
        - If the download fails but a (stale) cache exists, the stale cache is
          returned with a ``warning`` instead of failing.
        - If there is nothing usable at all, :class:`SjaurError` is raised.
        """
        have_cache = self.metadata_path.is_file()
        age = self.cache_age()
        if have_cache and not force and age is not None and age < CACHE_MAX_AGE:
            try:
                return MetadataBundle(self._load(), "cache", age)
            except CacheCorruptError:
                # Fresh but unreadable: fall through and try to replace it.
                have_cache = False
        try:
            data = self._download_and_replace(progress)
            return MetadataBundle(data, "downloaded", None)
        except SjaurError as error:
            if have_cache:
                try:
                    stale = self._load()
                except CacheCorruptError as corrupt:
                    raise SjaurError(
                        f"Download failed: {error}\n"
                        f"The existing cache is also unusable: {corrupt}"
                    ) from error
                return MetadataBundle(
                    stale,
                    "stale-cache",
                    age,
                    warning=(
                        f"Download failed: {error}\n"
                        f"Using existing cached metadata "
                        f"({humanize_delta(age) if age else 'unknown age'} old)."
                    ),
                )
            raise
    # -- internals ------------------------------------------------------------
    def _download_and_replace(self, progress: ProgressCallback | None) -> list[Any]:
        try:
            self.cache_dir.mkdir(parents=True, exist_ok=True)
        except OSError as error:
            raise SjaurError(
                f"Cannot create cache directory {self.cache_dir}: "
                f"{error.strerror or error}"
            ) from error
        tmp_download = self.cache_dir / f"{METADATA_FILENAME}.gz.tmp"
        tmp_plain = self.cache_dir / f"{METADATA_FILENAME}.tmp"
        try:
            download_file(self.url, tmp_download, progress)
            self._decompress(tmp_download, tmp_plain)
            # Validate the JSON before touching the existing cache.
            data = self._load_from(tmp_plain)
            os.replace(tmp_plain, self.metadata_path)
            return data
        finally:
            for tmp in (tmp_download, tmp_plain):
                try:
                    tmp.unlink(missing_ok=True)
                except OSError:
                    pass
    @staticmethod
    def _decompress(source: Path, dest: Path) -> None:
        try:
            with gzip.open(source, "rb") as src, dest.open("wb") as dst:
                shutil.copyfileobj(src, dst, length=COPY_BUFFER_SIZE)
        except (gzip.BadGzipFile, EOFError, OSError) as error:
            raise CacheCorruptError(f"Downloaded file is not valid gzip data: {error}") from error
    def _load(self) -> list[Any]:
        return self._load_from(self.metadata_path)
    @staticmethod
    def _load_from(path: Path) -> list[Any]:
        try:
            with path.open("r", encoding="utf-8") as fh:
                data = json.load(fh)
        except json.JSONDecodeError as error:
            raise CacheCorruptError(f"{path} is not valid JSON: {error}") from error
        except (OSError, UnicodeDecodeError) as error:
            raise CacheCorruptError(f"Cannot read {path}: {error}") from error
        if not isinstance(data, list):
            raise CacheCorruptError(f"{path} has an unexpected structure (expected a JSON array).")
        return data
@@ -0,0 +1,125 @@
 """Command line entry point for sjaur."""
 from __future__ import annotations
 import argparse
 from pathlib import Path
 from rich.console import Console
 from . import __version__
 from .cache import CacheManager, MetadataBundle, humanize_delta
 from .config import CACHE_MAX_AGE, DUMP_URL, RESULT_LIMIT, get_cache_dir
 from .errors import SjaurError
 from .metadata import parse_packages
 from .search import SearchEngine
 def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="sjaur",
        description="Interactive fuzzy search for the Arch User Repository (AUR).",
        epilog=(
            "Search syntax: type freely for fuzzy search; use '@maintainer' to "
            "fuzzy-search the maintainer field, e.g. '@johndoe firefox'."
        ),
    )
    parser.add_argument("--version", action="version", version=f"sjaur {__version__}")
    parser.add_argument(
        "--update",
        action="store_true",
        help="force a refresh of the cached AUR metadata, then exit",
    )
    parser.add_argument(
        "--cache-dir",
        type=Path,
        default=None,
        help="cache directory override (default: ~/.cache/sjaur)",
    )
    return parser
 def load_metadata(cache: CacheManager, console: Console) -> MetadataBundle:
    """Ensure usable metadata, showing download progress on the console."""
    age = cache.cache_age()
    if age is None or age >= CACHE_MAX_AGE:
        reason = (
            "AUR metadata cache is missing."
            if age is None
            else f"AUR metadata cache is outdated ({humanize_delta(age)} old)."
        )
        console.print(reason)
        console.print("Updating…")
        with console.status("Downloading AUR metadata…") as status:
            def progress(done: int, total: int) -> None:
                if total > 0:
                    status.update(f"Downloading AUR metadata… {done / total * 100:.0f}%")
                else:
                    status.update(f"Downloading AUR metadata… {done / (1024 * 1024):.1f} MiB")
            bundle = cache.ensure_fresh(progress=progress)
        if bundle.warning:
            console.print(f"[yellow]{bundle.warning}[/yellow]")
        return bundle
    return cache.ensure_fresh()
 def main(argv: list[str] | None = None) -> int:
    parser = build_parser()
    args = parser.parse_args(argv)
    console = Console(highlight=False)
    cache = CacheManager(args.cache_dir if args.cache_dir is not None else get_cache_dir())
    try:
        bundle = load_metadata(cache, console)
    except SjaurError as error:
        console.print(f"[bold red]Unable to load AUR metadata:[/bold red] {error}")
        console.print(
            f"\nNo usable metadata is available (source: {DUMP_URL}).\n"
            "Please check your network connection and try again."
        )
        return 1
    except KeyboardInterrupt:
        console.print("\nInterrupted.")
        return 130
    packages, skipped = parse_packages(bundle.data)
    if skipped:
        console.print(f"[yellow]warning:[/yellow] skipped {skipped} malformed package entries")
    if args.update:
        console.print(f"AUR metadata is up to date ({len(packages)} packages).")
        return 0
    if bundle.source == "downloaded":
        console.print(f"Downloaded fresh AUR metadata ({len(packages)} packages).")
    elif bundle.source == "stale-cache":
        console.print(f"Using existing cached metadata ({len(packages)} packages).")
    else:
        console.print(
            f"Using cached AUR metadata "
            f"({humanize_delta(bundle.cache_age) if bundle.cache_age else 'unknown age'} old, "
            f"{len(packages)} packages)."
        )
    with console.status("Building search index…"):
        engine = SearchEngine(packages)
    if bundle.source == "stale-cache":
        cache_note = "stale cache"
    elif bundle.source == "downloaded" or bundle.cache_age is None:
        cache_note = "fresh cache"
    else:
        cache_note = f"cache {humanize_delta(bundle.cache_age)} old"
    # Imported late so the TUI stack only loads once the metadata is ready.
    from .tui.app import SJaurApp
    SJaurApp(engine=engine, cache=cache, cache_note=cache_note).run()
    return 0
 if __name__ == "__main__":
    raise SystemExit(main())
@@ -0,0 +1,40 @@
 """Paths and tunable constants.
 The cache directory follows the XDG convention so that the effective default
 path is ``~/.cache/sjaur/`` on all platforms. ``SJAUR_CACHE_DIR`` overrides it
 (mainly useful for tests), and ``--cache-dir`` overrides both.
 """
 from __future__ import annotations
 import os
 from datetime import timedelta
 from pathlib import Path
 CACHE_DIR_ENV = "SJAUR_CACHE_DIR"
 XDG_CACHE_HOME_ENV = "XDG_CACHE_HOME"
 CACHE_DIR_NAME = "sjaur"
 #: Official extended AUR metadata archive (a "multiinfo"-shaped JSON array).
 DUMP_URL = "https://aur.archlinux.org/packages-meta-ext-v1.json.gz"
 METADATA_FILENAME = "packages-meta-ext-v1.json"
 #: A cached dump younger than this is used as-is; older ones trigger a refresh.
 CACHE_MAX_AGE = timedelta(hours=24)
 #: Delay between the last keystroke and the start of a search (keeps the TUI
 #: responsive while typing).
 SEARCH_DEBOUNCE_SECONDS = 0.18
 #: Maximum number of results shown for one search.
 RESULT_LIMIT = 200
 def get_cache_dir() -> Path:
    """Return the metadata cache directory (default: ``~/.cache/sjaur``)."""
    override = os.environ.get(CACHE_DIR_ENV)
    if override:
        return Path(override).expanduser()
    xdg = os.environ.get(XDG_CACHE_HOME_ENV)
    base = Path(xdg).expanduser() if xdg else Path.home() / ".cache"
    return base / CACHE_DIR_NAME
@@ -0,0 +1,21 @@
 """User-facing error types.
 Every error raised during normal operation is a :class:`SjaurError` carrying a
 human-readable message, so the CLI and TUI never have to show raw tracebacks
 for expected failures (network problems, corrupted caches, permission errors,
 ...).
 """
 from __future__ import annotations
 class SjaurError(Exception):
    """Base class for all expected, user-facing errors."""
 class DownloadError(SjaurError):
    """The metadata download failed (network, HTTP, or connection error)."""
 class CacheCorruptError(SjaurError):
    """The cached or downloaded metadata is unreadable or malformed."""
@@ -0,0 +1,120 @@
 """Installing AUR packages from the local metadata snapshot.
 The installer follows the traditional AUR flow with plain system tools:
    clone the package's git repository (or reuse/update an existing clone)
        -> review prompt (default: show the PKGBUILD)
        -> makepkg -si inside the clone directory
        -> pacman confirmation, sudo password, and build output appear
           directly in the terminal (the TUI suspends for the duration)
 Pure decision logic (clone/update/reuse choices) is separated from subprocess
 execution so it can be unit-tested without touching the network or the
 filesystem. Commands run with inherited stdio, so the user watches the build
 live and sudo/pacman can prompt for input.
 """
 from __future__ import annotations
 import shutil
 import subprocess
 from dataclasses import dataclass
 from pathlib import Path
 from .config import get_cache_dir
 from .errors import SjaurError
 AUR_GIT_URL = "https://aur.archlinux.org/{name}.git"
 #: Extra makepkg arguments: fetch and install missing build dependencies.
 MAKEPKG_DEP_ARGS = ("--syncdeps", "--install")
@dataclass(frozen=True)
 class InstallStep:
    """One planned action of the install workflow."""
    kind: str  # "clone" | "update" | "reuse" | "makepkg"
    name: str  # package name
 def builds_dir() -> Path:
    """Directory where package clones are built (``~/.cache/sjaur/build``)."""
    return get_cache_dir() / "build"
 def plan_install(name: str, target_root: Path, *, refresh: bool = False) -> list[InstallStep]:
    """Compute the steps needed to install ``name`` below ``target_root``.
    Pure function (no I/O except the existence check on the target path),
    which makes the workflow decisions easy to test.
    - No clone yet     -> clone fresh, then build.
    - ``refresh=True`` -> hard-reset and update the existing clone, then build.
    - Otherwise        -> reuse the existing clone, then build.
    """
    target = target_root / name
    if not target.is_dir():
        return [
            InstallStep("clone", name),
            InstallStep("makepkg", name),
        ]
    if refresh:
        return [
            InstallStep("update", name),
            InstallStep("makepkg", name),
        ]
    return [
        InstallStep("reuse", name),
        InstallStep("makepkg", name),
    ]
 def check_environment() -> list[str]:
    """Return human-readable problems that would break an install."""
    problems: list[str] = []
    for tool, hint in (
        ("git", "required to clone AUR package repositories"),
        ("makepkg", "required to build packages"),
        ("pacman", "required to install built packages"),
    ):
        if shutil.which(tool) is None:
            problems.append(f"'{tool}' not found in PATH — {hint}")
    return problems
 def run_step(step: InstallStep, target_root: Path) -> None:
    """Execute a planned step interactively.
    Raises :class:`SjaurError` with a human-readable message on failure.
    The caller is responsible for the interactive review step and for
    suspending the TUI so the subprocess owns the terminal.
    """
    target = target_root / step.name
    if step.kind == "clone":
        _run(["git", "clone", AUR_GIT_URL.format(name=step.name), str(target)])
    elif step.kind == "update":
        _run(["git", "reset", "--hard", "HEAD"], cwd=target)
        _run(["git", "pull", "--ff-only"], cwd=target)
    elif step.kind == "reuse":
        pass  # nothing to do
    elif step.kind == "makepkg":
        if not (target / "PKGBUILD").is_file():
            raise SjaurError(f"{target} does not contain a PKGBUILD.")
        _run(["makepkg", *MAKEPKG_DEP_ARGS], cwd=target)
    else:
        raise SjaurError(f"Unknown install step: {step.kind!r}")
 def _run(cmd: list[str], cwd: Path | None = None) -> None:
    """Run ``cmd`` with inherited stdio so output and prompts stay visible."""
    try:
        result = subprocess.run(cmd, cwd=cwd, check=False)
    except FileNotFoundError:
        raise SjaurError(f"'{cmd[0]}' not found in PATH.") from None
    if result.returncode != 0:
        raise SjaurError(
            f"'{' '.join(cmd)}' failed with exit code {result.returncode} "
            "(see the output above)."
        )
@@ -0,0 +1,137 @@
 """Parse the AUR metadata dump into typed package records.
 The extended dump is a JSON array of "multiinfo" objects. Fields that a
 package does not have are simply omitted by the server, and real-world data
 occasionally contains unexpected shapes — so parsing is deliberately
 tolerant: every field is coerced defensively and entries without a usable
 ``Name`` are skipped (and counted) instead of crashing the application.
 """
 from __future__ import annotations
 from dataclasses import dataclass
 from datetime import datetime, timezone
 from typing import Any
 from .errors import CacheCorruptError
 AUR_WEB_BASE = "https://aur.archlinux.org"
@dataclass(frozen=True, slots=True)
 class Package:
    """One AUR package as described by the extended metadata dump."""
    name: str
    version: str
    description: str
    maintainer: str
    votes: int
    popularity: float
    out_of_date: int | None
    last_modified: int
    first_submitted: int
    url: str
    url_path: str
    package_base: str
    keywords: tuple[str, ...]
    depends: tuple[str, ...]
    make_depends: tuple[str, ...]
    check_depends: tuple[str, ...]
    opt_depends: tuple[str, ...]
    provides: tuple[str, ...]
    conflicts: tuple[str, ...]
    replaces: tuple[str, ...]
    groups: tuple[str, ...]
    licenses: tuple[str, ...]
    @property
    def aur_url(self) -> str:
        return f"{AUR_WEB_BASE}/packages/{self.name}"
    @property
    def is_orphan(self) -> bool:
        return not self.maintainer
 def parse_packages(raw: Any) -> tuple[list[Package], int]:
    """Convert the raw JSON payload into :class:`Package` records.
    Returns ``(packages, skipped_count)``. Raises :class:`CacheCorruptError`
    if the payload is not a list at all.
    """
    if not isinstance(raw, list):
        raise CacheCorruptError("Metadata does not contain a package array.")
    packages: list[Package] = []
    skipped = 0
    for entry in raw:
        if not isinstance(entry, dict):
            skipped += 1
            continue
        name = entry.get("Name")
        if not isinstance(name, str) or not name.strip():
            skipped += 1
            continue
        packages.append(
            Package(
                name=name,
                version=_text(entry.get("Version")),
                description=_text(entry.get("Description")),
                maintainer=_text(entry.get("Maintainer")),
                votes=_int(entry.get("NumVotes")),
                popularity=_float(entry.get("Popularity")),
                out_of_date=_opt_int(entry.get("OutOfDate")),
                last_modified=_int(entry.get("LastModified")),
                first_submitted=_int(entry.get("FirstSubmitted")),
                url=_text(entry.get("URL")),
                url_path=_text(entry.get("URLPath")),
                package_base=_text(entry.get("PackageBase")) or name,
                keywords=_strings(entry.get("Keywords")),
                depends=_strings(entry.get("Depends")),
                make_depends=_strings(entry.get("MakeDepends")),
                check_depends=_strings(entry.get("CheckDepends")),
                opt_depends=_strings(entry.get("OptDepends")),
                provides=_strings(entry.get("Provides")),
                conflicts=_strings(entry.get("Conflicts")),
                replaces=_strings(entry.get("Replaces")),
                groups=_strings(entry.get("Groups")),
                licenses=_strings(entry.get("License")),
            )
        )
    return packages, skipped
 def format_timestamp(timestamp: int | None) -> str:
    """Render a unix timestamp (UTC) as a human-readable string."""
    if not timestamp:
        return "unknown"
    moment = datetime.fromtimestamp(timestamp, tz=timezone.utc)
    return moment.strftime("%Y-%m-%d %H:%M UTC")
 def _text(value: Any) -> str:
    return value if isinstance(value, str) else ""
 def _int(value: Any) -> int:
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        return 0
    return int(value)
 def _opt_int(value: Any) -> int | None:
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        return None
    return int(value)
 def _float(value: Any) -> float:
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        return 0.0
    return float(value)
 def _strings(value: Any) -> tuple[str, ...]:
    if not isinstance(value, list):
        return ()
    return tuple(item for item in value if isinstance(item, str) and item)
@@ -0,0 +1,105 @@
 """Query parsing.
 Raw user input is parsed into a :class:`SearchQuery`. Operators are
 registered in the :data:`OPERATORS` table mapping a typed prefix to the
 :class:`SearchQuery` attribute it feeds. The parser itself is generic, so
 adding future syntax (e.g. ``#keyword`` or ``depends:python``) only requires
 a new table entry plus a field on :class:`SearchQuery` — and until the search
 engine learns about such a field, its value is exposed via
 ``SearchQuery.filters``.
 Currently supported syntax::
    firefox                 free-text fuzzy search
    @max_musterman          maintainer-only fuzzy search
    @max_musterman firefox  maintainer search combined with free text
 Rules for ``@``:
 - Only the **first whitespace-delimited word** after ``@`` is the maintainer
  query; everything else is the normal search (``@john python editor`` -> the
  maintainer query is ``john``, the normal query is ``python editor``).
 - The word may be attached (``@john``) or separated (``@ john``).
 - The first ``@`` operator in the query wins; further ``@`` tokens are ignored.
 """
 from __future__ import annotations
 from dataclasses import dataclass
@dataclass(frozen=True)
 class SearchQuery:
    """A parsed search query.
    ``text``        free-text part of the query, fuzzy matched against all
                    package fields
    ``maintainer``  value of the ``@`` operator, if present
    ``filters``     values of additional/future operators as ``(field, value)``
                    pairs; the engine may grow support for these over time
    """
    text: str = ""
    maintainer: str | None = None
    filters: tuple[tuple[str, str], ...] = ()
 #: Operator registry: typed prefix -> :class:`SearchQuery` field name.
 #: Extending the search syntax starts here.
 OPERATORS: dict[str, str] = {
    "@": "maintainer",
 }
@dataclass(frozen=True)
 class _OperatorMatch:
    prefix: str
    field: str
 def _match_operator(token: str) -> _OperatorMatch | None:
    """Return the operator matching at the start of ``token``, if any.
    Longest prefix wins so that overlapping prefixes (e.g. ``@@`` and ``@``)
    can coexist. A token must *start* with the operator prefix, so ordinary
    words containing the character elsewhere (e.g. an email address) are left
    untouched.
    """
    for prefix in sorted(OPERATORS, key=len, reverse=True):
        if token.startswith(prefix):
            return _OperatorMatch(prefix, OPERATORS[prefix])
    return None
 def parse_query(raw: str) -> SearchQuery:
    """Parse a raw query string into a :class:`SearchQuery`."""
    tokens = raw.split()
    values: dict[str, str] = {}
    extra: list[tuple[str, str]] = []
    rest: list[str] = []
    index = 0
    while index < len(tokens):
        token = tokens[index]
        index += 1
        match = _match_operator(token)
        if match is None:
            rest.append(token)
            continue
        value = token[len(match.prefix) :]
        if not value and index < len(tokens):
            # Bare operator followed by a word: the next token is the value.
            value = tokens[index]
            index += 1
        if not value or match.field in values:
            # Ignore lone operators without a value, and extra occurrences
            # (the first operator of a kind wins).
            continue
        if match.field == "maintainer":
            values[match.field] = value
        else:
            extra.append((match.field, value))
    return SearchQuery(
        text=" ".join(rest),
        maintainer=values.get("maintainer"),
        filters=tuple(extra),
    )
@@ -0,0 +1,429 @@
 """Fuzzy search engine over the local AUR metadata snapshot.
 The engine builds lightweight search structures once at startup (lowercased
 fields and pre-joined "blobs" for low-priority fields) and then answers any
 number of queries from memory. Heavy fuzzy comparisons are delegated to
 rapidfuzz (C++); a fast Python pass combines bulk results and substring
 evidence into a deterministic score, computed only for candidate packages
 (see :meth:`SearchEngine.search`). Nothing here touches the disk or the
 network, so searches are safe to run in worker threads.
 Scoring model
 -------------
 The total score is **name-dominant** and fully deterministic::
    total = name_score * 10 + weighted secondary fields
 ``name_score`` (0..1000) follows the priority ladder::
    exact match          1000
    prefix match         900 - 12 * extra_length   (floor 660)
    word-boundary hit    800 - 0.5 * position
    plain substring      700 - 0.5 * position
    fuzzy match          up to 630 (rapidfuzz, cutoff 60)
 Secondary fields are scored on their own ladders (maintainer 600/520/420,
 description 560/500, keywords & provides 480/430, dependencies 380/330) and
 combined with weights 0.5 / 0.4 / 0.3 / 0.3 / 0.15. Their weighted sum is
 capped below 1000, so it can refine the ranking inside a name tier but can
 never outrank a better name match — exact name matches first, then prefixes,
 strong fuzzy name matches, maintainer, description, and the rest.
 With an ``@maintainer`` operator the maintainer similarity (0..100, fuzzy
 against the maintainer field) is combined with the normal package score::
    total = maintainer_score * 65 + normalized_package_score * 35
 Packages whose maintainer does not reach ``MAINTAINER_MATCH_THRESHOLD``
 similarity are excluded from ``@maintainer`` searches — but the match stays
 fuzzy (``@max_musterman`` still finds ``max_mustermann`` and
 ``max-musterman``).
 Ties are broken by votes (descending), then by name (ascending), so results
 are fully deterministic.
 """
 from __future__ import annotations
 import re
 from dataclasses import dataclass
 from typing import Sequence
 from rapidfuzz import fuzz, process
 from .config import RESULT_LIMIT
 from .metadata import Package
 from .query import SearchQuery
 NAME_WEIGHT = 1.0
 MAINTAINER_WEIGHT = 0.5
 DESCRIPTION_WEIGHT = 0.4
 KEYWORDS_WEIGHT = 0.3
 PROVIDES_WEIGHT = 0.3
 DEPS_WEIGHT = 0.15
 #: Multiplier that makes the name score dominate the weighted secondary sum.
 NAME_DOMINANCE = 10.0
 #: Maximum achievable "normal" package score, used to normalise it to 0..1000
 #: when combined with a maintainer query.
 SECONDARY_MAX = (
    600.0 * MAINTAINER_WEIGHT
    + 560.0 * DESCRIPTION_WEIGHT
    + 480.0 * KEYWORDS_WEIGHT
    + 480.0 * PROVIDES_WEIGHT
    + 380.0 * DEPS_WEIGHT
 )
 NORMAL_MAX_SCORE = 1000.0 * NAME_DOMINANCE + SECONDARY_MAX
 #: How strongly the maintainer score counts in an ``@maintainer`` search.
 MAINTAINER_QUERY_WEIGHT = 0.65
 PACKAGE_QUERY_WEIGHT = 1.0 - MAINTAINER_QUERY_WEIGHT
 #: Minimum rapidfuzz similarity (0..100) for a package to be considered a
 #: match for the ``@maintainer`` operator.
 MAINTAINER_MATCH_THRESHOLD = 55.0
 #: rapidfuzz cutoff for bulk fuzzy passes; anything weaker is treated as no
 #: match at all and saves work in the scoring pass.
 FUZZY_CUTOFF = 60.0
 #: Bulk fuzzy scores are scaled by this factor into the name-score ladder
 #: (a perfect 100 similarity maps to 630, below the plain-substring tier).
 NAME_FUZZY_SCALE = 6.3
 #: Bulk fuzzy scores are scaled by this factor into the description ladder.
 DESCRIPTION_FUZZY_SCALE = 4.5
 #: Minimum number of candidates before the (expensive) description fuzzy
 #: pass is skipped. With more candidates than this, name-dominated matches
 #: already fill the result window.
 DESC_FUZZY_MIN_CANDIDATES = RESULT_LIMIT * 2
 _SEPARATORS = set("-_.")
 _SPACES_RE = re.compile(r"[-_.]+")
@dataclass(frozen=True, slots=True)
 class SearchResult:
    """A package plus its final relevance score."""
    package: Package
    score: float
 def _substring_score(query: str, value: str) -> float:
    """Score how well ``query`` appears in ``value`` (both lowercase), 0..1000.
    Exact match > prefix (strongly preferring shorter values) >
    word-boundary hit > plain substring. The steep prefix penalty keeps
    ``firefox`` clearly above ``firefox-and-a-very-long-suffix``.
    """
    if not query:
        return 0.0
    if value == query:
        return 1000.0
    if value.startswith(query):
        return max(900.0 - 12.0 * min(len(value) - len(query), 20), 660.0)
    position = value.find(query)
    if position > 0:
        boundary = value[position - 1] in _SEPARATORS
        base = 800.0 if boundary else 700.0
        return base - min(position, 100) * 0.5
    return 0.0
 def _contains_any_token(tokens: tuple[str, ...], value: str) -> bool:
    for token in tokens:
        if token in value:
            return True
    return False
 def _contains_all_tokens(tokens: tuple[str, ...], value: str) -> bool:
    for token in tokens:
        if token not in value:
            return False
    return True
 class SearchEngine:
    """Answers fuzzy queries against an in-memory package list."""
    def __init__(self, packages: Sequence[Package]) -> None:
        self.packages: list[Package] = list(packages)
        self._index_by_name = {package.name: index for index, package in enumerate(self.packages)}
        self._names = [package.name.lower() for package in self.packages]
        # Separator-normalised names, so "python editor" also matches
        # the name "python-editor".
        self._names_spaced = [_SPACES_RE.sub(" ", name) for name in self._names]
        self._descriptions = [package.description.lower() for package in self.packages]
        self._maintainers = [package.maintainer.lower() for package in self.packages]
        self._keyword_blobs = [" ".join(p.keywords).lower() for p in self.packages]
        self._provide_blobs = [" ".join(p.provides).lower() for p in self.packages]
        self._deps_blobs = [
            " ".join(
                p.depends + p.make_depends + p.check_depends + p.opt_depends + p.conflicts + p.replaces
            ).lower()
            for p in self.packages
        ]
        # One combined blob per package for the cheap "any signal at all?"
        # prefilter. NUL separators stop matches from spanning fields.
        self._blobs = [
            "\x00".join(
                (
                    self._names[index],
                    self._names_spaced[index],
                    self._descriptions[index],
                    self._maintainers[index],
                    self._keyword_blobs[index],
                    self._provide_blobs[index],
                    self._deps_blobs[index],
                )
            )
            for index in range(len(self.packages))
        ]
        self._popular_order = sorted(
            range(len(self.packages)),
            key=lambda index: (-self.packages[index].votes, self._names[index]),
        )
    # -- public API -----------------------------------------------------------
    def search(self, query: SearchQuery, limit: int = RESULT_LIMIT) -> list[SearchResult]:
        """Search the snapshot for ``query`` and return up to ``limit`` hits.
        Runs in two phases to stay fast on the full AUR snapshot:
        1. A cheap gate pass collects candidate indices using substring
           evidence on the combined blobs, plus bulk fuzzy scores for names
           and (for ``@`` queries) maintainers.
        2. Only the candidates are scored in detail in Python.
        The expensive description fuzzy pass runs only when the candidate set
        is small (weak evidence); with plenty of candidates the top results
        are name-dominated anyway, so skipping it cannot change what is
        shown. Both phases depend only on the query, so results stay
        deterministic.
        """
        text = " ".join(query.text.split()).lower()
        maintainer = query.maintainer.lower() if query.maintainer else None
        if not text and maintainer is None:
            return self.top_packages(limit)
        tokens = tuple(text.split())
        # Bulk fuzzy scores (0..100, aligned with self.packages), computed in
        # C++ over all packages at once. The name pass is whole-string only
        # (partial matches would flood the results with suffix packages).
        name_fuzzy: list[float] | None = None
        if text:
            name_fuzzy = self._bulk_fuzzy(text, self._names, partial=False)
        maintainer_fuzzy: list[float] | None = None
        if maintainer is not None:
            maintainer_fuzzy = self._bulk_fuzzy(maintainer, self._maintainers)
        # -- phase 1: gather candidates cheaply --------------------------------
        candidates: set[int] = set()
        if text:
            blobs = self._blobs
            first = tokens[0]
            rest = tokens[1:]
            for index, blob in enumerate(blobs):
                if text in blob or first in blob:
                    candidates.add(index)
                    continue
                for token in rest:
                    if token in blob:
                        candidates.add(index)
                        break
            for index, score in enumerate(name_fuzzy or ()):
                if score > 0.0:
                    candidates.add(index)
        if maintainer_fuzzy is not None:
            for index, score in enumerate(maintainer_fuzzy):
                if score >= MAINTAINER_MATCH_THRESHOLD:
                    candidates.add(index)
        # Description fuzzy rescue for weak-evidence queries (typos in the
        # description, rare terms): partial_ratio over 120k descriptions is
        # the single most expensive pass, so only run it when needed.
        desc_fuzzy: list[float] | None = None
        if text and len(candidates) < DESC_FUZZY_MIN_CANDIDATES:
            desc_fuzzy = self._bulk_fuzzy(text, self._descriptions)
            for index, score in enumerate(desc_fuzzy):
                if score > 0.0:
                    candidates.add(index)
        # -- phase 2: detailed scoring of the candidates ------------------------
        results: list[SearchResult] = []
        packages = self.packages
        for index in candidates:
            maintainer_score = 0.0
            if maintainer_fuzzy is not None:
                maintainer_score = maintainer_fuzzy[index]
                if maintainer_score < MAINTAINER_MATCH_THRESHOLD:
                    continue
            if not text:
                total = maintainer_score * 10.0
            else:
                package_score = self._package_score(index, text, tokens, name_fuzzy, desc_fuzzy)
                if maintainer_fuzzy is not None:
                    # Candidates are already narrowed down by the maintainer
                    # threshold, so score every one of them: a package whose
                    # maintainer matches must not be hidden by the text gate.
                    total = maintainer_score * MAINTAINER_QUERY_WEIGHT * 10.0 + (
                        package_score / NORMAL_MAX_SCORE
                    ) * 1000.0 * PACKAGE_QUERY_WEIGHT
                else:
                    total = package_score
            if total > 0.0:
                results.append(SearchResult(package=packages[index], score=total))
        results.sort(key=lambda result: (-result.score, -result.package.votes, result.package.name))
        return results[:limit]
    def quick_filter(
        self, results: Sequence[SearchResult], text: str
    ) -> list[SearchResult]:
        """Instant client-side narrowing of previous results while typing.
        Keeps results that still match the typed text — either through
        substring evidence in the metadata blob or because the package name
        still fuzzy-matches (so typo queries like ``firfox`` do not flicker
        to empty while typing). This mirrors the gate conditions of
        :meth:`search`, so it is a conservative subset: the background search
        may still add lower-ranked matches afterwards. Unknown packages
        (e.g. results from an older engine) are dropped. Only the few hundred
        displayed results are examined, so this costs well under a
        millisecond.
        """
        text = " ".join(text.split()).lower()
        if not text:
            return list(results)
        tokens = tuple(text.split())
        scorers = [fuzz.ratio]
        if " " in text:
            scorers.append(fuzz.token_set_ratio)
        blobs = self._blobs
        names = self._names
        index_by_name = self._index_by_name
        narrowed: list[SearchResult] = []
        for result in results:
            index = index_by_name.get(result.package.name)
            if index is None:
                continue
            blob = blobs[index]
            if text in blob or (
                len(tokens) > 1 and all(token in blob for token in tokens)
            ):
                narrowed.append(result)
                continue
            name = names[index]
            if any(scorer(text, name, score_cutoff=FUZZY_CUTOFF) for scorer in scorers):
                narrowed.append(result)
        return narrowed
    def top_packages(self, limit: int = RESULT_LIMIT) -> list[SearchResult]:
        """The most-voted packages, used as the starting view for empty input."""
        return [
            SearchResult(package=self.packages[index], score=0.0)
            for index in self._popular_order[:limit]
        ]
    # -- internals ------------------------------------------------------------
    @staticmethod
    def _bulk_fuzzy(text: str, values: list[str], partial: bool = True) -> list[float]:
        """Best fuzzy similarity of ``text`` against every entry of ``values``.
        Returns plain Python floats in 0..100 (0 at or below the cutoff).
        """
        scorers = [fuzz.ratio]
        if partial:
            scorers.append(fuzz.partial_ratio)
        if " " in text.strip():
            scorers.append(fuzz.token_set_ratio)
        best: list[float] | None = None
        for scorer in scorers:
            scores = process.cdist([text], values, scorer=scorer, score_cutoff=FUZZY_CUTOFF)[0]
            best = scores if best is None else list(map(max, best, scores))
        return [float(score) for score in best] if best is not None else []
    def _package_score(
        self,
        index: int,
        text: str,
        tokens: tuple[str, ...],
        name_fuzzy: list[float] | None,
        desc_fuzzy: list[float] | None,
    ) -> float:
        # The fuzzy lists are only populated when the query has text.
        name_score = self._score_name(text, tokens, index, name_fuzzy[index] if name_fuzzy else 0.0)
        maintainer_score = self._score_maintainer(text, self._maintainers[index])
        description_score = self._score_description(
            text, tokens, self._descriptions[index], desc_fuzzy[index] if desc_fuzzy else 0.0
        )
        keywords_score = self._score_blob(text, tokens, self._keyword_blobs[index])
        provides_score = self._score_blob(text, tokens, self._provide_blobs[index])
        deps_score = self._score_blob(text, tokens, self._deps_blobs[index])
        return (
            name_score * NAME_WEIGHT * NAME_DOMINANCE
            + maintainer_score * MAINTAINER_WEIGHT
            + description_score * DESCRIPTION_WEIGHT
            + keywords_score * KEYWORDS_WEIGHT
            + provides_score * PROVIDES_WEIGHT
            + deps_score * DEPS_WEIGHT
        )
    def _score_name(self, text: str, tokens: tuple[str, ...], index: int, fuzzy: float) -> float:
        direct = _substring_score(text, self._names[index])
        if direct == 0.0:
            direct = _substring_score(text, self._names_spaced[index])
        token_score = 0.0
        if len(tokens) > 1:
            token_score = max(
                max(
                    _substring_score(token, self._names[index]),
                    _substring_score(token, self._names_spaced[index]),
                )
                * 0.9
                for token in tokens
            )
        return max(direct, token_score, fuzzy * NAME_FUZZY_SCALE)
    @staticmethod
    def _score_maintainer(text: str, maintainer: str) -> float:
        if not maintainer:
            return 0.0
        if maintainer == text:
            return 600.0
        if maintainer.startswith(text):
            return 520.0
        if text in maintainer:
            return 420.0
        return 0.0
    @staticmethod
    def _score_description(
        text: str, tokens: tuple[str, ...], description: str, fuzzy: float
    ) -> float:
        if not description:
            return 0.0
        position = description.find(text)
        if position != -1:
            return 560.0 - min(position, 100) * 0.5
        if len(tokens) > 1 and _contains_all_tokens(tokens, description):
            return 500.0
        return fuzzy * DESCRIPTION_FUZZY_SCALE
    @staticmethod
    def _score_blob(text: str, tokens: tuple[str, ...], blob: str) -> float:
        if not blob:
            return 0.0
        if text in blob:
            return 480.0
        if len(tokens) > 1 and _contains_all_tokens(tokens, blob):
            return 430.0
        return 0.0
@@ -0,0 +1 @@
 """Interactive terminal user interface for sjaur."""
@@ -0,0 +1,573 @@
 """The sjaur TUI application.
 Layout::
    ┌──────────────────────────────────────────────┐
    │ Search input                                 │
    ├──────────────────────────┬───────────────────┤
    │ Result list              │ Package details   │
    │                          │                   │
    ├──────────────────────────┴───────────────────┤
    │ status line                                  │
    └──────────────────────────────────────────────┘
    Footer with key bindings
 Searching runs in a background worker thread (debounced and generation-
 guarded so stale results can never overwrite newer ones), keeping the UI
 responsive even though every keystroke queries the full AUR snapshot.
 Installing (ctrl+i) is the opposite: the search worker is left alone while
 the main thread suspends the TUI, hands the terminal to the user, and drives
 git clone / PKGBUILD review / makepkg interactively.
 """
 from __future__ import annotations
 import os
 import subprocess
 import time
 from dataclasses import dataclass
 from rich.text import Text
 from textual import on, work
 from textual.app import App, ComposeResult, SuspendNotSupported
 from textual.binding import Binding
 from textual.containers import Horizontal, Vertical, VerticalScroll
 from textual.message import Message
 from textual.screen import ModalScreen
 from textual.timer import Timer
 from textual.widgets import Footer, Header, Input, ListItem, ListView, Static
 from .. import __version__
 from ..cache import CacheManager
 from ..config import RESULT_LIMIT, SEARCH_DEBOUNCE_SECONDS
 from ..errors import SjaurError
 from ..installer import check_environment, plan_install, run_step
 from ..metadata import Package, parse_packages
 from ..query import parse_query
 from ..search import SearchEngine, SearchResult
 from .details import render_package_details
 class SearchFinished(Message):
    """Posted by the search worker when a search completes."""
    def __init__(
        self, generation: int, key: str, results: list[SearchResult], elapsed: float
    ) -> None:
        self.generation = generation
        self.key = key
        self.results = results
        self.elapsed = elapsed
        super().__init__()
 class MetadataReloaded(Message):
    """Posted by the reload worker after the metadata was refreshed."""
    def __init__(self, engine: SearchEngine, note: str) -> None:
        self.engine = engine
        self.note = note
        super().__init__()
 class MetadataReloadFailed(Message):
    """Posted by the reload worker when refreshing the metadata failed."""
    def __init__(self, error: str) -> None:
        self.error = error
        super().__init__()
 class SearchInput(Input):
    """Search box; ``down`` and ``enter`` jump to the result list."""
    BINDINGS = [
        Binding("down", "app.focus_results", show=False),
        Binding("enter", "app.focus_results", show=False),
    ]
@dataclass(frozen=True)
 class InstallerConfirmed:
    """Result of the install dialog: which package to install, and whether
    an existing clone should be refreshed first."""
    package: Package
    refresh: bool
 class InstallerConfirmScreen(ModalScreen["InstallerConfirmed | None"]):
    """Confirmation dialog for installing the selected package."""
    CSS = """
    InstallerConfirmScreen { align: center middle; }
    #install-panel {
        width: 70%;
        border: solid $warning;
        background: $surface;
        padding: 1 2;
    }
    #install-panel .title { text-style: bold; margin-bottom: 1; }
    #install-panel .warning { text-style: italic; color: $warning; margin-bottom: 1; }
    #install-panel .keys { text-style: dim; }
    """
    BINDINGS = [
        Binding("y", "confirm", "Install"),
        Binding("u", "confirm_refresh", "Update & install"),
        Binding("n,escape", "cancel", "Cancel", priority=True),
    ]
    def __init__(self, package: Package) -> None:
        super().__init__()
        self.package = package
    def compose(self) -> ComposeResult:
        with Vertical(id="install-panel"):
            yield Static(
                f"Install [bold]{self.package.name}[/bold] {self.package.version}?",
                classes="title",
                markup=True,
            )
            if self.package.maintainer:
                yield Static(f"Maintainer: {self.package.maintainer}", classes="keys")
            else:
                yield Static(Text("This package is orphaned.", style="yellow"), classes="keys")
            yield Static(
                "The PKGBUILD will be shown for review before anything is built. "
                "Never build packages you do not understand.",
                classes="warning",
            )
            yield Static(
                "[bold]y[/bold] install · [bold]u[/bold] update clone & install · [bold]n[/bold] cancel",
                classes="keys",
                markup=True,
            )
    def action_confirm(self) -> None:
        self.dismiss(InstallerConfirmed(self.package, refresh=False))
    def action_confirm_refresh(self) -> None:
        self.dismiss(InstallerConfirmed(self.package, refresh=True))
    def action_cancel(self) -> None:
        self.dismiss(None)
 class HelpScreen(ModalScreen[None]):
    """Overlay documenting search syntax and key bindings."""
    CSS = """
    HelpScreen { align: center middle; }
    #help-panel {
        width: 90%;
        max-height: 90%;
        border: solid $primary;
        background: $surface;
        padding: 1 2;
    }
    """
    BINDINGS = [Binding("escape,f1", "app.pop_screen", "Close", priority=True)]
    def compose(self) -> ComposeResult:
        with Vertical(id="help-panel"):
            yield Static(HELP_TEXT, markup=True)
 HELP_TEXT = """\
 [bold]Search syntax[/bold]
  firefox                  fuzzy search across all package fields
  @max_musterman           fuzzy-search the maintainer field
  @max_musterman firefox   maintainer search combined with a normal search
  (only the first word after @ is the maintainer query)
 [bold]Keys[/bold]
  type                     search — results update as you type
  up / down                navigate results
  enter                    open package details (esc returns)
  tab / shift+tab          switch between search field and results
  page up / page down      scroll results or details
  home / end               jump to first / last result
  ctrl+r                   reload AUR metadata now
  i / ctrl+i               install the selected package
  f1                       this help
  esc                      clear search / leave current view
  ctrl+c                   quit
 """
 class SJaurApp(App[None]):
    """Interactive AUR search over the local metadata snapshot."""
    TITLE = f"sjaur {__version__}"
    SUB_TITLE = "AUR search"
    CSS = """
    #content { height: 1fr; }
    #left-pane { width: 45%; min-width: 44; }
    #right-pane { width: 1fr; border-left: solid $primary; }
    #search { height: 3; border-top: none; border-left: none; border-right: none; }
    #results { height: 1fr; }
    #details-scroll { height: 1fr; padding: 0 1; }
    #status { height: 1; text-style: dim; }
    #content.expanded #left-pane { display: none; }
    #content.expanded #right-pane { width: 100%; border-left: none; }
    .result-item { padding: 0 1; }
    """
    BINDINGS = [
        Binding("ctrl+c", "quit", "Quit", priority=True),
        Binding("ctrl+q", "quit", show=False, priority=True),
        Binding("escape", "escape", "Clear/Back", priority=True),
        Binding("f1", "help", "Help"),
        Binding("ctrl+r", "refresh_metadata", "Reload data"),
        # 'i' works whenever the results are focused; ctrl+i is kept for
        # terminals that can distinguish it from Tab.
        Binding("i", "install", "Install"),
        Binding("ctrl+i", "install", show=False),
    ]
    def __init__(
        self,
        engine: SearchEngine,
        cache: CacheManager | None = None,
        cache_note: str = "",
    ) -> None:
        super().__init__()
        self.engine = engine
        self.cache = cache
        self.cache_note = cache_note
        self._generation = 0
        self._debounce_timer: Timer | None = None
        self._expanded = False
        self._current_results: list[SearchResult] = []
        self._search_input: SearchInput | None = None
        #: Final results per finished query text, so backspacing or editing
        #: restores the previous list instantly without re-running the search.
        self._results_cache: dict[str, list[SearchResult]] = {}
    # -- layout ---------------------------------------------------------------
    def compose(self) -> ComposeResult:
        yield Header()
        with Horizontal(id="content"):
            with Vertical(id="left-pane"):
                yield SearchInput(
                    placeholder="Search the AUR…  (@maintainer supported)",
                    id="search",
                )
                yield ListView(id="results")
            with Vertical(id="right-pane"):
                yield VerticalScroll(Static(id="details"), id="details-scroll")
        yield Static(id="status")
        yield Footer()
    def on_mount(self) -> None:
        self._search_input = self.query_one("#search", SearchInput)
        self.query_one("#details", Static).update(
            Text("Start typing to search the AUR.\n\nSelect a package to see its details.", style="dim")
        )
        self._set_status(f"{len(self.engine.packages)} packages indexed" + (f" · {self.cache_note}" if self.cache_note else ""))
        self._search_input.focus()
        self._launch_search()
    # -- search ---------------------------------------------------------------
    @on(Input.Changed, "#search")
    def _on_search_changed(self, event: Input.Changed) -> None:
        # 1) Instant feedback while typing: narrow the displayed results to
        #    those still matching, or restore a cached result set.
        key = " ".join(event.value.split())
        query = parse_query(event.value)
        cached = self._results_cache.get(key)
        if cached is not None:
            self._show_results(cached, 0.0)
        else:
            self._show_results(
                self.engine.quick_filter(self._current_results, query.text), 0.0
            )
        # 2) Full (fuzzy, all-fields) search runs in the background after a
        #    short pause, refining what is on screen.
        self._schedule_search(key)
    def _schedule_search(self, key: str = "") -> None:
        if self._debounce_timer is not None:
            self._debounce_timer.stop()
        # A query whose results are already cached needs no background run.
        if key and key in self._results_cache:
            self._debounce_timer = None
            return
        self._debounce_timer = self.set_timer(SEARCH_DEBOUNCE_SECONDS, self._launch_search)
    def _launch_search(self) -> None:
        if self._debounce_timer is not None:
            self._debounce_timer.stop()
            self._debounce_timer = None
        if self._search_input is None:
            return
        self._generation += 1
        key = " ".join(self._search_input.value.split())
        self._run_search_worker(self._search_input.value, key, self._generation)
        self._set_status("Searching…")
    @work(thread=True, exclusive=True, group="search")
    def _run_search_worker(self, query_text: str, key: str, generation: int) -> None:
        started = time.perf_counter()
        results = self.engine.search(parse_query(query_text), limit=RESULT_LIMIT)
        elapsed = time.perf_counter() - started
        self.post_message(SearchFinished(generation, key, results, elapsed))
    @on(SearchFinished)
    def _on_search_finished(self, event: SearchFinished) -> None:
        if event.generation != self._generation:
            return  # a newer search superseded this one
        # An empty query means the popular-list was (re)built; stale cached
        # entries from a previous engine must go.
        if event.key == "":
            self._results_cache.clear()
        self._results_cache[event.key] = event.results
        self._show_results(event.results, event.elapsed)
    def _show_results(self, results: list[SearchResult], elapsed: float) -> None:
        self._current_results = results
        results_view = self.query_one("#results", ListView)
        previous_index = results_view.index
        results_view.clear()
        results_view.extend([self._build_list_item(result.package) for result in results])
        if results:
            if previous_index is not None and 0 <= previous_index < len(results):
                results_view.index = previous_index
            else:
                results_view.index = 0
            self._show_details(results[results_view.index or 0].package)
        else:
            self._show_details(None)
        if elapsed > 0.0:
            status = f"{len(results)} results · {elapsed * 1000:.0f} ms"
        else:
            status = f"{len(results)} results · live"
        self._set_status(status + (f" · {self.cache_note}" if self.cache_note else ""))
    @staticmethod
    def _build_list_item(package: Package) -> ListItem:
        line = Text()
        line.append(package.name, style="bold")
        if package.version:
            line.append(f"  {package.version}", style="dim")
        if package.votes:
            line.append(f"  +{package.votes}", style="yellow")
        if package.maintainer:
            line.append(f"  @{package.maintainer}", style="dim")
        if package.out_of_date:
            line.append("  OUT OF DATE", style="bold red")
        return ListItem(Static(line), classes="result-item")
    # -- details --------------------------------------------------------------
    @on(ListView.Highlighted, "#results")
    def _on_result_highlighted(self, event: ListView.Highlighted) -> None:
        index = self.query_one("#results", ListView).index
        if index is not None and 0 <= index < len(self._current_results):
            self._show_details(self._current_results[index].package)
    @on(ListView.Selected, "#results")
    def _on_result_selected(self, event: ListView.Selected) -> None:
        self._toggle_details()
    def _toggle_details(self) -> None:
        if self._expanded:
            self._collapse_details()
        else:
            self._expand_details()
    def _expand_details(self) -> None:
        self._expanded = True
        self.query_one("#content", Horizontal).add_class("expanded")
        self.query_one("#details-scroll", VerticalScroll).focus()
    def _collapse_details(self) -> None:
        self._expanded = False
        self.query_one("#content", Horizontal).remove_class("expanded")
        self.query_one("#results", ListView).focus()
    def _show_details(self, package: Package | None) -> None:
        if package is None:
            self.query_one("#details", Static).update(
                Text("No results.", style="dim italic")
            )
            return
        self.query_one("#details", Static).update(render_package_details(package))
        self.query_one("#details-scroll", VerticalScroll).scroll_home(animate=False)
    # -- key actions ----------------------------------------------------------
    def action_focus_results(self) -> None:
        if self._current_results:
            self.query_one("#results", ListView).focus()
    def action_escape(self) -> None:
        if self._expanded:
            self._collapse_details()
            return
        if self._search_input is not None:
            if self._search_input.value:
                self._search_input.value = ""  # triggers a fresh search
            self._search_input.focus()
    def action_help(self) -> None:
        self.push_screen(HelpScreen())
    # -- installing ------------------------------------------------------------
    def _selected_package(self) -> Package | None:
        index = self.query_one("#results", ListView).index
        if index is None or not (0 <= index < len(self._current_results)):
            return None
        return self._current_results[index].package
    def action_install(self) -> None:
        """Open the install dialog for the highlighted package."""
        package = self._selected_package()
        if package is None:
            self._set_status(Text("Nothing selected to install.", style="yellow"))
            return
        self.push_screen(
            InstallerConfirmScreen(package), callback=self._on_install_dialog_closed
        )
    def _on_install_dialog_closed(self, result: InstallerConfirmed | None) -> None:
        if result is not None:
            self._install_package(result.package, refresh=result.refresh)
    def _install_package(self, package: Package, *, refresh: bool) -> None:
        """Run the full install flow with the terminal handed over to makepkg."""
        problems = check_environment()
        if problems:
            self._set_status(Text(f"Cannot install: {'; '.join(problems)}", style="red"))
            return
        try:
            self._install_flow(package, refresh)
        except SjaurError as error:
            self._install_report(package.name, False, str(error))
        except KeyboardInterrupt:
            self._install_report(package.name, False, "Interrupted.")
        finally:
            self._set_status("")
        if self._search_input is not None:
            self._search_input.focus()
    def _install_flow(self, package: Package, refresh: bool) -> None:
        """Suspend the TUI and run clone/review/makepkg on the main thread."""
        from ..installer import builds_dir
        root = builds_dir()
        steps = plan_install(package.name, root, refresh=refresh)
        print(f"\n=== Installing {package.name} {package.version} ===")
        try:
            with self.suspend():
                for step in steps:
                    if step.kind == "clone":
                        print(f"--> Cloning {package.name} from the AUR…")
                    elif step.kind == "update":
                        print(f"--> Updating the existing clone of {package.name}…")
                    elif step.kind == "reuse":
                        print(f"--> Reusing the existing clone of {package.name}.")
                    elif step.kind == "makepkg":
                        print("--> Review the PKGBUILD before building.")
                        if not self._review_pkgbuild(root / package.name):
                            print("Install aborted.")
                            return
                        print(
                            f"--> Building and installing {package.name} "
                            "(you may be asked for your sudo password)…"
                        )
                    run_step(step, root)
        except SuspendNotSupported:
            # Headless environment (e.g. tests): still exercise the flow.
            for step in steps:
                run_step(step, root)
        self._install_report(package.name, True, "")
    @staticmethod
    def _review_pkgbuild(target: object) -> bool:
        """Show the PKGBUILD and ask for confirmation. Returns True to build."""
        from pathlib import Path
        pkgbuild = target / "PKGBUILD" if isinstance(target, Path) else None
        if pkgbuild is not None and pkgbuild.is_file():
            try:
                content = pkgbuild.read_text(encoding="utf-8", errors="replace")
            except OSError:
                content = None
            if content:
                print("\n--- PKGBUILD " + ("-" * 60))
                print(content)
                print("---" + ("-" * 72))
        while True:
            answer = input("Build and install this package? [y/N/e(dit)] ").strip().lower()
            if answer in ("y", "yes"):
                return True
            if answer in ("e", "edit"):
                editor = os.environ.get("EDITOR") or "vi"
                try:
                    subprocess.run([editor, str(pkgbuild)], check=False)
                except OSError as error:
                    print(f"Could not start editor: {error}")
                continue
            return False
    def _install_report(self, name: str, success: bool, message: str) -> None:
        """Report the install result on the suspended terminal.
        Falls back to the status line when there is no usable stdin/stdout
        (e.g. in headless test runs).
        """
        if success:
            text = f"✔ {name} was installed successfully."
        else:
            text = f"✘ Installing {name} failed: {message}"
        try:
            print(f"\n{text}")
            input("Press Enter to return to sjaur…")
        except OSError:
            self._set_status(Text(text, style="green" if success else "red"))
    # -- metadata reload ------------------------------------------------------
    def action_refresh_metadata(self) -> None:
        if self.cache is None:
            return
        self._set_status("Updating AUR metadata…")
        self._reload_metadata_worker()
    @work(thread=True, exclusive=True, group="metadata")
    def _reload_metadata_worker(self) -> None:
        try:
            assert self.cache is not None
            bundle = self.cache.ensure_fresh(force=True)
            packages, _skipped = parse_packages(bundle.data)
            engine = SearchEngine(packages)
            note = f"metadata updated · {len(packages)} packages"
            self.post_message(MetadataReloaded(engine, note))
        except SjaurError as error:
            self.post_message(MetadataReloadFailed(str(error)))
    @on(MetadataReloaded)
    def _on_metadata_reloaded(self, event: MetadataReloaded) -> None:
        self.engine = event.engine
        self.cache_note = event.note
        self._results_cache.clear()
        self._generation += 1  # invalidate in-flight searches against the old engine
        self._set_status(event.note)
        self._launch_search()
    @on(MetadataReloadFailed)
    def _on_metadata_reload_failed(self, event: MetadataReloadFailed) -> None:
        self._set_status(Text(f"Metadata update failed: {event.error}", style="red"))
    # -- helpers --------------------------------------------------------------
    def _set_status(self, content: str | Text) -> None:
        self.query_one("#status", Static).update(content)
@@ -0,0 +1,69 @@
 """Renderable for the package detail pane.
 All information comes from the local metadata snapshot; the detail view never
 performs a network request.
 """
 from __future__ import annotations
 from rich.table import Table
 from rich.text import Text
 from ..metadata import Package, format_timestamp
 def render_package_details(package: Package) -> Table:
    """Build the detail table for ``package``."""
    table = Table.grid(padding=(0, 2))
    table.add_column(style="dim", justify="right", no_wrap=True)
    table.add_column(ratio=1)
    table.add_row("Name", Text(package.name, style="bold"))
    if package.package_base and package.package_base != package.name:
        table.add_row("Package Base", package.package_base)
    table.add_row("Version", package.version or "unknown")
    if package.description:
        table.add_row("Description", package.description)
    if package.maintainer:
        table.add_row("Maintainer", package.maintainer)
    else:
        table.add_row("Maintainer", Text("orphaned", style="yellow"))
    table.add_row("Votes", str(package.votes))
    table.add_row("Popularity", f"{package.popularity:.2f}")
    table.add_row(
        "Last Modified",
        format_timestamp(package.last_modified) if package.last_modified else "unknown",
    )
    table.add_row(
        "First Submitted",
        format_timestamp(package.first_submitted) if package.first_submitted else "unknown",
    )
    if package.out_of_date:
        table.add_row(
            "Out of Date",
            Text(f"yes — since {format_timestamp(package.out_of_date)}", style="bold red"),
        )
    else:
        table.add_row("Out of Date", Text("no", style="green"))
    if package.url:
        table.add_row("URL", package.url)
    table.add_row("AUR Page", package.aur_url)
    if package.licenses:
        table.add_row("Licenses", ", ".join(package.licenses))
    if package.groups:
        table.add_row("Groups", ", ".join(package.groups))
    if package.keywords:
        table.add_row("Keywords", ", ".join(package.keywords))
    _add_section(table, "Depends", package.depends)
    _add_section(table, "Make Depends", package.make_depends)
    _add_section(table, "Check Depends", package.check_depends)
    _add_section(table, "Optional Depends", package.opt_depends)
    _add_section(table, "Provides", package.provides)
    _add_section(table, "Conflicts", package.conflicts)
    _add_section(table, "Replaces", package.replaces)
    return table
 def _add_section(table: Table, label: str, values: tuple[str, ...]) -> None:
    if values:
        table.add_row(label, ", ".join(values))
		`@@ -0,0 +1 @@`
							`"""Interactive terminal user interface for sjaur."""`