Coverage for src/local_deep_research/research_library/routes/library_routes.py: 94%
568 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-20 01:24 +0000
1"""
2Routes for Research Library and Download Manager
4Provides web endpoints for:
5- Library browsing and management
6- Download manager interface
7- API endpoints for downloads and queries
8"""
10import json
11import math
12from datetime import UTC, datetime
13from io import BytesIO
14from pathlib import Path
15from flask import (
16 Blueprint,
17 g,
18 jsonify,
19 request,
20 session,
21 Response,
22 send_file,
23 stream_with_context,
24)
25from loguru import logger
27from ...security.decorators import require_json_body
28from ...web.auth.decorators import login_required
29from ...web.utils.templates import render_template_with_defaults
30from ...database.session_context import get_user_db_session, safe_rollback
31from ...database.models.research import ResearchResource
32from ...database.models.library import (
33 Document as Document,
34 DocumentStatus,
35 DownloadQueue as LibraryDownloadQueue,
36 Collection,
37)
38from ...library.download_management import ResourceFilter
39from ..services.download_service import DownloadService
40from ..services.library_service import LibraryService
41from ..services.pdf_storage_manager import PDFStorageManager
42from ..utils import (
43 get_document_for_resource,
44 handle_api_error,
45 is_downloadable_domain,
46 is_downloadable_url,
47)
48from ...utilities.db_utils import get_settings_manager
49from ...config.paths import get_library_directory
50from ...web.exceptions import AuthenticationRequiredError
52# Create Blueprint
53library_bp = Blueprint("library", __name__, url_prefix="/library")
55# NOTE: Routes use session["username"] (not .get()) intentionally.
56# @login_required guarantees the key exists; direct access fails fast
57# if the decorator is ever removed.
60# Error handler for authentication errors
61@library_bp.errorhandler(Exception)
62def handle_web_api_exception(error):
63 """Handle WebAPIException and its subclasses."""
64 from ...web.exceptions import WebAPIException
66 if isinstance(error, WebAPIException):
67 return jsonify(error.to_dict()), error.status_code
68 # Re-raise other exceptions
69 raise error
72def get_authenticated_user_password(
73 username: str, flask_session_id: str | None = None
74) -> str:
75 """
76 Get authenticated user password from session store with fallback to g.user_password.
78 Args:
79 username: The username to get password for
80 flask_session_id: Optional Flask session ID. If not provided, uses session.get("session_id")
82 Returns:
83 str: The user's password
85 Raises:
86 AuthenticationRequiredError: If no password is available for the user
87 """
88 from ...database.session_passwords import session_password_store
90 session_id = flask_session_id or session.get("session_id")
92 # Try session password store first
93 try:
94 user_password = session_password_store.get_session_password(
95 username, session_id
96 )
97 if user_password:
98 logger.debug(
99 f"Retrieved user password from session store for user {username}"
100 )
101 return user_password
102 except Exception:
103 logger.exception("Failed to get user password from session store")
105 # Fallback to g.user_password (set by middleware if temp_auth was used)
106 user_password = getattr(g, "user_password", None)
107 if user_password:
108 logger.debug(
109 f"Retrieved user password from g.user_password fallback for user {username}"
110 )
111 return user_password
113 # No password available
114 logger.error(f"No user password available for user {username}")
115 raise AuthenticationRequiredError(
116 message="Authentication required: Please refresh the page and log in again to access encrypted database features.",
117 )
120# ============= Page Routes =============
123@library_bp.route("/")
124@login_required
125def library_page():
126 """Main library page showing downloaded documents."""
127 username = session["username"]
128 service = LibraryService(username)
130 # Get library settings
131 from ...utilities.db_utils import get_settings_manager
133 settings = get_settings_manager()
134 pdf_storage_mode = settings.get_setting(
135 "research_library.pdf_storage_mode", "database"
136 )
137 # Enable PDF storage button if mode is not "none"
138 enable_pdf_storage = pdf_storage_mode != "none"
139 shared_library = settings.get_setting(
140 "research_library.shared_library", False
141 )
143 # Get statistics
144 stats = service.get_library_stats()
146 # Get documents with optional filters
147 domain_filter = request.args.get("domain")
148 research_filter = request.args.get("research")
149 collection_filter = request.args.get("collection") # New collection filter
150 date_filter = request.args.get("date")
152 # Resolve collection_id once to avoid redundant DB lookups
153 from ...database.library_init import get_default_library_id
155 resolved_collection = collection_filter or get_default_library_id(username)
157 # Pagination
158 per_page = 100
159 total_docs = service.count_documents(
160 research_id=research_filter,
161 domain=domain_filter,
162 collection_id=resolved_collection,
163 date_filter=date_filter,
164 )
165 total_pages = max(1, math.ceil(total_docs / per_page))
166 page = request.args.get("page", 1, type=int)
167 page = max(1, min(page, total_pages))
168 offset = (page - 1) * per_page
170 documents = service.get_documents(
171 research_id=research_filter,
172 domain=domain_filter,
173 collection_id=resolved_collection,
174 date_filter=date_filter,
175 limit=per_page,
176 offset=offset,
177 )
179 # Get unique domains for filter dropdown
180 unique_domains = service.get_unique_domains()
182 # Get research list for filter dropdown
183 research_list = service.get_research_list_for_dropdown()
185 # Get collections list for filter dropdown
186 collections = service.get_all_collections()
188 # Find default library collection ID for semantic search
189 default_collection_id = next(
190 (c["id"] for c in collections if c.get("is_default")), None
191 )
193 return render_template_with_defaults(
194 "pages/library.html",
195 stats=stats,
196 documents=documents,
197 unique_domains=unique_domains,
198 research_list=research_list,
199 collections=collections,
200 selected_collection=collection_filter,
201 default_collection_id=default_collection_id,
202 storage_path=stats.get("storage_path", ""),
203 enable_pdf_storage=enable_pdf_storage,
204 pdf_storage_mode=pdf_storage_mode,
205 shared_library=shared_library,
206 page=page,
207 total_pages=total_pages,
208 selected_date=date_filter,
209 selected_research=research_filter,
210 selected_domain=domain_filter,
211 )
214@library_bp.route("/document/<string:document_id>")
215@login_required
216def document_details_page(document_id):
217 """Document details page showing all metadata and links."""
218 username = session["username"]
219 service = LibraryService(username)
221 # Get document details
222 document = service.get_document_by_id(document_id)
224 if not document:
225 return "Document not found", 404
227 return render_template_with_defaults(
228 "pages/document_details.html", document=document
229 )
232@library_bp.route("/download-manager")
233@login_required
234def download_manager_page():
235 """Download manager page for selecting and downloading research PDFs."""
236 username = session["username"]
237 service = LibraryService(username)
239 # Get library settings
240 from ...utilities.db_utils import get_settings_manager
242 settings = get_settings_manager()
243 pdf_storage_mode = settings.get_setting(
244 "research_library.pdf_storage_mode", "database"
245 )
246 # Enable PDF storage button if mode is not "none"
247 enable_pdf_storage = pdf_storage_mode != "none"
248 shared_library = settings.get_setting(
249 "research_library.shared_library", False
250 )
252 # Summary stats over ALL sessions (also used for page count)
253 per_page = 50
254 summary = service.get_download_manager_summary_stats()
255 total_pages = max(1, math.ceil(summary["total_researches"] / per_page))
257 # Pagination with upper-bound clamp
258 page = request.args.get("page", 1, type=int)
259 page = max(1, min(page, total_pages))
260 offset = (page - 1) * per_page
262 # Get paginated research sessions
263 research_list = service.get_research_list_with_stats(
264 limit=per_page, offset=offset
265 )
267 # Batch-fetch PDF previews and domain breakdowns (single query)
268 research_ids = [r["id"] for r in research_list]
269 previews = service.get_pdf_previews_batch(research_ids)
270 for research in research_list:
271 rid = research["id"]
272 data = previews.get(rid, {"pdf_sources": [], "domains": {}})
273 research["pdf_sources"] = data["pdf_sources"]
274 research["domains"] = data["domains"]
276 return render_template_with_defaults(
277 "pages/download_manager.html",
278 research_list=research_list,
279 total_researches=summary["total_researches"],
280 total_resources=summary["total_resources"],
281 already_downloaded=summary["already_downloaded"],
282 available_to_download=summary["available_to_download"],
283 enable_pdf_storage=enable_pdf_storage,
284 pdf_storage_mode=pdf_storage_mode,
285 shared_library=shared_library,
286 page=page,
287 total_pages=total_pages,
288 )
291# ============= API Routes =============
294@library_bp.route("/api/stats")
295@login_required
296def get_library_stats():
297 """Get library statistics."""
298 username = session["username"]
299 service = LibraryService(username)
300 stats = service.get_library_stats()
301 return jsonify(stats)
304@library_bp.route("/api/collections/list")
305@login_required
306def get_collections_list():
307 """Get list of all collections for dropdown selection."""
308 username = session["username"]
310 with get_user_db_session(username) as db_session:
311 collections = (
312 db_session.query(Collection).order_by(Collection.name).all()
313 )
315 return jsonify(
316 {
317 "success": True,
318 "collections": [
319 {
320 "id": col.id,
321 "name": col.name,
322 "description": col.description,
323 }
324 for col in collections
325 ],
326 }
327 )
330@library_bp.route("/api/documents")
331@login_required
332def get_documents():
333 """Get documents with filtering."""
334 username = session["username"]
335 service = LibraryService(username)
337 # Get filter parameters
338 research_id = request.args.get("research_id")
339 domain = request.args.get("domain")
340 file_type = request.args.get("file_type")
341 favorites_only = request.args.get("favorites") == "true"
342 search_query = request.args.get("search")
343 # Clamp pagination. SQLite treats LIMIT -1 as "no limit", so an
344 # unclamped negative limit (e.g. ?limit=-1) would bypass pagination and
345 # load the whole collection — including large Document.text_content
346 # bodies — into memory (#4560). Mirrors the clamp in history_routes.
347 limit = request.args.get("limit", 100, type=int)
348 limit = max(1, min(limit, 1000))
349 offset = max(0, request.args.get("offset", 0, type=int))
351 documents = service.get_documents(
352 research_id=research_id,
353 domain=domain,
354 file_type=file_type,
355 favorites_only=favorites_only,
356 search_query=search_query,
357 limit=limit,
358 offset=offset,
359 )
361 return jsonify({"documents": documents})
364@library_bp.route(
365 "/api/document/<string:document_id>/favorite", methods=["POST"]
366)
367@login_required
368def toggle_favorite(document_id):
369 """Toggle favorite status of a document."""
370 username = session["username"]
371 service = LibraryService(username)
372 is_favorite = service.toggle_favorite(document_id)
373 return jsonify({"favorite": is_favorite})
376# NOTE: DELETE /library/api/document/<id> is intentionally NOT defined here.
377# The canonical handler lives in deletion/routes/delete_routes.py
378# (delete_bp, url_prefix="/library/api") which adds the note-protection
379# guard (returns 403 for a note id, routing the caller to DELETE
380# /api/notes/<id>) and the richer cascade (filesystem + FAISS + chunk
381# stats). This blueprint previously also declared that exact path; since
382# library_bp is registered before delete_bp (app_factory.py), Werkzeug
383# resolved to THIS unguarded handler first, making the guard dead code and
384# letting a note Document be hard-deleted via the document API. The
385# duplicate route was removed so the guarded endpoint is reachable. The
386# route-resolution invariant is locked by a test in
387# tests/research_library/deletion (test_delete_document_route_is_guarded).
390@library_bp.route("/api/document/<string:document_id>/pdf-url")
391@login_required
392def get_pdf_url(document_id):
393 """Get URL for viewing PDF."""
394 # Return URL that will serve the PDF
395 return jsonify(
396 {
397 "url": f"/library/api/document/{document_id}/pdf",
398 "title": "Document", # Could fetch actual title
399 }
400 )
403@library_bp.route("/document/<string:document_id>/pdf")
404@login_required
405def view_pdf_page(document_id):
406 """Page for viewing PDF file - uses PDFStorageManager for retrieval."""
407 username = session["username"]
409 with get_user_db_session(username) as db_session:
410 # Get document from database
411 document = db_session.query(Document).filter_by(id=document_id).first()
413 if not document:
414 logger.warning(
415 f"Document ID {document_id} not found in database for user {username}"
416 )
417 return "Document not found", 404
419 logger.info(
420 f"Document {document_id}: title='{document.title}', "
421 f"file_path={document.file_path}"
422 )
424 # Get settings for PDF storage manager
425 settings = get_settings_manager(db_session)
426 storage_mode = settings.get_setting(
427 "research_library.pdf_storage_mode", "none"
428 )
429 library_root = (
430 Path(
431 settings.get_setting(
432 "research_library.storage_path",
433 str(get_library_directory()),
434 )
435 )
436 .expanduser()
437 .resolve()
438 )
440 # Use PDFStorageManager to load PDF (handles database and filesystem)
441 pdf_manager = PDFStorageManager(library_root, storage_mode)
442 pdf_bytes = pdf_manager.load_pdf(document, db_session)
444 if pdf_bytes:
445 logger.info(
446 f"Serving PDF for document {document_id} ({len(pdf_bytes)} bytes)"
447 )
448 return send_file(
449 BytesIO(pdf_bytes),
450 mimetype="application/pdf",
451 as_attachment=False,
452 download_name=document.filename or "document.pdf",
453 )
455 # No PDF found anywhere
456 logger.warning(f"No PDF available for document {document_id}")
457 return "PDF not available", 404
460@library_bp.route("/api/document/<string:document_id>/pdf")
461@login_required
462def serve_pdf_api(document_id):
463 """API endpoint for serving PDF file (kept for backward compatibility)."""
464 return view_pdf_page(document_id)
467@library_bp.route("/document/<string:document_id>/txt")
468@login_required
469def view_text_page(document_id):
470 """Page for viewing text content."""
471 username = session["username"]
473 with get_user_db_session(username) as db_session:
474 # Get document by ID (text now stored in Document.text_content)
475 document = db_session.query(Document).filter_by(id=document_id).first()
477 if not document:
478 logger.warning(f"Document not found for document ID {document_id}")
479 return "Document not found", 404
481 if not document.text_content:
482 logger.warning(f"Document {document_id} has no text content")
483 return "Text content not available", 404
485 logger.info(
486 f"Serving text content for document {document_id}: {len(document.text_content)} characters"
487 )
489 # Render as HTML page
490 return render_template_with_defaults(
491 "pages/document_text.html",
492 document_id=document_id,
493 title=document.title or "Document Text",
494 text_content=document.text_content,
495 extraction_method=document.extraction_method,
496 word_count=document.word_count,
497 )
500@library_bp.route("/api/document/<string:document_id>/text")
501@login_required
502def serve_text_api(document_id):
503 """API endpoint for serving text content (kept for backward compatibility)."""
504 username = session["username"]
506 with get_user_db_session(username) as db_session:
507 # Get document by ID (text now stored in Document.text_content)
508 document = db_session.query(Document).filter_by(id=document_id).first()
510 if not document:
511 logger.warning(f"Document not found for document ID {document_id}")
512 return jsonify({"error": "Document not found"}), 404
514 if not document.text_content:
515 logger.warning(f"Document {document_id} has no text content")
516 return jsonify({"error": "Text content not available"}), 404
518 logger.info(
519 f"Serving text content for document {document_id}: {len(document.text_content)} characters"
520 )
522 return jsonify(
523 {
524 "text_content": document.text_content,
525 "title": document.title or "Document",
526 "extraction_method": document.extraction_method,
527 "word_count": document.word_count,
528 }
529 )
532@library_bp.route("/api/open-folder", methods=["POST"])
533@login_required
534def open_folder():
535 """Open folder containing a document.
537 Security: This endpoint is disabled for server deployments.
538 It only makes sense for desktop usage where the server and client are on the same machine.
539 """
540 return jsonify(
541 {
542 "status": "error",
543 "message": "This feature is disabled. It is only available in desktop mode.",
544 }
545 ), 403
548@library_bp.route("/api/download/<int:resource_id>", methods=["POST"])
549@login_required
550def download_single_resource(resource_id):
551 """Download a single resource."""
552 username = session["username"]
553 user_password = get_authenticated_user_password(username)
555 with DownloadService(username, user_password) as service:
556 success, error = service.download_resource(resource_id)
557 if success:
558 return jsonify({"success": True})
559 logger.warning(f"Download failed for resource {resource_id}: {error}")
560 return jsonify(
561 {
562 "success": False,
563 "error": "Download failed. Please try again or contact support.",
564 }
565 ), 500
568@library_bp.route("/api/download-text/<int:resource_id>", methods=["POST"])
569@login_required
570def download_text_single(resource_id):
571 """Download a single resource as text file."""
572 try:
573 username = session["username"]
574 user_password = get_authenticated_user_password(username)
576 with DownloadService(username, user_password) as service:
577 success, error = service.download_as_text(resource_id)
579 # Sanitize error message - don't expose internal details
580 if not success:
581 if error: 581 ↛ 585line 581 didn't jump to line 585 because the condition on line 581 was always true
582 logger.warning(
583 f"Download as text failed for resource {resource_id}: {error}"
584 )
585 return jsonify(
586 {"success": False, "error": "Failed to download resource"}
587 )
589 return jsonify({"success": True, "error": None})
590 except AuthenticationRequiredError:
591 raise # Let blueprint error handler return 401
592 except Exception as e:
593 return handle_api_error(
594 f"downloading resource {resource_id} as text", e
595 )
598@library_bp.route("/api/download-all-text", methods=["POST"])
599@login_required
600def download_all_text():
601 """Download all undownloaded resources as text files."""
602 username = session["username"]
603 # Capture Flask session ID to avoid scoping issues in nested function
604 flask_session_id = session.get("session_id")
606 def generate():
607 # Get user password for database operations
608 try:
609 user_password = get_authenticated_user_password(
610 username, flask_session_id
611 )
612 except AuthenticationRequiredError:
613 logger.warning(
614 f"Authentication unavailable for user {username} - password not in session store"
615 )
616 yield f"data: {json.dumps({'progress': 0, 'current': 0, 'total': 0, 'error': 'Authentication required', 'complete': True})}\n\n"
617 return
619 download_service = DownloadService(username, user_password)
620 try:
621 # Get all undownloaded resources
622 with get_user_db_session(username) as session:
623 # Project only the columns the loop uses (id/url/title) rather
624 # than loading full ``ResearchResource`` entities — the
625 # ``content_preview`` Text column would otherwise materialize
626 # for every resource on this whole-table scan (#4560).
627 all_resources = session.query(
628 ResearchResource.id,
629 ResearchResource.url,
630 ResearchResource.title,
631 ).all()
632 # Filter to only downloadable resources (academic/PDF)
633 resources = [
634 r for r in all_resources if is_downloadable_url(r.url)
635 ]
637 # Filter resources that need text extraction
638 txt_path = Path(download_service.library_root) / "txt"
639 resources_to_process = []
641 # Pre-scan directory once to get all existing resource IDs
642 existing_resource_ids = set()
643 if txt_path.exists(): 643 ↛ 644line 643 didn't jump to line 644 because the condition on line 643 was never true
644 for txt_file in txt_path.glob("*.txt"):
645 # Extract resource ID from filename pattern *_{id}.txt
646 parts = txt_file.stem.rsplit("_", 1)
647 if len(parts) == 2:
648 try:
649 existing_resource_ids.add(int(parts[1]))
650 except ValueError:
651 pass
653 for resource in resources:
654 # Check if text file already exists using preloaded set
655 if resource.id not in existing_resource_ids: 655 ↛ 653line 655 didn't jump to line 653 because the condition on line 655 was always true
656 resources_to_process.append(resource)
658 total = len(resources_to_process)
659 current = 0
661 logger.info(f"Found {total} resources needing text extraction")
663 for resource in resources_to_process:
664 current += 1
665 progress = (
666 int((current / total) * 100) if total > 0 else 100
667 )
669 file_name = (
670 resource.title[:50]
671 if resource
672 else f"document_{current}.txt"
673 )
675 try:
676 success, error = download_service.download_as_text(
677 resource.id
678 )
680 if success:
681 status = "success"
682 error_msg = None
683 else:
684 status = "failed"
685 error_msg = error or "Text extraction failed"
687 except Exception as e:
688 logger.exception(
689 f"Error extracting text for resource {resource.id}"
690 )
691 status = "failed"
692 error_msg = (
693 f"Text extraction failed - {type(e).__name__}"
694 )
696 # Send update
697 update = {
698 "progress": progress,
699 "current": current,
700 "total": total,
701 "file": file_name,
702 "url": resource.url, # Add the URL for UI display
703 "status": status,
704 "error": error_msg,
705 }
706 yield f"data: {json.dumps(update)}\n\n"
708 # Send completion
709 yield f"data: {json.dumps({'complete': True, 'total': total})}\n\n"
710 finally:
711 from ...utilities.resource_utils import safe_close
713 safe_close(download_service, "download service")
715 return Response(
716 stream_with_context(generate()), mimetype="text/event-stream"
717 )
720@library_bp.route("/api/download-research/<research_id>", methods=["POST"])
721@login_required
722def download_research_pdfs(research_id):
723 """Queue all PDFs from a research session for download."""
724 username = session["username"]
725 user_password = get_authenticated_user_password(username)
727 with DownloadService(username, user_password) as service:
728 # Get optional collection_id from request body
729 data = request.json or {}
730 collection_id = data.get("collection_id")
732 queued = service.queue_research_downloads(research_id, collection_id)
734 # Start processing queue (in production, this would be a background task)
735 # For now, we'll process synchronously
736 # TODO: Integrate with existing queue processor
738 return jsonify({"success": True, "queued": queued})
741def _claim_download_queue_item(db_session, queue_item_id):
742 """Atomically claim a PENDING download-queue row before processing it.
744 Flip the row from PENDING to PROCESSING in a single conditional UPDATE and
745 return True only when this caller won the claim. Two concurrent
746 ``download_bulk`` streams for the same user (the user clicks Download
747 twice, or a second bulk run starts before the first finishes) both read
748 the same PENDING rows and, with no claim, both call ``download_resource``
749 for every row (issue #4691). The row-count guard makes the
750 PENDING -> PROCESSING transition the single point where exactly one stream
751 wins each row; the loser sees 0 updated rows and skips it.
752 """
753 claimed = (
754 db_session.query(LibraryDownloadQueue)
755 .filter_by(id=queue_item_id, status=DocumentStatus.PENDING)
756 .update(
757 {LibraryDownloadQueue.status: DocumentStatus.PROCESSING},
758 synchronize_session=False,
759 )
760 )
761 db_session.commit()
762 return claimed == 1
765def _release_download_queue_item(db_session, queue_item_id):
766 """Return a claimed row from PROCESSING back to PENDING so it stays
767 retryable when the download raised before a terminal COMPLETED/FAILED
768 status was recorded (issue #4691), matching the pre-fix behaviour where an
769 errored row was left PENDING. Scoped to PROCESSING so it only ever undoes
770 this stream's own claim.
771 """
772 db_session.query(LibraryDownloadQueue).filter_by(
773 id=queue_item_id, status=DocumentStatus.PROCESSING
774 ).update(
775 {LibraryDownloadQueue.status: DocumentStatus.PENDING},
776 synchronize_session=False,
777 )
778 db_session.commit()
781def _finalize_download_queue_item(db_session, queue_item_id, success):
782 """Record the outcome of a claimed row that returned without raising.
784 ``download_resource`` writes a terminal COMPLETED/FAILED status itself for
785 the rows it actually downloads, and this is a no-op for those: the row is
786 no longer PROCESSING, so the guard matches nothing and pdf-mode behaviour
787 is unchanged. It fires only on the paths that otherwise strand a claim
788 (issue #4691):
790 - ``mode="text_only"``: ``download_as_text`` and its call tree never touch
791 ``LibraryDownloadQueue``, so a successful extraction would leave the row
792 PROCESSING forever. Nothing in ``src/`` reaps PROCESSING rows, and a
793 successful extraction also writes a COMPLETED Document, which blocks both
794 the pre-pass reset and ``queue_all_undownloaded``'s outer-join filter.
795 - ``download_resource``'s already-downloaded early return, which reports
796 success before reaching its own queue update.
798 Pre-fix, both paths left the row PENDING and merely re-scanned it, so
799 failure returns to PENDING rather than FAILED to keep that retryability.
800 """
801 db_session.query(LibraryDownloadQueue).filter_by(
802 id=queue_item_id, status=DocumentStatus.PROCESSING
803 ).update(
804 {
805 LibraryDownloadQueue.status: (
806 DocumentStatus.COMPLETED if success else DocumentStatus.PENDING
807 ),
808 LibraryDownloadQueue.completed_at: (
809 datetime.now(UTC) if success else None
810 ),
811 },
812 synchronize_session=False,
813 )
814 db_session.commit()
817@library_bp.route("/api/download-bulk", methods=["POST"])
818@login_required
819@require_json_body()
820def download_bulk():
821 """Download PDFs or extract text from multiple research sessions."""
822 username = session["username"]
823 data = request.json
824 research_ids = data.get("research_ids", [])
825 mode = data.get("mode", "pdf") # pdf or text_only
826 collection_id = data.get(
827 "collection_id"
828 ) # Optional: target collection for downloads
830 if not research_ids:
831 return jsonify({"error": "No research IDs provided"}), 400
833 # Capture Flask session ID to avoid scoping issues in nested function
834 flask_session_id = session.get("session_id")
836 def generate():
837 """Generate progress updates as Server-Sent Events."""
838 # Get user password for database operations
839 try:
840 user_password = get_authenticated_user_password(
841 username, flask_session_id
842 )
843 except AuthenticationRequiredError:
844 logger.warning(
845 f"Authentication unavailable for user {username} - password not in session store"
846 )
847 yield f"data: {json.dumps({'progress': 0, 'current': 0, 'total': 0, 'error': 'Authentication required', 'complete': True})}\n\n"
848 return
850 download_service = DownloadService(username, user_password)
851 try:
852 total = 0
853 current = 0
854 queue_failures = 0
856 # Pre-queue and count in one merged loop. queue_research_downloads
857 # opens its own session, commits, and handles dedup internally:
858 # it skips resources that are already PENDING or already backed by
859 # a COMPLETED Document, and re-queues any remaining row (e.g. a
860 # prior FAILED attempt) back to PENDING. Calling it BEFORE counting
861 # ensures the total reflects what will actually be processed
862 # (issue #4660).
863 for research_id in research_ids:
864 try:
865 download_service.queue_research_downloads(
866 research_id, collection_id
867 )
868 except Exception:
869 queue_failures += 1
870 logger.exception(
871 f"Error queueing downloads for "
872 f"user={username} research={research_id} "
873 f"collection={collection_id}"
874 )
876 with get_user_db_session(username) as session:
877 count = (
878 session.query(LibraryDownloadQueue)
879 .filter_by(
880 research_id=research_id,
881 status=DocumentStatus.PENDING,
882 )
883 .count()
884 )
885 total += count
886 logger.debug(
887 f"Research {research_id}: {count} pending items in queue"
888 )
890 logger.info(f"Total pending downloads across all research: {total}")
891 yield f"data: {json.dumps({'progress': 0, 'current': 0, 'total': total})}\n\n"
893 # If the pre-pass yielded nothing to process, surface it as an
894 # error so the UI alerts instead of silently completing with
895 # "0 / 0 files" success. Distinguish queue failure from a
896 # legitimate "nothing left to download" state.
897 if total == 0:
898 if queue_failures > 0:
899 error_msg = (
900 f"Download failed to start: queueing failed for "
901 f"{queue_failures} of {len(research_ids)} "
902 f"research session(s). Check server logs."
903 )
904 else:
905 error_msg = (
906 "No new papers to download for the selected "
907 "research session(s)."
908 )
909 yield f"data: {json.dumps({'progress': 100, 'current': 0, 'total': 0, 'complete': True, 'error': error_msg})}\n\n"
910 return
912 # Process each research
913 for research_id in research_ids:
914 # Get queued downloads for this research
915 with get_user_db_session(username) as session:
916 # Get pending queue items for this research (pre-pass
917 # above already ensured items are queued)
918 queue_items = (
919 session.query(LibraryDownloadQueue)
920 .filter_by(
921 research_id=research_id,
922 status=DocumentStatus.PENDING,
923 )
924 .all()
925 )
927 # Process each queued item
928 for queue_item in queue_items:
929 # Atomically claim this row before downloading so a
930 # second concurrent download_bulk stream can't also
931 # process it (issue #4691). A row already claimed by
932 # another stream returns 0 updated rows; skip it.
933 queue_item_id = queue_item.id
934 if not _claim_download_queue_item( 934 ↛ 937line 934 didn't jump to line 937 because the condition on line 934 was never true
935 session, queue_item_id
936 ):
937 continue
939 current += 1
941 progress = (
942 int((current / total) * 100) if total > 0 else 100
943 )
945 # Attempt actual download with error handling
946 skip_reason = None
947 status = "skipped" # Default to skipped
948 success = False
949 error_msg = None
950 # Everything from the resource lookup onward runs
951 # inside the try below, so file_name needs a value
952 # even if that block raises before setting one.
953 file_name = f"document_{current}.pdf"
955 try:
956 # Get resource info. This sits INSIDE the try
957 # because the claim above is already committed as
958 # PROCESSING: any raise between the claim and this
959 # handler would leave the row that way forever,
960 # since nothing in src/ reaps PROCESSING rows and
961 # every reset path now refuses to touch them
962 # (issue #4691). ResearchResource.title is
963 # nullable, so resource.title[:50] raises TypeError
964 # on a resource whose title was never populated.
965 resource = session.get(
966 ResearchResource, queue_item.resource_id
967 )
968 if resource and resource.title:
969 file_name = resource.title[:50]
971 logger.debug(
972 f"Attempting {'PDF download' if mode == 'pdf' else 'text extraction'} for resource {queue_item.resource_id}"
973 )
975 # Call appropriate service method based on mode
976 if mode == "pdf":
977 result = download_service.download_resource(
978 queue_item.resource_id
979 )
980 else: # text_only
981 result = download_service.download_as_text(
982 queue_item.resource_id
983 )
985 # Handle new tuple return format
986 if isinstance(result, tuple): 986 ↛ 989line 986 didn't jump to line 989 because the condition on line 986 was always true
987 success, skip_reason = result
988 else:
989 success = result
990 skip_reason = None
992 # Finalize the claim for the paths that don't
993 # record a terminal status themselves (text_only,
994 # and download_resource's already-downloaded early
995 # return); a no-op when they already did (#4691).
996 _finalize_download_queue_item(
997 session, queue_item_id, success
998 )
1000 status = "success" if success else "skipped"
1001 if skip_reason and not success:
1002 error_msg = skip_reason
1003 logger.info(
1004 f"{'Download' if mode == 'pdf' else 'Text extraction'} skipped for resource {queue_item.resource_id}: {skip_reason}"
1005 )
1007 logger.debug(
1008 f"{'Download' if mode == 'pdf' else 'Text extraction'} result: success={success}, status={status}, skip_reason={skip_reason}"
1009 )
1010 except Exception as e:
1011 # Roll back FIRST: the next loop iteration's
1012 # session.get(ResearchResource, ...) at the
1013 # top of the for-body runs BEFORE the next
1014 # try/except, so a poisoned session here would
1015 # cascade into PendingRollbackError on the next
1016 # item before this handler ever runs again
1017 # (issue #3827).
1018 safe_rollback(session, "SSE download")
1019 # Release the claim so a later run can retry this
1020 # row: the download raised before
1021 # download_resource recorded a terminal status
1022 # (issue #4691).
1023 _release_download_queue_item(session, queue_item_id)
1024 # Log error but continue processing
1025 error_msg = str(e)
1026 error_type = type(e).__name__
1027 logger.warning(
1028 f"CAUGHT Download exception for resource {queue_item.resource_id}: {error_type}: {error_msg}"
1029 )
1030 # Check if this is a skip reason (not a real error)
1031 # Use error category + categorized message for user display
1032 if any(
1033 phrase in error_msg.lower()
1034 for phrase in [
1035 "paywall",
1036 "subscription",
1037 "not available",
1038 "not found",
1039 "no free",
1040 "embargoed",
1041 "forbidden",
1042 "not accessible",
1043 ]
1044 ):
1045 status = "skipped"
1046 skip_reason = f"Document not accessible (paywall or access restriction) - {error_type}"
1047 elif any(
1048 phrase in error_msg.lower()
1049 for phrase in [
1050 "failed to download",
1051 "could not",
1052 "invalid",
1053 "server",
1054 ]
1055 ):
1056 status = "failed"
1057 skip_reason = f"Download failed - {error_type}"
1058 else:
1059 status = "failed"
1060 skip_reason = (
1061 f"Processing failed - {error_type}"
1062 )
1063 success = False
1065 # Ensure skip_reason is set if we have an error message
1066 if error_msg and not skip_reason: 1066 ↛ 1067line 1066 didn't jump to line 1067 because the condition on line 1066 was never true
1067 skip_reason = f"Processing failed - {error_type}"
1068 logger.debug(
1069 f"Setting skip_reason from error_msg: {error_msg}"
1070 )
1072 # Send progress update
1073 update_data = {
1074 "progress": progress,
1075 "current": current,
1076 "total": total,
1077 "file": file_name,
1078 "status": status,
1079 }
1080 # Add skip reason if available
1081 if skip_reason:
1082 update_data["error"] = skip_reason
1083 logger.info(
1084 f"Sending skip reason to UI: {skip_reason}"
1085 )
1087 yield f"data: {json.dumps(update_data)}\n\n"
1089 yield f"data: {json.dumps({'progress': 100, 'current': current, 'total': total, 'complete': True})}\n\n"
1090 finally:
1091 from ...utilities.resource_utils import safe_close
1093 safe_close(download_service, "download service")
1095 return Response(
1096 stream_with_context(generate()),
1097 mimetype="text/event-stream",
1098 headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
1099 )
1102@library_bp.route("/api/research-list")
1103@login_required
1104def get_research_list():
1105 """Get list of research sessions for dropdowns."""
1106 username = session["username"]
1107 service = LibraryService(username)
1108 research_list = service.get_research_list_for_dropdown()
1109 return jsonify({"research": research_list})
1112@library_bp.route("/api/sync-library", methods=["POST"])
1113@login_required
1114def sync_library():
1115 """Sync library database with filesystem."""
1116 username = session["username"]
1117 service = LibraryService(username)
1118 stats = service.sync_library_with_filesystem()
1119 return jsonify(stats)
1122@library_bp.route("/api/mark-redownload", methods=["POST"])
1123@login_required
1124@require_json_body()
1125def mark_for_redownload():
1126 """Mark documents for re-download."""
1127 username = session["username"]
1128 service = LibraryService(username)
1130 data = request.json
1131 document_ids = data.get("document_ids", [])
1133 if not document_ids:
1134 return jsonify({"error": "No document IDs provided"}), 400
1136 count = service.mark_for_redownload(document_ids)
1137 return jsonify({"success": True, "marked": count})
1140@library_bp.route("/api/queue-all-undownloaded", methods=["POST"])
1141@login_required
1142def queue_all_undownloaded():
1143 """Queue all articles that haven't been downloaded yet."""
1144 username = session["username"]
1146 logger.info(f"queue_all_undownloaded called for user {username}")
1148 with get_user_db_session(username) as db_session:
1149 # Find all resources that don't have a completed download. Project only
1150 # the columns the filter/loop use (id/url/research_id) instead of full
1151 # ``ResearchResource`` entities so the ``content_preview`` Text column
1152 # isn't materialized for every row on this whole-table scan (#4560).
1153 undownloaded = (
1154 db_session.query(
1155 ResearchResource.id,
1156 ResearchResource.url,
1157 ResearchResource.research_id,
1158 )
1159 .outerjoin(
1160 Document,
1161 (
1162 (ResearchResource.id == Document.resource_id)
1163 | (ResearchResource.document_id == Document.id)
1164 )
1165 & (Document.status == "completed"),
1166 )
1167 .filter(Document.id.is_(None))
1168 .all()
1169 )
1171 logger.info(f"Found {len(undownloaded)} total undownloaded resources")
1173 # Get user password for encrypted database access
1174 user_password = get_authenticated_user_password(username)
1176 resource_filter = ResourceFilter(username, user_password)
1177 filter_results = resource_filter.filter_downloadable_resources(
1178 undownloaded
1179 )
1181 # Get detailed filtering summary
1182 filter_summary = resource_filter.get_filter_summary(undownloaded)
1183 skipped_info = resource_filter.get_skipped_resources_info(undownloaded)
1185 logger.info(f"Filter results: {filter_summary.to_dict()}")
1187 queued_count = 0
1188 research_ids = set()
1189 skipped_count = 0
1191 # Convert filter_results to dict for O(1) lookup instead of O(n²)
1192 filter_results_by_id = {r.resource_id: r for r in filter_results}
1194 for resource in undownloaded:
1195 # Check if resource passed the smart filter
1196 filter_result = filter_results_by_id.get(resource.id)
1198 if not filter_result or not filter_result.can_retry:
1199 skipped_count += 1
1200 if filter_result: 1200 ↛ 1201line 1200 didn't jump to line 1201 because the condition on line 1200 was never true
1201 logger.debug(
1202 f"Skipping resource {resource.id} due to retry policy: {filter_result.reason}"
1203 )
1204 else:
1205 logger.debug(
1206 f"Skipping resource {resource.id} - no filter result available"
1207 )
1208 continue
1210 # Check if it's downloadable using proper URL parsing
1211 if not resource.url:
1212 skipped_count += 1
1213 continue
1215 is_downloadable = is_downloadable_domain(resource.url)
1217 # Log what we're checking
1218 if resource.url and "pubmed" in resource.url.lower(): 1218 ↛ 1219line 1218 didn't jump to line 1219 because the condition on line 1218 was never true
1219 logger.info(f"Found PubMed URL: {resource.url[:100]}")
1221 if not is_downloadable: 1221 ↛ 1222line 1221 didn't jump to line 1222 because the condition on line 1221 was never true
1222 skipped_count += 1
1223 logger.debug(
1224 f"Skipping non-downloadable URL: {resource.url[:100] if resource.url else 'None'}"
1225 )
1226 continue
1228 # Check if already in queue (any status)
1229 existing_queue = (
1230 db_session.query(LibraryDownloadQueue)
1231 .filter_by(resource_id=resource.id)
1232 .first()
1233 )
1235 if existing_queue:
1236 # Reset a non-pending row back to PENDING so it is retried,
1237 # but never one a concurrent download stream has claimed:
1238 # PROCESSING satisfies `!= PENDING`, so an unguarded reset
1239 # un-claims an in-flight row and lets a second stream
1240 # re-download the same resource (issue #4691). The query above
1241 # selects in-flight rows, since a download that has not
1242 # finished has no completed Document yet. The guard lives
1243 # inside the UPDATE, not in an `if` above it, so the check and
1244 # the write cannot straddle another stream's claim.
1245 reset = (
1246 db_session.query(LibraryDownloadQueue)
1247 .filter(
1248 LibraryDownloadQueue.id == existing_queue.id,
1249 LibraryDownloadQueue.status != DocumentStatus.PENDING,
1250 LibraryDownloadQueue.status
1251 != DocumentStatus.PROCESSING,
1252 )
1253 .update(
1254 {
1255 LibraryDownloadQueue.status: DocumentStatus.PENDING,
1256 LibraryDownloadQueue.completed_at: None,
1257 },
1258 synchronize_session=False,
1259 )
1260 )
1261 # Counted either way, matching the pre-fix behaviour: a row
1262 # already PENDING, a row just reset, and a row another stream
1263 # is downloading are all "queued" to the caller.
1264 queued_count += 1
1265 research_ids.add(resource.research_id)
1266 if reset:
1267 logger.debug(
1268 f"Reset queue entry for resource {resource.id} to pending"
1269 )
1270 else:
1271 logger.debug(
1272 f"Resource {resource.id} already pending or in flight"
1273 )
1274 else:
1275 # Add new entry to queue
1276 queue_entry = LibraryDownloadQueue(
1277 resource_id=resource.id,
1278 research_id=resource.research_id,
1279 priority=0,
1280 status=DocumentStatus.PENDING,
1281 )
1282 db_session.add(queue_entry)
1283 queued_count += 1
1284 research_ids.add(resource.research_id)
1285 logger.debug(
1286 f"Added new queue entry for resource {resource.id}"
1287 )
1289 db_session.commit()
1291 logger.info(
1292 f"Queued {queued_count} articles for download, skipped {skipped_count} resources (including {filter_summary.permanently_failed_count} permanently failed and {filter_summary.temporarily_failed_count} temporarily failed)"
1293 )
1295 # Note: Removed synchronous download processing here to avoid blocking the HTTP request
1296 # Downloads will be processed via the SSE streaming endpoint or background tasks
1298 return jsonify(
1299 {
1300 "success": True,
1301 "queued": queued_count,
1302 "research_ids": list(research_ids),
1303 "total_undownloaded": len(undownloaded),
1304 "skipped": skipped_count,
1305 "filter_summary": filter_summary.to_dict(),
1306 "skipped_details": skipped_info,
1307 }
1308 )
1311@library_bp.route("/api/get-research-sources/<research_id>", methods=["GET"])
1312@login_required
1313def get_research_sources(research_id):
1314 """Get all sources for a research with snippets."""
1315 username = session["username"]
1317 sources = []
1318 with get_user_db_session(username) as db_session:
1319 # Get all resources for this research
1320 resources = (
1321 db_session.query(ResearchResource)
1322 .filter_by(research_id=research_id)
1323 .order_by(ResearchResource.created_at)
1324 .all()
1325 )
1327 for idx, resource in enumerate(resources, 1):
1328 # Check if document exists
1329 document = get_document_for_resource(db_session, resource)
1331 # Get domain from URL
1332 domain = ""
1333 if resource.url:
1334 try:
1335 from urllib.parse import urlparse
1337 domain = urlparse(resource.url).hostname or ""
1338 except (ValueError, AttributeError):
1339 # urlparse can raise ValueError for malformed URLs
1340 pass
1342 source_data = {
1343 "number": idx,
1344 "resource_id": resource.id,
1345 "url": resource.url,
1346 "title": resource.title or f"Source {idx}",
1347 "snippet": resource.content_preview or "",
1348 "domain": domain,
1349 "relevance_score": getattr(resource, "relevance_score", None),
1350 "downloaded": False,
1351 "document_id": None,
1352 "file_type": None,
1353 }
1355 if document and document.status == "completed":
1356 source_data.update(
1357 {
1358 "downloaded": True,
1359 "document_id": document.id,
1360 "file_type": document.file_type,
1361 "download_date": document.created_at.isoformat()
1362 if document.created_at
1363 else None,
1364 }
1365 )
1367 sources.append(source_data)
1369 return jsonify({"success": True, "sources": sources, "total": len(sources)})
1372@library_bp.route("/api/check-downloads", methods=["POST"])
1373@login_required
1374@require_json_body()
1375def check_downloads():
1376 """Check download status for a list of URLs."""
1377 username = session["username"]
1378 data = request.json
1379 research_id = data.get("research_id")
1380 urls = data.get("urls", [])
1382 if not research_id or not urls:
1383 return jsonify({"error": "Missing research_id or urls"}), 400
1385 download_status = {}
1387 with get_user_db_session(username) as db_session:
1388 # Get all resources for this research
1389 resources = (
1390 db_session.query(ResearchResource)
1391 .filter_by(research_id=research_id)
1392 .filter(ResearchResource.url.in_(urls))
1393 .all()
1394 )
1396 for resource in resources:
1397 # Check if document exists
1398 document = get_document_for_resource(db_session, resource)
1400 if document and document.status == "completed":
1401 download_status[resource.url] = {
1402 "downloaded": True,
1403 "document_id": document.id,
1404 "file_type": document.file_type,
1405 "title": document.title or resource.title,
1406 }
1407 else:
1408 download_status[resource.url] = {
1409 "downloaded": False,
1410 "resource_id": resource.id,
1411 }
1413 return jsonify({"download_status": download_status})
1416@library_bp.route("/api/download-source", methods=["POST"])
1417@login_required
1418@require_json_body()
1419def download_source():
1420 """Download a single source from a research."""
1421 username = session["username"]
1422 user_password = get_authenticated_user_password(username)
1423 data = request.json
1424 research_id = data.get("research_id")
1425 url = data.get("url")
1427 if not research_id or not url:
1428 return jsonify({"error": "Missing research_id or url"}), 400
1430 # Check if URL is downloadable
1431 if not is_downloadable_domain(url):
1432 return jsonify({"error": "URL is not from a downloadable domain"}), 400
1434 with get_user_db_session(username) as db_session:
1435 # Find the resource
1436 resource = (
1437 db_session.query(ResearchResource)
1438 .filter_by(research_id=research_id, url=url)
1439 .first()
1440 )
1442 if not resource:
1443 return jsonify({"error": "Resource not found"}), 404
1445 # Check if already downloaded
1446 existing = get_document_for_resource(db_session, resource)
1448 if existing and existing.status == "completed":
1449 return jsonify(
1450 {
1451 "success": True,
1452 "message": "Already downloaded",
1453 "document_id": existing.id,
1454 }
1455 )
1457 # Add to download queue
1458 queue_entry = (
1459 db_session.query(LibraryDownloadQueue)
1460 .filter_by(resource_id=resource.id)
1461 .first()
1462 )
1464 if not queue_entry: 1464 ↛ 1465line 1464 didn't jump to line 1465 because the condition on line 1464 was never true
1465 queue_entry = LibraryDownloadQueue(
1466 resource_id=resource.id,
1467 research_id=resource.research_id,
1468 priority=1, # Higher priority for manual downloads
1469 status=DocumentStatus.PENDING,
1470 )
1471 db_session.add(queue_entry)
1472 db_session.commit()
1473 else:
1474 # Reset the row so this manual download can claim it, but never
1475 # one a concurrent stream is already downloading: the reset is
1476 # followed immediately by a download in this request thread, so
1477 # un-claiming an in-flight row starts a second download of the
1478 # same resource (issue #4691). The already-downloaded early return
1479 # above cannot prevent that, because a download still in flight
1480 # has not written a completed Document yet.
1481 reset = (
1482 db_session.query(LibraryDownloadQueue)
1483 .filter(
1484 LibraryDownloadQueue.id == queue_entry.id,
1485 LibraryDownloadQueue.status != DocumentStatus.PROCESSING,
1486 )
1487 .update(
1488 {
1489 LibraryDownloadQueue.status: DocumentStatus.PENDING,
1490 LibraryDownloadQueue.priority: 1,
1491 },
1492 synchronize_session=False,
1493 )
1494 )
1495 db_session.commit()
1496 if not reset:
1497 return jsonify(
1498 {
1499 "success": False,
1500 "message": "Download already in progress",
1501 }
1502 ), 409
1504 queue_item_id = queue_entry.id
1506 # Claim the row before downloading, exactly as download_bulk does, so
1507 # a bulk stream that reaches this row between the reset above and the
1508 # download below cannot also process it (issue #4691).
1509 if not _claim_download_queue_item(db_session, queue_item_id): 1509 ↛ 1510line 1509 didn't jump to line 1510 because the condition on line 1509 was never true
1510 return jsonify(
1511 {"success": False, "message": "Download already in progress"}
1512 ), 409
1514 # Start download immediately
1515 try:
1516 with DownloadService(username, user_password) as service:
1517 success, message = service.download_resource(resource.id)
1518 except Exception:
1519 safe_rollback(db_session, "download_source")
1520 # Release the claim so a later run can retry this row: the
1521 # download raised before download_resource recorded a terminal
1522 # status (issue #4691).
1523 _release_download_queue_item(db_session, queue_item_id)
1524 raise
1526 # download_resource writes a terminal status for the rows it actually
1527 # downloads, so this is a no-op for those; it fires on the paths that
1528 # don't (its already-downloaded early return), which would otherwise
1529 # strand the claim in PROCESSING.
1530 _finalize_download_queue_item(db_session, queue_item_id, success)
1532 if success:
1533 return jsonify({"success": True, "message": "Download completed"})
1534 # Log internal message, but show only generic message to user
1535 return jsonify({"success": False, "message": "Download failed"})