From 92e5b6e3631cb6e8dc3a6cc1eb64f54b25910159 Mon Sep 17 00:00:00 2001 From: jonas H Date: Tue, 22 Sep 2026 20:20:37 +0200 Subject: [PATCH] add sjaur an simple TUI aur client --- prg/sjaur/src/sjaur/__init__.py | 5 + prg/sjaur/src/sjaur/__main__.py | 6 + prg/sjaur/src/sjaur/cache.py | 244 ++++++++++++ prg/sjaur/src/sjaur/cli.py | 125 ++++++ prg/sjaur/src/sjaur/config.py | 40 ++ prg/sjaur/src/sjaur/errors.py | 21 + prg/sjaur/src/sjaur/installer.py | 120 ++++++ prg/sjaur/src/sjaur/metadata.py | 137 +++++++ prg/sjaur/src/sjaur/py.typed | 0 prg/sjaur/src/sjaur/query.py | 105 +++++ prg/sjaur/src/sjaur/search.py | 429 +++++++++++++++++++++ prg/sjaur/src/sjaur/tui/__init__.py | 1 + prg/sjaur/src/sjaur/tui/app.py | 573 ++++++++++++++++++++++++++++ prg/sjaur/src/sjaur/tui/details.py | 69 ++++ 14 files changed, 1875 insertions(+) create mode 100644 prg/sjaur/src/sjaur/__init__.py create mode 100644 prg/sjaur/src/sjaur/__main__.py create mode 100644 prg/sjaur/src/sjaur/cache.py create mode 100644 prg/sjaur/src/sjaur/cli.py create mode 100644 prg/sjaur/src/sjaur/config.py create mode 100644 prg/sjaur/src/sjaur/errors.py create mode 100644 prg/sjaur/src/sjaur/installer.py create mode 100644 prg/sjaur/src/sjaur/metadata.py create mode 100644 prg/sjaur/src/sjaur/py.typed create mode 100644 prg/sjaur/src/sjaur/query.py create mode 100644 prg/sjaur/src/sjaur/search.py create mode 100644 prg/sjaur/src/sjaur/tui/__init__.py create mode 100644 prg/sjaur/src/sjaur/tui/app.py create mode 100644 prg/sjaur/src/sjaur/tui/details.py diff --git a/prg/sjaur/src/sjaur/__init__.py b/prg/sjaur/src/sjaur/__init__.py new file mode 100644 index 0000000..9ef9974 --- /dev/null +++ b/prg/sjaur/src/sjaur/__init__.py @@ -0,0 +1,5 @@ +"""sjaur — fast, fuzzy TUI search for the Arch User Repository.""" + +__version__ = "0.1.0" + +__all__ = ["__version__"] diff --git a/prg/sjaur/src/sjaur/__main__.py b/prg/sjaur/src/sjaur/__main__.py new file mode 100644 index 0000000..4545789 --- /dev/null +++ b/prg/sjaur/src/sjaur/__main__.py @@ -0,0 +1,6 @@ +"""Allow running the application via ``python -m sjaur``.""" + +from .cli import main + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/prg/sjaur/src/sjaur/cache.py b/prg/sjaur/src/sjaur/cache.py new file mode 100644 index 0000000..bfdf3cb --- /dev/null +++ b/prg/sjaur/src/sjaur/cache.py @@ -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 diff --git a/prg/sjaur/src/sjaur/cli.py b/prg/sjaur/src/sjaur/cli.py new file mode 100644 index 0000000..072518a --- /dev/null +++ b/prg/sjaur/src/sjaur/cli.py @@ -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()) diff --git a/prg/sjaur/src/sjaur/config.py b/prg/sjaur/src/sjaur/config.py new file mode 100644 index 0000000..f493afe --- /dev/null +++ b/prg/sjaur/src/sjaur/config.py @@ -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 diff --git a/prg/sjaur/src/sjaur/errors.py b/prg/sjaur/src/sjaur/errors.py new file mode 100644 index 0000000..350d306 --- /dev/null +++ b/prg/sjaur/src/sjaur/errors.py @@ -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.""" diff --git a/prg/sjaur/src/sjaur/installer.py b/prg/sjaur/src/sjaur/installer.py new file mode 100644 index 0000000..38bc11c --- /dev/null +++ b/prg/sjaur/src/sjaur/installer.py @@ -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)." + ) diff --git a/prg/sjaur/src/sjaur/metadata.py b/prg/sjaur/src/sjaur/metadata.py new file mode 100644 index 0000000..c605f1f --- /dev/null +++ b/prg/sjaur/src/sjaur/metadata.py @@ -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) diff --git a/prg/sjaur/src/sjaur/py.typed b/prg/sjaur/src/sjaur/py.typed new file mode 100644 index 0000000..e69de29 diff --git a/prg/sjaur/src/sjaur/query.py b/prg/sjaur/src/sjaur/query.py new file mode 100644 index 0000000..953a91c --- /dev/null +++ b/prg/sjaur/src/sjaur/query.py @@ -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), + ) diff --git a/prg/sjaur/src/sjaur/search.py b/prg/sjaur/src/sjaur/search.py new file mode 100644 index 0000000..079f663 --- /dev/null +++ b/prg/sjaur/src/sjaur/search.py @@ -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 diff --git a/prg/sjaur/src/sjaur/tui/__init__.py b/prg/sjaur/src/sjaur/tui/__init__.py new file mode 100644 index 0000000..8fcfef7 --- /dev/null +++ b/prg/sjaur/src/sjaur/tui/__init__.py @@ -0,0 +1 @@ +"""Interactive terminal user interface for sjaur.""" diff --git a/prg/sjaur/src/sjaur/tui/app.py b/prg/sjaur/src/sjaur/tui/app.py new file mode 100644 index 0000000..dad46b8 --- /dev/null +++ b/prg/sjaur/src/sjaur/tui/app.py @@ -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) diff --git a/prg/sjaur/src/sjaur/tui/details.py b/prg/sjaur/src/sjaur/tui/details.py new file mode 100644 index 0000000..4aa6bd5 --- /dev/null +++ b/prg/sjaur/src/sjaur/tui/details.py @@ -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))