Coverage for src/local_deep_research/research_library/services/library_service.py: 93%
475 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +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 and_, or_, func, case, select
17from sqlalchemy.orm import aliased, defer
19from ...constants import FILE_PATH_SENTINELS, FILE_PATH_TEXT_ONLY
20from ...utilities.arxiv import is_arxiv_paper_url
21from ...database.models.download_tracker import DownloadTracker
22from ...database.models.library import (
23 Collection,
24 Document,
25 DocumentBlob,
26 DocumentCollection,
27 DocumentStatus,
28)
29from ...database.models.metrics import ResearchRating
30from ...database.models.research import ResearchHistory, ResearchResource
31from ...database.session_context import get_user_db_session
32from ...config.paths import get_library_directory
33from ...utilities.db_utils import get_settings_manager
34from ...utilities.formatting import short_id
35from ...utilities.resource_utils import safe_close
36from ...security import url_authority_without_userinfo
37from ...utilities.sql_utils import escape_like as _escape_like
38from ..utils import (
39 document_awaits_pdf_clause,
40 document_counts_as_downloaded,
41 find_tracker_by_url,
42 find_tracker_for_document,
43 get_absolute_path_from_settings,
44 open_file_location,
45 resource_document_clause,
46)
48# Filter dropdowns (library page + ``/api/research-list``) load research
49# sessions into a client-side ``<select>``. Bound the query so a very large
50# history cannot pull every row into memory / the DOM on each page load
51# (#4560). The query already projects to three small columns, so this is a
52# safety cap rather than a crash fix; realistic histories are far smaller, and
53# comprehensive large-scale filtering is a server-side-search follow-up.
54_DROPDOWN_RESEARCH_LIMIT = 1000
56# ``get_unique_domains`` scans one URL row per downloaded document to build the
57# domain filter. Stream it in batches (``yield_per``) rather than materializing
58# every row at once, so a very large library cannot exhaust memory (#4560). The
59# distinct-domain set it accumulates is small regardless of library size.
60_DOMAIN_SCAN_BATCH_SIZE = 1000
63def _names_a_file(file_path: Optional[str]) -> bool:
64 """Whether a ``file_path`` value names a file on disk.
66 ``None``, ``"database"`` and the sentinels (``blob_deleted``,
67 ``text_only_not_stored``, ``metadata_only``) name no file.
68 """
69 return (
70 bool(file_path)
71 and file_path != "database"
72 and file_path not in FILE_PATH_SENTINELS
73 )
76def _tracker_describes_storage(
77 tracker_path: Optional[str], doc: Document
78) -> bool:
79 """Whether an identifying tracker's path names where ``doc`` stores its PDF.
81 The sync asks this only for a tracker whose content hash or explicit
82 document link identifies ``doc`` (``find_tracker_for_document``):
83 URL-equivalent addresses can hold different PDF versions, so a URL
84 match alone does not say whose file a tracker names. A download that
85 stores a PDF writes to the tracker where it stored it:
87 * filesystem storage: the file's path, which
88 ``PDFStorageManager.save_pdf`` also writes to the document, so the
89 two are equal (rows written before database storage was added,
90 8d7bb38d1a, copied the tracker's path into the document);
91 * database storage: ``"database"``, while ``save_pdf`` labels the
92 document ``storage_mode`` "database" and leaves its ``file_path``
93 None, as it has since database storage was added (8d7bb38d1a).
94 ``upgrade_to_pdf`` labels a document it stores a PDF in the same
95 way, and the content-hash dedup of ``_download_pdf`` writes
96 ``"database"`` for a matched document with that label too.
98 So the tracker describes a document when it names the document's own
99 path. An identifying tracker can still carry a stale path, and the
100 document's own path wins over it. A document without a path of its
101 own, a database-stored one or a legacy row that kept its path only in
102 the tracker, takes the path of the tracker that identifies it. Neither
103 ``"database"`` nor the document's label says that a blob was stored (a
104 failed-download placeholder keeps the column's ``"database"``
105 default), so the sync still checks the document's own blob.
106 """
107 if not doc.file_path:
108 return True
109 return tracker_path == doc.file_path
112def _missing_file_entry(
113 doc: Document, stored_path: Optional[str], resolved_path
114) -> Dict:
115 """The sync's ``missing_files`` entry for a document whose file is gone.
117 ``POST /library/api/sync-library`` returns these entries to the client.
118 ``path`` is the stored path, which is relative to the user's library
119 root (the library listing returns the same value as ``file_path``), or
120 ``"invalid"`` when it does not resolve inside that root
121 (``resolved_path`` is ``None``). For a document whose PDF its
122 identifying tracker records as stored in the database, or that is
123 itself labelled as stored there, the stored value is ``"database"``: a
124 fixed label that names no file, returned whatever ``resolved_path`` is.
125 It is never ``resolved_path`` itself: the absolute path is the server's
126 directory layout, which ``/library/api/check-downloads`` and the
127 library listing withhold from the client (#6462).
128 """
129 return {
130 "id": doc.id,
131 "title": doc.title,
132 "path": stored_path
133 if stored_path == "database" or (resolved_path and stored_path)
134 else "invalid",
135 "url": doc.original_url,
136 }
139def _counts_as_downloaded():
140 """SQL condition: a ``Document`` the Download Manager counts as downloaded.
142 A completed document, unless it awaits its PDF (see
143 ``document_awaits_pdf``): "Sync library" kept its text after its PDF
144 file went missing and reported the file missing, and "Download All New"
145 queues it, so the Download Manager counts it as available to download.
146 Without a matching document (the outer join's NULL row) the condition is
147 not true, so nothing is counted. ``document_counts_as_downloaded``
148 (``research_library.utils``) is the same rule for a loaded
149 ``Document``.
150 """
151 return and_(
152 Document.status == DocumentStatus.COMPLETED,
153 ~document_awaits_pdf_clause(),
154 )
157def _sql_falsy(column):
158 """SQL for Python's ``not value`` on a nullable text column: NULL or ''."""
159 return or_(column.is_(None), column == "")
162def _fallback_resource_id(document_model):
163 """Select one stable resource when no primary resource exists."""
164 return (
165 select(func.min(ResearchResource.id))
166 .where(ResearchResource.document_id == document_model.id)
167 .correlate(document_model)
168 .scalar_subquery()
169 )
172def _display_resource_id(document_model):
173 """Match the listing's primary-FK preference and stable fallback."""
174 primary_id = (
175 select(ResearchResource.id)
176 .where(ResearchResource.id == document_model.resource_id)
177 .correlate(document_model)
178 .scalar_subquery()
179 )
180 return func.coalesce(primary_id, _fallback_resource_id(document_model))
183class LibraryService:
184 """Service for managing and querying the document library."""
186 def __init__(self, username: str):
187 """Initialize library service for a user."""
188 self.username = username
190 def _has_blob_in_db(self, session, document_id: str) -> bool:
191 """Check if a PDF blob exists in the database for a document."""
192 return (
193 session.query(DocumentBlob.document_id)
194 .filter_by(document_id=document_id)
195 .first()
196 is not None
197 )
199 def _remaining_content(
200 self, session, doc: Document, settings_manager
201 ) -> Optional[str]:
202 """What a download stores itself, whatever its URL's tracker names.
204 Returns ``"pdf"`` when its PDF is still stored (a blob in the
205 database, or a file at the document's own ``file_path``),
206 ``"text"`` when only its extracted text is left, and ``None`` when
207 nothing is.
209 The document's ``storage_mode`` and its tracker's ``file_path`` can
210 both be stale: a re-download in "none" mode clears the tracker path
211 but keeps the stored PDF, and a failed-download placeholder keeps
212 the ``"database"`` column default after text is saved into it. The
213 tracker, which every document with the URL shares, can also name
214 another document's file. So this checks what is actually stored.
215 Deleting a document that holds any of it would also delete that
216 content, its chunks and its vectors.
217 """
218 if self._has_blob_in_db(session, doc.id):
219 return "pdf"
220 if _names_a_file(doc.file_path):
221 resolved = get_absolute_path_from_settings(
222 doc.file_path, self.username, settings_manager=settings_manager
223 )
224 if resolved and resolved.is_file():
225 return "pdf"
226 has_text = (
227 session.query(Document.id)
228 .filter(
229 Document.id == doc.id,
230 Document.text_content.isnot(None),
231 Document.text_content != "",
232 )
233 .first()
234 is not None
235 )
236 return "text" if has_text else None
238 def _keep_text_report_missing_pdf(
239 self,
240 doc: Document,
241 tracker,
242 stored_path: Optional[str],
243 resolved_path,
244 stats: Dict,
245 ) -> None:
246 """Keep a document's text, drop its dangling PDF and report it.
248 For a research download whose PDF file is gone but whose extracted
249 text is not, and that still has its research resource
250 (``resource_id``), from which the download manager can download
251 the PDF again. The document keeps its row, text, collection links,
252 chunks and vectors. It is labelled a PDF and stored the way
253 text-only documents are (``file_type`` "pdf", ``storage_mode``
254 "none", ``file_path`` ``FILE_PATH_TEXT_ONLY``), so the library no
255 longer offers a PDF that is not there, and ``tracker`` is reset
256 to not downloaded unless it is None (see ``_clear_missing_pdf``).
257 It is counted in ``files_missing`` and listed in ``missing_files``
258 under ``stored_path``, the path that named the missing file: the
259 document's own path or, for a document without one, the path of the
260 tracker that identifies it (see ``_missing_file_entry``), and
261 ``document_awaits_pdf`` holds for it, so the download manager
262 downloads its PDF into it again.
264 The label is set here because what went missing was a PDF: the
265 sync calls this only for a missing file that the document's own
266 ``file_path`` names, or that the identifying tracker of a document
267 without a path of its own names, and the file paths of documents and
268 trackers come only from PDF storage. A
269 failed-download placeholder that a PDF download completed before
270 ``_download_pdf`` labelled such documents "pdf" is still labelled
271 "unknown", and would otherwise never be downloaded again.
272 """
273 doc.file_type = "pdf"
274 doc.mime_type = "application/pdf"
275 self._clear_missing_pdf(doc, tracker, stats)
276 stats["files_missing"] += 1
277 stats["missing_files"].append(
278 _missing_file_entry(doc, stored_path, resolved_path)
279 )
281 def _clear_missing_pdf(self, doc: Document, tracker, stats: Dict) -> None:
282 """Make a document whose PDF file is gone a text-only document.
284 It is stored the way text-only documents are (``storage_mode``
285 "none", ``file_path`` ``FILE_PATH_TEXT_ONLY``), and ``tracker``,
286 unless it is None, is reset to not downloaded, so neither names the
287 missing file any more. The sync passes the tracker whose content
288 hash or explicit document link identifies this document (see
289 ``find_tracker_for_document``) when it names this document's path,
290 supplies the path of a document without one of its own, or has no
291 path. It passes None when no tracker identifies the document, or
292 when the identifying tracker names another path than the
293 document's own (see ``_tracker_describes_storage``); that tracker is
294 left as it is. The reset counts in ``trackers_updated`` only when it
295 changes the tracker, because a tracker without a path can be reset
296 already. A tracker identifies at most one document, so the sync
297 changes it at most once.
299 The sync calls this directly, without reporting the document, for
300 a download that keeps its extracted text but no longer has its
301 research resource (``resource_id`` is NULL, e.g. after
302 ``delete_resource``). No download path asks for such a document's
303 PDF (``document_awaits_pdf`` needs the resource), and it counts in
304 ``files_found``, as every text-only document does. A later download
305 of the same PDF bytes for another resource still reaches it through
306 the content-hash match in ``DownloadService._download_pdf``. In a
307 storage mode that keeps PDFs, that match stores the PDF into a
308 matched document stored without one (``storage_mode`` "none").
309 Left naming the missing file, the document made that match copy the
310 dangling path into the new resource's tracker and store nothing.
311 """
312 doc.storage_mode = "none"
313 doc.file_path = FILE_PATH_TEXT_ONLY
314 if tracker is not None and (
315 tracker.is_downloaded or tracker.file_path is not None
316 ):
317 tracker.is_downloaded = False
318 tracker.file_path = None
319 stats["trackers_updated"] += 1
321 def _get_safe_absolute_path(self, file_path: str) -> Optional[str]:
322 """
323 Get the absolute path for a file, safely handling invalid paths.
325 Args:
326 file_path: Relative file path from library root
328 Returns:
329 Absolute path as string, or None if path is invalid/unsafe
330 """
331 if not file_path or file_path in FILE_PATH_SENTINELS:
332 return None
333 abs_path = get_absolute_path_from_settings(file_path, self.username)
334 return str(abs_path) if abs_path else None
336 def _is_arxiv_url(self, url: str) -> bool:
337 """Whether the URL names one specific arXiv paper.
339 Paper identity, not host membership (#5617): every routing decision,
340 the downloader's ``can_handle`` and the citation source label follow
341 :func:`extract_arxiv_id`. A listing such as
342 ``https://arxiv.org/list/cs.AI/recent`` is on the host but carries no
343 identifier, so reporting ``is_arxiv: true`` for it told the UI
344 something the pipeline disagreed with (#6413).
345 """
346 return is_arxiv_paper_url(url)
348 def _is_pubmed_url(self, url: str) -> bool:
349 """Check if URL is from PubMed or NCBI domains."""
350 try:
351 parsed = urlparse(url)
352 hostname = parsed.hostname
353 if not hostname:
354 return False
356 # Check for pubmed.ncbi.nlm.nih.gov
357 if hostname == "pubmed.ncbi.nlm.nih.gov":
358 return True
360 # Check for ncbi.nlm.nih.gov with PMC path
361 if hostname == "ncbi.nlm.nih.gov" and "/pmc" in parsed.path:
362 return True
364 # Check for pubmed in subdomain
365 if "pubmed" in hostname:
366 return True
368 return False
369 except Exception:
370 return False
372 def _apply_date_filter(self, query, model_class, date_filter: str):
373 """Apply date range filter based on processed_at timestamp."""
374 from datetime import datetime, timedelta, timezone
376 now = datetime.now(timezone.utc)
377 if date_filter == "today":
378 cutoff = now.replace(hour=0, minute=0, second=0, microsecond=0)
379 elif date_filter == "week":
380 cutoff = now - timedelta(days=7)
381 elif date_filter == "month":
382 cutoff = now - timedelta(days=30)
383 else:
384 return query
385 return query.filter(model_class.processed_at >= cutoff)
387 def _apply_domain_filter(self, query, model_class, domain: str):
388 """Apply domain filter to query for Document.
390 The dropdown is fully data-driven (populated from get_unique_domains),
391 so the filter is a generic substring match against original_url.
392 """
393 pattern = f"%{_escape_like(domain)}%"
394 return query.filter(model_class.original_url.like(pattern, escape="\\"))
396 def _apply_search_filter(self, query, model_class, search_query: str):
397 """Apply the search filter to a Document query.
399 Matches the document's title, authors and DOI, the title of a linked
400 research resource, and the stored filename where that filename is
401 the name the document is listed under. ``ilike`` folds case for
402 ASCII letters only: on SQLite it is ``lower(...) LIKE lower(...)``.
404 ``get_documents`` lists a document as ``document_title``, which is
405 ``doc.title or (resource.title if resource else doc.filename)``, and
406 the Library card falls back to ``file_name`` when that is empty;
407 ``file_name`` is the filename unless the document has a real file
408 path. So the filename is the listed name only when the document has
409 no title and either no linked resource, or a linked resource with no
410 title and no real file path. Any other stored filename is not
411 searched: a titled document (every Zotero item) is listed under its
412 title, and a document listed under a resource title is found by that
413 title.
415 The caller must outer-join ``ResearchResource`` (``get_documents``
416 does).
417 """
418 search_pattern = f"%{_escape_like(search_query)}%"
419 listed_under_filename = and_(
420 _sql_falsy(model_class.title),
421 or_(
422 ResearchResource.id.is_(None),
423 and_(
424 ResearchResource.id == _display_resource_id(model_class),
425 _sql_falsy(ResearchResource.title),
426 or_(
427 _sql_falsy(model_class.file_path),
428 model_class.file_path.in_(FILE_PATH_SENTINELS),
429 ),
430 ),
431 ),
432 )
433 return query.filter(
434 or_(
435 model_class.title.ilike(search_pattern, escape="\\"),
436 and_(
437 listed_under_filename,
438 model_class.filename.ilike(search_pattern, escape="\\"),
439 ),
440 model_class.authors.ilike(search_pattern, escape="\\"),
441 model_class.doi.ilike(search_pattern, escape="\\"),
442 ResearchResource.title.ilike(search_pattern, escape="\\"),
443 )
444 )
446 def get_library_stats(self) -> Dict:
447 """Get overall library statistics."""
448 with get_user_db_session(self.username) as session:
449 # Get document counts
450 total_docs = session.query(Document).count()
451 total_pdfs = (
452 session.query(Document).filter_by(file_type="pdf").count()
453 )
455 # Get size stats
456 size_result = session.query(
457 func.sum(Document.file_size),
458 func.avg(Document.file_size),
459 ).first()
461 total_size = size_result[0] or 0
462 avg_size = size_result[1] or 0
464 # Get research stats
465 research_count = session.query(
466 func.count(func.distinct(Document.research_id))
467 ).scalar()
469 # Get domain stats - count unique domains from URLs
470 # Extract domain from original_url using SQL functions
471 from sqlalchemy import case, func as sql_func
473 # Count unique domains by extracting them from URLs
474 domain_subquery = session.query(
475 sql_func.distinct(
476 case(
477 (
478 Document.original_url.like("%arxiv.org%"),
479 "arxiv.org",
480 ),
481 (
482 Document.original_url.like("%pubmed%"),
483 "pubmed",
484 ),
485 (
486 Document.original_url.like("%ncbi.nlm.nih.gov%"),
487 "pubmed",
488 ),
489 else_="other",
490 )
491 )
492 ).subquery()
494 domain_count = (
495 session.query(sql_func.count())
496 .select_from(domain_subquery)
497 .scalar()
498 )
500 # Get download tracker stats
501 pending_downloads = (
502 session.query(DownloadTracker)
503 .filter_by(is_downloaded=False)
504 .count()
505 )
507 return {
508 "total_documents": total_docs,
509 "total_pdfs": total_pdfs,
510 "total_size_bytes": total_size,
511 "total_size_mb": total_size / (1024 * 1024)
512 if total_size
513 else 0,
514 "average_size_mb": avg_size / (1024 * 1024) if avg_size else 0,
515 "research_sessions": research_count,
516 "unique_domains": domain_count,
517 "pending_downloads": pending_downloads,
518 "storage_path": self._get_storage_path(),
519 }
521 def count_documents(
522 self,
523 research_id: Optional[str] = None,
524 domain: Optional[str] = None,
525 collection_id: Optional[str] = None,
526 date_filter: Optional[str] = None,
527 ) -> int:
528 """Count documents matching the given filters (for pagination)."""
529 with get_user_db_session(self.username) as session:
530 from ...database.library_init import get_default_library_id
532 if not collection_id: 532 ↛ 533line 532 didn't jump to line 533 because the condition on line 532 was never true
533 collection_id = get_default_library_id(self.username)
535 q = (
536 session.query(func.count(Document.id))
537 .join(
538 DocumentCollection,
539 Document.id == DocumentCollection.document_id,
540 )
541 .filter(DocumentCollection.collection_id == collection_id)
542 .filter(Document.status == "completed")
543 )
545 if research_id: 545 ↛ 546line 545 didn't jump to line 546 because the condition on line 545 was never true
546 q = q.filter(Document.research_id == research_id)
547 if domain: 547 ↛ 548line 547 didn't jump to line 548 because the condition on line 547 was never true
548 q = self._apply_domain_filter(q, Document, domain)
549 if date_filter: 549 ↛ 550line 549 didn't jump to line 550 because the condition on line 549 was never true
550 q = self._apply_date_filter(q, Document, date_filter)
552 return q.scalar() or 0
554 def get_documents(
555 self,
556 research_id: Optional[str] = None,
557 domain: Optional[str] = None,
558 file_type: Optional[str] = None,
559 favorites_only: bool = False,
560 search_query: Optional[str] = None,
561 collection_id: Optional[str] = None,
562 date_filter: Optional[str] = None,
563 limit: int = 100,
564 offset: int = 0,
565 ) -> List[Dict]:
566 """
567 Get documents with filtering options.
569 Returns enriched document information with research details.
570 """
571 with get_user_db_session(self.username) as session:
572 # Get default Library collection ID if not specified
573 from ...database.library_init import get_default_library_id
575 if not collection_id:
576 collection_id = get_default_library_id(self.username)
578 logger.info(
579 f"[LibraryService] Getting documents for collection_id: {collection_id}, research_id: {research_id}, domain: {domain}"
580 )
582 all_documents = []
584 # Step 1: subquery to get paginated document IDs.
585 # Pagination and sorting happen here on Document alone,
586 # avoiding non-determinism from outer-joining related tables.
587 doc_subq = (
588 session.query(Document.id)
589 .join(
590 DocumentCollection,
591 Document.id == DocumentCollection.document_id,
592 )
593 .filter(DocumentCollection.collection_id == collection_id)
594 )
596 # Apply filters
597 if research_id: 597 ↛ 598line 597 didn't jump to line 598 because the condition on line 597 was never true
598 doc_subq = doc_subq.filter(Document.research_id == research_id)
600 if domain: 600 ↛ 601line 600 didn't jump to line 601 because the condition on line 600 was never true
601 doc_subq = self._apply_domain_filter(doc_subq, Document, domain)
603 if date_filter: 603 ↛ 604line 603 didn't jump to line 604 because the condition on line 603 was never true
604 doc_subq = self._apply_date_filter(
605 doc_subq, Document, date_filter
606 )
608 if file_type: 608 ↛ 609line 608 didn't jump to line 609 because the condition on line 608 was never true
609 doc_subq = doc_subq.filter(Document.file_type == file_type)
611 if favorites_only: 611 ↛ 612line 611 didn't jump to line 612 because the condition on line 611 was never true
612 doc_subq = doc_subq.filter(Document.favorite.is_(True))
614 if search_query:
615 # _apply_search_filter references ResearchResource.title,
616 # so we must outerjoin it here. DISTINCT prevents fan-out
617 # from the outerjoin duplicating Document IDs.
618 doc_subq = doc_subq.outerjoin(
619 ResearchResource,
620 (Document.resource_id == ResearchResource.id)
621 | (ResearchResource.document_id == Document.id),
622 ).distinct()
623 doc_subq = self._apply_search_filter(
624 doc_subq, Document, search_query
625 )
627 # Filter to only completed documents
628 doc_subq = doc_subq.filter(Document.status == "completed")
630 # Sort at SQL level (SQLite-safe NULL handling)
631 doc_subq = doc_subq.order_by(
632 case((Document.processed_at.isnot(None), 0), else_=1),
633 Document.processed_at.desc(),
634 )
636 # Apply SQL-level pagination
637 doc_subq = doc_subq.offset(offset).limit(limit)
638 doc_id_subq = doc_subq.subquery()
640 # Step 2: join the paginated document IDs with related tables.
641 # Use two separate outer joins for ResearchResource to avoid
642 # the OR-condition join that can fan out to multiple rows:
643 # - ResourceByFK: matched via Document.resource_id (primary FK)
644 # - ResourceByDoc: matched via ResearchResource.document_id
645 # Prefer ResourceByFK; otherwise use the earliest back-linked
646 # resource, matching the filename search's display selection.
647 ResourceByFK = aliased(ResearchResource)
648 ResourceByDoc = aliased(ResearchResource)
650 query = (
651 session.query(
652 Document,
653 ResourceByFK,
654 ResourceByDoc,
655 ResearchHistory,
656 DocumentCollection,
657 )
658 .join(doc_id_subq, Document.id == doc_id_subq.c.id)
659 .join(
660 DocumentCollection,
661 Document.id == DocumentCollection.document_id,
662 )
663 .outerjoin(
664 ResourceByFK,
665 Document.resource_id == ResourceByFK.id,
666 )
667 .outerjoin(
668 ResourceByDoc,
669 ResourceByDoc.id == _fallback_resource_id(Document),
670 )
671 .outerjoin(
672 ResearchHistory,
673 Document.research_id == ResearchHistory.id,
674 )
675 .filter(DocumentCollection.collection_id == collection_id)
676 # Re-apply sort so final results are ordered
677 .order_by(
678 case((Document.processed_at.isnot(None), 0), else_=1),
679 Document.processed_at.desc(),
680 )
681 )
683 # Execute query
684 results = query.all()
685 logger.info(
686 f"[LibraryService] Found {len(results)} documents in collection {collection_id}"
687 )
689 # Batch-check blob existence to avoid N+1 queries
690 doc_ids = [row[0].id for row in results]
691 blob_ids = set()
692 if doc_ids:
693 blob_ids = {
694 r[0]
695 for r in session.query(DocumentBlob.document_id)
696 .filter(DocumentBlob.document_id.in_(doc_ids))
697 .all()
698 }
700 # Process results, defensively deduplicating by doc.id.
701 #
702 # Resolve one settings manager for the whole page (lazily, on the
703 # first document that actually needs a path resolved) instead of
704 # letting get_absolute_path_from_settings() fall back to the
705 # ambient get_settings_manager() per iteration (up to `limit`
706 # times). Each ambient call opens a DB session via
707 # get_db_session(), which is now tracked in a per-thread owner
708 # registry (utilities/db_utils.py) until this request's cleanup
709 # boundary — so on a busy multi-user instance, enough process-wide
710 # LRU evictions between iterations can pin one pooled connection
711 # per evicted-and-recreated session for the rest of the request
712 # (see changelog/#6095 follow-up). Creating it once here bounds
713 # that to at most one session for this call, mirroring the
714 # pattern already used in DownloadService (self.settings,
715 # threaded through get_absolute_path_from_settings calls).
716 seen_doc_ids = set()
717 settings_manager = None
718 try:
719 for (
720 doc,
721 res_by_fk,
722 res_by_doc,
723 research,
724 doc_collection,
725 ) in results:
726 if doc.id in seen_doc_ids:
727 continue
728 seen_doc_ids.add(doc.id)
729 # Prefer the resource matched via Document.resource_id FK;
730 # fall back to the one matched via ResearchResource.document_id.
731 resource = res_by_fk or res_by_doc
732 # Determine availability flags - use Document.file_path directly.
733 # Resolved here and NOT returned: the absolute path is the server's
734 # own directory layout, which /library/api/check-downloads already
735 # withholds for the same reason (#6462). Callers fetch the file
736 # through /document/{id}/pdf, so they never need it.
737 file_absolute_path = None
738 if (
739 doc.file_path
740 and doc.file_path not in FILE_PATH_SENTINELS
741 ):
742 if settings_manager is None:
743 settings_manager = get_settings_manager(
744 username=self.username
745 )
746 abs_path = get_absolute_path_from_settings(
747 doc.file_path,
748 self.username,
749 settings_manager=settings_manager,
750 )
751 if abs_path:
752 file_absolute_path = str(abs_path)
754 # Check if PDF is available (filesystem OR database)
755 has_pdf = bool(file_absolute_path)
756 # NULL storage_mode predates the column: a blob then
757 # means database storage, as infer_storage_mode() says.
758 if not has_pdf and doc.storage_mode in (None, "database"):
759 has_pdf = doc.id in blob_ids
760 has_text_db = bool(doc.text_content) # Text now in Document
762 # Use DocumentCollection from query results
763 has_rag_indexed = (
764 doc_collection.indexed if doc_collection else False
765 )
766 rag_chunk_count = (
767 doc_collection.chunk_count if doc_collection else 0
768 )
770 all_documents.append(
771 {
772 "id": doc.id,
773 "resource_id": doc.resource_id,
774 "research_id": doc.research_id,
775 # Document info
776 "document_title": doc.title
777 or (resource.title if resource else doc.filename),
778 "authors": doc.authors,
779 "published_date": doc.published_date,
780 "doi": doc.doi,
781 "arxiv_id": doc.arxiv_id,
782 "pmid": doc.pmid,
783 # File info
784 "file_path": doc.file_path,
785 "file_name": Path(doc.file_path).name
786 if doc.file_path
787 and doc.file_path not in FILE_PATH_SENTINELS
788 else doc.filename,
789 "file_size": doc.file_size,
790 "file_type": doc.file_type,
791 # URLs
792 "original_url": doc.original_url,
793 "domain": self._extract_domain(doc.original_url)
794 if doc.original_url
795 else "User Upload",
796 # Status
797 "download_status": doc.status or "completed",
798 "downloaded_at": doc.processed_at.isoformat()
799 if doc.processed_at
800 else (
801 doc.uploaded_at.isoformat()
802 if hasattr(doc, "uploaded_at")
803 and doc.uploaded_at
804 else None
805 ),
806 "favorite": doc.favorite
807 if hasattr(doc, "favorite")
808 else False,
809 "tags": doc.tags if hasattr(doc, "tags") else [],
810 # Research info (None for user uploads)
811 "research_title": research.title
812 or research.query[:80]
813 if research
814 else "User Upload",
815 "research_query": research.query
816 if research
817 else None,
818 "research_mode": research.mode
819 if research
820 else None,
821 "research_date": research.created_at
822 if research
823 else None,
824 # Classification flags
825 "is_arxiv": self._is_arxiv_url(doc.original_url)
826 if doc.original_url
827 else False,
828 "is_pubmed": self._is_pubmed_url(doc.original_url)
829 if doc.original_url
830 else False,
831 "is_pdf": doc.file_type == "pdf",
832 # Availability flags
833 "has_pdf": has_pdf,
834 "has_text_db": has_text_db,
835 "has_rag_indexed": has_rag_indexed,
836 "rag_chunk_count": rag_chunk_count,
837 }
838 )
840 # Sorting and pagination are now handled at SQL level
841 return all_documents
842 finally:
843 safe_close(
844 settings_manager, "settings manager", allow_none=True
845 )
847 def get_all_collections(self) -> List[Dict]:
848 """Get all collections with document and indexed document counts."""
849 with get_user_db_session(self.username) as session:
850 # Query collections with document counts and indexed counts
851 results = (
852 session.query(
853 Collection,
854 func.count(DocumentCollection.document_id).label(
855 "document_count"
856 ),
857 func.count(
858 case(
859 (
860 DocumentCollection.indexed == True, # noqa: E712
861 DocumentCollection.document_id,
862 ),
863 else_=None,
864 )
865 ).label("indexed_document_count"),
866 )
867 .outerjoin(
868 DocumentCollection,
869 Collection.id == DocumentCollection.collection_id,
870 )
871 .group_by(Collection.id)
872 .order_by(Collection.is_default.desc(), Collection.name)
873 .all()
874 )
876 logger.info(f"[LibraryService] Found {len(results)} collections")
878 collections = []
879 for collection, doc_count, indexed_count in results:
880 logger.debug(
881 f"[LibraryService] Collection: {collection.name} (ID: {collection.id}), documents: {doc_count}, indexed: {indexed_count}"
882 )
883 collections.append(
884 {
885 "id": collection.id,
886 "name": collection.name,
887 "description": collection.description,
888 "is_default": collection.is_default,
889 "document_count": doc_count or 0,
890 "indexed_document_count": indexed_count or 0,
891 }
892 )
894 return collections
896 def get_research_list_for_dropdown(self) -> List[Dict]:
897 """Get minimal research session info for filter dropdowns.
899 Returns only id, title, and query — no joins or aggregates — for at
900 most the ``_DROPDOWN_RESEARCH_LIMIT`` most-recent sessions, so a very
901 large history cannot load every row into memory / the DOM (#4560).
902 """
903 with get_user_db_session(self.username) as session:
904 results = (
905 session.query(
906 ResearchHistory.id,
907 ResearchHistory.title,
908 ResearchHistory.query,
909 )
910 .order_by(ResearchHistory.created_at.desc())
911 .limit(_DROPDOWN_RESEARCH_LIMIT)
912 .all()
913 )
914 return [
915 {"id": r.id, "title": r.title, "query": r.query}
916 for r in results
917 ]
919 def get_research_list_with_stats(
920 self,
921 limit: int = 0,
922 offset: int = 0,
923 ) -> List[Dict]:
924 """Get research sessions with download statistics.
926 ``downloaded_count`` leaves out a document that awaits its PDF (see
927 ``_counts_as_downloaded``).
929 Args:
930 limit: Maximum number of results (0 = no limit, for backwards compat).
931 offset: Number of rows to skip. Only applied when limit > 0.
932 """
933 with get_user_db_session(self.username) as session:
934 # Query research sessions with resource counts
935 query = (
936 session.query(
937 ResearchHistory,
938 func.count(func.distinct(ResearchResource.id)).label(
939 "total_resources"
940 ),
941 func.count(
942 func.distinct(
943 case(
944 (_counts_as_downloaded(), Document.id),
945 else_=None,
946 )
947 )
948 ).label("downloaded_count"),
949 func.count(
950 func.distinct(
951 case(
952 (
953 ResearchResource.url.like("%.pdf")
954 | ResearchResource.url.like("%arxiv.org%")
955 | ResearchResource.url.like(
956 "%ncbi.nlm.nih.gov/pmc%"
957 ),
958 ResearchResource.id,
959 ),
960 else_=None,
961 )
962 )
963 ).label("downloadable_count"),
964 )
965 .outerjoin(
966 ResearchResource,
967 ResearchHistory.id == ResearchResource.research_id,
968 )
969 .outerjoin(
970 Document,
971 resource_document_clause(),
972 )
973 .group_by(ResearchHistory.id)
974 .order_by(ResearchHistory.created_at.desc())
975 )
977 # Apply SQL-level pagination when limit is set
978 if limit > 0:
979 query = query.offset(offset).limit(limit)
981 results = query.all()
983 # Preload all ratings to avoid N+1 queries
984 research_ids = [r[0].id for r in results]
985 all_ratings = (
986 session.query(ResearchRating)
987 .filter(ResearchRating.research_id.in_(research_ids))
988 .all()
989 if research_ids
990 else []
991 )
992 ratings_by_research = {r.research_id: r for r in all_ratings}
994 # Batch domain queries to avoid N+1 (same pattern as ratings)
995 domain_case = case(
996 (
997 ResearchResource.url.like("%arxiv.org%"),
998 "arxiv.org",
999 ),
1000 (ResearchResource.url.like("%pubmed%"), "pubmed"),
1001 (
1002 ResearchResource.url.like("%ncbi.nlm.nih.gov%"),
1003 "pubmed",
1004 ),
1005 else_="other",
1006 )
1007 all_domains = (
1008 session.query(
1009 ResearchResource.research_id,
1010 domain_case.label("domain"),
1011 func.count().label("count"),
1012 )
1013 .filter(ResearchResource.research_id.in_(research_ids))
1014 .group_by(ResearchResource.research_id, domain_case)
1015 .all()
1016 if research_ids
1017 else []
1018 )
1019 domains_by_research: Dict[str, list] = {}
1020 for rid, domain, count in all_domains:
1021 domains_by_research.setdefault(rid, []).append((domain, count))
1023 research_list = []
1024 for (
1025 research,
1026 total_resources,
1027 downloaded_count,
1028 downloadable_count,
1029 ) in results:
1030 # Get rating from preloaded dict
1031 rating = ratings_by_research.get(research.id)
1033 # Get domain breakdown from preloaded dict
1034 domains = domains_by_research.get(research.id, [])
1036 research_list.append(
1037 {
1038 "id": research.id,
1039 "title": research.title,
1040 "query": research.query,
1041 "mode": research.mode,
1042 "status": research.status,
1043 "created_at": research.created_at,
1044 "duration_seconds": research.duration_seconds,
1045 "total_resources": total_resources or 0,
1046 "downloaded_count": downloaded_count or 0,
1047 "downloadable_count": downloadable_count or 0,
1048 "rating": rating.rating if rating else None,
1049 "top_domains": [(d, c) for d, c in domains if d],
1050 }
1051 )
1053 return research_list
1055 def get_download_manager_summary_stats(self) -> Dict:
1056 """Get aggregate download stats across ALL research sessions.
1058 This is a lightweight query that only returns totals — used for
1059 the download-manager header so stats remain accurate regardless
1060 of which page the user is viewing. A document that awaits its PDF
1061 does not count as downloaded (see ``_counts_as_downloaded``): it is
1062 left out of ``already_downloaded``, so it is not subtracted from
1063 ``available_to_download``.
1064 """
1065 with get_user_db_session(self.username) as session:
1066 row = (
1067 session.query(
1068 func.count(func.distinct(ResearchHistory.id)).label(
1069 "total_researches"
1070 ),
1071 func.count(func.distinct(ResearchResource.id)).label(
1072 "total_resources"
1073 ),
1074 func.count(
1075 func.distinct(
1076 case(
1077 (_counts_as_downloaded(), Document.id),
1078 else_=None,
1079 )
1080 )
1081 ).label("downloaded_count"),
1082 func.count(
1083 func.distinct(
1084 case(
1085 (
1086 ResearchResource.url.like("%.pdf")
1087 | ResearchResource.url.like("%arxiv.org%")
1088 | ResearchResource.url.like(
1089 "%ncbi.nlm.nih.gov/pmc%"
1090 ),
1091 ResearchResource.id,
1092 ),
1093 else_=None,
1094 )
1095 )
1096 ).label("downloadable_count"),
1097 )
1098 .select_from(ResearchHistory)
1099 .outerjoin(
1100 ResearchResource,
1101 ResearchHistory.id == ResearchResource.research_id,
1102 )
1103 .outerjoin(
1104 Document,
1105 resource_document_clause(),
1106 )
1107 .one()
1108 )
1110 total_researches = row.total_researches or 0
1111 total_resources = row.total_resources or 0
1112 downloaded = row.downloaded_count or 0
1113 downloadable = row.downloadable_count or 0
1115 return {
1116 "total_researches": total_researches,
1117 "total_resources": total_resources,
1118 "already_downloaded": downloaded,
1119 "available_to_download": max(downloadable - downloaded, 0),
1120 }
1122 def get_pdf_previews_batch(
1123 self, research_ids: List, limit_per_research: int = 10
1124 ) -> Dict[str, Dict]:
1125 """Batch-fetch PDF documents and domain breakdowns for multiple research sessions.
1127 Returns a dict keyed by research_id with:
1128 - "pdf_sources": list of document dicts (capped at limit_per_research),
1129 each with "document_title", "domain", "file_type" and
1130 "downloaded" (whether the Download Manager shows it as
1131 downloaded, with its "Downloaded" badge)
1132 - "domains": dict of domain -> {total, pdfs, downloaded}
1134 "downloaded" follows the rule of the page's other counts (see
1135 ``document_counts_as_downloaded``): a completed document counts,
1136 unless it awaits its PDF.
1137 """
1138 if not research_ids:
1139 return {}
1141 with get_user_db_session(self.username) as session:
1142 results = (
1143 session.query(Document, ResearchResource)
1144 .outerjoin(
1145 ResearchResource,
1146 (Document.resource_id == ResearchResource.id)
1147 | (ResearchResource.document_id == Document.id),
1148 )
1149 .filter(
1150 Document.research_id.in_(research_ids),
1151 Document.file_type == "pdf",
1152 )
1153 .order_by(Document.processed_at.desc())
1154 .limit(limit_per_research * len(research_ids))
1155 .all()
1156 )
1158 previews: Dict[str, Dict] = {}
1159 seen_doc_ids: set = set()
1160 for doc, resource in results:
1161 if doc.id in seen_doc_ids:
1162 continue
1163 seen_doc_ids.add(doc.id)
1165 rid = doc.research_id
1166 if rid not in previews:
1167 previews[rid] = {"pdf_sources": [], "domains": {}}
1169 entry = previews[rid]
1171 # Domain breakdown (within the SQL LIMIT budget); authority
1172 # minus userinfo, which the response would otherwise carry.
1173 # The helper never raises and returns "" when unparseable.
1174 domain = "unknown"
1175 if resource and resource.url:
1176 domain = (
1177 url_authority_without_userinfo(resource.url)
1178 or "unknown"
1179 )
1180 elif doc.original_url: 1180 ↛ 1186line 1180 didn't jump to line 1186 because the condition on line 1180 was always true
1181 domain = (
1182 url_authority_without_userinfo(doc.original_url)
1183 or "unknown"
1184 )
1186 if domain not in entry["domains"]:
1187 entry["domains"][domain] = {
1188 "total": 0,
1189 "pdfs": 0,
1190 "downloaded": 0,
1191 }
1192 entry["domains"][domain]["total"] += 1
1193 if doc.file_type == "pdf": 1193 ↛ 1195line 1193 didn't jump to line 1195 because the condition on line 1193 was always true
1194 entry["domains"][domain]["pdfs"] += 1
1195 downloaded = document_counts_as_downloaded(doc)
1196 if downloaded:
1197 entry["domains"][domain]["downloaded"] += 1
1199 # PDF sources preview (capped)
1200 if len(entry["pdf_sources"]) < limit_per_research:
1201 title = "Untitled"
1202 if resource and resource.title:
1203 title = resource.title
1204 elif doc.filename: 1204 ↛ 1207line 1204 didn't jump to line 1207 because the condition on line 1204 was always true
1205 title = doc.filename
1207 entry["pdf_sources"].append(
1208 {
1209 "document_title": title,
1210 "domain": domain,
1211 "file_type": doc.file_type,
1212 "downloaded": downloaded,
1213 }
1214 )
1216 return previews
1218 def get_document_by_id(self, doc_id: str) -> Optional[Dict]:
1219 """
1220 Get a specific document by its ID.
1222 Returns document information with file path.
1223 """
1224 with get_user_db_session(self.username) as session:
1225 # Find document - use outer joins to support both research downloads and user uploads
1226 result = (
1227 session.query(Document, ResearchResource, ResearchHistory)
1228 .outerjoin(
1229 ResearchResource,
1230 (Document.resource_id == ResearchResource.id)
1231 | (ResearchResource.document_id == Document.id),
1232 )
1233 .outerjoin(
1234 ResearchHistory,
1235 Document.research_id == ResearchHistory.id,
1236 )
1237 .filter(Document.id == doc_id)
1238 .first()
1239 )
1241 if result:
1242 # Found document
1243 doc, resource, research = result
1245 # Get RAG indexing status across all collections
1246 doc_collections = (
1247 session.query(DocumentCollection, Collection)
1248 .join(Collection)
1249 .filter(DocumentCollection.document_id == doc_id)
1250 .all()
1251 )
1253 # Check if indexed in any collection
1254 has_rag_indexed = any(
1255 dc.indexed for dc, coll in doc_collections
1256 )
1257 total_chunks = sum(
1258 dc.chunk_count for dc, coll in doc_collections if dc.indexed
1259 )
1261 # Build collections list
1262 collections_list = [
1263 {
1264 "id": coll.id,
1265 "name": coll.name,
1266 "indexed": dc.indexed,
1267 "chunk_count": dc.chunk_count,
1268 }
1269 for dc, coll in doc_collections
1270 ]
1272 # Calculate word count from text content
1273 word_count = (
1274 len(doc.text_content.split()) if doc.text_content else 0
1275 )
1277 # Check if PDF is available (database OR filesystem)
1278 has_pdf = bool(
1279 doc.file_path and doc.file_path not in FILE_PATH_SENTINELS
1280 )
1281 # NULL storage_mode predates the column: a blob then means
1282 # database storage, as infer_storage_mode() says.
1283 if not has_pdf and doc.storage_mode in (None, "database"):
1284 has_pdf = self._has_blob_in_db(session, doc.id)
1286 return {
1287 "id": doc.id,
1288 "resource_id": doc.resource_id,
1289 "research_id": doc.research_id,
1290 "document_title": doc.title
1291 or (resource.title if resource else doc.filename),
1292 "original_url": doc.original_url
1293 or (resource.url if resource else None),
1294 "file_path": doc.file_path,
1295 "file_name": Path(doc.file_path).name
1296 if doc.file_path
1297 and doc.file_path not in FILE_PATH_SENTINELS
1298 else doc.filename,
1299 "file_size": doc.file_size,
1300 "file_type": doc.file_type,
1301 "mime_type": doc.mime_type,
1302 "domain": self._extract_domain(resource.url)
1303 if resource
1304 else "User Upload",
1305 # The status's string value, which the details page
1306 # compares with 'completed' and prints: a loaded
1307 # DocumentStatus member never equals that string.
1308 "download_status": doc.status.value,
1309 "downloaded_at": doc.processed_at.isoformat()
1310 if doc.processed_at
1311 and hasattr(doc.processed_at, "isoformat")
1312 else str(doc.processed_at)
1313 if doc.processed_at
1314 else (
1315 doc.uploaded_at.isoformat()
1316 if hasattr(doc, "uploaded_at") and doc.uploaded_at
1317 else None
1318 ),
1319 "favorite": doc.favorite
1320 if hasattr(doc, "favorite")
1321 else False,
1322 "tags": doc.tags if hasattr(doc, "tags") else [],
1323 "research_title": research.query[:100]
1324 if research
1325 else "User Upload",
1326 "research_created_at": research.created_at
1327 if research and isinstance(research.created_at, str)
1328 else research.created_at.isoformat()
1329 if research and research.created_at
1330 else None,
1331 # Document fields
1332 "is_pdf": doc.file_type == "pdf",
1333 "has_pdf": has_pdf,
1334 "has_text_db": bool(doc.text_content),
1335 "has_rag_indexed": has_rag_indexed,
1336 "rag_chunk_count": total_chunks,
1337 "word_count": word_count,
1338 "collections": collections_list,
1339 }
1341 # Not found
1342 return None
1344 def toggle_favorite(self, document_id: str) -> bool:
1345 """Toggle favorite status of a document."""
1346 with get_user_db_session(self.username) as session:
1347 doc = session.get(Document, document_id)
1348 if doc:
1349 doc.favorite = not doc.favorite
1350 session.commit()
1351 return doc.favorite
1352 return False
1354 def delete_document(self, document_id: str) -> bool:
1355 """Delete a document from library (file and database entry).
1357 Refuses notes: they have their own deletion API
1358 (DELETE /api/notes/<id>) that performs version/link/synthesis
1359 cleanup. Hard-deleting a note here would bypass that and destroy
1360 the note's history. Defense-in-depth — the routed handler is the
1361 guarded DocumentDeletionService, but this is a public method.
1362 """
1363 from ...database.models.library import SourceType
1365 with get_user_db_session(self.username) as session:
1366 doc = session.get(Document, document_id)
1367 if not doc:
1368 return False
1370 note_source = (
1371 session.query(SourceType).filter_by(name="note").first()
1372 )
1373 if note_source and doc.source_type_id == note_source.id:
1374 logger.warning(
1375 f"Refused to delete note {short_id(document_id)} via the "
1376 "library document API; use DELETE /api/notes/<id>."
1377 )
1378 return False
1380 # A canonical URL can cover several historical PDF versions.
1381 # Use this document's storage; a tracker is only a legacy fallback
1382 # when its content hash or document link proves ownership.
1383 tracker = find_tracker_for_document(session, doc)
1384 stored_path = doc.file_path
1385 if (
1386 not stored_path
1387 and tracker
1388 and doc.storage_mode != "none"
1389 and not self._has_blob_in_db(session, doc.id)
1390 ):
1391 # A blob-backed or explicitly text-only document must not
1392 # borrow a stale physical tracker path. A legacy row with
1393 # neither a path nor a blob can still fall back to the
1394 # verified tracker's path, as the library sync does.
1395 stored_path = tracker.file_path
1397 # Delete physical file. Own-root only: never resolve via the
1398 # legacy shared root, where a colliding relative path can be
1399 # another tenant's file (#5521), consistent with the delete
1400 # sites in document_deletion.py / PDFStorageManager.delete_pdf.
1401 if stored_path and stored_path not in (
1402 *FILE_PATH_SENTINELS,
1403 "database",
1404 ):
1405 try:
1406 file_path = get_absolute_path_from_settings(
1407 stored_path,
1408 self.username,
1409 )
1410 if file_path and file_path.is_file():
1411 file_path.unlink()
1412 logger.info(f"Deleted file: {file_path}")
1413 except Exception:
1414 logger.exception("Failed to delete file")
1416 # Update tracker
1417 if tracker:
1418 tracker.is_downloaded = False
1419 tracker.file_path = None
1421 # Delete document and all related records
1422 from ..deletion.utils.cascade_helper import CascadeHelper
1424 CascadeHelper.delete_document_completely(session, document_id)
1425 session.commit()
1427 # Post-commit: purge the removed document's FAISS vectors + chunk rows
1428 # (delete_document_completely leaves DocumentChunk rows, so without this
1429 # the deleted content keeps surfacing in collection search). Best-effort;
1430 # the DB delete already committed. Mirrors sync_library_with_filesystem.
1431 try:
1432 from ..deletion.services.document_deletion import (
1433 DocumentDeletionService,
1434 )
1435 from ...database.models.library import DocumentChunk
1437 with get_user_db_session(self.username) as purge_session:
1438 coll_rows = (
1439 purge_session.query(DocumentChunk.collection_name)
1440 .filter_by(source_type="document", source_id=document_id)
1441 .distinct()
1442 .all()
1443 )
1444 collection_ids = [
1445 row[0].removeprefix("collection_")
1446 for row in coll_rows
1447 if row[0]
1448 ]
1449 DocumentDeletionService(self.username)._purge_document_rag(
1450 document_id, collection_ids, full_delete=True
1451 )
1452 except Exception:
1453 logger.opt(exception=True).error(
1454 "Failed to purge removed document {} (delete already committed)",
1455 document_id,
1456 )
1457 return True
1459 def open_file_location(self, document_id: str) -> bool:
1460 """Open the folder containing the document."""
1461 with get_user_db_session(self.username) as session:
1462 doc = session.get(Document, document_id)
1463 if not doc:
1464 return False
1466 tracker = None
1467 if doc.original_url:
1468 # B2 compat (#6758): match legacy-hash rows.
1469 tracker = find_tracker_by_url(session, doc.original_url)
1471 if tracker and tracker.file_path:
1472 # Resolve only within the per-user root (traversal + symlink safe).
1473 validated_path = get_absolute_path_from_settings(
1474 tracker.file_path, self.username
1475 )
1476 if validated_path and validated_path.is_file():
1477 return open_file_location(str(validated_path))
1478 logger.warning("Path validation failed or file missing")
1479 return False
1481 return False
1483 def get_unique_domains(self) -> List[str]:
1484 """Get sorted list of unique netlocs from all document URLs.
1486 Streams the URL column in batches (``yield_per``) and accumulates
1487 netlocs into a set, so a very large library is never fully
1488 materialized in memory at once (#4560). The query already projects to
1489 the single ``original_url`` column.
1490 """
1491 with get_user_db_session(self.username) as session:
1492 netlocs = set()
1493 rows = (
1494 session.query(Document.original_url)
1495 .filter(Document.original_url.isnot(None))
1496 .yield_per(_DOMAIN_SCAN_BATCH_SIZE)
1497 )
1498 for (url,) in rows:
1499 domain = self._extract_domain(url)
1500 if domain:
1501 netlocs.add(domain)
1502 return sorted(netlocs)
1504 def _extract_domain(self, url: Optional[str]) -> str:
1505 """Extract domain from URL.
1507 The authority minus userinfo: a stored URL's ``user:pass@`` would
1508 otherwise reach the library page's domain filter and the
1509 document payloads. The domain filter's substring match still
1510 finds a credentialed URL by its host. ``""`` for a ``None``
1511 (``ResearchResource.url`` is nullable), unparseable or
1512 ambiguous-authority URL; never raises.
1513 """
1514 return url_authority_without_userinfo(url)
1516 def _get_storage_path(self) -> str:
1517 """Get library storage path from settings (respects LDR_DATA_DIR)."""
1518 settings = get_settings_manager()
1519 return str(
1520 Path(
1521 settings.get_setting(
1522 "research_library.storage_path",
1523 str(get_library_directory()),
1524 )
1525 )
1526 .expanduser()
1527 .resolve()
1528 )
1530 def sync_library_with_filesystem(self) -> Dict:
1531 """
1532 Sync library database with filesystem.
1533 Check which PDF files exist and update database accordingly.
1535 Every document with the same URL shares that URL's tracker, and
1536 URL-equivalent addresses (``paper.pdf`` and
1537 ``paper.pdf?utm_source=old``) can hold different PDF versions, each
1538 in a document of its own. So a tracker supplies a path for a
1539 document, or is reset for it, only when its content hash or
1540 explicit document link identifies the document, with contradictory
1541 identity fields rejected (see ``find_tracker_for_document``), and
1542 only when it names where the document stores its PDF (see
1543 ``_tracker_describes_storage``): the document's own path, which wins
1544 over any other tracker path, or for a document without a path of
1545 its own, the tracker's ``"database"`` or legacy path. Each document
1546 is judged by what it stores itself, and a tracker that does not
1547 describe the document is left unchanged.
1549 A download whose PDF file is missing is handled by what it still
1550 stores (see ``_remaining_content``):
1552 * its PDF, as a database blob or a file at its own ``file_path``:
1553 kept unchanged;
1554 * only its extracted text, while its own ``file_path`` (or, without
1555 one, its identifying tracker's path) names a file, and it still
1556 has its research resource: kept with its text but made text-only,
1557 its identifying tracker reset if it names that file or no file,
1558 and counted in ``files_missing`` and ``missing_files``, so that
1559 the download manager downloads its PDF again (see
1560 ``_keep_text_report_missing_pdf``);
1561 * only its extracted text, while a file is named as above, with no
1562 research resource (e.g. a download whose resource was deleted):
1563 kept with its text and made text-only, its identifying tracker
1564 reset as above, and counted in ``files_found``. No download path
1565 asks for its PDF, so it is not reported, but a later download of
1566 the same PDF bytes stores the PDF in it (see
1567 ``_clear_missing_pdf``);
1568 * only its extracted text, with no file named (a "database" label
1569 without a blob): kept unchanged;
1570 * nothing: deleted, so that its URL can be downloaded again, and
1571 counted in ``files_missing``. Its identifying tracker is reset if
1572 it names the document's file. If that tracker is ``"database"``,
1573 it is reset only when, once every document has been checked, no
1574 other completed document under that tracker still holds a PDF
1575 blob, whether the tracker identifies that document or not: until
1576 then the URL's PDF is still stored in the database, as the
1577 tracker records.
1579 Left alone before any of these checks: a document whose own
1580 ``file_path`` is a sentinel, such as ``blob_deleted`` after "Remove
1581 PDF to save space", and a document whose ``storage_mode`` is "none"
1582 (a text-only document), whatever a tracker names. A document without
1583 a path of its own that no identifying tracker gives one is left
1584 alone too, unless it is labelled as stored in the database: its
1585 missing or ambiguous metadata cannot prove which file is absent.
1587 ``missing_files`` lists every document counted in ``files_missing``
1588 by the path stored for it, relative to the library root, never by
1589 the absolute path on the server (see ``_missing_file_entry``): the
1590 path that named its missing file, ``"database"`` for a document
1591 whose PDF was to be a database blob, or ``"invalid"`` for a path
1592 outside the library root.
1594 ``trackers_updated`` counts the trackers the sync changed, each
1595 once (see ``_clear_missing_pdf``).
1597 Returns:
1598 Statistics about the sync operation
1599 """
1600 with get_user_db_session(self.username) as session:
1601 # Sync only research downloads — uploads have no original_url
1602 # and no DownloadTracker, so they don't belong in this routine.
1603 # Don't load the large text_content body: this sync loop only
1604 # reads original_url/id/title (and may delete the row), so loading
1605 # every completed document's full text would needlessly exhaust
1606 # memory on a large library (#4560).
1607 documents = (
1608 session.query(Document)
1609 .filter_by(status=DocumentStatus.COMPLETED)
1610 .filter(Document.original_url.isnot(None))
1611 .options(defer(Document.text_content))
1612 .all()
1613 )
1615 stats = {
1616 "total_documents": len(documents),
1617 "files_found": 0,
1618 "files_missing": 0,
1619 "trackers_updated": 0,
1620 "missing_files": [],
1621 }
1622 # Documents deleted below — their FAISS vectors + chunk rows are
1623 # purged AFTER this transaction commits (see the post-commit block).
1624 removed_doc_ids: List[str] = []
1625 # The documents checked under each "database" tracker, by
1626 # tracker id, and the "database" trackers whose document was
1627 # deleted below. Whether such a tracker is reset is decided after
1628 # the loop, once every document under it has been checked.
1629 database_tracker_documents: Dict[int, List[str]] = {}
1630 emptied_database_trackers: Dict[int, DownloadTracker] = {}
1632 # Sync documents with filesystem.
1633 #
1634 # Resolve one settings manager for the whole sync (lazily, on the
1635 # first document that actually needs a path resolved) instead of
1636 # letting get_absolute_path_from_settings() fall back to the
1637 # ambient get_settings_manager() per document (this is a sync
1638 # `def`, so a large library means one iteration per completed
1639 # document, unbounded). Each ambient call opens a DB session that
1640 # is now tracked in a per-thread owner registry
1641 # (utilities/db_utils.py) until this request's cleanup boundary,
1642 # so enough process-wide LRU evictions between iterations can pin
1643 # one pooled connection per evicted-and-recreated session for the
1644 # rest of the request (see changelog/#6095 follow-up). Creating
1645 # it once here bounds that to at most one session, mirroring
1646 # DownloadService's self.settings pattern.
1647 settings_manager = None
1648 try:
1649 for doc in documents:
1650 # A URL has one tracker, shared by every document with
1651 # that URL: the documents of two researches that cite
1652 # the same paper, or a failed-download placeholder that
1653 # "Extract Text to DB" completed with text next to the
1654 # document that holds the PDF. URL-equivalent addresses,
1655 # such as ``paper.pdf`` and ``paper.pdf?utm_source=old``,
1656 # can also hold different PDF versions, each in a
1657 # document of its own. So a URL match alone does not say
1658 # whose file a tracker names: a tracker supplies a path
1659 # for this document, or is reset for it, only when its
1660 # content hash or explicit document link identifies the
1661 # document (find_tracker_for_document). The census of
1662 # "database" trackers below also counts every document
1663 # under the URL's tracker (``url_tracker``). B2 compat
1664 # (#6758): both lookups match legacy-hash rows too.
1665 url_tracker = find_tracker_by_url(session, doc.original_url)
1666 tracker = find_tracker_for_document(session, doc)
1667 for database_tracker in (url_tracker, tracker):
1668 if (
1669 database_tracker is not None
1670 and database_tracker.file_path == "database"
1671 ):
1672 database_tracker_documents.setdefault(
1673 database_tracker.id, []
1674 ).append(doc.id)
1675 # Explicit text-only mode and a sentinel path record a
1676 # PDF left out on purpose. Sentinel paths
1677 # (``blob_deleted``, ``text_only_not_stored``,
1678 # ``metadata_only``) name no file by design: resolving
1679 # one as a relative path always misses and would run
1680 # ``delete_document_completely`` on the text/chunks/
1681 # vectors the sentinel was protecting (issue #6760
1682 # follow-up). "Remove PDF to save space"
1683 # (delete_blob_only) writes ``blob_deleted`` to the
1684 # document but leaves the tracker naming the copy it
1685 # removed. That PDF was removed on purpose, so it is
1686 # neither missing nor to be downloaded again. Such a
1687 # document borrows no tracker path, not even from a
1688 # tracker that identifies it.
1689 if (
1690 doc.storage_mode == "none"
1691 or doc.file_path in FILE_PATH_SENTINELS
1692 ):
1693 stats["files_found"] += 1
1694 continue
1695 # The tracker's path is the one written last. It
1696 # describes this document when it names the document's
1697 # own path, which every download that stores a PDF file
1698 # writes to both. A document without a path of its own,
1699 # a database-stored one (its tracker records
1700 # "database") or a legacy row, takes the path of the
1701 # tracker that identifies it (see
1702 # _tracker_describes_storage). Any other document is
1703 # judged by what it stores itself, in the ``else``
1704 # branch, and leaves the tracker alone.
1705 # Judged by another document's file, it was counted
1706 # found while its own file was missing, or labelled a
1707 # PDF, made text-only and reported missing for a file it
1708 # never had, which the download manager then downloaded
1709 # into it as a second copy.
1710 if (
1711 tracker is not None
1712 and tracker.file_path
1713 and not _tracker_describes_storage(
1714 tracker.file_path, doc
1715 )
1716 ):
1717 tracker = None
1719 if tracker and tracker.file_path:
1720 # The tracker identifies this document and names
1721 # where it stores its PDF: its own path, the
1722 # database, or, for a document without a path of its
1723 # own, the tracker's legacy path. A sentinel there
1724 # names no file either: leave the document alone, as
1725 # above.
1726 if tracker.file_path in FILE_PATH_SENTINELS: 1726 ↛ 1727line 1726 didn't jump to line 1727 because the condition on line 1726 was never true
1727 stats["files_found"] += 1
1728 continue
1729 # Database-stored PDFs have no filesystem file to
1730 # sync — the blob is the source of truth. A present
1731 # blob counts as found. ``DocumentBlob`` is the
1732 # module's import: importing it here would make it a
1733 # local name of the whole function, which the
1734 # post-loop check below could read only after this
1735 # branch had run.
1736 if (
1737 tracker.file_path == "database"
1738 or doc.storage_mode == "database"
1739 ):
1740 has_blob = (
1741 session.query(DocumentBlob.document_id)
1742 .filter_by(document_id=doc.id)
1743 .first()
1744 is not None
1745 )
1746 if has_blob:
1747 stats["files_found"] += 1
1748 continue
1749 # Check if the file exists in the per-user root.
1750 if settings_manager is None:
1751 settings_manager = get_settings_manager(
1752 username=self.username
1753 )
1754 file_path = get_absolute_path_from_settings(
1755 tracker.file_path,
1756 self.username,
1757 settings_manager=settings_manager,
1758 )
1759 # "database" names no file: a file of that name in
1760 # the library root is not this document's PDF.
1761 if (
1762 _names_a_file(tracker.file_path)
1763 and file_path
1764 and file_path.is_file()
1765 ):
1766 stats["files_found"] += 1
1767 continue
1768 # The document's file is gone, or its blob is. What
1769 # else it stores decides.
1770 remaining = self._remaining_content(
1771 session, doc, settings_manager
1772 )
1773 if remaining == "text" and _names_a_file(
1774 tracker.file_path
1775 ):
1776 # Only the text is left of a PDF that was a
1777 # file. Keep the text and clear the dangling
1778 # PDF reference.
1779 if doc.resource_id is not None:
1780 # Its resource can download the PDF again:
1781 # report the PDF missing, under the stored
1782 # path that named the missing file.
1783 self._keep_text_report_missing_pdf(
1784 doc,
1785 tracker,
1786 tracker.file_path,
1787 file_path,
1788 stats,
1789 )
1790 else:
1791 # No download path asks for its PDF, but a
1792 # later download of the same bytes stores
1793 # the PDF in it (see _clear_missing_pdf).
1794 self._clear_missing_pdf(doc, tracker, stats)
1795 stats["files_found"] += 1
1796 elif remaining is not None:
1797 # Its PDF is still stored as a blob, or its text
1798 # is left and its path is the "database" label,
1799 # which names no file. Keep it and its tracker
1800 # unchanged.
1801 stats["files_found"] += 1
1802 else:
1803 # File missing or path invalid - mark for re-download
1804 stats["files_missing"] += 1
1805 stats["missing_files"].append(
1806 _missing_file_entry(
1807 doc, tracker.file_path, file_path
1808 )
1809 )
1811 if tracker.file_path == "database":
1812 # "database" records the URL's PDF for every
1813 # document of the URL stored there, and
1814 # another one can still hold its PDF blob.
1815 # Decided after the loop.
1816 emptied_database_trackers[tracker.id] = tracker
1817 else:
1818 # Reset tracker
1819 tracker.is_downloaded = False
1820 tracker.file_path = None
1821 stats["trackers_updated"] += 1
1823 # Delete the document entry so it can be re-queued
1824 from ..deletion.utils.cascade_helper import (
1825 CascadeHelper,
1826 )
1828 CascadeHelper.delete_document_completely(
1829 session, doc.id
1830 )
1831 removed_doc_ids.append(doc.id)
1832 else:
1833 # No tracker identifies the document, its tracker has
1834 # no path, or its tracker names another path than
1835 # the document's own: another file, or "database"
1836 # for a document with a file path. Only the
1837 # document's own tracker without a path is passed
1838 # on; otherwise ``tracker`` is None. A document
1839 # without a path of its own that is not labelled as
1840 # stored in the database names no storage to check,
1841 # e.g. a pre-upgrade row with an unset (NULL)
1842 # storage_mode. That missing or ambiguous metadata
1843 # cannot prove which file is absent, so the document
1844 # is left alone.
1845 if not doc.file_path and doc.storage_mode != "database":
1846 stats["files_found"] += 1
1847 continue
1848 # Any other label can be stale, e.g. a lost tracker
1849 # path after a "none"-mode re-download, or the
1850 # "database" default on a placeholder that later got
1851 # text. What the document still stores decides.
1852 if settings_manager is None and doc.file_path:
1853 settings_manager = get_settings_manager(
1854 username=self.username
1855 )
1856 remaining = self._remaining_content(
1857 session, doc, settings_manager
1858 )
1859 if remaining == "text" and _names_a_file(doc.file_path):
1860 # Its own PDF file is gone and only the text is
1861 # left. Keep the text and clear the dangling
1862 # PDF reference.
1863 if doc.resource_id is not None:
1864 # Its resource can download the PDF again:
1865 # report the PDF missing.
1866 self._keep_text_report_missing_pdf(
1867 doc,
1868 tracker,
1869 doc.file_path,
1870 get_absolute_path_from_settings(
1871 doc.file_path,
1872 self.username,
1873 settings_manager=settings_manager,
1874 ),
1875 stats,
1876 )
1877 else:
1878 # No download path asks for its PDF, but a
1879 # later download of the same bytes stores
1880 # the PDF in it (see _clear_missing_pdf).
1881 self._clear_missing_pdf(doc, tracker, stats)
1882 stats["files_found"] += 1
1883 continue
1884 if remaining is not None:
1885 # Its PDF is still stored, or its text is left
1886 # and it names no file (e.g. the "database"
1887 # default on a placeholder that later got text).
1888 stats["files_found"] += 1
1889 continue
1890 # Nothing of it is left: delete it so that its URL
1891 # can be downloaded again. A file its own path names
1892 # is reported missing, as the branch above reports
1893 # one that the tracker names. A document whose PDF
1894 # was to be a database blob (its label or its path
1895 # says "database") is reported as ``"database"``, as
1896 # the branch above reports one that a "database"
1897 # tracker describes.
1898 stats["files_missing"] += 1
1899 if _names_a_file(doc.file_path):
1900 stats["missing_files"].append(
1901 _missing_file_entry(
1902 doc,
1903 doc.file_path,
1904 get_absolute_path_from_settings(
1905 doc.file_path,
1906 self.username,
1907 settings_manager=settings_manager,
1908 ),
1909 )
1910 )
1911 elif ( 1911 ↛ 1918line 1911 didn't jump to line 1918 because the condition on line 1911 was always true
1912 doc.storage_mode == "database"
1913 or doc.file_path == "database"
1914 ):
1915 stats["missing_files"].append(
1916 _missing_file_entry(doc, "database", None)
1917 )
1918 from ..deletion.utils.cascade_helper import (
1919 CascadeHelper,
1920 )
1922 CascadeHelper.delete_document_completely(
1923 session, doc.id
1924 )
1925 removed_doc_ids.append(doc.id)
1926 finally:
1927 safe_close(
1928 settings_manager, "settings manager", allow_none=True
1929 )
1931 # A "database" tracker records that its URL's PDF is stored in
1932 # the database, which can hold it for several documents of the
1933 # URL. The one it identifies was deleted above with nothing
1934 # left. While another still holds a PDF blob, the record is true
1935 # and the tracker is left as it is; otherwise it is reset. A
1936 # document the tracker does not identify cannot reset it, but its
1937 # blob still keeps the record true. Every document under the
1938 # tracker has been checked by now, so the order in which the
1939 # sync read them does not change whether it is reset.
1940 removed = set(removed_doc_ids)
1941 for tracker_id, tracker in emptied_database_trackers.items():
1942 others = [
1943 document_id
1944 for document_id in database_tracker_documents[tracker_id]
1945 if document_id not in removed
1946 ]
1947 if others and (
1948 session.query(DocumentBlob.document_id)
1949 .filter(DocumentBlob.document_id.in_(others))
1950 .first()
1951 is not None
1952 ):
1953 continue
1954 tracker.is_downloaded = False
1955 tracker.file_path = None
1956 stats["trackers_updated"] += 1
1958 session.commit()
1959 logger.info(
1960 f"Library sync completed: {stats['files_found']} found, {stats['files_missing']} missing"
1961 )
1963 # Post-commit: purge the removed documents' FAISS vectors + chunk rows.
1964 # delete_document_completely leaves DocumentChunk rows (no FK), so
1965 # without this the removed doc's vectors and chunk_text linger and keep
1966 # surfacing in collection search. Reuse DocumentDeletionService's
1967 # post-commit purge (per-collection vector removal + guaranteed chunk-row
1968 # backstop), run after the session above has closed.
1969 if removed_doc_ids:
1970 from ..deletion.services.document_deletion import (
1971 DocumentDeletionService,
1972 )
1973 from ...database.models.library import DocumentChunk
1975 deleter = DocumentDeletionService(self.username)
1976 for doc_id in removed_doc_ids:
1977 # Best-effort per document: the sync's DB deletes already
1978 # committed, so a purge hiccup here must not raise out of the
1979 # sync (which would report the whole operation as failed) nor
1980 # abort the remaining documents' purges.
1981 try:
1982 with get_user_db_session(self.username) as purge_session:
1983 coll_rows = (
1984 purge_session.query(DocumentChunk.collection_name)
1985 .filter_by(source_type="document", source_id=doc_id)
1986 .distinct()
1987 .all()
1988 )
1989 collection_ids = [
1990 row[0].removeprefix("collection_")
1991 for row in coll_rows
1992 if row[0]
1993 ]
1994 deleter._purge_document_rag(
1995 doc_id, collection_ids, full_delete=True
1996 )
1997 except Exception:
1998 logger.opt(exception=True).error(
1999 "Failed to purge removed document {} after library sync "
2000 "(delete already committed)",
2001 doc_id,
2002 )
2004 return stats
2006 def mark_for_redownload(self, document_ids: List[str]) -> int:
2007 """
2008 Mark specific documents for re-download.
2010 Args:
2011 document_ids: List of document IDs to mark for re-download
2013 Returns:
2014 Number of documents marked
2015 """
2016 with get_user_db_session(self.username) as session:
2017 count = 0
2018 for doc_id in document_ids:
2019 doc = session.get(Document, doc_id)
2020 if doc:
2021 if not doc.original_url:
2022 # Uploads have no source URL and no DownloadTracker,
2023 # so re-download is not meaningful for them.
2024 logger.warning(
2025 f"Skipping mark-for-redownload on {doc_id[:8]}…: "
2026 "document has no original_url (likely user upload)"
2027 )
2028 continue
2029 # Get tracker and reset it. B2 compat (#6758): match
2030 # legacy-hash rows so a stale tracker is not left
2031 # is_downloaded=True while the document is PENDING.
2032 tracker = find_tracker_by_url(session, doc.original_url)
2034 if tracker: 2034 ↛ 2039line 2034 didn't jump to line 2039 because the condition on line 2034 was always true
2035 tracker.is_downloaded = False
2036 tracker.file_path = None
2038 # Mark document as pending
2039 doc.status = DocumentStatus.PENDING
2040 count += 1
2042 session.commit()
2043 logger.info(f"Marked {count} documents for re-download")
2044 return count