Coverage for src/local_deep_research/web/routers/library.py: 96%
652 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +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"""
10from fastapi import APIRouter, Depends, Request
11from fastapi.responses import (
12 HTMLResponse,
13 JSONResponse,
14 Response,
15)
16from ..dependencies.auth import require_auth
17from ..dependencies.threadpool import (
18 WorkerCleanupStreamingResponse,
19 run_db_sync,
20)
21from ..utils.range_response import in_memory_range_response
22from ..template_config import templates
23from ..utils.download_headers import build_content_disposition
24from ..utils.pagination import OFFSET_LIMIT_ERROR, parse_bounded_offset
26import os
27import json
28from datetime import UTC, datetime
29from pathlib import Path
31from loguru import logger
33from ...database.session_context import get_user_db_session
34from ...database.models.research import ResearchResource
35from ...database.models.library import (
36 Document as Document,
37 DocumentStatus,
38 DownloadQueue as LibraryDownloadQueue,
39 Collection,
40)
41from ...library.download_management import ResourceFilter
42from ...security.url_validator import URLValidator
43from ...research_library.services.download_service import DownloadService
44from ...research_library.services.library_service import LibraryService
45from ...research_library.services.pdf_storage_manager import (
46 PDFStorageManager,
47 resolve_pdf_storage_mode,
48)
49from ...security import failure_reason_for_log, redact_url_for_log
50from ...research_library.utils import (
51 apply_user_subdir,
52 document_awaits_pdf_clause,
53 document_counts_as_downloaded,
54 download_claim_is_live,
55 get_document_for_resource,
56 resource_document_clause,
57 handle_api_error,
58 is_downloadable_domain,
59 is_downloadable_url,
60)
61from ...utilities.db_utils import get_settings_manager
62from ...config.paths import get_library_directory
63from ...security.log_sanitizer import redact_and_bound_for_log
64from ...settings.exceptions import SettingsValueError
65from ...web.exceptions import AuthenticationRequiredError
66from ..dependencies.json_body import read_json_dict
67from .notes import _log_value_preview
68from typing import Annotated
70# Create the router
71router = APIRouter(prefix="/library", tags=["library"])
73#: Upper bound for user-supplied id/url lists on the manual-selection
74#: routes (check-downloads, mark-redownload). Far above any real bulk
75#: selection and a safe margin under SQLite's 250,000 bind-parameter
76#: ceiling, so an oversized list can never reach the .in_() filters.
77_MAX_STRING_LIST_ITEMS = 1000
78#: Higher bound for download-bulk only: the locked download-all
79#: contract (test_library_api_response_shapes) requires lists above
80#: one thousand to stream, so the functional floor sits above the
81#: manual cap. download_bulk never binds these ids into a SQL query —
82#: it loops per id, opening a session and queueing downloads for each
83#: — so the real cost axis is per-item work, not a bind-parameter
84#: ceiling. 100,000 bounds that per-item work to a size the "queue
85#: everything undownloaded" flow (queue-all-undownloaded's
86#: research_ids) can legitimately reach.
87_MAX_BULK_LIST_ITEMS = 100_000
88#: Upper bound per item; 2048 characters covers any URL or id the
89#: UI ever sends while bounding per-request memory.
90_MAX_STRING_LIST_ITEM_LENGTH = 2048
93def _string_list_error(
94 value, field_name: str, max_items: int = _MAX_STRING_LIST_ITEMS
95) -> JSONResponse | None:
96 """Return a 400 response unless *value* is a bounded non-empty-string list.
98 Caps both the item count and per-item length. Only
99 ``check-downloads`` feeds the list into a SQLAlchemy ``.in_()``
100 filter, whose bind-parameter ceiling this helper's default
101 ``max_items`` stays a safe margin under; ``download-bulk`` and
102 ``mark-redownload`` loop over the list per item instead, so their
103 exposure is unbounded per-item work rather than a bind-parameter
104 crash. Routes that must preserve larger lists (download-all) pass
105 a higher ``max_items`` sized to that per-item cost instead of the
106 bind ceiling.
108 The count check runs before the per-item type/length sweeps so an
109 oversized list 400s in O(1) instead of paying an O(n) scan first.
110 """
111 if not isinstance(value, list):
112 return JSONResponse(
113 {"error": f"{field_name} must be a list of non-empty strings"},
114 status_code=400,
115 )
116 if len(value) > max_items:
117 return JSONResponse(
118 {"error": (f"{field_name} must contain at most {max_items} items")},
119 status_code=400,
120 )
121 if any(not isinstance(item, str) or not item.strip() for item in value):
122 return JSONResponse(
123 {"error": f"{field_name} must be a list of non-empty strings"},
124 status_code=400,
125 )
126 if any(len(item) > _MAX_STRING_LIST_ITEM_LENGTH for item in value):
127 return JSONResponse(
128 {
129 "error": (
130 f"{field_name} items must be at most"
131 f" {_MAX_STRING_LIST_ITEM_LENGTH} characters"
132 )
133 },
134 status_code=400,
135 )
136 return None
139# NOTE: Routes use username (not .get()) intentionally.
140# Depends(require_auth) guarantees the key exists; direct access fails
141# fast if the dependency is ever removed.
144# Error handler for authentication errors
145def handle_web_api_exception(error):
146 """Handle WebAPIException and its subclasses."""
147 from ...web.exceptions import WebAPIException
149 if isinstance(error, WebAPIException):
150 return error.to_dict(), error.status_code
151 # Re-raise other exceptions
152 raise error
155def get_authenticated_user_password(
156 username: str, session_id: str | None = None
157) -> str:
158 """
159 Get authenticated user password from session store.
161 Args:
162 username: The username to get password for
163 session_id: Optional session ID for direct lookup.
165 Returns:
166 str: The user's password
168 Raises:
169 AuthenticationRequiredError: If no password is available for the user
170 """
171 from ...database.session_passwords import session_password_store
173 # Try specific session first
174 if session_id:
175 try:
176 user_password = session_password_store.get_session_password(
177 username, session_id
178 )
179 if user_password:
180 return user_password
181 except Exception:
182 logger.exception("Failed to get user password from session store")
184 # Try any session for this user (works without Flask)
185 user_password = session_password_store.get_any_session_password(username)
186 if user_password:
187 return user_password
189 # No password available
190 logger.error(f"No user password available for user {username}")
191 raise AuthenticationRequiredError(
192 message="Authentication required: Please refresh the page and log in again to access encrypted database features.",
193 )
196# ============= Page Routes =============
199@router.get("/")
200def library_page(
201 request: Request,
202 username: Annotated[str, Depends(require_auth)],
203):
204 """Main library page showing downloaded documents."""
205 import math
207 try:
208 with get_user_db_session(username) as db_session:
209 settings = get_settings_manager(db_session, username)
210 pdf_storage_mode = resolve_pdf_storage_mode(
211 settings.get_setting(
212 "research_library.pdf_storage_mode", "database"
213 )
214 )
215 shared_library = settings.get_setting(
216 "research_library.shared_library", False
217 )
218 except SettingsValueError:
219 # An invalid configured value must not render the page as if the
220 # defaults were configured. The app's handler shows the diagnostic.
221 raise
222 except Exception:
223 pdf_storage_mode = "database"
224 shared_library = False
226 enable_pdf_storage = pdf_storage_mode != "none"
228 # Filters
229 domain_filter = request.query_params.get("domain")
230 research_filter = request.query_params.get("research")
231 collection_filter = request.query_params.get("collection")
232 date_filter = request.query_params.get("date")
234 # Build context with graceful degradation
235 stats = {}
236 documents = []
237 unique_domains = []
238 research_list = []
239 collections = []
240 page = 1
241 total_pages = 1
242 # Degrading gracefully is deliberate here (main let the exception
243 # reach Flask's 500 handler), but an empty `documents` list is
244 # ambiguous: the template's empty state would tell a user whose
245 # library failed to load that they have no documents yet, and invite
246 # them to go download some. Track *why* the list is empty so the page
247 # can say "we could not load this" instead of asserting something
248 # false about their data.
249 load_error = False
251 try:
252 service = LibraryService(username)
253 stats = service.get_library_stats()
255 from ...database.library_init import get_default_library_id
257 resolved_collection = collection_filter or get_default_library_id(
258 username
259 )
261 # Pagination
262 per_page = 100
263 total_docs = service.count_documents(
264 research_id=research_filter,
265 domain=domain_filter,
266 collection_id=resolved_collection,
267 date_filter=date_filter,
268 )
269 total_pages = max(1, math.ceil(total_docs / per_page))
270 try:
271 page = int(request.query_params.get("page", "1"))
272 except ValueError:
273 page = 1
274 page = max(1, min(page, total_pages))
275 offset = (page - 1) * per_page
277 documents = service.get_documents(
278 research_id=research_filter,
279 domain=domain_filter,
280 collection_id=resolved_collection,
281 date_filter=date_filter,
282 limit=per_page,
283 offset=offset,
284 )
285 unique_domains = service.get_unique_domains()
286 research_list = service.get_research_list_for_dropdown()
287 collections = service.get_all_collections()
288 except Exception:
289 logger.exception("Error loading library data")
290 load_error = True
292 default_collection_id = next(
293 (c["id"] for c in collections if c.get("is_default")), None
294 )
296 return templates.TemplateResponse(
297 request=request,
298 name="pages/library.html",
299 context={
300 "stats": stats,
301 "documents": documents,
302 "load_error": load_error,
303 "unique_domains": unique_domains,
304 "research_list": research_list,
305 "collections": collections,
306 "selected_collection": collection_filter,
307 "default_collection_id": default_collection_id,
308 "storage_path": stats.get("storage_path", ""),
309 "enable_pdf_storage": enable_pdf_storage,
310 "pdf_storage_mode": pdf_storage_mode,
311 "shared_library": shared_library,
312 "page": page,
313 "total_pages": total_pages,
314 "selected_date": date_filter,
315 "selected_research": research_filter,
316 "selected_domain": domain_filter,
317 },
318 )
321@router.get("/document/{document_id}")
322def document_details_page(
323 request: Request,
324 document_id,
325 username: Annotated[str, Depends(require_auth)],
326):
327 """Document details page showing all metadata and links."""
328 service = LibraryService(username)
330 # Get document details
331 document = service.get_document_by_id(document_id)
333 if not document:
334 # Browser navigation: return text/html, not JSON. These four page
335 # routes are reached from library.html as ordinary <a href> links, so a
336 # stale link showed the user a raw {"error": ...} body in the browser's
337 # JSON viewer. main returned `"Document not found", 404` (text/html)
338 # here; the sibling /api/ routes keep JSON.
339 return HTMLResponse("Document not found", status_code=404)
341 # HTML escaping does not make a stored javascript:/data: URL safe to
342 # navigate. Keep rejected values visible as text, but never as links.
343 original_url = document.get("original_url")
344 original_source_url = (
345 original_url if URLValidator.is_safe_http_link(original_url) else None
346 )
348 return templates.TemplateResponse(
349 request=request,
350 name="pages/document_details.html",
351 context={
352 "request": request,
353 "document": document,
354 "original_source_url": original_source_url,
355 },
356 )
359@router.get("/download-manager")
360def download_manager_page(
361 request: Request,
362 username: Annotated[str, Depends(require_auth)],
363):
364 """Download manager page for selecting and downloading research PDFs."""
365 import math
367 service = LibraryService(username)
368 with get_user_db_session(username) as db_session:
369 settings = get_settings_manager(db_session, username)
370 pdf_storage_mode = resolve_pdf_storage_mode(
371 settings.get_setting(
372 "research_library.pdf_storage_mode", "database"
373 )
374 )
375 shared_library = settings.get_setting(
376 "research_library.shared_library", False
377 )
378 enable_pdf_storage = pdf_storage_mode != "none"
380 # Summary stats over ALL sessions (also used for page count)
381 per_page = 50
382 summary = service.get_download_manager_summary_stats()
383 total_pages = max(1, math.ceil(summary["total_researches"] / per_page))
385 # Pagination with upper-bound clamp
386 try:
387 page = int(request.query_params.get("page", "1"))
388 except ValueError:
389 page = 1
390 page = max(1, min(page, total_pages))
391 offset = (page - 1) * per_page
393 # Get paginated research sessions
394 research_list = service.get_research_list_with_stats(
395 limit=per_page, offset=offset
396 )
398 # Batch-fetch PDF previews and domain breakdowns (single query)
399 research_ids = [r["id"] for r in research_list]
400 previews = service.get_pdf_previews_batch(research_ids)
401 for research in research_list:
402 rid = research["id"]
403 data = previews.get(rid, {"pdf_sources": [], "domains": {}})
404 research["pdf_sources"] = data["pdf_sources"]
405 research["domains"] = data["domains"]
407 return templates.TemplateResponse(
408 request=request,
409 name="pages/download_manager.html",
410 context={
411 "request": request,
412 "research_list": research_list,
413 "total_researches": summary["total_researches"],
414 "total_resources": summary["total_resources"],
415 "already_downloaded": summary["already_downloaded"],
416 "available_to_download": summary["available_to_download"],
417 "enable_pdf_storage": enable_pdf_storage,
418 "pdf_storage_mode": pdf_storage_mode,
419 "shared_library": shared_library,
420 "page": page,
421 "total_pages": total_pages,
422 },
423 )
426# ============= API Routes =============
429@router.get("/api/stats")
430def get_library_stats(
431 request: Request, username: Annotated[str, Depends(require_auth)]
432):
433 """Get library statistics."""
434 service = LibraryService(username)
435 return service.get_library_stats()
438@router.get("/api/collections/list")
439def get_collections_list(
440 request: Request, username: Annotated[str, Depends(require_auth)]
441):
442 """Get list of all collections for dropdown selection."""
444 with get_user_db_session(username) as db_session:
445 collections = (
446 db_session.query(Collection).order_by(Collection.name).all()
447 )
449 return {
450 "success": True,
451 "collections": [
452 {
453 "id": col.id,
454 "name": col.name,
455 "description": col.description,
456 }
457 for col in collections
458 ],
459 }
462@router.get("/api/documents")
463def get_documents(
464 request: Request, username: Annotated[str, Depends(require_auth)]
465):
466 """Get documents with filtering."""
467 service = LibraryService(username)
469 # Get filter parameters
470 research_id = request.query_params.get("research_id")
471 domain = request.query_params.get("domain")
472 file_type = request.query_params.get("file_type")
473 favorites_only = request.query_params.get("favorites") == "true"
474 search_query = request.query_params.get("search")
475 # Clamp pagination. SQLite treats LIMIT -1 as "no limit", so an
476 # unclamped negative limit (e.g. ?limit=-1) would bypass pagination and
477 # load the whole collection — including large Document.text_content
478 # bodies — into memory (#4560). Mirrors the clamp in history_routes.
479 try:
480 limit = max(1, min(int(request.query_params.get("limit", 100)), 1000))
481 except (TypeError, ValueError):
482 limit = 100
483 try:
484 offset = parse_bounded_offset(request.query_params.get("offset"))
485 except ValueError:
486 return JSONResponse({"error": OFFSET_LIMIT_ERROR}, status_code=400)
488 documents = service.get_documents(
489 research_id=research_id,
490 domain=domain,
491 file_type=file_type,
492 favorites_only=favorites_only,
493 search_query=search_query,
494 limit=limit,
495 offset=offset,
496 )
498 return {"documents": documents}
501@router.post("/api/document/{document_id}/favorite")
502def toggle_favorite(
503 request: Request,
504 document_id,
505 username: Annotated[str, Depends(require_auth)],
506):
507 """Toggle favorite status of a document."""
508 service = LibraryService(username)
509 is_favorite = service.toggle_favorite(document_id)
510 return {"favorite": is_favorite}
513@router.get("/api/document/{document_id}/pdf-url")
514def get_pdf_url(
515 request: Request,
516 document_id,
517 username: Annotated[str, Depends(require_auth)],
518):
519 """Get URL for viewing PDF."""
520 # Return URL that will serve the PDF
521 return {
522 "url": f"/library/api/document/{document_id}/pdf",
523 "title": "Document", # Could fetch actual title
524 }
527def _serve_pdf(
528 request: Request, document_id, username: str, *, html_errors: bool
529):
530 """Shared PDF-loading core for the page route and its /api/ sibling.
532 ``html_errors`` picks the not-found/not-available error shape: the
533 browser-navigation page route wants text/html (a stale <a href> link
534 should not show a raw JSON body), while the /api/ sibling keeps the
535 JSON contract every other /api/ route has -- same rule its /text
536 sibling (serve_text_api) already follows.
537 """
539 def _error(message: str) -> Response:
540 if html_errors:
541 return HTMLResponse(message, status_code=404)
542 return JSONResponse({"error": message}, status_code=404)
544 with get_user_db_session(username) as db_session:
545 # Get document from database
546 document = db_session.query(Document).filter_by(id=document_id).first()
548 if not document:
549 logger.warning(
550 f"Document ID {_log_value_preview(document_id)} not found in database for user {username}"
551 )
552 # Browser navigation: return text/html, not JSON. These four page
553 # routes are reached from library.html as ordinary <a href> links, so a
554 # stale link showed the user a raw {"error": ...} body in the browser's
555 # JSON viewer. main returned `"Document not found", 404` (text/html)
556 # here; the sibling /api/ routes keep JSON.
557 return _error("Document not found")
559 logger.info(
560 f"Document {document_id}: title='{document.title}', "
561 f"file_path={document.file_path}"
562 )
564 # Get settings for PDF storage manager
565 settings = get_settings_manager(db_session)
566 storage_mode = settings.get_setting(
567 "research_library.pdf_storage_mode", "none"
568 )
569 # Expand configured environment variables before choosing the user's
570 # library directory, matching the download/storage path resolution.
571 base_root = (
572 Path(
573 os.path.expandvars(
574 settings.get_setting(
575 "research_library.storage_path",
576 str(get_library_directory()),
577 )
578 )
579 )
580 .expanduser()
581 .resolve()
582 )
583 shared_library = settings.get_setting(
584 "research_library.shared_library", False
585 )
586 # Serve only from the requesting user's root. A matching relative
587 # filename in the shared legacy directory does not prove ownership.
588 library_root = apply_user_subdir(base_root, username, shared_library)
590 # Use PDFStorageManager to load PDF (handles database and filesystem)
591 pdf_manager = PDFStorageManager(library_root, storage_mode)
592 pdf_bytes = pdf_manager.load_pdf(document, db_session)
594 if pdf_bytes:
595 logger.info(
596 f"Serving PDF for document {document_id} ({len(pdf_bytes)} bytes)"
597 )
598 filename = document.filename or "document.pdf"
599 # Plain Response, not StreamingResponse: the bytes are already
600 # fully in memory, and iterating a BytesIO yields *lines* — it
601 # splits on every 0x0A, which in binary PDF data occurs roughly
602 # every 256 bytes, so a single view became tens of thousands of
603 # threadpool dispatches and sent no Content-Length.
604 return in_memory_range_response(
605 content=pdf_bytes,
606 request=request,
607 media_type="application/pdf",
608 headers={
609 "Content-Disposition": build_content_disposition(
610 filename, inline=True
611 )
612 },
613 )
615 # No PDF found anywhere
616 logger.warning(f"No PDF available for document {document_id}")
617 # Browser navigation, same rule as the not-found branch above: main
618 # returned `"PDF not available", 404` (text/html) here. Reached
619 # routinely, not just via stale links -- "Remove the PDF file to save
620 # space" deletes the blob while leaving the row, so an open tab or a
621 # bookmark lands here. The /api/ sibling keeps JSON.
622 return _error("PDF not available")
625@router.get("/document/{document_id}/pdf")
626def view_pdf_page(
627 request: Request,
628 document_id,
629 username: Annotated[str, Depends(require_auth)],
630):
631 """Page for viewing PDF file - uses PDFStorageManager for retrieval."""
632 return _serve_pdf(request, document_id, username, html_errors=True)
635@router.get("/api/document/{document_id}/pdf")
636def serve_pdf_api(
637 request: Request,
638 document_id,
639 username: Annotated[str, Depends(require_auth)],
640):
641 """API endpoint for serving PDF file (kept for backward compatibility).
643 Unlike the page route above, a missing document or missing PDF blob
644 here answers JSON, not text/html -- this is an /api/ route, and its
645 /api/document/{id}/text sibling (serve_text_api) already keeps that
646 contract on the same two error branches.
647 """
648 return _serve_pdf(request, document_id, username, html_errors=False)
651@router.get("/document/{document_id}/txt")
652def view_text_page(
653 request: Request,
654 document_id,
655 username: Annotated[str, Depends(require_auth)],
656):
657 """Page for viewing text content."""
659 with get_user_db_session(username) as db_session:
660 # Get document by ID (text now stored in Document.text_content)
661 document = db_session.query(Document).filter_by(id=document_id).first()
663 if not document:
664 logger.warning(
665 f"Document not found for document ID {_log_value_preview(document_id)}"
666 )
667 # Browser navigation: return text/html, not JSON. These four page
668 # routes are reached from library.html as ordinary <a href> links, so a
669 # stale link showed the user a raw {"error": ...} body in the browser's
670 # JSON viewer. main returned `"Document not found", 404` (text/html)
671 # here; the sibling /api/ routes keep JSON.
672 return HTMLResponse("Document not found", status_code=404)
674 if not document.text_content:
675 logger.warning(f"Document {document_id} has no text content")
676 # Browser navigation, same rule as the not-found branch above:
677 # main returned `"Text content not available", 404` (text/html)
678 # here. serve_text_api below keeps main's jsonify() shape.
679 return HTMLResponse("Text content not available", status_code=404)
681 logger.info(
682 f"Serving text content for document {document_id}: {len(document.text_content)} characters"
683 )
685 # Render as HTML page
686 return templates.TemplateResponse(
687 request=request,
688 name="pages/document_text.html",
689 context={
690 "document_id": document_id,
691 "title": document.title or "Document Text",
692 "text_content": document.text_content,
693 "extraction_method": document.extraction_method,
694 "word_count": document.word_count,
695 },
696 )
699@router.get("/api/document/{document_id}/text")
700def serve_text_api(
701 request: Request,
702 document_id,
703 username: Annotated[str, Depends(require_auth)],
704):
705 """API endpoint for serving text content (kept for backward compatibility)."""
707 with get_user_db_session(username) as db_session:
708 # Get document by ID (text now stored in Document.text_content)
709 document = db_session.query(Document).filter_by(id=document_id).first()
711 if not document:
712 logger.warning(
713 f"Document not found for document ID {_log_value_preview(document_id)}"
714 )
715 return JSONResponse(
716 {"error": "Document not found"}, status_code=404
717 )
719 if not document.text_content:
720 logger.warning(f"Document {document_id} has no text content")
721 return JSONResponse(
722 {"error": "Text content not available"}, status_code=404
723 )
725 logger.info(
726 f"Serving text content for document {document_id}: {len(document.text_content)} characters"
727 )
729 return {
730 "text_content": document.text_content,
731 "title": document.title or "Document",
732 "extraction_method": document.extraction_method,
733 "word_count": document.word_count,
734 }
737@router.post("/api/open-folder")
738def open_folder(
739 request: Request, username: Annotated[str, Depends(require_auth)]
740):
741 """Open folder containing a document.
743 Security: This endpoint is disabled for server deployments.
744 It only makes sense for desktop usage where the server and client are on the same machine.
745 """
746 return JSONResponse(
747 {
748 "status": "error",
749 "message": "This feature is disabled. It is only available in desktop mode.",
750 },
751 status_code=403,
752 )
755@router.post("/api/download/{resource_id}")
756def download_single_resource(
757 request: Request,
758 # Typed int, matching Flask's <int:resource_id> converter: without it a
759 # non-numeric segment is passed straight through to the service layer
760 # instead of being rejected as a 422 by the router.
761 resource_id: int,
762 username: Annotated[str, Depends(require_auth)],
763):
764 """Download a single resource."""
765 user_password = get_authenticated_user_password(username)
767 with DownloadService(username, user_password) as service:
768 success, error = service.download_resource(resource_id)
769 if success:
770 return {"success": True}
771 if error == "Resource not found":
772 # download_resource() returns this exact string (never raises)
773 # when resource_id doesn't exist — that's a 404, not a generic
774 # download failure. Without this branch every nonexistent
775 # resource_id collapsed to the 500 below.
776 return JSONResponse(
777 {"success": False, "error": "Resource not found"},
778 status_code=404,
779 )
780 # The reason may be exception text that keeps the URL's path and
781 # query; download_service logs the cause itself where it is caught.
782 logger.warning(
783 f"Download failed for resource {resource_id}: "
784 f"{failure_reason_for_log(error)}"
785 )
786 return JSONResponse(
787 {
788 "success": False,
789 "error": "Download failed. Please try again or contact support.",
790 },
791 status_code=500,
792 )
795@router.post("/api/download-text/{resource_id}")
796def download_text_single(
797 request: Request,
798 # Typed int, matching Flask's <int:resource_id> converter: without it a
799 # non-numeric segment is passed straight through to the service layer
800 # instead of being rejected as a 422 by the router.
801 resource_id: int,
802 username: Annotated[str, Depends(require_auth)],
803):
804 """Download a single resource as text file."""
805 try:
806 user_password = get_authenticated_user_password(username)
808 with DownloadService(username, user_password) as service:
809 success, error = service.download_as_text(resource_id)
811 # Sanitize error message - don't expose internal details
812 if not success:
813 if error:
814 logger.warning(
815 f"Download as text failed for resource {resource_id}: "
816 f"{failure_reason_for_log(error)}"
817 )
818 return {
819 "success": False,
820 "error": "Failed to download resource",
821 }
823 return {"success": True, "error": None}
824 except AuthenticationRequiredError:
825 raise # Let the global WebAPIException handler return 401
826 except Exception as e:
827 return handle_api_error(
828 f"downloading resource {resource_id} as text", e
829 )
832@router.post("/api/download-all-text")
833def download_all_text(
834 request: Request, username: Annotated[str, Depends(require_auth)]
835):
836 """Download all undownloaded resources as text files."""
837 # Capture Flask session ID to avoid scoping issues in nested function
838 session_id = request.session.get("session_id")
840 def generate():
841 # Get user password for database operations
842 try:
843 user_password = get_authenticated_user_password(
844 username, session_id
845 )
846 except AuthenticationRequiredError:
847 logger.warning(
848 f"Authentication unavailable for user {username} - password not in session store"
849 )
850 yield f"data: {json.dumps({'progress': 0, 'current': 0, 'total': 0, 'error': 'Authentication required', 'complete': True})}\n\n"
851 return
853 # This DownloadService (and the SettingsManager session inside
854 # download_service.settings) is built ONCE, before the loop, and
855 # lives across every ``yield`` below. That crosses a cleanup
856 # boundary the wrapper's per-chunk cleanup does NOT know about:
857 # ``WorkerCleanupStreamingResponse`` wraps this generator in
858 # ``iterate_sync_with_cleanup``, whose ``thread_cleanup()`` runs
859 # after EVERY ``next()`` (i.e. after every chunk, not just at the
860 # end) and closes every session cached or tracked on this thread
861 # (``utilities/db_utils.py``). The first chunk that touches
862 # ``download_service.settings`` therefore gets its underlying
863 # SQLAlchemy session closed by that per-chunk cleanup the moment the
864 # chunk's ``yield`` returns -- but the ``SettingsManager``/``Session``
865 # OBJECTS themselves are untouched, only their pooled connection is
866 # released, so the next chunk's use of ``download_service.settings``
867 # transparently reopens a fresh connection (ordinary SQLAlchemy
868 # session behavior after ``close()``). That connection then sits in
869 # neither the LRU cache nor the owner-thread registry until stream
870 # end, when ``safe_close(download_service, ...)`` below closes it
871 # via ``SettingsManager.close()`` -- exactly where it was released
872 # pre-#6095, so this is NOT a leak, just a "cleanup at every
873 # boundary" invariant violated for the longest-lived object in this
874 # stream. Documented rather than restructured: giving this loop its
875 # own re-created-per-chunk settings manager would fix the invariant
876 # but changes this route's session lifetime shape for no behavioral
877 # gain, and isn't warranted without a demonstrated leak.
878 download_service = DownloadService(username, user_password)
879 try:
880 # Fetch the resource rows in a SHORT-LIVED session, snapshotting
881 # only the columns the loop uses into plain tuples. The download
882 # loop below holds NO session while it yields — the held session is
883 # only needed for this query, and keeping a get_user_db_session
884 # scope open across a yield risks the Starlette/anyio thread-
885 # affinity corruption fixed in download_bulk (its __enter__ and
886 # __exit__ can land on different pooled threads). Project only
887 # id/url/title so the content_preview Text column doesn't
888 # materialize on this whole-table scan (#4560).
889 with get_user_db_session(username) as session:
890 all_resources = [
891 (r.id, r.url, r.title)
892 for r in session.query(
893 ResearchResource.id,
894 ResearchResource.url,
895 ResearchResource.title,
896 ).all()
897 ]
899 # Filter to only downloadable resources (academic/PDF)
900 resources = [
901 (rid, url, title)
902 for (rid, url, title) in all_resources
903 if is_downloadable_url(url)
904 ]
906 # Pre-scan directory once to get all existing resource IDs
907 txt_path = Path(download_service.library_root) / "txt"
908 existing_resource_ids = set()
909 if txt_path.exists():
910 # nosemgrep: semgrep.rules.path-traversal-risk, reason: Only enumerate fixed txt names under this authenticated user's resolved library directory
911 for txt_file in txt_path.glob("*.txt"):
912 # Extract resource ID from filename pattern *_{id}.txt
913 parts = txt_file.stem.rsplit("_", 1)
914 if len(parts) == 2: 914 ↛ 911line 914 didn't jump to line 911 because the condition on line 914 was always true
915 try:
916 existing_resource_ids.add(int(parts[1]))
917 except ValueError:
918 pass
920 # Filter resources that need text extraction
921 resources_to_process = [
922 (rid, url, title)
923 for (rid, url, title) in resources
924 if rid not in existing_resource_ids
925 ]
927 total = len(resources_to_process)
928 current = 0
930 logger.info(f"Found {total} resources needing text extraction")
932 for rid, url, title in resources_to_process:
933 current += 1
934 progress = int((current / total) * 100) if total > 0 else 100
936 file_name = title[:50] if title else f"document_{current}.txt"
938 try:
939 # download_as_text opens its own session internally.
940 # Its every error-return path (see download_service.py
941 # download_as_text/_try_api_text_extraction/
942 # _fallback_pdf_extraction) is either a static string or
943 # a fixed token from _client_safe_download_message()
944 # (never str(e) — sanitize_error_for_client() scrubs
945 # credentials only and leaves SQL text intact) — no raw
946 # exception text reaches `error` here.
947 success, error = download_service.download_as_text(rid)
949 if success:
950 status = "success"
951 error_msg = None
952 else:
953 status = "failed"
954 error_msg = error or "Text extraction failed"
956 except Exception as e:
957 # Type name only, no traceback: an exception escaping
958 # the fetch can embed the raw resource URL in its text.
959 logger.warning(
960 f"Error extracting text for resource {rid} "
961 f"({type(e).__name__})"
962 )
963 status = "failed"
964 # CWE-209: only the exception CLASS NAME is surfaced,
965 # never str(e) — same "class name only" mitigation as
966 # journal_quality/downloader.py (CodeQL alerts 7650/7684).
967 error_msg = f"Text extraction failed - {type(e).__name__}"
969 # Send update
970 update = {
971 "progress": progress,
972 "current": current,
973 "total": total,
974 "file": file_name,
975 "url": url, # Add the URL for UI display
976 "status": status,
977 "error": error_msg,
978 }
979 yield f"data: {json.dumps(update)}\n\n"
981 # Send completion
982 yield f"data: {json.dumps({'complete': True, 'total': total})}\n\n"
983 finally:
984 from ...utilities.resource_utils import safe_close
986 safe_close(download_service, "download service")
988 # CWE-209 (CodeQL "Information exposure through an exception"): the SSE
989 # body streamed here never carries a raw exception message — see the
990 # `except Exception as e:` block in generate() above, which only ever
991 # surfaces `type(e).__name__` or a fixed token from
992 # download_service._client_safe_download_message (sanitize_error_for_client
993 # alone would not suffice — it scrubs credentials only, not SQL/paths).
994 return WorkerCleanupStreamingResponse(
995 generate(),
996 media_type="text/event-stream",
997 headers={
998 "Cache-Control": "no-cache, no-transform",
999 "X-Accel-Buffering": "no",
1000 },
1001 )
1004@router.post("/api/download-research/{research_id}")
1005async def download_research_pdfs(
1006 request: Request,
1007 research_id,
1008 username: Annotated[str, Depends(require_auth)],
1009):
1010 """Queue all PDFs from a research session for download."""
1011 user_password = get_authenticated_user_password(username)
1012 data, body_error = await read_json_dict(
1013 request, "simple", "Request body must be valid JSON"
1014 )
1015 if body_error is not None:
1016 return body_error
1017 collection_id = data.get("collection_id")
1019 # `queue_research_downloads` does sync SQLAlchemy work. Keep it off
1020 # the uvicorn event loop.
1021 def _queue_sync():
1022 with DownloadService(username, user_password) as service:
1023 return service.queue_research_downloads(research_id, collection_id)
1025 queued = await run_db_sync(_queue_sync)
1026 return {"success": True, "queued": queued}
1029def _claim_download_queue_item(db_session, queue_item_id):
1030 """Atomically claim a PENDING download-queue row before processing it.
1032 Flip the row from PENDING to PROCESSING in a single conditional UPDATE and
1033 return True only when this caller won the claim. Two concurrent
1034 ``download_bulk`` streams for the same user (the user clicks Download
1035 twice, or a second bulk run starts before the first finishes) both read
1036 the same PENDING rows and, with no claim, both call ``download_resource``
1037 for every row (issue #4691). The row-count guard makes the
1038 PENDING -> PROCESSING transition the single point where exactly one stream
1039 wins each row; the loser sees 0 updated rows and skips it.
1041 The claim also records its time in ``last_attempt_at``. Nothing reclaims
1042 a row left PROCESSING, e.g. by a restart during the download, and the
1043 queueing of a document that awaits its PDF (see ``document_awaits_pdf``)
1044 lets another resource refill it only while no download holds one of its
1045 resources. ``download_claim_is_live`` tells a held row from one a restart
1046 left behind by this time.
1047 """
1048 claimed = (
1049 db_session.query(LibraryDownloadQueue)
1050 .filter_by(id=queue_item_id, status=DocumentStatus.PENDING)
1051 .update(
1052 {
1053 LibraryDownloadQueue.status: DocumentStatus.PROCESSING,
1054 LibraryDownloadQueue.last_attempt_at: datetime.now(UTC),
1055 },
1056 synchronize_session=False,
1057 )
1058 )
1059 db_session.commit()
1060 return claimed == 1
1063def _release_download_queue_item(db_session, queue_item_id):
1064 """Return a claimed row from PROCESSING back to PENDING so it stays
1065 retryable when the download raised before a terminal COMPLETED/FAILED
1066 status was recorded (issue #4691), matching the pre-fix behaviour where an
1067 errored row was left PENDING. Scoped to PROCESSING so it only ever undoes
1068 this stream's own claim.
1069 """
1070 db_session.query(LibraryDownloadQueue).filter_by(
1071 id=queue_item_id, status=DocumentStatus.PROCESSING
1072 ).update(
1073 {LibraryDownloadQueue.status: DocumentStatus.PENDING},
1074 synchronize_session=False,
1075 )
1076 db_session.commit()
1079def _finalize_download_queue_item(db_session, queue_item_id, success):
1080 """Record the outcome of a claimed row that returned without raising.
1082 ``download_resource`` writes a terminal COMPLETED/FAILED status itself for
1083 the rows it actually downloads, and this is a no-op for those: the row is
1084 no longer PROCESSING, so the guard matches nothing and pdf-mode behaviour
1085 is unchanged. It fires only on the paths that otherwise strand a claim
1086 (issue #4691):
1088 - ``mode="text_only"``: ``download_as_text`` and its call tree never touch
1089 ``LibraryDownloadQueue``, so a successful extraction would leave the row
1090 PROCESSING forever. Nothing in ``src/`` reaps PROCESSING rows, and a
1091 successful extraction also writes a COMPLETED Document, which blocks both
1092 the pre-pass reset and ``queue_all_undownloaded``'s outer-join filter.
1093 - ``download_resource``'s already-downloaded early return, which reports
1094 success before reaching its own queue update.
1096 Pre-fix, both paths left the row PENDING and merely re-scanned it, so
1097 failure returns to PENDING rather than FAILED to keep that retryability.
1098 """
1099 db_session.query(LibraryDownloadQueue).filter_by(
1100 id=queue_item_id, status=DocumentStatus.PROCESSING
1101 ).update(
1102 {
1103 LibraryDownloadQueue.status: (
1104 DocumentStatus.COMPLETED if success else DocumentStatus.PENDING
1105 ),
1106 LibraryDownloadQueue.completed_at: (
1107 datetime.now(UTC) if success else None
1108 ),
1109 },
1110 synchronize_session=False,
1111 )
1112 db_session.commit()
1115@router.post("/api/download-bulk")
1116async def download_bulk(
1117 request: Request, username: Annotated[str, Depends(require_auth)]
1118):
1119 """Download PDFs or extract text from multiple research sessions."""
1120 data, body_error = await read_json_dict(
1121 request, "simple", "Request body must be valid JSON"
1122 )
1123 if body_error is not None:
1124 return body_error
1125 research_ids = data.get("research_ids", [])
1126 mode = data.get("mode", "pdf") # pdf or text_only
1127 collection_id = data.get(
1128 "collection_id"
1129 ) # Optional: target collection for downloads
1131 if not research_ids:
1132 return JSONResponse(
1133 {"error": "No research IDs provided"}, status_code=400
1134 )
1135 # Download-all selections legitimately exceed the manual-selection
1136 # cap (locked contract: lists above one thousand must stream), so
1137 # this route uses the higher bulk ceiling — bounded to a size the
1138 # "queue everything undownloaded" flow can legitimately reach, not
1139 # to a SQL bind-parameter limit (this route loops per id rather
1140 # than binding the list into a query).
1141 if error := _string_list_error(
1142 research_ids, "research_ids", max_items=_MAX_BULK_LIST_ITEMS
1143 ):
1144 return error
1145 if not isinstance(mode, str) or mode not in {"pdf", "text_only"}:
1146 return JSONResponse(
1147 {"error": "mode must be either 'pdf' or 'text_only'"},
1148 status_code=400,
1149 )
1151 # Capture Flask session ID to avoid scoping issues in nested function
1152 session_id = request.session.get("session_id")
1154 def generate():
1155 """Generate progress updates as Server-Sent Events."""
1156 # Get user password for database operations
1157 try:
1158 user_password = get_authenticated_user_password(
1159 username, session_id
1160 )
1161 except AuthenticationRequiredError:
1162 logger.warning(
1163 f"Authentication unavailable for user {username} - password not in session store"
1164 )
1165 yield f"data: {json.dumps({'progress': 0, 'current': 0, 'total': 0, 'error': 'Authentication required', 'complete': True})}\n\n"
1166 return
1168 # Same per-chunk-cleanup caveat as download_all_text() above:
1169 # download_service.settings' session gets closed (connection
1170 # released, object untouched) by iterate_sync_with_cleanup's
1171 # per-chunk thread_cleanup() after the first chunk that uses it, and
1172 # transparently reopens a fresh connection on later chunks. Not a
1173 # leak -- safe_close(download_service, ...) below releases it at
1174 # stream end exactly as pre-#6095 -- just documented here rather
1175 # than restructured; see the fuller note there.
1176 download_service = DownloadService(username, user_password)
1177 try:
1178 total = 0
1179 current = 0
1180 queue_failures = 0
1182 # Pre-queue and count in one merged loop. queue_research_downloads
1183 # opens its own session, commits, and handles dedup internally:
1184 # it skips resources that are already PENDING or already backed by
1185 # a COMPLETED Document, and re-queues any remaining row (e.g. a
1186 # prior FAILED attempt) back to PENDING. Calling it BEFORE counting
1187 # ensures the total reflects what will actually be processed
1188 # (issue #4660). In PDF mode it also queues a COMPLETED Document
1189 # that still needs its PDF (see document_awaits_pdf), through
1190 # one resource per such document across all the research
1191 # sessions of this run (run_research_ids); a text extraction
1192 # already has that document's text, so text_only mode skips it.
1193 # The set is built once for the run: each call only looks it up.
1194 run_research_ids = frozenset(research_ids)
1195 for research_id in research_ids:
1196 try:
1197 download_service.queue_research_downloads(
1198 research_id,
1199 collection_id,
1200 text_only=mode == "text_only",
1201 run_research_ids=run_research_ids,
1202 )
1203 except Exception:
1204 queue_failures += 1
1205 logger.exception(
1206 "Error queueing downloads for "
1207 "user={} research={} collection={}",
1208 username,
1209 research_id,
1210 _log_value_preview(collection_id),
1211 )
1213 with get_user_db_session(username) as session:
1214 count = (
1215 session.query(LibraryDownloadQueue)
1216 .filter_by(
1217 research_id=research_id,
1218 status=DocumentStatus.PENDING,
1219 )
1220 .count()
1221 )
1222 total += count
1223 logger.debug(
1224 f"Research {research_id}: {count} pending items in queue"
1225 )
1227 logger.info(f"Total pending downloads across all research: {total}")
1228 yield f"data: {json.dumps({'progress': 0, 'current': 0, 'total': total})}\n\n"
1230 # If the pre-pass yielded nothing to process, surface it as an
1231 # error so the UI alerts instead of silently completing with
1232 # "0 / 0 files" success. Distinguish queue failure from a
1233 # legitimate "nothing left to download" state.
1234 if total == 0:
1235 if queue_failures > 0:
1236 error_msg = (
1237 f"Download failed to start: queueing failed for "
1238 f"{queue_failures} of {len(research_ids)} "
1239 f"research session(s). Check server logs."
1240 )
1241 else:
1242 error_msg = (
1243 "No new papers to download for the selected "
1244 "research session(s)."
1245 )
1246 yield f"data: {json.dumps({'progress': 100, 'current': 0, 'total': 0, 'complete': True, 'error': error_msg})}\n\n"
1247 return
1249 # Process each research
1250 for research_id in research_ids:
1251 # Get queued downloads for this research in a SHORT-LIVED
1252 # session, snapshotting only the ids we need. The per-item work
1253 # below opens its own short-lived sessions so that NO
1254 # get_user_db_session scope is ever held across a ``yield``:
1255 # generate() is a sync generator that Starlette drives via
1256 # iterate_in_threadpool (one anyio dispatch per next()), so a
1257 # session whose __enter__/__exit__ straddled a yield could run
1258 # on two different pooled OS threads and leave the thread-local
1259 # scope_depth counter permanently corrupted.
1260 with get_user_db_session(username) as session:
1261 queue_items = [
1262 (qi.id, qi.resource_id)
1263 for qi in (
1264 session.query(LibraryDownloadQueue)
1265 .filter_by(
1266 research_id=research_id,
1267 status=DocumentStatus.PENDING,
1268 )
1269 .all()
1270 )
1271 ]
1273 # Process each queued item
1274 for queue_item_id, resource_id in queue_items:
1275 # Atomically claim this row before downloading so a
1276 # second concurrent download_bulk stream can't also
1277 # process it (issue #4691). A row already claimed by
1278 # another stream returns 0 updated rows; skip it.
1279 with get_user_db_session(username) as session:
1280 claimed = _claim_download_queue_item(
1281 session, queue_item_id
1282 )
1283 if not claimed: 1283 ↛ 1284line 1283 didn't jump to line 1284 because the condition on line 1283 was never true
1284 continue
1286 current += 1
1288 progress = (
1289 int((current / total) * 100) if total > 0 else 100
1290 )
1292 # Attempt actual download with error handling
1293 skip_reason = None
1294 status = "skipped" # Default to skipped
1295 success = False
1296 error_msg = None
1297 error_type = None
1298 # Everything from the resource lookup onward runs
1299 # inside the try below, so file_name needs a value
1300 # even if that block raises before setting one.
1301 file_name = f"document_{current}.pdf"
1303 try:
1304 # Get resource info in its own short-lived session.
1305 # This sits INSIDE the try because the claim above is
1306 # already committed as PROCESSING: any raise between
1307 # the claim and this handler would leave the row that
1308 # way forever, since nothing in src/ reaps PROCESSING
1309 # rows and every reset path now refuses to touch them
1310 # (issue #4691). ResearchResource.title is nullable, so
1311 # resource.title[:50] raises TypeError on a resource
1312 # whose title was never populated.
1313 with get_user_db_session(username) as session:
1314 resource = session.get(
1315 ResearchResource, resource_id
1316 )
1317 if resource and resource.title:
1318 file_name = resource.title[:50]
1320 logger.debug(
1321 f"Attempting {'PDF download' if mode == 'pdf' else 'text extraction'} for resource {resource_id}"
1322 )
1324 # Call appropriate service method based on mode.
1325 # DownloadService opens its own session internally; no
1326 # session is held here across the blocking download.
1327 if mode == "pdf":
1328 result = download_service.download_resource(
1329 resource_id
1330 )
1331 else: # text_only
1332 result = download_service.download_as_text(
1333 resource_id
1334 )
1336 # Handle new tuple return format
1337 if isinstance(result, tuple): 1337 ↛ 1340line 1337 didn't jump to line 1340 because the condition on line 1337 was always true
1338 success, skip_reason = result
1339 else:
1340 success = result
1341 skip_reason = None
1343 # Finalize the claim for the paths that don't
1344 # record a terminal status themselves (text_only,
1345 # and download_resource's already-downloaded early
1346 # return); a no-op when they already did (#4691).
1347 with get_user_db_session(username) as session:
1348 _finalize_download_queue_item(
1349 session, queue_item_id, success
1350 )
1352 status = "success" if success else "skipped"
1353 if skip_reason and not success:
1354 error_msg = skip_reason
1355 logger.info(
1356 f"{'Download' if mode == 'pdf' else 'Text extraction'} skipped for resource {resource_id}: {failure_reason_for_log(skip_reason)}"
1357 )
1359 logger.debug(
1360 f"{'Download' if mode == 'pdf' else 'Text extraction'} result: success={success}, status={status}, skip_reason={failure_reason_for_log(skip_reason)}"
1361 )
1362 except Exception as e:
1363 # Release the claim so a later run can retry this row:
1364 # the download raised before download_resource recorded
1365 # a terminal status (issue #4691). A fresh short-lived
1366 # session — the download's own session is unrelated and
1367 # each op above already closed its scope, so there is no
1368 # held session to roll back.
1369 try:
1370 with get_user_db_session(username) as session:
1371 _release_download_queue_item(
1372 session, queue_item_id
1373 )
1374 except Exception:
1375 logger.exception(
1376 "Failed to release download-queue claim"
1377 )
1378 # Log error but continue processing. `error_msg`
1379 # (str(e)) is used ONLY for the keyword classification
1380 # below (`.lower()` checks) — it is never logged
1381 # (an exception escaping download_resource /
1382 # download_as_text can embed the raw resource URL:
1383 # urllib3 "with url: ...", SQLAlchemy
1384 # "[parameters: ...]") and never written into
1385 # `skip_reason`, which the yield below sends to the
1386 # client. Only `error_type` (the exception CLASS NAME)
1387 # reaches the log and skip_reason, same "class name
1388 # only" mitigation as journal_quality/downloader.py
1389 # (CodeQL alerts 7650/7684) — CWE-209.
1390 error_msg = str(e)
1391 error_type = type(e).__name__
1392 logger.warning(
1393 f"CAUGHT Download exception for resource {resource_id}: {error_type}"
1394 )
1395 # Check if this is a skip reason (not a real error)
1396 # Use error category + categorized message for user display
1397 if any(
1398 phrase in error_msg.lower()
1399 for phrase in [
1400 "paywall",
1401 "subscription",
1402 "not available",
1403 "not found",
1404 "no free",
1405 "embargoed",
1406 "forbidden",
1407 "not accessible",
1408 ]
1409 ):
1410 status = "skipped"
1411 skip_reason = f"Document not accessible (paywall or access restriction) - {error_type}"
1412 elif any(
1413 phrase in error_msg.lower()
1414 for phrase in [
1415 "failed to download",
1416 "could not",
1417 "invalid",
1418 "server",
1419 ]
1420 ):
1421 status = "failed"
1422 skip_reason = f"Download failed - {error_type}"
1423 else:
1424 status = "failed"
1425 skip_reason = f"Processing failed - {error_type}"
1426 success = False
1428 # Ensure skip_reason is set if we have an error message
1429 if error_msg and not skip_reason: 1429 ↛ 1430line 1429 didn't jump to line 1430 because the condition on line 1429 was never true
1430 skip_reason = f"Processing failed - {error_type}"
1431 logger.debug(
1432 f"Setting skip_reason from error type: {error_type}"
1433 )
1435 # Send progress update
1436 update_data = {
1437 "progress": progress,
1438 "current": current,
1439 "total": total,
1440 "file": file_name,
1441 "status": status,
1442 }
1443 # Add skip reason if available
1444 if skip_reason:
1445 update_data["error"] = skip_reason
1446 logger.info(
1447 "Sending skip reason to UI: "
1448 f"{failure_reason_for_log(skip_reason)}"
1449 )
1451 yield f"data: {json.dumps(update_data)}\n\n"
1453 yield f"data: {json.dumps({'progress': 100, 'current': current, 'total': total, 'complete': True})}\n\n"
1454 finally:
1455 from ...utilities.resource_utils import safe_close
1457 safe_close(download_service, "download service")
1459 # CWE-209 (CodeQL "Information exposure through an exception"): the SSE
1460 # body streamed here never carries a raw exception message. `skip_reason`
1461 # either comes straight from download_service (already a static string or
1462 # a fixed token from _client_safe_download_message() — see
1463 # download_resource / download_as_text and their helpers in
1464 # download_service.py; sanitize_error_for_client() alone would not
1465 # suffice since it scrubs credentials only, not SQL/paths) or, on the
1466 # `except Exception as e:` path above, is built from `type(e).__name__`
1467 # only — never the raw `error_msg = str(e)`.
1468 return WorkerCleanupStreamingResponse(
1469 generate(),
1470 media_type="text/event-stream",
1471 headers={
1472 "Cache-Control": "no-cache, no-transform",
1473 "X-Accel-Buffering": "no",
1474 },
1475 )
1478@router.get("/api/research-list")
1479def get_research_list(
1480 request: Request, username: Annotated[str, Depends(require_auth)]
1481):
1482 """Get the lightweight research list used to populate a dropdown.
1484 Deliberately NOT ``get_research_list_with_stats``: that runs a 3-table
1485 GROUP BY aggregate plus follow-up ratings and domain-breakdown queries,
1486 which a <select> needing only id/title/query does not use. Main replaced
1487 this call for exactly that reason.
1488 """
1489 service = LibraryService(username)
1490 research_list = service.get_research_list_for_dropdown()
1491 return {"research": research_list}
1494@router.post("/api/sync-library")
1495def sync_library(
1496 request: Request, username: Annotated[str, Depends(require_auth)]
1497):
1498 """Sync library database with filesystem."""
1499 service = LibraryService(username)
1500 return service.sync_library_with_filesystem()
1503@router.post("/api/mark-redownload")
1504async def mark_for_redownload(
1505 request: Request, username: Annotated[str, Depends(require_auth)]
1506):
1507 """Mark documents for re-download."""
1508 data, body_error = await read_json_dict(
1509 request, "simple", "Request body must be valid JSON"
1510 )
1511 if body_error is not None:
1512 return body_error
1513 document_ids = data.get("document_ids", [])
1515 if not document_ids:
1516 return JSONResponse(
1517 {"error": "No document IDs provided"}, status_code=400
1518 )
1519 if error := _string_list_error(document_ids, "document_ids"):
1520 return error
1522 def _mark_sync():
1523 service = LibraryService(username)
1524 return service.mark_for_redownload(document_ids)
1526 count = await run_db_sync(_mark_sync)
1527 return {"success": True, "marked": count}
1530def _one_resource_per_awaiting_document(
1531 db_session, resources, username, password
1532):
1533 """Keep one of the resources that would refill the same document.
1535 A document that awaits its PDF (see ``document_awaits_pdf``) does not
1536 count as a download, so "Download All New" finds its own resource
1537 (``Document.resource_id``) and every resource linked to it by content
1538 (``ResearchResource.document_id``). Each of them downloads into that
1539 document (``get_document_for_resource``), and one download refills it.
1540 So of the ``resources`` that would refill the same document only one is
1541 kept, the one ``queue_research_downloads`` would leave the document to
1542 (see ``DownloadService._refilled_by_another_resource``): one that a
1543 download holds (a PROCESSING queue entry, see
1544 ``download_claim_is_live``), else a viable one that is already PENDING
1545 in the download queue, else the one that ranks first in
1546 ``DownloadService.refill_rank``. So a resource that cannot succeed (its
1547 URL cannot be queued, the retry manager holds it back, or its latest
1548 attempt failed) is kept only when it is being downloaded or no viable
1549 one is there. A resource whose PROCESSING entry a restart left behind
1550 comes last, since no reset touches that entry. The others count as
1551 skipped. (Once the document holds its PDF again,
1552 ``DownloadService.download_resource`` reports the document's own
1553 resource as already downloaded, and for a linked one whose URL the
1554 egress policy allows, ``DownloadService._download_pdf`` reuses that
1555 stored PDF instead of fetching it.)
1557 ``resources`` have passed the route's retry-policy and downloadability
1558 checks. The ranking also asks the egress policy and the resources'
1559 retry records, through a ``DownloadService`` for ``username``, which is
1560 built only when some document has more than one of them.
1562 Returns the kept resources in their order, and how many were left out.
1563 """
1564 awaiting = {
1565 resource_id: (document_id, owner_id)
1566 for resource_id, document_id, owner_id in (
1567 db_session.query(
1568 ResearchResource.id, Document.id, Document.resource_id
1569 )
1570 .join(
1571 Document,
1572 resource_document_clause(),
1573 )
1574 .filter(document_awaits_pdf_clause())
1575 .all()
1576 )
1577 }
1578 candidates = {}
1579 for resource in resources:
1580 if resource.id in awaiting:
1581 document_id = awaiting[resource.id][0]
1582 candidates.setdefault(document_id, []).append(resource)
1583 shared = [rows for rows in candidates.values() if len(rows) > 1]
1584 if not shared:
1585 return resources, 0
1587 entries = {
1588 entry.resource_id: entry
1589 for entry in db_session.query(
1590 LibraryDownloadQueue.resource_id,
1591 LibraryDownloadQueue.status,
1592 LibraryDownloadQueue.last_attempt_at,
1593 )
1594 .filter(
1595 LibraryDownloadQueue.resource_id.in_(
1596 [row.id for rows in shared for row in rows]
1597 ),
1598 LibraryDownloadQueue.status.in_(
1599 (DocumentStatus.PENDING, DocumentStatus.PROCESSING)
1600 ),
1601 )
1602 .all()
1603 }
1604 left_out = set()
1605 with DownloadService(username, password) as download_service:
1606 for rows in shared:
1607 owner_id = awaiting[rows[0].id][1]
1608 choices = []
1609 for row in rows:
1610 rank = download_service.refill_rank(row.id, row.url, owner_id)
1611 entry = entries.get(row.id)
1612 if entry is None:
1613 tier = 2
1614 elif entry.status == DocumentStatus.PROCESSING:
1615 tier = (
1616 0
1617 if download_claim_is_live(entry.last_attempt_at)
1618 else 3
1619 )
1620 else:
1621 tier = 1 if rank.viable else 2
1622 choices.append(((tier, rank), row.id))
1623 kept_id = min(choices)[1]
1624 left_out.update(row.id for row in rows if row.id != kept_id)
1625 for resource_id in sorted(left_out):
1626 logger.debug(
1627 f"Skipping resource {resource_id}: another resource of its "
1628 "document is queued to download its PDF"
1629 )
1630 kept = [resource for resource in resources if resource.id not in left_out]
1631 return kept, len(left_out)
1634@router.post("/api/queue-all-undownloaded")
1635def queue_all_undownloaded(
1636 request: Request, username: Annotated[str, Depends(require_auth)]
1637):
1638 """Queue all articles that haven't been downloaded yet."""
1640 logger.info(f"queue_all_undownloaded called for user {username}")
1642 with get_user_db_session(username) as db_session:
1643 # Find all resources that don't have a completed download. Project only
1644 # the columns the filter/loop use (id/url/research_id) instead of full
1645 # ``ResearchResource`` entities so the ``content_preview`` Text column
1646 # isn't materialized for every row on this whole-table scan (#4560).
1647 # A completed document that still needs its PDF (the library sync
1648 # kept its text after its PDF file went missing) is not counted as
1649 # a download, see document_awaits_pdf. Only one of the resources
1650 # that would refill such a document is queued, see
1651 # _one_resource_per_awaiting_document.
1652 undownloaded = (
1653 db_session.query(
1654 ResearchResource.id,
1655 ResearchResource.url,
1656 ResearchResource.research_id,
1657 )
1658 .outerjoin(
1659 Document,
1660 resource_document_clause()
1661 & (Document.status == "completed")
1662 & ~document_awaits_pdf_clause(),
1663 )
1664 .filter(Document.id.is_(None))
1665 .all()
1666 )
1668 logger.info(f"Found {len(undownloaded)} total undownloaded resources")
1670 # Get user password for encrypted database access
1671 user_password = get_authenticated_user_password(username)
1673 resource_filter = ResourceFilter(username, user_password)
1674 filter_results = resource_filter.filter_downloadable_resources(
1675 undownloaded
1676 )
1678 # Get detailed filtering summary
1679 filter_summary = resource_filter.get_filter_summary(undownloaded)
1680 skipped_info = resource_filter.get_skipped_resources_info(undownloaded)
1682 logger.info(f"Filter results: {filter_summary.to_dict()}")
1684 queued_count = 0
1685 research_ids = set()
1686 skipped_count = 0
1688 # Convert filter_results to dict for O(1) lookup instead of O(n²)
1689 filter_results_by_id = {r.resource_id: r for r in filter_results}
1691 # The resources that pass every check below, in scan order.
1692 eligible = []
1693 for resource in undownloaded:
1694 # Check if resource passed the smart filter
1695 filter_result = filter_results_by_id.get(resource.id)
1697 if not filter_result or not filter_result.can_retry:
1698 skipped_count += 1
1699 if filter_result:
1700 # A permanently failed row's reason embeds its stored
1701 # failure_message, which rows written before the
1702 # domain fix may hold with URL userinfo.
1703 logger.debug(
1704 f"Skipping resource {resource.id} due to retry policy: "
1705 f"{failure_reason_for_log(filter_result.reason)}"
1706 )
1707 else:
1708 logger.debug(
1709 f"Skipping resource {resource.id} - no filter result available"
1710 )
1711 continue
1713 # Check if it's downloadable using proper URL parsing
1714 if not resource.url:
1715 skipped_count += 1
1716 continue
1718 is_downloadable = is_downloadable_domain(resource.url)
1720 # Log what we're checking
1721 if resource.url and "pubmed" in resource.url.lower(): 1721 ↛ 1722line 1721 didn't jump to line 1722 because the condition on line 1721 was never true
1722 logger.info(
1723 "Found PubMed URL: {}",
1724 # (#6938: redacted before it is cut.)
1725 redact_and_bound_for_log(
1726 redact_url_for_log(resource.url), 100
1727 ),
1728 )
1730 if not is_downloadable: 1730 ↛ 1731line 1730 didn't jump to line 1731 because the condition on line 1730 was never true
1731 skipped_count += 1
1732 logger.opt(lazy=True).debug(
1733 "Skipping non-downloadable URL: {}",
1734 lambda: redact_and_bound_for_log(
1735 redact_url_for_log(resource.url), 100
1736 ),
1737 )
1738 continue
1740 eligible.append(resource)
1742 # One download refills a document that awaits its PDF, so only one
1743 # of the resources that would refill the same document is queued.
1744 eligible, covered = _one_resource_per_awaiting_document(
1745 db_session, eligible, username, user_password
1746 )
1747 skipped_count += covered
1749 for resource in eligible:
1750 # Check if already in queue (any status)
1751 existing_queue = (
1752 db_session.query(LibraryDownloadQueue)
1753 .filter_by(resource_id=resource.id)
1754 .first()
1755 )
1757 if existing_queue:
1758 # Reset a non-pending row back to PENDING so it is retried,
1759 # but never one a concurrent download stream has claimed:
1760 # PROCESSING satisfies `!= PENDING`, so an unguarded reset
1761 # un-claims an in-flight row and lets a second stream
1762 # re-download the same resource (issue #4691). The query above
1763 # selects in-flight rows, since a download that has not
1764 # finished has no completed Document yet. The guard lives
1765 # inside the UPDATE, not in an `if` above it, so the check and
1766 # the write cannot straddle another stream's claim.
1767 reset = (
1768 db_session.query(LibraryDownloadQueue)
1769 .filter(
1770 LibraryDownloadQueue.id == existing_queue.id,
1771 LibraryDownloadQueue.status != DocumentStatus.PENDING,
1772 LibraryDownloadQueue.status
1773 != DocumentStatus.PROCESSING,
1774 )
1775 .update(
1776 {
1777 LibraryDownloadQueue.status: DocumentStatus.PENDING,
1778 LibraryDownloadQueue.completed_at: None,
1779 },
1780 synchronize_session=False,
1781 )
1782 )
1783 # Counted either way, matching the pre-fix behaviour: a row
1784 # already PENDING, a row just reset, and a row another stream
1785 # is downloading are all "queued" to the caller.
1786 queued_count += 1
1787 research_ids.add(resource.research_id)
1788 if reset:
1789 logger.debug(
1790 f"Reset queue entry for resource {resource.id} to pending"
1791 )
1792 else:
1793 logger.debug(
1794 f"Resource {resource.id} already pending or in flight"
1795 )
1796 else:
1797 # Add new entry to queue
1798 queue_entry = LibraryDownloadQueue(
1799 resource_id=resource.id,
1800 research_id=resource.research_id,
1801 priority=0,
1802 status=DocumentStatus.PENDING,
1803 )
1804 db_session.add(queue_entry)
1805 queued_count += 1
1806 research_ids.add(resource.research_id)
1807 logger.debug(
1808 f"Added new queue entry for resource {resource.id}"
1809 )
1811 db_session.commit()
1813 logger.info(
1814 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)"
1815 )
1817 # Note: Removed synchronous download processing here to avoid blocking the HTTP request
1818 # Downloads will be processed via the SSE streaming endpoint or background tasks
1820 return {
1821 "success": True,
1822 "queued": queued_count,
1823 "research_ids": list(research_ids),
1824 "total_undownloaded": len(undownloaded),
1825 "skipped": skipped_count,
1826 "filter_summary": filter_summary.to_dict(),
1827 "skipped_details": skipped_info,
1828 }
1831@router.get("/api/get-research-sources/{research_id}")
1832def get_research_sources(
1833 request: Request,
1834 research_id,
1835 username: Annotated[str, Depends(require_auth)],
1836):
1837 """Get all sources for a research with snippets."""
1839 sources = []
1840 with get_user_db_session(username) as db_session:
1841 # Get all resources for this research
1842 resources = (
1843 db_session.query(ResearchResource)
1844 .filter_by(research_id=research_id)
1845 .order_by(ResearchResource.created_at)
1846 .all()
1847 )
1849 for idx, resource in enumerate(resources, 1):
1850 # Check if document exists
1851 document = get_document_for_resource(db_session, resource)
1853 # Get domain from URL
1854 domain = ""
1855 if resource.url:
1856 try:
1857 from urllib.parse import urlparse
1859 domain = urlparse(resource.url).hostname or ""
1860 except (ValueError, AttributeError):
1861 # urlparse can raise ValueError for malformed URLs
1862 pass
1864 source_data = {
1865 "number": idx,
1866 "resource_id": resource.id,
1867 "url": resource.url,
1868 "title": resource.title or f"Source {idx}",
1869 "snippet": resource.content_preview or "",
1870 "domain": domain,
1871 "relevance_score": getattr(resource, "relevance_score", None),
1872 "downloaded": False,
1873 "document_id": None,
1874 "file_type": None,
1875 }
1877 # The Download Manager's rule: a completed document, unless it
1878 # awaits its PDF (see document_counts_as_downloaded). A loaded
1879 # status never equals the string "completed".
1880 if document and document_counts_as_downloaded(document):
1881 source_data.update(
1882 {
1883 "downloaded": True,
1884 "document_id": document.id,
1885 "file_type": document.file_type,
1886 "download_date": document.created_at.isoformat()
1887 if document.created_at
1888 else None,
1889 }
1890 )
1892 sources.append(source_data)
1894 return {"success": True, "sources": sources, "total": len(sources)}
1897@router.post("/api/check-downloads")
1898async def check_downloads(
1899 request: Request, username: Annotated[str, Depends(require_auth)]
1900):
1901 """Check download status for a list of URLs."""
1902 data, body_error = await read_json_dict(
1903 request, "simple", "Request body must be valid JSON"
1904 )
1905 if body_error is not None:
1906 return body_error
1907 research_id = data.get("research_id")
1908 urls = data.get("urls", [])
1910 if not isinstance(research_id, str) or not research_id.strip() or not urls:
1911 return JSONResponse(
1912 {"error": "Missing research_id or urls"}, status_code=400
1913 )
1914 if error := _string_list_error(urls, "urls"):
1915 return error
1917 def _query_sync():
1918 download_status = {}
1919 with get_user_db_session(username) as db_session:
1920 resources = (
1921 db_session.query(ResearchResource)
1922 .filter_by(research_id=research_id)
1923 .filter(ResearchResource.url.in_(urls))
1924 .all()
1925 )
1926 for resource in resources:
1927 document = get_document_for_resource(db_session, resource)
1928 # A document that awaits its PDF is reported not downloaded,
1929 # with the resource_id to download it by (see
1930 # document_counts_as_downloaded).
1931 if document and document_counts_as_downloaded(document):
1932 # Do NOT return document.file_path — it's the absolute
1933 # server path, which leaks directory layout to any
1934 # authenticated user. The client fetches via
1935 # `/document/{id}/pdf` or `/document/{id}/text` using
1936 # document_id, so the path is never needed client-side.
1937 download_status[resource.url] = {
1938 "downloaded": True,
1939 "document_id": document.id,
1940 "file_type": document.file_type,
1941 "title": document.title or resource.title,
1942 }
1943 else:
1944 download_status[resource.url] = {
1945 "downloaded": False,
1946 "resource_id": resource.id,
1947 }
1948 return download_status
1950 download_status = await run_db_sync(_query_sync)
1951 return {"download_status": download_status}
1954@router.post("/api/download-source")
1955async def download_source(
1956 request: Request, username: Annotated[str, Depends(require_auth)]
1957):
1958 """Download a single source from a research."""
1959 user_password = get_authenticated_user_password(username)
1960 data, body_error = await read_json_dict(
1961 request, "simple", "Request body must be valid JSON"
1962 )
1963 if body_error is not None:
1964 return body_error
1965 research_id = data.get("research_id")
1966 url = data.get("url")
1968 if not research_id or not url:
1969 return JSONResponse(
1970 {"error": "Missing research_id or url"}, status_code=400
1971 )
1973 # Check if URL is downloadable
1974 if not is_downloadable_domain(url):
1975 return JSONResponse(
1976 {"error": "URL is not from a downloadable domain"}, status_code=400
1977 )
1979 def _download_sync():
1980 """The DB + network work — all blocking, run in a threadpool."""
1981 with get_user_db_session(username) as db_session:
1982 # Find the resource
1983 resource = (
1984 db_session.query(ResearchResource)
1985 .filter_by(research_id=research_id, url=url)
1986 .first()
1987 )
1989 if not resource:
1990 return JSONResponse(
1991 {"error": "Resource not found"}, status_code=404
1992 )
1994 # Check if already downloaded, by the Download Manager's rule
1995 # (see document_counts_as_downloaded). A document that awaits
1996 # its PDF is downloaded again, into the same document.
1997 existing = get_document_for_resource(db_session, resource)
1999 if existing and document_counts_as_downloaded(existing):
2000 return {
2001 "success": True,
2002 "message": "Already downloaded",
2003 "document_id": existing.id,
2004 }
2006 # Add to download queue
2007 queue_entry = (
2008 db_session.query(LibraryDownloadQueue)
2009 .filter_by(resource_id=resource.id)
2010 .first()
2011 )
2013 if not queue_entry:
2014 queue_entry = LibraryDownloadQueue(
2015 resource_id=resource.id,
2016 research_id=resource.research_id,
2017 priority=1, # Higher priority for manual downloads
2018 status=DocumentStatus.PENDING,
2019 )
2020 db_session.add(queue_entry)
2021 db_session.commit()
2022 else:
2023 # Reset the row so this manual download can claim it, but never
2024 # one a concurrent stream is already downloading: the reset is
2025 # followed immediately by a download in this request thread, so
2026 # un-claiming an in-flight row starts a second download of the
2027 # same resource (issue #4691). The already-downloaded early
2028 # return above cannot prevent that, because a download still in
2029 # flight has not written a completed Document yet.
2030 reset = (
2031 db_session.query(LibraryDownloadQueue)
2032 .filter(
2033 LibraryDownloadQueue.id == queue_entry.id,
2034 LibraryDownloadQueue.status
2035 != DocumentStatus.PROCESSING,
2036 )
2037 .update(
2038 {
2039 LibraryDownloadQueue.status: DocumentStatus.PENDING,
2040 LibraryDownloadQueue.priority: 1,
2041 },
2042 synchronize_session=False,
2043 )
2044 )
2045 db_session.commit()
2046 if not reset:
2047 return JSONResponse(
2048 {
2049 "success": False,
2050 "message": "Download already in progress",
2051 },
2052 status_code=409,
2053 )
2055 queue_item_id = queue_entry.id
2056 # The claim commits and expires ORM rows. Snapshot the resource
2057 # ID first so no attribute access can fail between claiming and
2058 # the download. Base re-read resource.id after claiming, inside
2059 # the try: that refresh raised, and although the handler caught
2060 # it and released the claim, the download itself never ran.
2061 resource_id = resource.id
2063 # Claim the row before downloading, exactly as download_bulk does,
2064 # so a bulk stream that reaches this row between the reset above and
2065 # the download below cannot also process it (issue #4691).
2066 if not _claim_download_queue_item(db_session, queue_item_id):
2067 return JSONResponse(
2068 {
2069 "success": False,
2070 "message": "Download already in progress",
2071 },
2072 status_code=409,
2073 )
2075 # The claim above is committed. Leave the thread-local read scope
2076 # before the remote download, which may occupy this worker for minutes.
2077 try:
2078 with DownloadService(username, user_password) as service:
2079 success, message = service.download_resource(resource_id)
2080 except Exception:
2081 # A fresh short-lived session — the download ran outside the read
2082 # scope above, so there is no held session to roll back. Pass the
2083 # password captured at request start: by the time a multi-minute
2084 # transfer ends, logout / password change / the idle sweep may
2085 # have cleared the session password store, and the store-less
2086 # lookup would raise DatabaseSessionError before the release,
2087 # stranding the row in PROCESSING (issue #4691). Guard the
2088 # release so a failure here cannot mask the original exception.
2089 try:
2090 with get_user_db_session(username, user_password) as db_session:
2091 # Release the claim so a later run can retry this row:
2092 # the download raised before download_resource recorded a
2093 # terminal status (issue #4691).
2094 _release_download_queue_item(db_session, queue_item_id)
2095 except Exception:
2096 logger.exception("Failed to release download-queue claim")
2097 raise
2099 # download_resource writes a terminal status for the rows it actually
2100 # downloads, so this is a no-op for those; it fires on early returns,
2101 # which would otherwise strand the claim in PROCESSING. The password
2102 # must be passed for the same reason as the release above.
2103 with get_user_db_session(username, user_password) as db_session:
2104 _finalize_download_queue_item(db_session, queue_item_id, success)
2106 if success:
2107 return {"success": True, "message": "Download completed"}
2108 # Log internal message, but show only generic message to user
2109 return {"success": False, "message": "Download failed"}
2111 # Offload the blocking DB + HTTP work so we don't stall the event loop.
2112 return await run_db_sync(_download_sync)