245 lines
7.9 KiB
Python
245 lines
7.9 KiB
Python
"""FTS5 session search bridge (M3).
|
|
|
|
Bridges the ``search {query, scope, chat_id?, thread_id?}`` frame to the hermes
|
|
session store (SQLite + FTS5, ``hermes_state.py`` / ``hermes_state_search.py``)
|
|
and returns ``search.results`` hits.
|
|
|
|
The session DB (``get_hermes_home()/"state.db"``) is opened **read-only** --
|
|
search never writes to the store. Two query paths:
|
|
|
|
* **FTS5** (primary): ``messages_fts MATCH <sanitized>`` with BM25 ranking.
|
|
* **LIKE** (fallback): when the FTS5 table is absent (FTS disabled / fresh DB)
|
|
or the MATCH raises, a substring scan over ``messages.content``.
|
|
|
|
Scope:
|
|
* ``"all"`` -- every channel/thread/session.
|
|
* ``"chat"`` -- restrict to the given ``chat_id`` (and optional ``thread_id``).
|
|
|
|
Each hit: ``{message_id, chat_id, thread_id, role, snippet, ts}`` where
|
|
``message_id`` is the session-store row id (string) and ``ts`` is epoch
|
|
milliseconds. The app navigates to the hit's channel/thread and matches the
|
|
message by timestamp to scroll + highlight.
|
|
|
|
Privacy: search is local to the user's own hermes home; no data leaves the
|
|
machine.
|
|
|
|
Milestone M3.
|
|
"""
|
|
|
|
import contextlib
|
|
import logging
|
|
import re
|
|
import sqlite3
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
MAX_QUERY_CHARS = 200
|
|
DEFAULT_LIMIT = 20
|
|
MAX_LIMIT = 100
|
|
|
|
# FTS5 special chars (mirror of hermes_state_search._FTS5_SPECIAL_CHARS) for the
|
|
# fallback sanitizer when the real one can't be imported.
|
|
_FTS5_SPECIAL_CHARS = "+{}():\"^@/#&|~[]<>,;!?$=\\'"
|
|
_FTS5_SPECIAL_RE = re.compile(f"[{re.escape(_FTS5_SPECIAL_CHARS)}]")
|
|
|
|
|
|
def _sanitize(query: str) -> str:
|
|
"""Sanitize user input for a safe FTS5 MATCH.
|
|
|
|
Prefers the gateway's own sanitizer (exact parity with session_search);
|
|
falls back to a simplified strip-and-quote pass when it can't be imported.
|
|
"""
|
|
q = (query or "").strip()
|
|
if not q:
|
|
return ""
|
|
try:
|
|
from hermes_state_search import SessionSearchMixin
|
|
|
|
return SessionSearchMixin._sanitize_fts5_query(q)
|
|
except Exception:
|
|
return _sanitize_fallback(q)
|
|
|
|
|
|
def _sanitize_fallback(query: str) -> str:
|
|
q = query[:MAX_QUERY_CHARS]
|
|
q = _FTS5_SPECIAL_RE.sub(" ", q)
|
|
if "%" in q:
|
|
q = q.replace("%", " ")
|
|
q = re.sub(r"\*+", "*", q)
|
|
q = re.sub(r"(^|\s)\*", r"\1", q)
|
|
q = re.sub(r"(?i)^(AND|OR|NOT)\b\s*", "", q.strip())
|
|
q = re.sub(r"(?i)\s+(AND|OR|NOT)\s*$", "", q.strip())
|
|
q = re.sub(r"\b(\w+(?:[._-]\w+)+)\b", r'"\1"', q)
|
|
return q.strip()
|
|
|
|
|
|
def _fts_available(conn: sqlite3.Connection) -> bool:
|
|
try:
|
|
row = conn.execute(
|
|
"SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'messages_fts' LIMIT 1"
|
|
).fetchone()
|
|
return row is not None
|
|
except sqlite3.Error:
|
|
return False
|
|
|
|
|
|
def _scope_clauses(
|
|
scope: str, chat_id: str | None, thread_id: str | None
|
|
) -> tuple[list[str], list[Any]]:
|
|
"""Build the scope WHERE clauses + params (empty for scope='all')."""
|
|
clauses: list[str] = []
|
|
params: list[Any] = []
|
|
if scope == "chat" and chat_id:
|
|
clauses.append("s.chat_id = ?")
|
|
params.append(chat_id)
|
|
if thread_id:
|
|
clauses.append("s.thread_id = ?")
|
|
params.append(thread_id)
|
|
return clauses, params
|
|
|
|
|
|
def _row_to_hit(row: sqlite3.Row) -> dict[str, Any]:
|
|
ts = row["timestamp"]
|
|
try:
|
|
ts_ms = int(float(ts) * 1000)
|
|
except (TypeError, ValueError):
|
|
ts_ms = 0
|
|
return {
|
|
"message_id": str(row["id"]),
|
|
"chat_id": row["chat_id"],
|
|
"thread_id": row["thread_id"],
|
|
"role": row["role"],
|
|
"snippet": row["snippet"] or "",
|
|
"ts": ts_ms,
|
|
}
|
|
|
|
|
|
def _fts_query(
|
|
conn: sqlite3.Connection,
|
|
query: str,
|
|
scope: str,
|
|
chat_id: str | None,
|
|
thread_id: str | None,
|
|
limit: int,
|
|
) -> list[dict[str, Any]]:
|
|
where = ["messages_fts MATCH ?", "(m.active = 1 OR m.compacted = 1)"]
|
|
params: list[Any] = [query]
|
|
scope_clauses, scope_params = _scope_clauses(scope, chat_id, thread_id)
|
|
where.extend(scope_clauses)
|
|
params.extend(scope_params)
|
|
params.extend([limit])
|
|
# The f-string only splices a fixed set of static WHERE fragments; every
|
|
# user value is bound via ``?`` placeholders (see execute below).
|
|
sql = f"""
|
|
SELECT
|
|
m.id,
|
|
m.role,
|
|
snippet(messages_fts, -1, '>>>', '<<<', '...', 40) AS snippet,
|
|
m.timestamp,
|
|
s.chat_id,
|
|
s.thread_id
|
|
FROM messages_fts
|
|
JOIN messages m ON m.id = messages_fts.rowid
|
|
JOIN sessions s ON s.id = m.session_id
|
|
WHERE {" AND ".join(where)}
|
|
ORDER BY rank
|
|
LIMIT ?
|
|
"""
|
|
# Safe: every value is bound via ``?`` placeholders (no user data in SQL).
|
|
# pi-lens-ignore: python-sql-injection
|
|
rows = conn.execute(sql, params).fetchall()
|
|
return [_row_to_hit(r) for r in rows]
|
|
|
|
|
|
def _like_query(
|
|
conn: sqlite3.Connection,
|
|
query: str,
|
|
scope: str,
|
|
chat_id: str | None,
|
|
thread_id: str | None,
|
|
limit: int,
|
|
) -> list[dict[str, Any]]:
|
|
"""Substring fallback when FTS5 is unavailable."""
|
|
# First plain word of the query is the LIKE needle (best-effort).
|
|
needle = re.split(r"\s+", query.strip(), maxsplit=1)[0].strip('"')
|
|
if not needle:
|
|
return []
|
|
like = f"%{needle}%"
|
|
where = ["(m.active = 1 OR m.compacted = 1)", "m.content LIKE ?"]
|
|
params: list[Any] = [like]
|
|
scope_clauses, scope_params = _scope_clauses(scope, chat_id, thread_id)
|
|
where.extend(scope_clauses)
|
|
params.extend(scope_params)
|
|
params.extend([limit])
|
|
# The f-string only splices a fixed set of static WHERE fragments; every
|
|
# user value is bound via ``?`` placeholders (see execute below).
|
|
sql = f"""
|
|
SELECT
|
|
m.id,
|
|
m.role,
|
|
substr(m.content, max(1, instr(m.content, ?) - 40), 120) AS snippet,
|
|
m.timestamp,
|
|
s.chat_id,
|
|
s.thread_id
|
|
FROM messages m
|
|
JOIN sessions s ON s.id = m.session_id
|
|
WHERE {" AND ".join(where)}
|
|
ORDER BY m.timestamp DESC
|
|
LIMIT ?
|
|
"""
|
|
# The needle appears twice (LIKE + instr); params order: like, scope..., needle, limit
|
|
full_params = [like, *scope_params, needle, limit]
|
|
# Safe: every value is bound via ``?`` placeholders (no user data in SQL).
|
|
# pi-lens-ignore: python-sql-injection
|
|
rows = conn.execute(sql, full_params).fetchall()
|
|
return [_row_to_hit(r) for r in rows]
|
|
|
|
|
|
def search(
|
|
db_path: Path,
|
|
query: str,
|
|
scope: str = "all",
|
|
chat_id: str | None = None,
|
|
thread_id: str | None = None,
|
|
limit: int = DEFAULT_LIMIT,
|
|
) -> list[dict[str, Any]]:
|
|
"""Run a scoped search over the session store. Returns a list of hits.
|
|
|
|
Never raises: any DB/FTS error yields an empty result (the caller sends an
|
|
empty ``search.results``).
|
|
"""
|
|
sanitized = _sanitize(query)
|
|
if not sanitized:
|
|
return []
|
|
db_path = Path(db_path)
|
|
if not db_path.exists():
|
|
return []
|
|
limit = max(1, min(int(limit or DEFAULT_LIMIT), MAX_LIMIT))
|
|
scope = (scope or "all").strip().lower()
|
|
if scope not in ("all", "chat"):
|
|
scope = "all"
|
|
|
|
try:
|
|
conn = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True)
|
|
except sqlite3.Error as e:
|
|
logger.warning("iris search: open failed: %s", e)
|
|
return []
|
|
conn.row_factory = sqlite3.Row
|
|
try:
|
|
if _fts_available(conn):
|
|
try:
|
|
return _fts_query(conn, sanitized, scope, chat_id, thread_id, limit)
|
|
except sqlite3.Error as e:
|
|
logger.debug("iris search: FTS5 failed, using LIKE: %s", e)
|
|
return _like_query(conn, sanitized, scope, chat_id, thread_id, limit)
|
|
except sqlite3.Error as e:
|
|
logger.warning("iris search: query failed: %s", e)
|
|
return []
|
|
finally:
|
|
# Best-effort: a close failure on a read-only connection is not
|
|
# actionable (nothing to roll back).
|
|
with contextlib.suppress(Exception):
|
|
conn.close()
|