chemenu/tools/chemenu/search/service.py

"""The search core, with no CLI attached.

`run_search()` and the corpus loader used to live in `commands/search.py`,
which imports `typer` at module level and `rich` through `_util`. Any
in-process caller therefore dragged the whole CLI head in behind it - which
made the "two third-party packages" the read core actually needs (`yaml`,
`jsonschema`) an accounting fiction rather than a fact about the import graph.

Nothing here imports `typer`, `rich`, or anything under `chemenu.commands`.
That is the boundary, and it is worth keeping: `commands/search.py` is now the
adapter that turns these values into terminal output and these exceptions into
exit codes, and the MCP server (Gitea #19) is a second adapter over the same
functions rather than a second implementation of them.

Errors are raised, never printed: `PredicateError` for a bad `--field`,
`RipgrepMissing`/`RipgrepFailed` for the backend. All of them are
`ChemenuError` - see `chemenu/errors.py`.
"""
from __future__ import annotations

from pathlib import Path

from chemenu import config
from chemenu.frontmatter_io import read_page_with_error
from chemenu.kb_scan import iter_kb_pages
from chemenu.page import Page
from chemenu.search import filters
from chemenu.search.base import page_key
from chemenu.search.fuse import reciprocal_rank_fusion
from chemenu.search.ripgrep import build_hit
from chemenu.search.types import SearchHit, SearchQuery, SearchResult


def load_pages_by_path(kb_dir: Path | None = None, root: Path | None = None) -> dict[str, Page]:
    """Every page under `kb/`, keyed by repo-relative path.

    Path-keyed rather than title-keyed on purpose: `load_kb_pages()` drops one
    of two pages sharing a stem, and search should still find both - a
    duplicate title is a lint finding, not a reason to hide a page.
    """
    kb_dir = kb_dir or config.KB_DIR
    root = root or config.ROOT
    pages: dict[str, Page] = {}
    for path in iter_kb_pages(kb_dir):
        frontmatter, body, error = read_page_with_error(path)
        pages[page_key(path, root)] = Page(
            path=path, frontmatter=frontmatter, body=body, frontmatter_error=error
        )
    return pages


def unreadable_pages(pages: dict[str, Page]) -> list[dict[str, str]]:
    """The pages whose frontmatter could not be used, as `{path, reason}`.

    Reported rather than swallowed. Such a page has no `kind` and no readable
    frontmatter at all, so it silently drops out of every positive `--field`
    predicate - including the sweeps that exist to find pages in exactly that
    state (`!sources`, `provenance=general`). Saying nothing makes it look
    like a page that did not match; an empty block is excluded, because a
    page can legitimately carry one.
    """
    return [
        {"path": key, "reason": page.frontmatter_error}
        for key, page in sorted(pages.items())
        if page.frontmatter_error and page.frontmatter_error != "empty frontmatter block"
    ]


def _sort_key(hit: SearchHit, field: str):
    value = hit.as_dict().get(field)
    if value is None:
        # Missing values sort last in either direction rather than crashing on
        # a None comparison.
        return (1, "")
    if isinstance(value, (int, float)):
        return (0, value)
    return (0, str(value).lower())


def sort_hits(hits: list[SearchHit], sort: str | None) -> list[SearchHit]:
    """Sort by a hit field. A leading `-` reverses, e.g. `--sort -modified`."""
    if not sort:
        return hits
    descending = sort.startswith("-")
    field = sort.lstrip("-")
    ordered = sorted(hits, key=lambda h: _sort_key(h, field), reverse=descending)
    return ordered


def run_search(
    query: SearchQuery,
    pages: dict[str, Page],
    backends: list,
    kb_dir: Path | None = None,
) -> SearchResult:
    """Answer a query. Pure: no I/O beyond whatever a backend does.

    Returns the truncated hits *and* the number there were before the limit,
    because the caller cannot recover the second from the first - see
    `SearchResult`.
    """
    filters.validate_fields(query.predicates, pages)

    if query.text:
        rankings = [backend.search(query, pages) for backend in backends]
        hits = rankings[0] if len(rankings) == 1 else reciprocal_rank_fusion(rankings)
        allowed = filters.apply_predicates(pages, query.predicates, kb_dir)
        hits = [hit for hit in hits if hit.path in allowed]
    else:
        selected = filters.apply_predicates(pages, query.predicates, kb_dir)
        hits = [
            build_hit(page, key, [], query, backend="frontmatter", kb_dir=kb_dir)
            for key, page in selected.items()
        ]
        hits.sort(key=lambda h: h.title.lower())

    hits = sort_hits(hits, query.sort)
    total = len(hits)
    return SearchResult(
        hits=hits[: query.limit] if query.limit else hits,
        total=total,
        limit=query.limit,
    )