Coverage for src/local_deep_research/web_search_engines/engines/search_engine_library.py: 97%
212 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +0000
1"""
2Library RAG Search Engine
4Provides semantic search over the user's personal research library using RAG.
5"""
7import json
8import os
9from typing import List, Dict, Any, Optional
11from ..search_engine_base import BaseSearchEngine, Exposure, Sensitivity
12from ...security.secure_logging import logger
13from ...constants import (
14 DEFAULT_LOCAL_SEARCH_CHUNK_OVERLAP,
15 DEFAULT_LOCAL_SEARCH_CHUNK_SIZE,
16 DEFAULT_LOCAL_SEARCH_DISTANCE_METRIC,
17 DEFAULT_LOCAL_SEARCH_INDEX_TYPE,
18 DEFAULT_LOCAL_SEARCH_MODEL,
19 DEFAULT_LOCAL_SEARCH_NORMALIZE_VECTORS,
20 DEFAULT_LOCAL_SEARCH_PROVIDER,
21 DEFAULT_LOCAL_SEARCH_SPLITTER_TYPE,
22 DEFAULT_LOCAL_SEARCH_TEXT_SEPARATORS,
23 DEFAULT_LOCAL_SEARCH_TEXT_SEPARATORS_JSON,
24 SNIPPET_LENGTH_LONG,
25)
26from ...research_library.services.library_rag_service import LibraryRAGService
27from ...research_library.services.library_service import LibraryService
28from ...config.thread_settings import get_setting_from_snapshot
29from ...utilities.chunk_anchor import (
30 build_chunk_anchor_url,
31 extract_chunk_index,
32 extract_document_id,
33 is_safe_document_id,
34)
35from ...utilities.llm_utils import get_server_url
36from ...utilities.type_utils import to_bool
37from ...database.models.library import Collection, Document
38from ...research_library.services.pdf_storage_manager import PDFStorageManager
39from ...database.session_context import get_user_db_session
40from ...config.paths import get_library_directory
43class LibraryRAGSearchEngine(BaseSearchEngine):
44 """
45 Search engine that queries the user's research library using RAG/semantic search.
46 """
48 # Mark as local RAG engine for tool selection and safety checks
49 is_local = True
51 # Egress (ADR-0007): a local document store — sensitive data, kept on-box.
52 egress_sensitivity = Sensitivity.SENSITIVE
53 egress_exposure = Exposure.CONTAINED
55 def __init__(
56 self,
57 llm: Optional[Any] = None,
58 max_filtered_results: Optional[int] = None,
59 max_results: int = 10,
60 settings_snapshot: Optional[Dict[str, Any]] = None,
61 search_snippets_only: bool = True,
62 **kwargs,
63 ):
64 """
65 Initialize the library RAG search engine.
67 Args:
68 llm: Language model for relevance filtering
69 max_filtered_results: Maximum number of results to keep after filtering
70 max_results: Maximum number of search results
71 settings_snapshot: Settings snapshot from thread context
72 search_snippets_only: named explicitly because create_search_engine
73 filters its params against this signature and drops the rest
74 **kwargs: Additional engine-specific parameters
75 """
76 super().__init__(
77 llm=llm,
78 max_filtered_results=max_filtered_results,
79 max_results=max_results,
80 settings_snapshot=settings_snapshot,
81 search_snippets_only=search_snippets_only,
82 **kwargs,
83 )
84 self.username = (
85 settings_snapshot.get("_username") if settings_snapshot else None
86 )
87 # Self-stamp the policy registry name so direct instantiations get
88 # the runtime egress backstop without caller cooperation (the
89 # factory stamps registry engines, but routes/services may build
90 # this engine directly). CollectionSearchEngine overrides this
91 # with its collection key after calling super().__init__.
92 self._engine_name = "library"
94 if not self.username:
95 logger.warning(
96 "Library RAG search engine initialized without username"
97 )
99 # Get RAG configuration from settings. NB: keyword arguments — the
100 # signature is (key, default, username, settings_snapshot); these
101 # calls used to pass (key, settings_snapshot, default) positionally,
102 # so the snapshot was never consulted and, outside a thread settings
103 # context, each setting silently resolved to the snapshot DICT
104 # itself (the misplaced "default").
105 self.embedding_model = get_setting_from_snapshot(
106 "local_search_embedding_model",
107 default=DEFAULT_LOCAL_SEARCH_MODEL,
108 settings_snapshot=settings_snapshot,
109 )
110 self.embedding_provider = get_setting_from_snapshot(
111 "local_search_embedding_provider",
112 default=DEFAULT_LOCAL_SEARCH_PROVIDER,
113 settings_snapshot=settings_snapshot,
114 )
115 self.chunk_size = get_setting_from_snapshot(
116 "local_search_chunk_size",
117 default=DEFAULT_LOCAL_SEARCH_CHUNK_SIZE,
118 settings_snapshot=settings_snapshot,
119 )
120 self.chunk_overlap = get_setting_from_snapshot(
121 "local_search_chunk_overlap",
122 default=DEFAULT_LOCAL_SEARCH_CHUNK_OVERLAP,
123 settings_snapshot=settings_snapshot,
124 )
125 self.splitter_type = get_setting_from_snapshot(
126 "local_search_splitter_type",
127 default=DEFAULT_LOCAL_SEARCH_SPLITTER_TYPE,
128 settings_snapshot=settings_snapshot,
129 )
130 raw_text_separators = get_setting_from_snapshot(
131 "local_search_text_separators",
132 default=DEFAULT_LOCAL_SEARCH_TEXT_SEPARATORS_JSON,
133 settings_snapshot=settings_snapshot,
134 )
135 if isinstance(raw_text_separators, str):
136 try:
137 self.text_separators = json.loads(raw_text_separators)
138 except (json.JSONDecodeError, TypeError):
139 self.text_separators = DEFAULT_LOCAL_SEARCH_TEXT_SEPARATORS
140 elif isinstance(raw_text_separators, list): 140 ↛ 141line 140 didn't jump to line 141 because the condition on line 140 was never true
141 self.text_separators = raw_text_separators
142 else:
143 self.text_separators = DEFAULT_LOCAL_SEARCH_TEXT_SEPARATORS
145 self.distance_metric = get_setting_from_snapshot(
146 "local_search_distance_metric",
147 default=DEFAULT_LOCAL_SEARCH_DISTANCE_METRIC,
148 settings_snapshot=settings_snapshot,
149 )
150 self.normalize_vectors = to_bool(
151 get_setting_from_snapshot(
152 "local_search_normalize_vectors",
153 default=DEFAULT_LOCAL_SEARCH_NORMALIZE_VECTORS,
154 settings_snapshot=settings_snapshot,
155 ),
156 default=DEFAULT_LOCAL_SEARCH_NORMALIZE_VECTORS,
157 )
158 self.index_type = get_setting_from_snapshot(
159 "local_search_index_type",
160 default=DEFAULT_LOCAL_SEARCH_INDEX_TYPE,
161 settings_snapshot=settings_snapshot,
162 )
164 # Fallback server URL from the settings snapshot (see
165 # get_server_url). Nothing reads this attribute yet, so no link is
166 # built from it.
167 self.server_url = get_server_url(settings_snapshot)
169 # Local RAG retrieval is vector-similarity ranked and can return many
170 # near-duplicate chunks per query — enough to flood an agent's context
171 # across research iterations. Apply a RAG-specific cap (default 20),
172 # independent of the global ``search.max_results`` (which targets web
173 # engines). This OVERRIDES the inherited max_results for local search.
174 # Default to the inherited max_results so an ABSENT setting preserves
175 # the caller's value (programmatic callers, tests); real web/CLI runs
176 # carry the shipped default of 20 via default_settings.json, so they ARE
177 # capped. A malformed value also falls back to no change.
178 rag_max_results = get_setting_from_snapshot(
179 "search.rag.max_results",
180 default=self.max_results,
181 settings_snapshot=settings_snapshot,
182 )
183 try:
184 rag_max_results = int(rag_max_results)
185 except (TypeError, ValueError):
186 rag_max_results = self.max_results
187 if rag_max_results > 0: 187 ↛ 199line 187 didn't jump to line 199 because the condition on line 187 was always true
188 self.max_results = rag_max_results
190 # Embeddings already rank by cosine similarity, so the LLM relevance
191 # filter is OPT-IN here (unlike keyword engines that hardcode
192 # needs_llm_relevance_filter=True). Default on: prune
193 # semantically-near-but-irrelevant chunks. The factory only ever SETS
194 # enable_llm_relevance_filter=True (for should_filter engines) and never
195 # unsets it — and RAG engines are not should_filter (no
196 # needs_llm_relevance_filter; dynamic ``collection_<uuid>`` name) — so
197 # this assignment is authoritative. The base filter is a no-op without
198 # an llm, so it degrades gracefully.
199 self.enable_llm_relevance_filter = to_bool(
200 get_setting_from_snapshot(
201 "search.rag.enable_relevance_filter",
202 default=True,
203 settings_snapshot=settings_snapshot,
204 ),
205 default=True,
206 )
208 def search(
209 self,
210 query: str,
211 limit: int = 10,
212 llm_callback=None,
213 extra_params: Optional[Dict[str, Any]] = None,
214 ) -> List[Dict[str, Any]]:
215 """
216 Search the library using semantic search.
218 Args:
219 query: Search query
220 limit: Maximum number of results to return
221 llm_callback: Optional LLM callback for processing results
222 extra_params: Additional search parameters
224 Returns:
225 List of search results with title, url, snippet, etc.
226 """
227 if not self.username:
228 logger.error("Cannot search library without username")
229 return []
231 # search() can be called directly (bypassing run()), so apply the
232 # runtime egress backstop here too. Memoized per snapshot, so the
233 # run() -> _get_previews() -> search() path re-checks for free.
234 # Raises PolicyDeniedError on denial — deliberately BEFORE the
235 # broad try/except below.
236 self._verify_egress_scope()
238 try:
239 # Initialize services
240 library_service = LibraryService(username=self.username)
242 # Get all collections for this user
243 collections = library_service.get_all_collections()
244 if not collections:
245 logger.info("No collections found for user")
246 return []
248 # Search across all collections and merge results. Each
249 # (SearchResult, relevance, metadata) tuple is pre-scored per
250 # its OWN collection's metric before merging, so a collection
251 # indexed with cosine/dot_product doesn't get compared on the
252 # wrong scale against one indexed with l2 (see the
253 # relevance-formula comment below).
254 all_results = []
255 failed_collections = []
256 collection_names = {
257 c.get("id"): c.get("name") or f"Collection {c.get('id')}"
258 for c in collections
259 if c.get("id")
260 }
261 for collection in collections:
262 collection_id = collection.get("id")
263 if not collection_id:
264 continue
266 try:
267 # Get the Collection record directly for frozen embedding settings
268 with get_user_db_session(self.username) as session:
269 collection_record = (
270 session.query(Collection)
271 .filter_by(id=collection_id)
272 .first()
273 )
275 if (
276 not collection_record
277 or not collection_record.embedding_model
278 ):
279 logger.debug(
280 f"No embedding settings found for collection {collection_id}"
281 )
282 continue
284 # Get embedding settings directly from Collection
285 embedding_model = collection_record.embedding_model
286 provider = collection_record.embedding_model_type
287 if provider is not None:
288 embedding_provider = (
289 provider.value
290 if hasattr(provider, "value")
291 else str(provider)
292 )
293 else:
294 embedding_provider = self.embedding_provider
295 coll_chunk_size = getattr(
296 collection_record, "chunk_size", None
297 )
298 chunk_size = (
299 int(coll_chunk_size)
300 if coll_chunk_size is not None
301 else int(
302 self.chunk_size
303 or DEFAULT_LOCAL_SEARCH_CHUNK_SIZE
304 )
305 )
306 coll_chunk_overlap = getattr(
307 collection_record, "chunk_overlap", None
308 )
309 chunk_overlap = (
310 int(coll_chunk_overlap)
311 if coll_chunk_overlap is not None
312 else int(
313 self.chunk_overlap
314 if self.chunk_overlap is not None
315 else DEFAULT_LOCAL_SEARCH_CHUNK_OVERLAP
316 )
317 )
318 splitter_type = (
319 getattr(collection_record, "splitter_type", None)
320 or getattr(self, "splitter_type", None)
321 or DEFAULT_LOCAL_SEARCH_SPLITTER_TYPE
322 )
323 text_separators = (
324 collection_record.text_separators
325 if getattr(
326 collection_record, "text_separators", None
327 )
328 is not None
329 else getattr(self, "text_separators", None)
330 )
331 # Thread the stored normalization flag AND distance
332 # metric through, exactly like CollectionSearchEngine.
333 # Without normalize_vectors, LibraryRAGService defaults
334 # True and would L2-normalize the query against a
335 # raw-vector collection, flipping the top hit. Without
336 # distance_metric, every collection is labeled "cosine"
337 # and l2/dot_product hits get the wrong [0,1] mapping in
338 # the cross-collection merge below. NULL → prior default.
339 coll_normalize = getattr(
340 collection_record, "normalize_vectors", None
341 )
342 normalize_vectors = (
343 to_bool(
344 coll_normalize,
345 default=DEFAULT_LOCAL_SEARCH_NORMALIZE_VECTORS,
346 )
347 if coll_normalize is not None
348 else getattr(
349 self,
350 "normalize_vectors",
351 DEFAULT_LOCAL_SEARCH_NORMALIZE_VECTORS,
352 )
353 )
354 distance_metric = (
355 getattr(collection_record, "distance_metric", None)
356 or getattr(self, "distance_metric", None)
357 or DEFAULT_LOCAL_SEARCH_DISTANCE_METRIC
358 )
359 # Thread the stored index_type too (like metric/
360 # normalize): a legacy NULL-index_type row that is
361 # physically HNSW would otherwise fall back to "flat",
362 # mislabelling the store. NULL → "flat".
363 index_type = (
364 getattr(collection_record, "index_type", None)
365 or getattr(self, "index_type", None)
366 or DEFAULT_LOCAL_SEARCH_INDEX_TYPE
367 )
369 # Create RAG service with the collection's embedding settings
370 with LibraryRAGService(
371 username=self.username,
372 embedding_model=embedding_model,
373 embedding_provider=embedding_provider,
374 chunk_size=chunk_size,
375 chunk_overlap=chunk_overlap,
376 splitter_type=splitter_type,
377 text_separators=text_separators,
378 normalize_vectors=normalize_vectors,
379 distance_metric=distance_metric,
380 index_type=index_type,
381 ) as rag_service:
382 # Index-scoped EXISTS check instead of the full stats query
383 if not rag_service.has_indexed_documents(collection_id):
384 logger.debug(
385 f"No indexed documents in collection {collection_id}"
386 )
387 continue
389 # Search this collection's vector index via the
390 # new int-id-keyed store (chunk text is rehydrated
391 # from the encrypted DB by id — never read from the
392 # vector store itself; see the SECURITY INVARIANT
393 # in vector_stores/base.py).
394 search_results = rag_service.search(
395 query, collection_id, limit
396 )
398 # Tag each result with collection ID and pre-score
399 # relevance per THIS collection's metric (so cosine
400 # and L2 sort compatibly on merge).
401 for r in search_results:
402 # Map r.distance to a [0, 1] relevance score:
403 # - cosine/dot_product: (d+1)/2 clamped to [0,1]
404 # - l2 / other: 1/(1+d) in (0, 1]
405 # Must match faiss_store._build_base_index metric
406 # selection (cosine/dot_product -> IP, else L2).
407 relevance = (
408 max(0.0, min(1.0, (r.distance + 1.0) / 2.0))
409 if r.metric in ("cosine", "dot_product")
410 else 1.0 / (1.0 + r.distance)
411 )
412 all_results.append((r, relevance, collection_id))
414 except Exception as e:
415 safe_msg = self._scrub_error(e)
416 logger.exception(
417 f"Error searching collection {collection_id} ({type(e).__name__}): {safe_msg}"
418 )
419 failed_collections.append(collection_id)
420 continue
422 # Fail loudly if *every* collection errored (returning []
423 # masks outright service failure as "no results"). Partial
424 # failures with 0 hits also raise: distinguishing a genuine
425 # "not found" from "the one collection with answers crashed"
426 # is impossible without this check.
427 if failed_collections and not all_results:
428 failed_names = [
429 collection_names.get(cid) or str(cid)
430 for cid in failed_collections
431 ]
432 self._raise_collections_failed(failed_names)
434 if failed_collections and all_results:
435 failed_names = [
436 collection_names.get(cid) or str(cid)
437 for cid in failed_collections
438 ]
439 logger.warning(
440 f"Library search returned partial results: {len(failed_collections)} of "
441 f"{len(collections)} collection(s) failed during search: {failed_names}"
442 )
444 if not all_results:
445 logger.info("No results found across any collections")
446 return []
448 # Sort by pre-scored relevance (descending) and take top limit
449 all_results.sort(key=lambda x: x[1], reverse=True)
450 top_results = all_results[:limit]
452 # Convert to search result format
453 results = []
454 for r, relevance, coll_id in top_results:
455 metadata = dict(r.metadata or {})
456 metadata["collection_id"] = coll_id
457 metadata["collection_name"] = (
458 collection_names.get(coll_id) or "Unknown"
459 )
461 # Get document ID
462 doc_id = (
463 r.source_id
464 or metadata.get("source_id")
465 or metadata.get("document_id")
466 )
468 # Get title
469 title = (
470 r.document_title
471 or metadata.get("document_title")
472 or metadata.get("title")
473 or (f"Document {doc_id}" if doc_id else "Untitled")
474 )
476 # Create snippet from content
477 snippet = (
478 r.text[:SNIPPET_LENGTH_LONG] + "..."
479 if len(r.text) > SNIPPET_LENGTH_LONG
480 else r.text
481 )
483 # Generate document URL. Validate chunk_idx and sanitise
484 # doc_id through the shared helpers so a malformed chunk index
485 # (UUID, boolean, negative) or path-traversal doc_id cannot leak
486 # into the citation.
487 chunk_idx = extract_chunk_index(metadata)
488 # Authoritative id FIRST. ``extract_document_id`` scans its
489 # first mapping before any later one, so passing ``metadata``
490 # first would let a chunk's denormalised metadata override
491 # ``r.source_id`` — the DocumentChunk FK that decides which
492 # document the chunk actually belongs to. Pre-#5381 the
493 # column always won; a divergent metadata id would otherwise
494 # point the citation at a different document.
495 sanitised_doc_id = extract_document_id(
496 {"source_id": doc_id} if doc_id else None,
497 metadata,
498 )
499 document_url = self._get_document_url(
500 sanitised_doc_id, chunk_index=chunk_idx
501 )
503 result = {
504 "title": title,
505 "snippet": snippet,
506 "url": document_url,
507 "link": document_url, # Add "link" for source extraction
508 "source": "library",
509 "source_type": "library",
510 # Authoritative document id at the TOP level, not only
511 # nested in ``metadata``: SearchResultsCollector rebuilds
512 # this citation URL and would otherwise have nothing but
513 # the denormalised metadata to go on.
514 "source_id": doc_id,
515 "relevance_score": float(relevance),
516 "metadata": metadata,
517 }
519 results.append(result)
521 logger.info(f"Library RAG search returned {len(results)} results")
522 return results
524 except Exception as e:
525 # Re-raise so run() records the failure in metrics instead of
526 # treating a failed search as "no matching documents".
527 safe_msg = self._scrub_error(e)
528 logger.exception(
529 f"Error searching library RAG ({type(e).__name__}): {safe_msg}"
530 )
531 raise
533 @staticmethod
534 def _raise_collections_failed(failed_collections: List[str]) -> None:
535 """Signal zero results caused (at least partly) by search failures."""
536 raise RuntimeError(
537 f"Library search returned no results and failed for "
538 f"{len(failed_collections)} collection(s): {failed_collections}"
539 )
541 def _get_previews(
542 self,
543 query: str,
544 limit: Optional[int] = None,
545 llm_callback=None,
546 extra_params: Optional[Dict[str, Any]] = None,
547 ) -> List[Dict[str, Any]]:
548 """
549 Get preview results for the query.
550 Delegates to the search method, using the configured max_results
551 when no explicit limit is provided.
552 """
553 if limit is None:
554 limit = self.max_results
555 return self.search(query, limit, llm_callback, extra_params)
557 def _get_full_content(
558 self, relevant_items: List[Dict[str, Any]]
559 ) -> List[Dict[str, Any]]:
560 """
561 Get full content for relevant library documents or chunks.
562 Retrieves chunk text if a chunk index is targeted, else complete document text.
563 """
564 if not self.username:
565 logger.error("Cannot retrieve full content without username")
566 return relevant_items
568 try:
569 from ...database.models.library import Document, DocumentChunk
570 from ...database.session_context import get_user_db_session
572 # Retrieve content for each document or targeted chunk
573 for item in relevant_items:
574 metadata = item.get("metadata") or {}
575 doc_id = (
576 item.get("source_id")
577 or metadata.get("source_id")
578 or metadata.get("document_id")
579 )
580 if not doc_id:
581 continue
583 chunk_idx = extract_chunk_index(metadata)
585 # Get document/chunk text from database
586 with get_user_db_session(self.username) as db_session:
587 if chunk_idx is not None:
588 query = db_session.query(DocumentChunk).filter(
589 DocumentChunk.source_id == doc_id,
590 DocumentChunk.chunk_index == chunk_idx,
591 )
592 coll_id = metadata.get("collection_id") or item.get(
593 "collection_id"
594 )
595 if coll_id:
596 coll_name = (
597 coll_id
598 if str(coll_id).startswith("collection_")
599 else f"collection_{coll_id}"
600 )
601 query = query.filter(
602 DocumentChunk.collection_name == coll_name
603 )
604 elif metadata.get("collection_name") and str( 604 ↛ 607line 604 didn't jump to line 607 because the condition on line 604 was never true
605 metadata.get("collection_name")
606 ).startswith("collection_"):
607 query = query.filter(
608 DocumentChunk.collection_name
609 == metadata.get("collection_name")
610 )
612 chunk = query.order_by(
613 DocumentChunk.collection_name, DocumentChunk.id
614 ).first()
616 if chunk is not None:
617 text = chunk.chunk_text or ""
618 else:
619 text = f"Chunk {chunk_idx} not found for document {doc_id}."
621 item["content"] = text
622 # BaseCitationHandler reads "full_content" and falls
623 # back to the truncated "snippet".
624 item["full_content"] = text
625 item["full_content_is_complete"] = chunk is not None
626 item["snippet"] = (
627 text[:SNIPPET_LENGTH_LONG] + "..."
628 if len(text) > SNIPPET_LENGTH_LONG
629 else text
630 )
631 logger.debug(
632 f"Retrieved chunk {chunk_idx} content for document {doc_id}"
633 )
634 else:
635 document = (
636 db_session.query(Document)
637 .filter_by(id=doc_id)
638 .first()
639 )
641 if document and document.text_content: 641 ↛ 573line 641 didn't jump to line 573
642 # Replace snippet with full content
643 item["content"] = document.text_content
644 item["full_content"] = document.text_content
645 item["full_content_is_complete"] = True
646 item["snippet"] = (
647 document.text_content[:SNIPPET_LENGTH_LONG]
648 + "..."
649 if len(document.text_content)
650 > SNIPPET_LENGTH_LONG
651 else document.text_content
652 )
653 logger.debug(
654 f"Retrieved full content for document {doc_id}"
655 )
657 return relevant_items
659 except Exception as e:
660 safe_msg = self._scrub_error(e)
661 logger.exception(
662 f"Error retrieving full content from library ({type(e).__name__}): {safe_msg}"
663 )
664 return relevant_items
666 def _get_document_url(
667 self, doc_id: Optional[str], chunk_index: Optional[int] = None
668 ) -> str:
669 """Get the URL for viewing a document, targeting a specific chunk if provided."""
670 if not doc_id:
671 return "#"
672 # Validate here rather than trusting the caller. Both in-tree call
673 # sites pre-sanitise via ``extract_document_id``, but this method
674 # BUILDS a URL path, so an unvalidated id would be interpolated
675 # straight into it — a ``../..`` id yields a traversal route on all
676 # three spellings below, not just the chunk one.
677 if not is_safe_document_id(str(doc_id)):
678 return "#"
680 if chunk_index is not None:
681 # Route through the shared builder instead of re-deriving the
682 # rule. The hand-rolled test this replaces accepted any
683 # non-negative int, so it had no ``MAX_CHUNK_INDEX`` bound —
684 # a third, weaker copy of a contract ``chunk_anchor`` calls
685 # itself the single source of truth for. On rejection fall
686 # through to the plain document route rather than emitting an
687 # anchor that cannot name a chunk.
688 anchored = build_chunk_anchor_url("", doc_id, chunk_index)
689 if anchored is not None:
690 return anchored
692 document_url = f"/library/document/{doc_id}"
693 try:
694 with get_user_db_session(self.username) as session:
695 document = session.query(Document).filter_by(id=doc_id).first()
696 if document: 696 ↛ 732line 696 didn't jump to line 732
697 from pathlib import Path
698 from ...research_library.utils import apply_user_subdir
700 library_root = get_setting_from_snapshot(
701 "research_library.storage_path",
702 default=str(get_library_directory()),
703 settings_snapshot=self.settings_snapshot,
704 )
705 base_root = (
706 Path(os.path.expandvars(library_root))
707 .expanduser()
708 .resolve()
709 )
710 shared_library = get_setting_from_snapshot(
711 "research_library.shared_library",
712 default=False,
713 settings_snapshot=self.settings_snapshot,
714 )
715 # PDF availability uses the same per-user root as serving.
716 per_user_root = apply_user_subdir(
717 base_root, self.username, shared_library
718 )
719 if PDFStorageManager.pdf_exists(
720 per_user_root,
721 document,
722 session,
723 ):
724 document_url = f"/library/document/{doc_id}/pdf"
725 except Exception as e:
726 safe_msg = self._scrub_error(e)
727 logger.warning(
728 f"Error getting document URL for {doc_id} "
729 f"({type(e).__name__}): {safe_msg}"
730 )
732 return document_url
734 def close(self):
735 """Clean up resources."""
736 pass