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