"""Conversation search provider — queries comm_conversations table.""" from __future__ import annotations import logging import uuid from typing import Any from sqlalchemy import text from sqlalchemy.ext.asyncio import AsyncSession from app.plugins.builtins.unified_search.base_provider import BaseSearchProvider logger = logging.getLogger(__name__) class ConversationSearchProvider(BaseSearchProvider): """Search provider for CommConversation entities.""" entity_type = "conversation" async def _search_fts_filtered( self, db: AsyncSession, tsquery: str, tenant_id: uuid.UUID, limit: int, visible_ids: set[uuid.UUID] | None, ) -> list[dict[str, Any]]: """Full-text search on comm_conversations.title and last_msg_preview.""" if visible_ids is not None: sql = text( """ SELECT c.*, ts_rank( to_tsvector('pg_catalog.german', coalesce(c.title, '') || ' ' || coalesce(c.last_msg_preview, '') ), to_tsquery('pg_catalog.german', :q) ) AS rank FROM comm_conversations c WHERE c.tenant_id = :tid AND c.deleted_at IS NULL AND to_tsvector('pg_catalog.german', coalesce(c.title, '') || ' ' || coalesce(c.last_msg_preview, '') ) @@ to_tsquery('pg_catalog.german', :q) AND c.id = ANY(:visible_ids) ORDER BY rank DESC LIMIT :lim """ ) result = await db.execute( sql, { "q": tsquery, "tid": tenant_id, "lim": limit, "visible_ids": list(visible_ids), }, ) else: sql = text( """ SELECT c.*, ts_rank( to_tsvector('pg_catalog.german', coalesce(c.title, '') || ' ' || coalesce(c.last_msg_preview, '') ), to_tsquery('pg_catalog.german', :q) ) AS rank FROM comm_conversations c WHERE c.tenant_id = :tid AND c.deleted_at IS NULL AND to_tsvector('pg_catalog.german', coalesce(c.title, '') || ' ' || coalesce(c.last_msg_preview, '') ) @@ to_tsquery('pg_catalog.german', :q) ORDER BY rank DESC LIMIT :lim """ ) result = await db.execute( sql, {"q": tsquery, "tid": tenant_id, "lim": limit}, ) rows = result.mappings().all() return [dict(r) for r in rows] async def _search_vector_filtered( self, db: AsyncSession, embedding: list[float], tenant_id: uuid.UUID, limit: int, visible_ids: set[uuid.UUID] | None, ) -> list[dict[str, Any]]: """Semantic search — comm_conversations table has no embedding column yet.""" return [] async def get_embedding_text( self, db: AsyncSession, entity_id: uuid.UUID, tenant_id: uuid.UUID ) -> str: """Get text for embedding generation.""" sql = text( """ SELECT title, last_msg_preview FROM comm_conversations WHERE id = :eid AND tenant_id = :tid """ ) result = await db.execute(sql, {"eid": entity_id, "tid": tenant_id}) row = result.mappings().first() if not row: return "" parts = [ row.get("title", ""), row.get("last_msg_preview", ""), ] return " ".join(str(p) for p in parts if p) def to_search_result(self, entity: object) -> dict[str, Any]: """Convert conversation to search result dict.""" if isinstance(entity, dict): title = entity.get("title", "") or "" preview = entity.get("last_msg_preview", "") or "" entity_id = str(entity.get("id", "")) is_direct = entity.get("is_direct", False) else: title = getattr(entity, "title", "") or "" preview = getattr(entity, "last_msg_preview", "") or "" entity_id = str(getattr(entity, "id", "")) is_direct = getattr(entity, "is_direct", False) return { "entity_type": self.entity_type, "entity_id": entity_id, "title": title, "snippet": preview[:200], "score": 0.0, "data": {"is_direct": is_direct}, }