Coverage for src/local_deep_research/web/routers/library_search.py: 94%
178 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-06 15:42 +0000
1"""
2Semantic Search Routes
4Provides endpoints for:
5- Research history collection management and indexing
6- Semantic search across any library collection
7"""
9from fastapi import APIRouter, Depends, Request
10from fastapi.responses import JSONResponse
11from ..dependencies.auth import require_auth
12from ..dependencies.threadpool import run_db_sync
15from ...database.models.library import Collection, Document
17from ...research_library.utils import handle_api_error
18from ..dependencies.json_body import json_body_error
19from typing import Annotated
21router = APIRouter(prefix="/library", tags=["search"])
23# =============================================================================
24# Research History Collection & Indexing
25# =============================================================================
28@router.get("/api/research-history/collection")
29def get_research_history_collection(
30 request: Request, username: Annotated[str, Depends(require_auth)]
31):
32 """
33 Get the Research History collection info and indexing status.
35 Returns collection ID and statistics about indexed vs total research.
36 Counts are derived from DocumentCollection membership (matching the
37 collection page) rather than source_type_id filtering.
38 """
39 from ...constants import ResearchStatus
40 from ...database.models.library import DocumentCollection
41 from ...database.models.research import ResearchHistory
42 from ...database.session_context import get_user_db_session
43 from ...database.session_passwords import session_password_store
44 from ...research_library.search.services.research_history_indexer import (
45 ResearchHistoryIndexer,
46 )
48 session_id = request.session.get("session_id")
50 db_password = None
51 if session_id:
52 db_password = (
53 session_password_store.get_session_password( # gitleaks:allow
54 username, session_id
55 )
56 )
58 try:
59 indexer = ResearchHistoryIndexer(username, db_password)
60 collection_id = indexer.get_or_create_collection()
62 with get_user_db_session(username, db_password) as db_session:
63 # Total completed research with report content
64 total_research = (
65 db_session.query(ResearchHistory)
66 .filter(ResearchHistory.status == ResearchStatus.COMPLETED)
67 .filter(ResearchHistory.report_content.isnot(None))
68 .filter(ResearchHistory.report_content != "")
69 .count()
70 )
72 # Research entries represented in this collection
73 # (via Document → DocumentCollection join, matching collection page)
74 indexed_research = (
75 db_session.query(Document.research_id)
76 .join(
77 DocumentCollection,
78 DocumentCollection.document_id == Document.id,
79 )
80 .filter(DocumentCollection.collection_id == collection_id)
81 .filter(Document.research_id.isnot(None))
82 .distinct()
83 .count()
84 )
86 # Document counts in collection
87 total_documents = (
88 db_session.query(DocumentCollection)
89 .filter(DocumentCollection.collection_id == collection_id)
90 .count()
91 )
92 indexed_documents = (
93 db_session.query(DocumentCollection)
94 .filter(DocumentCollection.collection_id == collection_id)
95 .filter(DocumentCollection.indexed == True) # noqa: E712
96 .count()
97 )
99 return {
100 "success": True,
101 "collection_id": collection_id,
102 "total_research": total_research,
103 "indexed_research": indexed_research,
104 "total_documents": total_documents,
105 "indexed_documents": indexed_documents,
106 }
108 except Exception as e:
109 return handle_api_error("getting research history collection", e)
112@router.post("/api/research-history/convert-all")
113async def convert_all_research(
114 request: Request, username: Annotated[str, Depends(require_auth)]
115):
116 """
117 Convert all completed research entries into library Documents.
119 Unlike the SSE index endpoint this is a synchronous JSON endpoint that
120 creates Document rows (and DocumentCollection memberships) without
121 triggering FAISS / RAG indexing. Call this before the SSE index endpoint
122 to avoid nested-session problems on SQLite.
124 Request JSON (optional):
125 force: If true, re-convert even already-converted entries (default false)
127 Returns:
128 JSON with converted, skipped, failed counts and collection_id
129 """
130 from ...database.session_passwords import session_password_store
131 from ...research_library.search.services.research_history_indexer import (
132 ResearchHistoryIndexer,
133 )
135 data = await request.json()
136 if not isinstance(data, dict):
137 return json_body_error("success", "Request body must be valid JSON")
138 force = data.get("force", False)
139 if not isinstance(force, bool):
140 return json_body_error("success", "force must be a boolean")
142 session_id = request.session.get("session_id")
144 db_password = None
145 if session_id:
146 db_password = (
147 session_password_store.get_session_password( # gitleaks:allow
148 username, session_id
149 )
150 )
152 def _convert():
153 # convert_all_research walks every completed research with sync
154 # SQLCipher sessions — minutes of work for a large library. Run
155 # on the threadpool so it cannot stall the event loop serving
156 # every other request and WebSocket.
157 indexer = ResearchHistoryIndexer(username, db_password)
158 return indexer.convert_all_research(force=force)
160 try:
161 result = await run_db_sync(_convert)
162 return {"success": True, **result}
164 except Exception as e:
165 return handle_api_error("converting all research", e)
168@router.post("/api/research/{research_id}/add-to-collection")
169async def add_research_to_collection(
170 request: Request,
171 research_id,
172 username: Annotated[str, Depends(require_auth)],
173):
174 """
175 Add a research entry to a specific collection.
177 This allows users to organize research into custom collections
178 in addition to the default Research History collection.
180 Args:
181 research_id: UUID of the research to add
183 Request JSON:
184 collection_id: UUID of the target collection (required)
185 """
186 session_id = request.session.get("session_id")
187 data = await request.json()
188 if not isinstance(data, dict): 188 ↛ 189line 188 didn't jump to line 189 because the condition on line 188 was never true
189 return json_body_error("success", "Request body must be valid JSON")
190 return await run_db_sync(
191 _add_research_to_collection_sync,
192 data,
193 research_id,
194 username,
195 session_id,
196 )
199def _add_research_to_collection_sync(data, research_id, username, session_id):
200 from ...database.session_context import get_user_db_session
201 from ...database.session_passwords import session_password_store
202 from ...research_library.search.services.research_history_indexer import (
203 ResearchHistoryIndexer,
204 )
206 db_password = None
207 if session_id:
208 db_password = (
209 session_password_store.get_session_password( # gitleaks:allow
210 username, session_id
211 )
212 )
214 collection_id = data.get("collection_id")
216 if not collection_id:
217 return JSONResponse(
218 {
219 "success": False,
220 "error": "collection_id is required",
221 },
222 status_code=400,
223 )
225 try:
226 # Verify collection exists
227 with get_user_db_session(username, db_password) as db_session:
228 collection = (
229 db_session.query(Collection)
230 .filter(Collection.id == collection_id)
231 .first()
232 )
233 if not collection:
234 return JSONResponse(
235 {
236 "success": False,
237 "error": "Collection not found",
238 },
239 status_code=404,
240 )
242 collection_name = collection.name
244 indexer = ResearchHistoryIndexer(username, db_password)
245 result = indexer.index_research(
246 research_id,
247 collection_id=collection_id,
248 )
250 if result["status"] == "error": 250 ↛ 251line 250 didn't jump to line 251 because the condition on line 250 was never true
251 return JSONResponse(
252 {
253 "success": False,
254 "error": result.get("error", "Operation failed."),
255 },
256 status_code=400,
257 )
259 result["collection_name"] = collection_name
260 return {"success": True, **result}
262 except Exception as e:
263 return handle_api_error("adding research to collection", e)
266# =============================================================================
267# Collection Search (generic — works for any collection type)
268# =============================================================================
271@router.post("/api/collections/{collection_id}/search")
272async def search_collection(
273 request: Request,
274 collection_id,
275 username: Annotated[str, Depends(require_auth)],
276):
277 """Search any collection using semantic similarity.
279 Delegates to CollectionSearchEngine instead of reimplementing FAISS search.
281 Request JSON:
282 query: Search query string
283 limit: Maximum number of results (default 10)
284 """
285 session_id = request.session.get("session_id")
286 data = await request.json()
287 if not isinstance(data, dict):
288 return json_body_error("success", "Request body must be valid JSON")
289 return await run_db_sync(
290 _search_collection_sync, data, collection_id, username, session_id
291 )
294def _search_collection_sync(data, collection_id, username, session_id):
295 from ...database.session_context import get_user_db_session
296 from ...database.session_passwords import session_password_store
297 from ...web_search_engines.engines.search_engine_collection import (
298 CollectionSearchEngine,
299 )
301 db_password = None
302 if session_id:
303 db_password = (
304 session_password_store.get_session_password( # gitleaks:allow
305 username, session_id
306 )
307 )
308 raw_query = data.get("query", "")
309 if not isinstance(raw_query, str):
310 return json_body_error("success", "query must be a string")
311 query = raw_query.strip()
313 if len(query) > 10000: 313 ↛ 314line 313 didn't jump to line 314 because the condition on line 313 was never true
314 return JSONResponse(
315 {
316 "success": False,
317 "error": "Query too long (max 10000 characters)",
318 },
319 status_code=400,
320 )
322 try:
323 limit = max(1, min(int(data.get("limit", 10)), 50))
324 except (TypeError, ValueError):
325 limit = 10
327 if not query:
328 return JSONResponse(
329 {"success": False, "error": "Query is required"}, status_code=400
330 )
332 try:
333 # Verify collection exists and get its type
334 with get_user_db_session(username, db_password) as db_session:
335 collection = (
336 db_session.query(Collection).filter_by(id=collection_id).first()
337 )
338 if not collection:
339 return JSONResponse(
340 {"success": False, "error": "Collection not found"},
341 status_code=404,
342 )
343 collection_type = collection.collection_type
344 collection_name = collection.name
346 # Delegate to CollectionSearchEngine
347 engine = CollectionSearchEngine(
348 collection_id=collection_id,
349 collection_name=collection_name,
350 max_results=limit * 2,
351 settings_snapshot={"_username": username},
352 )
353 raw_results = engine.search(query, limit=limit * 2)
355 # Transform CollectionSearchEngine format -> API format
356 results = []
357 for r in raw_results:
358 meta = r.get("metadata", {})
359 results.append(
360 {
361 "document_id": meta.get("document_id")
362 or meta.get("source_id"),
363 "title": r.get("title", "Untitled"),
364 "snippet": r.get("snippet", ""),
365 "similarity": round(r.get("relevance_score", 0) * 100, 1),
366 "url": meta.get("source"),
367 }
368 )
369 if len(results) >= limit: 369 ↛ 370line 369 didn't jump to line 370 because the condition on line 369 was never true
370 break
372 # For research_history collections, enrich with report/source type
373 if collection_type == "research_history":
374 _enrich_with_research_metadata(results, username, db_password)
376 # Always enrich with document-level metadata (file type, domain)
377 _enrich_with_document_metadata(results, username, db_password)
379 return {"success": True, "results": results, "query": query}
381 except Exception as e:
382 return handle_api_error("searching collection", e)
385def _enrich_with_research_metadata(results, username, db_password):
386 """Add report/source type and research context to search results."""
387 from ...database.models.library import SourceType
388 from ...database.models.research import ResearchHistory
389 from ...database.session_context import get_user_db_session
391 doc_ids = [r["document_id"] for r in results if r.get("document_id")]
392 if not doc_ids: 392 ↛ 393line 392 didn't jump to line 393 because the condition on line 392 was never true
393 return
395 with get_user_db_session(username, db_password) as db_session:
396 rows = (
397 db_session.query(
398 Document.id.label("document_id"),
399 SourceType.name.label("source_type_name"),
400 ResearchHistory.title.label("research_title"),
401 ResearchHistory.query.label("research_query"),
402 ResearchHistory.created_at.label("research_created_at"),
403 Document.research_id,
404 )
405 .outerjoin(SourceType, Document.source_type_id == SourceType.id)
406 .outerjoin(
407 ResearchHistory,
408 Document.research_id == ResearchHistory.id,
409 )
410 .filter(Document.id.in_(doc_ids))
411 .all()
412 )
413 lookup = {row.document_id: row for row in rows}
415 for result in results:
416 row = lookup.get(result.get("document_id"))
417 if row:
418 result["type"] = (
419 "report"
420 if row.source_type_name == "research_report"
421 else "source"
422 )
423 result["research_id"] = row.research_id
424 result["research_title"] = row.research_title or (
425 row.research_query[:100] if row.research_query else ""
426 )
427 result["research_query"] = row.research_query
428 result["research_created_at"] = (
429 row.research_created_at
430 if isinstance(row.research_created_at, str)
431 else row.research_created_at.isoformat()
432 if row.research_created_at
433 else None
434 )
435 else:
436 result["type"] = "source"
437 result["research_id"] = None
438 result["research_title"] = ""
439 result["research_query"] = None
440 result["research_created_at"] = None
443def _enrich_with_document_metadata(results, username, db_password):
444 """Add file type, domain, and creation date to search results."""
445 from urllib.parse import urlparse
447 from ...database.session_context import get_user_db_session
449 doc_ids = [r["document_id"] for r in results if r.get("document_id")]
450 if not doc_ids:
451 return
453 with get_user_db_session(username, db_password) as db_session:
454 rows = (
455 db_session.query(
456 Document.id.label("document_id"),
457 Document.file_type,
458 Document.original_url,
459 Document.created_at,
460 )
461 .filter(Document.id.in_(doc_ids))
462 .all()
463 )
464 lookup = {row.document_id: row for row in rows}
466 for result in results:
467 row = lookup.get(result.get("document_id"))
468 if row:
469 result["file_type"] = row.file_type
470 result["created_at"] = (
471 row.created_at
472 if isinstance(row.created_at, str)
473 else row.created_at.isoformat()
474 if row.created_at
475 else None
476 )
477 if row.original_url:
478 try:
479 result["domain"] = urlparse(row.original_url).netloc
480 except (ValueError, AttributeError):
481 result["domain"] = "unknown"
482 else:
483 result["domain"] = None
484 else:
485 result.setdefault("file_type", "unknown")
486 result.setdefault("domain", None)
487 result.setdefault("created_at", None)