Coverage for src/local_deep_research/web/routers/library_search.py: 96%
175 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +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, read_json_dict
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, body_error = await read_json_dict(
136 request, "success", "Request body must be valid JSON"
137 )
138 if body_error is not None:
139 return body_error
140 force = data.get("force", False)
141 if not isinstance(force, bool):
142 return json_body_error("success", "force must be a boolean")
144 session_id = request.session.get("session_id")
146 db_password = None
147 if session_id:
148 db_password = (
149 session_password_store.get_session_password( # gitleaks:allow
150 username, session_id
151 )
152 )
154 def _convert():
155 # convert_all_research walks every completed research with sync
156 # SQLCipher sessions — minutes of work for a large library. Run
157 # on the threadpool so it cannot stall the event loop serving
158 # every other request and WebSocket.
159 indexer = ResearchHistoryIndexer(username, db_password)
160 return indexer.convert_all_research(force=force)
162 try:
163 result = await run_db_sync(_convert)
164 return {"success": True, **result}
166 except Exception as e:
167 return handle_api_error("converting all research", e)
170@router.post("/api/research/{research_id}/add-to-collection")
171async def add_research_to_collection(
172 request: Request,
173 research_id,
174 username: Annotated[str, Depends(require_auth)],
175):
176 """
177 Add a research entry to a specific collection.
179 This allows users to organize research into custom collections
180 in addition to the default Research History collection.
182 Args:
183 research_id: UUID of the research to add
185 Request JSON:
186 collection_id: UUID of the target collection (required)
187 """
188 session_id = request.session.get("session_id")
189 data, body_error = await read_json_dict(
190 request, "success", "Request body must be valid JSON"
191 )
192 if body_error is not None:
193 return body_error
194 return await run_db_sync(
195 _add_research_to_collection_sync,
196 data,
197 research_id,
198 username,
199 session_id,
200 )
203def _add_research_to_collection_sync(data, research_id, username, session_id):
204 from ...database.session_context import get_user_db_session
205 from ...database.session_passwords import session_password_store
206 from ...research_library.search.services.research_history_indexer import (
207 ResearchHistoryIndexer,
208 )
210 db_password = None
211 if session_id:
212 db_password = (
213 session_password_store.get_session_password( # gitleaks:allow
214 username, session_id
215 )
216 )
218 collection_id = data.get("collection_id")
220 if not collection_id:
221 return JSONResponse(
222 {
223 "success": False,
224 "error": "collection_id is required",
225 },
226 status_code=400,
227 )
229 try:
230 # Verify collection exists
231 with get_user_db_session(username, db_password) as db_session:
232 collection = (
233 db_session.query(Collection)
234 .filter(Collection.id == collection_id)
235 .first()
236 )
237 if not collection:
238 return JSONResponse(
239 {
240 "success": False,
241 "error": "Collection not found",
242 },
243 status_code=404,
244 )
246 collection_name = collection.name
248 indexer = ResearchHistoryIndexer(username, db_password)
249 result = indexer.index_research(
250 research_id,
251 collection_id=collection_id,
252 )
254 if result["status"] == "error": 254 ↛ 255line 254 didn't jump to line 255 because the condition on line 254 was never true
255 return JSONResponse(
256 {
257 "success": False,
258 "error": result.get("error", "Operation failed."),
259 },
260 status_code=400,
261 )
263 result["collection_name"] = collection_name
264 return {"success": True, **result}
266 except Exception as e:
267 return handle_api_error("adding research to collection", e)
270# =============================================================================
271# Collection Search (generic — works for any collection type)
272# =============================================================================
275@router.post("/api/collections/{collection_id}/search")
276async def search_collection(
277 request: Request,
278 collection_id,
279 username: Annotated[str, Depends(require_auth)],
280):
281 """Search any collection using semantic similarity.
283 Delegates to CollectionSearchEngine instead of reimplementing FAISS search.
285 Request JSON:
286 query: Search query string
287 limit: Maximum number of results (default 10)
288 """
289 session_id = request.session.get("session_id")
290 data, body_error = await read_json_dict(
291 request, "success", "Request body must be valid JSON"
292 )
293 if body_error is not None:
294 return body_error
295 return await run_db_sync(
296 _search_collection_sync, data, collection_id, username, session_id
297 )
300def _search_collection_sync(data, collection_id, username, session_id):
301 from ...database.session_context import get_user_db_session
302 from ...database.session_passwords import session_password_store
303 from ...web_search_engines.engines.search_engine_collection import (
304 CollectionSearchEngine,
305 )
307 db_password = None
308 if session_id:
309 db_password = (
310 session_password_store.get_session_password( # gitleaks:allow
311 username, session_id
312 )
313 )
314 raw_query = data.get("query", "")
315 if not isinstance(raw_query, str):
316 return json_body_error("success", "query must be a string")
317 query = raw_query.strip()
319 if len(query) > 10000: 319 ↛ 320line 319 didn't jump to line 320 because the condition on line 319 was never true
320 return JSONResponse(
321 {
322 "success": False,
323 "error": "Query too long (max 10000 characters)",
324 },
325 status_code=400,
326 )
328 try:
329 limit = max(1, min(int(data.get("limit", 10)), 50))
330 except (TypeError, ValueError):
331 limit = 10
333 if not query:
334 return JSONResponse(
335 {"success": False, "error": "Query is required"}, status_code=400
336 )
338 try:
339 # Verify collection exists and get its type
340 with get_user_db_session(username, db_password) as db_session:
341 collection = (
342 db_session.query(Collection).filter_by(id=collection_id).first()
343 )
344 if not collection:
345 return JSONResponse(
346 {"success": False, "error": "Collection not found"},
347 status_code=404,
348 )
349 collection_type = collection.collection_type
350 collection_name = collection.name
352 # Delegate to CollectionSearchEngine
353 engine = CollectionSearchEngine(
354 collection_id=collection_id,
355 collection_name=collection_name,
356 max_results=limit * 2,
357 settings_snapshot={"_username": username},
358 )
359 raw_results = engine.search(query, limit=limit * 2)
361 # Transform CollectionSearchEngine format -> API format
362 results = []
363 for r in raw_results:
364 meta = r.get("metadata", {})
365 results.append(
366 {
367 "document_id": meta.get("document_id")
368 or meta.get("source_id"),
369 "title": r.get("title", "Untitled"),
370 "snippet": r.get("snippet", ""),
371 "similarity": round(r.get("relevance_score", 0) * 100, 1),
372 "url": meta.get("source"),
373 }
374 )
375 if len(results) >= limit: 375 ↛ 376line 375 didn't jump to line 376 because the condition on line 375 was never true
376 break
378 # For research_history collections, enrich with report/source type
379 if collection_type == "research_history":
380 _enrich_with_research_metadata(results, username, db_password)
382 # Always enrich with document-level metadata (file type, domain)
383 _enrich_with_document_metadata(results, username, db_password)
385 return {"success": True, "results": results, "query": query}
387 except Exception as e:
388 return handle_api_error("searching collection", e)
391def _enrich_with_research_metadata(results, username, db_password):
392 """Add report/source type and research context to search results."""
393 from ...database.models.library import SourceType
394 from ...database.models.research import ResearchHistory
395 from ...database.session_context import get_user_db_session
397 doc_ids = [r["document_id"] for r in results if r.get("document_id")]
398 if not doc_ids: 398 ↛ 399line 398 didn't jump to line 399 because the condition on line 398 was never true
399 return
401 with get_user_db_session(username, db_password) as db_session:
402 rows = (
403 db_session.query(
404 Document.id.label("document_id"),
405 SourceType.name.label("source_type_name"),
406 ResearchHistory.title.label("research_title"),
407 ResearchHistory.query.label("research_query"),
408 ResearchHistory.created_at.label("research_created_at"),
409 Document.research_id,
410 )
411 .outerjoin(SourceType, Document.source_type_id == SourceType.id)
412 .outerjoin(
413 ResearchHistory,
414 Document.research_id == ResearchHistory.id,
415 )
416 .filter(Document.id.in_(doc_ids))
417 .all()
418 )
419 lookup = {row.document_id: row for row in rows}
421 for result in results:
422 row = lookup.get(result.get("document_id"))
423 if row:
424 result["type"] = (
425 "report"
426 if row.source_type_name == "research_report"
427 else "source"
428 )
429 result["research_id"] = row.research_id
430 result["research_title"] = row.research_title or (
431 row.research_query[:100] if row.research_query else ""
432 )
433 result["research_query"] = row.research_query
434 result["research_created_at"] = (
435 row.research_created_at
436 if isinstance(row.research_created_at, str)
437 else row.research_created_at.isoformat()
438 if row.research_created_at
439 else None
440 )
441 else:
442 result["type"] = "source"
443 result["research_id"] = None
444 result["research_title"] = ""
445 result["research_query"] = None
446 result["research_created_at"] = None
449def _enrich_with_document_metadata(results, username, db_password):
450 """Add file type, domain, and creation date to search results."""
451 from ...database.session_context import get_user_db_session
452 from ...security import url_authority_without_userinfo
454 doc_ids = [r["document_id"] for r in results if r.get("document_id")]
455 if not doc_ids:
456 return
458 with get_user_db_session(username, db_password) as db_session:
459 rows = (
460 db_session.query(
461 Document.id.label("document_id"),
462 Document.file_type,
463 Document.original_url,
464 Document.created_at,
465 )
466 .filter(Document.id.in_(doc_ids))
467 .all()
468 )
469 lookup = {row.document_id: row for row in rows}
471 for result in results:
472 row = lookup.get(result.get("document_id"))
473 if row:
474 result["file_type"] = row.file_type
475 result["created_at"] = (
476 row.created_at
477 if isinstance(row.created_at, str)
478 else row.created_at.isoformat()
479 if row.created_at
480 else None
481 )
482 if row.original_url:
483 # Authority minus userinfo, which the JSON would
484 # otherwise carry; "" when unparseable or when the
485 # authority is ambiguous (the helper never raises).
486 result["domain"] = url_authority_without_userinfo(
487 row.original_url
488 )
489 else:
490 result["domain"] = None
491 else:
492 result.setdefault("file_type", "unknown")
493 result.setdefault("domain", None)
494 result.setdefault("created_at", None)