Coverage for src/local_deep_research/web/routers/library.py: 93%
591 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-06 15:42 +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 ..template_config import templates
23import os
24import json
25from datetime import UTC, datetime
26from pathlib import Path
28from loguru import logger
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 ...research_library.services.download_service import DownloadService
40from ...research_library.services.library_service import LibraryService
41from ...research_library.services.pdf_storage_manager import PDFStorageManager
42from ...research_library.utils import (
43 apply_user_subdir,
44 get_document_for_resource,
45 handle_api_error,
46 is_downloadable_domain,
47 is_downloadable_url,
48)
49from ...utilities.db_utils import get_settings_manager
50from ...config.paths import get_library_directory
51from ...web.exceptions import AuthenticationRequiredError
52from ..dependencies.json_body import json_body_error
53from typing import Annotated
55# Create the router
56router = APIRouter(prefix="/library", tags=["library"])
59def _string_list_error(value, field_name: str) -> JSONResponse | None:
60 """Return a 400 response unless *value* is a non-empty-string list."""
61 if not isinstance(value, list) or any(
62 not isinstance(item, str) or not item.strip() for item in value
63 ):
64 return JSONResponse(
65 {"error": f"{field_name} must be a list of non-empty strings"},
66 status_code=400,
67 )
68 return None
71# NOTE: Routes use username (not .get()) intentionally.
72# Depends(require_auth) guarantees the key exists; direct access fails
73# fast if the dependency is ever removed.
76# Error handler for authentication errors
77def handle_web_api_exception(error):
78 """Handle WebAPIException and its subclasses."""
79 from ...web.exceptions import WebAPIException
81 if isinstance(error, WebAPIException):
82 return error.to_dict(), error.status_code
83 # Re-raise other exceptions
84 raise error
87def get_authenticated_user_password(
88 username: str, session_id: str | None = None
89) -> str:
90 """
91 Get authenticated user password from session store.
93 Args:
94 username: The username to get password for
95 session_id: Optional session ID for direct lookup.
97 Returns:
98 str: The user's password
100 Raises:
101 AuthenticationRequiredError: If no password is available for the user
102 """
103 from ...database.session_passwords import session_password_store
105 # Try specific session first
106 if session_id:
107 try:
108 user_password = session_password_store.get_session_password(
109 username, session_id
110 )
111 if user_password:
112 return user_password
113 except Exception:
114 logger.exception("Failed to get user password from session store")
116 # Try any session for this user (works without Flask)
117 user_password = session_password_store.get_any_session_password(username)
118 if user_password:
119 return user_password
121 # No password available
122 logger.error(f"No user password available for user {username}")
123 raise AuthenticationRequiredError(
124 message="Authentication required: Please refresh the page and log in again to access encrypted database features.",
125 )
128# ============= Page Routes =============
131@router.get("/")
132def library_page(
133 request: Request,
134 username: Annotated[str, Depends(require_auth)],
135):
136 """Main library page showing downloaded documents."""
137 import math
139 try:
140 with get_user_db_session(username) as db_session:
141 settings = get_settings_manager(db_session, username)
142 pdf_storage_mode = settings.get_setting(
143 "research_library.pdf_storage_mode", "database"
144 )
145 shared_library = settings.get_setting(
146 "research_library.shared_library", False
147 )
148 except Exception:
149 pdf_storage_mode = "database"
150 shared_library = False
152 enable_pdf_storage = pdf_storage_mode != "none"
154 # Filters
155 domain_filter = request.query_params.get("domain")
156 research_filter = request.query_params.get("research")
157 collection_filter = request.query_params.get("collection")
158 date_filter = request.query_params.get("date")
160 # Build context with graceful degradation
161 stats = {}
162 documents = []
163 unique_domains = []
164 research_list = []
165 collections = []
166 page = 1
167 total_pages = 1
168 # Degrading gracefully is deliberate here (main let the exception
169 # reach Flask's 500 handler), but an empty `documents` list is
170 # ambiguous: the template's empty state would tell a user whose
171 # library failed to load that they have no documents yet, and invite
172 # them to go download some. Track *why* the list is empty so the page
173 # can say "we could not load this" instead of asserting something
174 # false about their data.
175 load_error = False
177 try:
178 service = LibraryService(username)
179 stats = service.get_library_stats()
181 from ...database.library_init import get_default_library_id
183 resolved_collection = collection_filter or get_default_library_id(
184 username
185 )
187 # Pagination
188 per_page = 100
189 total_docs = service.count_documents(
190 research_id=research_filter,
191 domain=domain_filter,
192 collection_id=resolved_collection,
193 date_filter=date_filter,
194 )
195 total_pages = max(1, math.ceil(total_docs / per_page))
196 try:
197 page = int(request.query_params.get("page", "1"))
198 except ValueError:
199 page = 1
200 page = max(1, min(page, total_pages))
201 offset = (page - 1) * per_page
203 documents = service.get_documents(
204 research_id=research_filter,
205 domain=domain_filter,
206 collection_id=resolved_collection,
207 date_filter=date_filter,
208 limit=per_page,
209 offset=offset,
210 )
211 unique_domains = service.get_unique_domains()
212 research_list = service.get_research_list_for_dropdown()
213 collections = service.get_all_collections()
214 except Exception:
215 logger.exception("Error loading library data")
216 load_error = True
218 default_collection_id = next(
219 (c["id"] for c in collections if c.get("is_default")), None
220 )
222 return templates.TemplateResponse(
223 request=request,
224 name="pages/library.html",
225 context={
226 "stats": stats,
227 "documents": documents,
228 "load_error": load_error,
229 "unique_domains": unique_domains,
230 "research_list": research_list,
231 "collections": collections,
232 "selected_collection": collection_filter,
233 "default_collection_id": default_collection_id,
234 "storage_path": stats.get("storage_path", ""),
235 "enable_pdf_storage": enable_pdf_storage,
236 "pdf_storage_mode": pdf_storage_mode,
237 "shared_library": shared_library,
238 "page": page,
239 "total_pages": total_pages,
240 "selected_date": date_filter,
241 "selected_research": research_filter,
242 "selected_domain": domain_filter,
243 },
244 )
247@router.get("/document/{document_id}")
248def document_details_page(
249 request: Request,
250 document_id,
251 username: Annotated[str, Depends(require_auth)],
252):
253 """Document details page showing all metadata and links."""
254 service = LibraryService(username)
256 # Get document details
257 document = service.get_document_by_id(document_id)
259 if not document:
260 # Browser navigation: return text/html, not JSON. These four page
261 # routes are reached from library.html as ordinary <a href> links, so a
262 # stale link showed the user a raw {"error": ...} body in the browser's
263 # JSON viewer. main returned `"Document not found", 404` (text/html)
264 # here; the sibling /api/ routes keep JSON.
265 return HTMLResponse("Document not found", status_code=404)
267 return templates.TemplateResponse(
268 request=request,
269 name="pages/document_details.html",
270 context={"request": request, "document": document},
271 )
274@router.get("/download-manager")
275def download_manager_page(
276 request: Request,
277 username: Annotated[str, Depends(require_auth)],
278):
279 """Download manager page for selecting and downloading research PDFs."""
280 import math
282 service = LibraryService(username)
283 with get_user_db_session(username) as db_session:
284 settings = get_settings_manager(db_session, username)
285 pdf_storage_mode = settings.get_setting(
286 "research_library.pdf_storage_mode", "database"
287 )
288 shared_library = settings.get_setting(
289 "research_library.shared_library", False
290 )
291 enable_pdf_storage = pdf_storage_mode != "none"
293 # Summary stats over ALL sessions (also used for page count)
294 per_page = 50
295 summary = service.get_download_manager_summary_stats()
296 total_pages = max(1, math.ceil(summary["total_researches"] / per_page))
298 # Pagination with upper-bound clamp
299 try:
300 page = int(request.query_params.get("page", "1"))
301 except ValueError:
302 page = 1
303 page = max(1, min(page, total_pages))
304 offset = (page - 1) * per_page
306 # Get paginated research sessions
307 research_list = service.get_research_list_with_stats(
308 limit=per_page, offset=offset
309 )
311 # Batch-fetch PDF previews and domain breakdowns (single query)
312 research_ids = [r["id"] for r in research_list]
313 previews = service.get_pdf_previews_batch(research_ids)
314 for research in research_list: 314 ↛ 315line 314 didn't jump to line 315 because the loop on line 314 never started
315 rid = research["id"]
316 data = previews.get(rid, {"pdf_sources": [], "domains": {}})
317 research["pdf_sources"] = data["pdf_sources"]
318 research["domains"] = data["domains"]
320 return templates.TemplateResponse(
321 request=request,
322 name="pages/download_manager.html",
323 context={
324 "request": request,
325 "research_list": research_list,
326 "total_researches": summary["total_researches"],
327 "total_resources": summary["total_resources"],
328 "already_downloaded": summary["already_downloaded"],
329 "available_to_download": summary["available_to_download"],
330 "enable_pdf_storage": enable_pdf_storage,
331 "pdf_storage_mode": pdf_storage_mode,
332 "shared_library": shared_library,
333 "page": page,
334 "total_pages": total_pages,
335 },
336 )
339# ============= API Routes =============
342@router.get("/api/stats")
343def get_library_stats(
344 request: Request, username: Annotated[str, Depends(require_auth)]
345):
346 """Get library statistics."""
347 service = LibraryService(username)
348 return service.get_library_stats()
351@router.get("/api/collections/list")
352def get_collections_list(
353 request: Request, username: Annotated[str, Depends(require_auth)]
354):
355 """Get list of all collections for dropdown selection."""
357 with get_user_db_session(username) as db_session:
358 collections = (
359 db_session.query(Collection).order_by(Collection.name).all()
360 )
362 return {
363 "success": True,
364 "collections": [
365 {
366 "id": col.id,
367 "name": col.name,
368 "description": col.description,
369 }
370 for col in collections
371 ],
372 }
375@router.get("/api/documents")
376def get_documents(
377 request: Request, username: Annotated[str, Depends(require_auth)]
378):
379 """Get documents with filtering."""
380 service = LibraryService(username)
382 # Get filter parameters
383 research_id = request.query_params.get("research_id")
384 domain = request.query_params.get("domain")
385 file_type = request.query_params.get("file_type")
386 favorites_only = request.query_params.get("favorites") == "true"
387 search_query = request.query_params.get("search")
388 # Clamp pagination. SQLite treats LIMIT -1 as "no limit", so an
389 # unclamped negative limit (e.g. ?limit=-1) would bypass pagination and
390 # load the whole collection — including large Document.text_content
391 # bodies — into memory (#4560). Mirrors the clamp in history_routes.
392 try:
393 limit = max(1, min(int(request.query_params.get("limit", 100)), 1000))
394 except (TypeError, ValueError):
395 limit = 100
396 try:
397 offset = max(0, int(request.query_params.get("offset", 0)))
398 except (TypeError, ValueError):
399 offset = 0
401 documents = service.get_documents(
402 research_id=research_id,
403 domain=domain,
404 file_type=file_type,
405 favorites_only=favorites_only,
406 search_query=search_query,
407 limit=limit,
408 offset=offset,
409 )
411 return {"documents": documents}
414@router.post("/api/document/{document_id}/favorite")
415def toggle_favorite(
416 request: Request,
417 document_id,
418 username: Annotated[str, Depends(require_auth)],
419):
420 """Toggle favorite status of a document."""
421 service = LibraryService(username)
422 is_favorite = service.toggle_favorite(document_id)
423 return {"favorite": is_favorite}
426@router.get("/api/document/{document_id}/pdf-url")
427def get_pdf_url(
428 request: Request,
429 document_id,
430 username: Annotated[str, Depends(require_auth)],
431):
432 """Get URL for viewing PDF."""
433 # Return URL that will serve the PDF
434 return {
435 "url": f"/library/api/document/{document_id}/pdf",
436 "title": "Document", # Could fetch actual title
437 }
440def _serve_pdf(
441 request: Request, document_id, username: str, *, html_errors: bool
442):
443 """Shared PDF-loading core for the page route and its /api/ sibling.
445 ``html_errors`` picks the not-found/not-available error shape: the
446 browser-navigation page route wants text/html (a stale <a href> link
447 should not show a raw JSON body), while the /api/ sibling keeps the
448 JSON contract every other /api/ route has -- same rule its /text
449 sibling (serve_text_api) already follows.
450 """
452 def _error(message: str) -> Response:
453 if html_errors:
454 return HTMLResponse(message, status_code=404)
455 return JSONResponse({"error": message}, status_code=404)
457 with get_user_db_session(username) as db_session:
458 # Get document from database
459 document = db_session.query(Document).filter_by(id=document_id).first()
461 if not document:
462 logger.warning(
463 f"Document ID {document_id} not found in database for user {username}"
464 )
465 # Browser navigation: return text/html, not JSON. These four page
466 # routes are reached from library.html as ordinary <a href> links, so a
467 # stale link showed the user a raw {"error": ...} body in the browser's
468 # JSON viewer. main returned `"Document not found", 404` (text/html)
469 # here; the sibling /api/ routes keep JSON.
470 return _error("Document not found")
472 logger.info(
473 f"Document {document_id}: title='{document.title}', "
474 f"file_path={document.file_path}"
475 )
477 # Get settings for PDF storage manager
478 settings = get_settings_manager(db_session)
479 storage_mode = settings.get_setting(
480 "research_library.pdf_storage_mode", "none"
481 )
482 # expandvars() matters here and was dropped in the port of main's
483 # #5537. `library_root` below survives without it because
484 # apply_user_subdir() re-expands whatever it is handed, but
485 # `legacy_root` is passed this value RAW -- so with a storage_path
486 # containing an env-var token the legacy fallback would search a
487 # directory literally named e.g. "$HOME", and pre-isolation PDFs
488 # would silently stop being findable.
489 base_root = (
490 Path(
491 os.path.expandvars(
492 settings.get_setting(
493 "research_library.storage_path",
494 str(get_library_directory()),
495 )
496 )
497 )
498 .expanduser()
499 .resolve()
500 )
501 shared_library = settings.get_setting(
502 "research_library.shared_library", False
503 )
504 # Serve from the per-user root, falling back to the legacy shared root
505 # for PDFs downloaded before per-user isolation (issue #5521).
506 #
507 # Ported from #5537, which landed on the Flask library_routes.py this
508 # router replaced. Passing the bare shared root here (as this route
509 # did) resolves a document id against a directory every user writes
510 # into: two users' document ids collide by construction, so one user's
511 # id could serve another's file, and the legacy-fallback gate was
512 # bypassed rather than consulted.
513 library_root = apply_user_subdir(base_root, username, shared_library)
515 # Use PDFStorageManager to load PDF (handles database and filesystem)
516 pdf_manager = PDFStorageManager(
517 library_root, storage_mode, legacy_root=base_root
518 )
519 pdf_bytes = pdf_manager.load_pdf(document, db_session)
521 if pdf_bytes:
522 logger.info(
523 f"Serving PDF for document {document_id} ({len(pdf_bytes)} bytes)"
524 )
525 filename = document.filename or "document.pdf"
526 # Filenames come from the DB (originally user-supplied at
527 # upload). A quote or CR/LF in the raw value would break out
528 # of the header. Use RFC 5987 encoding for a safe round-trip.
529 from urllib.parse import quote as _url_quote
531 safe_filename = _url_quote(filename, safe="")
532 # Plain Response, not StreamingResponse: the bytes are already
533 # fully in memory, and iterating a BytesIO yields *lines* — it
534 # splits on every 0x0A, which in binary PDF data occurs roughly
535 # every 256 bytes, so a single view became tens of thousands of
536 # threadpool dispatches and sent no Content-Length.
537 return Response(
538 content=pdf_bytes,
539 media_type="application/pdf",
540 headers={
541 "Content-Disposition": (
542 f"inline; filename*=UTF-8''{safe_filename}"
543 )
544 },
545 )
547 # No PDF found anywhere
548 logger.warning(f"No PDF available for document {document_id}")
549 # Browser navigation, same rule as the not-found branch above: main
550 # returned `"PDF not available", 404` (text/html) here. Reached
551 # routinely, not just via stale links -- "Remove the PDF file to save
552 # space" deletes the blob while leaving the row, so an open tab or a
553 # bookmark lands here. The /api/ sibling keeps JSON.
554 return _error("PDF not available")
557@router.get("/document/{document_id}/pdf")
558def view_pdf_page(
559 request: Request,
560 document_id,
561 username: Annotated[str, Depends(require_auth)],
562):
563 """Page for viewing PDF file - uses PDFStorageManager for retrieval."""
564 return _serve_pdf(request, document_id, username, html_errors=True)
567@router.get("/api/document/{document_id}/pdf")
568def serve_pdf_api(
569 request: Request,
570 document_id,
571 username: Annotated[str, Depends(require_auth)],
572):
573 """API endpoint for serving PDF file (kept for backward compatibility).
575 Unlike the page route above, a missing document or missing PDF blob
576 here answers JSON, not text/html -- this is an /api/ route, and its
577 /api/document/{id}/text sibling (serve_text_api) already keeps that
578 contract on the same two error branches.
579 """
580 return _serve_pdf(request, document_id, username, html_errors=False)
583@router.get("/document/{document_id}/txt")
584def view_text_page(
585 request: Request,
586 document_id,
587 username: Annotated[str, Depends(require_auth)],
588):
589 """Page for viewing text content."""
591 with get_user_db_session(username) as db_session:
592 # Get document by ID (text now stored in Document.text_content)
593 document = db_session.query(Document).filter_by(id=document_id).first()
595 if not document:
596 logger.warning(f"Document not found for document ID {document_id}")
597 # Browser navigation: return text/html, not JSON. These four page
598 # routes are reached from library.html as ordinary <a href> links, so a
599 # stale link showed the user a raw {"error": ...} body in the browser's
600 # JSON viewer. main returned `"Document not found", 404` (text/html)
601 # here; the sibling /api/ routes keep JSON.
602 return HTMLResponse("Document not found", status_code=404)
604 if not document.text_content:
605 logger.warning(f"Document {document_id} has no text content")
606 # Browser navigation, same rule as the not-found branch above:
607 # main returned `"Text content not available", 404` (text/html)
608 # here. serve_text_api below keeps main's jsonify() shape.
609 return HTMLResponse("Text content not available", status_code=404)
611 logger.info(
612 f"Serving text content for document {document_id}: {len(document.text_content)} characters"
613 )
615 # Render as HTML page
616 return templates.TemplateResponse(
617 request=request,
618 name="pages/document_text.html",
619 context={
620 "document_id": document_id,
621 "title": document.title or "Document Text",
622 "text_content": document.text_content,
623 "extraction_method": document.extraction_method,
624 "word_count": document.word_count,
625 },
626 )
629@router.get("/api/document/{document_id}/text")
630def serve_text_api(
631 request: Request,
632 document_id,
633 username: Annotated[str, Depends(require_auth)],
634):
635 """API endpoint for serving text content (kept for backward compatibility)."""
637 with get_user_db_session(username) as db_session:
638 # Get document by ID (text now stored in Document.text_content)
639 document = db_session.query(Document).filter_by(id=document_id).first()
641 if not document:
642 logger.warning(f"Document not found for document ID {document_id}")
643 return JSONResponse(
644 {"error": "Document not found"}, status_code=404
645 )
647 if not document.text_content:
648 logger.warning(f"Document {document_id} has no text content")
649 return JSONResponse(
650 {"error": "Text content not available"}, status_code=404
651 )
653 logger.info(
654 f"Serving text content for document {document_id}: {len(document.text_content)} characters"
655 )
657 return {
658 "text_content": document.text_content,
659 "title": document.title or "Document",
660 "extraction_method": document.extraction_method,
661 "word_count": document.word_count,
662 }
665@router.post("/api/open-folder")
666def open_folder(
667 request: Request, username: Annotated[str, Depends(require_auth)]
668):
669 """Open folder containing a document.
671 Security: This endpoint is disabled for server deployments.
672 It only makes sense for desktop usage where the server and client are on the same machine.
673 """
674 return JSONResponse(
675 {
676 "status": "error",
677 "message": "This feature is disabled. It is only available in desktop mode.",
678 },
679 status_code=403,
680 )
683@router.post("/api/download/{resource_id}")
684def download_single_resource(
685 request: Request,
686 # Typed int, matching Flask's <int:resource_id> converter: without it a
687 # non-numeric segment is passed straight through to the service layer
688 # instead of being rejected as a 422 by the router.
689 resource_id: int,
690 username: Annotated[str, Depends(require_auth)],
691):
692 """Download a single resource."""
693 user_password = get_authenticated_user_password(username)
695 with DownloadService(username, user_password) as service:
696 success, error = service.download_resource(resource_id)
697 if success:
698 return {"success": True}
699 if error == "Resource not found":
700 # download_resource() returns this exact string (never raises)
701 # when resource_id doesn't exist — that's a 404, not a generic
702 # download failure. Without this branch every nonexistent
703 # resource_id collapsed to the 500 below.
704 return JSONResponse(
705 {"success": False, "error": "Resource not found"},
706 status_code=404,
707 )
708 logger.warning(f"Download failed for resource {resource_id}: {error}")
709 return JSONResponse(
710 {
711 "success": False,
712 "error": "Download failed. Please try again or contact support.",
713 },
714 status_code=500,
715 )
718@router.post("/api/download-text/{resource_id}")
719def download_text_single(
720 request: Request,
721 # Typed int, matching Flask's <int:resource_id> converter: without it a
722 # non-numeric segment is passed straight through to the service layer
723 # instead of being rejected as a 422 by the router.
724 resource_id: int,
725 username: Annotated[str, Depends(require_auth)],
726):
727 """Download a single resource as text file."""
728 try:
729 user_password = get_authenticated_user_password(username)
731 with DownloadService(username, user_password) as service:
732 success, error = service.download_as_text(resource_id)
734 # Sanitize error message - don't expose internal details
735 if not success:
736 if error:
737 logger.warning(
738 f"Download as text failed for resource {resource_id}: {error}"
739 )
740 return {
741 "success": False,
742 "error": "Failed to download resource",
743 }
745 return {"success": True, "error": None}
746 except AuthenticationRequiredError:
747 raise # Let the global WebAPIException handler return 401
748 except Exception as e:
749 return handle_api_error(
750 f"downloading resource {resource_id} as text", e
751 )
754@router.post("/api/download-all-text")
755def download_all_text(
756 request: Request, username: Annotated[str, Depends(require_auth)]
757):
758 """Download all undownloaded resources as text files."""
759 # Capture Flask session ID to avoid scoping issues in nested function
760 session_id = request.session.get("session_id")
762 def generate():
763 # Get user password for database operations
764 try:
765 user_password = get_authenticated_user_password(
766 username, session_id
767 )
768 except AuthenticationRequiredError:
769 logger.warning(
770 f"Authentication unavailable for user {username} - password not in session store"
771 )
772 yield f"data: {json.dumps({'progress': 0, 'current': 0, 'total': 0, 'error': 'Authentication required', 'complete': True})}\n\n"
773 return
775 # This DownloadService (and the SettingsManager session inside
776 # download_service.settings) is built ONCE, before the loop, and
777 # lives across every ``yield`` below. That crosses a cleanup
778 # boundary the wrapper's per-chunk cleanup does NOT know about:
779 # ``WorkerCleanupStreamingResponse`` wraps this generator in
780 # ``iterate_sync_with_cleanup``, whose ``thread_cleanup()`` runs
781 # after EVERY ``next()`` (i.e. after every chunk, not just at the
782 # end) and closes every session cached or tracked on this thread
783 # (``utilities/db_utils.py``). The first chunk that touches
784 # ``download_service.settings`` therefore gets its underlying
785 # SQLAlchemy session closed by that per-chunk cleanup the moment the
786 # chunk's ``yield`` returns -- but the ``SettingsManager``/``Session``
787 # OBJECTS themselves are untouched, only their pooled connection is
788 # released, so the next chunk's use of ``download_service.settings``
789 # transparently reopens a fresh connection (ordinary SQLAlchemy
790 # session behavior after ``close()``). That connection then sits in
791 # neither the LRU cache nor the owner-thread registry until stream
792 # end, when ``safe_close(download_service, ...)`` below closes it
793 # via ``SettingsManager.close()`` -- exactly where it was released
794 # pre-#6095, so this is NOT a leak, just a "cleanup at every
795 # boundary" invariant violated for the longest-lived object in this
796 # stream. Documented rather than restructured: giving this loop its
797 # own re-created-per-chunk settings manager would fix the invariant
798 # but changes this route's session lifetime shape for no behavioral
799 # gain, and isn't warranted without a demonstrated leak.
800 download_service = DownloadService(username, user_password)
801 try:
802 # Fetch the resource rows in a SHORT-LIVED session, snapshotting
803 # only the columns the loop uses into plain tuples. The download
804 # loop below holds NO session while it yields — the held session is
805 # only needed for this query, and keeping a get_user_db_session
806 # scope open across a yield risks the Starlette/anyio thread-
807 # affinity corruption fixed in download_bulk (its __enter__ and
808 # __exit__ can land on different pooled threads). Project only
809 # id/url/title so the content_preview Text column doesn't
810 # materialize on this whole-table scan (#4560).
811 with get_user_db_session(username) as session:
812 all_resources = [
813 (r.id, r.url, r.title)
814 for r in session.query(
815 ResearchResource.id,
816 ResearchResource.url,
817 ResearchResource.title,
818 ).all()
819 ]
821 # Filter to only downloadable resources (academic/PDF)
822 resources = [
823 (rid, url, title)
824 for (rid, url, title) in all_resources
825 if is_downloadable_url(url)
826 ]
828 # Pre-scan directory once to get all existing resource IDs
829 txt_path = Path(download_service.library_root) / "txt"
830 existing_resource_ids = set()
831 if txt_path.exists():
832 for txt_file in txt_path.glob("*.txt"):
833 # Extract resource ID from filename pattern *_{id}.txt
834 parts = txt_file.stem.rsplit("_", 1)
835 if len(parts) == 2: 835 ↛ 832line 835 didn't jump to line 832 because the condition on line 835 was always true
836 try:
837 existing_resource_ids.add(int(parts[1]))
838 except ValueError:
839 pass
841 # Filter resources that need text extraction
842 resources_to_process = [
843 (rid, url, title)
844 for (rid, url, title) in resources
845 if rid not in existing_resource_ids
846 ]
848 total = len(resources_to_process)
849 current = 0
851 logger.info(f"Found {total} resources needing text extraction")
853 for rid, url, title in resources_to_process:
854 current += 1
855 progress = int((current / total) * 100) if total > 0 else 100
857 file_name = title[:50] if title else f"document_{current}.txt"
859 try:
860 # download_as_text opens its own session internally.
861 # Its every error-return path (see download_service.py
862 # download_as_text/_try_api_text_extraction/
863 # _fallback_pdf_extraction) is either a static string or
864 # already scrubbed via sanitize_error_for_client() — no
865 # raw exception text reaches `error` here.
866 success, error = download_service.download_as_text(rid)
868 if success:
869 status = "success"
870 error_msg = None
871 else:
872 status = "failed"
873 error_msg = error or "Text extraction failed"
875 except Exception as e:
876 logger.exception(
877 f"Error extracting text for resource {rid}"
878 )
879 status = "failed"
880 # CWE-209: only the exception CLASS NAME is surfaced,
881 # never str(e) — same "class name only" mitigation as
882 # journal_quality/downloader.py (CodeQL alerts 7650/7684).
883 error_msg = f"Text extraction failed - {type(e).__name__}"
885 # Send update
886 update = {
887 "progress": progress,
888 "current": current,
889 "total": total,
890 "file": file_name,
891 "url": url, # Add the URL for UI display
892 "status": status,
893 "error": error_msg,
894 }
895 yield f"data: {json.dumps(update)}\n\n"
897 # Send completion
898 yield f"data: {json.dumps({'complete': True, 'total': total})}\n\n"
899 finally:
900 from ...utilities.resource_utils import safe_close
902 safe_close(download_service, "download service")
904 # CWE-209 (CodeQL "Information exposure through an exception"): the SSE
905 # body streamed here never carries a raw exception message — see the
906 # `except Exception as e:` block in generate() above, which only ever
907 # surfaces `type(e).__name__` or an already-scrubbed string from
908 # download_service.
909 return WorkerCleanupStreamingResponse(
910 generate(),
911 media_type="text/event-stream",
912 headers={
913 "Cache-Control": "no-cache, no-transform",
914 "X-Accel-Buffering": "no",
915 },
916 )
919@router.post("/api/download-research/{research_id}")
920async def download_research_pdfs(
921 request: Request,
922 research_id,
923 username: Annotated[str, Depends(require_auth)],
924):
925 """Queue all PDFs from a research session for download."""
926 user_password = get_authenticated_user_password(username)
927 data = await request.json()
928 if not isinstance(data, dict):
929 return json_body_error("simple", "Request body must be valid JSON")
930 collection_id = data.get("collection_id")
932 # `queue_research_downloads` does sync SQLAlchemy work. Keep it off
933 # the uvicorn event loop.
934 def _queue_sync():
935 with DownloadService(username, user_password) as service:
936 return service.queue_research_downloads(research_id, collection_id)
938 queued = await run_db_sync(_queue_sync)
939 return {"success": True, "queued": queued}
942def _claim_download_queue_item(db_session, queue_item_id):
943 """Atomically claim a PENDING download-queue row before processing it.
945 Flip the row from PENDING to PROCESSING in a single conditional UPDATE and
946 return True only when this caller won the claim. Two concurrent
947 ``download_bulk`` streams for the same user (the user clicks Download
948 twice, or a second bulk run starts before the first finishes) both read
949 the same PENDING rows and, with no claim, both call ``download_resource``
950 for every row (issue #4691). The row-count guard makes the
951 PENDING -> PROCESSING transition the single point where exactly one stream
952 wins each row; the loser sees 0 updated rows and skips it.
953 """
954 claimed = (
955 db_session.query(LibraryDownloadQueue)
956 .filter_by(id=queue_item_id, status=DocumentStatus.PENDING)
957 .update(
958 {LibraryDownloadQueue.status: DocumentStatus.PROCESSING},
959 synchronize_session=False,
960 )
961 )
962 db_session.commit()
963 return claimed == 1
966def _release_download_queue_item(db_session, queue_item_id):
967 """Return a claimed row from PROCESSING back to PENDING so it stays
968 retryable when the download raised before a terminal COMPLETED/FAILED
969 status was recorded (issue #4691), matching the pre-fix behaviour where an
970 errored row was left PENDING. Scoped to PROCESSING so it only ever undoes
971 this stream's own claim.
972 """
973 db_session.query(LibraryDownloadQueue).filter_by(
974 id=queue_item_id, status=DocumentStatus.PROCESSING
975 ).update(
976 {LibraryDownloadQueue.status: DocumentStatus.PENDING},
977 synchronize_session=False,
978 )
979 db_session.commit()
982def _finalize_download_queue_item(db_session, queue_item_id, success):
983 """Record the outcome of a claimed row that returned without raising.
985 ``download_resource`` writes a terminal COMPLETED/FAILED status itself for
986 the rows it actually downloads, and this is a no-op for those: the row is
987 no longer PROCESSING, so the guard matches nothing and pdf-mode behaviour
988 is unchanged. It fires only on the paths that otherwise strand a claim
989 (issue #4691):
991 - ``mode="text_only"``: ``download_as_text`` and its call tree never touch
992 ``LibraryDownloadQueue``, so a successful extraction would leave the row
993 PROCESSING forever. Nothing in ``src/`` reaps PROCESSING rows, and a
994 successful extraction also writes a COMPLETED Document, which blocks both
995 the pre-pass reset and ``queue_all_undownloaded``'s outer-join filter.
996 - ``download_resource``'s already-downloaded early return, which reports
997 success before reaching its own queue update.
999 Pre-fix, both paths left the row PENDING and merely re-scanned it, so
1000 failure returns to PENDING rather than FAILED to keep that retryability.
1001 """
1002 db_session.query(LibraryDownloadQueue).filter_by(
1003 id=queue_item_id, status=DocumentStatus.PROCESSING
1004 ).update(
1005 {
1006 LibraryDownloadQueue.status: (
1007 DocumentStatus.COMPLETED if success else DocumentStatus.PENDING
1008 ),
1009 LibraryDownloadQueue.completed_at: (
1010 datetime.now(UTC) if success else None
1011 ),
1012 },
1013 synchronize_session=False,
1014 )
1015 db_session.commit()
1018@router.post("/api/download-bulk")
1019async def download_bulk(
1020 request: Request, username: Annotated[str, Depends(require_auth)]
1021):
1022 """Download PDFs or extract text from multiple research sessions."""
1023 data = await request.json()
1024 if not isinstance(data, dict): 1024 ↛ 1025line 1024 didn't jump to line 1025 because the condition on line 1024 was never true
1025 return json_body_error("simple", "Request body must be valid JSON")
1026 research_ids = data.get("research_ids", [])
1027 mode = data.get("mode", "pdf") # pdf or text_only
1028 collection_id = data.get(
1029 "collection_id"
1030 ) # Optional: target collection for downloads
1032 if not research_ids:
1033 return JSONResponse(
1034 {"error": "No research IDs provided"}, status_code=400
1035 )
1036 if error := _string_list_error(research_ids, "research_ids"):
1037 return error
1038 if not isinstance(mode, str) or mode not in {"pdf", "text_only"}:
1039 return JSONResponse(
1040 {"error": "mode must be either 'pdf' or 'text_only'"},
1041 status_code=400,
1042 )
1044 # Capture Flask session ID to avoid scoping issues in nested function
1045 session_id = request.session.get("session_id")
1047 def generate():
1048 """Generate progress updates as Server-Sent Events."""
1049 # Get user password for database operations
1050 try:
1051 user_password = get_authenticated_user_password(
1052 username, session_id
1053 )
1054 except AuthenticationRequiredError:
1055 logger.warning(
1056 f"Authentication unavailable for user {username} - password not in session store"
1057 )
1058 yield f"data: {json.dumps({'progress': 0, 'current': 0, 'total': 0, 'error': 'Authentication required', 'complete': True})}\n\n"
1059 return
1061 # Same per-chunk-cleanup caveat as download_all_text() above:
1062 # download_service.settings' session gets closed (connection
1063 # released, object untouched) by iterate_sync_with_cleanup's
1064 # per-chunk thread_cleanup() after the first chunk that uses it, and
1065 # transparently reopens a fresh connection on later chunks. Not a
1066 # leak -- safe_close(download_service, ...) below releases it at
1067 # stream end exactly as pre-#6095 -- just documented here rather
1068 # than restructured; see the fuller note there.
1069 download_service = DownloadService(username, user_password)
1070 try:
1071 total = 0
1072 current = 0
1073 queue_failures = 0
1075 # Pre-queue and count in one merged loop. queue_research_downloads
1076 # opens its own session, commits, and handles dedup internally:
1077 # it skips resources that are already PENDING or already backed by
1078 # a COMPLETED Document, and re-queues any remaining row (e.g. a
1079 # prior FAILED attempt) back to PENDING. Calling it BEFORE counting
1080 # ensures the total reflects what will actually be processed
1081 # (issue #4660).
1082 for research_id in research_ids:
1083 try:
1084 download_service.queue_research_downloads(
1085 research_id, collection_id
1086 )
1087 except Exception:
1088 queue_failures += 1
1089 logger.exception(
1090 f"Error queueing downloads for "
1091 f"user={username} research={research_id} "
1092 f"collection={collection_id}"
1093 )
1095 with get_user_db_session(username) as session:
1096 count = (
1097 session.query(LibraryDownloadQueue)
1098 .filter_by(
1099 research_id=research_id,
1100 status=DocumentStatus.PENDING,
1101 )
1102 .count()
1103 )
1104 total += count
1105 logger.debug(
1106 f"Research {research_id}: {count} pending items in queue"
1107 )
1109 logger.info(f"Total pending downloads across all research: {total}")
1110 yield f"data: {json.dumps({'progress': 0, 'current': 0, 'total': total})}\n\n"
1112 # If the pre-pass yielded nothing to process, surface it as an
1113 # error so the UI alerts instead of silently completing with
1114 # "0 / 0 files" success. Distinguish queue failure from a
1115 # legitimate "nothing left to download" state.
1116 if total == 0:
1117 if queue_failures > 0:
1118 error_msg = (
1119 f"Download failed to start: queueing failed for "
1120 f"{queue_failures} of {len(research_ids)} "
1121 f"research session(s). Check server logs."
1122 )
1123 else:
1124 error_msg = (
1125 "No new papers to download for the selected "
1126 "research session(s)."
1127 )
1128 yield f"data: {json.dumps({'progress': 100, 'current': 0, 'total': 0, 'complete': True, 'error': error_msg})}\n\n"
1129 return
1131 # Process each research
1132 for research_id in research_ids:
1133 # Get queued downloads for this research in a SHORT-LIVED
1134 # session, snapshotting only the ids we need. The per-item work
1135 # below opens its own short-lived sessions so that NO
1136 # get_user_db_session scope is ever held across a ``yield``:
1137 # generate() is a sync generator that Starlette drives via
1138 # iterate_in_threadpool (one anyio dispatch per next()), so a
1139 # session whose __enter__/__exit__ straddled a yield could run
1140 # on two different pooled OS threads and leave the thread-local
1141 # scope_depth counter permanently corrupted.
1142 with get_user_db_session(username) as session:
1143 queue_items = [
1144 (qi.id, qi.resource_id)
1145 for qi in (
1146 session.query(LibraryDownloadQueue)
1147 .filter_by(
1148 research_id=research_id,
1149 status=DocumentStatus.PENDING,
1150 )
1151 .all()
1152 )
1153 ]
1155 # Process each queued item
1156 for queue_item_id, resource_id in queue_items:
1157 # Atomically claim this row before downloading so a
1158 # second concurrent download_bulk stream can't also
1159 # process it (issue #4691). A row already claimed by
1160 # another stream returns 0 updated rows; skip it.
1161 with get_user_db_session(username) as session:
1162 claimed = _claim_download_queue_item(
1163 session, queue_item_id
1164 )
1165 if not claimed: 1165 ↛ 1166line 1165 didn't jump to line 1166 because the condition on line 1165 was never true
1166 continue
1168 current += 1
1170 progress = (
1171 int((current / total) * 100) if total > 0 else 100
1172 )
1174 # Attempt actual download with error handling
1175 skip_reason = None
1176 status = "skipped" # Default to skipped
1177 success = False
1178 error_msg = None
1179 error_type = None
1180 # Everything from the resource lookup onward runs
1181 # inside the try below, so file_name needs a value
1182 # even if that block raises before setting one.
1183 file_name = f"document_{current}.pdf"
1185 try:
1186 # Get resource info in its own short-lived session.
1187 # This sits INSIDE the try because the claim above is
1188 # already committed as PROCESSING: any raise between
1189 # the claim and this handler would leave the row that
1190 # way forever, since nothing in src/ reaps PROCESSING
1191 # rows and every reset path now refuses to touch them
1192 # (issue #4691). ResearchResource.title is nullable, so
1193 # resource.title[:50] raises TypeError on a resource
1194 # whose title was never populated.
1195 with get_user_db_session(username) as session:
1196 resource = session.get(
1197 ResearchResource, resource_id
1198 )
1199 if resource and resource.title:
1200 file_name = resource.title[:50]
1202 logger.debug(
1203 f"Attempting {'PDF download' if mode == 'pdf' else 'text extraction'} for resource {resource_id}"
1204 )
1206 # Call appropriate service method based on mode.
1207 # DownloadService opens its own session internally; no
1208 # session is held here across the blocking download.
1209 if mode == "pdf":
1210 result = download_service.download_resource(
1211 resource_id
1212 )
1213 else: # text_only
1214 result = download_service.download_as_text(
1215 resource_id
1216 )
1218 # Handle new tuple return format
1219 if isinstance(result, tuple): 1219 ↛ 1222line 1219 didn't jump to line 1222 because the condition on line 1219 was always true
1220 success, skip_reason = result
1221 else:
1222 success = result
1223 skip_reason = None
1225 # Finalize the claim for the paths that don't
1226 # record a terminal status themselves (text_only,
1227 # and download_resource's already-downloaded early
1228 # return); a no-op when they already did (#4691).
1229 with get_user_db_session(username) as session:
1230 _finalize_download_queue_item(
1231 session, queue_item_id, success
1232 )
1234 status = "success" if success else "skipped"
1235 if skip_reason and not success:
1236 error_msg = skip_reason
1237 logger.info(
1238 f"{'Download' if mode == 'pdf' else 'Text extraction'} skipped for resource {resource_id}: {skip_reason}"
1239 )
1241 logger.debug(
1242 f"{'Download' if mode == 'pdf' else 'Text extraction'} result: success={success}, status={status}, skip_reason={skip_reason}"
1243 )
1244 except Exception as e:
1245 # Release the claim so a later run can retry this row:
1246 # the download raised before download_resource recorded
1247 # a terminal status (issue #4691). A fresh short-lived
1248 # session — the download's own session is unrelated and
1249 # each op above already closed its scope, so there is no
1250 # held session to roll back.
1251 try:
1252 with get_user_db_session(username) as session:
1253 _release_download_queue_item(
1254 session, queue_item_id
1255 )
1256 except Exception:
1257 logger.exception(
1258 "Failed to release download-queue claim"
1259 )
1260 # Log error but continue processing. `error_msg`
1261 # (str(e)) is used ONLY for the keyword classification
1262 # below (`.lower()` checks) and the server log line —
1263 # it is never written into `skip_reason`, which the
1264 # yield below sends to the client. Only `error_type`
1265 # (the exception CLASS NAME) reaches skip_reason, same
1266 # "class name only" mitigation as
1267 # journal_quality/downloader.py (CodeQL alerts
1268 # 7650/7684) — CWE-209.
1269 error_msg = str(e)
1270 error_type = type(e).__name__
1271 logger.warning(
1272 f"CAUGHT Download exception for resource {resource_id}: {error_type}: {error_msg}"
1273 )
1274 # Check if this is a skip reason (not a real error)
1275 # Use error category + categorized message for user display
1276 if any(
1277 phrase in error_msg.lower()
1278 for phrase in [
1279 "paywall",
1280 "subscription",
1281 "not available",
1282 "not found",
1283 "no free",
1284 "embargoed",
1285 "forbidden",
1286 "not accessible",
1287 ]
1288 ):
1289 status = "skipped"
1290 skip_reason = f"Document not accessible (paywall or access restriction) - {error_type}"
1291 elif any(
1292 phrase in error_msg.lower()
1293 for phrase in [
1294 "failed to download",
1295 "could not",
1296 "invalid",
1297 "server",
1298 ]
1299 ):
1300 status = "failed"
1301 skip_reason = f"Download failed - {error_type}"
1302 else:
1303 status = "failed"
1304 skip_reason = f"Processing failed - {error_type}"
1305 success = False
1307 # Ensure skip_reason is set if we have an error message
1308 if error_msg and not skip_reason: 1308 ↛ 1309line 1308 didn't jump to line 1309 because the condition on line 1308 was never true
1309 skip_reason = f"Processing failed - {error_type}"
1310 logger.debug(
1311 f"Setting skip_reason from error_msg: {error_msg}"
1312 )
1314 # Send progress update
1315 update_data = {
1316 "progress": progress,
1317 "current": current,
1318 "total": total,
1319 "file": file_name,
1320 "status": status,
1321 }
1322 # Add skip reason if available
1323 if skip_reason:
1324 update_data["error"] = skip_reason
1325 logger.info(f"Sending skip reason to UI: {skip_reason}")
1327 yield f"data: {json.dumps(update_data)}\n\n"
1329 yield f"data: {json.dumps({'progress': 100, 'current': current, 'total': total, 'complete': True})}\n\n"
1330 finally:
1331 from ...utilities.resource_utils import safe_close
1333 safe_close(download_service, "download service")
1335 # CWE-209 (CodeQL "Information exposure through an exception"): the SSE
1336 # body streamed here never carries a raw exception message. `skip_reason`
1337 # either comes straight from download_service (already a static string or
1338 # sanitize_error_for_client()-scrubbed — see download_resource /
1339 # download_as_text and their helpers in download_service.py) or, on the
1340 # `except Exception as e:` path above, is built from `type(e).__name__`
1341 # only — never the raw `error_msg = str(e)`.
1342 return WorkerCleanupStreamingResponse(
1343 generate(),
1344 media_type="text/event-stream",
1345 headers={
1346 "Cache-Control": "no-cache, no-transform",
1347 "X-Accel-Buffering": "no",
1348 },
1349 )
1352@router.get("/api/research-list")
1353def get_research_list(
1354 request: Request, username: Annotated[str, Depends(require_auth)]
1355):
1356 """Get the lightweight research list used to populate a dropdown.
1358 Deliberately NOT ``get_research_list_with_stats``: that runs a 3-table
1359 GROUP BY aggregate plus follow-up ratings and domain-breakdown queries,
1360 which a <select> needing only id/title/query does not use. Main replaced
1361 this call for exactly that reason.
1362 """
1363 service = LibraryService(username)
1364 research_list = service.get_research_list_for_dropdown()
1365 return {"research": research_list}
1368@router.post("/api/sync-library")
1369def sync_library(
1370 request: Request, username: Annotated[str, Depends(require_auth)]
1371):
1372 """Sync library database with filesystem."""
1373 service = LibraryService(username)
1374 return service.sync_library_with_filesystem()
1377@router.post("/api/mark-redownload")
1378async def mark_for_redownload(
1379 request: Request, username: Annotated[str, Depends(require_auth)]
1380):
1381 """Mark documents for re-download."""
1382 data = await request.json()
1383 if not isinstance(data, dict): 1383 ↛ 1384line 1383 didn't jump to line 1384 because the condition on line 1383 was never true
1384 return json_body_error("simple", "Request body must be valid JSON")
1385 document_ids = data.get("document_ids", [])
1387 if not document_ids:
1388 return JSONResponse(
1389 {"error": "No document IDs provided"}, status_code=400
1390 )
1391 if error := _string_list_error(document_ids, "document_ids"):
1392 return error
1394 def _mark_sync():
1395 service = LibraryService(username)
1396 return service.mark_for_redownload(document_ids)
1398 count = await run_db_sync(_mark_sync)
1399 return {"success": True, "marked": count}
1402@router.post("/api/queue-all-undownloaded")
1403def queue_all_undownloaded(
1404 request: Request, username: Annotated[str, Depends(require_auth)]
1405):
1406 """Queue all articles that haven't been downloaded yet."""
1408 logger.info(f"queue_all_undownloaded called for user {username}")
1410 with get_user_db_session(username) as db_session:
1411 # Find all resources that don't have a completed download. Project only
1412 # the columns the filter/loop use (id/url/research_id) instead of full
1413 # ``ResearchResource`` entities so the ``content_preview`` Text column
1414 # isn't materialized for every row on this whole-table scan (#4560).
1415 undownloaded = (
1416 db_session.query(
1417 ResearchResource.id,
1418 ResearchResource.url,
1419 ResearchResource.research_id,
1420 )
1421 .outerjoin(
1422 Document,
1423 (
1424 (ResearchResource.id == Document.resource_id)
1425 | (ResearchResource.document_id == Document.id)
1426 )
1427 & (Document.status == "completed"),
1428 )
1429 .filter(Document.id.is_(None))
1430 .all()
1431 )
1433 logger.info(f"Found {len(undownloaded)} total undownloaded resources")
1435 # Get user password for encrypted database access
1436 user_password = get_authenticated_user_password(username)
1438 resource_filter = ResourceFilter(username, user_password)
1439 filter_results = resource_filter.filter_downloadable_resources(
1440 undownloaded
1441 )
1443 # Get detailed filtering summary
1444 filter_summary = resource_filter.get_filter_summary(undownloaded)
1445 skipped_info = resource_filter.get_skipped_resources_info(undownloaded)
1447 logger.info(f"Filter results: {filter_summary.to_dict()}")
1449 queued_count = 0
1450 research_ids = set()
1451 skipped_count = 0
1453 # Convert filter_results to dict for O(1) lookup instead of O(n²)
1454 filter_results_by_id = {r.resource_id: r for r in filter_results}
1456 for resource in undownloaded:
1457 # Check if resource passed the smart filter
1458 filter_result = filter_results_by_id.get(resource.id)
1460 if not filter_result or not filter_result.can_retry:
1461 skipped_count += 1
1462 if filter_result:
1463 logger.debug(
1464 f"Skipping resource {resource.id} due to retry policy: {filter_result.reason}"
1465 )
1466 else:
1467 logger.debug(
1468 f"Skipping resource {resource.id} - no filter result available"
1469 )
1470 continue
1472 # Check if it's downloadable using proper URL parsing
1473 if not resource.url:
1474 skipped_count += 1
1475 continue
1477 is_downloadable = is_downloadable_domain(resource.url)
1479 # Log what we're checking
1480 if resource.url and "pubmed" in resource.url.lower(): 1480 ↛ 1481line 1480 didn't jump to line 1481 because the condition on line 1480 was never true
1481 logger.info(f"Found PubMed URL: {resource.url[:100]}")
1483 if not is_downloadable: 1483 ↛ 1484line 1483 didn't jump to line 1484 because the condition on line 1483 was never true
1484 skipped_count += 1
1485 logger.debug(
1486 f"Skipping non-downloadable URL: {resource.url[:100] if resource.url else 'None'}"
1487 )
1488 continue
1490 # Check if already in queue (any status)
1491 existing_queue = (
1492 db_session.query(LibraryDownloadQueue)
1493 .filter_by(resource_id=resource.id)
1494 .first()
1495 )
1497 if existing_queue:
1498 # Reset a non-pending row back to PENDING so it is retried,
1499 # but never one a concurrent download stream has claimed:
1500 # PROCESSING satisfies `!= PENDING`, so an unguarded reset
1501 # un-claims an in-flight row and lets a second stream
1502 # re-download the same resource (issue #4691). The query above
1503 # selects in-flight rows, since a download that has not
1504 # finished has no completed Document yet. The guard lives
1505 # inside the UPDATE, not in an `if` above it, so the check and
1506 # the write cannot straddle another stream's claim.
1507 reset = (
1508 db_session.query(LibraryDownloadQueue)
1509 .filter(
1510 LibraryDownloadQueue.id == existing_queue.id,
1511 LibraryDownloadQueue.status != DocumentStatus.PENDING,
1512 LibraryDownloadQueue.status
1513 != DocumentStatus.PROCESSING,
1514 )
1515 .update(
1516 {
1517 LibraryDownloadQueue.status: DocumentStatus.PENDING,
1518 LibraryDownloadQueue.completed_at: None,
1519 },
1520 synchronize_session=False,
1521 )
1522 )
1523 # Counted either way, matching the pre-fix behaviour: a row
1524 # already PENDING, a row just reset, and a row another stream
1525 # is downloading are all "queued" to the caller.
1526 queued_count += 1
1527 research_ids.add(resource.research_id)
1528 if reset:
1529 logger.debug(
1530 f"Reset queue entry for resource {resource.id} to pending"
1531 )
1532 else:
1533 logger.debug(
1534 f"Resource {resource.id} already pending or in flight"
1535 )
1536 else:
1537 # Add new entry to queue
1538 queue_entry = LibraryDownloadQueue(
1539 resource_id=resource.id,
1540 research_id=resource.research_id,
1541 priority=0,
1542 status=DocumentStatus.PENDING,
1543 )
1544 db_session.add(queue_entry)
1545 queued_count += 1
1546 research_ids.add(resource.research_id)
1547 logger.debug(
1548 f"Added new queue entry for resource {resource.id}"
1549 )
1551 db_session.commit()
1553 logger.info(
1554 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)"
1555 )
1557 # Note: Removed synchronous download processing here to avoid blocking the HTTP request
1558 # Downloads will be processed via the SSE streaming endpoint or background tasks
1560 return {
1561 "success": True,
1562 "queued": queued_count,
1563 "research_ids": list(research_ids),
1564 "total_undownloaded": len(undownloaded),
1565 "skipped": skipped_count,
1566 "filter_summary": filter_summary.to_dict(),
1567 "skipped_details": skipped_info,
1568 }
1571@router.get("/api/get-research-sources/{research_id}")
1572def get_research_sources(
1573 request: Request,
1574 research_id,
1575 username: Annotated[str, Depends(require_auth)],
1576):
1577 """Get all sources for a research with snippets."""
1579 sources = []
1580 with get_user_db_session(username) as db_session:
1581 # Get all resources for this research
1582 resources = (
1583 db_session.query(ResearchResource)
1584 .filter_by(research_id=research_id)
1585 .order_by(ResearchResource.created_at)
1586 .all()
1587 )
1589 for idx, resource in enumerate(resources, 1):
1590 # Check if document exists
1591 document = get_document_for_resource(db_session, resource)
1593 # Get domain from URL
1594 domain = ""
1595 if resource.url:
1596 try:
1597 from urllib.parse import urlparse
1599 domain = urlparse(resource.url).hostname or ""
1600 except (ValueError, AttributeError):
1601 # urlparse can raise ValueError for malformed URLs
1602 pass
1604 source_data = {
1605 "number": idx,
1606 "resource_id": resource.id,
1607 "url": resource.url,
1608 "title": resource.title or f"Source {idx}",
1609 "snippet": resource.content_preview or "",
1610 "domain": domain,
1611 "relevance_score": getattr(resource, "relevance_score", None),
1612 "downloaded": False,
1613 "document_id": None,
1614 "file_type": None,
1615 }
1617 if document and document.status == "completed":
1618 source_data.update(
1619 {
1620 "downloaded": True,
1621 "document_id": document.id,
1622 "file_type": document.file_type,
1623 "download_date": document.created_at.isoformat()
1624 if document.created_at
1625 else None,
1626 }
1627 )
1629 sources.append(source_data)
1631 return {"success": True, "sources": sources, "total": len(sources)}
1634@router.post("/api/check-downloads")
1635async def check_downloads(
1636 request: Request, username: Annotated[str, Depends(require_auth)]
1637):
1638 """Check download status for a list of URLs."""
1639 data = await request.json()
1640 if not isinstance(data, dict): 1640 ↛ 1641line 1640 didn't jump to line 1641 because the condition on line 1640 was never true
1641 return json_body_error("simple", "Request body must be valid JSON")
1642 research_id = data.get("research_id")
1643 urls = data.get("urls", [])
1645 if not isinstance(research_id, str) or not research_id.strip() or not urls:
1646 return JSONResponse(
1647 {"error": "Missing research_id or urls"}, status_code=400
1648 )
1649 if error := _string_list_error(urls, "urls"):
1650 return error
1652 def _query_sync():
1653 download_status = {}
1654 with get_user_db_session(username) as db_session:
1655 resources = (
1656 db_session.query(ResearchResource)
1657 .filter_by(research_id=research_id)
1658 .filter(ResearchResource.url.in_(urls))
1659 .all()
1660 )
1661 for resource in resources:
1662 document = get_document_for_resource(db_session, resource)
1663 if document and document.status == "completed":
1664 # Do NOT return document.file_path — it's the absolute
1665 # server path, which leaks directory layout to any
1666 # authenticated user. The client fetches via
1667 # `/document/{id}/pdf` or `/document/{id}/text` using
1668 # document_id, so the path is never needed client-side.
1669 download_status[resource.url] = {
1670 "downloaded": True,
1671 "document_id": document.id,
1672 "file_type": document.file_type,
1673 "title": document.title or resource.title,
1674 }
1675 else:
1676 download_status[resource.url] = {
1677 "downloaded": False,
1678 "resource_id": resource.id,
1679 }
1680 return download_status
1682 download_status = await run_db_sync(_query_sync)
1683 return {"download_status": download_status}
1686@router.post("/api/download-source")
1687async def download_source(
1688 request: Request, username: Annotated[str, Depends(require_auth)]
1689):
1690 """Download a single source from a research."""
1691 user_password = get_authenticated_user_password(username)
1692 data = await request.json()
1693 if not isinstance(data, dict): 1693 ↛ 1694line 1693 didn't jump to line 1694 because the condition on line 1693 was never true
1694 return json_body_error("simple", "Request body must be valid JSON")
1695 research_id = data.get("research_id")
1696 url = data.get("url")
1698 if not research_id or not url:
1699 return JSONResponse(
1700 {"error": "Missing research_id or url"}, status_code=400
1701 )
1703 # Check if URL is downloadable
1704 if not is_downloadable_domain(url):
1705 return JSONResponse(
1706 {"error": "URL is not from a downloadable domain"}, status_code=400
1707 )
1709 def _download_sync():
1710 """The DB + network work — all blocking, run in a threadpool."""
1711 with get_user_db_session(username) as db_session:
1712 # Find the resource
1713 resource = (
1714 db_session.query(ResearchResource)
1715 .filter_by(research_id=research_id, url=url)
1716 .first()
1717 )
1719 if not resource:
1720 return JSONResponse(
1721 {"error": "Resource not found"}, status_code=404
1722 )
1724 # Check if already downloaded
1725 existing = get_document_for_resource(db_session, resource)
1727 if existing and existing.status == "completed":
1728 return {
1729 "success": True,
1730 "message": "Already downloaded",
1731 "document_id": existing.id,
1732 }
1734 # Add to download queue
1735 queue_entry = (
1736 db_session.query(LibraryDownloadQueue)
1737 .filter_by(resource_id=resource.id)
1738 .first()
1739 )
1741 if not queue_entry:
1742 queue_entry = LibraryDownloadQueue(
1743 resource_id=resource.id,
1744 research_id=resource.research_id,
1745 priority=1, # Higher priority for manual downloads
1746 status=DocumentStatus.PENDING,
1747 )
1748 db_session.add(queue_entry)
1749 db_session.commit()
1750 else:
1751 # Reset the row so this manual download can claim it, but never
1752 # one a concurrent stream is already downloading: the reset is
1753 # followed immediately by a download in this request thread, so
1754 # un-claiming an in-flight row starts a second download of the
1755 # same resource (issue #4691). The already-downloaded early
1756 # return above cannot prevent that, because a download still in
1757 # flight has not written a completed Document yet.
1758 reset = (
1759 db_session.query(LibraryDownloadQueue)
1760 .filter(
1761 LibraryDownloadQueue.id == queue_entry.id,
1762 LibraryDownloadQueue.status
1763 != DocumentStatus.PROCESSING,
1764 )
1765 .update(
1766 {
1767 LibraryDownloadQueue.status: DocumentStatus.PENDING,
1768 LibraryDownloadQueue.priority: 1,
1769 },
1770 synchronize_session=False,
1771 )
1772 )
1773 db_session.commit()
1774 if not reset:
1775 return JSONResponse(
1776 {
1777 "success": False,
1778 "message": "Download already in progress",
1779 },
1780 status_code=409,
1781 )
1783 queue_item_id = queue_entry.id
1785 # Claim the row before downloading, exactly as download_bulk does,
1786 # so a bulk stream that reaches this row between the reset above and
1787 # the download below cannot also process it (issue #4691).
1788 if not _claim_download_queue_item(db_session, queue_item_id):
1789 return JSONResponse(
1790 {
1791 "success": False,
1792 "message": "Download already in progress",
1793 },
1794 status_code=409,
1795 )
1797 # Start download immediately
1798 try:
1799 with DownloadService(username, user_password) as service:
1800 success, message = service.download_resource(resource.id)
1801 except Exception:
1802 safe_rollback(db_session, "download_source")
1803 # Release the claim so a later run can retry this row: the
1804 # download raised before download_resource recorded a terminal
1805 # status (issue #4691).
1806 _release_download_queue_item(db_session, queue_item_id)
1807 raise
1809 # download_resource writes a terminal status for the rows it
1810 # actually downloads, so this is a no-op for those; it fires on the
1811 # paths that don't (its already-downloaded early return), which
1812 # would otherwise strand the claim in PROCESSING.
1813 _finalize_download_queue_item(db_session, queue_item_id, success)
1815 if success:
1816 return {"success": True, "message": "Download completed"}
1817 # Log internal message, but show only generic message to user
1818 return {"success": False, "message": "Download failed"}
1820 # Offload the blocking DB + HTTP work so we don't stall the event loop.
1821 return await run_db_sync(_download_sync)