tinyweb/handlers/search.py

from db import get_db, return_db, get_setting, get_site_name, clean_url
from templates import esc
from ._helpers import _sanitize_fts_query, _paginate, _get_page_tags, _respond, _page_nav, PER_PAGE


def handle_search(query):
    q = query.get("q", [""])[0].strip()
    page = _paginate(query)
    offset = (page - 1) * PER_PAGE
    db = get_db()
    try:
        count = db.execute("SELECT count(*) FROM pages").fetchone()[0]
        name = get_site_name()

        result_html = ""
        trusted_html = ""
        if q:
            try:
                fts_q = _sanitize_fts_query(q)
                bm25_rows = db.execute(
                    "SELECT p.id, p.url, p.title, p.body, p.note "
                    "FROM pages_fts f JOIN pages p ON f.rowid = p.id "
                    "WHERE pages_fts MATCH ? "
                    "ORDER BY bm25(pages_fts, 10.0, 1.0, 5.0, 3.0) LIMIT 100",
                    (fts_q,),
                ).fetchall()
            except Exception:
                bm25_rows = []

            bm25_ids = [r["id"] for r in bm25_rows]
            chunk_snippets = {}
            if get_setting("semantic_search", "0") == "1":
                try:
                    from embeddings import hybrid_search
                    use_reranker = get_setting("use_reranker", "1") == "1"
                    fused = hybrid_search(q, bm25_ids, limit=100, db=db, use_reranker=use_reranker)
                    fused_ids = [pid for pid, _ in fused]
                    chunk_snippets = {pid: text for pid, text in fused if text}
                except Exception:
                    fused_ids = bm25_ids
            else:
                fused_ids = bm25_ids

            total_results = len(fused_ids)
            page_ids = fused_ids[offset:offset + PER_PAGE]

            if page_ids:
                placeholders = ",".join("?" * len(page_ids))
                all_rows = db.execute(
                    f"SELECT id, url, title, body, note, summary FROM pages WHERE id IN ({placeholders})",
                    page_ids,
                ).fetchall()
                row_map = {r["id"]: r for r in all_rows}
                rows = [row_map[pid] for pid in page_ids if pid in row_map]
            else:
                rows = []

            if rows:
                for r in rows:
                    note_html = ""
                    if r["note"]:
                        note_html = f'<div class="note"><em>{esc(r["note"])}</em></div>'
                    tags = _get_page_tags(r["id"], db)
                    tags_html = ""
                    if tags:
                        tag_links = " ".join(f'<a href="/tags/{esc(t)}" class="tag">[{esc(t)}]</a>' for t in tags)
                        tags_html = f'<div class="tags">{tag_links}</div>'
                    snip_html = f'<br>{esc(r["summary"])}' if r["summary"] else ""
                    result_html += (
                        f'<div class="result">'
                        f'<a href="{esc(r["url"])}" rel="noreferrer noopener">{esc(r["title"])}</a><br>'
                        f'<small>{esc(r["url"])}</small>'
                        f'{snip_html}'
                        f'{note_html}{tags_html}'
                        f'</div>'
                    )
            else:
                result_html = "<p>No results in your index.</p>"

            words = q.lower().split()
            all_links = db.execute(
                "SELECT l.url, l.label, p.title AS source_title "
                "FROM links l JOIN pages p ON l.page_id = p.id",
            ).fetchall()
            indexed_urls = set(r["url"] for r in rows) if rows else set()
            seen = set()
            trusted = []
            for l in all_links:
                if l["url"] in indexed_urls or l["url"] in seen:
                    continue
                if any(w in l["label"].lower() for w in words):
                    seen.add(l["url"])
                    trusted.append(l)
                if len(trusted) >= 20:
                    break

            if trusted:
                items = ""
                for l in trusted:
                    items += (
                        f'<li><a href="{esc(clean_url(l["url"]))}" rel="noreferrer noopener">{esc(l["label"])}</a> '
                        f'<small>— from {esc(l["source_title"])}</small></li>'
                    )
                trusted_html = (
                    f'<details class="trusted">'
                    f'<summary>from your trusted sites ({len(trusted)})</summary>'
                    f'<ul>{items}</ul>'
                    f'</details>'
                )

            try:
                remote_rows = db.execute(
                    "SELECT rp.url, rp.title, rp.note, s.name AS source_name "
                    "FROM remote_pages_fts rpf "
                    "JOIN remote_pages rp ON rpf.rowid = rp.id "
                    "JOIN subscriptions s ON rp.subscription_id = s.id "
                    "WHERE remote_pages_fts MATCH ? ORDER BY rank LIMIT 50",
                    (_sanitize_fts_query(q),),
                ).fetchall()
            except Exception:
                remote_rows = []

        remote_html = ""
        if q and remote_rows:
            by_source = {}
            for r in remote_rows:
                source = r["source_name"] or "unknown"
                by_source.setdefault(source, []).append(r)
            for source, items in by_source.items():
                source_items = ""
                for r in items:
                    note_html = f' — <em>{esc(r["note"])}</em>' if r["note"] else ""
                    source_items += (
                        f'<li><a href="{esc(clean_url(r["url"]))}" rel="noreferrer noopener">{esc(r["title"])}</a>'
                        f'{note_html} <small>({esc(clean_url(r["url"]))})</small></li>'
                    )
                remote_html += (
                    f'<details class="remote" open>'
                    f'<summary>from {esc(source)} ({len(items)})</summary>'
                    f'<ul>{source_items}</ul>'
                    f'</details>'
                )
    finally:
        return_db(db)
    sub_count = ""
    if q and remote_rows:
        sub_count = f" + {len(remote_rows)} from subscriptions"
    welcome_html = ""
    if count == 0 and not q:
        welcome_html = (
            '<section style="margin-top:1.5rem;max-width:40em">'
            '<p>Your index is empty.</p>'
            '<p>tinyweb is a personal search engine for pages you save. '
            'The index stays on your machine; so does every search.</p>'
            '<p>From here: <a href="/add">add a page</a>, '
            '<a href="/style">get the bookmarklet</a>, or '
            '<a href="/subscriptions">subscribe to another instance</a>.</p>'
            '</section>'
        )
    return _respond(
        f'<form method="get" action="/">'
        f'<input name="q" value="{esc(q)}" placeholder="search your index" size="40">'
        f' <button type="submit">search</button>'
        f'</form>'
        f'<p class="meta">{count} pages indexed'
        f' · <a href="/add">+ add url</a></p>'
        f'{welcome_html}'
        f'{result_html}'
        f'{_page_nav(page, total_results, f"/?q={esc(q)}") if q else ""}'
        f'{trusted_html}{remote_html}'
    )