Spaces:
Paused
Paused
| """ | |
| Search Service - Hybrid Semantic and Full-Text Search for evidence and system data. | |
| """ | |
| import json | |
| import logging | |
| import os | |
| import sqlite3 | |
| from typing import Any | |
| from app.services.ai.ai_service import get_ai_service | |
| logger = logging.getLogger(__name__) | |
| class SearchService: | |
| """Unified search engine supporting semantic (AI) and keyword (FTS) queries""" | |
| def __init__(self, fts_db_path: str = "data/evidence_index.db"): | |
| self.fts_db_path = fts_db_path | |
| os.makedirs(os.path.dirname(fts_db_path), exist_ok=True) | |
| self._init_fts() | |
| def _init_fts(self): | |
| """Initialize SQLite FTS for keyword search fallback""" | |
| try: | |
| with sqlite3.connect(self.fts_db_path) as conn: | |
| conn.execute( | |
| """ | |
| CREATE TABLE IF NOT EXISTS evidence_index ( | |
| id INTEGER PRIMARY KEY, | |
| evidence_id TEXT UNIQUE, | |
| content TEXT, | |
| metadata TEXT | |
| ) | |
| """ | |
| ) | |
| # Simplified FTS setup | |
| conn.execute( | |
| "CREATE VIRTUAL TABLE IF NOT EXISTS evidence_fts USING fts5(content, content=evidence_index)" | |
| ) | |
| except Exception as e: | |
| logger.error(f"Failed to init FTS: {e}") | |
| async def semantic_search( | |
| self, query: str, limit: int = 10 | |
| ) -> list[dict[str, Any]]: | |
| """AI-powered semantic search via vector embeddings""" | |
| try: | |
| ai_service = await get_ai_service() | |
| return await ai_service.semantic_search(query, limit) | |
| except Exception as e: | |
| logger.error(f"Semantic search failed: {e}") | |
| return [] | |
| async def index_evidence( | |
| self, evidence_id: str, content: str, metadata: dict[str, Any] | |
| ): | |
| """Index evidence in both AI vector store and local FTS""" | |
| try: | |
| # 1. AI Indexing | |
| ai_service = await get_ai_service() | |
| await ai_service.add_document( | |
| doc_id=evidence_id, content=content, metadata=metadata | |
| ) | |
| # 2. Local FTS Indexing | |
| with sqlite3.connect(self.fts_db_path) as conn: | |
| conn.execute( | |
| "INSERT OR REPLACE INTO evidence_index (evidence_id, content, metadata) VALUES (?, ?, ?)", | |
| (evidence_id, content, json.dumps(metadata)), | |
| ) | |
| except Exception as e: | |
| logger.error(f"Failed to index evidence {evidence_id}: {e}") | |
| # singleton | |
| search_service = SearchService() | |
| # legacy alias | |
| evidence_search_index = search_service | |