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