Coverage for src/local_deep_research/advanced_search_system/filters/journal_reputation_filter.py: 70%
455 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-20 01:24 +0000
1"""
2Journal reputation filter with tiered quality scoring.
4Scores journals 1-10 and filters academic search results by quality.
5Uses bundled bibliometric data for most journals; LLM analysis is an
6opt-in last resort. Predatory journals are auto-removed.
8Scoring tiers (tried in order, first match wins):
10 1. Predatory check — auto-removes blacklisted journals/publishers
11 (whitelist override prevents false positives)
12 2. OpenAlex — h-index, quartile, DOAJ from ~217K bundled sources;
13 preprint repos can be lifted via institution affiliations
14 3. DOAJ — quality floor (5) for listed OA journals
15 3.5 Institutions — author affiliation lookup when no venue matched
16 (capped at 6, never beats a real venue)
18 --- DB cache check (only for cached LLM results from previous runs) ---
20 3.6 LLM cleanup — LLM canonicalises the name, then retries Tier 2
21 (opt-in via ``enable_llm_scoring``)
22 4. LLM analysis — SearXNG web search + LLM scoring (opt-in, expensive);
23 disabled after 2 consecutive failures
24 Conference — name-pattern heuristic for unmatched conferences
26Unknown journals that no tier can score receive a low-confidence score (3).
27"""
29import re
30import threading
31import time
32import unicodedata
33from datetime import timedelta
34from typing import Any, Dict, List, Optional
36from langchain_core.language_models.chat_models import BaseChatModel
37from loguru import logger
38from sqlalchemy.orm import Session
40from ...utilities.type_utils import unwrap_setting
41from ...config.llm_config import get_llm
42from ...constants import VALID_QUALITY_SCORES
43from ...database.models import Journal
44from ...database.session_context import get_user_db_session
46# normalize_name applies NFKC + lower + strip — must match the
47# migration backfill/dedupe expressions in 0006_journal_quality_system.py
48# and the reference DB builder in journal_quality/db.py so name_lower
49# is single-valued across every writer.
50from ...journal_quality.db import get_db as get_journal_data_manager
51from ...journal_quality.scoring import normalize_name
52from ...security.egress.policy import PolicyDeniedError
53from ...security.log_sanitizer import strip_control_chars
54from ...utilities.llm_utils import get_model_identifier
55from ...utilities.resource_utils import safe_close
56from ...utilities.thread_context import get_search_context
57from ...web_search_engines.search_engine_factory import create_search_engine
58from .base_filter import BaseFilter
59from ...constants import DEFAULT_SEARCH_TOOL
62# Patterns that indicate a venue is a conference, not a journal.
63# Used as a fallback when DOI enrichment and OpenAlex lookup both miss.
64_CONFERENCE_PATTERNS = [
65 re.compile(r"\b(?:proceedings|proc\.)\b", re.I),
66 re.compile(r"\b(?:conference|conf\.)\b", re.I),
67 re.compile(r"\b(?:symposium|symp\.)\b", re.I),
68 re.compile(r"\bworkshop\b", re.I),
69 re.compile(
70 r"\b(?:ICML|NeurIPS|NIPS|AAAI|CVPR|ICLR|ACL|EMNLP|NAACL|ECCV|ICCV|ICSE|SIGMOD|VLDB|KDD|WWW|SIGIR|CIKM|WSDM|RecSys|ISCA|MICRO|ASPLOS|OSDI|SOSP|NSDI|USENIX)\b"
71 ),
72]
75def _is_likely_conference(name: str) -> bool:
76 """Detect if a venue name is likely a conference based on common patterns."""
77 return any(p.search(name) for p in _CONFERENCE_PATTERNS)
80def _sanitize_name(name: str) -> str:
81 """Sanitize a journal name for safe use in logs and LLM prompts.
83 Args:
84 name: Raw journal name string, potentially containing control
85 characters, excessive length, or quotes.
87 Returns:
88 Sanitized string safe for use in logs and LLM prompts.
89 """
90 # Strip control + format characters. Covers C0/C1 (log injection),
91 # bidi overrides (U+202A-E, U+2066-9), zero-width / joiner chars
92 # (U+200B-F, U+2060-4, U+FEFF), Arabic letter mark, and digit-shape
93 # controls — the comprehensive pattern audited in log_sanitizer,
94 # not the narrow C0/C1-only regex we used to have here.
95 name = strip_control_chars(name)
96 # Normalize Unicode (prevents lookalike bypasses)
97 name = unicodedata.normalize("NFKC", name)
98 # Limit length (prevents resource exhaustion in prompts)
99 if len(name) > 500: 99 ↛ 100line 99 didn't jump to line 100 because the condition on line 99 was never true
100 name = name[:500] + "..."
101 # Strip quotes that could break prompt structure
102 name = name.replace('"', "'")
103 return name.strip()
106def _format_affiliations(affiliations: list, max_n: int = 3) -> str:
107 """Render an affiliation list as a compact human-readable string for
108 log lines. Accepts the same shapes as ``score_from_affiliations``
109 (plain strings or dicts with a ``name`` key) and truncates after
110 ``max_n`` entries so a 20-author paper doesn't blow up the log.
111 """
112 if not affiliations: 112 ↛ 114line 112 didn't jump to line 114 because the condition on line 112 was always true
113 return "(none)"
114 names: list[str] = []
115 for aff in affiliations:
116 if isinstance(aff, str):
117 names.append(aff)
118 elif isinstance(aff, dict):
119 nm = aff.get("name") or aff.get("display_name")
120 if nm:
121 names.append(nm)
122 if not names:
123 return "(unknown)"
124 shown = names[:max_n]
125 suffix = "" if len(names) <= max_n else f" (+{len(names) - max_n} more)"
126 return ", ".join(shown) + suffix
129_bg_fetch_lock = threading.Lock()
130_bg_fetch_thread: Optional[threading.Thread] = None
133def _start_background_journal_fetch() -> None:
134 """Kick off ``ensure_journal_data(auto_download=True)`` in a daemon
135 thread the first time a search hits the pending path.
137 The worker returns immediately if another thread is already in
138 flight (``_bg_fetch_thread.is_alive()``) — so 30 concurrent filter
139 workers can't each spawn their own download. The ``ensure_journal_data``
140 TTL cache provides a second line of defence.
142 Daemon thread so it doesn't block process exit.
143 """
144 global _bg_fetch_thread
145 with _bg_fetch_lock:
146 if _bg_fetch_thread is not None and _bg_fetch_thread.is_alive():
147 logger.debug(
148 "journal-data background fetch already in flight — "
149 "not spawning a second thread"
150 )
151 return
153 def _run():
154 try:
155 # Late import so this module doesn't pay the cost of
156 # importing the downloader at its own import time.
157 from ...journal_quality.downloader import (
158 ensure_journal_data,
159 )
161 logger.info(
162 "journal-data background fetch: starting "
163 "(triggered by filter pending path)"
164 )
165 ensure_journal_data(auto_download=True)
166 logger.info("journal-data background fetch: done")
167 except Exception:
168 logger.exception(
169 "journal-data background fetch crashed — "
170 "next search will retry"
171 )
173 _bg_fetch_thread = threading.Thread(
174 target=_run,
175 name="journal-data-bg-fetch",
176 daemon=True,
177 )
178 _bg_fetch_thread.start()
181class JournalFilterError(Exception):
182 """
183 Custom exception for errors related to journal filtering.
184 """
187class JournalReputationFilter(BaseFilter):
188 """
189 A filter for academic results that considers the reputation of journals.
191 Uses a tiered scoring approach: bundled data (OpenAlex, DOAJ, predatory
192 lists) for most journals, with LLM-based analysis via SearXNG as a
193 fallback for truly unknown journals.
195 Predatory journals are **automatically removed** from results.
196 """
198 def __init__(
199 self,
200 model: BaseChatModel | None = None,
201 reliability_threshold: int | None = None,
202 max_context: int | None = None,
203 exclude_non_published: bool | None = None,
204 quality_reanalysis_period: timedelta | None = None,
205 settings_snapshot: Dict[str, Any] | None = None,
206 ):
207 """Initialize the journal reputation filter.
209 Args:
210 model: The LLM model to use for Tier 4 analysis. If None,
211 the default LLM from settings will be used.
212 reliability_threshold: Minimum quality score (1-10) for a
213 result to pass. Read from settings if not specified.
214 max_context: Maximum characters of source content for LLM
215 quality evaluation.
216 exclude_non_published: If True, exclude results that don't
217 have an associated journal publication reference.
218 quality_reanalysis_period: Period after which cached journal
219 quality assessments are refreshed.
220 settings_snapshot: Settings snapshot for thread context.
221 """
222 super().__init__(model)
224 self._owns_llm = self.model is None
225 if self.model is None:
226 # Forward the snapshot so the LLM PEP at llm_config.py:295
227 # can evaluate ``llm.require_local_endpoint`` / egress scope.
228 # Dropping it here previously bypassed the PEP entirely on
229 # search engines that constructed this filter with no llm.
230 self.model = get_llm(settings_snapshot=settings_snapshot)
232 # Import here to avoid circular import
233 from ...config.search_config import get_setting_from_snapshot
235 self.__threshold = reliability_threshold
236 if self.__threshold is None:
237 self.__threshold = int(
238 get_setting_from_snapshot(
239 "search.journal_reputation.threshold",
240 2,
241 settings_snapshot=settings_snapshot,
242 )
243 )
244 self.__max_context = max_context
245 if self.__max_context is None:
246 self.__max_context = int(
247 get_setting_from_snapshot(
248 "search.journal_reputation.max_context",
249 3000,
250 settings_snapshot=settings_snapshot,
251 )
252 )
253 self.__exclude_non_published = exclude_non_published
254 if self.__exclude_non_published is None:
255 self.__exclude_non_published = bool(
256 get_setting_from_snapshot(
257 "search.journal_reputation.exclude_non_published",
258 False,
259 settings_snapshot=settings_snapshot,
260 )
261 )
262 self.__quality_reanalysis_period = quality_reanalysis_period
263 if self.__quality_reanalysis_period is None:
264 self.__quality_reanalysis_period = timedelta(
265 days=int(
266 get_setting_from_snapshot(
267 "search.journal_reputation.reanalysis_period",
268 365,
269 settings_snapshot=settings_snapshot,
270 )
271 )
272 )
274 self.__settings_snapshot = settings_snapshot
276 # SearXNG for Tier 4 (LLM fallback). Not strictly required anymore
277 # since bundled data covers most journals.
278 self.__engine = create_search_engine(
279 "searxng", llm=self.model, settings_snapshot=settings_snapshot
280 )
281 self.__searxng_available = self.__engine is not None and getattr(
282 self.__engine, "is_available", False
283 )
284 if not self.__searxng_available:
285 logger.info(
286 "SearXNG not available — Tier 4 (LLM analysis) disabled. "
287 "Bundled data tiers still active."
288 )
290 # Fail-fast counter for SearXNG failures within a batch.
291 # Stored in `threading.local()` so concurrent `filter_results`
292 # calls on the same cached filter instance (the parallel search
293 # engine reuses instances across worker threads — see
294 # concurrent engine threads) can't clobber each other's
295 # counter. Per-thread state is reset at the top of every
296 # `filter_results` invocation.
297 self.__tls = threading.local()
299 # Lock serializing access to the shared SearXNG engine for
300 # Tier 4. BaseSearchEngine instances keep mutable bookkeeping
301 # state (_last_results_count, _search_results, rate-limit
302 # tracker) on self; two concurrent .run() calls on the same
303 # instance would clobber that state. Tier 4 is rarely hit in
304 # practice (requires enable_llm_scoring=True + SearXNG +
305 # bundled-data miss), so the lock's contention cost is
306 # negligible compared to the correctness guarantee.
307 self.__engine_lock = threading.Lock()
309 # Journal data manager (loads bundled datasets lazily)
310 self.__data_manager = get_journal_data_manager()
312 # ------------------------------------------------------------------
313 # Thread-local fail-fast counter accessors
314 # ------------------------------------------------------------------
316 def __searxng_failures(self) -> int:
317 return getattr(self.__tls, "searxng_failures", 0)
319 def __reset_searxng_failures(self) -> None:
320 self.__tls.searxng_failures = 0
322 def __bump_searxng_failures(self) -> int:
323 n = self.__searxng_failures() + 1
324 self.__tls.searxng_failures = n
325 return n
327 def close(self) -> None:
328 """Close the SearXNG engine and LLM client."""
329 if hasattr(self, "_JournalReputationFilter__engine"): 329 ↛ 333line 329 didn't jump to line 333 because the condition on line 329 was always true
330 # allow_none=True: SearXNG is optional (Tier 4 only), so
331 # __engine is None in the common no-SearXNG configuration.
332 safe_close(self.__engine, "SearXNG engine", allow_none=True)
333 if self._owns_llm: 333 ↛ 334line 333 didn't jump to line 334 because the condition on line 333 was never true
334 safe_close(self.model, "journal filter LLM")
336 def _should_skip_journal_fetch_for_scope(self) -> bool:
337 """Return True when the active egress scope forbids public
338 fetches (PRIVATE_ONLY / STRICT). The journal-data sources
339 (OpenAlex, DOAJ, JabRef) are all public, so under these scopes
340 the daemon thread would burn cycles and (worse) attempt egress
341 the user explicitly opted out of.
342 """
343 snapshot = self.__settings_snapshot
344 if not snapshot:
345 return False
346 try:
347 from ...security.egress.policy import (
348 EgressScope,
349 PolicyDeniedError,
350 context_from_snapshot,
351 )
353 primary_raw = unwrap_setting(
354 snapshot.get("search.tool", DEFAULT_SEARCH_TOOL)
355 )
356 ctx = context_from_snapshot(
357 snapshot, primary_raw or DEFAULT_SEARCH_TOOL
358 )
359 return ctx.scope in (EgressScope.PRIVATE_ONLY, EgressScope.STRICT)
360 except PolicyDeniedError:
361 # Corrupt/unknown scope value — we cannot certify that a public
362 # journal fetch is permitted, so SKIP the fetch (fail closed).
363 # This matches the hardened sibling
364 # notifications.manager._filter_urls_by_egress_policy, which also
365 # refuses rather than proceeds when the policy is unevaluable.
366 logger.bind(policy_audit=True).warning(
367 "journal fetch skipped: egress policy unevaluable "
368 "(corrupt scope) — failing closed"
369 )
370 return True
371 except Exception:
372 # Other errors (missing key, snapshot shape) — fail open; the
373 # per-URL runtime check still gates each individual fetch.
374 return False
376 @classmethod
377 def create_default(
378 cls,
379 model: BaseChatModel | None = None,
380 *,
381 engine_name: str,
382 settings_snapshot: Dict[str, Any] | None = None,
383 ) -> Optional["JournalReputationFilter"]:
384 """Initializes a default configuration of the filter based on settings.
386 SearXNG is not required — the filter works with bundled data alone.
387 SearXNG enables the optional Tier 4 (LLM analysis) for journals
388 not found in the bundled datasets.
390 Args:
391 model: Optional LLM model for Tier 4 analysis.
392 engine_name: Search engine configuration key (e.g. "arxiv").
393 settings_snapshot: Optional frozen settings dict.
395 Returns:
396 A configured JournalReputationFilter, or None if filtering
397 is disabled in settings for this engine.
398 """
399 from ...config.search_config import get_setting_from_snapshot
401 try:
402 enabled = get_setting_from_snapshot(
403 f"search.engine.web.{engine_name}.journal_reputation.enabled",
404 True,
405 settings_snapshot=settings_snapshot,
406 )
407 logger.info(
408 f"Journal filter create_default: engine={engine_name}, "
409 f"enabled={enabled} (type={type(enabled).__name__})"
410 )
411 if not bool(enabled):
412 logger.info(
413 f"Journal filter disabled for {engine_name} in settings"
414 )
415 return None
417 filt = JournalReputationFilter(
418 model=model, settings_snapshot=settings_snapshot
419 )
420 logger.info(
421 f"Journal filter created for {engine_name} — "
422 f"threshold={filt._JournalReputationFilter__threshold}"
423 )
424 return filt
425 except PolicyDeniedError:
426 # The egress policy refused the LLM the filter needs. This
427 # must propagate so the user sees a clear policy refusal —
428 # silently returning None would let unfiltered results
429 # through under ``llm.require_local_endpoint=True``.
430 raise
431 except Exception:
432 # Any other failure — settings read, filter init — returns
433 # None rather than silently defaulting to enabled. A separate
434 # silent ``except Exception: enabled = True`` branch used to
435 # wrap only the settings read, which hid legitimate
436 # configuration errors; per CLAUDE.md, fallbacks have to be
437 # explicit, not default-on.
438 logger.exception(
439 f"Failed to configure journal filter for {engine_name}; "
440 "results will not be journal-quality filtered for this engine"
441 )
442 return None
444 @staticmethod
445 def __db_session() -> Session | None:
446 """Get a database session using the current search context credentials.
448 Returns:
449 SQLAlchemy Session context manager for the user's database, or
450 ``None`` if no search context is available (e.g. when called
451 from the preview filter phase, before per-user thread context
452 has been propagated). Callers should treat ``None`` as
453 "skip the DB operation".
454 """
455 context = get_search_context()
456 if context is None:
457 return None
458 username = context.get("username")
459 password = context.get("user_password")
460 return get_user_db_session(username=username, password=password)
462 # ------------------------------------------------------------------
463 # Journal name cleaning (with LRU cache to avoid duplicate LLM calls)
464 # ------------------------------------------------------------------
466 def __clean_journal_name(self, journal_name: str) -> str:
467 """Clean journal name to normalize for deduplication and lookup.
469 Deterministic regex-based cleaning only: strips volume / page /
470 year / month references, then tries a JabRef abbreviation
471 expansion. This method does NOT call the LLM — it is the cheap,
472 instant cleaning path that every Tier runs through first.
474 The separate ``__llm_clean_journal_name`` method is the LLM
475 fallback, invoked later only as a salvage step when the bundled
476 data tiers all miss and ``enable_llm_scoring`` is on.
477 Abbreviations not in the JabRef list and location suffixes
478 ("ICML 2023, Honolulu") only get LLM-cleaned on that salvage
479 path; this method returns them unchanged.
481 Args:
482 journal_name: Raw journal name from search results.
484 Returns:
485 Cleaned, normalized journal name.
486 """
487 # Sanitize first (strips control chars, normalizes Unicode)
488 journal_name = _sanitize_name(journal_name)
489 # Regex handles volume/page/year stripping (instant)
490 cleaned = self.__regex_clean_journal_name(journal_name)
492 # Try JabRef abbreviation expansion (deterministic, instant)
493 expanded = self.__data_manager.expand_abbreviation(cleaned)
494 if expanded:
495 logger.debug(f"Abbreviation expanded: '{cleaned}' → '{expanded}'")
496 return expanded
498 if cleaned != journal_name:
499 logger.debug(
500 f"Regex-cleaned journal name: '{journal_name}' → '{cleaned}'"
501 )
502 return cleaned
504 @staticmethod
505 def __regex_clean_journal_name(name: str) -> str:
506 """
507 Fast regex-based journal name normalization. Strips volume, issue,
508 page, year, and month references. No LLM needed.
509 """
510 months = (
511 "january|february|march|april|may|june|july|"
512 "august|september|october|november|december|"
513 "jan|feb|mar|apr|jun|jul|aug|sep|oct|nov|dec"
514 )
516 # Strip leading/trailing whitespace
517 name = name.strip()
519 # Strip a leading [bracketed-original-language] prefix that
520 # MEDLINE uses for non-English journals:
521 # "[Rinsho ketsueki] The Japanese journal of clinical hematology"
522 # → "The Japanese journal of clinical hematology"
523 name = re.sub(r"^\[[^\]]+\]\s*", "", name)
525 # Strip trailing publisher suffixes that some search engines glue
526 # onto the journal name (e.g. "Information Fusion Elsevier" or
527 # "Cell Press"). Conservative — only the handful of well-known
528 # academic publishers, anchored at end of string with a leading
529 # space so we don't eat them mid-name.
530 name = re.sub(
531 r"\s+(?:Elsevier|Springer|Wiley|Nature\s+Publishing|"
532 r"Cell\s+Press|MDPI|Sage|Taylor\s*&?\s*Francis|"
533 r"Oxford\s+University\s+Press|Cambridge\s+University\s+Press|"
534 r"IEEE|ACM|Routledge|Frontiers)\s*$",
535 "",
536 name,
537 flags=re.IGNORECASE,
538 )
540 # Strip a leading 4-digit year ("2015 Plasma Phys. ..." → "Plasma Phys. ...")
541 name = re.sub(r"^(?:19|20)\d{2}\s+", "", name)
543 # Strip a leading ordinal volume marker, e.g.
544 # "31st Conference on Neural Information Processing Systems" → "Conference on …"
545 # Without this the OpenAlex name lookup fails on most conference
546 # entries because the canonical name has no ordinal prefix.
547 name = re.sub(
548 r"^\d+(?:st|nd|rd|th)\s+",
549 "",
550 name,
551 flags=re.IGNORECASE,
552 )
554 # Remove month+year references FIRST (before bare year strip),
555 # so "September 2023" is consumed as a unit and we don't leave
556 # the month behind as an orphan word.
557 name = re.sub(
558 rf",?\s*\b(?:{months})\b\.?\s+(?:19|20)?\d{{2,4}}",
559 "",
560 name,
561 flags=re.IGNORECASE,
562 )
564 # Remove volume/issue/page refs: "Vol. 12", "Issue 3", "pp. 100-200"
565 name = re.sub(
566 r",?\s*(?:vol(?:ume)?\.?\s*\d+|"
567 r"issue\s*\d+|"
568 r"no\.?\s*\d+|"
569 r"pp?\.?\s*\d+[\s–-]*\d*|"
570 r"pages?\s*\d+[\s–-]*\d*)",
571 "",
572 name,
573 flags=re.IGNORECASE,
574 )
575 # Remove volume(issue) patterns: "141(5)" — and bare "(15)" issues.
576 name = re.sub(r",?\s*\d+\(\d+\)", "", name)
577 name = re.sub(r"\s*\(\d+\)\s*", " ", name)
578 # Remove year references: "(2023)", ", 2023". Anchored to 19xx/20xx
579 # so 4-digit page numbers like ", 1063" in "106335" aren't eaten.
580 name = re.sub(r"\s*[\(,]\s*(?:19|20)\d{2}\b\s*\)?", "", name)
581 # Remove bare trailing citation data: ", 95, 146802" style
582 # Only strips when there's a comma before the first number
583 # (preserves "NeurIPS 2023" where space-number is part of the name)
584 name = re.sub(r",\s*\d[\d,\s]*$", "", name)
585 # Strip trailing alphanumeric volume markers: "E48", "R569", "L102"
586 # (single uppercase letter followed by digits at end of string).
587 name = re.sub(r"\s+[A-Z]\d+\s*$", "", name)
588 # Strip trailing volume/page debris like "170 266-275", "71: 1-10",
589 # "151:48-60", or a bare trailing volume "116". Repeated to peel
590 # multiple chunks. Stops when only the journal name remains. We
591 # require the run to start with whitespace or punctuation so we
592 # don't eat the "2023" of "NeurIPS 2023" — the trailing-year regex
593 # below handles that case.
594 prev = None
595 while prev != name:
596 prev = name
597 name = re.sub(
598 r"[\s,;:]+\d+(?:\s*[:\-–]\s*\d+)?(?:\s*[-–]\s*\d+)?\s*$",
599 "",
600 name,
601 )
602 # Strip leftover trailing month name (no year) — happens when the
603 # year was stripped by another regex first.
604 name = re.sub(
605 rf",?\s*\b(?:{months})\b\.?\s*$",
606 "",
607 name,
608 flags=re.IGNORECASE,
609 )
610 # Strip leftover bare volume/page keyword at the end ("p", "pp",
611 # "vol", "vol.", "no", "no.") that survives when the number got
612 # truncated upstream by the search engine result preview.
613 name = re.sub(
614 r",?\s*\b(?:vol(?:ume)?|pp?|no)\b\.?\s*$",
615 "",
616 name,
617 flags=re.IGNORECASE,
618 )
619 # Strip empty / whitespace-only trailing parens — arXiv
620 # journal_ref fields sometimes end with "()" where the citation
621 # year was stripped upstream, e.g. "Physical Review Research ()".
622 name = re.sub(r"\s*\(\s*\)\s*$", "", name)
623 # Strip geographic qualifiers: "(London)", "(New York)", "(US)"
624 # Only strip parenthesized suffixes that contain no digits
625 # (preserves "NeurIPS (2023)" which is handled by the year regex)
626 name = re.sub(r"\s*\([^()0-9]+\)\s*$", "", name)
627 # Strip trailing truncated volume/page markers: ", v", ", p",
628 # ", vol" — these appear when the search engine preview cut the
629 # citation mid-keyword ("Plasma Physics and Controlled Fusion,
630 # vol. 63, no. 8" → "Plasma Physics and Controlled Fusion, v").
631 name = re.sub(
632 r",\s*(?:v|vol|p|pp|no|n)\.?\s*$",
633 "",
634 name,
635 flags=re.IGNORECASE,
636 )
637 # Remove trailing punctuation and whitespace
638 name = re.sub(r"[,;.\s]+$", "", name)
639 # Strip a trailing 4-digit year/volume (conferences: "NeurIPS 2023"
640 # → "NeurIPS"). Comes after the punctuation strip so any trailing
641 # comma/period is already gone, and after the parenthesized-year
642 # regex above so we don't double-process "(2023)".
643 name = re.sub(r"\s+\d{4}\s*$", "", name)
644 # Normalize "&" → "and" for consistent matching
645 name = re.sub(r"\s*&\s*", " and ", name)
646 # Normalize internal whitespace
647 return re.sub(r"\s+", " ", name).strip()
649 def __llm_clean_journal_name(self, journal_name: str) -> Optional[str]:
650 """LLM-based fallback for canonicalizing unusual journal names.
652 The regex + JabRef abbreviation tiers handle the common cases
653 (volume/year/page stripping, well-known abbreviations like
654 "Phys. Rev. Lett." → "Physical Review Letters"). They cannot
655 handle locations ("ICML 2023, Honolulu"), unusual abbreviations
656 not in the JabRef list, or non-English title transliterations.
658 This is gated behind ``enable_llm_scoring`` so it never fires
659 unless the user opted into the Tier 4 LLM path. Called only as a
660 salvage step when bundled tiers all miss, so the LLM bill is
661 bounded by the number of *unrecognised* journals per query, not
662 every journal.
664 Args:
665 journal_name: A name that the regex tier could not match
666 against any bundled dataset.
668 Returns:
669 A canonicalised name from the LLM, or ``None`` if the call
670 failed or the response was empty.
671 """
672 prompt = (
673 f"Clean up the following journal or conference name:\n\n"
674 f'"{journal_name}"\n\n'
675 "Remove any references to volumes, pages, months, or years. "
676 "Expand common abbreviations. For conferences, remove "
677 "locations. Output only the clean name, no explanation."
678 )
679 try:
680 response = self.model.invoke(prompt)
681 content = getattr(response, "content", None) or response
682 cleaned = str(content).strip().strip('"').strip("'")
683 if not cleaned: 683 ↛ 684line 683 didn't jump to line 684 because the condition on line 683 was never true
684 return None
685 return cleaned
686 except (
687 ConnectionError,
688 TimeoutError,
689 ValueError,
690 ) as e:
691 # Network / service / parse failures are expected and
692 # recoverable — caller falls back to the regex-cleaned name.
693 # Surface at WARNING (not silent / not DEBUG) so they're
694 # visible during triage without flooding info-level logs.
695 # Log only the exception class name; the message can carry
696 # request-specific data (URLs, prompts) that doesn't belong
697 # in operational logs.
698 logger.warning(
699 f"LLM name cleaning failed for '{journal_name}' "
700 f"({type(e).__name__}); using regex-cleaned version"
701 )
702 return None
704 # ------------------------------------------------------------------
705 # Tier 4: LLM-based analysis (last resort)
706 # ------------------------------------------------------------------
708 def __analyze_journal_reputation(self, journal_name: str) -> int:
709 """Analyze journal reputation via 1 SearXNG search + 1 LLM call.
711 This is Tier 4 — the last-resort scoring path. Only used when
712 the journal is not found in bundled data (OpenAlex, DOAJ, predatory).
713 Uses a single web search for context, then a single LLM call to score.
715 Args:
716 journal_name: Cleaned journal name to research.
718 Returns:
719 Reputation score between 1 and 10.
721 Raises:
722 ValueError: If the LLM response cannot be parsed as a score.
723 """
724 logger.info(f"Tier 4: LLM analysis for journal '{journal_name}'...")
726 # Single SearXNG search for journal info. Serialize access to
727 # the shared SearXNG engine to prevent two threads from
728 # clobbering its instance state (_last_results_count,
729 # _search_results, rate tracker).
730 query = f'"{journal_name}" academic journal impact factor quartile'
731 with self.__engine_lock:
732 results = self.__engine.run(query)
734 # Extract snippets from search results
735 snippets = []
736 for r in results[:10]:
737 snippet = r.get("snippet", "") or r.get("content", "")
738 if snippet: 738 ↛ 736line 738 didn't jump to line 736 because the condition on line 738 was always true
739 snippets.append(snippet)
740 journal_info_text = "\n".join(snippets)
742 if not journal_info_text: 742 ↛ 743line 742 didn't jump to line 743 because the condition on line 742 was never true
743 logger.warning(
744 f"No SearXNG results for '{journal_name}' — "
745 f"cannot score via Tier 4"
746 )
747 raise ValueError(f"No search results for journal '{journal_name}'")
749 # Truncate to fit context
750 if len(journal_info_text) > self.__max_context: 750 ↛ 751line 750 didn't jump to line 751 because the condition on line 750 was never true
751 journal_info_text = journal_info_text[: self.__max_context] + "..."
753 # Single LLM call to score. Wording mirrors the long-standing
754 # original prompt — earlier code review flagged that arbitrary
755 # rewrites of this prompt have a real chance of regressing the
756 # Q1/Q2/Q3 calibration the rest of the code depends on.
757 prompt = f"""
758You are a research assistant helping to assess the reliability and
759reputability of scientific journals. A reputable journal should be
760peer-reviewed, not predatory, and high-impact. Please review the
761following information on the journal "{journal_name}" and output a
762reputability score between 1 and 10, where 1-3 is not reputable and
763probably predatory, 4-6 is reputable but low-impact (Q2 or Q3),
764and 7-10 is reputable Q1 journals. Only output the number, do not
765provide any explanation or other output.
767JOURNAL INFORMATION:
769{journal_info_text}
770"""
772 response = self.model.invoke(prompt).content
773 logger.debug(f"Tier 4 LLM response for '{journal_name}': {response}")
775 match = re.search(r"\d+", response.strip())
776 if match is None:
777 logger.warning(
778 f"Failed to parse score from LLM response for "
779 f"'{journal_name}': {response!r}"
780 )
781 raise ValueError(
782 "Failed to parse reputation score from LLM response."
783 )
785 reputation_score = int(match.group())
786 if reputation_score not in VALID_QUALITY_SCORES:
787 # Scoring tiers emit {1,4,5,6,7,8,10}; LLM returning anything
788 # else is almost certainly prompt drift. Treat as a parse
789 # failure so the existing failure counter + circuit breaker
790 # observe it, rather than snapping to the nearest bucket and
791 # silently masking the problem.
792 logger.warning(
793 f"LLM returned out-of-set score {reputation_score} for "
794 f"'{journal_name}' (expected one of "
795 f"{sorted(VALID_QUALITY_SCORES)}); treating as parse failure"
796 )
797 raise ValueError(
798 f"LLM returned out-of-set score {reputation_score}."
799 )
800 return reputation_score
802 # ------------------------------------------------------------------
803 # Database operations
804 # ------------------------------------------------------------------
806 def __save_llm_score_to_db(self, *, name: str, quality: int) -> None:
807 """Cache a Tier 4 LLM score for future research runs.
809 Only Tier 4 (LLM) results are cached — Tiers 1–3 read directly
810 from the read-only reference DB on every scoring pass. The
811 lookup predicate filters on ``score_source == "llm"`` and the
812 current ``quality_model`` so stale scores from a superseded LLM
813 don't get served.
815 No-op during the preview filter phase (no thread context).
816 """
817 session_ctx = self.__db_session()
818 if session_ctx is None: 818 ↛ 820line 818 didn't jump to line 820 because the condition on line 818 was always true
819 return
820 try:
821 self._save_journal_to_db_inner(
822 session_ctx, name=name, quality=quality
823 )
824 except Exception:
825 # Score is still valid and returned to the caller — only
826 # the cache write failed.
827 logger.exception(
828 f"Failed to cache LLM score for '{name}' — "
829 f"score is still returned but won't be cached."
830 )
832 def _save_journal_to_db_inner(
833 self, session_ctx, *, name: str, quality: int
834 ) -> None:
835 """Race-safe upsert of the Tier 4 LLM cache row.
837 Mirrors the Paper upsert pattern: select-then-insert in a
838 savepoint, and on IntegrityError (a concurrent writer created
839 the row first) roll back the savepoint and re-fetch. This
840 prevents the pre-fix bug where two concurrent scorings of the
841 same journal collided on the UNIQUE(name) constraint and the
842 exception handler left the cache empty.
843 """
844 from sqlalchemy.exc import IntegrityError
846 now = int(time.time())
847 model_id = get_model_identifier(self.model)
849 with session_ctx as db_session:
850 journal = db_session.query(Journal).filter_by(name=name).first()
851 if journal is not None:
852 journal.quality = quality
853 journal.score_source = "llm"
854 journal.quality_analysis_time = now
855 journal.quality_model = model_id
856 # name_lower MUST go through normalize_name, not bare
857 # .lower(). Bare lowercase leaves U+2122 (TM), ligatures,
858 # fullwidth letters intact; NFKC collapses them. The
859 # migration backfill (0006:257) uses the same normalization;
860 # divergence produces silent cache misses and, on migration,
861 # UNIQUE violations that abort the upgrade.
862 journal.name_lower = normalize_name(name)
863 try:
864 db_session.commit()
865 except Exception:
866 db_session.rollback()
867 logger.warning(
868 f"Failed to update cached LLM score for '{name}'"
869 )
870 return
872 sp = db_session.begin_nested()
873 try:
874 journal = Journal(
875 name=name,
876 name_lower=normalize_name(name),
877 quality=quality,
878 score_source="llm",
879 quality_model=model_id,
880 quality_analysis_time=now,
881 )
882 db_session.add(journal)
883 db_session.flush()
884 sp.commit()
885 db_session.commit()
886 return
887 except IntegrityError:
888 sp.rollback()
890 # Competing writer inserted first; re-fetch and update.
891 journal = db_session.query(Journal).filter_by(name=name).first()
892 if journal is None:
893 # Genuinely unexpected — UNIQUE violation with no row.
894 logger.warning(
895 f"IntegrityError on Journal '{name}' insert but "
896 f"row not found on re-fetch; skipping cache write."
897 )
898 return
899 journal.quality = quality
900 journal.score_source = "llm"
901 journal.quality_analysis_time = now
902 journal.quality_model = model_id
903 journal.name_lower = normalize_name(name)
904 try:
905 db_session.commit()
906 except Exception:
907 db_session.rollback()
908 logger.warning(f"Failed to merge cached LLM score for '{name}'")
910 # ------------------------------------------------------------------
911 # Tiered scoring for a single journal
912 # ------------------------------------------------------------------
914 def __score_journal(
915 self, journal_name: str, result: Dict[str, Any]
916 ) -> tuple[int | None, str | None]:
917 """
918 Score a journal using the tiered approach.
920 Returns ``(score, source_tag)``:
921 - ``score`` is the 1-10 quality value, or ``None`` if the
922 journal is predatory (signal to auto-remove).
923 - ``source_tag`` identifies which tier produced the score:
924 ``"openalex"``, ``"doaj"``, ``"institution"``, ``"llm"``
925 (Tier 4 live scoring OR cache hit on a prior LLM row),
926 ``"conference"``, or ``"low_confidence"`` (no tier matched).
927 ``None`` when predatory.
929 The tag is attached to the result dict for rendering. Nothing
930 is frozen on the Paper row; the dashboard resolves current
931 quality live from ``journals.quality`` (Tier 4) or the bundled
932 reference DB (Tier 1-3) so a re-scored journal propagates
933 automatically.
934 """
935 dm = self.__data_manager
937 # Extract IDs from result for richer lookups
938 issn = result.get("issn")
939 openalex_sid = result.get("openalex_source_id")
940 publisher = result.get("publisher")
942 # --- Tier 1: Predatory check ---
943 is_pred, pred_source = dm.is_predatory(
944 journal_name=journal_name,
945 publisher_name=publisher,
946 )
947 if is_pred:
948 # Check whitelist override (avoids false positives)
949 if dm.is_whitelisted(issn=issn, name=journal_name):
950 logger.debug(
951 f"Tier 1: '{journal_name}' is on predatory list "
952 f"({pred_source}) but whitelisted — not removing"
953 )
954 else:
955 logger.warning(
956 f"Tier 1: PREDATORY — removing results from "
957 f"'{journal_name}' (source: {pred_source})"
958 )
959 return None, None # Signal auto-remove
961 # --- Tier 2: OpenAlex snapshot ---
962 oa_entry = dm.lookup_openalex(
963 source_id=openalex_sid, issn=issn, name=journal_name
964 )
965 if oa_entry:
966 h_idx = oa_entry.get("h_index")
967 oa_doaj = oa_entry.get("is_in_doaj", False)
968 oa_type = oa_entry.get("type", "journal")
969 oa_quartile = oa_entry.get("quartile")
970 score = dm.derive_quality_score(
971 h_index=h_idx,
972 quartile=oa_quartile,
973 is_in_doaj=oa_doaj,
974 source_type=oa_type,
975 )
976 if score is not None: 976 ↛ 1006line 976 didn't jump to line 1006 because the condition on line 976 was always true
977 logger.debug(
978 f"Tier 2 (OpenAlex): '{journal_name}' → "
979 f"score {score}/10 "
980 f"(quartile: {oa_quartile or '—'}, h-index: {h_idx})"
981 )
982 # Tier 2 results are NOT cached to the user DB — the
983 # read-only reference DB is already a 100–300µs lookup,
984 # so a second-level cache adds no value. Only Tier 4
985 # (LLM) results are cached, below.
986 # Preprint repositories (arxiv, biorxiv, ssrn, ...) get a
987 # low Tier-2 floor because they aren't peer-reviewed. If
988 # the authors are at a strong institution, lift the score
989 # via the institution tier — taking max so a real venue
990 # match (≥6) is never demoted. Only applies to repository
991 # source types and only when score is weak (≤5).
992 if oa_type == "repository" and score <= 5: 992 ↛ 993line 992 didn't jump to line 993 because the condition on line 992 was never true
993 affs = result.get("affiliations")
994 if affs:
995 inst = dm.score_from_affiliations(affs)
996 if inst is not None and inst > score:
997 logger.debug(
998 f"Tier 2+3.5 (preprint lift): "
999 f"'{journal_name}' {score} → {inst} via "
1000 f"institutions: {_format_affiliations(affs)}"
1001 )
1002 return inst, "institution"
1003 return score, "openalex"
1005 # --- Tier 3: DOAJ ---
1006 if issn:
1007 doaj_entry = dm.lookup_doaj(issn=issn)
1008 if doaj_entry:
1009 score = dm.derive_quality_score(is_in_doaj=True)
1010 logger.debug(
1011 f"Tier 3 (DOAJ): '{journal_name}' → score {score}/10"
1012 )
1013 return score, "doaj"
1015 # --- Tier 3.5: Institution lookup ---
1016 # When the venue tiers couldn't score the paper, fall back to
1017 # author affiliations. This is the *only* trust signal we have
1018 # for preprints with no journal_ref or for venues OpenAlex
1019 # doesn't index. Score is capped at 6 inside score_from_affiliations
1020 # so institution alone never beats a real venue match.
1021 affiliations = result.get("affiliations")
1022 if affiliations: 1022 ↛ 1023line 1022 didn't jump to line 1023 because the condition on line 1022 was never true
1023 inst_score = dm.score_from_affiliations(affiliations)
1024 if inst_score is not None:
1025 logger.debug(
1026 f"Tier 3.5 (Institution): '{journal_name}' → "
1027 f"score {inst_score}/10 from institutions: "
1028 f"{_format_affiliations(affiliations)}"
1029 )
1030 return inst_score, "institution"
1032 # --- DB cache: check for cached LLM results before expensive tiers ---
1033 # Tiers 1-3 use bundled data (instant, no caching needed).
1034 # Only Tier 4 (LLM) results are expensive and worth caching.
1035 # The predicate filters on quality_model so scores from a
1036 # superseded LLM model miss the cache and re-score, and
1037 # `cached.quality` is validated against VALID_QUALITY_SCORES to
1038 # evict any pre-fix rows that stored an out-of-set value.
1039 session_ctx = self.__db_session()
1040 if session_ctx is not None:
1041 try:
1042 with session_ctx as session:
1043 cached = (
1044 session.query(Journal)
1045 .filter_by(name=journal_name)
1046 .filter(Journal.score_source == "llm")
1047 .filter(
1048 Journal.quality_model
1049 == get_model_identifier(self.model)
1050 )
1051 .first()
1052 )
1053 if cached is not None: 1053 ↛ 1088line 1053 didn't jump to line 1088
1054 is_fresh = (
1055 time.time() - cached.quality_analysis_time
1056 ) < self.__quality_reanalysis_period.total_seconds()
1057 is_valid = cached.quality in VALID_QUALITY_SCORES
1059 if is_fresh and is_valid:
1060 logger.info(
1061 f"DB cache hit: '{journal_name}' → "
1062 f"score {cached.quality}/10 [cached LLM]"
1063 )
1064 # Cache hit on a Journal row written by the
1065 # LLM path — tag as "llm" so the dashboard
1066 # surfaces it as a Tier 4 verdict.
1067 return cached.quality, "llm"
1068 if is_fresh and not is_valid: 1068 ↛ 1088line 1068 didn't jump to line 1088
1069 logger.warning(
1070 f"Cached score {cached.quality} for "
1071 f"'{journal_name}' not in valid set "
1072 f"{sorted(VALID_QUALITY_SCORES)}; rescoring"
1073 )
1074 # Expired or invalid — fall through to re-evaluate
1075 except Exception:
1076 logger.exception(
1077 f"DB cache read failed for '{journal_name}', "
1078 f"continuing with LLM tiers"
1079 )
1081 # --- Tier 4: LLM analysis (last resort) ---
1082 # Off by default — bundled data covers 217K+ sources and this tier
1083 # adds significant latency (1 SearXNG search + 1 LLM call per
1084 # unknown journal). Users opt in via the
1085 # `search.journal_reputation.enable_llm_scoring` setting. The
1086 # __searxng_available and consecutive-failures checks below remain
1087 # as runtime safety nets even when the user enabled it.
1088 from ...config.search_config import get_setting_from_snapshot
1090 _enable_tier4 = bool(
1091 get_setting_from_snapshot(
1092 "search.journal_reputation.enable_llm_scoring",
1093 False,
1094 settings_snapshot=self.__settings_snapshot,
1095 )
1096 )
1098 # Tier 3.6: LLM-based name cleanup salvage. Gated behind the same
1099 # opt-in flag as Tier 4. Asks the LLM to canonicalise the name
1100 # (handles abbreviations and locations the regex can't), then
1101 # retries the cheap bundled tiers. This costs one extra LLM call
1102 # per unknown journal but can avoid Tier 4's full SearXNG search.
1103 if _enable_tier4:
1104 relabeled = self.__llm_clean_journal_name(journal_name)
1105 if relabeled and relabeled != journal_name: 1105 ↛ 1134line 1105 didn't jump to line 1134 because the condition on line 1105 was always true
1106 logger.debug(
1107 f"Tier 3.6 (LLM cleanup): '{journal_name}' → "
1108 f"'{relabeled}', retrying bundled tiers"
1109 )
1110 oa_retry = dm.lookup_openalex(name=relabeled)
1111 if oa_retry:
1112 h_idx = oa_retry.get("h_index")
1113 oa_doaj = oa_retry.get("is_in_doaj", False)
1114 score = dm.derive_quality_score(
1115 h_index=h_idx,
1116 quartile=oa_retry.get("quartile"),
1117 is_in_doaj=oa_doaj,
1118 source_type=oa_retry.get("type", "journal"),
1119 )
1120 if score is not None: 1120 ↛ 1134line 1120 didn't jump to line 1134 because the condition on line 1120 was always true
1121 logger.info(
1122 f"Tier 3.6 (LLM cleanup → OpenAlex): "
1123 f"'{journal_name}' (as '{relabeled}') → "
1124 f"score {score}/10"
1125 )
1126 # Tier 3.6 is a Tier 2 retry under a cleaned
1127 # name; the result is effectively Tier 2 data
1128 # and is NOT cached in the user DB (reference
1129 # DB lookups are already instant). Tagged
1130 # "openalex" — only the NAME came from the LLM,
1131 # not the score.
1132 return score, "openalex"
1134 if (
1135 _enable_tier4
1136 and self.__searxng_available
1137 and self.__searxng_failures() < 2
1138 ):
1139 try:
1140 quality = self.__analyze_journal_reputation(journal_name)
1141 self.__reset_searxng_failures()
1142 # There used to be a +1 "DOAJ Seal bonus" here. DOAJ
1143 # retired the Seal in April 2025, so the bonus could
1144 # only ever fire on stale pre-2025 data and was removed.
1145 self.__save_llm_score_to_db(name=journal_name, quality=quality)
1146 logger.debug(
1147 f"Tier 4 (LLM): '{journal_name}' → "
1148 f"score {quality}/10 "
1149 f"[via SearXNG + LLM analysis]"
1150 )
1151 return quality, "llm"
1152 except Exception:
1153 failures = self.__bump_searxng_failures()
1154 logger.exception(
1155 f"Tier 4 failed for '{journal_name}'. "
1156 f"Consecutive failures: {failures}"
1157 )
1158 if failures >= 2:
1159 logger.warning(
1160 "Tier 4 disabled for remaining journals in "
1161 "this batch (2 consecutive failures)."
1162 )
1164 # --- Conference heuristic (for papers without DOI or OpenAlex match) ---
1165 # Guard: many high-tier journals start with "Proceedings of …"
1166 # (PNAS, Royal Society A/B, AMS, LMS, …). The bare `proceedings`
1167 # token in `_CONFERENCE_PATTERNS` would otherwise classify them
1168 # as Q3 conferences and throw away their real h-index. Skip the
1169 # heuristic for these — they fall through to the unknown-journal
1170 # score (3) and the user's threshold decides what to do.
1171 if journal_name.lower().lstrip().startswith("proceedings of "): 1171 ↛ 1172line 1171 didn't jump to line 1172 because the condition on line 1171 was never true
1172 logger.debug(
1173 f"Conference heuristic: skipped for '{journal_name}' "
1174 f"(starts with 'Proceedings of' — likely a journal, "
1175 f"not a conference)"
1176 )
1177 elif _is_likely_conference(journal_name): 1177 ↛ 1178line 1177 didn't jump to line 1178 because the condition on line 1177 was never true
1178 score = dm.derive_quality_score(source_type="conference")
1179 logger.debug(
1180 f"Conference heuristic: '{journal_name}' → "
1181 f"score {score}/10 (detected as conference by name pattern)"
1182 )
1183 return score, "conference"
1185 # No tier could score this journal — neither OpenAlex/DOAJ
1186 # venue match nor Tier 3.5 institution salvage produced a
1187 # signal. Score it as low-confidence (3) so the default
1188 # threshold (4) actually filters it out. Distinct from
1189 # predatory (1) — these are merely unknown, not blacklisted.
1190 affs_for_log = result.get("affiliations") or []
1191 logger.debug(
1192 f"No scoring data for '{journal_name}' — flagging as "
1193 f"low-confidence (score 3); tried institutions: "
1194 f"{_format_affiliations(affs_for_log)}"
1195 )
1196 return 3, "low_confidence"
1198 # ------------------------------------------------------------------
1199 # Main filter entry point
1200 # ------------------------------------------------------------------
1202 def filter_results(
1203 self, results: List[Dict], query: str, **kwargs
1204 ) -> List[Dict]:
1205 """Filter results by journal quality, with deduplication."""
1206 logger.info(
1207 f"Journal filter: processing {len(results)} results "
1208 f"(threshold={self.__threshold})"
1209 )
1210 # Fail-soft during a fresh install: if the reference DB file
1211 # isn't on disk yet, don't score anything. Every journal would
1212 # otherwise fall through to the "no scoring data" branch and
1213 # be marked as score 3, which is semantically wrong — we don't
1214 # know the journal is unknown, we just haven't loaded the data
1215 # yet. Tag each result with the QUALITY_PENDING sentinel so the
1216 # report renderer can show the user a helpful note pointing
1217 # them at /metrics/journals.
1218 #
1219 # We probe the DB file path directly rather than calling
1220 # ``data_manager.available`` — the latter has side effects
1221 # (triggers `_ensure_engine`, which tries to lazy-build the DB
1222 # and can block on an in-flight download for several minutes).
1223 # A cached engine on the data manager means either the file
1224 # was successfully opened earlier, or a test fixture has
1225 # injected an in-memory DB — either way we're good to score.
1226 try:
1227 from ...config.paths import get_journal_data_directory
1229 db_file = get_journal_data_directory() / "journal_quality.db"
1230 on_disk = db_file.exists() and db_file.stat().st_size > 0
1231 engine_cached = (
1232 getattr(self.__data_manager, "_engine", None) is not None
1233 )
1234 db_ready = on_disk or engine_cached
1235 logger.info(
1236 f"Journal filter: db_ready={db_ready} "
1237 f"(on_disk={on_disk}, engine_cached={engine_cached})"
1238 )
1239 except Exception:
1240 logger.exception(
1241 "Journal filter: db-ready probe raised; "
1242 "assuming DB not ready (pending)."
1243 )
1244 db_ready = False
1245 if not db_ready: 1245 ↛ 1246line 1245 didn't jump to line 1246 because the condition on line 1245 was never true
1246 from ...utilities.search_utilities import QUALITY_PENDING
1248 # Fire-and-forget the download in a daemon thread so the
1249 # pending-marker copy ("by the time you check, it may
1250 # already be done") is actually true. Without this the
1251 # filter would just tag every search "pending" forever
1252 # and nobody would ever fetch the data unless the user
1253 # clicked Download manually. The 30-second TTL cache in
1254 # ensure_journal_data prevents multiple concurrent filter
1255 # workers from all racing to spawn the same download.
1256 #
1257 # Egress policy: skip the background fetch under
1258 # PRIVATE_ONLY or STRICT — those scopes shouldn't egress at
1259 # all, and the journal data sources (OpenAlex / DOAJ /
1260 # JabRef bulk downloads) are all public hosts.
1261 if self._should_skip_journal_fetch_for_scope():
1262 logger.bind(policy_audit=True).info(
1263 "journal-data background fetch skipped: egress scope "
1264 "forbids public fetches"
1265 )
1266 else:
1267 _start_background_journal_fetch()
1269 # Respect exclude_non_published — results without a venue
1270 # still get dropped in pending mode, same as in the full
1271 # scoring path. Only venued results carry the marker.
1272 out = []
1273 tagged = 0
1274 dropped = 0
1275 for r in results:
1276 if r.get("journal_ref"):
1277 r.setdefault("journal_quality", QUALITY_PENDING)
1278 out.append(r)
1279 tagged += 1
1280 elif not self.__exclude_non_published:
1281 out.append(r)
1282 else:
1283 dropped += 1
1284 logger.warning(
1285 f"Journal filter: reference DB not yet built — "
1286 f"tagged {tagged} result(s) with QUALITY_PENDING, "
1287 f"kept {len(out) - tagged} venueless, "
1288 f"dropped {dropped} (exclude_non_published). "
1289 f"Background download triggered if not already running."
1290 )
1291 return out
1293 # Initialize `filtered` outside the try so the predatory-safe
1294 # fallback in the except handler can always reference it even
1295 # if the crash happens before Pass-1 populates anything.
1296 filtered: list = []
1298 try:
1299 # Reset the per-thread fail-fast counter for each batch.
1300 # The counter lives in `threading.local()` so concurrent
1301 # callers on the same filter instance don't clobber each
1302 # other (see Bug A3).
1303 self.__reset_searxng_failures()
1305 # Pass 1: collect the richest metadata per journal (the result
1306 # with ISSN/source_id) so scoring uses the best available data.
1307 journal_best_result: Dict[str, Dict] = {}
1308 results_with_journals: list[tuple[Dict, str]] = []
1310 def _handle_no_venue(result: Dict) -> None:
1311 """Institution-salvage then exclude_non_published policy.
1313 If Tier 3.5 salvages a score from author affiliations,
1314 the result carries that numeric score. Otherwise we
1315 tag it with ``QUALITY_PREPRINT`` so the report renderer
1316 can show a "preprint — not in journal catalog" label
1317 instead of leaving the quality column ambiguously blank.
1318 """
1319 from ...utilities.search_utilities import QUALITY_PREPRINT
1321 affs = result.get("affiliations")
1322 if affs: 1322 ↛ 1323line 1322 didn't jump to line 1323 because the condition on line 1322 was never true
1323 inst_score = self.__data_manager.score_from_affiliations(
1324 affs
1325 )
1326 if (
1327 inst_score is not None
1328 and inst_score >= self.__threshold
1329 ):
1330 result["journal_quality"] = inst_score
1331 logger.debug(
1332 f"Tier 3.5 (Institution, no venue): "
1333 f"'{result.get('title', '')[:60]}' → "
1334 f"score {inst_score}/10 from institutions: "
1335 f"{_format_affiliations(affs)}"
1336 )
1337 filtered.append(result)
1338 return
1339 if not self.__exclude_non_published:
1340 result.setdefault("journal_quality", QUALITY_PREPRINT)
1341 filtered.append(result)
1343 # Per-batch cache for journal name cleaning — avoids redundant
1344 # regex + abbreviation DB lookups when multiple results come
1345 # from the same raw journal_ref string (common in OpenAlex
1346 # result batches).
1347 _name_cache: Dict[str, str] = {}
1349 for result in results:
1350 journal_ref = result.get("journal_ref")
1351 # Strip whitespace-only refs — " " is truthy but has no
1352 # meaningful content for the filter.
1353 if isinstance(journal_ref, str):
1354 journal_ref = journal_ref.strip()
1355 if not journal_ref:
1356 _handle_no_venue(result)
1357 continue
1359 # Use per-batch cache to skip redundant cleaning for
1360 # repeated raw journal_ref values in the same batch.
1361 clean_name = _name_cache.get(journal_ref)
1362 if clean_name is None: 1362 ↛ 1368line 1362 didn't jump to line 1368 because the condition on line 1362 was always true
1363 clean_name = self.__clean_journal_name(journal_ref)
1364 _name_cache[journal_ref] = clean_name
1365 # Cleanup can reduce a volume/page-only ref to "". Treat
1366 # that the same as "no venue" rather than bucketing all
1367 # affected results under a degenerate empty-string key.
1368 if not clean_name: 1368 ↛ 1369line 1368 didn't jump to line 1369 because the condition on line 1368 was never true
1369 _handle_no_venue(result)
1370 continue
1372 results_with_journals.append((result, clean_name))
1374 if clean_name not in journal_best_result:
1375 journal_best_result[clean_name] = result
1376 else:
1377 prev = journal_best_result[clean_name]
1378 if ( 1378 ↛ 1384line 1378 didn't jump to line 1384 because the condition on line 1378 was never true
1379 not prev.get("issn")
1380 and not prev.get("openalex_source_id")
1381 ) and (
1382 result.get("issn") or result.get("openalex_source_id")
1383 ):
1384 journal_best_result[clean_name] = result
1386 # Pass 2: score each unique journal once, then filter
1387 journal_scores: Dict[str, tuple[int | None, str | None]] = {}
1389 for result, clean_name in results_with_journals:
1390 if clean_name not in journal_scores:
1391 journal_scores[clean_name] = self.__score_journal(
1392 clean_name, journal_best_result[clean_name]
1393 )
1395 score, source_tag = journal_scores[clean_name]
1397 if score is None:
1398 # Predatory → auto-remove. Include original journal_ref
1399 # and URL in the log so false-positive reports can be
1400 # debugged without re-running the query.
1401 logger.warning(
1402 f"Auto-removed predatory: "
1403 f"title='{result.get('title', '')[:80]}' "
1404 f"journal_ref='{result.get('journal_ref', '')!r}' "
1405 f"cleaned='{clean_name}' "
1406 f"url={result.get('link') or result.get('url') or '—'}"
1407 )
1408 continue
1410 if score >= self.__threshold:
1411 result["journal_quality"] = score
1412 # Cleaned name that keyed the successful score —
1413 # persisted on Paper.container_title so the dashboard
1414 # can GROUP BY it and enrich from the reference DB.
1415 result["journal_name_matched"] = clean_name
1416 # Source tag — propagates to the rendered quality
1417 # tag in the research output. Nothing is frozen on
1418 # the Paper row; the dashboard resolves current
1419 # quality live from journals.quality (Tier 4) or
1420 # the bundled reference DB (Tier 1-3).
1421 result["journal_quality_source"] = source_tag
1422 filtered.append(result)
1424 predatory_count = sum(
1425 1 for s, _ in journal_scores.values() if s is None
1426 )
1427 passed_count = sum(
1428 1
1429 for s, _ in journal_scores.values()
1430 if s is not None and s >= self.__threshold
1431 )
1432 below_count = sum(
1433 1
1434 for s, _ in journal_scores.values()
1435 if s is not None and s < self.__threshold
1436 )
1437 logger.info(
1438 f"Journal quality filter: {len(results)} → "
1439 f"{len(filtered)} results | "
1440 f"{len(journal_scores)} unique journals scored | "
1441 f"{passed_count} passed, {below_count} below threshold, "
1442 f"{predatory_count} predatory removed"
1443 )
1444 return filtered
1446 except Exception:
1447 # Safety net: a filter crash should not kill the entire search,
1448 # but it MUST NOT re-admit predatory journals either.
1449 # `filtered` is predatory-free by construction (the pass-2 loop
1450 # `continue`s on predatory scores). Returning `results` — the
1451 # original unfiltered list — would leak predatory sources that
1452 # Tier 1 had already caught. Prefer losing in-flight non-
1453 # predatory results over breaking the predatory-removal
1454 # safety contract. Logged at ERROR so the root cause surfaces.
1455 logger.exception(
1456 "Journal quality filtering failed — returning partial "
1457 "(predatory-free) results. This is a bug that should be "
1458 "investigated."
1459 )
1460 return filtered