Coverage for src/local_deep_research/journal_quality/db.py: 55%
669 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-06 15:42 +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 os
29import secrets
30import sqlite3
31import sys
32import threading
33import time
34from contextlib import closing, contextmanager
35from pathlib import Path
36from typing import Iterable, Iterator, Optional
38from loguru import logger
39from sqlalchemy import create_engine, func, inspect, or_, select
40from sqlalchemy.engine import Engine
41from sqlalchemy.exc import DatabaseError, OperationalError
42from sqlalchemy.orm import Session, sessionmaker
44from .models import (
45 Abbreviation,
46 Institution,
47 JournalQualityBase,
48 PredatoryHijacked,
49 PredatoryJournal,
50 PredatoryPublisher,
51 Source,
52)
53from ..constants import PREDATORY_WHITELIST_HINDEX
54from ..utilities.citation_normalizer import normalize_issn
55from ..utilities.sql_utils import escape_like
56from .scoring import (
57 derive_quality_score,
58 institution_score_from_h_index,
59 normalize_name,
60)
62DB_FILENAME = "journal_quality.db"
63_BATCH_SIZE = 5000
65# Bump when the reference DB schema (models.py) changes in a way that
66# requires a rebuild even if the upstream data version hasn't changed.
67# Stamped as SQLite `PRAGMA user_version` during build_db; checked on
68# _ensure_engine. Separate from JOURNAL_DATA_VERSION (downloader.py)
69# which tracks upstream source-data freshness.
70# v4: dropped Source.has_doaj_seal — DOAJ retired the Seal in April 2025.
71JOURNAL_QUALITY_SCHEMA_VERSION = 4
73# Quality tier → score range, used by the dashboard tier filter.
74_TIER_RANGES = {
75 "elite": (9, 10),
76 "strong": (7, 8),
77 "moderate": (5, 6),
78 "low": (3, 4),
79 "predatory": (1, 2),
80}
82# Columns safe to use in ORDER BY (prevents injection via the dashboard
83# `sort` query parameter).
84_SORT_COLUMNS = frozenset(
85 {
86 "name",
87 "quality",
88 "quartile",
89 "h_index",
90 "impact_factor",
91 "score_source",
92 "source_type",
93 "publisher",
94 "is_predatory",
95 }
96)
98# Max length for user-supplied search strings. Even with LIKE
99# wildcards escaped, a 10 KB pattern against 217K rows is slow enough
100# to matter for CPU budget under concurrent requests.
101_MAX_SEARCH_LEN = 100
104# ---------------------------------------------------------------------------
105# Read-only accessor
106# ---------------------------------------------------------------------------
109class JournalQualityDB:
110 """Read-only SQLAlchemy 2.0 accessor for `journal_quality.db`.
112 All filter hot-path methods return plain dicts (not mapped Source
113 objects) so call sites in `journal_reputation_filter.py` keep the
114 same call shape they had against the dict-based predecessor. The
115 dashboard methods can return either dicts or Source instances —
116 they're called once per page view so the ORM overhead is fine.
117 """
119 def __init__(self) -> None:
120 self._engine: Optional[Engine] = None
121 self._SessionLocal: Optional[sessionmaker[Session]] = None
122 # RLock (not Lock): _ensure_engine holds the lock while calling
123 # _build_or_raise → build_db → reset_db → self.reset(), which
124 # re-acquires the same lock. A non-reentrant Lock would deadlock
125 # the very first request on a fresh install.
126 self._lock = threading.RLock()
127 # Whether we've already logged a stale-data-version warning for
128 # this engine lifetime. Prevents log spam — one WARNING per
129 # server start is enough to surface the problem to admins.
130 self._stale_version_warned = False
132 # --- engine + session lifecycle ---
134 def _resolve_db_path(self) -> Path:
135 from ..config.paths import get_journal_data_directory
137 return get_journal_data_directory() / DB_FILENAME
139 def _ensure_engine(self) -> None:
140 # Acquire lock BEFORE first read to avoid DCLP publication hazard.
141 # With the GIL this is safe on CPython but explicit locking makes
142 # the happens-before relationship clear and portable.
143 with self._lock:
144 if self._engine is not None:
145 return
146 path = self._resolve_db_path()
147 if not path.exists():
148 self._build_or_raise(path)
149 else:
150 # Validate existing file before wiring up the read-only
151 # engine. Catches two failure modes at open time instead
152 # of letting them propagate to first query:
153 # 1. Schema drift — ORM changed since this file was
154 # built (PRAGMA user_version mismatch) → rebuild.
155 # 2. Corruption — file exists but isn't a valid DB
156 # (truncated build, disk error) → rebuild.
157 if not self._validate_existing_db(path):
158 self._build_or_raise(path)
160 # mode=ro + immutable=1: SQLite physically refuses writes,
161 # skips locking entirely, and reads via mmap. The OS page
162 # cache holds one shared resident copy of the hot pages.
163 #
164 # Use a creator callback because SQLAlchemy's URL parser
165 # eats the ?mode=ro&immutable=1 query string before it can
166 # reach sqlite3. The creator builds the connection directly
167 # with the SQLite URI flags intact.
168 def _make_ro_conn() -> sqlite3.Connection:
169 return sqlite3.connect(
170 f"file:{path}?mode=ro&immutable=1",
171 uri=True,
172 )
174 # NullPool: each Session checks out an isolated read-only
175 # connection that is closed when the session context exits.
176 # With mode=ro&immutable=1, SQLite connection opens are
177 # trivial (no lock overhead) and the OS page cache
178 # shares data across connections. NullPool ensures that
179 # concurrent filter_results threads never share a single
180 # sqlite3.Connection handle (eliminating the cursor race condition
181 # that caused sqlite3.InterfaceError / IndexError under StaticPool).
182 from sqlalchemy.pool import NullPool
184 engine = create_engine(
185 "sqlite://",
186 creator=_make_ro_conn,
187 poolclass=NullPool,
188 echo=False,
189 )
190 session_local = sessionmaker(bind=engine, expire_on_commit=False)
191 # Publish both together so readers never see engine-without-session.
192 self._engine = engine
193 self._SessionLocal = session_local
194 logger.info(f"Opened journal_quality.db (read-only): {path}")
195 # One-shot check: is the DATA version (the one the sources
196 # JSON + build logic produce) behind the bundled latest?
197 # Schema drift is already handled by `_validate_existing_db`
198 # via ``PRAGMA user_version``. A data-version mismatch is a
199 # different concern: the DB schema is fine, but the scoring
200 # logic (e.g. the repository cap) or source snapshots have
201 # been updated since this file was built. The hot path
202 # (filter scoring) would silently serve stale scores if we
203 # didn't surface the mismatch anywhere but the admin
204 # dashboard. Log once, don't auto-rebuild — user consent
205 # via the dashboard "Download Data" button remains the
206 # explicit refresh trigger.
207 self._warn_on_stale_data_version(path.parent)
209 def _warn_on_stale_data_version(self, data_dir: Path) -> None:
210 """Log WARNING once if ``version.json`` is behind ``JOURNAL_DATA_VERSION``."""
211 if self._stale_version_warned:
212 return
213 # Lazy import to avoid any downloader → db cycle even though
214 # today's module graph doesn't have one.
215 from .downloader import JOURNAL_DATA_VERSION
217 version_file = data_dir / "version.json"
218 if not version_file.exists():
219 return # Brand-new install — the dashboard's banner handles this.
220 try:
221 with open(version_file, encoding="utf-8") as f:
222 info = json.load(f)
223 installed = info.get("version")
224 except (json.JSONDecodeError, OSError):
225 return # Malformed — dashboard's banner surfaces; don't double-log.
226 if installed and installed != JOURNAL_DATA_VERSION:
227 logger.warning(
228 f"journal_quality data version is stale: on-disk={installed!r} "
229 f"bundled-latest={JOURNAL_DATA_VERSION!r}. "
230 f"Scoring is continuing with the older data. Visit "
231 f"/metrics/journals and click 'Download Data' to refresh."
232 )
233 self._stale_version_warned = True
235 def _validate_existing_db(self, path: Path) -> bool:
236 """Return True if the existing DB file is usable as-is.
238 A version of 0 means the file was built before schema stamping
239 existed and is grandfathered in — we don't force a rebuild just
240 because the stamp is missing. A non-zero version that doesn't
241 match the current schema is a real drift signal and triggers a
242 rebuild. File-open errors also trigger a rebuild.
243 """
244 from ..utilities.resource_utils import safe_close
246 conn: Optional[sqlite3.Connection] = None
247 is_valid = False
248 try:
249 conn = sqlite3.connect(f"file:{path}?mode=ro", uri=True)
250 version = conn.execute("PRAGMA user_version").fetchone()[0]
251 if version != 0 and version != JOURNAL_QUALITY_SCHEMA_VERSION:
252 logger.warning(
253 f"journal_quality.db schema_version={version}, "
254 f"expected {JOURNAL_QUALITY_SCHEMA_VERSION} — "
255 f"rebuilding"
256 )
257 else:
258 # Cheap sanity check — confirms the file is a valid DB.
259 conn.execute("SELECT 1 FROM sqlite_master LIMIT 1").fetchone()
260 is_valid = True
261 except (sqlite3.DatabaseError, OSError):
262 logger.exception(
263 f"journal_quality.db at {path} is unusable; rebuilding"
264 )
265 finally:
266 if conn is not None:
267 safe_close(conn, "journal_quality validate")
269 if not is_valid:
270 self._unlink_unusable_db(path)
271 return is_valid
273 @staticmethod
274 def _unlink_unusable_db(path: Path) -> None:
275 """Best-effort cleanup of a corrupted / schema-drifted DB file.
277 Corruption was already logged by the caller (``_validate_existing_db``).
278 Both operations below are best-effort — on failure we *log and
279 continue* rather than raise, because the build path will rebuild
280 the file regardless. But we don't silence: if chmod / unlink
281 fails (permissions, read-only mount, file held open on Windows)
282 the next build will likely also fail and the user needs the
283 warning to diagnose the real problem.
284 """
285 try:
286 # bearer:disable python_lang_file_permissions
287 os.chmod(path, 0o644)
288 except OSError:
289 logger.warning(
290 f"Could not chmod 0644 on unusable DB {path} before "
291 f"unlink (continuing to unlink attempt)"
292 )
293 try:
294 path.unlink()
295 except OSError:
296 logger.warning(
297 f"Could not unlink unusable DB {path} (will be "
298 f"overwritten on next build)"
299 )
301 def _build_or_raise(self, path: Path) -> None:
302 """Lazy-build the DB on first access if it's missing.
304 Under the test suite's mock-only mode (``LDR_TESTING_WITH_MOCKS``,
305 defaulted to true by ``tests/conftest.py``) this refuses to
306 *download*. The method is reached implicitly — any read through
307 ``_ensure_engine`` when the file is missing or fails validation — so
308 a test that never intended to exercise the download path could
309 otherwise trigger a live, multi-hundred-MB fetch from OpenAlex/DOAJ
310 inside a request. ``timeout_method="thread"`` cannot interrupt a
311 blocked socket, so the xdist worker dies instead of the test failing.
313 It gates ``auto_download`` rather than refusing outright, and the
314 difference matters: ``ensure_journal_data`` returns immediately when
315 the gz snapshots are already on disk, which is a purely local,
316 network-free path. Refusing before that call would also block a
317 developer or CI job that has the snapshots, turning a working offline
318 build into a failure. With ``auto_download=False`` the local build
319 still happens and only the network is off the table.
321 Tests that DO want the real download orchestration call
322 ``download_journal_data()`` directly with the per-source fetchers
323 mocked (see ``tests/journal_quality/test_downloader.py``), which this
324 does not touch.
325 """
326 from ..settings.env_definitions.testing import testing_with_mocks
327 from .downloader import ensure_journal_data
329 mock_mode = testing_with_mocks()
330 data_dir, available = ensure_journal_data(auto_download=not mock_mode)
331 if not available: 331 ↛ 346line 331 didn't jump to line 346 because the condition on line 331 was always true
332 if mock_mode: 332 ↛ 341line 332 didn't jump to line 341 because the condition on line 332 was always true
333 raise FileNotFoundError(
334 "Journal data files not available and "
335 "LDR_TESTING_WITH_MOCKS=true — refusing to download "
336 "from OpenAlex/DOAJ during a test run. Seed "
337 "journal_quality.db directly, or place the gz "
338 "snapshots in the data directory, if this test needs "
339 "real data."
340 )
341 raise FileNotFoundError(
342 "Journal data files not available. "
343 "Check your network connection or download manually "
344 "from the dashboard."
345 )
346 logger.info(f"Building {DB_FILENAME} from data files...")
347 build_db(data_dir=data_dir, output_path=path)
349 @contextmanager
350 def session(self) -> Iterator[Session]:
351 """Yield a read-only SQLAlchemy session.
353 If the underlying file becomes corrupt mid-session (e.g. a
354 rebuild ran and this engine is pointed at a now-unlinked inode),
355 DatabaseError propagates but we drop the cached engine so the
356 next call rebuilds cleanly instead of failing forever.
357 """
358 self._ensure_engine()
359 if self._SessionLocal is None: 359 ↛ 360line 359 didn't jump to line 360 because the condition on line 359 was never true
360 raise RuntimeError("JournalQualityDB engine failed to initialize")
361 from ..utilities.resource_utils import safe_close
363 s = self._SessionLocal()
364 try:
365 yield s
366 except (OperationalError, DatabaseError):
367 logger.exception("journal_quality.db error — resetting engine")
368 self.reset()
369 raise
370 finally:
371 safe_close(s, "journal_quality session")
373 @property
374 def available(self) -> bool:
375 try:
376 self._ensure_engine()
377 return True
378 except FileNotFoundError:
379 return False
381 def reset(self) -> None:
382 """Drop the cached engine — call after `build_db` rebuilds the file."""
383 with self._lock:
384 if self._engine is not None:
385 self._engine.dispose()
386 self._engine = None
387 self._SessionLocal = None
388 logger.info("Reset journal_quality.db engine")
390 # --- filter hot path: return plain dicts ---
392 def lookup_openalex(
393 self,
394 *,
395 source_id: Optional[str] = None,
396 issn: Optional[str] = None,
397 name: Optional[str] = None,
398 ) -> Optional[dict]:
399 """Look up a source by OpenAlex ID, ISSN, or name.
401 Returns a dict with the same shape the dict-based predecessor
402 produced (`name`, `type`, `h_index`, `impact_factor`,
403 `is_in_doaj`, `publisher`, `issn_l`) so the filter code at
404 `journal_reputation_filter.py` doesn't need to change.
405 """
406 issn = normalize_issn(issn)
407 try:
408 self._ensure_engine()
409 except FileNotFoundError:
410 return None
411 with self.session() as s:
412 row = self._lookup_source_row(s, source_id, issn, name)
413 return _source_to_lookup_dict(row) if row else None
415 # Alias used by the dashboard / future call sites
416 lookup_source = lookup_openalex
418 def count_predatory_by_names(self, names: Iterable[str]) -> int:
419 """Count how many of the given journal names are flagged predatory.
421 One SQL round-trip using ``WHERE name_lower IN (…) AND is_predatory = TRUE``.
422 Names are normalized (NFKC + lower + strip) so the caller can pass raw
423 display names; matches the normalization used at build time.
425 Used by the per-user metrics dashboard to report a global "N predatory
426 journals across all your research" stat without making N round trips
427 to the reference DB. Returns 0 if the reference DB is missing or if
428 ``names`` is empty.
430 .. note::
431 Deliberately unchunked. SQLite's ``SQLITE_MAX_VARIABLE_NUMBER``
432 has been 250,000 since SQLite 3.32 (2020); Python 3.11 ships
433 with 3.45.1. A heavy user with 100k distinct container_titles
434 is still well under the limit. Re-confirmed in the PR #3081
435 audit — no chunking needed.
436 """
437 normed = {normalize_name(n) for n in names if n}
438 normed.discard("")
439 if not normed:
440 return 0
441 try:
442 self._ensure_engine()
443 except FileNotFoundError:
444 return 0
445 with self.session() as s:
446 stmt = select(func.count(Source.id)).where(
447 Source.name_lower.in_(normed),
448 Source.is_predatory.is_(True),
449 )
450 result = s.execute(stmt).scalar()
451 return int(result or 0)
453 def lookup_sources_batch(self, names: Iterable[str]) -> dict:
454 """Batch-look-up multiple journal names in one query.
456 Takes an iterable of raw display names and returns a
457 ``{normalized_name: dashboard_dict}`` map for every name that
458 matched a Source. Names that didn't match are simply absent
459 from the result (caller decides how to handle misses).
461 Dashboard hot path: the ``/api/journals/user-research`` endpoint
462 collects up to 200 unique ``container_title`` values from the
463 user's Papers and hands them in here — one SQL round-trip vs.
464 200 per-row lookups.
466 Normalization matches ``normalize_name`` (NFKC + lower + strip)
467 so the reference DB's ``name_lower`` column hits directly. No
468 "the " / "proceedings of" fallback tiers — those live in
469 ``_lookup_source_row`` for precision per-call; the batch path
470 is for dashboard display where a miss is acceptable.
472 Chunked at 900 params per chunk. Defensive: SQLite's actual
473 limit (``SQLITE_MAX_VARIABLE_NUMBER``) has been 250,000 since
474 3.32 (2020) — we could easily put the whole batch in one IN —
475 but 900 keeps us well under any older embedded-SQLite ceiling
476 a deployment might pin to.
477 """
478 normed = [normalize_name(n) for n in names if n]
479 normed = [n for n in normed if n]
480 if not normed: 480 ↛ 481line 480 didn't jump to line 481 because the condition on line 480 was never true
481 return {}
482 try:
483 self._ensure_engine()
484 except FileNotFoundError:
485 return {}
486 # De-duplicate while preserving insertion order for stable
487 # iteration in tests.
488 seen: set = set()
489 uniq: list = []
490 for n in normed:
491 if n not in seen: 491 ↛ 490line 491 didn't jump to line 490 because the condition on line 491 was always true
492 seen.add(n)
493 uniq.append(n)
495 out: dict = {}
496 CHUNK = 900
497 with self.session() as s:
498 for i in range(0, len(uniq), CHUNK):
499 batch = uniq[i : i + CHUNK]
500 stmt = select(Source).where(Source.name_lower.in_(batch))
501 for row in s.scalars(stmt):
502 out[row.name_lower] = _source_to_dashboard_dict(row)
503 return out
505 def lookup_doaj(self, *, issn: Optional[str] = None) -> Optional[dict]:
506 issn = normalize_issn(issn)
507 if not issn:
508 return None
509 try:
510 self._ensure_engine()
511 except FileNotFoundError:
512 return None
513 with self.session() as s:
514 stmt = (
515 select(Source)
516 .where(Source.issn == issn, Source.is_in_doaj.is_(True))
517 .limit(1)
518 )
519 row = s.scalars(stmt).first()
520 if row is None:
521 return None
522 return {
523 "name": row.name,
524 "publisher": row.publisher,
525 }
527 def is_in_doaj(self, issn: Optional[str]) -> bool:
528 return self.lookup_doaj(issn=issn) is not None
530 def is_predatory(
531 self,
532 *,
533 journal_name: Optional[str] = None,
534 publisher_name: Optional[str] = None,
535 ) -> tuple[bool, Optional[str]]:
536 """Check if a journal/publisher is on the predatory list.
538 Looks up the dedicated predatory tables (NOT just the
539 `is_predatory` flag on `Source`), so checks work for arbitrary
540 input names that aren't in OpenAlex.
541 """
542 try:
543 self._ensure_engine()
544 except FileNotFoundError:
545 return False, None
546 with self.session() as s:
547 if journal_name: 547 ↛ 554line 547 didn't jump to line 554 because the condition on line 547 was always true
548 norm = normalize_name(journal_name)
549 if s.get(PredatoryJournal, norm) is not None: 549 ↛ 550line 549 didn't jump to line 550 because the condition on line 549 was never true
550 return True, "stop-predatory-journals"
551 if s.get(PredatoryHijacked, norm) is not None: 551 ↛ 552line 551 didn't jump to line 552 because the condition on line 551 was never true
552 return True, "stop-predatory-hijacked"
554 if publisher_name: 554 ↛ 555line 554 didn't jump to line 555 because the condition on line 554 was never true
555 pub_norm = normalize_name(publisher_name)
556 if s.get(PredatoryPublisher, pub_norm) is not None:
557 return True, "stop-predatory-publishers"
558 # Substring scan over long entries (~1162 rows)
559 stmt = select(PredatoryPublisher.name_lower).where(
560 PredatoryPublisher.is_long.is_(True)
561 )
562 for (entry,) in s.execute(stmt).all():
563 if pub_norm in entry or entry in pub_norm:
564 return True, "stop-predatory-publishers"
566 return False, None
568 def is_whitelisted(
569 self,
570 *,
571 issn: Optional[str] = None,
572 name: Optional[str] = None,
573 ) -> bool:
574 if self.is_in_doaj(issn):
575 return True
576 oa = self.lookup_openalex(issn=issn, name=name)
577 if oa and (oa.get("h_index") or 0) > PREDATORY_WHITELIST_HINDEX:
578 return True
579 return False
581 def lookup_institution(
582 self,
583 *,
584 ror_id: Optional[str] = None,
585 openalex_id: Optional[str] = None,
586 name: Optional[str] = None,
587 ) -> Optional[dict]:
588 """Look up an institution.
590 Order: openalex_id → ror → name. Returns a dict with full-name
591 keys (``name``, ``country``, ``type``, ``h_index``,
592 ``impact_factor``, ``works_count``, ``cited_by_count``, ``ror_id``)
593 or ``None`` if no match. The on-disk snapshot uses one-character
594 keys (``n``, ``c``, etc.) for space efficiency; the accessor
595 returns full names instead for legibility and schema robustness.
596 """
597 try:
598 self._ensure_engine()
599 except FileNotFoundError:
600 return None
601 with self.session() as s:
602 row: Optional[Institution] = None
604 if openalex_id:
605 sid = openalex_id.split("/")[-1]
606 row = s.get(Institution, sid)
608 if row is None and ror_id:
609 ror = ror_id.rstrip("/").split("/")[-1]
610 stmt = (
611 select(Institution)
612 .where(Institution.ror_id == ror)
613 .limit(1)
614 )
615 row = s.scalars(stmt).first()
617 if row is None and name:
618 norm = normalize_name(name)
619 stmt = (
620 select(Institution)
621 .where(Institution.name_lower == norm)
622 .limit(1)
623 )
624 row = s.scalars(stmt).first()
626 return _institution_to_dict(row) if row else None
628 def score_from_affiliations(self, affiliations: list) -> Optional[int]:
629 """Derive a score from author affiliations in ONE SQL query."""
630 if not affiliations:
631 return None
633 openalex_ids: list[str] = []
634 ror_ids: list[str] = []
635 names: list[str] = []
637 for aff in affiliations:
638 if isinstance(aff, str):
639 names.append(normalize_name(aff))
640 elif isinstance(aff, dict):
641 if oid := (aff.get("openalex_id") or aff.get("id")):
642 openalex_ids.append(oid.split("/")[-1])
643 if rid := aff.get("ror"):
644 ror_ids.append(rid.rstrip("/").split("/")[-1])
645 if nm := aff.get("name"):
646 names.append(normalize_name(nm))
648 if not (openalex_ids or ror_ids or names):
649 return None
651 try:
652 self._ensure_engine()
653 except FileNotFoundError:
654 return None
656 clauses = []
657 if openalex_ids:
658 clauses.append(Institution.openalex_id.in_(openalex_ids))
659 if ror_ids:
660 clauses.append(Institution.ror_id.in_(ror_ids))
661 if names:
662 clauses.append(Institution.name_lower.in_(names))
664 with self.session() as s:
665 stmt = select(func.max(Institution.h_index)).where(
666 or_(*clauses), Institution.h_index.is_not(None)
667 )
668 best_h = s.scalar(stmt)
670 # Single source of truth for institution scoring lives in
671 # scoring.py — delegate so the build phase, the runtime filter,
672 # and this affiliation-salvage path can never disagree.
673 return institution_score_from_h_index(best_h)
675 # Static passthrough so the filter can call dm.derive_quality_score(...)
676 # without importing from .scoring directly. Single home for the
677 # scoring rules in scoring.py.
678 derive_quality_score = staticmethod(derive_quality_score)
680 def expand_abbreviation(self, name: str) -> Optional[str]:
681 if not name:
682 return None
683 try:
684 self._ensure_engine()
685 except FileNotFoundError:
686 return None
687 normalized = normalize_name(name)
688 with self.session() as s:
689 row = s.get(Abbreviation, normalized)
690 if row is not None: 690 ↛ 691line 690 didn't jump to line 691 because the condition on line 690 was never true
691 return row.full_name
692 no_dots = normalized.replace(".", "").strip()
693 if no_dots != normalized: 693 ↛ 694line 693 didn't jump to line 694 because the condition on line 693 was never true
694 row = s.get(Abbreviation, no_dots)
695 if row is not None:
696 return row.full_name
697 return None
699 # --- internal source lookup with name fallbacks ---
701 def _lookup_source_row(
702 self,
703 s: Session,
704 source_id: Optional[str],
705 issn: Optional[str],
706 name: Optional[str],
707 ) -> Optional[Source]:
708 if source_id: 708 ↛ 709line 708 didn't jump to line 709 because the condition on line 708 was never true
709 sid = source_id.split("/")[-1] if "/" in source_id else source_id
710 stmt = (
711 select(Source).where(Source.openalex_source_id == sid).limit(1)
712 )
713 row = s.scalars(stmt).first()
714 if row is not None:
715 return row
717 if issn:
718 stmt = select(Source).where(Source.issn == issn).limit(1)
719 row = s.scalars(stmt).first()
720 if row is not None: 720 ↛ 723line 720 didn't jump to line 723 because the condition on line 720 was always true
721 return row
723 if name: 723 ↛ 773line 723 didn't jump to line 773 because the condition on line 723 was always true
724 norm = normalize_name(name)
725 row = self._fetch_by_name_lower(s, norm)
726 if row is not None: 726 ↛ 727line 726 didn't jump to line 727 because the condition on line 726 was never true
727 return row
728 # Try with/without "the " prefix (~5K journals have it)
729 if norm.startswith("the "): 729 ↛ 730line 729 didn't jump to line 730 because the condition on line 729 was never true
730 row = self._fetch_by_name_lower(s, norm[4:])
731 else:
732 row = self._fetch_by_name_lower(s, "the " + norm)
733 if row is not None: 733 ↛ 734line 733 didn't jump to line 734 because the condition on line 733 was never true
734 return row
735 # Strip "proceedings of (the) (conference on) " prefix
736 stripped = norm
737 for prefix in (
738 "proceedings of the conference on ",
739 "proceedings of the ",
740 "proceedings of ",
741 ):
742 if stripped.startswith(prefix): 742 ↛ 743line 742 didn't jump to line 743 because the condition on line 742 was never true
743 stripped = stripped[len(prefix) :]
744 break
745 if stripped != norm: 745 ↛ 746line 745 didn't jump to line 746 because the condition on line 745 was never true
746 row = self._fetch_by_name_lower(s, stripped)
747 if row is not None:
748 return row
750 # MEDLINE-style "Title : long subtitle" — try the segment
751 # before the colon. Catches PubMed names like
752 # "Molecular therapy : the journal of the American Society..."
753 # → "Molecular therapy"
754 if " : " in norm: 754 ↛ 755line 754 didn't jump to line 755 because the condition on line 754 was never true
755 head = norm.split(" : ", 1)[0].strip()
756 if head and head != norm:
757 row = self._fetch_by_name_lower(s, head)
758 if row is not None:
759 return row
761 # MEDLINE-style "Title. Section name" — try the segment
762 # before the first period. Catches PubMed names like
763 # "Molecular therapy. Methods and clinical development"
764 # but only when the head is meaningfully shorter (we don't
765 # want to match "Nat" from "Nat. Commun.").
766 if "." in norm: 766 ↛ 767line 766 didn't jump to line 767 because the condition on line 766 was never true
767 head = norm.split(".", 1)[0].strip()
768 if head and len(head) >= 6 and head != norm:
769 row = self._fetch_by_name_lower(s, head)
770 if row is not None:
771 return row
773 return None
775 @staticmethod
776 def _fetch_by_name_lower(s: Session, name_lower: str) -> Optional[Source]:
777 stmt = select(Source).where(Source.name_lower == name_lower).limit(1)
778 return s.scalars(stmt).first()
780 # --- dashboard queries ---
782 def get_summary(self) -> dict:
783 if not self.available:
784 return {
785 "total": 0,
786 "avg_quality": 0,
787 "avg_h_index": None,
788 "predatory_count": 0,
789 "doaj_count": 0,
790 "llm_count": 0,
791 }
793 with self.session() as s:
794 row = s.execute(
795 select(
796 func.count().label("total"),
797 func.round(func.avg(Source.quality), 1).label(
798 "avg_quality"
799 ),
800 func.round(func.avg(Source.h_index)).label("avg_h_index"),
801 func.sum(func.iif(Source.is_predatory, 1, 0)).label(
802 "predatory_count"
803 ),
804 func.sum(func.iif(Source.is_in_doaj, 1, 0)).label(
805 "doaj_count"
806 ),
807 func.sum(
808 func.iif(Source.score_source == "llm", 1, 0)
809 ).label("llm_count"),
810 )
811 ).first()
812 return dict(row._mapping) if row else {}
814 def get_quality_distribution(self) -> dict[str, int]:
815 if not self.available:
816 return {}
817 with self.session() as s:
818 rows = s.execute(
819 select(Source.quality, func.count().label("cnt"))
820 .where(Source.quality.is_not(None))
821 .group_by(Source.quality)
822 .order_by(Source.quality)
823 ).all()
824 return {str(q): c for q, c in rows}
826 def get_source_distribution(self) -> dict[str, int]:
827 if not self.available:
828 return {}
829 with self.session() as s:
830 rows = s.execute(
831 select(
832 func.coalesce(Source.score_source, "unknown").label("src"),
833 func.count().label("cnt"),
834 ).group_by(Source.score_source)
835 ).all()
836 return {row.src: row.cnt for row in rows}
838 def get_journals_page(
839 self,
840 *,
841 page: int = 1,
842 per_page: int = 50,
843 search: str = "",
844 tier: str = "",
845 score_source: str = "",
846 sort: str = "quality",
847 order: str = "desc",
848 ) -> tuple[list[dict], int]:
849 if not self.available: 849 ↛ 850line 849 didn't jump to line 850 because the condition on line 849 was never true
850 return [], 0
852 if sort not in _SORT_COLUMNS:
853 sort = "quality"
854 if order not in ("asc", "desc"):
855 order = "desc"
857 wheres: list = []
858 if search:
859 needle = escape_like(normalize_name(search)[:_MAX_SEARCH_LEN])
860 wheres.append(Source.name_lower.like(f"%{needle}%", escape="\\"))
861 if tier and tier in _TIER_RANGES: 861 ↛ 862line 861 didn't jump to line 862 because the condition on line 861 was never true
862 lo, hi = _TIER_RANGES[tier]
863 wheres.append(Source.quality.between(lo, hi))
864 if score_source:
865 wheres.append(Source.score_source == score_source)
867 sort_col = getattr(Source, sort)
868 order_clause = (
869 sort_col.desc().nulls_last()
870 if order == "desc"
871 else sort_col.asc().nulls_last()
872 )
874 offset = (max(1, page) - 1) * per_page
876 with self.session() as s:
877 total = (
878 s.scalar(
879 select(func.count()).select_from(Source).where(*wheres)
880 )
881 or 0
882 )
883 rows = s.scalars(
884 select(Source)
885 .where(*wheres)
886 .order_by(order_clause)
887 .limit(per_page)
888 .offset(offset)
889 ).all()
891 return [_source_to_dashboard_dict(r) for r in rows], total
893 def get_institutions_page(
894 self,
895 *,
896 page: int = 1,
897 per_page: int = 50,
898 search: str = "",
899 sort: str = "h_index",
900 order: str = "desc",
901 ) -> tuple[list[dict], int]:
902 if not self.available: 902 ↛ 903line 902 didn't jump to line 903 because the condition on line 902 was never true
903 return [], 0
905 # Defensive allowlist — matches the pattern in get_journals_page.
906 # The ternary below is already safe (non-"desc" falls through to
907 # .asc()), but the explicit check prevents future refactors from
908 # accidentally interpolating a tainted value into SQL.
909 if order not in ("asc", "desc"): 909 ↛ 910line 909 didn't jump to line 910 because the condition on line 909 was never true
910 order = "desc"
912 wheres = []
913 if search: 913 ↛ 919line 913 didn't jump to line 919 because the condition on line 913 was always true
914 needle = escape_like(normalize_name(search)[:_MAX_SEARCH_LEN])
915 wheres.append(
916 Institution.name_lower.like(f"%{needle}%", escape="\\")
917 )
919 sort_col = (
920 Institution.h_index if sort == "h_index" else Institution.name
921 )
922 order_clause = (
923 sort_col.desc().nulls_last()
924 if order == "desc"
925 else sort_col.asc().nulls_last()
926 )
928 offset = (max(1, page) - 1) * per_page
930 with self.session() as s:
931 total = (
932 s.scalar(
933 select(func.count()).select_from(Institution).where(*wheres)
934 )
935 or 0
936 )
937 rows = s.scalars(
938 select(Institution)
939 .where(*wheres)
940 .order_by(order_clause)
941 .limit(per_page)
942 .offset(offset)
943 ).all()
945 return [_institution_to_dashboard_dict(r) for r in rows], total
948# ---------------------------------------------------------------------------
949# Dict adapters — keep filter/dashboard call sites unchanged
950# ---------------------------------------------------------------------------
953def _source_to_lookup_dict(row: Source) -> dict:
954 """Convert a Source row to the dict shape `lookup_openalex` produces.
956 Includes `openalex_source_id` so dashboard / test code can chain
957 a follow-up `lookup_source(source_id=...)` call. Also exposes
958 ``quartile`` so the filter can store it on the per-user Journal row
959 and feed it into score derivation.
960 """
961 return {
962 "name": row.name,
963 "type": row.source_type,
964 "h_index": row.h_index,
965 "impact_factor": row.impact_factor,
966 "is_in_doaj": row.is_in_doaj,
967 "publisher": row.publisher,
968 "issn_l": row.issn,
969 "openalex_source_id": row.openalex_source_id,
970 "quartile": row.quartile,
971 }
974def _source_to_dashboard_dict(row: Source) -> dict:
975 return {
976 "name": row.name,
977 "quality": row.quality,
978 "quartile": row.quartile,
979 "cited_by_count": row.cited_by_count,
980 "h_index": row.h_index,
981 "impact_factor": (
982 round(row.impact_factor, 2) if row.impact_factor else None
983 ),
984 "is_in_doaj": bool(row.is_in_doaj),
985 "is_predatory": bool(row.is_predatory),
986 "predatory_source": row.predatory_source,
987 "score_source": row.score_source,
988 "source_type": row.source_type,
989 "publisher": row.publisher,
990 "issn": row.issn,
991 "openalex_source_id": row.openalex_source_id,
992 }
995def _institution_to_dict(row: Institution) -> dict:
996 """Public accessor shape for `lookup_institution`.
998 The on-disk JSON snapshot uses one-character keys (``n``, ``c``,
999 ``t``, …) purely for space efficiency — 200k institutions × seven
1000 long field names adds real bytes. Callers of the accessor don't
1001 care about on-disk layout, so here we return the full names to
1002 keep the public API legible and robust to future schema tweaks.
1003 """
1004 return {
1005 "name": row.name,
1006 "country": row.country,
1007 "type": row.type,
1008 "h_index": row.h_index,
1009 "impact_factor": row.impact_factor,
1010 "works_count": row.works_count,
1011 "cited_by_count": row.cited_by_count,
1012 "ror_id": row.ror_id,
1013 }
1016def _institution_to_dashboard_dict(row: Institution) -> dict:
1017 return {
1018 "openalex_id": row.openalex_id,
1019 "name": row.name,
1020 "ror_id": row.ror_id,
1021 "country": row.country,
1022 "type": row.type,
1023 "h_index": row.h_index,
1024 "impact_factor": row.impact_factor,
1025 "works_count": row.works_count,
1026 "cited_by_count": row.cited_by_count,
1027 }
1030# ---------------------------------------------------------------------------
1031# Module singleton
1032# ---------------------------------------------------------------------------
1035_db: Optional[JournalQualityDB] = None
1036_db_lock = threading.Lock()
1039def get_db() -> JournalQualityDB:
1040 """Get or create the singleton `JournalQualityDB`."""
1041 global _db
1042 if _db is None:
1043 with _db_lock:
1044 if _db is None: 1044 ↛ 1046line 1044 didn't jump to line 1046
1045 _db = JournalQualityDB()
1046 return _db
1049# Backwards-compat aliases used by web/routers/metrics.py and a couple of tests
1050get_journal_reference_db = get_db
1051JournalReferenceDB = JournalQualityDB
1054def reset_db() -> None:
1055 """Reset the cached engine after a build_db rebuild.
1057 Held under `_db_lock` so a concurrent `get_db()` call can't see a
1058 half-disposed singleton — without the lock, Thread B could pass
1059 the `if _db is None` check in `get_db()` while Thread A is still
1060 inside `_db.reset()`, then call `_ensure_engine()` which short-
1061 circuits on the still-set `_engine` and hands back a disposed
1062 pool. The lock makes the read-then-reset pair atomic with respect
1063 to `get_db()`'s lazy-init path.
1064 """
1065 global _db
1066 with _db_lock:
1067 if _db is not None:
1068 _db.reset()
1071# ---------------------------------------------------------------------------
1072# Build phase — the ONLY writer
1073# ---------------------------------------------------------------------------
1076def build_db(
1077 data_dir: Optional[Path] = None,
1078 output_path: Optional[Path] = None,
1079) -> None:
1080 """Compile `journal_quality.db` from the gzipped JSON sources.
1082 Opens a SHORT-LIVED writable engine, creates the schema, populates
1083 every table from the gz files, runs ANALYZE + VACUUM, closes the
1084 engine, then `chmod 0o444` the file.
1085 """
1086 start = time.time()
1088 if data_dir is None:
1089 from ..config.paths import get_journal_data_directory
1091 data_dir = get_journal_data_directory()
1092 if output_path is None:
1093 output_path = data_dir / DB_FILENAME
1095 logger.info(
1096 "Building journal quality reference DB (one-time, "
1097 "~30s, decompresses ~25 MB of bundled data)…"
1098 )
1100 # Sweep stale temp files from prior crashed builds so they don't
1101 # accumulate. Any .tmp-* older than 1h is assumed dead.
1102 _sweep_stale_tmp_files(output_path.parent, output_path.name)
1104 # Build into a unique temp path, then os.replace() atomically at
1105 # the end. A random suffix (not a fixed .tmp) lets concurrent
1106 # builders write to separate files instead of racing on the same
1107 # path — os.replace picks a winner atomically and neither corrupts
1108 # the live file.
1109 tmp_path = output_path.with_name(
1110 f"{output_path.name}.tmp-{os.getpid()}-{secrets.token_hex(4)}"
1111 )
1113 write_url = f"sqlite:///{tmp_path}"
1114 engine = create_engine(write_url, connect_args={"check_same_thread": False})
1116 try:
1117 # Pragmas for fast bulk insert. `journal_mode = OFF` plus
1118 # `synchronous = OFF` is deliberately unsafe for general use but
1119 # correct here because durability is guaranteed by the temp-file
1120 # + os.replace() pattern around this block: we write to a unique
1121 # `.tmp-PID-RAND` path, and on any crash mid-build the incomplete
1122 # temp file is orphaned (and swept by `_sweep_stale_tmp_files()`
1123 # on the next build). The live file is only ever moved into place
1124 # by the atomic `os.replace()` at the bottom of this function —
1125 # it never sees a partial write. Do NOT copy this pragma set
1126 # elsewhere without the same atomic rename discipline.
1127 with engine.connect() as conn:
1128 conn.exec_driver_sql("PRAGMA journal_mode = OFF")
1129 conn.exec_driver_sql("PRAGMA synchronous = OFF")
1130 conn.exec_driver_sql("PRAGMA cache_size = -64000")
1131 conn.exec_driver_sql("PRAGMA page_size = 4096")
1133 JournalQualityBase.metadata.create_all(engine)
1135 SessionWrite = sessionmaker(bind=engine)
1136 with SessionWrite() as session:
1137 sources = _load_openalex(data_dir)
1138 doaj_data = _load_doaj(data_dir)
1139 pred_data = _load_predatory(data_dir)
1140 institutions = _load_institutions(data_dir)
1141 abbreviations = _load_abbreviations(data_dir)
1143 _populate_predatory(session, pred_data)
1144 _populate_sources(session, sources, doaj_data, pred_data)
1145 _populate_institutions(session, institutions)
1146 _populate_abbreviations(session, abbreviations)
1147 session.commit()
1149 with engine.connect() as conn:
1150 conn.exec_driver_sql("ANALYZE")
1151 conn.exec_driver_sql("VACUUM")
1152 # Stamp schema version so _ensure_engine can detect drift
1153 # without depending on the external version.json.
1154 conn.exec_driver_sql(
1155 f"PRAGMA user_version = {JOURNAL_QUALITY_SCHEMA_VERSION}"
1156 )
1157 except Exception:
1158 engine.dispose()
1159 if tmp_path.exists():
1160 try:
1161 # bearer:disable python_lang_file_permissions
1162 os.chmod(tmp_path, 0o644)
1163 tmp_path.unlink()
1164 except OSError:
1165 logger.exception(f"Failed to clean up tmp DB at {tmp_path}")
1166 raise
1168 engine.dispose()
1170 # Atomically swap tmp into place. os.replace is atomic on POSIX and
1171 # overwrites an existing output_path if present.
1172 if output_path.exists():
1173 # Prior file is chmod 0444 from the previous build — relax it
1174 # so os.replace can overwrite. Best-effort: if chmod fails
1175 # (e.g. read-only mount), os.replace will raise and surface
1176 # the real problem. Log so the cause is visible.
1177 try:
1178 # bearer:disable python_lang_file_permissions
1179 os.chmod(output_path, 0o644)
1180 except OSError:
1181 logger.warning(
1182 f"Could not chmod 0644 on existing {output_path} before "
1183 f"os.replace; if the replace fails this is likely why"
1184 )
1185 os.replace(tmp_path, output_path)
1187 # OS-level read-only flag — third layer of write protection.
1188 # POSIX chmod is a no-op on Windows, so we also set the Windows
1189 # read-only file attribute via SetFileAttributesW. The pre-commit
1190 # hook check-journal-quality-readonly.py remains the primary
1191 # defense against accidental writable opens.
1192 # bearer:disable python_lang_file_permissions
1193 os.chmod(output_path, 0o444)
1194 if sys.platform == "win32":
1195 try:
1196 import ctypes
1198 # FILE_ATTRIBUTE_READONLY = 0x1
1199 ok = ctypes.windll.kernel32.SetFileAttributesW(
1200 str(output_path), 0x1
1201 )
1202 if not ok:
1203 logger.warning(
1204 f"SetFileAttributesW failed on {output_path.name}; "
1205 "readonly pre-commit hook is the sole defense."
1206 )
1207 except Exception:
1208 logger.warning(
1209 f"Could not set Windows readonly attribute on "
1210 f"{output_path.name}"
1211 )
1213 elapsed = time.time() - start
1214 size_mb = output_path.stat().st_size / (1024 * 1024)
1215 with closing(
1216 sqlite3.connect(f"file:{output_path}?mode=ro&immutable=1", uri=True)
1217 ) as _count_conn:
1218 source_count = _count_conn.execute(
1219 "SELECT COUNT(*) FROM sources"
1220 ).fetchone()[0]
1221 logger.info(
1222 f"Journal quality DB ready: {source_count} sources, "
1223 f"{size_mb:.1f} MB in {elapsed:.1f}s ({output_path.name}, chmod 0o444)"
1224 )
1226 reset_db()
1229def _sweep_stale_tmp_files(directory: Path, base_name: str) -> None:
1230 """Remove journal_quality.db.tmp-* files older than 1h.
1232 Per-file OSError (vanished between glob+stat, no permission, etc.)
1233 is logged at debug — the sweep is best-effort and shouldn't stop
1234 the build, but silent-pass on filesystem errors hides the cause of
1235 accumulating stale tmp files that would otherwise eat disk over
1236 time.
1237 """
1238 if not directory.exists():
1239 return
1240 cutoff = time.time() - 3600
1241 for tmp in directory.glob(f"{base_name}.tmp-*"):
1242 try:
1243 if tmp.stat().st_mtime < cutoff:
1244 tmp.unlink()
1245 logger.info(f"Swept stale temp build file: {tmp.name}")
1246 except OSError:
1247 logger.debug(f"Could not sweep stale tmp file {tmp.name}")
1250# ---------------------------------------------------------------------------
1251# Source-data loaders (used by build_db only)
1252# ---------------------------------------------------------------------------
1255def _load_openalex(data_dir: Path) -> dict:
1256 path = data_dir / "openalex_sources.json.gz"
1257 if not path.exists():
1258 raise FileNotFoundError(f"OpenAlex source file not found: {path}")
1259 with gzip.open(path, "rt", encoding="utf-8") as f:
1260 data = json.load(f)
1261 sources = data.get("s", data.get("sources", {}))
1262 logger.info(f"Loaded {len(sources)} OpenAlex sources")
1263 return dict(sources)
1266def _load_doaj(data_dir: Path) -> dict:
1267 path = data_dir / "doaj_journals.json"
1268 if not path.exists():
1269 logger.warning(f"{path} not found — DOAJ cross-ref will be skipped")
1270 return {}
1271 with open(path, encoding="utf-8") as f:
1272 data = json.load(f)
1273 journals = data.get("journals", {})
1274 logger.info(f"Loaded {len(journals)} DOAJ entries")
1275 return dict(journals)
1278def _load_predatory(data_dir: Path) -> dict:
1279 """Returns {journals: set, publishers: set, hijacked: set, long_pubs: list}."""
1280 path = data_dir / "predatory.json"
1281 if not path.exists():
1282 logger.warning(f"{path} not found — predatory check will be skipped")
1283 return {
1284 "journals": set(),
1285 "publishers": set(),
1286 "hijacked": set(),
1287 "long_pubs": [],
1288 }
1290 with open(path, encoding="utf-8") as f:
1291 data = json.load(f)
1293 journal_names = {
1294 normalize_name(e.get("name", ""))
1295 for e in data.get("journals", [])
1296 if e.get("name", "").strip()
1297 }
1298 publisher_names = {
1299 normalize_name(e.get("name", ""))
1300 for e in data.get("publishers", [])
1301 if e.get("name", "").strip()
1302 }
1303 hijacked_names = {
1304 normalize_name(e.get("hijacked_name", ""))
1305 for e in data.get("hijacked", [])
1306 if e.get("hijacked_name", "").strip()
1307 }
1308 long_pubs = [
1309 normalize_name(e.get("name", ""))
1310 for e in data.get("publishers", [])
1311 if len(e.get("name", "").strip()) >= 10
1312 ]
1313 logger.info(
1314 f"Loaded predatory: {len(journal_names)} journals, "
1315 f"{len(publisher_names)} publishers, "
1316 f"{len(hijacked_names)} hijacked"
1317 )
1318 return {
1319 "journals": journal_names,
1320 "publishers": publisher_names,
1321 "hijacked": hijacked_names,
1322 "long_pubs": long_pubs,
1323 }
1326def _load_institutions(data_dir: Path) -> dict:
1327 path = data_dir / "openalex_institutions.json.gz"
1328 if not path.exists():
1329 logger.warning(f"{path} not found — institution tier will be empty")
1330 return {}
1331 with gzip.open(path, "rt", encoding="utf-8") as f:
1332 data = json.load(f)
1333 institutions = data.get("i", {})
1334 logger.info(f"Loaded {len(institutions)} institutions")
1335 return dict(institutions)
1338def _load_abbreviations(data_dir: Path) -> dict:
1339 path = data_dir / "jabref_abbreviations.json.gz"
1340 if not path.exists():
1341 logger.warning(
1342 f"{path} not found — abbreviation expansion will be empty"
1343 )
1344 return {}
1345 with gzip.open(path, "rt", encoding="utf-8") as f:
1346 data = json.load(f)
1347 mappings = data.get("abbrev_to_full", {})
1348 logger.info(f"Loaded {len(mappings)} abbreviation mappings")
1349 return dict(mappings)
1352# ---------------------------------------------------------------------------
1353# Table populators
1354# ---------------------------------------------------------------------------
1357def _populate_predatory(session: Session, pred: dict) -> None:
1358 long_set = set(pred.get("long_pubs", []))
1360 journals = [{"name_lower": n} for n in pred.get("journals", set()) if n]
1361 if journals:
1362 session.bulk_insert_mappings(inspect(PredatoryJournal), journals)
1364 hijacked = [{"name_lower": n} for n in pred.get("hijacked", set()) if n]
1365 if hijacked:
1366 session.bulk_insert_mappings(inspect(PredatoryHijacked), hijacked)
1368 publishers = [
1369 {"name_lower": n, "is_long": n in long_set}
1370 for n in pred.get("publishers", set())
1371 if n
1372 ]
1373 if publishers:
1374 session.bulk_insert_mappings(inspect(PredatoryPublisher), publishers)
1376 logger.info(
1377 f"Inserted predatory tables: "
1378 f"{len(journals)} journals, "
1379 f"{len(publishers)} publishers, "
1380 f"{len(hijacked)} hijacked"
1381 )
1384def _populate_sources(
1385 session: Session,
1386 sources: dict,
1387 doaj_data: dict,
1388 pred: dict,
1389) -> None:
1390 """Build Source rows with cross-referenced DOAJ + predatory flags."""
1391 type_map = {"j": "journal", "c": "conference"}
1392 pred_journals = pred.get("journals", set())
1393 pred_publishers = pred.get("publishers", set())
1394 pred_hijacked = pred.get("hijacked", set())
1396 # Dedup key is (name_lower, issn or "") so journals with separate
1397 # print and electronic ISSNs in OpenAlex both survive instead of
1398 # collapsing onto one row.
1399 seen: dict[tuple[str, str], dict] = {}
1401 for source_id, compact in sources.items():
1402 name = (compact.get("n") or "").strip()
1403 if not name: 1403 ↛ 1404line 1403 didn't jump to line 1404 because the condition on line 1403 was never true
1404 continue
1406 name_lower = normalize_name(name)
1407 issn = normalize_issn(compact.get("i"))
1408 publisher = compact.get("p") or None
1409 h_index = compact.get("h")
1410 impact_factor = compact.get("if")
1411 cited_by_count = compact.get("cb")
1412 source_type = type_map.get(compact.get("t", ""), compact.get("t", ""))
1414 doaj_entry = doaj_data.get(issn) if issn else None
1415 is_in_doaj = doaj_entry is not None
1417 is_pred = name_lower in pred_journals
1418 pred_source = "stop-predatory-journals" if is_pred else None
1419 if not is_pred and publisher:
1420 pub_norm = normalize_name(publisher)
1421 if pub_norm in pred_publishers: 1421 ↛ 1422line 1421 didn't jump to line 1422 because the condition on line 1421 was never true
1422 is_pred = True
1423 pred_source = "stop-predatory-publishers"
1424 if not is_pred and name_lower in pred_hijacked:
1425 is_pred = True
1426 pred_source = "stop-predatory-hijacked"
1428 # Whitelist override
1429 if is_pred and ( 1429 ↛ 1432line 1429 didn't jump to line 1432 because the condition on line 1429 was never true
1430 is_in_doaj or (h_index or 0) > PREDATORY_WHITELIST_HINDEX
1431 ):
1432 is_pred = False
1433 pred_source = None
1435 quality = derive_quality_score(
1436 h_index=h_index,
1437 is_in_doaj=is_in_doaj,
1438 is_predatory=is_pred,
1439 source_type=source_type,
1440 )
1442 rec = {
1443 "name": name,
1444 "name_lower": name_lower,
1445 "issn": issn,
1446 "openalex_source_id": source_id,
1447 "source_type": source_type,
1448 "publisher": publisher,
1449 "h_index": h_index,
1450 "impact_factor": impact_factor,
1451 "cited_by_count": cited_by_count,
1452 "quartile": None, # filled in by the post-pass below
1453 "quality": quality,
1454 "is_in_doaj": is_in_doaj,
1455 "is_predatory": is_pred,
1456 "predatory_source": pred_source,
1457 "score_source": "openalex",
1458 }
1460 key = (name_lower, issn or "")
1461 prev = seen.get(key)
1462 if prev is None or (h_index or 0) > (prev.get("h_index") or 0): 1462 ↛ 1401line 1462 didn't jump to line 1401 because the condition on line 1462 was always true
1463 seen[key] = rec
1465 # Second pass: DOAJ-only journals (not in OpenAlex). Without this
1466 # we lose ~4-7K small open-access venues. Keyed by name_lower so
1467 # we don't double-insert anything OpenAlex already covered.
1468 openalex_names = {k[0] for k in seen.keys()}
1469 doaj_added = 0
1470 for issn, doaj_entry in doaj_data.items():
1471 name = (doaj_entry.get("name") or "").strip()
1472 if not name: 1472 ↛ 1473line 1472 didn't jump to line 1473 because the condition on line 1472 was never true
1473 continue
1474 name_lower = normalize_name(name)
1475 if name_lower in openalex_names:
1476 continue
1477 publisher = doaj_entry.get("publisher") or None
1478 quality = derive_quality_score(
1479 h_index=None,
1480 is_in_doaj=True,
1481 is_predatory=False,
1482 source_type="journal",
1483 )
1484 seen[(name_lower, issn or "")] = {
1485 "name": name,
1486 "name_lower": name_lower,
1487 "issn": issn,
1488 "openalex_source_id": None,
1489 "source_type": "journal",
1490 "publisher": publisher,
1491 "h_index": None,
1492 "impact_factor": None,
1493 "cited_by_count": None,
1494 "quartile": None,
1495 "quality": quality,
1496 "is_in_doaj": True,
1497 "is_predatory": False,
1498 "predatory_source": None,
1499 "score_source": "doaj",
1500 }
1501 openalex_names.add(name_lower)
1502 doaj_added += 1
1504 records = list(seen.values())
1506 # Derive quartile (Q1–Q4) from cited_by_count percentile within each
1507 # source_type. Field-specific quartiles would be more accurate but
1508 # require per-source topic data that 4–7×s the snapshot size, so we
1509 # use global per-type percentiles as a defensible approximation
1510 # given the license constraint that ruled out SJR.
1511 by_type: dict[str, list[dict]] = {}
1512 for r in records:
1513 if r.get("cited_by_count") is None:
1514 continue # NULL quartile for entries without citation data
1515 by_type.setdefault(r.get("source_type") or "", []).append(r)
1516 for type_records in by_type.values():
1517 type_records.sort(key=lambda r: r["cited_by_count"])
1518 n = len(type_records)
1519 if n == 0: 1519 ↛ 1520line 1519 didn't jump to line 1520 because the condition on line 1519 was never true
1520 continue
1521 for rank, r in enumerate(type_records):
1522 pct = rank / n # 0.0 = lowest, ~1.0 = highest
1523 if pct >= 0.75:
1524 r["quartile"] = "Q1"
1525 elif pct >= 0.50:
1526 r["quartile"] = "Q2"
1527 elif pct >= 0.25:
1528 r["quartile"] = "Q3"
1529 else:
1530 r["quartile"] = "Q4"
1532 # Re-derive quality now that quartile is available. The first-pass
1533 # `quality` values above were computed without quartile and are
1534 # therefore suboptimal — a Q1 journal without h-index data would
1535 # have scored `None` (fall-through) instead of 8. The runtime filter
1536 # code in journal_reputation_filter.py does pass quartile, so the
1537 # stored column should agree with the live score.
1538 for r in records:
1539 r["quality"] = derive_quality_score(
1540 h_index=r.get("h_index"),
1541 quartile=r.get("quartile"),
1542 is_in_doaj=r.get("is_in_doaj") or False,
1543 is_predatory=r.get("is_predatory") or False,
1544 source_type=r.get("source_type"),
1545 )
1547 logger.info(
1548 f"Inserting {len(records)} source records ({doaj_added} DOAJ-only)..."
1549 )
1550 for i in range(0, len(records), _BATCH_SIZE):
1551 session.bulk_insert_mappings(
1552 inspect(Source), records[i : i + _BATCH_SIZE]
1553 )
1556def _populate_institutions(session: Session, institutions: dict) -> None:
1557 records: list[dict] = []
1558 for inst_id, compact in institutions.items():
1559 name = (compact.get("n") or "").strip()
1560 if not name: 1560 ↛ 1561line 1560 didn't jump to line 1561 because the condition on line 1560 was never true
1561 continue
1562 records.append(
1563 {
1564 "openalex_id": inst_id,
1565 "name": name,
1566 "name_lower": normalize_name(name),
1567 "ror_id": compact.get("r"),
1568 "country": compact.get("c"),
1569 "type": compact.get("t"),
1570 "h_index": compact.get("h"),
1571 "impact_factor": compact.get("if"),
1572 "works_count": compact.get("w"),
1573 "cited_by_count": compact.get("cb"),
1574 }
1575 )
1576 logger.info(f"Inserting {len(records)} institution records...")
1577 for i in range(0, len(records), _BATCH_SIZE):
1578 session.bulk_insert_mappings(
1579 inspect(Institution), records[i : i + _BATCH_SIZE]
1580 )
1583def _populate_abbreviations(session: Session, mappings: dict) -> None:
1584 records: list[dict] = []
1585 seen: set[str] = set()
1586 for abbrev, full in mappings.items():
1587 norm = normalize_name(abbrev)
1588 if not norm or norm in seen:
1589 continue
1590 seen.add(norm)
1591 records.append({"abbrev_lower": norm, "full_name": full})
1592 logger.info(f"Inserting {len(records)} abbreviation records...")
1593 for i in range(0, len(records), _BATCH_SIZE):
1594 session.bulk_insert_mappings(
1595 inspect(Abbreviation), records[i : i + _BATCH_SIZE]
1596 )
1599# ---------------------------------------------------------------------------
1600# Backwards-compat shim for the old build_reference_db name
1601# ---------------------------------------------------------------------------
1604def build_reference_db(
1605 data_dir: Optional[Path] = None,
1606 output_path: Optional[Path] = None,
1607) -> None:
1608 """Deprecated alias for `build_db`."""
1609 build_db(data_dir=data_dir, output_path=output_path)