Coverage for src/local_deep_research/research_library/utils/__init__.py: 91%
116 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-19 23:35 +0000
1"""Shared utility functions for the Research Library."""
3import hashlib
4import os
5import subprocess
6import sys
7from pathlib import Path
8from typing import Optional
9from urllib.parse import urlparse
11from flask import jsonify
12from loguru import logger
14from ...config.paths import get_library_directory
15from ...database.models.library import Document, DocumentCollection
16from ...security.path_validator import PathValidator
19def escape_like(text: str) -> str:
20 """Escape SQL LIKE/ILIKE wildcards so user input matches literally.
22 ``%`` and ``_`` are LIKE wildcards and ``\\`` is the escape character;
23 without escaping, a query like ``my_note`` would treat ``_`` as "any
24 character" and ``%`` would match anything. Use the result with
25 ``.like(pattern, escape="\\\\")`` / ``.ilike(pattern, escape="\\\\")``.
27 This is the single source of truth for LIKE-escaping across the library
28 and notes query paths (previously copy-pasted in NoteService,
29 LibraryService and unified_search_routes).
30 """
31 return (
32 (text or "")
33 .replace("\\", "\\\\")
34 .replace("%", "\\%")
35 .replace("_", "\\_")
36 )
39def is_downloadable_domain(url: str) -> bool:
40 """Check if URL is from a downloadable academic domain using proper URL parsing."""
41 try:
42 if not url:
43 return False
45 parsed = urlparse(url.lower())
46 hostname = parsed.hostname or ""
47 path = parsed.path or ""
48 query = parsed.query or ""
50 # Check for direct PDF files
51 if path.endswith(".pdf") or ".pdf?" in url.lower():
52 return True
54 # List of downloadable academic domains
55 downloadable_domains = [
56 "arxiv.org",
57 "biorxiv.org",
58 "medrxiv.org",
59 "ncbi.nlm.nih.gov",
60 "pubmed.ncbi.nlm.nih.gov",
61 "europepmc.org",
62 "semanticscholar.org",
63 "researchgate.net",
64 "academia.edu",
65 "sciencedirect.com",
66 "springer.com",
67 "nature.com",
68 "wiley.com",
69 "ieee.org",
70 "acm.org",
71 "plos.org",
72 "frontiersin.org",
73 "mdpi.com",
74 "acs.org",
75 "rsc.org",
76 "tandfonline.com",
77 "sagepub.com",
78 "oxford.com",
79 "cambridge.org",
80 "bmj.com",
81 "nejm.org",
82 "thelancet.com",
83 "jamanetwork.com",
84 "annals.org",
85 "ahajournals.org",
86 "cell.com",
87 "science.org",
88 "pnas.org",
89 "elifesciences.org",
90 "embopress.org",
91 "journals.asm.org",
92 "microbiologyresearch.org",
93 "jvi.asm.org",
94 "genome.cshlp.org",
95 "genetics.org",
96 "g3journal.org",
97 "plantphysiol.org",
98 "plantcell.org",
99 "aspb.org",
100 "bioone.org",
101 "company-of-biologists.org",
102 "biologists.org",
103 "jeb.biologists.org",
104 "dmm.biologists.org",
105 "bio.biologists.org",
106 "doi.org",
107 "ssrn.com",
108 "openreview.net",
109 ]
111 # Check if hostname matches any downloadable domain
112 for domain in downloadable_domains:
113 if hostname == domain or hostname.endswith("." + domain):
114 return True
116 # Special case for PubMed which might appear in path
117 if "pubmed" in hostname or "/pubmed/" in path: 117 ↛ 118line 117 didn't jump to line 118 because the condition on line 117 was never true
118 return True
120 # Check for PDF in path or query parameters
121 if "/pdf/" in path or "type=pdf" in query or "format=pdf" in query:
122 return True
124 return False
126 except Exception:
127 logger.warning(f"Error parsing URL {url}")
128 return False
131def is_downloadable_url(url: str) -> bool:
132 """Check if a URL is downloadable (academic domain or direct PDF link).
134 This is the single source of truth for downloadability checks.
135 Combines domain checking with PDF extension/path detection.
137 Args:
138 url: The URL to check
140 Returns:
141 True if the URL is from a downloadable academic domain or is a direct PDF link
142 """
143 return is_downloadable_domain(url)
146def get_document_for_resource(session, resource):
147 """Get Document for a ResearchResource.
149 Checks resource.document_id first (library resources point directly
150 to existing Documents), falls back to Document.resource_id lookup
151 (web downloads create Documents with resource_id set).
152 """
153 if resource.document_id:
154 return (
155 session.query(Document).filter_by(id=resource.document_id).first()
156 )
157 return session.query(Document).filter_by(resource_id=resource.id).first()
160def get_url_hash(url: str) -> str:
161 """
162 Generate a SHA256 hash of a URL.
164 Args:
165 url: The URL to hash
167 Returns:
168 The SHA256 hash of the URL
169 """
170 return hashlib.sha256(url.lower().encode()).hexdigest()
173def ensure_in_collection(
174 session, document_id: str, collection_id: str
175) -> "DocumentCollection":
176 """Get or create a DocumentCollection link between a document and a collection.
178 Args:
179 session: SQLAlchemy session
180 document_id: UUID of the document
181 collection_id: UUID of the collection
183 Returns:
184 The existing or newly created DocumentCollection row
185 """
186 existing = (
187 session.query(DocumentCollection)
188 .filter_by(document_id=document_id, collection_id=collection_id)
189 .first()
190 )
191 if existing:
192 return existing
194 doc_collection = DocumentCollection(
195 document_id=document_id,
196 collection_id=collection_id,
197 indexed=False,
198 )
199 session.add(doc_collection)
200 return doc_collection
203def get_library_storage_path(username: str) -> Path:
204 """
205 Get the storage path for a user's library.
207 Uses the settings system which respects environment variable overrides:
208 - research_library.storage_path: Base path for library storage
209 - research_library.shared_library: If true, all users share the same directory
211 Args:
212 username: The username
214 Returns:
215 Path to the library storage directory
216 """
217 from ...utilities.db_utils import get_settings_manager
219 settings = get_settings_manager()
221 # Get the base path from settings (uses centralized path, respects LDR_DATA_DIR)
222 base_path = (
223 Path(
224 settings.get_setting(
225 "research_library.storage_path",
226 str(get_library_directory()),
227 )
228 )
229 .expanduser()
230 .resolve()
231 )
233 # Check if shared library mode is enabled
234 shared_library = settings.get_setting(
235 "research_library.shared_library", False
236 )
238 if shared_library:
239 # Shared mode: all users use the same directory
240 base_path.mkdir(parents=True, exist_ok=True)
241 return base_path
242 # Default: user isolation with subdirectories
243 user_dir = base_path / username
244 user_dir.mkdir(parents=True, exist_ok=True)
245 return user_dir
248def open_file_location(file_path: str) -> bool:
249 """
250 Open the file location in the system file manager.
252 Args:
253 file_path: Path to the file
255 Returns:
256 True if successful, False otherwise
257 """
258 try:
259 # Validate path is safe (blocks system dirs, path traversal)
260 validated = PathValidator.validate_local_filesystem_path(file_path)
261 folder = str(validated.parent)
262 if sys.platform == "win32": 262 ↛ 263line 262 didn't jump to line 263 because the condition on line 262 was never true
263 os.startfile(folder)
264 elif sys.platform == "darwin": # macOS
265 result = subprocess.run(
266 ["open", folder], capture_output=True, text=True, shell=False
267 )
268 if result.returncode != 0: 268 ↛ 269line 268 didn't jump to line 269 because the condition on line 268 was never true
269 logger.error(f"Failed to open folder on macOS: {result.stderr}")
270 return False
271 else: # Linux
272 result = subprocess.run(
273 ["xdg-open", folder],
274 capture_output=True,
275 text=True,
276 shell=False,
277 )
278 if result.returncode != 0:
279 logger.error(f"Failed to open folder on Linux: {result.stderr}")
280 return False
281 return True
282 except Exception:
283 logger.exception("Failed to open file location")
284 return False
287def get_absolute_library_path(
288 relative_path: str, username: str
289) -> Optional[Path]:
290 """
291 Get the absolute path from a relative library path.
293 Uses PathValidator to prevent path traversal attacks.
295 Args:
296 relative_path: The relative path from library root
297 username: The username
299 Returns:
300 The absolute path, or None if the path is unsafe
301 """
302 library_root = get_library_storage_path(username)
303 try:
304 # Use PathValidator to prevent path traversal attacks
305 safe_path = PathValidator.validate_safe_path(
306 relative_path, str(library_root)
307 )
308 if safe_path is None: 308 ↛ 309line 308 didn't jump to line 309 because the condition on line 308 was never true
309 return None
310 result = Path(safe_path)
311 if result.is_symlink():
312 logger.warning(f"Symlink blocked: {relative_path}")
313 return None
314 return result
315 except ValueError:
316 logger.warning(f"Path traversal blocked: {relative_path}")
317 return None
320def get_absolute_path_from_settings(relative_path: str) -> Optional[Path]:
321 """
322 Get absolute path using settings manager for library root.
324 Uses PathValidator to prevent path traversal attacks.
326 Args:
327 relative_path: The relative path from library root
329 Returns:
330 The absolute path, or None if the path is unsafe
331 """
332 from ...utilities.db_utils import get_settings_manager
334 settings = get_settings_manager()
335 library_root = (
336 Path(
337 settings.get_setting(
338 "research_library.storage_path",
339 str(get_library_directory()),
340 )
341 )
342 .expanduser()
343 .resolve()
344 )
346 if not relative_path:
347 return library_root
349 try:
350 # Use PathValidator to prevent path traversal attacks
351 safe_path = PathValidator.validate_safe_path(
352 relative_path, str(library_root)
353 )
354 if safe_path is None: 354 ↛ 355line 354 didn't jump to line 355 because the condition on line 354 was never true
355 return None
356 result = Path(safe_path)
357 if result.is_symlink():
358 logger.warning(f"Symlink blocked: {relative_path}")
359 return None
360 return result
361 except ValueError:
362 logger.warning(f"Path traversal blocked: {relative_path}")
363 return None
366def handle_api_error(operation: str, error: Exception, status_code: int = 500):
367 """
368 Handle API errors consistently - log internally, return generic message to user.
370 This prevents information exposure by logging full error details internally
371 while returning a generic message to the user.
373 Args:
374 operation: Description of the operation that failed (for logging)
375 error: The exception that occurred
376 status_code: HTTP status code to return (default: 500)
378 Returns:
379 Flask JSON response tuple (response, status_code)
380 """
381 # Log the full error internally with stack trace
382 logger.exception(f"Error during {operation}")
384 # Return generic message to user (no internal details exposed)
385 return jsonify(
386 {
387 "success": False,
388 "error": "An internal error occurred. Please try again or contact support.",
389 }
390 ), status_code