67 lines
2.4 KiB
Python
67 lines
2.4 KiB
Python
"""Search routes for the EPUB search web UI."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from dataclasses import replace
|
|
from time import perf_counter
|
|
from typing import TYPE_CHECKING, Annotated
|
|
|
|
from fastapi import APIRouter, Form, Request
|
|
from fastapi.responses import HTMLResponse
|
|
|
|
from python.ebook_search.answer import answer_query
|
|
from python.ebook_search.api.web import templates
|
|
from python.ebook_search.search import search_ebooks
|
|
from python.ebook_search.timing import runtime_step_from_start
|
|
|
|
if TYPE_CHECKING:
|
|
from fastapi import FastAPI
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
def register_search_routes(app: FastAPI) -> None:
|
|
"""Register search routes on the app."""
|
|
app.include_router(router)
|
|
|
|
|
|
@router.post("/search", response_class=HTMLResponse)
|
|
def search(
|
|
request: Request,
|
|
query: Annotated[str, Form()],
|
|
rerank: Annotated[str | None, Form()] = None,
|
|
) -> HTMLResponse:
|
|
"""Run a search and render HTMX results."""
|
|
try:
|
|
response = search_ebooks(request.app.state.engine, query, request.app.state.config, rerank=rerank == "true")
|
|
except Exception as error:
|
|
logger.exception("ebook_search_request_failed")
|
|
return templates.TemplateResponse(request, "partials/error.html", {"message": str(error)}, status_code=500)
|
|
|
|
answer_start = perf_counter()
|
|
if request.app.state.config.answer_enabled:
|
|
try:
|
|
answer = answer_query(query, response.results, request.app.state.config)
|
|
except RuntimeError as error:
|
|
logger.warning("ebook_answer_request_failed_falling_back error=%s", error)
|
|
answer = "Answer generation failed. Source chunks are still shown below."
|
|
else:
|
|
logger.info("ebook_answer_skipped_disabled")
|
|
answer = "Answer generation is disabled. Source chunks are shown below."
|
|
answer_step_name = "Answer generation" if request.app.state.config.answer_enabled else "Answer skipped"
|
|
response = replace(
|
|
response,
|
|
timings=(*response.timings, runtime_step_from_start(answer_step_name, answer_start)),
|
|
)
|
|
|
|
logger.info(
|
|
"ebook_search_request_complete results=%s rank_label=%s runtime_ms=%.1f",
|
|
len(response.results),
|
|
response.rank_label,
|
|
response.total_runtime_ms,
|
|
)
|
|
return templates.TemplateResponse(request, "partials/results.html", {"answer": answer, "response": response})
|