diff --git a/document-parser/api/ingestion.py b/document-parser/api/ingestion.py index 2f48e7d..8915ce6 100644 --- a/document-parser/api/ingestion.py +++ b/document-parser/api/ingestion.py @@ -5,9 +5,14 @@ from __future__ import annotations import logging from typing import Annotated -from fastapi import APIRouter, Depends, HTTPException, Request +from fastapi import APIRouter, Depends, HTTPException, Query, Request -from api.schemas import IngestionResponse, IngestionStatusResponse +from api.schemas import ( + IngestionResponse, + IngestionStatusResponse, + SearchResponse, + SearchResultItem, +) from services.analysis_service import AnalysisService from services.ingestion_service import IngestionService @@ -77,6 +82,38 @@ async def delete_ingested_document(doc_id: str, ingestion: IngestionDep) -> None @router.get("/status", response_model=IngestionStatusResponse) async def ingestion_status(request: Request) -> IngestionStatusResponse: - """Check if the ingestion pipeline is available.""" - available = request.app.state.ingestion_service is not None - return IngestionStatusResponse(available=available) + """Check if the ingestion pipeline is available and OpenSearch is connected.""" + svc = request.app.state.ingestion_service + if svc is None: + return IngestionStatusResponse(available=False, opensearch_connected=False) + + connected = await svc.ping() + return IngestionStatusResponse(available=True, opensearch_connected=connected) + + +@router.get("/search", response_model=SearchResponse) +async def search_chunks( + ingestion: IngestionDep, + q: str = Query(..., min_length=1, description="Search query"), + doc_id: str | None = Query(None, description="Filter by document ID"), + k: int = Query(20, ge=1, le=100, description="Max results"), +) -> SearchResponse: + """Full-text search across indexed chunks. + + Returns matching chunks with content and metadata. + Optionally filter by document ID. + """ + results = await ingestion.search_fulltext(q, k=k, doc_id=doc_id) + items = [ + SearchResultItem( + doc_id=r.chunk.doc_id, + filename=r.chunk.filename, + content=r.chunk.content, + chunk_index=r.chunk.chunk_index, + page_number=r.chunk.page_number, + score=r.score, + headings=r.chunk.headings, + ) + for r in results + ] + return SearchResponse(results=items, total=len(items), query=q) diff --git a/document-parser/api/schemas.py b/document-parser/api/schemas.py index 808b70b..1f283e3 100644 --- a/document-parser/api/schemas.py +++ b/document-parser/api/schemas.py @@ -190,3 +190,23 @@ class IngestionResponse(_CamelModel): class IngestionStatusResponse(_CamelModel): available: bool + opensearch_connected: bool = False + + +class SearchResultItem(_CamelModel): + """A single search result with content and metadata.""" + + doc_id: str + filename: str + content: str + chunk_index: int + page_number: int + score: float + headings: list[str] = [] + highlights: list[str] = [] + + +class SearchResponse(_CamelModel): + results: list[SearchResultItem] + total: int + query: str diff --git a/document-parser/services/ingestion_service.py b/document-parser/services/ingestion_service.py index ca29c1e..cce6b98 100644 --- a/document-parser/services/ingestion_service.py +++ b/document-parser/services/ingestion_service.py @@ -164,3 +164,27 @@ class IngestionService: k=k, doc_id=doc_id, ) + + async def search_fulltext( + self, + query: str, + *, + k: int = 20, + doc_id: str | None = None, + ) -> list: + """Full-text keyword search in indexed chunks.""" + return await self._vector_store.search_fulltext( + self._config.index_name, + query, + k=k, + doc_id=doc_id, + ) + + async def ping(self) -> bool: + """Check if the OpenSearch cluster is reachable.""" + try: + info = await self._vector_store._client.info() + return bool(info) + except Exception: + logger.debug("OpenSearch ping failed", exc_info=True) + return False diff --git a/document-parser/tests/test_ingestion_api.py b/document-parser/tests/test_ingestion_api.py index ee124a8..0c330c4 100644 --- a/document-parser/tests/test_ingestion_api.py +++ b/document-parser/tests/test_ingestion_api.py @@ -112,3 +112,65 @@ class TestIngestionDisabled: tc = TestClient(app) resp = tc.post("/api/ingestion/job-1") assert resp.status_code == 503 + + +class TestStatusOpenSearch: + def test_status_with_opensearch_connected( + self, client: TestClient, mock_ingestion_service: AsyncMock + ) -> None: + mock_ingestion_service.ping.return_value = True + resp = client.get("/api/ingestion/status") + assert resp.status_code == 200 + data = resp.json() + assert data["available"] is True + assert data["opensearchConnected"] is True + + def test_status_with_opensearch_disconnected( + self, client: TestClient, mock_ingestion_service: AsyncMock + ) -> None: + mock_ingestion_service.ping.return_value = False + resp = client.get("/api/ingestion/status") + assert resp.status_code == 200 + data = resp.json() + assert data["available"] is True + assert data["opensearchConnected"] is False + + +class TestSearchEndpoint: + def test_search_success(self, client: TestClient, mock_ingestion_service: AsyncMock) -> None: + from domain.vector_schema import IndexedChunk, SearchResult + + chunk = IndexedChunk( + doc_id="doc-1", + filename="test.pdf", + content="hello world", + embedding=[], + chunk_index=0, + chunk_type="text", + page_number=1, + headings=["Intro"], + ) + mock_ingestion_service.search_fulltext.return_value = [ + SearchResult(chunk=chunk, score=0.95) + ] + resp = client.get("/api/ingestion/search", params={"q": "hello"}) + assert resp.status_code == 200 + data = resp.json() + assert data["total"] == 1 + assert data["query"] == "hello" + assert data["results"][0]["content"] == "hello world" + assert data["results"][0]["score"] == 0.95 + + def test_search_empty_query(self, client: TestClient) -> None: + resp = client.get("/api/ingestion/search", params={"q": ""}) + assert resp.status_code == 422 + + def test_search_with_doc_filter( + self, client: TestClient, mock_ingestion_service: AsyncMock + ) -> None: + mock_ingestion_service.search_fulltext.return_value = [] + resp = client.get("/api/ingestion/search", params={"q": "test", "doc_id": "doc-1"}) + assert resp.status_code == 200 + mock_ingestion_service.search_fulltext.assert_awaited_once_with( + "test", k=20, doc_id="doc-1" + ) diff --git a/document-parser/tests/test_ingestion_service.py b/document-parser/tests/test_ingestion_service.py index cbea903..dff801d 100644 --- a/document-parser/tests/test_ingestion_service.py +++ b/document-parser/tests/test_ingestion_service.py @@ -140,6 +140,44 @@ class TestSearch: mock_vector_store.search_similar.assert_awaited_once() +class TestSearchFulltext: + async def test_delegates_to_vector_store( + self, service: IngestionService, mock_vector_store: AsyncMock + ) -> None: + mock_vector_store.search_fulltext.return_value = [] + await service.search_fulltext("hello world", k=5) + mock_vector_store.search_fulltext.assert_awaited_once_with( + "test-idx", "hello world", k=5, doc_id=None + ) + + async def test_filters_by_doc_id( + self, service: IngestionService, mock_vector_store: AsyncMock + ) -> None: + mock_vector_store.search_fulltext.return_value = [] + await service.search_fulltext("hello", doc_id="doc-1") + mock_vector_store.search_fulltext.assert_awaited_once_with( + "test-idx", "hello", k=20, doc_id="doc-1" + ) + + +class TestPing: + async def test_ping_success( + self, service: IngestionService, mock_vector_store: AsyncMock + ) -> None: + mock_vector_store._client = AsyncMock() + mock_vector_store._client.info.return_value = {"cluster_name": "test"} + result = await service.ping() + assert result is True + + async def test_ping_failure( + self, service: IngestionService, mock_vector_store: AsyncMock + ) -> None: + mock_vector_store._client = AsyncMock() + mock_vector_store._client.info.side_effect = ConnectionError("down") + result = await service.ping() + assert result is False + + class TestEnsureIndex: async def test_calls_vector_store( self, service: IngestionService, mock_vector_store: AsyncMock diff --git a/frontend/src/features/ingestion/api.ts b/frontend/src/features/ingestion/api.ts index 4cdb345..ac749c4 100644 --- a/frontend/src/features/ingestion/api.ts +++ b/frontend/src/features/ingestion/api.ts @@ -8,6 +8,23 @@ export interface IngestionResult { export interface IngestionStatus { available: boolean + opensearchConnected: boolean +} + +export interface SearchResultItem { + docId: string + filename: string + content: string + chunkIndex: number + pageNumber: number + score: number + headings: string[] +} + +export interface SearchResponse { + results: SearchResultItem[] + total: number + query: string } export function ingestAnalysis(jobId: string): Promise { @@ -23,3 +40,13 @@ export function deleteIngested(docId: string): Promise { export function fetchIngestionStatus(): Promise { return apiFetch('/api/ingestion/status') } + +export function searchChunks( + query: string, + options: { docId?: string; k?: number } = {}, +): Promise { + const params = new URLSearchParams({ q: query }) + if (options.docId) params.set('doc_id', options.docId) + if (options.k) params.set('k', String(options.k)) + return apiFetch(`/api/ingestion/search?${params}`) +} diff --git a/frontend/src/features/ingestion/store.ts b/frontend/src/features/ingestion/store.ts index 4fbcd14..9b2620b 100644 --- a/frontend/src/features/ingestion/store.ts +++ b/frontend/src/features/ingestion/store.ts @@ -2,35 +2,68 @@ import { defineStore } from 'pinia' import { ref } from 'vue' import * as api from './api' +export type IngestionStep = 'embedding' | 'indexing' | 'done' + export const useIngestionStore = defineStore('ingestion', () => { const available = ref(false) + const opensearchConnected = ref(false) const ingesting = ref(false) const error = ref(null) /** Map of docId → chunks indexed count (tracks which docs are ingested) */ const ingestedDocs = ref>({}) + /** Current step of the ingestion pipeline (null when idle) */ + const currentStep = ref(null) + /** Search results */ + const searchResults = ref([]) + const searchQuery = ref('') + const searching = ref(false) + + let _pollTimer: ReturnType | null = null async function checkAvailability(): Promise { try { const status = await api.fetchIngestionStatus() available.value = status.available + opensearchConnected.value = status.opensearchConnected } catch { available.value = false + opensearchConnected.value = false + } + } + + function startPolling(intervalMs = 30_000): void { + stopPolling() + _pollTimer = setInterval(checkAvailability, intervalMs) + } + + function stopPolling(): void { + if (_pollTimer) { + clearInterval(_pollTimer) + _pollTimer = null } } async function ingest(jobId: string): Promise { ingesting.value = true error.value = null + currentStep.value = 'embedding' try { + currentStep.value = 'indexing' const result = await api.ingestAnalysis(jobId) + currentStep.value = 'done' ingestedDocs.value[result.docId] = result.chunksIndexed return result } catch (e) { error.value = (e as Error).message || 'Ingestion failed' console.error('Ingestion failed', e) + currentStep.value = null return null } finally { ingesting.value = false + // Reset step after a short delay so the user sees the "done" state + setTimeout(() => { + currentStep.value = null + }, 2000) } } @@ -44,13 +77,46 @@ export const useIngestionStore = defineStore('ingestion', () => { } } + async function search(query: string, docId?: string): Promise { + if (!query.trim()) { + searchResults.value = [] + searchQuery.value = '' + return + } + searching.value = true + searchQuery.value = query + try { + const resp = await api.searchChunks(query, { docId }) + searchResults.value = resp.results + } catch (e) { + console.error('Search failed', e) + searchResults.value = [] + } finally { + searching.value = false + } + } + + function clearSearch(): void { + searchResults.value = [] + searchQuery.value = '' + } + return { available, + opensearchConnected, ingesting, error, ingestedDocs, + currentStep, + searchResults, + searchQuery, + searching, checkAvailability, + startPolling, + stopPolling, ingest, deleteIngested, + search, + clearSearch, } }) diff --git a/frontend/src/pages/DocumentsPage.vue b/frontend/src/pages/DocumentsPage.vue index 3256bff..a4759e5 100644 --- a/frontend/src/pages/DocumentsPage.vue +++ b/frontend/src/pages/DocumentsPage.vue @@ -30,6 +30,46 @@ + +