"""
FHIR Document Search Service.
Provides full-text and vector search over clinical notes.
"""
from typing import List, Dict, Any, Optional
from .base import BaseSearchService
class FHIRSearchService(BaseSearchService):
"""Service for searching FHIR clinical documents."""
def search_documents(self, query: str, limit: int = 10) -> List[Dict[str, Any]]:
"""
Search FHIR documents by text content.
Args:
query: Search keywords
limit: Maximum results
Returns:
List of matching documents with previews
"""
_, cursor = self.connect()
# Extract keywords for scoring
keywords = query.lower().split()
# SQL to find documents (parameterized)
sql = """
SELECT ResourceID, TextContent
FROM SQLUser.FHIRDocuments
WHERE (TextContent %MATCHES(?))
"""
# Note: Depending on IRIS version/setup, %MATCHES or LIKE might be used.
# Current MCP server uses manual filtering in Python for text search,
# but we'll try a more performant SQL approach if supported.
# Fallback to fetching and filtering if needed.
# For pragmatic refactor, we'll follow the pattern in fhir_graphrag_mcp_server.py
# which performs keyword matching on TextContent.
sql = "SELECT ID, TextContent FROM SQLUser.FHIRDocuments"
cursor.execute(sql)
results = []
for row in cursor.fetchall():
doc_id, text_content = row
if not text_content:
continue
text_lower = text_content.lower()
score = sum(text_lower.count(kw) for kw in keywords)
if score > 0:
results.append({
"fhir_id": str(doc_id),
"preview": text_content[:200] + "..." if len(text_content) > 200 else text_content,
"relevance": score,
"full_text": text_content
})
# Sort by relevance
results.sort(key=lambda x: x["relevance"], reverse=True)
return results[:limit]
def get_document_details(self, fhir_id: str) -> Optional[Dict[str, Any]]:
"""Retrieve full document details."""
_, cursor = self.connect()
sql = "SELECT TextContent FROM SQLUser.FHIRDocuments WHERE ID = ?"
cursor.execute(sql, [fhir_id])
row = cursor.fetchone()
if row:
return {
"fhir_id": fhir_id,
"text_content": row[0]
}
return None
