Coverage for src/local_deep_research/research_library/services/library_service.py: 90%
393 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-06 15:42 +0000
1"""
2Library Management Service
4Handles querying and managing the downloaded document library:
5- Search and filter documents
6- Get statistics and analytics
7- Manage collections and favorites
8- Handle file operations
9"""
11from pathlib import Path
12from typing import Dict, List, Optional
13from urllib.parse import urlparse
15from loguru import logger
16from sqlalchemy import or_, func, case
17from sqlalchemy.orm import aliased, defer
19from ...constants import FILE_PATH_SENTINELS
20from ...database.models.download_tracker import DownloadTracker
21from ...database.models.library import (
22 Collection,
23 Document,
24 DocumentBlob,
25 DocumentCollection,
26 DocumentStatus,
27)
28from ...database.models.metrics import ResearchRating
29from ...database.models.research import ResearchHistory, ResearchResource
30from ...database.session_context import get_user_db_session
31from ...config.paths import get_library_directory
32from ...utilities.db_utils import get_settings_manager
33from ...utilities.resource_utils import safe_close
34from ...utilities.sql_utils import escape_like as _escape_like
35from ..utils import (
36 get_absolute_path_from_settings,
37 get_url_hash,
38 open_file_location,
39)
41# Filter dropdowns (library page + ``/api/research-list``) load research
42# sessions into a client-side ``<select>``. Bound the query so a very large
43# history cannot pull every row into memory / the DOM on each page load
44# (#4560). The query already projects to three small columns, so this is a
45# safety cap rather than a crash fix; realistic histories are far smaller, and
46# comprehensive large-scale filtering is a server-side-search follow-up.
47_DROPDOWN_RESEARCH_LIMIT = 1000
49# ``get_unique_domains`` scans one URL row per downloaded document to build the
50# domain filter. Stream it in batches (``yield_per``) rather than materializing
51# every row at once, so a very large library cannot exhaust memory (#4560). The
52# distinct-domain set it accumulates is small regardless of library size.
53_DOMAIN_SCAN_BATCH_SIZE = 1000
56class LibraryService:
57 """Service for managing and querying the document library."""
59 def __init__(self, username: str):
60 """Initialize library service for a user."""
61 self.username = username
63 def _has_blob_in_db(self, session, document_id: str) -> bool:
64 """Check if a PDF blob exists in the database for a document."""
65 return (
66 session.query(DocumentBlob.document_id)
67 .filter_by(document_id=document_id)
68 .first()
69 is not None
70 )
72 def _get_safe_absolute_path(self, file_path: str) -> Optional[str]:
73 """
74 Get the absolute path for a file, safely handling invalid paths.
76 Args:
77 file_path: Relative file path from library root
79 Returns:
80 Absolute path as string, or None if path is invalid/unsafe
81 """
82 if not file_path or file_path in FILE_PATH_SENTINELS:
83 return None
84 abs_path = get_absolute_path_from_settings(file_path, self.username)
85 return str(abs_path) if abs_path else None
87 def _is_arxiv_url(self, url: str) -> bool:
88 """Check if URL is from arXiv domain."""
89 try:
90 hostname = urlparse(url).hostname
91 return bool(
92 hostname
93 and (hostname == "arxiv.org" or hostname.endswith(".arxiv.org"))
94 )
95 except Exception:
96 return False
98 def _is_pubmed_url(self, url: str) -> bool:
99 """Check if URL is from PubMed or NCBI domains."""
100 try:
101 parsed = urlparse(url)
102 hostname = parsed.hostname
103 if not hostname:
104 return False
106 # Check for pubmed.ncbi.nlm.nih.gov
107 if hostname == "pubmed.ncbi.nlm.nih.gov":
108 return True
110 # Check for ncbi.nlm.nih.gov with PMC path
111 if hostname == "ncbi.nlm.nih.gov" and "/pmc" in parsed.path:
112 return True
114 # Check for pubmed in subdomain
115 if "pubmed" in hostname:
116 return True
118 return False
119 except Exception:
120 return False
122 def _apply_date_filter(self, query, model_class, date_filter: str):
123 """Apply date range filter based on processed_at timestamp."""
124 from datetime import datetime, timedelta, timezone
126 now = datetime.now(timezone.utc)
127 if date_filter == "today":
128 cutoff = now.replace(hour=0, minute=0, second=0, microsecond=0)
129 elif date_filter == "week":
130 cutoff = now - timedelta(days=7)
131 elif date_filter == "month":
132 cutoff = now - timedelta(days=30)
133 else:
134 return query
135 return query.filter(model_class.processed_at >= cutoff)
137 def _apply_domain_filter(self, query, model_class, domain: str):
138 """Apply domain filter to query for Document.
140 The dropdown is fully data-driven (populated from get_unique_domains),
141 so the filter is a generic substring match against original_url.
142 """
143 pattern = f"%{_escape_like(domain)}%"
144 return query.filter(model_class.original_url.like(pattern, escape="\\"))
146 def _apply_search_filter(self, query, model_class, search_query: str):
147 """Apply search filter to query for Document."""
148 search_pattern = f"%{_escape_like(search_query)}%"
149 return query.filter(
150 or_(
151 model_class.title.ilike(search_pattern, escape="\\"),
152 model_class.authors.ilike(search_pattern, escape="\\"),
153 model_class.doi.ilike(search_pattern, escape="\\"),
154 ResearchResource.title.ilike(search_pattern, escape="\\"),
155 )
156 )
158 def get_library_stats(self) -> Dict:
159 """Get overall library statistics."""
160 with get_user_db_session(self.username) as session:
161 # Get document counts
162 total_docs = session.query(Document).count()
163 total_pdfs = (
164 session.query(Document).filter_by(file_type="pdf").count()
165 )
167 # Get size stats
168 size_result = session.query(
169 func.sum(Document.file_size),
170 func.avg(Document.file_size),
171 ).first()
173 total_size = size_result[0] or 0
174 avg_size = size_result[1] or 0
176 # Get research stats
177 research_count = session.query(
178 func.count(func.distinct(Document.research_id))
179 ).scalar()
181 # Get domain stats - count unique domains from URLs
182 # Extract domain from original_url using SQL functions
183 from sqlalchemy import case, func as sql_func
185 # Count unique domains by extracting them from URLs
186 domain_subquery = session.query(
187 sql_func.distinct(
188 case(
189 (
190 Document.original_url.like("%arxiv.org%"),
191 "arxiv.org",
192 ),
193 (
194 Document.original_url.like("%pubmed%"),
195 "pubmed",
196 ),
197 (
198 Document.original_url.like("%ncbi.nlm.nih.gov%"),
199 "pubmed",
200 ),
201 else_="other",
202 )
203 )
204 ).subquery()
206 domain_count = (
207 session.query(sql_func.count())
208 .select_from(domain_subquery)
209 .scalar()
210 )
212 # Get download tracker stats
213 pending_downloads = (
214 session.query(DownloadTracker)
215 .filter_by(is_downloaded=False)
216 .count()
217 )
219 return {
220 "total_documents": total_docs,
221 "total_pdfs": total_pdfs,
222 "total_size_bytes": total_size,
223 "total_size_mb": total_size / (1024 * 1024)
224 if total_size
225 else 0,
226 "average_size_mb": avg_size / (1024 * 1024) if avg_size else 0,
227 "research_sessions": research_count,
228 "unique_domains": domain_count,
229 "pending_downloads": pending_downloads,
230 "storage_path": self._get_storage_path(),
231 }
233 def count_documents(
234 self,
235 research_id: Optional[str] = None,
236 domain: Optional[str] = None,
237 collection_id: Optional[str] = None,
238 date_filter: Optional[str] = None,
239 ) -> int:
240 """Count documents matching the given filters (for pagination)."""
241 with get_user_db_session(self.username) as session:
242 from ...database.library_init import get_default_library_id
244 if not collection_id: 244 ↛ 245line 244 didn't jump to line 245 because the condition on line 244 was never true
245 collection_id = get_default_library_id(self.username)
247 q = (
248 session.query(func.count(Document.id))
249 .join(
250 DocumentCollection,
251 Document.id == DocumentCollection.document_id,
252 )
253 .filter(DocumentCollection.collection_id == collection_id)
254 .filter(Document.status == "completed")
255 )
257 if research_id: 257 ↛ 258line 257 didn't jump to line 258 because the condition on line 257 was never true
258 q = q.filter(Document.research_id == research_id)
259 if domain: 259 ↛ 260line 259 didn't jump to line 260 because the condition on line 259 was never true
260 q = self._apply_domain_filter(q, Document, domain)
261 if date_filter: 261 ↛ 262line 261 didn't jump to line 262 because the condition on line 261 was never true
262 q = self._apply_date_filter(q, Document, date_filter)
264 return q.scalar() or 0
266 def get_documents(
267 self,
268 research_id: Optional[str] = None,
269 domain: Optional[str] = None,
270 file_type: Optional[str] = None,
271 favorites_only: bool = False,
272 search_query: Optional[str] = None,
273 collection_id: Optional[str] = None,
274 date_filter: Optional[str] = None,
275 limit: int = 100,
276 offset: int = 0,
277 ) -> List[Dict]:
278 """
279 Get documents with filtering options.
281 Returns enriched document information with research details.
282 """
283 with get_user_db_session(self.username) as session:
284 # Get default Library collection ID if not specified
285 from ...database.library_init import get_default_library_id
287 if not collection_id:
288 collection_id = get_default_library_id(self.username)
290 logger.info(
291 f"[LibraryService] Getting documents for collection_id: {collection_id}, research_id: {research_id}, domain: {domain}"
292 )
294 all_documents = []
296 # Step 1: subquery to get paginated document IDs.
297 # Pagination and sorting happen here on Document alone,
298 # avoiding non-determinism from outer-joining related tables.
299 doc_subq = (
300 session.query(Document.id)
301 .join(
302 DocumentCollection,
303 Document.id == DocumentCollection.document_id,
304 )
305 .filter(DocumentCollection.collection_id == collection_id)
306 )
308 # Apply filters
309 if research_id: 309 ↛ 310line 309 didn't jump to line 310 because the condition on line 309 was never true
310 doc_subq = doc_subq.filter(Document.research_id == research_id)
312 if domain: 312 ↛ 313line 312 didn't jump to line 313 because the condition on line 312 was never true
313 doc_subq = self._apply_domain_filter(doc_subq, Document, domain)
315 if date_filter: 315 ↛ 316line 315 didn't jump to line 316 because the condition on line 315 was never true
316 doc_subq = self._apply_date_filter(
317 doc_subq, Document, date_filter
318 )
320 if file_type: 320 ↛ 321line 320 didn't jump to line 321 because the condition on line 320 was never true
321 doc_subq = doc_subq.filter(Document.file_type == file_type)
323 if favorites_only: 323 ↛ 324line 323 didn't jump to line 324 because the condition on line 323 was never true
324 doc_subq = doc_subq.filter(Document.favorite.is_(True))
326 if search_query:
327 # _apply_search_filter references ResearchResource.title,
328 # so we must outerjoin it here. DISTINCT prevents fan-out
329 # from the outerjoin duplicating Document IDs.
330 doc_subq = doc_subq.outerjoin(
331 ResearchResource,
332 (Document.resource_id == ResearchResource.id)
333 | (ResearchResource.document_id == Document.id),
334 ).distinct()
335 doc_subq = self._apply_search_filter(
336 doc_subq, Document, search_query
337 )
339 # Filter to only completed documents
340 doc_subq = doc_subq.filter(Document.status == "completed")
342 # Sort at SQL level (SQLite-safe NULL handling)
343 doc_subq = doc_subq.order_by(
344 case((Document.processed_at.isnot(None), 0), else_=1),
345 Document.processed_at.desc(),
346 )
348 # Apply SQL-level pagination
349 doc_subq = doc_subq.offset(offset).limit(limit)
350 doc_id_subq = doc_subq.subquery()
352 # Step 2: join the paginated document IDs with related tables.
353 # Use two separate outer joins for ResearchResource to avoid
354 # the OR-condition join that can fan out to multiple rows:
355 # - ResourceByFK: matched via Document.resource_id (primary FK)
356 # - ResourceByDoc: matched via ResearchResource.document_id
357 # We prefer ResourceByFK; fall back to ResourceByDoc in Python.
358 ResourceByFK = aliased(ResearchResource)
359 ResourceByDoc = aliased(ResearchResource)
361 query = (
362 session.query(
363 Document,
364 ResourceByFK,
365 ResourceByDoc,
366 ResearchHistory,
367 DocumentCollection,
368 )
369 .join(doc_id_subq, Document.id == doc_id_subq.c.id)
370 .join(
371 DocumentCollection,
372 Document.id == DocumentCollection.document_id,
373 )
374 .outerjoin(
375 ResourceByFK,
376 Document.resource_id == ResourceByFK.id,
377 )
378 .outerjoin(
379 ResourceByDoc,
380 ResourceByDoc.document_id == Document.id,
381 )
382 .outerjoin(
383 ResearchHistory,
384 Document.research_id == ResearchHistory.id,
385 )
386 .filter(DocumentCollection.collection_id == collection_id)
387 # Re-apply sort so final results are ordered
388 .order_by(
389 case((Document.processed_at.isnot(None), 0), else_=1),
390 Document.processed_at.desc(),
391 )
392 )
394 # Execute query
395 results = query.all()
396 logger.info(
397 f"[LibraryService] Found {len(results)} documents in collection {collection_id}"
398 )
400 # Batch-check blob existence to avoid N+1 queries
401 doc_ids = [row[0].id for row in results]
402 blob_ids = set()
403 if doc_ids:
404 blob_ids = {
405 r[0]
406 for r in session.query(DocumentBlob.document_id)
407 .filter(DocumentBlob.document_id.in_(doc_ids))
408 .all()
409 }
411 # Process results — deduplicate by doc.id since the ResourceByDoc
412 # outer join can fan out when multiple ResearchResource rows
413 # point to the same document via document_id.
414 #
415 # Resolve one settings manager for the whole page (lazily, on the
416 # first document that actually needs a path resolved) instead of
417 # letting get_absolute_path_from_settings() fall back to the
418 # ambient get_settings_manager() per iteration (up to `limit`
419 # times). Each ambient call opens a DB session via
420 # get_db_session(), which is now tracked in a per-thread owner
421 # registry (utilities/db_utils.py) until this request's cleanup
422 # boundary — so on a busy multi-user instance, enough process-wide
423 # LRU evictions between iterations can pin one pooled connection
424 # per evicted-and-recreated session for the rest of the request
425 # (see changelog/#6095 follow-up). Creating it once here bounds
426 # that to at most one session for this call, mirroring the
427 # pattern already used in DownloadService (self.settings,
428 # threaded through get_absolute_path_from_settings calls).
429 seen_doc_ids = set()
430 settings_manager = None
431 try:
432 for (
433 doc,
434 res_by_fk,
435 res_by_doc,
436 research,
437 doc_collection,
438 ) in results:
439 if doc.id in seen_doc_ids:
440 continue
441 seen_doc_ids.add(doc.id)
442 # Prefer the resource matched via Document.resource_id FK;
443 # fall back to the one matched via ResearchResource.document_id.
444 resource = res_by_fk or res_by_doc
445 # Determine availability flags - use Document.file_path directly
446 file_absolute_path = None
447 if (
448 doc.file_path
449 and doc.file_path not in FILE_PATH_SENTINELS
450 ):
451 if settings_manager is None:
452 settings_manager = get_settings_manager(
453 username=self.username
454 )
455 abs_path = get_absolute_path_from_settings(
456 doc.file_path,
457 self.username,
458 settings_manager=settings_manager,
459 )
460 if abs_path:
461 file_absolute_path = str(abs_path)
463 # Check if PDF is available (filesystem OR database)
464 has_pdf = bool(file_absolute_path)
465 if not has_pdf and doc.storage_mode == "database":
466 has_pdf = doc.id in blob_ids
467 has_text_db = bool(doc.text_content) # Text now in Document
469 # Use DocumentCollection from query results
470 has_rag_indexed = (
471 doc_collection.indexed if doc_collection else False
472 )
473 rag_chunk_count = (
474 doc_collection.chunk_count if doc_collection else 0
475 )
477 all_documents.append(
478 {
479 "id": doc.id,
480 "resource_id": doc.resource_id,
481 "research_id": doc.research_id,
482 # Document info
483 "document_title": doc.title
484 or (resource.title if resource else doc.filename),
485 "authors": doc.authors,
486 "published_date": doc.published_date,
487 "doi": doc.doi,
488 "arxiv_id": doc.arxiv_id,
489 "pmid": doc.pmid,
490 # File info
491 "file_path": doc.file_path,
492 "file_absolute_path": file_absolute_path,
493 "file_name": Path(doc.file_path).name
494 if doc.file_path
495 and doc.file_path not in FILE_PATH_SENTINELS
496 else doc.filename,
497 "file_size": doc.file_size,
498 "file_type": doc.file_type,
499 # URLs
500 "original_url": doc.original_url,
501 "domain": self._extract_domain(doc.original_url)
502 if doc.original_url
503 else "User Upload",
504 # Status
505 "download_status": doc.status or "completed",
506 "downloaded_at": doc.processed_at.isoformat()
507 if doc.processed_at
508 else (
509 doc.uploaded_at.isoformat()
510 if hasattr(doc, "uploaded_at")
511 and doc.uploaded_at
512 else None
513 ),
514 "favorite": doc.favorite
515 if hasattr(doc, "favorite")
516 else False,
517 "tags": doc.tags if hasattr(doc, "tags") else [],
518 # Research info (None for user uploads)
519 "research_title": research.title
520 or research.query[:80]
521 if research
522 else "User Upload",
523 "research_query": research.query
524 if research
525 else None,
526 "research_mode": research.mode
527 if research
528 else None,
529 "research_date": research.created_at
530 if research
531 else None,
532 # Classification flags
533 "is_arxiv": self._is_arxiv_url(doc.original_url)
534 if doc.original_url
535 else False,
536 "is_pubmed": self._is_pubmed_url(doc.original_url)
537 if doc.original_url
538 else False,
539 "is_pdf": doc.file_type == "pdf",
540 # Availability flags
541 "has_pdf": has_pdf,
542 "has_text_db": has_text_db,
543 "has_rag_indexed": has_rag_indexed,
544 "rag_chunk_count": rag_chunk_count,
545 }
546 )
548 # Sorting and pagination are now handled at SQL level
549 return all_documents
550 finally:
551 safe_close(
552 settings_manager, "settings manager", allow_none=True
553 )
555 def get_all_collections(self) -> List[Dict]:
556 """Get all collections with document and indexed document counts."""
557 with get_user_db_session(self.username) as session:
558 # Query collections with document counts and indexed counts
559 results = (
560 session.query(
561 Collection,
562 func.count(DocumentCollection.document_id).label(
563 "document_count"
564 ),
565 func.count(
566 case(
567 (
568 DocumentCollection.indexed == True, # noqa: E712
569 DocumentCollection.document_id,
570 ),
571 else_=None,
572 )
573 ).label("indexed_document_count"),
574 )
575 .outerjoin(
576 DocumentCollection,
577 Collection.id == DocumentCollection.collection_id,
578 )
579 .group_by(Collection.id)
580 .order_by(Collection.is_default.desc(), Collection.name)
581 .all()
582 )
584 logger.info(f"[LibraryService] Found {len(results)} collections")
586 collections = []
587 for collection, doc_count, indexed_count in results:
588 logger.debug(
589 f"[LibraryService] Collection: {collection.name} (ID: {collection.id}), documents: {doc_count}, indexed: {indexed_count}"
590 )
591 collections.append(
592 {
593 "id": collection.id,
594 "name": collection.name,
595 "description": collection.description,
596 "is_default": collection.is_default,
597 "document_count": doc_count or 0,
598 "indexed_document_count": indexed_count or 0,
599 }
600 )
602 return collections
604 def get_research_list_for_dropdown(self) -> List[Dict]:
605 """Get minimal research session info for filter dropdowns.
607 Returns only id, title, and query — no joins or aggregates — for at
608 most the ``_DROPDOWN_RESEARCH_LIMIT`` most-recent sessions, so a very
609 large history cannot load every row into memory / the DOM (#4560).
610 """
611 with get_user_db_session(self.username) as session:
612 results = (
613 session.query(
614 ResearchHistory.id,
615 ResearchHistory.title,
616 ResearchHistory.query,
617 )
618 .order_by(ResearchHistory.created_at.desc())
619 .limit(_DROPDOWN_RESEARCH_LIMIT)
620 .all()
621 )
622 return [
623 {"id": r.id, "title": r.title, "query": r.query}
624 for r in results
625 ]
627 def get_research_list_with_stats(
628 self,
629 limit: int = 0,
630 offset: int = 0,
631 ) -> List[Dict]:
632 """Get research sessions with download statistics.
634 Args:
635 limit: Maximum number of results (0 = no limit, for backwards compat).
636 offset: Number of rows to skip. Only applied when limit > 0.
637 """
638 with get_user_db_session(self.username) as session:
639 # Query research sessions with resource counts
640 query = (
641 session.query(
642 ResearchHistory,
643 func.count(func.distinct(ResearchResource.id)).label(
644 "total_resources"
645 ),
646 func.count(
647 func.distinct(
648 case(
649 (Document.status == "completed", Document.id),
650 else_=None,
651 )
652 )
653 ).label("downloaded_count"),
654 func.count(
655 func.distinct(
656 case(
657 (
658 ResearchResource.url.like("%.pdf")
659 | ResearchResource.url.like("%arxiv.org%")
660 | ResearchResource.url.like(
661 "%ncbi.nlm.nih.gov/pmc%"
662 ),
663 ResearchResource.id,
664 ),
665 else_=None,
666 )
667 )
668 ).label("downloadable_count"),
669 )
670 .outerjoin(
671 ResearchResource,
672 ResearchHistory.id == ResearchResource.research_id,
673 )
674 .outerjoin(
675 Document,
676 (ResearchResource.id == Document.resource_id)
677 | (ResearchResource.document_id == Document.id),
678 )
679 .group_by(ResearchHistory.id)
680 .order_by(ResearchHistory.created_at.desc())
681 )
683 # Apply SQL-level pagination when limit is set
684 if limit > 0:
685 query = query.offset(offset).limit(limit)
687 results = query.all()
689 # Preload all ratings to avoid N+1 queries
690 research_ids = [r[0].id for r in results]
691 all_ratings = (
692 session.query(ResearchRating)
693 .filter(ResearchRating.research_id.in_(research_ids))
694 .all()
695 if research_ids
696 else []
697 )
698 ratings_by_research = {r.research_id: r for r in all_ratings}
700 # Batch domain queries to avoid N+1 (same pattern as ratings)
701 domain_case = case(
702 (
703 ResearchResource.url.like("%arxiv.org%"),
704 "arxiv.org",
705 ),
706 (ResearchResource.url.like("%pubmed%"), "pubmed"),
707 (
708 ResearchResource.url.like("%ncbi.nlm.nih.gov%"),
709 "pubmed",
710 ),
711 else_="other",
712 )
713 all_domains = (
714 session.query(
715 ResearchResource.research_id,
716 domain_case.label("domain"),
717 func.count().label("count"),
718 )
719 .filter(ResearchResource.research_id.in_(research_ids))
720 .group_by(ResearchResource.research_id, domain_case)
721 .all()
722 if research_ids
723 else []
724 )
725 domains_by_research: Dict[str, list] = {}
726 for rid, domain, count in all_domains:
727 domains_by_research.setdefault(rid, []).append((domain, count))
729 research_list = []
730 for (
731 research,
732 total_resources,
733 downloaded_count,
734 downloadable_count,
735 ) in results:
736 # Get rating from preloaded dict
737 rating = ratings_by_research.get(research.id)
739 # Get domain breakdown from preloaded dict
740 domains = domains_by_research.get(research.id, [])
742 research_list.append(
743 {
744 "id": research.id,
745 "title": research.title,
746 "query": research.query,
747 "mode": research.mode,
748 "status": research.status,
749 "created_at": research.created_at,
750 "duration_seconds": research.duration_seconds,
751 "total_resources": total_resources or 0,
752 "downloaded_count": downloaded_count or 0,
753 "downloadable_count": downloadable_count or 0,
754 "rating": rating.rating if rating else None,
755 "top_domains": [(d, c) for d, c in domains if d],
756 }
757 )
759 return research_list
761 def get_download_manager_summary_stats(self) -> Dict:
762 """Get aggregate download stats across ALL research sessions.
764 This is a lightweight query that only returns totals — used for
765 the download-manager header so stats remain accurate regardless
766 of which page the user is viewing.
767 """
768 with get_user_db_session(self.username) as session:
769 row = (
770 session.query(
771 func.count(func.distinct(ResearchHistory.id)).label(
772 "total_researches"
773 ),
774 func.count(func.distinct(ResearchResource.id)).label(
775 "total_resources"
776 ),
777 func.count(
778 func.distinct(
779 case(
780 (Document.status == "completed", Document.id),
781 else_=None,
782 )
783 )
784 ).label("downloaded_count"),
785 func.count(
786 func.distinct(
787 case(
788 (
789 ResearchResource.url.like("%.pdf")
790 | ResearchResource.url.like("%arxiv.org%")
791 | ResearchResource.url.like(
792 "%ncbi.nlm.nih.gov/pmc%"
793 ),
794 ResearchResource.id,
795 ),
796 else_=None,
797 )
798 )
799 ).label("downloadable_count"),
800 )
801 .select_from(ResearchHistory)
802 .outerjoin(
803 ResearchResource,
804 ResearchHistory.id == ResearchResource.research_id,
805 )
806 .outerjoin(
807 Document,
808 (ResearchResource.id == Document.resource_id)
809 | (ResearchResource.document_id == Document.id),
810 )
811 .one()
812 )
814 total_researches = row.total_researches or 0
815 total_resources = row.total_resources or 0
816 downloaded = row.downloaded_count or 0
817 downloadable = row.downloadable_count or 0
819 return {
820 "total_researches": total_researches,
821 "total_resources": total_resources,
822 "already_downloaded": downloaded,
823 "available_to_download": max(downloadable - downloaded, 0),
824 }
826 def get_pdf_previews_batch(
827 self, research_ids: List, limit_per_research: int = 10
828 ) -> Dict[str, Dict]:
829 """Batch-fetch PDF documents and domain breakdowns for multiple research sessions.
831 Returns a dict keyed by research_id with:
832 - "pdf_sources": list of document dicts (capped at limit_per_research)
833 - "domains": dict of domain -> {total, pdfs, downloaded}
834 """
835 if not research_ids:
836 return {}
838 with get_user_db_session(self.username) as session:
839 results = (
840 session.query(Document, ResearchResource)
841 .outerjoin(
842 ResearchResource,
843 (Document.resource_id == ResearchResource.id)
844 | (ResearchResource.document_id == Document.id),
845 )
846 .filter(
847 Document.research_id.in_(research_ids),
848 Document.file_type == "pdf",
849 )
850 .order_by(Document.processed_at.desc())
851 .limit(limit_per_research * len(research_ids))
852 .all()
853 )
855 previews: Dict[str, Dict] = {}
856 seen_doc_ids: set = set()
857 for doc, resource in results:
858 if doc.id in seen_doc_ids:
859 continue
860 seen_doc_ids.add(doc.id)
862 rid = doc.research_id
863 if rid not in previews:
864 previews[rid] = {"pdf_sources": [], "domains": {}}
866 entry = previews[rid]
868 # Domain breakdown (within the SQL LIMIT budget)
869 domain = "unknown"
870 if resource and resource.url:
871 try:
872 domain = urlparse(resource.url).netloc or "unknown"
873 except Exception:
874 logger.debug("Failed to parse resource URL for domain")
875 elif doc.original_url: 875 ↛ 881line 875 didn't jump to line 881 because the condition on line 875 was always true
876 try:
877 domain = urlparse(doc.original_url).netloc or "unknown"
878 except Exception:
879 logger.debug("Failed to parse document URL for domain")
881 if domain not in entry["domains"]:
882 entry["domains"][domain] = {
883 "total": 0,
884 "pdfs": 0,
885 "downloaded": 0,
886 }
887 entry["domains"][domain]["total"] += 1
888 if doc.file_type == "pdf": 888 ↛ 890line 888 didn't jump to line 890 because the condition on line 888 was always true
889 entry["domains"][domain]["pdfs"] += 1
890 if doc.status == "completed":
891 entry["domains"][domain]["downloaded"] += 1
893 # PDF sources preview (capped)
894 if len(entry["pdf_sources"]) < limit_per_research:
895 title = "Untitled"
896 if resource and resource.title:
897 title = resource.title
898 elif doc.filename: 898 ↛ 901line 898 didn't jump to line 901 because the condition on line 898 was always true
899 title = doc.filename
901 entry["pdf_sources"].append(
902 {
903 "document_title": title,
904 "domain": domain,
905 "file_type": doc.file_type,
906 "download_status": doc.status or "unknown",
907 }
908 )
910 return previews
912 def get_document_by_id(self, doc_id: str) -> Optional[Dict]:
913 """
914 Get a specific document by its ID.
916 Returns document information with file path.
917 """
918 with get_user_db_session(self.username) as session:
919 # Find document - use outer joins to support both research downloads and user uploads
920 result = (
921 session.query(Document, ResearchResource, ResearchHistory)
922 .outerjoin(
923 ResearchResource,
924 (Document.resource_id == ResearchResource.id)
925 | (ResearchResource.document_id == Document.id),
926 )
927 .outerjoin(
928 ResearchHistory,
929 Document.research_id == ResearchHistory.id,
930 )
931 .filter(Document.id == doc_id)
932 .first()
933 )
935 if result:
936 # Found document
937 doc, resource, research = result
939 # Get RAG indexing status across all collections
940 doc_collections = (
941 session.query(DocumentCollection, Collection)
942 .join(Collection)
943 .filter(DocumentCollection.document_id == doc_id)
944 .all()
945 )
947 # Check if indexed in any collection
948 has_rag_indexed = any(
949 dc.indexed for dc, coll in doc_collections
950 )
951 total_chunks = sum(
952 dc.chunk_count for dc, coll in doc_collections if dc.indexed
953 )
955 # Build collections list
956 collections_list = [
957 {
958 "id": coll.id,
959 "name": coll.name,
960 "indexed": dc.indexed,
961 "chunk_count": dc.chunk_count,
962 }
963 for dc, coll in doc_collections
964 ]
966 # Calculate word count from text content
967 word_count = (
968 len(doc.text_content.split()) if doc.text_content else 0
969 )
971 # Check if PDF is available (database OR filesystem)
972 has_pdf = bool(
973 doc.file_path and doc.file_path not in FILE_PATH_SENTINELS
974 )
975 if not has_pdf and doc.storage_mode == "database":
976 has_pdf = self._has_blob_in_db(session, doc.id)
978 return {
979 "id": doc.id,
980 "resource_id": doc.resource_id,
981 "research_id": doc.research_id,
982 "document_title": doc.title
983 or (resource.title if resource else doc.filename),
984 "original_url": doc.original_url
985 or (resource.url if resource else None),
986 "file_path": doc.file_path,
987 "file_absolute_path": self._get_safe_absolute_path(
988 doc.file_path
989 ),
990 "file_name": Path(doc.file_path).name
991 if doc.file_path
992 and doc.file_path not in FILE_PATH_SENTINELS
993 else doc.filename,
994 "file_size": doc.file_size,
995 "file_type": doc.file_type,
996 "mime_type": doc.mime_type,
997 "domain": self._extract_domain(resource.url)
998 if resource
999 else "User Upload",
1000 "download_status": doc.status,
1001 "downloaded_at": doc.processed_at.isoformat()
1002 if doc.processed_at
1003 and hasattr(doc.processed_at, "isoformat")
1004 else str(doc.processed_at)
1005 if doc.processed_at
1006 else (
1007 doc.uploaded_at.isoformat()
1008 if hasattr(doc, "uploaded_at") and doc.uploaded_at
1009 else None
1010 ),
1011 "favorite": doc.favorite
1012 if hasattr(doc, "favorite")
1013 else False,
1014 "tags": doc.tags if hasattr(doc, "tags") else [],
1015 "research_title": research.query[:100]
1016 if research
1017 else "User Upload",
1018 "research_created_at": research.created_at
1019 if research and isinstance(research.created_at, str)
1020 else research.created_at.isoformat()
1021 if research and research.created_at
1022 else None,
1023 # Document fields
1024 "is_pdf": doc.file_type == "pdf",
1025 "has_pdf": has_pdf,
1026 "has_text_db": bool(doc.text_content),
1027 "has_rag_indexed": has_rag_indexed,
1028 "rag_chunk_count": total_chunks,
1029 "word_count": word_count,
1030 "collections": collections_list,
1031 }
1033 # Not found
1034 return None
1036 def toggle_favorite(self, document_id: str) -> bool:
1037 """Toggle favorite status of a document."""
1038 with get_user_db_session(self.username) as session:
1039 doc = session.get(Document, document_id)
1040 if doc:
1041 doc.favorite = not doc.favorite
1042 session.commit()
1043 return doc.favorite
1044 return False
1046 def delete_document(self, document_id: str) -> bool:
1047 """Delete a document from library (file and database entry).
1049 Refuses notes: they have their own deletion API
1050 (DELETE /api/notes/<id>) that performs version/link/synthesis
1051 cleanup. Hard-deleting a note here would bypass that and destroy
1052 the note's history. Defense-in-depth — the routed handler is the
1053 guarded DocumentDeletionService, but this is a public method.
1054 """
1055 from ...database.models.library import SourceType
1057 with get_user_db_session(self.username) as session:
1058 doc = session.get(Document, document_id)
1059 if not doc:
1060 return False
1062 note_source = (
1063 session.query(SourceType).filter_by(name="note").first()
1064 )
1065 if note_source and doc.source_type_id == note_source.id:
1066 logger.warning(
1067 f"Refused to delete note {document_id[:8]}... via the "
1068 "library document API; use DELETE /api/notes/<id>."
1069 )
1070 return False
1072 # Get file path from tracker (only if document has original_url)
1073 tracker = None
1074 if doc.original_url:
1075 tracker = (
1076 session.query(DownloadTracker)
1077 .filter_by(url_hash=self._get_url_hash(doc.original_url))
1078 .first()
1079 )
1081 # Delete physical file. Own-root only: never resolve via the
1082 # legacy shared root, where a colliding relative path can be
1083 # another tenant's file (#5521), consistent with the delete
1084 # sites in document_deletion.py / PDFStorageManager.delete_pdf.
1085 if tracker and tracker.file_path:
1086 try:
1087 file_path = get_absolute_path_from_settings(
1088 tracker.file_path,
1089 self.username,
1090 allow_legacy_fallback=False,
1091 )
1092 if file_path and file_path.is_file():
1093 file_path.unlink()
1094 logger.info(f"Deleted file: {file_path}")
1095 except Exception:
1096 logger.exception("Failed to delete file")
1098 # Update tracker
1099 if tracker:
1100 tracker.is_downloaded = False
1101 tracker.file_path = None
1103 # Delete document and all related records
1104 from ..deletion.utils.cascade_helper import CascadeHelper
1106 CascadeHelper.delete_document_completely(session, document_id)
1107 session.commit()
1109 # Post-commit: purge the removed document's FAISS vectors + chunk rows
1110 # (delete_document_completely leaves DocumentChunk rows, so without this
1111 # the deleted content keeps surfacing in collection search). Best-effort;
1112 # the DB delete already committed. Mirrors sync_library_with_filesystem.
1113 try:
1114 from ..deletion.services.document_deletion import (
1115 DocumentDeletionService,
1116 )
1117 from ...database.models.library import DocumentChunk
1119 with get_user_db_session(self.username) as purge_session:
1120 coll_rows = (
1121 purge_session.query(DocumentChunk.collection_name)
1122 .filter_by(source_type="document", source_id=document_id)
1123 .distinct()
1124 .all()
1125 )
1126 collection_ids = [
1127 row[0].removeprefix("collection_")
1128 for row in coll_rows
1129 if row[0]
1130 ]
1131 DocumentDeletionService(self.username)._purge_document_rag(
1132 document_id, collection_ids, full_delete=True
1133 )
1134 except Exception:
1135 logger.opt(exception=True).error(
1136 "Failed to purge removed document {} (delete already committed)",
1137 document_id,
1138 )
1139 return True
1141 def open_file_location(self, document_id: str) -> bool:
1142 """Open the folder containing the document."""
1143 with get_user_db_session(self.username) as session:
1144 doc = session.get(Document, document_id)
1145 if not doc:
1146 return False
1148 tracker = None
1149 if doc.original_url:
1150 tracker = (
1151 session.query(DownloadTracker)
1152 .filter_by(url_hash=self._get_url_hash(doc.original_url))
1153 .first()
1154 )
1156 if tracker and tracker.file_path:
1157 # Resolve within the per-user root (traversal + symlink safe),
1158 # with legacy-shared fallback for pre-isolation files (#5521).
1159 validated_path = get_absolute_path_from_settings(
1160 tracker.file_path, self.username
1161 )
1162 if validated_path and validated_path.is_file():
1163 return open_file_location(str(validated_path))
1164 logger.warning("Path validation failed or file missing")
1165 return False
1167 return False
1169 def get_unique_domains(self) -> List[str]:
1170 """Get sorted list of unique netlocs from all document URLs.
1172 Streams the URL column in batches (``yield_per``) and accumulates
1173 netlocs into a set, so a very large library is never fully
1174 materialized in memory at once (#4560). The query already projects to
1175 the single ``original_url`` column.
1176 """
1177 with get_user_db_session(self.username) as session:
1178 netlocs = set()
1179 rows = (
1180 session.query(Document.original_url)
1181 .filter(Document.original_url.isnot(None))
1182 .yield_per(_DOMAIN_SCAN_BATCH_SIZE)
1183 )
1184 for (url,) in rows:
1185 domain = self._extract_domain(url)
1186 if domain:
1187 netlocs.add(domain)
1188 return sorted(netlocs)
1190 def _extract_domain(self, url: str) -> str:
1191 """Extract domain from URL."""
1192 from urllib.parse import urlparse
1194 try:
1195 return urlparse(url).netloc
1196 except (ValueError, AttributeError):
1197 return ""
1199 def _get_url_hash(self, url: str) -> str:
1200 """Generate hash for URL."""
1201 import re
1203 # Normalize URL
1204 url = re.sub(r"^https?://", "", url)
1205 url = re.sub(r"^www\.", "", url)
1206 url = url.rstrip("/")
1208 return get_url_hash(url)
1210 def _get_storage_path(self) -> str:
1211 """Get library storage path from settings (respects LDR_DATA_DIR)."""
1212 settings = get_settings_manager()
1213 return str(
1214 Path(
1215 settings.get_setting(
1216 "research_library.storage_path",
1217 str(get_library_directory()),
1218 )
1219 )
1220 .expanduser()
1221 .resolve()
1222 )
1224 def sync_library_with_filesystem(self) -> Dict:
1225 """
1226 Sync library database with filesystem.
1227 Check which PDF files exist and update database accordingly.
1229 Returns:
1230 Statistics about the sync operation
1231 """
1232 with get_user_db_session(self.username) as session:
1233 # Sync only research downloads — uploads have no original_url
1234 # and no DownloadTracker, so they don't belong in this routine
1235 # (and the destructive `else` branch below would otherwise
1236 # silently delete every uploaded document).
1237 # Don't load the large text_content body: this sync loop only
1238 # reads original_url/id/title (and may delete the row), so loading
1239 # every completed document's full text would needlessly exhaust
1240 # memory on a large library (#4560).
1241 documents = (
1242 session.query(Document)
1243 .filter_by(status=DocumentStatus.COMPLETED)
1244 .filter(Document.original_url.isnot(None))
1245 .options(defer(Document.text_content))
1246 .all()
1247 )
1249 stats = {
1250 "total_documents": len(documents),
1251 "files_found": 0,
1252 "files_missing": 0,
1253 "trackers_updated": 0,
1254 "missing_files": [],
1255 }
1256 # Documents deleted below — their FAISS vectors + chunk rows are
1257 # purged AFTER this transaction commits (see the post-commit block).
1258 removed_doc_ids: List[str] = []
1260 # Sync documents with filesystem.
1261 #
1262 # Resolve one settings manager for the whole sync (lazily, on the
1263 # first document that actually needs a path resolved) instead of
1264 # letting get_absolute_path_from_settings() fall back to the
1265 # ambient get_settings_manager() per document (this is a sync
1266 # `def`, so a large library means one iteration per completed
1267 # document, unbounded). Each ambient call opens a DB session that
1268 # is now tracked in a per-thread owner registry
1269 # (utilities/db_utils.py) until this request's cleanup boundary,
1270 # so enough process-wide LRU evictions between iterations can pin
1271 # one pooled connection per evicted-and-recreated session for the
1272 # rest of the request (see changelog/#6095 follow-up). Creating
1273 # it once here bounds that to at most one session, mirroring
1274 # DownloadService's self.settings pattern.
1275 settings_manager = None
1276 try:
1277 for doc in documents:
1278 # Get download tracker
1279 tracker = (
1280 session.query(DownloadTracker)
1281 .filter_by(
1282 url_hash=self._get_url_hash(doc.original_url)
1283 )
1284 .first()
1285 )
1287 if tracker and tracker.file_path:
1288 # Check if file exists (per-user, legacy-shared fallback)
1289 if settings_manager is None:
1290 settings_manager = get_settings_manager(
1291 username=self.username
1292 )
1293 file_path = get_absolute_path_from_settings(
1294 tracker.file_path,
1295 self.username,
1296 settings_manager=settings_manager,
1297 )
1298 if file_path and file_path.is_file():
1299 stats["files_found"] += 1
1300 else:
1301 # File missing or path invalid - mark for re-download
1302 stats["files_missing"] += 1
1303 stats["missing_files"].append(
1304 {
1305 "id": doc.id,
1306 "title": doc.title,
1307 "path": str(file_path)
1308 if file_path
1309 else "invalid",
1310 "url": doc.original_url,
1311 }
1312 )
1314 # Reset tracker
1315 tracker.is_downloaded = False
1316 tracker.file_path = None
1318 # Delete the document entry so it can be re-queued
1319 from ..deletion.utils.cascade_helper import (
1320 CascadeHelper,
1321 )
1323 CascadeHelper.delete_document_completely(
1324 session, doc.id
1325 )
1326 removed_doc_ids.append(doc.id)
1327 stats["trackers_updated"] += 1
1328 else:
1329 # No tracker or path - delete the document entry
1330 stats["files_missing"] += 1
1331 from ..deletion.utils.cascade_helper import (
1332 CascadeHelper,
1333 )
1335 CascadeHelper.delete_document_completely(
1336 session, doc.id
1337 )
1338 removed_doc_ids.append(doc.id)
1339 finally:
1340 safe_close(
1341 settings_manager, "settings manager", allow_none=True
1342 )
1344 session.commit()
1345 logger.info(
1346 f"Library sync completed: {stats['files_found']} found, {stats['files_missing']} missing"
1347 )
1349 # Post-commit: purge the removed documents' FAISS vectors + chunk rows.
1350 # delete_document_completely leaves DocumentChunk rows (no FK), so
1351 # without this the removed doc's vectors and chunk_text linger and keep
1352 # surfacing in collection search. Reuse DocumentDeletionService's
1353 # post-commit purge (per-collection vector removal + guaranteed chunk-row
1354 # backstop), run after the session above has closed.
1355 if removed_doc_ids:
1356 from ..deletion.services.document_deletion import (
1357 DocumentDeletionService,
1358 )
1359 from ...database.models.library import DocumentChunk
1361 deleter = DocumentDeletionService(self.username)
1362 for doc_id in removed_doc_ids:
1363 # Best-effort per document: the sync's DB deletes already
1364 # committed, so a purge hiccup here must not raise out of the
1365 # sync (which would report the whole operation as failed) nor
1366 # abort the remaining documents' purges.
1367 try:
1368 with get_user_db_session(self.username) as purge_session:
1369 coll_rows = (
1370 purge_session.query(DocumentChunk.collection_name)
1371 .filter_by(source_type="document", source_id=doc_id)
1372 .distinct()
1373 .all()
1374 )
1375 collection_ids = [
1376 row[0].removeprefix("collection_")
1377 for row in coll_rows
1378 if row[0]
1379 ]
1380 deleter._purge_document_rag(
1381 doc_id, collection_ids, full_delete=True
1382 )
1383 except Exception:
1384 logger.opt(exception=True).error(
1385 "Failed to purge removed document {} after library sync "
1386 "(delete already committed)",
1387 doc_id,
1388 )
1390 return stats
1392 def mark_for_redownload(self, document_ids: List[str]) -> int:
1393 """
1394 Mark specific documents for re-download.
1396 Args:
1397 document_ids: List of document IDs to mark for re-download
1399 Returns:
1400 Number of documents marked
1401 """
1402 with get_user_db_session(self.username) as session:
1403 count = 0
1404 for doc_id in document_ids:
1405 doc = session.get(Document, doc_id)
1406 if doc:
1407 if not doc.original_url:
1408 # Uploads have no source URL and no DownloadTracker,
1409 # so re-download is not meaningful for them.
1410 logger.warning(
1411 f"Skipping mark-for-redownload on {doc_id[:8]}…: "
1412 "document has no original_url (likely user upload)"
1413 )
1414 continue
1415 # Get tracker and reset it
1416 tracker = (
1417 session.query(DownloadTracker)
1418 .filter_by(
1419 url_hash=self._get_url_hash(doc.original_url)
1420 )
1421 .first()
1422 )
1424 if tracker: 1424 ↛ 1429line 1424 didn't jump to line 1429 because the condition on line 1424 was always true
1425 tracker.is_downloaded = False
1426 tracker.file_path = None
1428 # Mark document as pending
1429 doc.status = DocumentStatus.PENDING
1430 count += 1
1432 session.commit()
1433 logger.info(f"Marked {count} documents for re-download")
1434 return count