S.H.O.N.A.R._Desktop_Companion/backend/shonar/services/search.py

"""Full-text search across recordings (M9).

Postgres uses the tsvector columns from migration ``fts0000000001``
(title/notes, transcript text, summary content, tag names). SQLite (the
desktop bundled-lite engine) falls back to a substring scan; libraries
there are single-user and small, and the endpoint contract is identical.

This module is the whole SearchBackend seam — a Meilisearch/OpenSearch
implementation would replace it, not the callers.
"""

from __future__ import annotations

import uuid
from dataclasses import dataclass

from sqlalchemy import select, text
from sqlalchemy.ext.asyncio import AsyncSession

from shonar.db.models import Recording, RecordingTag, Summary, Tag, Transcript

# ``scope`` values accepted by the endpoint.
SCOPES = ("all", "title", "notes", "transcript", "summary", "tag")

# Hit ordering: a title hit outranks a transcript hit.
_FIELD_RANK = {"title": 0, "tag": 1, "notes": 2, "summary": 3, "transcript": 4}


@dataclass
class SearchHit:
    recording: Recording
    field: str  # where the best match landed
    snippet: str


def _snippet_around(value: str, at: int, width: int = 160) -> str:
    """~``width`` chars centred on ``at``, word-bounded, with ellipses."""
    half = width // 2
    start = max(0, at - half)
    end = min(len(value), at + half)
    if start > 0:
        start = value.rfind(" ", 0, start) + 1 or start
    if end < len(value):
        nxt = value.find(" ", end)
        end = nxt if nxt != -1 else end
    prefix = "…" if start > 0 else ""
    suffix = "…" if end < len(value) else ""
    return f"{prefix}{value[start:end].strip()}{suffix}"


async def tag_names(session: AsyncSession, recording_id: uuid.UUID) -> list[str]:
    return list(
        await session.scalars(
            select(Tag.name)
            .join(RecordingTag, RecordingTag.tag_id == Tag.id)
            .where(RecordingTag.recording_id == recording_id)
        )
    )


async def _current_transcript_text(session: AsyncSession, recording_id: uuid.UUID) -> str:
    row = await session.scalar(
        select(Transcript.text)
        .where(Transcript.recording_id == recording_id, Transcript.superseded_at.is_(None))
        .order_by(Transcript.version.desc())
    )
    return row or ""


def _summary_flat(content: dict | None) -> str:
    if not content:
        return ""
    parts = [str(content.get("short", "")), str(content.get("detailed", ""))]
    for key in ("key_points", "decisions", "action_items", "questions"):
        v = content.get(key)
        if isinstance(v, list):
            parts.extend(str(x) for x in v)
    return " ".join(p for p in parts if p)


# --- SQLite fallback ----------------------------------------------------------


async def _search_sqlite(
    session: AsyncSession, user_id: uuid.UUID, q: str, scope: str,
    limit: int, offset: int,
) -> tuple[list[SearchHit], int]:
    needles = [t for t in q.lower().split() if t]
    if not needles:
        return [], 0
    recs = list(
        await session.scalars(
            select(Recording)
            .where(Recording.user_id == user_id, Recording.deleted_at.is_(None))
            .order_by(Recording.recorded_at.desc())
        )
    )
    hits: list[SearchHit] = []
    for rec in recs:
        fields: list[tuple[str, str]] = []
        if scope in ("all", "title"):
            fields.append(("title", rec.title or ""))
        if scope in ("all", "notes"):
            fields.append(("notes", rec.notes or ""))
        if scope in ("all", "tag"):
            fields.append(("tag", " ".join(await tag_names(session, rec.id))))
        if scope in ("all", "transcript"):
            fields.append(("transcript", await _current_transcript_text(session, rec.id)))
        if scope in ("all", "summary"):
            content = await session.scalar(
                select(Summary.content)
                .where(Summary.recording_id == rec.id, Summary.superseded_at.is_(None))
                .order_by(Summary.version.desc())
            )
            fields.append(("summary", _summary_flat(content)))
        # A field matches when EVERY needle appears in it (AND semantics,
        # matching plainto_tsquery on the Postgres side).
        best: SearchHit | None = None
        for field, value in fields:
            low = value.lower()
            if not all(n in low for n in needles):
                continue
            at = low.find(needles[0])
            cand = SearchHit(
                recording=rec, field=field, snippet=_snippet_around(value, max(at, 0))
            )
            if best is None or _FIELD_RANK[field] < _FIELD_RANK[best.field]:
                best = cand
        if best is not None:
            hits.append(best)
    hits.sort(key=lambda h: (_FIELD_RANK[h.field], h.recording.recorded_at), reverse=False)
    return hits[offset : offset + limit], len(hits)


# --- PostgreSQL tsvector path -------------------------------------------------

# CTE ``q`` carries the parsed tsquery so it is computed once. Notes are not
# in the recordings search_vector weights the way we want headlines, so notes
# match by substring like the SQLite path (title/notes share the vector; the
# field classifier prefers 'title' when the vector hits).
_PG_MATCHES = """
WITH q AS (SELECT plainto_tsquery('simple', :q) AS ts),
lt AS (
  SELECT DISTINCT ON (recording_id) recording_id, search_vector, text
  FROM transcripts WHERE superseded_at IS NULL
  ORDER BY recording_id, version DESC
),
ls AS (
  SELECT DISTINCT ON (recording_id) recording_id, content, search_vector
  FROM summaries WHERE superseded_at IS NULL
  ORDER BY recording_id, version DESC
),
tm AS (
  SELECT DISTINCT rt.recording_id,
         ts_headline('simple', t.name, q.ts,
                     'StartSel=,StopSel=,MaxFragments=0') AS snip
  FROM recording_tags rt
  JOIN tags t ON t.id = rt.tag_id AND t.user_id = :uid, q
),
matched AS (
  SELECT r.id AS id,
    CASE
      WHEN :want_title AND r.search_vector @@ q.ts
           AND coalesce(r.title, '') <> ''
        THEN 'title'
      WHEN :want_tag AND tm.recording_id IS NOT NULL THEN 'tag'
      WHEN :want_notes AND coalesce(r.notes, '') ILIKE '%' || :raw || '%' THEN 'notes'
      WHEN :want_summary AND ls.search_vector @@ q.ts THEN 'summary'
      WHEN :want_transcript AND lt.search_vector @@ q.ts THEN 'transcript'
    END AS field,
    CASE
      WHEN :want_title AND r.search_vector @@ q.ts
           AND coalesce(r.title, '') <> ''
        THEN ts_headline('simple', r.title, q.ts,
                         'StartSel=,StopSel=,MaxFragments=0,MaxWords=25')
      WHEN :want_tag AND tm.recording_id IS NOT NULL THEN tm.snip
      WHEN :want_notes AND coalesce(r.notes, '') ILIKE '%' || :raw || '%'
        THEN left(r.notes, 200)
      WHEN :want_summary AND ls.search_vector @@ q.ts
        THEN ts_headline('simple',
              coalesce(ls.content->>'short', '') || ' ' || coalesce(ls.content->>'detailed', ''),
              q.ts, 'StartSel=,StopSel=,MaxFragments=1,MinWords=10,MaxWords=25')
      WHEN :want_transcript AND lt.search_vector @@ q.ts
        THEN ts_headline('simple', coalesce(lt.text, ''), q.ts,
                         'StartSel=,StopSel=,MaxFragments=1,MinWords=10,MaxWords=25')
    END AS snippet
  FROM recordings r
  CROSS JOIN q
  LEFT JOIN lt ON lt.recording_id = r.id
  LEFT JOIN ls ON ls.recording_id = r.id
  LEFT JOIN tm ON tm.recording_id = r.id
  WHERE r.user_id = :uid AND r.deleted_at IS NULL
)
SELECT id, field, snippet FROM matched
WHERE field IS NOT NULL
ORDER BY CASE field WHEN 'title' THEN 0 WHEN 'tag' THEN 1 WHEN 'notes' THEN 2
                    WHEN 'summary' THEN 3 ELSE 4 END,
         id
LIMIT :limit OFFSET :offset
"""

_PG_COUNT = """
WITH q AS (SELECT plainto_tsquery('simple', :q) AS ts),
lt AS (
  SELECT DISTINCT ON (recording_id) recording_id, search_vector
  FROM transcripts WHERE superseded_at IS NULL
),
ls AS (
  SELECT DISTINCT ON (recording_id) recording_id, search_vector
  FROM summaries WHERE superseded_at IS NULL
),
tm AS (
  SELECT DISTINCT rt.recording_id
  FROM recording_tags rt
  JOIN tags t ON t.id = rt.tag_id AND t.user_id = :uid, q
  WHERE t.search_vector @@ q.ts
)
SELECT count(*)
FROM recordings r
CROSS JOIN q
LEFT JOIN lt ON lt.recording_id = r.id
LEFT JOIN ls ON ls.recording_id = r.id
LEFT JOIN tm ON tm.recording_id = r.id
WHERE r.user_id = :uid AND r.deleted_at IS NULL
  AND (
    (:want_title AND r.search_vector @@ q.ts) OR
    (:want_transcript AND lt.search_vector @@ q.ts) OR
    (:want_summary AND ls.search_vector @@ q.ts) OR
    (:want_tag AND tm.recording_id IS NOT NULL) OR
    (:want_notes AND coalesce(r.notes, '') ILIKE '%' || :raw || '%')
  )
"""


async def _search_postgres(
    session: AsyncSession, user_id: uuid.UUID, q: str, scope: str,
    limit: int, offset: int,
) -> tuple[list[SearchHit], int]:
    # NOTE: the title field matches anything the recordings vector hits
    # (title + notes); notes-only hits surface under 'title' headlines from
    # the title text. Acceptable precision tradeoff for a GIN-indexed path.
    params = {
        "uid": user_id,
        "q": q,
        "raw": q,
        "limit": limit,
        "offset": offset,
        "want_title": scope in ("all", "title"),
        "want_transcript": scope in ("all", "transcript"),
        "want_summary": scope in ("all", "summary"),
        "want_tag": scope in ("all", "tag"),
        "want_notes": scope in ("all", "notes"),
    }
    rows = (await session.execute(text(_PG_MATCHES), params)).all()
    total = await session.scalar(text(_PG_COUNT), params) or 0
    if not rows:
        return [], total
    ids = [r[0] for r in rows]
    by_id = {
        rec.id: rec
        for rec in (
            await session.scalars(select(Recording).where(Recording.id.in_(ids)))
        ).all()
    }
    hits = [
        SearchHit(recording=by_id[r.id], field=r.field, snippet=r.snippet or "")
        for r in rows
        if r.id in by_id
    ]
    return hits, total


# --- public API ----------------------------------------------------------------


async def search_recordings(
    session: AsyncSession,
    user_id: uuid.UUID,
    q: str,
    *,
    scope: str = "all",
    limit: int = 20,
    offset: int = 0,
) -> tuple[list[SearchHit], int]:
    """Returns (page of hits ordered by field rank, total). Empty q → no hits."""
    q = q.strip()
    if not q:
        return [], 0
    dialect = session.bind.dialect.name if session.bind else "sqlite"
    if dialect == "postgresql":
        return await _search_postgres(session, user_id, q, scope, limit, offset)
    return await _search_sqlite(session, user_id, q, scope, limit, offset)