Coverage for src/local_deep_research/journal_quality/downloader.py: 81%
223 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-06 15:42 +0000
1"""Bulk-fetch and lazy-load journal-quality datasets.
3This module is now a thin orchestration layer over the
4`data_sources` package. Each academic dataset is a `DataSource`
5subclass; this module just iterates `ALL_SOURCES` to drive the bulk
6download flow (the dashboard "Download Data" button) and to compute the
7status payload returned by the `/metrics/api/journal-data/status`
8endpoint.
10Public API (kept stable for existing callers and tests):
11- `JOURNAL_DATA_VERSION`
12- `get_journal_data_status()`
13- `download_journal_data(force=False)`
14- `ensure_journal_data(auto_download=True)`
16Test-patch surface (kept stable for tests that mock by string path):
17- `_get_data_dir`
18- `_fetch_openalex_sources` (re-export shim)
19- `_fetch_doaj_journals` (re-export shim)
21Data source registry lives in `.data_sources` and is the single source
22of truth for source metadata, fetch logic, and download policy.
23"""
25from __future__ import annotations
27import json
28import os
29import time
30from pathlib import Path
31from typing import Optional
33from loguru import logger
35from .data_sources import ALL_SOURCES, get_source
37JOURNAL_DATA_VERSION = "v4"
39_SENTINEL = ".downloading"
42# If the sentinel is older than this, assume the previous download
43# crashed mid-way (thread died before the `finally` cleanup could run)
44# and reclaim it. The expected wall-clock is ~7 minutes, so 20 minutes
45# is a generous safety margin that still catches stuck sentinels.
46_SENTINEL_STALE_SECS = 20 * 60
48# Shared progress state for the dashboard's status endpoint.
49# A module-level dict is deliberate: there's only ever one concurrent
50# download (enforced by the O_EXCL sentinel), and the status endpoint
51# reads it on a best-effort basis. Structure:
52# state: "idle" | "running" | "error" | "success"
53# started_at: epoch seconds, or None
54# finished_at: epoch seconds, or None
55# sources: {src_key: {name, state, detail, count}}
56# db_build: {state, detail}
57# error_msg: str or None
58#
59# Per-source entries track independent parallel downloads. Writes are
60# atomic at the per-key level in CPython, so workers can update their
61# own sub-dict without a lock; the main thread composes the overall
62# `state` / `error_msg` after joining.
63_download_state: dict = {
64 "state": "idle",
65 "started_at": None,
66 "finished_at": None,
67 "sources": {},
68 "db_build": {"state": "pending", "detail": ""},
69 "error_msg": None,
70 # Per-source final counts from the most recently COMPLETED download.
71 # Set to a dict only on the success path; explicitly invalidated to
72 # None on every other return in `download_journal_data` (up-to-date,
73 # disk-space, sentinel-race, required-failure, DB-build-failure) so
74 # callers cannot read stale counts from a prior successful run.
75 # Callers rendering a user-facing summary should prefer this
76 # structured field over parsing the `(success, message)` tuple's
77 # string.
78 "counts": None,
79}
82def get_download_state() -> dict:
83 """Return a copy of the current download progress state.
85 A shallow dict copy is not enough — callers (the status endpoint)
86 serialize this to JSON and would otherwise race with live updates.
87 We copy the nested dicts too.
88 """
89 counts = _download_state["counts"]
90 return {
91 "state": _download_state["state"],
92 "started_at": _download_state["started_at"],
93 "finished_at": _download_state["finished_at"],
94 "db_build": dict(_download_state["db_build"]),
95 "error_msg": _download_state["error_msg"],
96 "sources": {k: dict(v) for k, v in _download_state["sources"].items()},
97 "counts": dict(counts) if counts is not None else None,
98 }
101def _set_source_state(key: str, **updates) -> None:
102 """Update a single source's state entry. Safe to call from worker
103 threads because the only writer of ``sources[key]`` is that key's
104 own worker (CPython guarantees dict-item writes are atomic).
105 """
106 entry = _download_state["sources"].setdefault(
107 key, {"name": key, "state": "pending", "detail": "", "count": 0}
108 )
109 entry.update(updates)
110 logger.info(
111 f"journal-data progress: {entry.get('name', key)} "
112 f"{entry.get('state', '?')} {entry.get('detail', '')}".rstrip()
113 )
116def _get_data_dir() -> Path:
117 """Get the journal data directory (user-writable).
119 Kept as a module-level function so tests can patch it via
120 `mock.patch("...journal_data_downloader._get_data_dir", ...)`.
121 """
122 from ..config.paths import get_journal_data_directory
124 return get_journal_data_directory()
127def _clear_orphan_sentinel_on_startup() -> None:
128 """Remove any ``.downloading`` sentinel left over from a previous
129 process that got killed mid-download.
131 The sentinel is created inside ``download_journal_data`` and cleaned
132 up in its ``finally`` block. If the process is SIGKILLed (or crashes
133 hard enough that the ``finally`` doesn't run) the file sits on disk
134 forever. Every subsequent call then sees "Download already in
135 progress" and bows out, even though nothing is actually downloading.
137 Called once at import time. A fresh process can't possibly own an
138 in-progress download, so any pre-existing sentinel is by definition
139 an orphan. The 20-minute ``_SENTINEL_STALE_SECS`` recovery path
140 still exists for the "process still alive but hung" case.
142 Swallows all exceptions — a misread on startup must not break the
143 module. Tests that run concurrently in the same process
144 (test_concurrent_download_blocked) set the sentinel deliberately
145 *after* import, so this runs once and doesn't interfere.
146 """
147 try:
148 sentinel = _get_data_dir() / _SENTINEL
149 if sentinel.exists(): 149 ↛ 150line 149 didn't jump to line 150 because the condition on line 149 was never true
150 sentinel.unlink()
151 logger.warning(
152 f"Cleared orphan {_SENTINEL} sentinel from a previous run "
153 f"(the old process was killed mid-download; a fresh process "
154 f"cannot own an in-progress download)."
155 )
156 except Exception:
157 logger.exception(
158 "Could not clear orphan sentinel on startup; "
159 "new downloads may be blocked until the 20-minute stale timer."
160 )
163_clear_orphan_sentinel_on_startup()
166# ---------------------------------------------------------------------------
167# Test-patch shims for `_fetch_openalex_sources` and `_fetch_doaj_journals`
168#
169# Tests in `tests/journal_quality/test_downloader.py` and
170# `tests/journal_quality/test_downloader_exception_sanitization.py` patch
171# these names by string path. The bodies have moved into `OpenAlexSource.fetch`
172# and `DOAJSource.fetch`, but we expose module-level wrappers so the
173# existing patches keep intercepting calls. The bulk download loop below
174# routes both sources through these wrappers (not directly through
175# `.fetch()`) so that `mock.patch` substitutions take effect.
176# ---------------------------------------------------------------------------
179def _fetch_openalex_sources(data_dir: Path, progress_cb=None) -> int:
180 return get_source("openalex").fetch(data_dir, progress_cb=progress_cb)
183def _fetch_doaj_journals(data_dir: Path, progress_cb=None) -> int:
184 return get_source("doaj").fetch(data_dir, progress_cb=progress_cb)
187def _fetch_predatory(data_dir: Path, progress_cb=None) -> int:
188 return get_source("predatory").fetch(data_dir, progress_cb=progress_cb)
191def _fetch_jabref_abbreviations(data_dir: Path, progress_cb=None) -> int:
192 return get_source("jabref").fetch(data_dir, progress_cb=progress_cb)
195def _fetch_institutions(data_dir: Path, progress_cb=None) -> int:
196 return get_source("institutions").fetch(data_dir, progress_cb=progress_cb)
199# Map source key → module-level shim *name*, resolved at call time so
200# `mock.patch` substitutions on the module attribute take effect (capturing
201# function objects in a dict at import time would defeat patching).
202_FETCH_SHIM_NAMES = {
203 "openalex": "_fetch_openalex_sources",
204 "doaj": "_fetch_doaj_journals",
205 "predatory": "_fetch_predatory",
206 "jabref": "_fetch_jabref_abbreviations",
207 "institutions": "_fetch_institutions",
208}
211# ---------------------------------------------------------------------------
212# Status
213# ---------------------------------------------------------------------------
216def get_journal_data_status() -> dict:
217 """Return the status payload for the dashboard data sources banner.
219 Shape (kept stable for existing JS consumers and tests):
220 {
221 "available": bool, # OpenAlex source OR compiled DB present
222 "version": Optional[str],
223 "latest_version": str,
224 "needs_update": bool,
225 "files": dict[str, bool], # legacy: filename → present
226 "sources": list[dict], # per-source detail for the banner
227 "data_dir": str,
228 }
229 """
230 data_dir = _get_data_dir()
231 version_file = data_dir / "version.json"
233 installed_version: Optional[str] = None
234 if version_file.exists():
235 try:
236 with open(version_file, encoding="utf-8") as f:
237 info = json.load(f)
238 installed_version = info.get("version")
239 except (json.JSONDecodeError, OSError):
240 pass
242 files = {src.filename: src.is_present(data_dir) for src in ALL_SOURCES}
243 sources = [src.status_dict(data_dir) for src in ALL_SOURCES]
245 # Available if the (required) OpenAlex source exists OR the compiled
246 # reference DB exists. The compiled DB is useful even when the source
247 # JSON has been deleted, since the dashboard can still query it.
248 has_source = files.get("openalex_sources.json.gz", False)
249 has_db = (data_dir / "journal_quality.db").exists() or (
250 data_dir / "journal_reference.db"
251 ).exists()
253 return {
254 "available": has_source or has_db,
255 "version": installed_version,
256 "latest_version": JOURNAL_DATA_VERSION,
257 # `needs_update` is True both when no version is installed at all
258 # (first run — show the download CTA) and when the installed
259 # version is older than the bundled latest. The previous `and`
260 # spelling silently hid the first-run case from the dashboard.
261 "needs_update": (
262 installed_version is None
263 or installed_version != JOURNAL_DATA_VERSION
264 ),
265 "files": files,
266 "sources": sources,
267 "data_dir": str(data_dir),
268 # Live progress for the dashboard's status indicator. The client
269 # polls this endpoint while a download is in flight; see
270 # downloadJournalData() in journal_quality.html.
271 "download_progress": get_download_state(),
272 }
275# ---------------------------------------------------------------------------
276# Bulk download (dashboard "Download Data" button)
277# ---------------------------------------------------------------------------
280def download_journal_data(force: bool = False) -> tuple[bool, str]:
281 """Fetch every registered data source into the user data directory.
283 Iterates `ALL_SOURCES` in order. Sources marked `required=True`
284 (OpenAlex) abort the batch on failure; `required=False` sources are
285 best-effort and continue on error.
287 Args:
288 force: Re-fetch even if data exists and is current version.
290 Returns:
291 (success, message) tuple. Message format is
292 `"Fetched <N1> <label1> + <N2> <label2> + ... in <S>s"` so the
293 existing test substring assertions ("100 OpenAlex", "50 DOAJ")
294 continue to match.
295 """
296 data_dir = _get_data_dir()
297 sentinel = data_dir / _SENTINEL
299 if not force:
300 status = get_journal_data_status()
301 if status["available"] and not status["needs_update"]:
302 # No fresh fetch ran, so invalidate any counts cached from a
303 # previous in-process download. Callers keying a "what just
304 # happened" summary off `counts` must see None here.
305 _download_state["counts"] = None
306 return True, "Journal data is already up to date"
308 # Disk-space pre-check. The five data sources uncompress to ~1 GB
309 # intermediate, plus the compiled DB. Fail fast with a clear message
310 # rather than crashing mid-download and leaving a corrupt tmp file.
311 import shutil as _shutil
313 from ..constants import JOURNAL_QUALITY_MIN_FREE_DISK_BYTES
315 try:
316 free_bytes = _shutil.disk_usage(str(data_dir)).free
317 except OSError:
318 logger.warning(
319 f"Could not check free disk space for {data_dir}; proceeding."
320 )
321 free_bytes = None
322 if (
323 free_bytes is not None
324 and free_bytes < JOURNAL_QUALITY_MIN_FREE_DISK_BYTES
325 ):
326 # No fetch ran → invalidate any stale counts from a prior call
327 # so `get_download_state()["counts"]` cannot leak them.
328 _download_state["counts"] = None
329 return False, (
330 f"Insufficient disk space: "
331 f"{free_bytes / (1024**3):.1f} GB available, "
332 f"{JOURNAL_QUALITY_MIN_FREE_DISK_BYTES / (1024**3):.0f} GB required."
333 )
335 # Atomic sentinel creation (O_CREAT | O_EXCL). Replaces the previous
336 # exists()+touch() TOCTOU race so two concurrent download triggers
337 # (dashboard click + scheduler) cannot both proceed.
338 #
339 # Stale-sentinel recovery. Two triggers are checked each call:
340 #
341 # 1. PID-based: the sentinel holds the PID of the process that
342 # created it. If that PID is not alive, the owner process
343 # crashed or was killed and the sentinel is orphan.
344 # 2. Age-based (fallback): if the sentinel is older than
345 # _SENTINEL_STALE_SECS we reclaim it even if the PID check is
346 # inconclusive (exotic environments, unreadable sentinel).
347 #
348 # The startup hook in _clear_orphan_sentinel_on_startup handles the
349 # common case of "server was restarted mid-download"; these two
350 # runtime checks cover the case where the server is still running
351 # but the download worker thread itself crashed out of the sentinel.
353 def _sentinel_owner_alive() -> bool:
354 try:
355 owner_pid = int(sentinel.read_text(encoding="utf-8").strip())
356 except (OSError, ValueError):
357 return False # unreadable / malformed → treat as orphan
358 if owner_pid == os.getpid(): 358 ↛ 363line 358 didn't jump to line 363 because the condition on line 358 was always true
359 # Same process. Something is wrong (we should never race
360 # with ourselves — the module-level lock guards that) but
361 # err on the side of "alive" to avoid self-nuking.
362 return True
363 try:
364 os.kill(owner_pid, 0) # signal 0 = liveness probe
365 except ProcessLookupError:
366 return False
367 except PermissionError:
368 # Process exists, owned by another user — treat as alive.
369 return True
370 except OSError:
371 return False
372 return True
374 def _try_claim_sentinel() -> bool:
375 """Create the sentinel + stamp our PID. Returns True on success."""
376 try:
377 with sentinel.open("x", encoding="utf-8") as f:
378 f.write(str(os.getpid()))
379 return True
380 except FileExistsError:
381 return False
383 if not _try_claim_sentinel():
384 try:
385 age = time.time() - sentinel.stat().st_mtime
386 except OSError:
387 age = 0
388 orphan = not _sentinel_owner_alive()
389 if orphan or age > _SENTINEL_STALE_SECS: 389 ↛ 390line 389 didn't jump to line 390 because the condition on line 389 was never true
390 reason = (
391 "owner process not alive"
392 if orphan
393 else f"age {age:.0f}s > {_SENTINEL_STALE_SECS}s"
394 )
395 logger.warning(
396 f"Reclaiming stale .downloading sentinel ({reason}); "
397 "previous download likely crashed without cleanup."
398 )
399 sentinel.unlink(missing_ok=True)
400 if not _try_claim_sentinel():
401 # Lost a race with another caller reclaiming the same
402 # stale sentinel — bow out cleanly.
403 _download_state["counts"] = None
404 return False, "Download already in progress"
405 else:
406 _download_state["counts"] = None
407 return False, "Download already in progress"
408 try:
409 start = time.time()
410 counts: dict[str, int] = {}
411 parts: list[str] = []
413 # Reset per-source state to a clean "pending" row for each
414 # known source. The dashboard renders one row per entry.
415 _download_state["state"] = "running"
416 _download_state["started_at"] = start
417 _download_state["finished_at"] = None
418 _download_state["error_msg"] = None
419 # Invalidate counts from any prior run — callers inspecting the
420 # structured summary must not see stale data if this download
421 # fails or is still running.
422 _download_state["counts"] = None
423 _download_state["db_build"] = {"state": "pending", "detail": ""}
424 _download_state["sources"] = {
425 src.key: {
426 "name": src.name,
427 "state": "pending",
428 "detail": "",
429 "percent": 0,
430 "count": 0,
431 "required": src.required,
432 }
433 for src in ALL_SOURCES
434 }
436 def _fetch_one(src):
437 """Worker: run one source's fetch, mirror state as it goes."""
438 _set_source_state(
439 src.key, state="running", detail="downloading", percent=5
440 )
442 # Per-partition callback: the chunked sources (openalex
443 # sources + institutions) call this on every partition so
444 # the dashboard's bar moves smoothly. One-shot sources
445 # don't call it and stay at the initial 5% → final 100%.
446 def _on_progress(done, total, detail):
447 pct = int(5 + (done / total) * 90) if total > 0 else 5
448 _set_source_state(
449 src.key,
450 state="running",
451 detail=detail,
452 percent=max(5, min(95, pct)),
453 )
455 try:
456 shim_name = _FETCH_SHIM_NAMES.get(src.key)
457 if shim_name: 457 ↛ 464line 457 didn't jump to line 464 because the condition on line 457 was always true
458 # bearer:disable python_lang_code_injection
459 # _FETCH_SHIM_NAMES is a hardcoded dict (line 188); the
460 # late-bound globals() lookup is needed for mock.patch
461 # compatibility in tests. No user input reaches the key.
462 n = globals()[shim_name](data_dir, progress_cb=_on_progress)
463 else:
464 n = src.fetch(data_dir, progress_cb=_on_progress)
465 _set_source_state(
466 src.key,
467 state="success",
468 detail=f"{n} {src.count_label}",
469 percent=100,
470 count=n,
471 )
472 return (src, n, None)
473 except Exception as exc:
474 logger.exception(
475 f"{src.name} fetch failed "
476 f"({'required' if src.required else 'optional'})"
477 )
478 _set_source_state(
479 src.key,
480 state="error",
481 detail=exc.__class__.__name__,
482 percent=100,
483 )
484 return (src, 0, exc)
486 # Parallel fetch — every source streams from a different host
487 # (openalex S3, DOAJ CSV, raw.githubusercontent.com, api.openalex.org
488 # for institutions). No single-host contention; the wall-clock is
489 # dominated by the slowest source (OpenAlex snapshot ~30-60 s).
490 from concurrent.futures import ThreadPoolExecutor
492 with ThreadPoolExecutor(
493 max_workers=max(1, len(ALL_SOURCES)),
494 thread_name_prefix="journal-dl",
495 ) as pool:
496 results = list(pool.map(_fetch_one, ALL_SOURCES))
498 required_failure = None
499 for src, n, exc in results:
500 counts[src.key] = n
501 parts.append(f"{n} {src.count_label}")
502 if exc is not None and src.required:
503 required_failure = (src, exc)
505 if required_failure is not None:
506 src, _exc = required_failure
507 msg = f"Failed to fetch {src.name}. Check your network connection."
508 _download_state["state"] = "error"
509 _download_state["error_msg"] = msg
510 _download_state["finished_at"] = time.time()
511 return False, msg
513 # Write version marker. Per-source key names are preserved for
514 # any external consumer that might read them, even though no
515 # production code does today.
516 version_file = data_dir / "version.json"
517 with open(version_file, "w", encoding="utf-8") as f:
518 json.dump(
519 {
520 "version": JOURNAL_DATA_VERSION,
521 "downloaded_at": time.strftime("%Y-%m-%dT%H:%M:%SZ"),
522 "openalex_count": counts.get("openalex", 0),
523 "doaj_count": counts.get("doaj", 0),
524 "jabref_count": counts.get("jabref", 0),
525 "predatory_count": counts.get("predatory", 0),
526 },
527 f,
528 )
530 # Rebuild journal_quality.db synchronously from the freshly
531 # downloaded gz files. The build is the ONLY writer of this
532 # file; everything else opens it read-only via mode=ro.
533 # Also clean up any leftover legacy journal_reference.db files
534 # from before the rename so existing installs don't carry junk.
535 legacy_db = data_dir / "journal_reference.db"
536 if legacy_db.exists(): 536 ↛ 537line 536 didn't jump to line 537 because the condition on line 536 was never true
537 try:
538 import os as _os
540 # bearer:disable python_lang_file_permissions
541 _os.chmod(legacy_db, 0o644)
542 legacy_db.unlink()
543 logger.info(
544 "Removed legacy journal_reference.db "
545 "(replaced by journal_quality.db)"
546 )
547 except OSError:
548 logger.exception("Could not remove legacy DB")
550 _download_state["db_build"] = {
551 "state": "running",
552 "detail": "parsing bundled data",
553 }
554 db_build_error: Optional[str] = None
555 try:
556 from .db import DB_FILENAME, build_db
558 new_db_file = data_dir / DB_FILENAME
559 if new_db_file.exists(): 559 ↛ 560line 559 didn't jump to line 560 because the condition on line 559 was never true
560 import os as _os
562 # bearer:disable python_lang_file_permissions
563 _os.chmod(new_db_file, 0o644)
564 new_db_file.unlink()
565 build_db(data_dir=data_dir, output_path=new_db_file)
566 except Exception as exc:
567 logger.exception(
568 "Failed to rebuild journal_quality.db; "
569 "the runtime accessor will lazy-build on next access"
570 )
571 # SchemaDriftError messages are developer-authored literals
572 # (no SQL, paths, or stack fragments) so they're safe to
573 # surface — operators need to see *which* field drifted to
574 # act on it. For any other exception, fall back to the
575 # class name only, per CodeQL "Information exposure through
576 # an exception" (alerts 7650, 7684). The full trace always
577 # stays in logger.exception above (server-side only).
578 from .data_sources.openalex import SchemaDriftError
580 if isinstance(exc, SchemaDriftError):
581 db_build_error = str(exc)
582 else:
583 db_build_error = exc.__class__.__name__
585 elapsed = time.time() - start
586 if db_build_error:
587 msg = (
588 f"Downloaded data ({' + '.join(parts)}) in {elapsed:.0f}s "
589 f"but DB build failed ({db_build_error}). "
590 f"Lazy-build will retry on next access."
591 )
592 _download_state["db_build"] = {
593 "state": "error",
594 "detail": db_build_error,
595 }
596 _download_state["state"] = "error"
597 _download_state["error_msg"] = msg
598 _download_state["finished_at"] = time.time()
599 return False, msg
601 success_msg = f"Fetched {' + '.join(parts)} in {elapsed:.0f}s"
602 _download_state["db_build"] = {
603 "state": "success",
604 "detail": "ready",
605 }
606 _download_state["state"] = "success"
607 _download_state["error_msg"] = None
608 _download_state["finished_at"] = time.time()
609 # Publish structured counts for callers that want to render a
610 # user-facing summary without echoing `success_msg`. All values
611 # are ints populated from source `.fetch()` returns above.
612 _download_state["counts"] = dict(counts)
613 return True, success_msg
615 finally:
616 sentinel.unlink(missing_ok=True)
619_ensure_cache: Optional[tuple[float, tuple[Optional[Path], bool]]] = None
620_ENSURE_CACHE_TTL = 30.0 # seconds
623def ensure_journal_data(
624 auto_download: bool = True,
625) -> tuple[Optional[Path], bool]:
626 """Ensure journal data is available, optionally triggering a bulk fetch.
628 Returns:
629 (data_dir, is_available) — data_dir is None if unavailable.
631 Thundering-herd guard: when a search runs, every search engine's
632 reputation-filter worker (~30 threads) calls this concurrently.
633 Without the cache, 29 of them race to create the sentinel and
634 each logs a WARNING. One call does the real work; the rest get
635 the cached answer for 30 seconds. The success path (data files
636 present) is already fast — we only cache the negative / race
637 result, which is the noisy path.
638 """
639 global _ensure_cache
641 user_dir = _get_data_dir()
642 if (user_dir / "openalex_sources.json.gz").exists():
643 # Positive path is cheap (one stat call) — no need to cache.
644 return user_dir, True
646 now = time.time()
647 if _ensure_cache is not None:
648 ts, cached = _ensure_cache
649 if now - ts < _ENSURE_CACHE_TTL:
650 return cached
652 if auto_download:
653 logger.info(
654 "Journal data not found — fetching from upstream sources..."
655 )
656 success, message = download_journal_data()
657 if success: 657 ↛ 658line 657 didn't jump to line 658 because the condition on line 657 was never true
658 logger.info(message)
659 _ensure_cache = (now, (user_dir, True))
660 return user_dir, True
661 logger.warning(f"Journal data fetch failed: {message}")
663 result: tuple[Optional[Path], bool] = (None, False)
664 _ensure_cache = (now, result)
665 return result