CLI search for search.nixos.org ES backend with improved ranking

_README.md

nix-search.py

Command-line search for search.nixos.org's Elasticsearch backend (nixos-search). Python 3 stdlib only.

Builds the same query as the web frontend, plus ranking tweaks (disable with --upstream-ranking):

• Exact/prefix boost on the attribute name: gcc, docker rank first.
• Phrase bonus in descriptions for multi-word queries ("web browser").
• Shorter attribute names win ties (nodejs_26 over nodejs-slim_26).

Usage

./nix-search.py firefox
./nix-search.py -t option services.nginx enable
./nix-search.py -c nixos-25.11 -n 20 gcc
./nix-search.py -- python3 -numpy      # exclude words with -word (after --)
./nix-search.py --json ripgrep         # raw ES response
./nix-search.py --upstream-ranking gcc # frontend ranking

Before / after

Top 5 on nixos-unstable. Upstream = current frontend ranking.

query	upstream	improved
gcc	gccNGPackages_15.gcc, gcc15, gccgo16, gccNGPackages_15.gcc-unwrapped, libgcc	gcc, gcc15, gcc16, gcc13, gccgo
docker	docker-client, docker_25, docker, docker-gc, docker-ls	docker, docker_25, docker-client, docker-gc, docker-ls
node	nodejs-slim_26, nodejs-slim_22, nodejs-slim, graalvmPackages.graalnodejs, nodejs_26	nodejs_26, nodejs_22, nodejs-slim, nodejs-slim_26, nodejs-slim_22
web browser	browser-sync, web-eid-app, haskellPackages.web-output, haskellPackages.web-browser-in-haskell, python313Packages.webdriver-manager	haskellPackages.web-browser-in-haskell, w3m, lynx, surf, w3m-nox
password manager	gorilla-bin, haskellPackages.passman, lesspass-cli, haskellPackages.lesspass, _1password-gui	enpass, pypass, gorilla-bin, prs, lesspass-cli

Limitations

• No popularity data in the index: "text editor" cannot rank vim/emacs first.
• Only current releases and nixos-unstable exist in the cluster.

nix-search.py

#!/usr/bin/env python3
"""Command-line search against the nixos-search Elasticsearch backend.

Replicates the query built by frontend/src/Search.elm plus optional extra
ranking boosts (exact/prefix name match, description phrase match, shortest
attribute name tie-breaker).
"""

import argparse
import base64
import json
import sys
import urllib.request
from typing import Any

DEFAULT_URL = "https://nixos-search-7-1733963800.us-east-1.bonsaisearch.net"
DEFAULT_USERNAME = "aWVSALXpZv"
DEFAULT_PASSWORD = "X8gPHnzL52wFEekuxsfQ9cSh"  # public read-only credentials
DEFAULT_SCHEMA_VERSION = 48

# type -> (main keyword field, weighted search fields, phrase-boost fields)
CONFIG = {
    "package": (
        "package_attr_name",
        [
            ("package_attr_name", 9.0),
            ("package_programs", 9.0),
            ("package_pname", 6.0),
            ("package_description", 1.3),
            ("package_longDescription", 1.0),
            ("flake_name", 0.5),
        ],
        ["package_description^3", "package_longDescription^1"],
    ),
    "option": (
        "option_name",
        [("option_name", 6.0), ("option_description", 1.0)],
        ["option_description^3"],
    ),
}


def word_variants(words: list[str]) -> list[str]:
    """Underscore/dash variants of each word, deduplicated, order preserved."""
    variants = [v for w in words for v in (w.replace("_", "-"), w.replace("-", "_"), w)]
    return list(dict.fromkeys(variants))


def wildcard(field: str, word: str) -> dict[str, Any]:
    return {"wildcard": {field: {"value": f"*{word}*", "case_insensitive": True}}}


def build_body(query: str, size: int, doc_type: str, improved: bool) -> dict[str, Any]:
    main_field, fields, phrase_fields = CONFIG[doc_type]
    words = query.lower().split()
    negative = [w[1:] for w in words if w.startswith("-")]
    positive = [w for w in words if not w.startswith("-")]

    all_fields = [
        f
        for field, score in fields
        for f in (f"{field}^{score}", f"{field}.*^{score * 0.6}")
    ]
    dis_max_queries: list[dict[str, Any]] = [
        {
            "multi_match": {
                "type": "cross_fields",
                "query": " ".join(positive),
                "analyzer": "whitespace",
                "auto_generate_synonyms_phrase_query": False,
                "operator": "and",
                "fields": all_fields,
            }
        }
    ] + [wildcard(main_field, w) for w in word_variants(positive)]

    # Extra ranking signals on top of the frontend query.
    should: list[dict[str, Any]] = []
    if improved and positive:
        joined = "".join(positive)
        # Exact and prefix matches on the keyword name field float canonical
        # attrs like "gcc" above nested/variant attrs that otherwise tie.
        should = [
            {"term": {main_field: {"value": joined, "boost": 100.0}}},
            {
                "prefix": {
                    main_field: {
                        "value": joined,
                        "boost": 20.0,
                        "case_insensitive": True,
                    }
                }
            },
        ]
        if len(positive) > 1:
            # Multi-word queries are usually descriptive ("text editor"):
            # give a fixed bonus (constant_score, so short descriptions do not
            # win via length norms) when the words appear as a phrase.
            should.append(
                {
                    "constant_score": {
                        "filter": {
                            "multi_match": {
                                "type": "phrase",
                                "query": " ".join(positive),
                                "fields": phrase_fields,
                            }
                        },
                        "boost": 80.0,
                    }
                }
            )

    body: dict[str, Any] = {
        "from": 0,
        "size": size,
        "sort": ["_score"],
        "query": {
            "bool": {
                "should": should,
                "filter": [{"term": {"type": {"value": doc_type}}}],
                "must_not": [wildcard(main_field, w) for w in word_variants(negative)],
                "must": [{"dis_max": {"tie_breaker": 0.7, "queries": dis_max_queries}}],
            }
        },
    }
    if improved and doc_type == "package":
        # Tie-breaker: variants (nodejs_22, nodejs-slim, ...) score identically;
        # prefer the shortest attribute name among the top hits.
        body["rescore"] = {
            "window_size": 100,
            "query": {
                "rescore_query": {
                    "function_score": {
                        "script_score": {
                            "script": {
                                "source": "1.0 / doc['package_attr_name'].value.length()"
                            }
                        }
                    }
                },
                "rescore_query_weight": 20.0,
            },
        }
    return body


def search(args: argparse.Namespace) -> dict[str, Any]:
    body = build_body(
        " ".join(args.query), args.size, args.type, not args.upstream_ranking
    )
    index = f"latest-{args.schema_version}-{args.channel}"
    auth = base64.b64encode(f"{args.username}:{args.password}".encode()).decode()
    req = urllib.request.Request(
        f"{args.url}/{index}/_search",
        data=json.dumps(body).encode(),
        headers={"Content-Type": "application/json", "Authorization": f"Basic {auth}"},
        method="POST",
    )
    try:
        with urllib.request.urlopen(req, timeout=30) as resp:
            return json.loads(resp.read())  # type: ignore[no-any-return]
    except urllib.error.HTTPError as e:
        if e.code == 404:
            sys.exit(
                f"error: index {index} not found; "
                "channel may be EOL (try nixos-unstable or a current release)"
            )
        raise


def print_results(data: dict[str, Any], doc_type: str) -> None:
    hits = data["hits"]["hits"]
    total = data["hits"]["total"]["value"]
    print(
        f"# {total} results, showing {len(hits)}, took {data['took']}ms\n",
        file=sys.stderr,
    )
    for hit in hits:
        src = hit["_source"]
        if doc_type == "package":
            name = f"{src['package_attr_name']} ({src.get('package_pversion', '')})"
            desc = src.get("package_description")
        else:
            name = src["option_name"]
            desc = src.get("option_description")
        print(f"{name}  [score {hit['_score']:.1f}]")
        if desc and desc.strip():
            print(f"    {desc.strip().splitlines()[0][:120]}")


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Search packages/options like search.nixos.org"
    )
    parser.add_argument(
        "query",
        nargs="+",
        help="search words; prefix with - to exclude (use `--` before them)",
    )
    parser.add_argument("-c", "--channel", default="nixos-unstable")
    parser.add_argument("-n", "--size", type=int, default=15)
    parser.add_argument(
        "-t", "--type", choices=["package", "option"], default="package"
    )
    parser.add_argument("--url", default=DEFAULT_URL)
    parser.add_argument("--username", default=DEFAULT_USERNAME)
    parser.add_argument("--password", default=DEFAULT_PASSWORD)
    parser.add_argument("--schema-version", type=int, default=DEFAULT_SCHEMA_VERSION)
    parser.add_argument("--json", action="store_true", help="dump raw JSON response")
    parser.add_argument(
        "--upstream-ranking",
        action="store_true",
        help="use the exact frontend query without extra ranking boosts",
    )
    args = parser.parse_args()

    data = search(args)
    if args.json:
        json.dump(data, sys.stdout, indent=2)
    else:
        print_results(data, args.type)


if __name__ == "__main__":
    main()