Coverage for src/local_deep_research/journal_quality/db.py: 95%
747 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +0000
1"""Read-only SQLAlchemy accessor for the compiled journal-quality DB.
3INVARIANT: this module's `JournalQualityDB` class **never writes**.
4The runtime engine is opened with SQLite URI flags `mode=ro` and
5`immutable=1`, the file is `chmod 0o444` after every build, and a
6pre-commit hook bans cross-module opens of the file without `mode=ro`.
8The only writer is `build_db()` in this same module, which opens its
9own short-lived writable engine, populates the schema, runs ANALYZE
10+ VACUUM, closes the engine, and chmods the file back to 0o444.
12The DB compiles five gzipped JSON snapshots (downloaded by
13`journal_quality.downloader`) into one queryable file:
15- OpenAlex sources → `sources` table (with predatory + DOAJ flags)
16- Stop Predatory Journals → `predatory_journals/_publishers/_hijacked`
17- DOAJ → cross-referenced into `sources`
18- JabRef abbreviations → `abbreviations` table
19- OpenAlex Institutions → `institutions` table
21Built fresh on every download, no migrations.
22"""
24from __future__ import annotations
26import gzip
27import json
28import math
29import os
30import secrets
31import sqlite3
32import sys
33import threading
34import time
35from contextlib import closing, contextmanager
36from pathlib import Path, PurePath
37from typing import Iterable, Iterator, Optional
38from urllib.parse import quote
40from loguru import logger
41from sqlalchemy import create_engine, func, inspect, or_, select
42from sqlalchemy.engine import Engine
43from sqlalchemy.exc import DatabaseError, OperationalError
44from sqlalchemy.orm import Session, sessionmaker
46from .models import (
47 Abbreviation,
48 Institution,
49 JournalQualityBase,
50 PredatoryHijacked,
51 PredatoryJournal,
52 PredatoryPublisher,
53 Source,
54)
55from ..constants import PREDATORY_WHITELIST_HINDEX
56from ..utilities.citation_normalizer import normalize_issn
57from ..utilities.sql_utils import escape_like
58from .scoring import (
59 derive_quality_score,
60 institution_score_from_h_index,
61 normalize_name,
62)
64DB_FILENAME = "journal_quality.db"
65_BATCH_SIZE = 5000
67# Bump when the reference DB schema (models.py) changes in a way that
68# requires a rebuild even if the upstream data version hasn't changed.
69# Stamped as SQLite `PRAGMA user_version` during build_db; checked on
70# _ensure_engine. Separate from JOURNAL_DATA_VERSION (downloader.py)
71# which tracks upstream source-data freshness.
72# v4: dropped Source.has_doaj_seal — DOAJ retired the Seal in April 2025.
73JOURNAL_QUALITY_SCHEMA_VERSION = 4
75# Quality tier → score range, used by the dashboard tier filter.
76_TIER_RANGES = {
77 "elite": (9, 10),
78 "strong": (7, 8),
79 "moderate": (5, 6),
80 "low": (3, 4),
81 "predatory": (1, 2),
82}
84# Columns safe to use in ORDER BY (prevents injection via the dashboard
85# `sort` query parameter).
86_SORT_COLUMNS = frozenset(
87 {
88 "name",
89 "quality",
90 "quartile",
91 "h_index",
92 "impact_factor",
93 "score_source",
94 "source_type",
95 "publisher",
96 "is_predatory",
97 }
98)
100# Max length for user-supplied search strings. Even with LIKE
101# wildcards escaped, a 10 KB pattern against 217K rows is slow enough
102# to matter for CPU budget under concurrent requests.
103_MAX_SEARCH_LEN = 100
106def sqlite_readonly_uri(path: PurePath, *, immutable: bool = False) -> str:
107 """Build a ``mode=ro`` SQLite URI for an absolute ``path``.
109 With ``immutable=True`` the query string is ``mode=ro&immutable=1``
110 (the runtime engine's flags). Every SQLite URI this package opens
111 goes through this one helper so that no call site interpolates a raw
112 path into a ``file:`` URI.
114 The URI always has an EMPTY authority (``file://`` + path), because
115 SQLite rejects any authority other than empty or ``localhost`` with
116 "invalid uri authority" unless it was compiled with
117 ``SQLITE_ALLOW_URI_AUTHORITY`` — which the stock builds are not.
118 ``Path.as_uri()`` is therefore unusable here: for a Windows UNC path
119 (``\\\\server\\share\\...``, and a mapped drive that ``resolve()``
120 turns into one) it yields ``file://server/share/...``, which SQLite
121 refuses to open at all. The UNC host goes into the PATH instead —
122 ``file:////server/share/...`` — which SQLite decodes to
123 ``//server/share/...``, a path Windows opens as UNC. A drive path
124 ``C:/...`` becomes ``file:///C:/...``, SQLite's documented Windows
125 form, and a POSIX path ``/a/b`` becomes ``file:///a/b``.
127 The path is percent-encoded with only ``/`` and ``:`` left literal,
128 so ``?``, ``#``, ``%``, spaces and non-ASCII characters all
129 round-trip through SQLite's URI decoder. What gets encoded is the
130 path's filesystem BYTES (``os.fsencode``), not its ``str`` form: on
131 POSIX a directory name that is not valid UTF-8 decodes to lone
132 surrogates (``b"\\xe9"`` -> ``"\\udce9"``), which ``quote(str)``
133 refuses with ``UnicodeEncodeError``; percent-encoding the bytes makes
134 SQLite open exactly the on-disk name, as a plain
135 ``sqlite3.connect(str_path)`` does. On Windows ``os.fsencode`` is
136 UTF-8, the same bytes ``quote(str)`` would produce.
138 Interpolating the raw path into ``f"file:{path}?mode=ro"`` instead
139 lets a literal ``?`` or ``#`` in a data-directory name truncate the
140 URI before ``?mode=ro`` — silently dropping the read-only flag and,
141 on a database that does not otherwise exist at the truncated path,
142 creating a stray 0-byte file there.
144 Takes a ``PurePath`` so the Windows forms are unit-testable on any
145 OS via ``PureWindowsPath``; the caller resolves the path first.
146 """
147 posix = PurePath(path).as_posix()
148 if not posix.startswith("/"):
149 # Windows drive path "C:/x" -> "/C:/x". A UNC path's as_posix()
150 # already starts with "//server/share", giving four slashes.
151 posix = "/" + posix
152 query = "mode=ro&immutable=1" if immutable else "mode=ro"
153 return "file://" + quote(os.fsencode(posix), safe=b"/:") + "?" + query
156#: ``journal_db_file_status`` results.
157JOURNAL_DB_VALID = "valid"
158JOURNAL_DB_INVALID = "invalid"
159JOURNAL_DB_UNOPENABLE = "unopenable"
162def journal_db_file_status(path: Path) -> str:
163 """Classify ``path`` as a compiled journal-quality DB, read-only.
165 Returns one of:
167 - ``JOURNAL_DB_VALID`` — the file opened and is usable as-is.
168 - ``JOURNAL_DB_INVALID`` — the file is missing, or it OPENED but is
169 not usable: a non-zero ``PRAGMA user_version`` that does not match
170 the current schema (a real drift signal), or content SQLite
171 rejects as a database (corruption). Rebuilding is the right fix.
172 - ``JOURNAL_DB_UNOPENABLE`` — the file exists but the read-only open
173 itself failed (permissions, a URI or path form this SQLite build
174 refuses, an unreachable share). Nothing is known about the file's
175 contents, so callers must NOT treat it as corrupt: deleting it on
176 this signal would destroy a possibly-valid DB over an environment
177 problem.
179 A version of 0 means the file was built before schema stamping
180 existed and is grandfathered in — we don't force a rebuild just
181 because the stamp is missing.
183 Side-effect-free on the target file, deliberately: it opens with the
184 SQLite URI flag ``mode=ro`` (see ``sqlite_readonly_uri``), which
185 never creates the target path, reads ``PRAGMA user_version``,
186 touches ``sqlite_master`` to confirm the file really is a database,
187 and closes the connection. It removes nothing and builds nothing.
188 The one caveat is SQLite's own: opening a WAL-mode database — even
189 read-only — can still create that database's ``-wal``/``-shm`` side
190 files, since materializing them is how SQLite reads a WAL database
191 at all.
193 That is what makes it safe to call from a read path that only needs
194 to know whether a read WOULD trigger a rebuild — see
195 ``web/routers/metrics.py``'s ``_journal_ref_db_read_would_download``.
196 ``JournalQualityDB._validate_existing_db`` wraps it with the
197 destructive half (unlinking an INVALID file), which a read route
198 must not do.
199 """
200 from ..utilities.resource_utils import safe_close
202 try:
203 resolved = Path(path).resolve()
204 uri = sqlite_readonly_uri(resolved)
205 conn = sqlite3.connect(uri, uri=True)
206 except (sqlite3.DatabaseError, OSError, ValueError):
207 if not Path(path).exists():
208 # Nothing to protect — a missing file is simply "not a
209 # usable DB", and a build creating it destroys nothing.
210 return JOURNAL_DB_INVALID
211 logger.exception(
212 f"journal_quality.db at {path} exists but could not be "
213 f"opened read-only"
214 )
215 return JOURNAL_DB_UNOPENABLE
217 status = JOURNAL_DB_INVALID
218 try:
219 version = conn.execute("PRAGMA user_version").fetchone()[0]
220 if version != 0 and version != JOURNAL_QUALITY_SCHEMA_VERSION:
221 logger.warning(
222 f"journal_quality.db schema_version={version}, "
223 f"expected {JOURNAL_QUALITY_SCHEMA_VERSION} — "
224 f"a rebuild is needed"
225 )
226 else:
227 # Cheap sanity check — confirms the file is a valid DB.
228 conn.execute("SELECT 1 FROM sqlite_master LIMIT 1").fetchone()
229 status = JOURNAL_DB_VALID
230 except (sqlite3.DatabaseError, OSError):
231 logger.exception(f"journal_quality.db at {path} is unusable")
232 finally:
233 safe_close(conn, "journal_quality validate")
234 return status
237def journal_db_file_is_valid(path: Path) -> bool:
238 """Return True if ``path`` is a usable compiled journal-quality DB.
240 Read-only boolean view of ``journal_db_file_status``: both INVALID
241 and UNOPENABLE report False, since neither can be served as-is.
242 Callers that would DESTROY the file on False must use the status
243 function instead and leave an UNOPENABLE file alone.
244 """
245 return journal_db_file_status(path) == JOURNAL_DB_VALID
248# ---------------------------------------------------------------------------
249# Read-only accessor
250# ---------------------------------------------------------------------------
253class JournalQualityDB:
254 """Read-only SQLAlchemy 2.0 accessor for `journal_quality.db`.
256 All filter hot-path methods return plain dicts (not mapped Source
257 objects) so call sites in `journal_reputation_filter.py` keep the
258 same call shape they had against the dict-based predecessor. The
259 dashboard methods can return either dicts or Source instances —
260 they're called once per page view so the ORM overhead is fine.
261 """
263 def __init__(self) -> None:
264 self._engine: Optional[Engine] = None
265 self._SessionLocal: Optional[sessionmaker[Session]] = None
266 # RLock (not Lock): _ensure_engine holds the lock while calling
267 # _build_or_raise → build_db → reset_db → self.reset(), which
268 # re-acquires the same lock. A non-reentrant Lock would deadlock
269 # the very first request on a fresh install.
270 self._lock = threading.RLock()
271 # Whether we've already logged a stale-data-version warning for
272 # this engine lifetime. Prevents log spam — one WARNING per
273 # server start is enough to surface the problem to admins.
274 self._stale_version_warned = False
276 # --- engine + session lifecycle ---
278 def _resolve_db_path(self) -> Path:
279 from ..config.paths import get_journal_data_directory
281 return get_journal_data_directory() / DB_FILENAME
283 def _ensure_engine(self) -> None:
284 # Acquire lock BEFORE first read to avoid DCLP publication hazard.
285 # With the GIL this is safe on CPython but explicit locking makes
286 # the happens-before relationship clear and portable.
287 with self._lock:
288 if self._engine is not None:
289 return
290 path = self._resolve_db_path()
291 if not path.exists():
292 self._build_or_raise(path)
293 else:
294 # Validate existing file before wiring up the read-only
295 # engine. Catches two failure modes at open time instead
296 # of letting them propagate to first query:
297 # 1. Schema drift — ORM changed since this file was
298 # built (PRAGMA user_version mismatch) → rebuild.
299 # 2. Corruption — file exists but isn't a valid DB
300 # (truncated build, disk error) → rebuild.
301 if not self._validate_existing_db(path):
302 self._build_or_raise(path)
304 # mode=ro + immutable=1: SQLite physically refuses writes,
305 # skips locking entirely, and reads via mmap. The OS page
306 # cache holds one shared resident copy of the hot pages.
307 #
308 # Use a creator callback because SQLAlchemy's URL parser
309 # eats the ?mode=ro&immutable=1 query string before it can
310 # reach sqlite3. The creator builds the connection directly
311 # with the SQLite URI flags intact.
312 # Percent-encoded, empty-authority URI (see
313 # ``sqlite_readonly_uri``): a raw ``f"file:{path}?..."`` lets
314 # a ``?``/``#`` in the data directory truncate the URI and
315 # drop mode=ro/immutable, and mis-decodes a literal ``%XX``.
316 engine_uri = sqlite_readonly_uri(path.resolve(), immutable=True)
318 def _make_ro_conn() -> sqlite3.Connection:
319 return sqlite3.connect(engine_uri, uri=True)
321 # NullPool: each Session checks out an isolated read-only
322 # connection that is closed when the session context exits.
323 # With mode=ro&immutable=1, SQLite connection opens are
324 # trivial (no lock overhead) and the OS page cache
325 # shares data across connections. NullPool ensures that
326 # concurrent filter_results threads never share a single
327 # sqlite3.Connection handle (eliminating the cursor race condition
328 # that caused sqlite3.InterfaceError / IndexError under StaticPool).
329 from sqlalchemy.pool import NullPool
331 engine = create_engine(
332 "sqlite://",
333 creator=_make_ro_conn,
334 poolclass=NullPool,
335 echo=False,
336 )
337 session_local = sessionmaker(bind=engine, expire_on_commit=False)
338 # Publish both together so readers never see engine-without-session.
339 self._engine = engine
340 self._SessionLocal = session_local
341 logger.info(f"Opened journal_quality.db (read-only): {path}")
342 # One-shot check: is the DATA version (the one the sources
343 # JSON + build logic produce) behind the bundled latest?
344 # Schema drift is already handled by `_validate_existing_db`
345 # via ``PRAGMA user_version``. A data-version mismatch is a
346 # different concern: the DB schema is fine, but the scoring
347 # logic (e.g. the repository cap) or source snapshots have
348 # been updated since this file was built. The hot path
349 # (filter scoring) would silently serve stale scores if we
350 # didn't surface the mismatch anywhere but the admin
351 # dashboard. Log once, don't auto-rebuild — user consent
352 # via the dashboard "Download Data" button remains the
353 # explicit refresh trigger.
354 self._warn_on_stale_data_version(path.parent)
356 def _warn_on_stale_data_version(self, data_dir: Path) -> None:
357 """Log WARNING once if ``version.json`` is behind ``JOURNAL_DATA_VERSION``."""
358 if self._stale_version_warned:
359 return
360 # Lazy import to avoid any downloader → db cycle even though
361 # today's module graph doesn't have one.
362 from .downloader import JOURNAL_DATA_VERSION
364 version_file = data_dir / "version.json"
365 if not version_file.exists():
366 return # Brand-new install — the dashboard's banner handles this.
367 try:
368 with open(version_file, encoding="utf-8") as f:
369 info = json.load(f)
370 installed = info.get("version")
371 except (json.JSONDecodeError, OSError):
372 return # Malformed — dashboard's banner surfaces; don't double-log.
373 if installed and installed != JOURNAL_DATA_VERSION:
374 logger.warning(
375 f"journal_quality data version is stale: on-disk={installed!r} "
376 f"bundled-latest={JOURNAL_DATA_VERSION!r}. "
377 f"Scoring is continuing with the older data. Visit "
378 f"/metrics/journals and click 'Download Data' to refresh."
379 )
380 self._stale_version_warned = True
382 def _validate_existing_db(self, path: Path) -> bool:
383 """Return True if the existing DB file is usable as-is.
385 Thin wrapper over the module-level ``journal_db_file_status``
386 probe plus the destructive half: a file that opened but is
387 unusable is removed here so the caller's rebuild starts from a
388 clean slate. A file that exists but could not be opened at all
389 is left alone and raises ``FileNotFoundError`` (reference DB
390 unavailable) instead. Callers
391 that only want to KNOW whether the file is usable — a read route
392 deciding whether a read would trigger a build — must use the
393 probe directly rather than this method, which would delete their
394 DB as a side effect of asking.
395 """
396 status = journal_db_file_status(path)
397 if status == JOURNAL_DB_UNOPENABLE:
398 # The file exists but could not even be opened read-only, so
399 # nothing says its contents are bad. Unlinking it (and then
400 # rebuilding or re-downloading over it) would destroy a
401 # possibly-valid DB over a permissions / path-form problem;
402 # under a forbidding egress scope the rebuild would then
403 # refuse and leave the user with no DB at all. Report the
404 # reference DB as unavailable instead — FileNotFoundError is
405 # the signal every reader here already degrades on — and
406 # leave the file in place.
407 raise FileNotFoundError(
408 f"{DB_FILENAME} exists at {path} but could not be opened "
409 f"read-only; leaving it in place rather than rebuilding"
410 )
411 is_valid = status == JOURNAL_DB_VALID
412 if not is_valid:
413 self._unlink_unusable_db(path)
414 return is_valid
416 @staticmethod
417 def _unlink_unusable_db(path: Path) -> None:
418 """Best-effort cleanup of a corrupted / schema-drifted DB file.
420 Corruption was already logged by the caller (``_validate_existing_db``).
421 Both operations below are best-effort — on failure we *log and
422 continue* rather than raise, because the build path will rebuild
423 the file regardless. But we don't silence: if chmod / unlink
424 fails (permissions, read-only mount, file held open on Windows)
425 the next build will likely also fail and the user needs the
426 warning to diagnose the real problem.
427 """
428 try:
429 # bearer:disable python_lang_file_permissions
430 os.chmod(path, 0o644)
431 except OSError:
432 logger.warning(
433 f"Could not chmod 0644 on unusable DB {path} before "
434 f"unlink (continuing to unlink attempt)"
435 )
436 try:
437 path.unlink()
438 except OSError:
439 logger.warning(
440 f"Could not unlink unusable DB {path} (will be "
441 f"overwritten on next build)"
442 )
444 def _build_or_raise(self, path: Path) -> None:
445 """Lazy-build the DB on first access if it's missing.
447 Under the test suite's mock-only mode (``LDR_TESTING_WITH_MOCKS``,
448 defaulted to true by ``tests/conftest.py``) this refuses to
449 *download*. The method is reached implicitly — any read through
450 ``_ensure_engine`` when the file is missing or fails validation — so
451 a test that never intended to exercise the download path could
452 otherwise trigger a live, multi-hundred-MB fetch from OpenAlex/DOAJ
453 inside a request. ``timeout_method="thread"`` cannot interrupt a
454 blocked socket, so the xdist worker dies instead of the test failing.
456 It gates ``auto_download`` rather than refusing outright, and the
457 difference matters: ``ensure_journal_data`` returns immediately when
458 the gz snapshots are already on disk, which is a purely local,
459 network-free path. Refusing before that call would also block a
460 developer or CI job that has the snapshots, turning a working offline
461 build into a failure. With ``auto_download=False`` the local build
462 still happens and only the network is off the table.
464 Tests that DO want the real download orchestration call
465 ``download_journal_data()`` directly with the per-source fetchers
466 mocked (see ``tests/journal_quality/test_downloader.py``), which this
467 does not touch.
468 """
469 from ..settings.env_definitions.testing import testing_with_mocks
470 from .downloader import ensure_journal_data
472 mock_mode = testing_with_mocks()
473 # Third gate, alongside mock mode: the active egress scope. The
474 # dataset sources (OpenAlex/DOAJ/GitHub) are all public, so under
475 # PRIVATE_ONLY/STRICT the lazy build must refuse the network
476 # download, like the dashboard button and the filter's background
477 # fetch already do. The parity is close but not exact, and the
478 # difference matters: both siblings resolve the user's PERSISTED
479 # settings (the button the literal `policy.egress_scope`, the
480 # filter that scope with ADAPTIVE resolved against the primary
481 # engine) and fail closed when the scope cannot be parsed,
482 # while this gate can only see a context armed on the calling
483 # thread. Where no context is armed it allows the download —
484 # request threads (Starlette runs sync routes on a threadpool
485 # that arms none) and a run whose settings snapshot failed to
486 # resolve (`search_system.py` deliberately leaves the backstop
487 # unarmed there) both land in that case. Entry points that can
488 # reach the persisted settings gate on them themselves: the
489 # dashboard read routes in `web/routers/metrics.py` resolve the
490 # scope a run would get (ADAPTIVE included), since a page load
491 # would otherwise reach this build ungated.
492 # A purely local build from already-present snapshots stays
493 # allowed — only the network is off the table.
494 from ..security.egress.audit_hook import get_active_context
495 from ..security.egress.policy import EgressScope
497 ctx = get_active_context()
498 scope_forbids_public = ctx is not None and ctx.scope in (
499 EgressScope.PRIVATE_ONLY,
500 EgressScope.STRICT,
501 )
502 data_dir, available = ensure_journal_data(
503 auto_download=not mock_mode and not scope_forbids_public
504 )
505 if not available:
506 if mock_mode:
507 raise FileNotFoundError(
508 "Journal data files not available and "
509 "LDR_TESTING_WITH_MOCKS=true — refusing to download "
510 "from OpenAlex/DOAJ during a test run. Seed "
511 "journal_quality.db directly, or place the gz "
512 "snapshots in the data directory, if this test needs "
513 "real data."
514 )
515 if scope_forbids_public:
516 # Recorded here rather than next to the gate above: with
517 # the snapshots already on disk `ensure_journal_data`
518 # reports them available and the local build proceeds, so
519 # nothing was refused and no policy record is due.
520 logger.bind(policy_audit=True).info(
521 "journal-data lazy build refused the network download: "
522 "active egress scope forbids public fetches"
523 )
524 raise FileNotFoundError(
525 "Journal data files not available and the active "
526 "egress scope forbids the public download. Place "
527 "the gz snapshots in the data directory, or change "
528 "the egress scope and then download from the "
529 "dashboard."
530 )
531 raise FileNotFoundError(
532 "Journal data files not available. "
533 "Check your network connection or download manually "
534 "from the dashboard."
535 )
536 logger.info(f"Building {DB_FILENAME} from data files...")
537 build_db(data_dir=data_dir, output_path=path)
539 @contextmanager
540 def session(self) -> Iterator[Session]:
541 """Yield a read-only SQLAlchemy session.
543 If the underlying file becomes corrupt mid-session (e.g. a
544 rebuild ran and this engine is pointed at a now-unlinked inode),
545 DatabaseError propagates but we drop the cached engine so the
546 next call rebuilds cleanly instead of failing forever.
547 """
548 self._ensure_engine()
549 if self._SessionLocal is None: 549 ↛ 550line 549 didn't jump to line 550 because the condition on line 549 was never true
550 raise RuntimeError("JournalQualityDB engine failed to initialize")
551 from ..utilities.resource_utils import safe_close
553 s = self._SessionLocal()
554 try:
555 yield s
556 except (OperationalError, DatabaseError):
557 logger.exception("journal_quality.db error — resetting engine")
558 self.reset()
559 raise
560 finally:
561 safe_close(s, "journal_quality session")
563 @property
564 def available(self) -> bool:
565 try:
566 self._ensure_engine()
567 return True
568 except FileNotFoundError:
569 return False
571 def reset(self) -> None:
572 """Drop the cached engine — call after `build_db` rebuilds the file."""
573 with self._lock:
574 if self._engine is not None:
575 self._engine.dispose()
576 self._engine = None
577 self._SessionLocal = None
578 logger.info("Reset journal_quality.db engine")
580 # --- filter hot path: return plain dicts ---
582 def lookup_openalex(
583 self,
584 *,
585 source_id: Optional[str] = None,
586 issn: Optional[str] = None,
587 name: Optional[str] = None,
588 ) -> Optional[dict]:
589 """Look up a source by OpenAlex ID, ISSN, or name.
591 Returns a dict with the same shape the dict-based predecessor
592 produced (`name`, `type`, `h_index`, `impact_factor`,
593 `is_in_doaj`, `publisher`, `issn_l`) so the filter code at
594 `journal_reputation_filter.py` doesn't need to change.
595 """
596 issn = normalize_issn(issn)
597 try:
598 self._ensure_engine()
599 except FileNotFoundError:
600 return None
601 with self.session() as s:
602 row = self._lookup_source_row(s, source_id, issn, name)
603 return _source_to_lookup_dict(row) if row else None
605 # Alias used by the dashboard / future call sites
606 lookup_source = lookup_openalex
608 def count_predatory_by_names(self, names: Iterable[str]) -> int:
609 """Count how many of the given journal names are flagged predatory.
611 One SQL round-trip using ``WHERE name_lower IN (…) AND is_predatory = TRUE``.
612 Names are normalized (NFKC + lower + strip) so the caller can pass raw
613 display names; matches the normalization used at build time.
615 Used by the per-user metrics dashboard to report a global "N predatory
616 journals across all your research" stat without making N round trips
617 to the reference DB. Returns 0 if the reference DB is missing or if
618 ``names`` is empty.
620 .. note::
621 Deliberately unchunked. SQLite's ``SQLITE_MAX_VARIABLE_NUMBER``
622 has been 250,000 since SQLite 3.32 (2020); Python 3.11 ships
623 with 3.45.1. A heavy user with 100k distinct container_titles
624 is still well under the limit. Re-confirmed in the PR #3081
625 audit — no chunking needed.
626 """
627 normed = {normalize_name(n) for n in names if n}
628 normed.discard("")
629 if not normed: 629 ↛ 630line 629 didn't jump to line 630 because the condition on line 629 was never true
630 return 0
631 try:
632 self._ensure_engine()
633 except FileNotFoundError:
634 return 0
635 with self.session() as s:
636 stmt = select(func.count(Source.id)).where(
637 Source.name_lower.in_(normed),
638 Source.is_predatory.is_(True),
639 )
640 result = s.execute(stmt).scalar()
641 return int(result or 0)
643 def lookup_sources_batch(self, names: Iterable[str]) -> dict:
644 """Batch-look-up multiple journal names in one query.
646 Takes an iterable of raw display names and returns a
647 ``{normalized_name: dashboard_dict}`` map for every name that
648 matched a Source. Names that didn't match are simply absent
649 from the result (caller decides how to handle misses).
651 Dashboard hot path: the ``/api/journals/user-research`` endpoint
652 collects up to 200 unique ``container_title`` values from the
653 user's Papers and hands them in here — one SQL round-trip vs.
654 200 per-row lookups.
656 Normalization matches ``normalize_name`` (NFKC + lower + strip)
657 so the reference DB's ``name_lower`` column hits directly. No
658 "the " / "proceedings of" fallback tiers — those live in
659 ``_lookup_source_row`` for precision per-call; the batch path
660 is for dashboard display where a miss is acceptable.
662 Chunked at 900 params per chunk. Defensive: SQLite's actual
663 limit (``SQLITE_MAX_VARIABLE_NUMBER``) has been 250,000 since
664 3.32 (2020) — we could easily put the whole batch in one IN —
665 but 900 keeps us well under any older embedded-SQLite ceiling
666 a deployment might pin to.
667 """
668 normed = [normalize_name(n) for n in names if n]
669 normed = [n for n in normed if n]
670 if not normed: 670 ↛ 671line 670 didn't jump to line 671 because the condition on line 670 was never true
671 return {}
672 try:
673 self._ensure_engine()
674 except FileNotFoundError:
675 return {}
676 # De-duplicate while preserving insertion order for stable
677 # iteration in tests.
678 seen: set = set()
679 uniq: list = []
680 for n in normed:
681 if n not in seen:
682 seen.add(n)
683 uniq.append(n)
685 out: dict = {}
686 CHUNK = 900
687 with self.session() as s:
688 for i in range(0, len(uniq), CHUNK):
689 batch = uniq[i : i + CHUNK]
690 stmt = select(Source).where(Source.name_lower.in_(batch))
691 for row in s.scalars(stmt):
692 out[row.name_lower] = _source_to_dashboard_dict(row)
693 return out
695 def lookup_doaj(self, *, issn: Optional[str] = None) -> Optional[dict]:
696 issn = normalize_issn(issn)
697 if not issn:
698 return None
699 try:
700 self._ensure_engine()
701 except FileNotFoundError:
702 return None
703 with self.session() as s:
704 stmt = (
705 select(Source)
706 .where(Source.issn == issn, Source.is_in_doaj.is_(True))
707 .limit(1)
708 )
709 row = s.scalars(stmt).first()
710 if row is None:
711 return None
712 return {
713 "name": row.name,
714 "publisher": row.publisher,
715 }
717 def is_in_doaj(self, issn: Optional[str]) -> bool:
718 return self.lookup_doaj(issn=issn) is not None
720 def is_predatory(
721 self,
722 *,
723 journal_name: Optional[str] = None,
724 publisher_name: Optional[str] = None,
725 ) -> tuple[bool, Optional[str]]:
726 """Check if a journal/publisher is on the predatory list.
728 Looks up the dedicated predatory tables (NOT just the
729 `is_predatory` flag on `Source`), so checks work for arbitrary
730 input names that aren't in OpenAlex.
731 """
732 try:
733 self._ensure_engine()
734 except FileNotFoundError:
735 return False, None
736 with self.session() as s:
737 if journal_name:
738 norm = normalize_name(journal_name)
739 if s.get(PredatoryJournal, norm) is not None:
740 return True, "stop-predatory-journals"
741 if s.get(PredatoryHijacked, norm) is not None:
742 return True, "stop-predatory-hijacked"
744 if publisher_name:
745 pub_norm = normalize_name(publisher_name)
746 if s.get(PredatoryPublisher, pub_norm) is not None:
747 return True, "stop-predatory-publishers"
748 # Substring scan over long entries (~1162 rows)
749 stmt = select(PredatoryPublisher.name_lower).where(
750 PredatoryPublisher.is_long.is_(True)
751 )
752 for (entry,) in s.execute(stmt).all():
753 if pub_norm in entry or entry in pub_norm:
754 return True, "stop-predatory-publishers"
756 return False, None
758 def is_whitelisted(
759 self,
760 *,
761 issn: Optional[str] = None,
762 name: Optional[str] = None,
763 ) -> bool:
764 if self.is_in_doaj(issn):
765 return True
766 oa = self.lookup_openalex(issn=issn, name=name)
767 if oa and (oa.get("h_index") or 0) > PREDATORY_WHITELIST_HINDEX:
768 return True
769 return False
771 def lookup_institution(
772 self,
773 *,
774 ror_id: Optional[str] = None,
775 openalex_id: Optional[str] = None,
776 name: Optional[str] = None,
777 ) -> Optional[dict]:
778 """Look up an institution.
780 Order: openalex_id → ror → name. Returns a dict with full-name
781 keys (``name``, ``country``, ``type``, ``h_index``,
782 ``impact_factor``, ``works_count``, ``cited_by_count``, ``ror_id``)
783 or ``None`` if no match. The on-disk snapshot uses one-character
784 keys (``n``, ``c``, etc.) for space efficiency; the accessor
785 returns full names instead for legibility and schema robustness.
786 """
787 try:
788 self._ensure_engine()
789 except FileNotFoundError:
790 return None
791 with self.session() as s:
792 row: Optional[Institution] = None
794 if openalex_id:
795 sid = openalex_id.split("/")[-1]
796 row = s.get(Institution, sid)
798 if row is None and ror_id:
799 ror = ror_id.rstrip("/").split("/")[-1]
800 stmt = (
801 select(Institution)
802 .where(Institution.ror_id == ror)
803 .limit(1)
804 )
805 row = s.scalars(stmt).first()
807 if row is None and name:
808 norm = normalize_name(name)
809 stmt = (
810 select(Institution)
811 .where(Institution.name_lower == norm)
812 .limit(1)
813 )
814 row = s.scalars(stmt).first()
816 return _institution_to_dict(row) if row else None
818 def score_from_affiliations(self, affiliations: list) -> Optional[int]:
819 """Derive a score from author affiliations in ONE SQL query."""
820 if not affiliations: 820 ↛ 821line 820 didn't jump to line 821 because the condition on line 820 was never true
821 return None
823 openalex_ids: list[str] = []
824 ror_ids: list[str] = []
825 names: list[str] = []
827 for aff in affiliations:
828 if isinstance(aff, str):
829 names.append(normalize_name(aff))
830 elif isinstance(aff, dict): 830 ↛ 827line 830 didn't jump to line 827 because the condition on line 830 was always true
831 if oid := (aff.get("openalex_id") or aff.get("id")):
832 openalex_ids.append(oid.split("/")[-1])
833 if rid := aff.get("ror"):
834 ror_ids.append(rid.rstrip("/").split("/")[-1])
835 if nm := aff.get("name"):
836 names.append(normalize_name(nm))
838 if not (openalex_ids or ror_ids or names): 838 ↛ 839line 838 didn't jump to line 839 because the condition on line 838 was never true
839 return None
841 try:
842 self._ensure_engine()
843 except FileNotFoundError:
844 return None
846 clauses = []
847 if openalex_ids:
848 clauses.append(Institution.openalex_id.in_(openalex_ids))
849 if ror_ids:
850 clauses.append(Institution.ror_id.in_(ror_ids))
851 if names:
852 clauses.append(Institution.name_lower.in_(names))
854 with self.session() as s:
855 stmt = select(func.max(Institution.h_index)).where(
856 or_(*clauses), Institution.h_index.is_not(None)
857 )
858 best_h = s.scalar(stmt)
860 # Single source of truth for institution scoring lives in
861 # scoring.py — delegate so the build phase, the runtime filter,
862 # and this affiliation-salvage path can never disagree.
863 return institution_score_from_h_index(best_h)
865 # Static passthrough so the filter can call dm.derive_quality_score(...)
866 # without importing from .scoring directly. Single home for the
867 # scoring rules in scoring.py.
868 derive_quality_score = staticmethod(derive_quality_score)
870 def expand_abbreviation(self, name: str) -> Optional[str]:
871 if not name:
872 return None
873 try:
874 self._ensure_engine()
875 except FileNotFoundError:
876 return None
877 normalized = normalize_name(name)
878 with self.session() as s:
879 row = s.get(Abbreviation, normalized)
880 if row is not None: 880 ↛ 881line 880 didn't jump to line 881 because the condition on line 880 was never true
881 return row.full_name
882 no_dots = normalized.replace(".", "").strip()
883 if no_dots != normalized:
884 row = s.get(Abbreviation, no_dots)
885 if row is not None: 885 ↛ 887line 885 didn't jump to line 887
886 return row.full_name
887 return None
889 # --- internal source lookup with name fallbacks ---
891 def _lookup_source_row(
892 self,
893 s: Session,
894 source_id: Optional[str],
895 issn: Optional[str],
896 name: Optional[str],
897 ) -> Optional[Source]:
898 if source_id:
899 sid = source_id.split("/")[-1] if "/" in source_id else source_id
900 stmt = (
901 select(Source).where(Source.openalex_source_id == sid).limit(1)
902 )
903 row = s.scalars(stmt).first()
904 if row is not None: 904 ↛ 907line 904 didn't jump to line 907 because the condition on line 904 was always true
905 return row
907 if issn:
908 stmt = select(Source).where(Source.issn == issn).limit(1)
909 row = s.scalars(stmt).first()
910 if row is not None: 910 ↛ 913line 910 didn't jump to line 913 because the condition on line 910 was always true
911 return row
913 if name: 913 ↛ 963line 913 didn't jump to line 963 because the condition on line 913 was always true
914 norm = normalize_name(name)
915 row = self._fetch_by_name_lower(s, norm)
916 if row is not None:
917 return row
918 # Try with/without "the " prefix (~5K journals have it)
919 if norm.startswith("the "):
920 row = self._fetch_by_name_lower(s, norm[4:])
921 else:
922 row = self._fetch_by_name_lower(s, "the " + norm)
923 if row is not None:
924 return row
925 # Strip "proceedings of (the) (conference on) " prefix
926 stripped = norm
927 for prefix in (
928 "proceedings of the conference on ",
929 "proceedings of the ",
930 "proceedings of ",
931 ):
932 if stripped.startswith(prefix):
933 stripped = stripped[len(prefix) :]
934 break
935 if stripped != norm:
936 row = self._fetch_by_name_lower(s, stripped)
937 if row is not None: 937 ↛ 944line 937 didn't jump to line 944 because the condition on line 937 was always true
938 return row
940 # MEDLINE-style "Title : long subtitle" — try the segment
941 # before the colon. Catches PubMed names like
942 # "Molecular therapy : the journal of the American Society..."
943 # → "Molecular therapy"
944 if " : " in norm:
945 head = norm.split(" : ", 1)[0].strip()
946 if head and head != norm: 946 ↛ 956line 946 didn't jump to line 956 because the condition on line 946 was always true
947 row = self._fetch_by_name_lower(s, head)
948 if row is not None: 948 ↛ 956line 948 didn't jump to line 956 because the condition on line 948 was always true
949 return row
951 # MEDLINE-style "Title. Section name" — try the segment
952 # before the first period. Catches PubMed names like
953 # "Molecular therapy. Methods and clinical development"
954 # but only when the head is meaningfully shorter (we don't
955 # want to match "Nat" from "Nat. Commun.").
956 if "." in norm:
957 head = norm.split(".", 1)[0].strip()
958 if head and len(head) >= 6 and head != norm: 958 ↛ 963line 958 didn't jump to line 963 because the condition on line 958 was always true
959 row = self._fetch_by_name_lower(s, head)
960 if row is not None: 960 ↛ 963line 960 didn't jump to line 963 because the condition on line 960 was always true
961 return row
963 return None
965 @staticmethod
966 def _fetch_by_name_lower(s: Session, name_lower: str) -> Optional[Source]:
967 stmt = select(Source).where(Source.name_lower == name_lower).limit(1)
968 return s.scalars(stmt).first()
970 # --- dashboard queries ---
972 def get_summary(self) -> dict:
973 if not self.available:
974 return {
975 "total": 0,
976 "avg_quality": 0,
977 "avg_h_index": None,
978 "predatory_count": 0,
979 "doaj_count": 0,
980 "llm_count": 0,
981 }
983 with self.session() as s:
984 row = s.execute(
985 select(
986 func.count().label("total"),
987 func.round(func.avg(Source.quality), 1).label(
988 "avg_quality"
989 ),
990 func.round(func.avg(Source.h_index)).label("avg_h_index"),
991 func.sum(func.iif(Source.is_predatory, 1, 0)).label(
992 "predatory_count"
993 ),
994 func.sum(func.iif(Source.is_in_doaj, 1, 0)).label(
995 "doaj_count"
996 ),
997 func.sum(
998 func.iif(Source.score_source == "llm", 1, 0)
999 ).label("llm_count"),
1000 )
1001 ).first()
1002 return dict(row._mapping) if row else {}
1004 def get_quality_distribution(self) -> dict[str, int]:
1005 if not self.available:
1006 return {}
1007 with self.session() as s:
1008 rows = s.execute(
1009 select(Source.quality, func.count().label("cnt"))
1010 .where(Source.quality.is_not(None))
1011 .group_by(Source.quality)
1012 .order_by(Source.quality)
1013 ).all()
1014 return {str(q): c for q, c in rows}
1016 def get_source_distribution(self) -> dict[str, int]:
1017 if not self.available:
1018 return {}
1019 with self.session() as s:
1020 rows = s.execute(
1021 select(
1022 func.coalesce(Source.score_source, "unknown").label("src"),
1023 func.count().label("cnt"),
1024 ).group_by(Source.score_source)
1025 ).all()
1026 return {row.src: row.cnt for row in rows}
1028 def get_journals_page(
1029 self,
1030 *,
1031 page: int = 1,
1032 per_page: int = 50,
1033 search: str = "",
1034 tier: str = "",
1035 score_source: str = "",
1036 sort: str = "quality",
1037 order: str = "desc",
1038 ) -> tuple[list[dict], int]:
1039 if not self.available:
1040 return [], 0
1042 if sort not in _SORT_COLUMNS:
1043 sort = "quality"
1044 if order not in ("asc", "desc"):
1045 order = "desc"
1047 wheres: list = []
1048 if search:
1049 needle = escape_like(normalize_name(search)[:_MAX_SEARCH_LEN])
1050 wheres.append(Source.name_lower.like(f"%{needle}%", escape="\\"))
1051 if tier and tier in _TIER_RANGES:
1052 lo, hi = _TIER_RANGES[tier]
1053 wheres.append(Source.quality.between(lo, hi))
1054 if score_source:
1055 wheres.append(Source.score_source == score_source)
1057 sort_col = getattr(Source, sort)
1058 order_clause = (
1059 sort_col.desc().nulls_last()
1060 if order == "desc"
1061 else sort_col.asc().nulls_last()
1062 )
1064 offset = (max(1, page) - 1) * per_page
1066 with self.session() as s:
1067 total = (
1068 s.scalar(
1069 select(func.count()).select_from(Source).where(*wheres)
1070 )
1071 or 0
1072 )
1073 rows = s.scalars(
1074 select(Source)
1075 .where(*wheres)
1076 .order_by(order_clause)
1077 .limit(per_page)
1078 .offset(offset)
1079 ).all()
1081 return [_source_to_dashboard_dict(r) for r in rows], total
1083 def get_institutions_page(
1084 self,
1085 *,
1086 page: int = 1,
1087 per_page: int = 50,
1088 search: str = "",
1089 sort: str = "h_index",
1090 order: str = "desc",
1091 ) -> tuple[list[dict], int]:
1092 if not self.available:
1093 return [], 0
1095 # Defensive allowlist — matches the pattern in get_journals_page.
1096 # The ternary below is already safe (non-"desc" falls through to
1097 # .asc()), but the explicit check prevents future refactors from
1098 # accidentally interpolating a tainted value into SQL.
1099 if order not in ("asc", "desc"):
1100 order = "desc"
1102 wheres = []
1103 if search: 1103 ↛ 1109line 1103 didn't jump to line 1109 because the condition on line 1103 was always true
1104 needle = escape_like(normalize_name(search)[:_MAX_SEARCH_LEN])
1105 wheres.append(
1106 Institution.name_lower.like(f"%{needle}%", escape="\\")
1107 )
1109 sort_col = (
1110 Institution.h_index if sort == "h_index" else Institution.name
1111 )
1112 order_clause = (
1113 sort_col.desc().nulls_last()
1114 if order == "desc"
1115 else sort_col.asc().nulls_last()
1116 )
1118 offset = (max(1, page) - 1) * per_page
1120 with self.session() as s:
1121 total = (
1122 s.scalar(
1123 select(func.count()).select_from(Institution).where(*wheres)
1124 )
1125 or 0
1126 )
1127 rows = s.scalars(
1128 select(Institution)
1129 .where(*wheres)
1130 .order_by(order_clause)
1131 .limit(per_page)
1132 .offset(offset)
1133 ).all()
1135 return [_institution_to_dashboard_dict(r) for r in rows], total
1138# ---------------------------------------------------------------------------
1139# Dict adapters — keep filter/dashboard call sites unchanged
1140# ---------------------------------------------------------------------------
1143def _source_to_lookup_dict(row: Source) -> dict:
1144 """Convert a Source row to the dict shape `lookup_openalex` produces.
1146 Includes `openalex_source_id` so dashboard / test code can chain
1147 a follow-up `lookup_source(source_id=...)` call. Also exposes
1148 ``quartile`` so the filter can store it on the per-user Journal row
1149 and feed it into score derivation.
1150 """
1151 return {
1152 "name": row.name,
1153 "type": row.source_type,
1154 "h_index": row.h_index,
1155 "impact_factor": row.impact_factor,
1156 "is_in_doaj": row.is_in_doaj,
1157 "publisher": row.publisher,
1158 "issn_l": row.issn,
1159 "openalex_source_id": row.openalex_source_id,
1160 "quartile": row.quartile,
1161 }
1164def _source_to_dashboard_dict(row: Source) -> dict:
1165 return {
1166 "name": row.name,
1167 "quality": row.quality,
1168 "quartile": row.quartile,
1169 "cited_by_count": row.cited_by_count,
1170 "h_index": row.h_index,
1171 "impact_factor": (
1172 round(row.impact_factor, 2) if row.impact_factor else None
1173 ),
1174 "is_in_doaj": bool(row.is_in_doaj),
1175 "is_predatory": bool(row.is_predatory),
1176 "predatory_source": row.predatory_source,
1177 "score_source": row.score_source,
1178 "source_type": row.source_type,
1179 "publisher": row.publisher,
1180 "issn": row.issn,
1181 "openalex_source_id": row.openalex_source_id,
1182 }
1185def _institution_to_dict(row: Institution) -> dict:
1186 """Public accessor shape for `lookup_institution`.
1188 The on-disk JSON snapshot uses one-character keys (``n``, ``c``,
1189 ``t``, …) purely for space efficiency — 200k institutions × seven
1190 long field names adds real bytes. Callers of the accessor don't
1191 care about on-disk layout, so here we return the full names to
1192 keep the public API legible and robust to future schema tweaks.
1193 """
1194 return {
1195 "name": row.name,
1196 "country": row.country,
1197 "type": row.type,
1198 "h_index": row.h_index,
1199 "impact_factor": row.impact_factor,
1200 "works_count": row.works_count,
1201 "cited_by_count": row.cited_by_count,
1202 "ror_id": row.ror_id,
1203 }
1206def _institution_to_dashboard_dict(row: Institution) -> dict:
1207 return {
1208 "openalex_id": row.openalex_id,
1209 "name": row.name,
1210 "ror_id": row.ror_id,
1211 "country": row.country,
1212 "type": row.type,
1213 "h_index": row.h_index,
1214 "impact_factor": row.impact_factor,
1215 "works_count": row.works_count,
1216 "cited_by_count": row.cited_by_count,
1217 }
1220# ---------------------------------------------------------------------------
1221# Module singleton
1222# ---------------------------------------------------------------------------
1225_db: Optional[JournalQualityDB] = None
1226_db_lock = threading.Lock()
1229def get_db() -> JournalQualityDB:
1230 """Get or create the singleton `JournalQualityDB`."""
1231 global _db
1232 if _db is None:
1233 with _db_lock:
1234 if _db is None: 1234 ↛ 1236line 1234 didn't jump to line 1236
1235 _db = JournalQualityDB()
1236 return _db
1239# Backwards-compat aliases used by web/routers/metrics.py and a couple of tests
1240get_journal_reference_db = get_db
1241JournalReferenceDB = JournalQualityDB
1244def reset_db() -> None:
1245 """Reset the cached engine after a build_db rebuild.
1247 Held under `_db_lock` so a concurrent `get_db()` call can't see a
1248 half-disposed singleton — without the lock, Thread B could pass
1249 the `if _db is None` check in `get_db()` while Thread A is still
1250 inside `_db.reset()`, then call `_ensure_engine()` which short-
1251 circuits on the still-set `_engine` and hands back a disposed
1252 pool. The lock makes the read-then-reset pair atomic with respect
1253 to `get_db()`'s lazy-init path.
1254 """
1255 global _db
1256 with _db_lock:
1257 if _db is not None:
1258 _db.reset()
1261# ---------------------------------------------------------------------------
1262# Build phase — the ONLY writer
1263# ---------------------------------------------------------------------------
1266def build_db(
1267 data_dir: Optional[Path] = None,
1268 output_path: Optional[Path] = None,
1269) -> None:
1270 """Compile `journal_quality.db` from the gzipped JSON sources.
1272 Opens a SHORT-LIVED writable engine, creates the schema, populates
1273 every table from the gz files, runs ANALYZE + VACUUM, closes the
1274 engine, then `chmod 0o444` the file.
1275 """
1276 start = time.time()
1278 if data_dir is None: 1278 ↛ 1279line 1278 didn't jump to line 1279 because the condition on line 1278 was never true
1279 from ..config.paths import get_journal_data_directory
1281 data_dir = get_journal_data_directory()
1282 if output_path is None: 1282 ↛ 1283line 1282 didn't jump to line 1283 because the condition on line 1282 was never true
1283 output_path = data_dir / DB_FILENAME
1285 logger.info(
1286 "Building journal quality reference DB (one-time, "
1287 "~30s, decompresses ~25 MB of bundled data)…"
1288 )
1290 # Sweep stale temp files from prior crashed builds so they don't
1291 # accumulate. Any .tmp-* older than 1h is assumed dead.
1292 _sweep_stale_tmp_files(output_path.parent, output_path.name)
1294 # Build into a unique temp path, then os.replace() atomically at
1295 # the end. A random suffix (not a fixed .tmp) lets concurrent
1296 # builders write to separate files instead of racing on the same
1297 # path — os.replace picks a winner atomically and neither corrupts
1298 # the live file.
1299 tmp_path = output_path.with_name(
1300 f"{output_path.name}.tmp-{os.getpid()}-{secrets.token_hex(4)}"
1301 )
1303 # Connect by plain filename through a creator, never via a
1304 # ``sqlite:///{tmp_path}`` URL: SQLAlchemy's URL parser treats a
1305 # ``?`` in the data-directory name as the start of a query string,
1306 # so the writer would open (and create) a truncated path and the
1307 # final ``os.replace`` would fail. A plain (non-URI) filename has no
1308 # delimiters. QueuePool is what the URL form selected for a file DB.
1309 from sqlalchemy.pool import QueuePool
1311 def _make_write_conn() -> sqlite3.Connection:
1312 return sqlite3.connect(os.fspath(tmp_path), check_same_thread=False)
1314 engine = create_engine(
1315 "sqlite://", creator=_make_write_conn, poolclass=QueuePool
1316 )
1318 try:
1319 # Pragmas for fast bulk insert. `journal_mode = OFF` plus
1320 # `synchronous = OFF` is deliberately unsafe for general use but
1321 # correct here because durability is guaranteed by the temp-file
1322 # + os.replace() pattern around this block: we write to a unique
1323 # `.tmp-PID-RAND` path, and on any crash mid-build the incomplete
1324 # temp file is orphaned (and swept by `_sweep_stale_tmp_files()`
1325 # on the next build). The live file is only ever moved into place
1326 # by the atomic `os.replace()` at the bottom of this function —
1327 # it never sees a partial write. Do NOT copy this pragma set
1328 # elsewhere without the same atomic rename discipline.
1329 with engine.connect() as conn:
1330 conn.exec_driver_sql("PRAGMA journal_mode = OFF")
1331 conn.exec_driver_sql("PRAGMA synchronous = OFF")
1332 conn.exec_driver_sql("PRAGMA cache_size = -64000")
1333 conn.exec_driver_sql("PRAGMA page_size = 4096")
1335 JournalQualityBase.metadata.create_all(engine)
1337 SessionWrite = sessionmaker(bind=engine)
1338 with SessionWrite() as session:
1339 sources = _load_openalex(data_dir)
1340 doaj_data = _load_doaj(data_dir)
1341 pred_data = _load_predatory(data_dir)
1342 institutions = _load_institutions(data_dir)
1343 abbreviations = _load_abbreviations(data_dir)
1345 _populate_predatory(session, pred_data)
1346 _populate_sources(session, sources, doaj_data, pred_data)
1347 _populate_institutions(session, institutions)
1348 _populate_abbreviations(session, abbreviations)
1349 session.commit()
1351 with engine.connect() as conn:
1352 conn.exec_driver_sql("ANALYZE")
1353 conn.exec_driver_sql("VACUUM")
1354 # Stamp schema version so _ensure_engine can detect drift
1355 # without depending on the external version.json.
1356 conn.exec_driver_sql(
1357 # nosemgrep: semgrep.rules.sql-string-concatenation, reason: Schema version is the module integer constant JOURNAL_QUALITY_SCHEMA_VERSION
1358 f"PRAGMA user_version = {JOURNAL_QUALITY_SCHEMA_VERSION}"
1359 )
1360 except Exception:
1361 engine.dispose()
1362 if tmp_path.exists(): 1362 ↛ 1369line 1362 didn't jump to line 1369 because the condition on line 1362 was always true
1363 try:
1364 # bearer:disable python_lang_file_permissions
1365 os.chmod(tmp_path, 0o644)
1366 tmp_path.unlink()
1367 except OSError:
1368 logger.exception(f"Failed to clean up tmp DB at {tmp_path}")
1369 raise
1371 engine.dispose()
1373 # Atomically swap tmp into place. os.replace is atomic on POSIX and
1374 # overwrites an existing output_path if present.
1375 if output_path.exists():
1376 # Prior file is chmod 0444 from the previous build — relax it
1377 # so os.replace can overwrite. Best-effort: if chmod fails
1378 # (e.g. read-only mount), os.replace will raise and surface
1379 # the real problem. Log so the cause is visible.
1380 try:
1381 # bearer:disable python_lang_file_permissions
1382 os.chmod(output_path, 0o644)
1383 except OSError:
1384 logger.warning(
1385 f"Could not chmod 0644 on existing {output_path} before "
1386 f"os.replace; if the replace fails this is likely why"
1387 )
1388 os.replace(tmp_path, output_path)
1390 # OS-level read-only flag — third layer of write protection.
1391 # POSIX chmod is a no-op on Windows, so we also set the Windows
1392 # read-only file attribute via SetFileAttributesW. The pre-commit
1393 # hook check-journal-quality-readonly.py remains the primary
1394 # defense against accidental writable opens.
1395 # bearer:disable python_lang_file_permissions
1396 os.chmod(output_path, 0o444)
1397 if sys.platform == "win32": 1397 ↛ 1398line 1397 didn't jump to line 1398 because the condition on line 1397 was never true
1398 try:
1399 import ctypes
1401 # FILE_ATTRIBUTE_READONLY = 0x1
1402 ok = ctypes.windll.kernel32.SetFileAttributesW(
1403 str(output_path), 0x1
1404 )
1405 if not ok:
1406 logger.warning(
1407 f"SetFileAttributesW failed on {output_path.name}; "
1408 "readonly pre-commit hook is the sole defense."
1409 )
1410 except Exception:
1411 logger.warning(
1412 f"Could not set Windows readonly attribute on "
1413 f"{output_path.name}"
1414 )
1416 elapsed = time.time() - start
1417 size_mb = output_path.stat().st_size / (1024 * 1024)
1418 with closing(
1419 sqlite3.connect(
1420 sqlite_readonly_uri(Path(output_path).resolve(), immutable=True),
1421 uri=True,
1422 )
1423 ) as _count_conn:
1424 source_count = _count_conn.execute(
1425 "SELECT COUNT(*) FROM sources"
1426 ).fetchone()[0]
1427 logger.info(
1428 f"Journal quality DB ready: {source_count} sources, "
1429 f"{size_mb:.1f} MB in {elapsed:.1f}s ({output_path.name}, chmod 0o444)"
1430 )
1432 reset_db()
1435def _sweep_stale_tmp_files(directory: Path, base_name: str) -> None:
1436 """Remove journal_quality.db.tmp-* files older than 1h.
1438 Per-file OSError (vanished between glob+stat, no permission, etc.)
1439 is logged at debug — the sweep is best-effort and shouldn't stop
1440 the build, but silent-pass on filesystem errors hides the cause of
1441 accumulating stale tmp files that would otherwise eat disk over
1442 time.
1443 """
1444 if not directory.exists():
1445 return
1446 cutoff = time.time() - 3600
1447 for tmp in directory.glob(f"{base_name}.tmp-*"):
1448 try:
1449 if tmp.stat().st_mtime < cutoff:
1450 tmp.unlink()
1451 logger.info(f"Swept stale temp build file: {tmp.name}")
1452 except OSError:
1453 logger.debug(f"Could not sweep stale tmp file {tmp.name}")
1456# ---------------------------------------------------------------------------
1457# Source-data loaders (used by build_db only)
1458# ---------------------------------------------------------------------------
1460# The gzipped snapshots are this pipeline's own output, but a replaced or
1461# corrupted file must not expand without limit into one string. The bound is
1462# derived from the compressed size: compact JSON of these records compresses
1463# well under 32:1, so a file past the bound is refused rather than loaded.
1464_GZIP_SNAPSHOT_RATIO = 32
1465_GZIP_SNAPSHOT_SLACK_CHARS = 1024 * 1024
1466_GZIP_SNAPSHOT_MAX_CHARS = 4 * 1024**3
1469def _load_gzip_json(path: Path) -> dict:
1470 """Parse a gzipped JSON snapshot, refusing one that decompresses too far."""
1471 cap = min(
1472 path.stat().st_size * _GZIP_SNAPSHOT_RATIO + _GZIP_SNAPSHOT_SLACK_CHARS,
1473 _GZIP_SNAPSHOT_MAX_CHARS,
1474 )
1475 with gzip.open(path, "rt", encoding="utf-8") as f:
1476 text = f.read(cap + 1)
1477 if len(text) > cap:
1478 raise ValueError(
1479 f"{path.name} decompresses past {cap:,} characters, "
1480 f"{_GZIP_SNAPSHOT_RATIO}x its compressed size; refusing to load it"
1481 )
1482 data = json.loads(text)
1483 if not isinstance(data, dict):
1484 raise ValueError(f"{path.name} is not a JSON object")
1485 return data
1488def _load_openalex(data_dir: Path) -> dict:
1489 # Lazy import to avoid any downloader <-> db module-level cycle,
1490 # matching the other lazy imports of `.downloader` in this file.
1491 from .downloader import REQUIRED_SNAPSHOT_FILENAME
1493 path = data_dir / REQUIRED_SNAPSHOT_FILENAME
1494 if not path.exists():
1495 raise FileNotFoundError(f"OpenAlex source file not found: {path}")
1496 data = _load_gzip_json(path)
1497 sources = data.get("s", data.get("sources", {}))
1498 logger.info(f"Loaded {len(sources)} OpenAlex sources")
1499 return dict(sources)
1502def _load_doaj(data_dir: Path) -> dict:
1503 path = data_dir / "doaj_journals.json"
1504 if not path.exists():
1505 logger.warning(f"{path} not found — DOAJ cross-ref will be skipped")
1506 return {}
1507 with open(path, encoding="utf-8") as f:
1508 data = json.load(f)
1509 journals = data.get("journals", {})
1510 logger.info(f"Loaded {len(journals)} DOAJ entries")
1511 return dict(journals)
1514def _load_predatory(data_dir: Path) -> dict:
1515 """Returns {journals: set, publishers: set, hijacked: set, long_pubs: list}."""
1516 path = data_dir / "predatory.json"
1517 if not path.exists():
1518 logger.warning(f"{path} not found — predatory check will be skipped")
1519 return {
1520 "journals": set(),
1521 "publishers": set(),
1522 "hijacked": set(),
1523 "long_pubs": [],
1524 }
1526 with open(path, encoding="utf-8") as f:
1527 data = json.load(f)
1529 journal_names = {
1530 normalize_name(e.get("name", ""))
1531 for e in data.get("journals", [])
1532 if e.get("name", "").strip()
1533 }
1534 publisher_names = {
1535 normalize_name(e.get("name", ""))
1536 for e in data.get("publishers", [])
1537 if e.get("name", "").strip()
1538 }
1539 hijacked_names = {
1540 normalize_name(e.get("hijacked_name", ""))
1541 for e in data.get("hijacked", [])
1542 if e.get("hijacked_name", "").strip()
1543 }
1544 long_pubs = [
1545 normalize_name(e.get("name", ""))
1546 for e in data.get("publishers", [])
1547 if len(e.get("name", "").strip()) >= 10
1548 ]
1549 logger.info(
1550 f"Loaded predatory: {len(journal_names)} journals, "
1551 f"{len(publisher_names)} publishers, "
1552 f"{len(hijacked_names)} hijacked"
1553 )
1554 return {
1555 "journals": journal_names,
1556 "publishers": publisher_names,
1557 "hijacked": hijacked_names,
1558 "long_pubs": long_pubs,
1559 }
1562def _load_institutions(data_dir: Path) -> dict:
1563 path = data_dir / "openalex_institutions.json.gz"
1564 if not path.exists():
1565 logger.warning(f"{path} not found — institution tier will be empty")
1566 return {}
1567 data = _load_gzip_json(path)
1568 institutions = data.get("i", {})
1569 logger.info(f"Loaded {len(institutions)} institutions")
1570 return dict(institutions)
1573def _load_abbreviations(data_dir: Path) -> dict:
1574 path = data_dir / "jabref_abbreviations.json.gz"
1575 if not path.exists():
1576 logger.warning(
1577 f"{path} not found — abbreviation expansion will be empty"
1578 )
1579 return {}
1580 data = _load_gzip_json(path)
1581 mappings = data.get("abbrev_to_full", {})
1582 logger.info(f"Loaded {len(mappings)} abbreviation mappings")
1583 return dict(mappings)
1586# ---------------------------------------------------------------------------
1587# Table populators
1588# ---------------------------------------------------------------------------
1591def _populate_predatory(session: Session, pred: dict) -> None:
1592 long_set = set(pred.get("long_pubs", []))
1594 journals = [{"name_lower": n} for n in pred.get("journals", set()) if n]
1595 if journals:
1596 session.bulk_insert_mappings(inspect(PredatoryJournal), journals)
1598 hijacked = [{"name_lower": n} for n in pred.get("hijacked", set()) if n]
1599 if hijacked:
1600 session.bulk_insert_mappings(inspect(PredatoryHijacked), hijacked)
1602 publishers = [
1603 {"name_lower": n, "is_long": n in long_set}
1604 for n in pred.get("publishers", set())
1605 if n
1606 ]
1607 if publishers:
1608 session.bulk_insert_mappings(inspect(PredatoryPublisher), publishers)
1610 logger.info(
1611 f"Inserted predatory tables: "
1612 f"{len(journals)} journals, "
1613 f"{len(publishers)} publishers, "
1614 f"{len(hijacked)} hijacked"
1615 )
1618def _as_number(value):
1619 """Coerce a dataset-supplied numeric field; junk types become None.
1621 The datasets come from a public upstream over HTTPS — the same
1622 hostile-upstream threat model as the manifest URL allowlist (and,
1623 once merged, the pending digest verification, decompression caps
1624 and https-pin PRs). Junk-typed numerics must land as NULL, not
1625 crash the build's comparisons or the dashboard reads. That
1626 includes values SQLite cannot round-trip or store: floats that
1627 are inf/NaN (``json.loads('1e999')`` yields inf) and ints
1628 outside the int64 range an INTEGER column can hold.
1629 """
1630 if isinstance(value, bool) or not isinstance(value, (int, float)):
1631 return None
1632 if isinstance(value, int):
1633 if not -(2**63) <= value <= 2**63 - 1:
1634 return None
1635 elif not math.isfinite(value):
1636 return None
1637 return value
1640def _as_text(value):
1641 """Coerce a dataset-supplied text field; non-strings become None."""
1642 if isinstance(value, str):
1643 stripped = value.strip()
1644 if stripped:
1645 return stripped
1646 return None
1649# Row-count floors for the two populators below. They mirror the
1650# fetch-stage floors that already guard the raw snapshot
1651# (``_MIN_OPENALEX_SOURCES = 10_000`` in
1652# ``data_sources/openalex.py``, ``_MIN_INSTITUTIONS = 50_000`` in
1653# ``data_sources/institutions.py``) — but those floors count records
1654# reaching the snapshot, not how many of those this module's coercion
1655# then drops for a blank/non-string name. A fetch that clears its own
1656# floor can still collapse here: e.g. every fetched record sharing a
1657# well-typed but empty ``display_name`` passes the fetch floor and
1658# then loses 100% of rows to the ``if not name: continue`` skip below,
1659# silently producing a near-empty (or fully empty) table with no
1660# error. Set to half the corresponding fetch floor: normal attrition
1661# from a healthy OpenAlex snapshot is a small fraction of blank names,
1662# so losing more than half of an otherwise fetch-floor-sized snapshot
1663# to name/type coercion is itself evidence of a collapsed or hostile
1664# feed, not ordinary data quality noise.
1665#
1666# The check only applies once the input is already at fetch-floor
1667# scale (``len(sources) >= _MIN_BUILD_SOURCES``, same pattern for
1668# institutions) — mirroring the fetch stage's own schema-drift check
1669# (``openalex.py``'s ``if len(journal_sample) >= _SCHEMA_SAMPLE_SIZE``
1670# guard), which likewise only fires once the sample is large enough
1671# to be meaningful. A hand-built test fixture of a handful of records
1672# could never pass the fetch-stage floor in the first place, so it
1673# never reaches build-floor scale either and the check does not fire
1674# for it; only a real (or realistically-sized) snapshot that
1675# collapses during populate does.
1676_MIN_BUILD_SOURCES = 5_000 # half of openalex.py's _MIN_OPENALEX_SOURCES
1677_MIN_BUILD_INSTITUTIONS = 25_000 # half of institutions.py's _MIN_INSTITUTIONS
1680def _populate_sources(
1681 session: Session,
1682 sources: dict,
1683 doaj_data: dict,
1684 pred: dict,
1685) -> None:
1686 """Build Source rows with cross-referenced DOAJ + predatory flags."""
1687 type_map = {"j": "journal", "c": "conference"}
1688 pred_journals = pred.get("journals", set())
1689 pred_publishers = pred.get("publishers", set())
1690 pred_hijacked = pred.get("hijacked", set())
1692 # Dedup key is (name_lower, issn or "") so journals with separate
1693 # print and electronic ISSNs in OpenAlex both survive instead of
1694 # collapsing onto one row.
1695 seen: dict[tuple[str, str], dict] = {}
1696 skipped = 0
1698 for source_id, compact in sources.items():
1699 name = _as_text(compact.get("n"))
1700 if not name:
1701 skipped += 1
1702 continue
1704 name_lower = normalize_name(name)
1705 issn = normalize_issn(_as_text(compact.get("i")))
1706 publisher = _as_text(compact.get("p"))
1707 h_index = _as_number(compact.get("h"))
1708 impact_factor = _as_number(compact.get("if"))
1709 cited_by_count = _as_number(compact.get("cb"))
1710 type_tag = _as_text(compact.get("t")) or ""
1711 source_type = type_map.get(type_tag, type_tag)
1713 doaj_entry = doaj_data.get(issn) if issn else None
1714 is_in_doaj = doaj_entry is not None
1716 is_pred = name_lower in pred_journals
1717 pred_source = "stop-predatory-journals" if is_pred else None
1718 if not is_pred and publisher:
1719 pub_norm = normalize_name(publisher)
1720 if pub_norm in pred_publishers:
1721 is_pred = True
1722 pred_source = "stop-predatory-publishers"
1723 if not is_pred and name_lower in pred_hijacked:
1724 is_pred = True
1725 pred_source = "stop-predatory-hijacked"
1727 # Whitelist override
1728 if is_pred and (
1729 is_in_doaj or (h_index or 0) > PREDATORY_WHITELIST_HINDEX
1730 ):
1731 is_pred = False
1732 pred_source = None
1734 quality = derive_quality_score(
1735 h_index=h_index,
1736 is_in_doaj=is_in_doaj,
1737 is_predatory=is_pred,
1738 source_type=source_type,
1739 )
1741 rec = {
1742 "name": name,
1743 "name_lower": name_lower,
1744 "issn": issn,
1745 "openalex_source_id": source_id,
1746 "source_type": source_type,
1747 "publisher": publisher,
1748 "h_index": h_index,
1749 "impact_factor": impact_factor,
1750 "cited_by_count": cited_by_count,
1751 "quartile": None, # filled in by the post-pass below
1752 "quality": quality,
1753 "is_in_doaj": is_in_doaj,
1754 "is_predatory": is_pred,
1755 "predatory_source": pred_source,
1756 "score_source": "openalex",
1757 }
1759 key = (name_lower, issn or "")
1760 prev = seen.get(key)
1761 if prev is None or (h_index or 0) > (prev.get("h_index") or 0): 1761 ↛ 1698line 1761 didn't jump to line 1698 because the condition on line 1761 was always true
1762 seen[key] = rec
1764 if skipped:
1765 logger.warning(
1766 f"OpenAlex sources: skipped {skipped:,} of {len(sources):,} "
1767 "records with a blank or non-string name"
1768 )
1769 kept = len(sources) - skipped
1770 if len(sources) >= _MIN_BUILD_SOURCES and kept < _MIN_BUILD_SOURCES: 1770 ↛ 1771line 1770 didn't jump to line 1771 because the condition on line 1770 was never true
1771 raise RuntimeError(
1772 f"OpenAlex sources: suspiciously few records survived "
1773 f"ingestion ({kept:,} < {_MIN_BUILD_SOURCES:,} of "
1774 f"{len(sources):,} fetched); refusing to overwrite existing data"
1775 )
1777 # Second pass: DOAJ-only journals (not in OpenAlex). Without this
1778 # we lose ~4-7K small open-access venues. Keyed by name_lower so
1779 # we don't double-insert anything OpenAlex already covered.
1780 openalex_names = {k[0] for k in seen.keys()}
1781 doaj_added = 0
1782 for issn, doaj_entry in doaj_data.items():
1783 name = (doaj_entry.get("name") or "").strip()
1784 if not name:
1785 continue
1786 name_lower = normalize_name(name)
1787 if name_lower in openalex_names:
1788 continue
1789 publisher = doaj_entry.get("publisher") or None
1790 quality = derive_quality_score(
1791 h_index=None,
1792 is_in_doaj=True,
1793 is_predatory=False,
1794 source_type="journal",
1795 )
1796 seen[(name_lower, issn or "")] = {
1797 "name": name,
1798 "name_lower": name_lower,
1799 "issn": issn,
1800 "openalex_source_id": None,
1801 "source_type": "journal",
1802 "publisher": publisher,
1803 "h_index": None,
1804 "impact_factor": None,
1805 "cited_by_count": None,
1806 "quartile": None,
1807 "quality": quality,
1808 "is_in_doaj": True,
1809 "is_predatory": False,
1810 "predatory_source": None,
1811 "score_source": "doaj",
1812 }
1813 openalex_names.add(name_lower)
1814 doaj_added += 1
1816 records = list(seen.values())
1818 # Derive quartile (Q1–Q4) from cited_by_count percentile within each
1819 # source_type. Field-specific quartiles would be more accurate but
1820 # require per-source topic data that 4–7×s the snapshot size, so we
1821 # use global per-type percentiles as a defensible approximation
1822 # given the license constraint that ruled out SJR.
1823 by_type: dict[str, list[dict]] = {}
1824 for r in records:
1825 if r.get("cited_by_count") is None:
1826 continue # NULL quartile for entries without citation data
1827 by_type.setdefault(r.get("source_type") or "", []).append(r)
1828 for type_records in by_type.values():
1829 type_records.sort(key=lambda r: r["cited_by_count"])
1830 n = len(type_records)
1831 if n == 0: 1831 ↛ 1832line 1831 didn't jump to line 1832 because the condition on line 1831 was never true
1832 continue
1833 for rank, r in enumerate(type_records):
1834 pct = rank / n # 0.0 = lowest, ~1.0 = highest
1835 if pct >= 0.75:
1836 r["quartile"] = "Q1"
1837 elif pct >= 0.50:
1838 r["quartile"] = "Q2"
1839 elif pct >= 0.25:
1840 r["quartile"] = "Q3"
1841 else:
1842 r["quartile"] = "Q4"
1844 # Re-derive quality now that quartile is available. The first-pass
1845 # `quality` values above were computed without quartile and are
1846 # therefore suboptimal — a Q1 journal without h-index data would
1847 # have scored `None` (fall-through) instead of 8. The runtime filter
1848 # code in journal_reputation_filter.py does pass quartile, so the
1849 # stored column should agree with the live score.
1850 for r in records:
1851 r["quality"] = derive_quality_score(
1852 h_index=r.get("h_index"),
1853 quartile=r.get("quartile"),
1854 is_in_doaj=r.get("is_in_doaj") or False,
1855 is_predatory=r.get("is_predatory") or False,
1856 source_type=r.get("source_type"),
1857 )
1859 logger.info(
1860 f"Inserting {len(records)} source records ({doaj_added} DOAJ-only)..."
1861 )
1862 for i in range(0, len(records), _BATCH_SIZE):
1863 session.bulk_insert_mappings(
1864 inspect(Source), records[i : i + _BATCH_SIZE]
1865 )
1868def _populate_institutions(session: Session, institutions: dict) -> None:
1869 records: list[dict] = []
1870 skipped = 0
1871 for inst_id, compact in institutions.items():
1872 name = _as_text(compact.get("n"))
1873 if not name:
1874 skipped += 1
1875 continue
1876 records.append(
1877 {
1878 "openalex_id": inst_id,
1879 "name": name,
1880 "name_lower": normalize_name(name),
1881 "ror_id": _as_text(compact.get("r")),
1882 "country": _as_text(compact.get("c")),
1883 "type": _as_text(compact.get("t")),
1884 "h_index": _as_number(compact.get("h")),
1885 "impact_factor": _as_number(compact.get("if")),
1886 "works_count": _as_number(compact.get("w")),
1887 "cited_by_count": _as_number(compact.get("cb")),
1888 }
1889 )
1890 if skipped:
1891 logger.warning(
1892 f"OpenAlex institutions: skipped {skipped:,} of "
1893 f"{len(institutions):,} records with a blank or non-string name"
1894 )
1895 kept = len(records)
1896 if ( 1896 ↛ 1900line 1896 didn't jump to line 1900 because the condition on line 1896 was never true
1897 len(institutions) >= _MIN_BUILD_INSTITUTIONS
1898 and kept < _MIN_BUILD_INSTITUTIONS
1899 ):
1900 raise RuntimeError(
1901 f"OpenAlex institutions: suspiciously few records survived "
1902 f"ingestion ({kept:,} < {_MIN_BUILD_INSTITUTIONS:,} of "
1903 f"{len(institutions):,} fetched); refusing to overwrite "
1904 "existing data"
1905 )
1906 logger.info(f"Inserting {len(records)} institution records...")
1907 for i in range(0, len(records), _BATCH_SIZE):
1908 session.bulk_insert_mappings(
1909 inspect(Institution), records[i : i + _BATCH_SIZE]
1910 )
1913def _populate_abbreviations(session: Session, mappings: dict) -> None:
1914 records: list[dict] = []
1915 seen: set[str] = set()
1916 for abbrev, full in mappings.items():
1917 norm = normalize_name(abbrev)
1918 if not norm or norm in seen:
1919 continue
1920 seen.add(norm)
1921 records.append({"abbrev_lower": norm, "full_name": full})
1922 logger.info(f"Inserting {len(records)} abbreviation records...")
1923 for i in range(0, len(records), _BATCH_SIZE):
1924 session.bulk_insert_mappings(
1925 inspect(Abbreviation), records[i : i + _BATCH_SIZE]
1926 )
1929# ---------------------------------------------------------------------------
1930# Backwards-compat shim for the old build_reference_db name
1931# ---------------------------------------------------------------------------
1934def build_reference_db(
1935 data_dir: Optional[Path] = None,
1936 output_path: Optional[Path] = None,
1937) -> None:
1938 """Deprecated alias for `build_db`."""
1939 build_db(data_dir=data_dir, output_path=output_path)