Coverage for src/local_deep_research/advanced_search_system/filters/journal_reputation

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.json_utils import get_llm_response_text

55from ...utilities.llm_utils import get_model_identifier

56from ...utilities.resource_utils import safe_close

57from ...utilities.thread_context import get_search_context

58from ...web_search_engines.search_engine_factory import create_search_engine

59from .base_filter import BaseFilter

60from ...constants import DEFAULT_SEARCH_TOOL

63# Patterns that indicate a venue is a conference, not a journal.

64# Used as a fallback when DOI enrichment and OpenAlex lookup both miss.

65_CONFERENCE_PATTERNS = [

66 re.compile(r"\b(?:proceedings|proc\.)\b", re.I),

67 re.compile(r"\b(?:conference|conf\.)\b", re.I),

68 re.compile(r"\b(?:symposium|symp\.)\b", re.I),

69 re.compile(r"\bworkshop\b", re.I),

70 re.compile(

72 ),

73]

76def _is_likely_conference(name: str) -> bool:

77 """Detect if a venue name is likely a conference based on common patterns."""

78 return any(p.search(name) for p in _CONFERENCE_PATTERNS)

81def _sanitize_name(name: str) -> str:

82 """Sanitize a journal name for safe use in logs and LLM prompts.

84 Args:

85 name: Raw journal name string, potentially containing control

86 characters, excessive length, or quotes.

88 Returns:

89 Sanitized string safe for use in logs and LLM prompts.

90 """

91 # Strip control + format characters. Covers C0/C1 (log injection),

92 # bidi overrides (U+202A-E, U+2066-9), zero-width / joiner chars

93 # (U+200B-F, U+2060-4, U+FEFF), Arabic letter mark, and digit-shape

94 # controls — the comprehensive pattern audited in log_sanitizer,

95 # not the narrow C0/C1-only regex we used to have here.

96 name = strip_control_chars(name)

97 # Normalize Unicode (prevents lookalike bypasses)

98 name = unicodedata.normalize("NFKC", name)

99 # Limit length (prevents resource exhaustion in prompts)

100 if len(name) > 500: 100 ↛ 101line 100 didn't jump to line 101 because the condition on line 100 was never true

101 name = name[:500] + "..."

102 # Strip quotes that could break prompt structure

103 name = name.replace('"', "'")

104 return name.strip()

105

106

107def _format_affiliations(affiliations: list, max_n: int = 3) -> str:

108 """Render an affiliation list as a compact human-readable string for

109 log lines. Accepts the same shapes as ``score_from_affiliations``

110 (plain strings or dicts with a ``name`` key) and truncates after

111 ``max_n`` entries so a 20-author paper doesn't blow up the log.

112 """

113 if not affiliations: 113 ↛ 115line 113 didn't jump to line 115 because the condition on line 113 was always true

114 return "(none)"

115 names: list[str] = []

116 for aff in affiliations:

117 if isinstance(aff, str):

118 names.append(aff)

119 elif isinstance(aff, dict):

120 nm = aff.get("name") or aff.get("display_name")

121 if nm:

122 names.append(nm)

123 if not names:

124 return "(unknown)"

125 shown = names[:max_n]

126 suffix = "" if len(names) <= max_n else f" (+{len(names) - max_n} more)"

127 return ", ".join(shown) + suffix

128

129

130_bg_fetch_lock = threading.Lock()

131_bg_fetch_thread: Optional[threading.Thread] = None

132

133

134def _start_background_journal_fetch() -> None:

135 """Kick off ``ensure_journal_data(auto_download=True)`` in a daemon

136 thread the first time a search hits the pending path.

137

138 The worker returns immediately if another thread is already in

139 flight (``_bg_fetch_thread.is_alive()``) — so 30 concurrent filter

140 workers can't each spawn their own download. The ``ensure_journal_data``

141 TTL cache provides a second line of defence.

142

143 Daemon thread so it doesn't block process exit.

144 """

145 global _bg_fetch_thread

146 from ...settings.env_definitions.testing import testing_with_mocks

147

148 if testing_with_mocks():

149 # Test suite's mock-only mode (defaulted to true by

150 # tests/conftest.py). This trigger is fire-and-forget and

151 # implicit — nothing calling into the filter's pending path

152 # explicitly asked for a live download — so refuse to spawn

153 # the background thread rather than silently reaching out to

154 # OpenAlex/DOAJ/etc. mid-test. Tests exercising the real

155 # download path call downloader.download_journal_data()

156 # directly with the per-source fetchers mocked.

157 logger.debug(

158 "journal-data background fetch skipped: LDR_TESTING_WITH_MOCKS=true"

159 )

160 return

161 with _bg_fetch_lock:

162 if _bg_fetch_thread is not None and _bg_fetch_thread.is_alive():

163 logger.debug(

164 "journal-data background fetch already in flight — "

165 "not spawning a second thread"

166 )

167 return

168

169 def _run():

170 try:

171 # Late import so this module doesn't pay the cost of

172 # importing the downloader at its own import time.

173 from ...journal_quality.downloader import (

174 ensure_journal_data,

175 )

176

177 logger.info(

178 "journal-data background fetch: starting "

179 "(triggered by filter pending path)"

180 )

181 ensure_journal_data(auto_download=True)

182 logger.info("journal-data background fetch: done")

183 except Exception:

184 logger.exception(

185 "journal-data background fetch crashed — "

186 "next search will retry"

187 )

188

189 _bg_fetch_thread = threading.Thread(

190 target=_run,

191 name="journal-data-bg-fetch",

192 daemon=True,

193 )

194 _bg_fetch_thread.start()

195

196

197class JournalFilterError(Exception):

198 """

199 Custom exception for errors related to journal filtering.

200 """

201

202

203class JournalReputationFilter(BaseFilter):

204 """

205 A filter for academic results that considers the reputation of journals.

206

207 Uses a tiered scoring approach: bundled data (OpenAlex, DOAJ, predatory

208 lists) for most journals, with LLM-based analysis via SearXNG as a

209 fallback for truly unknown journals.

210

211 Predatory journals are **automatically removed** from results.

212 """

213

214 def __init__(

215 self,

216 model: BaseChatModel | None = None,

217 reliability_threshold: int | None = None,

218 max_context: int | None = None,

219 exclude_non_published: bool | None = None,

220 quality_reanalysis_period: timedelta | None = None,

221 settings_snapshot: Dict[str, Any] | None = None,

222 ):

223 """Initialize the journal reputation filter.

224

225 Args:

226 model: The LLM model to use for Tier 4 analysis. If None,

227 the default LLM from settings will be used.

228 reliability_threshold: Minimum quality score (1-10) for a

229 result to pass. Read from settings if not specified.

230 max_context: Maximum characters of source content for LLM

231 quality evaluation.

232 exclude_non_published: If True, exclude results that don't

233 have an associated journal publication reference.

234 quality_reanalysis_period: Period after which cached journal

235 quality assessments are refreshed.

236 settings_snapshot: Settings snapshot for thread context.

237 """

238 super().__init__(model)

239

240 self._owns_llm = self.model is None

241 if self.model is None:

242 # Forward the snapshot so the LLM PEP at llm_config.py:295

243 # can evaluate ``llm.require_local_endpoint`` / egress scope.

244 # Dropping it here previously bypassed the PEP entirely on

245 # search engines that constructed this filter with no llm.

246 self.model = get_llm(settings_snapshot=settings_snapshot)

247

248 # Import here to avoid circular import

249 from ...config.search_config import get_setting_from_snapshot

250

251 self.__threshold = reliability_threshold

252 if self.__threshold is None:

253 self.__threshold = int(

254 get_setting_from_snapshot(

255 "search.journal_reputation.threshold",

256 2,

257 settings_snapshot=settings_snapshot,

258 )

259 )

260 self.__max_context = max_context

261 if self.__max_context is None:

262 self.__max_context = int(

263 get_setting_from_snapshot(

264 "search.journal_reputation.max_context",

265 3000,

266 settings_snapshot=settings_snapshot,

267 )

268 )

269 self.__exclude_non_published = exclude_non_published

270 if self.__exclude_non_published is None:

271 self.__exclude_non_published = bool(

272 get_setting_from_snapshot(

273 "search.journal_reputation.exclude_non_published",

274 False,

275 settings_snapshot=settings_snapshot,

276 )

277 )

278 self.__quality_reanalysis_period = quality_reanalysis_period

279 if self.__quality_reanalysis_period is None:

280 self.__quality_reanalysis_period = timedelta(

281 days=int(

282 get_setting_from_snapshot(

283 "search.journal_reputation.reanalysis_period",

284 365,

285 settings_snapshot=settings_snapshot,

286 )

287 )

288 )

289

290 self.__settings_snapshot = settings_snapshot

291

292 # SearXNG for Tier 4 (LLM fallback). Not strictly required anymore

293 # since bundled data covers most journals.

294 self.__engine = create_search_engine(

295 "searxng", llm=self.model, settings_snapshot=settings_snapshot

296 )

297 self.__searxng_available = self.__engine is not None and getattr(

298 self.__engine, "_is_available", False

299 )

300 if not self.__searxng_available:

301 logger.info(

302 "SearXNG not available — Tier 4 (LLM analysis) disabled. "

303 "Bundled data tiers still active."

304 )

305

306 # Fail-fast counter for SearXNG failures within a batch.

307 # Stored in `threading.local()` so concurrent `filter_results`

308 # calls on the same cached filter instance (the parallel search

309 # engine reuses instances across worker threads — see

310 # concurrent engine threads) can't clobber each other's

311 # counter. Per-thread state is reset at the top of every

312 # `filter_results` invocation.

313 self.__tls = threading.local()

314

315 # Lock serializing access to the shared SearXNG engine for

316 # Tier 4. BaseSearchEngine instances keep mutable bookkeeping

317 # state (_last_results_count, _search_results, rate-limit

318 # tracker) on self; two concurrent .run() calls on the same

319 # instance would clobber that state. Tier 4 is rarely hit in

320 # practice (requires enable_llm_scoring=True + SearXNG +

321 # bundled-data miss), so the lock's contention cost is

322 # negligible compared to the correctness guarantee.

323 self.__engine_lock = threading.Lock()

324

325 # Journal data manager (loads bundled datasets lazily)

326 self.__data_manager = get_journal_data_manager()

327

328 # ------------------------------------------------------------------

329 # Thread-local fail-fast counter accessors

330 # ------------------------------------------------------------------

331

332 def __searxng_failures(self) -> int:

333 return getattr(self.__tls, "searxng_failures", 0)

334

335 def __reset_searxng_failures(self) -> None:

336 self.__tls.searxng_failures = 0

337

338 def __bump_searxng_failures(self) -> int:

339 n = self.__searxng_failures() + 1

340 self.__tls.searxng_failures = n

341 return n

342

343 def close(self) -> None:

344 """Close the SearXNG engine and LLM client."""

345 if hasattr(self, "_JournalReputationFilter__engine"): 345 ↛ 349line 345 didn't jump to line 349 because the condition on line 345 was always true

346 # allow_none=True: SearXNG is optional (Tier 4 only), so

347 # __engine is None in the common no-SearXNG configuration.

348 safe_close(self.__engine, "SearXNG engine", allow_none=True)

349 if self._owns_llm: 349 ↛ 350line 349 didn't jump to line 350 because the condition on line 349 was never true

350 safe_close(self.model, "journal filter LLM")

351

352 def _should_skip_journal_fetch_for_scope(self) -> bool:

353 """Return True when the active egress scope forbids public

354 fetches (PRIVATE_ONLY / STRICT). The journal-data sources

355 (OpenAlex, DOAJ, JabRef) are all public, so under these scopes

356 the daemon thread would burn cycles and (worse) attempt egress

357 the user explicitly opted out of.

358 """

359 snapshot = self.__settings_snapshot

360 if not snapshot:

361 return False

362 try:

363 from ...security.egress.policy import (

364 EgressScope,

365 PolicyDeniedError,

366 context_from_snapshot,

367 )

368 from ...search_system import username_from_snapshot

369

370 primary_raw = unwrap_setting(

371 snapshot.get("search.tool", DEFAULT_SEARCH_TOOL)

372 )

373 # Thread username so a per-user private retriever primary

374 # resolves PRIVATE_ONLY and the public journal fetch is skipped.

375 ctx = context_from_snapshot(

376 snapshot,

377 primary_raw or DEFAULT_SEARCH_TOOL,

378 username=username_from_snapshot(snapshot),

379 )

380 return ctx.scope in (EgressScope.PRIVATE_ONLY, EgressScope.STRICT)

381 except PolicyDeniedError:

382 # Corrupt/unknown scope value — we cannot certify that a public

383 # journal fetch is permitted, so SKIP the fetch (fail closed).

384 # This matches the hardened sibling

385 # notifications.manager._filter_urls_by_egress_policy, which also

386 # refuses rather than proceeds when the policy is unevaluable.

387 logger.bind(policy_audit=True).warning(

388 "journal fetch skipped: egress policy unevaluable "

389 "(corrupt scope) — failing closed"

390 )

391 return True

392 except Exception:

393 # Other errors (missing key, snapshot shape) — fail open; the

394 # per-URL runtime check still gates each individual fetch.

395 return False

396

397 @classmethod

398 def create_default(

399 cls,

400 model: BaseChatModel | None = None,

401 *,

402 engine_name: str,

403 settings_snapshot: Dict[str, Any] | None = None,

404 ) -> Optional["JournalReputationFilter"]:

405 """Initializes a default configuration of the filter based on settings.

406

407 SearXNG is not required — the filter works with bundled data alone.

408 SearXNG enables the optional Tier 4 (LLM analysis) for journals

409 not found in the bundled datasets.

410

411 Args:

412 model: Optional LLM model for Tier 4 analysis.

413 engine_name: Search engine configuration key (e.g. "arxiv").

414 settings_snapshot: Optional frozen settings dict.

415

416 Returns:

417 A configured JournalReputationFilter, or None if filtering

418 is disabled in settings for this engine.

419 """

420 from ...config.search_config import get_setting_from_snapshot

421

422 try:

423 enabled = get_setting_from_snapshot(

424 f"search.engine.web.{engine_name}.journal_reputation.enabled",

425 True,

426 settings_snapshot=settings_snapshot,

427 )

428 logger.info(

429 f"Journal filter create_default: engine={engine_name}, "

430 f"enabled={enabled} (type={type(enabled).__name__})"

431 )

432 if not bool(enabled):

433 logger.info(

434 f"Journal filter disabled for {engine_name} in settings"

435 )

436 return None

437

438 filt = JournalReputationFilter(

439 model=model, settings_snapshot=settings_snapshot

440 )

441 logger.info(

442 f"Journal filter created for {engine_name} — "

443 f"threshold={filt._JournalReputationFilter__threshold}"

444 )

445 return filt

446 except PolicyDeniedError:

447 # The egress policy refused the LLM the filter needs. This

448 # must propagate so the user sees a clear policy refusal —

449 # silently returning None would let unfiltered results

450 # through under ``llm.require_local_endpoint=True``.

451 raise

452 except Exception:

453 # Any other failure — settings read, filter init — returns

454 # None rather than silently defaulting to enabled. A separate

455 # silent ``except Exception: enabled = True`` branch used to

456 # wrap only the settings read, which hid legitimate

457 # configuration errors; per CLAUDE.md, fallbacks have to be

458 # explicit, not default-on.

459 logger.exception(

460 f"Failed to configure journal filter for {engine_name}; "

461 "results will not be journal-quality filtered for this engine"

462 )

463 return None

464

465 @staticmethod

466 def __db_session() -> Session | None:

467 """Get a database session using the current search context credentials.

468

469 Returns:

470 SQLAlchemy Session context manager for the user's database, or

471 ``None`` if no search context is available (e.g. when called

472 from the preview filter phase, before per-user thread context

473 has been propagated). Callers should treat ``None`` as

474 "skip the DB operation".

475 """

476 context = get_search_context()

477 if context is None:

478 return None

479 username = context.get("username")

480 password = context.get("user_password")

481 return get_user_db_session(username=username, password=password)

482

483 # ------------------------------------------------------------------

484 # Journal name cleaning (with LRU cache to avoid duplicate LLM calls)

485 # ------------------------------------------------------------------

486

487 def __clean_journal_name(self, journal_name: str) -> str:

488 """Clean journal name to normalize for deduplication and lookup.

489

490 Deterministic regex-based cleaning only: strips volume / page /

491 year / month references, then tries a JabRef abbreviation

492 expansion. This method does NOT call the LLM — it is the cheap,

493 instant cleaning path that every Tier runs through first.

494

495 The separate ``__llm_clean_journal_name`` method is the LLM

496 fallback, invoked later only as a salvage step when the bundled

497 data tiers all miss and ``enable_llm_scoring`` is on.

498 Abbreviations not in the JabRef list and location suffixes

499 ("ICML 2023, Honolulu") only get LLM-cleaned on that salvage

500 path; this method returns them unchanged.

501

502 Args:

503 journal_name: Raw journal name from search results.

504

505 Returns:

506 Cleaned, normalized journal name.

507 """

508 # Sanitize first (strips control chars, normalizes Unicode)

509 journal_name = _sanitize_name(journal_name)

510 # Regex handles volume/page/year stripping (instant)

511 cleaned = self.__regex_clean_journal_name(journal_name)

512

513 # Try JabRef abbreviation expansion (deterministic, instant)

514 expanded = self.__data_manager.expand_abbreviation(cleaned)

515 if expanded:

516 logger.debug(f"Abbreviation expanded: '{cleaned}' → '{expanded}'")

517 return expanded

518

519 if cleaned != journal_name:

520 logger.debug(

521 f"Regex-cleaned journal name: '{journal_name}' → '{cleaned}'"

522 )

523 return cleaned

524

525 @staticmethod

526 def __regex_clean_journal_name(name: str) -> str:

527 """

528 Fast regex-based journal name normalization. Strips volume, issue,

529 page, year, and month references. No LLM needed.

530 """

531 months = (

534 "jan|feb|mar|apr|jun|jul|aug|sep|oct|nov|dec"

535 )

536

537 # Strip leading/trailing whitespace

538 name = name.strip()

539

540 # Strip a leading [bracketed-original-language] prefix that

541 # MEDLINE uses for non-English journals:

542 # "[Rinsho ketsueki] The Japanese journal of clinical hematology"

543 # → "The Japanese journal of clinical hematology"

544 name = re.sub(r"^\[[^\]]+\]\s*", "", name)

545

546 # Strip trailing publisher suffixes that some search engines glue

547 # onto the journal name (e.g. "Information Fusion Elsevier" or

548 # "Cell Press"). Conservative — only the handful of well-known

549 # academic publishers, anchored at end of string with a leading

550 # space so we don't eat them mid-name.

551 name = re.sub(

552 r"\s+(?:Elsevier|Springer|Wiley|Nature\s+Publishing|"

553 r"Cell\s+Press|MDPI|Sage|Taylor\s*&?\s*Francis|"

554 r"Oxford\s+University\s+Press|Cambridge\s+University\s+Press|"

555 r"IEEE|ACM|Routledge|Frontiers)\s*$",

556 "",

557 name,

558 flags=re.IGNORECASE,

559 )

560

561 # Strip a leading 4-digit year ("2015 Plasma Phys. ..." → "Plasma Phys. ...")

562 name = re.sub(r"^(?:19|20)\d{2}\s+", "", name)

563

564 # Strip a leading ordinal volume marker, e.g.

565 # "31st Conference on Neural Information Processing Systems" → "Conference on …"

566 # Without this the OpenAlex name lookup fails on most conference

567 # entries because the canonical name has no ordinal prefix.

568 name = re.sub(

569 r"^\d+(?:st|nd|rd|th)\s+",

570 "",

571 name,

572 flags=re.IGNORECASE,

573 )

574

575 # Remove month+year references FIRST (before bare year strip),

576 # so "September 2023" is consumed as a unit and we don't leave

577 # the month behind as an orphan word.

578 name = re.sub(

579 rf",?\s*\b(?:{months})\b\.?\s+(?:19|20)?\d{{2,4}}",

580 "",

581 name,

582 flags=re.IGNORECASE,

583 )

584

585 # Remove volume/issue/page refs: "Vol. 12", "Issue 3", "pp. 100-200"

586 name = re.sub(

587 r",?\s*(?:vol(?:ume)?\.?\s*\d+|"

588 r"issue\s*\d+|"

589 r"no\.?\s*\d+|"

590 r"pp?\.?\s*\d+[\s–-]*\d*|"

591 r"pages?\s*\d+[\s–-]*\d*)",

592 "",

593 name,

594 flags=re.IGNORECASE,

595 )

596 # Remove volume(issue) patterns: "141(5)" — and bare "(15)" issues.

597 name = re.sub(r",?\s*\d+$\d+$", "", name)

598 name = re.sub(r"\s*$\d+$\s*", " ", name)

599 # Remove year references: "(2023)", ", 2023". Anchored to 19xx/20xx

600 # so 4-digit page numbers like ", 1063" in "106335" aren't eaten.

601 name = re.sub(r"\s*[$,]\s*(?:19|20)\d{2}\b\s*$?", "", name)

602 # Remove bare trailing citation data: ", 95, 146802" style

603 # Only strips when there's a comma before the first number

604 # (preserves "NeurIPS 2023" where space-number is part of the name)

605 name = re.sub(r",\s*\d[\d,\s]*$", "", name)

606 # Strip trailing alphanumeric volume markers: "E48", "R569", "L102"

607 # (single uppercase letter followed by digits at end of string).

608 name = re.sub(r"\s+[A-Z]\d+\s*$", "", name)

609 # Strip trailing volume/page debris like "170 266-275", "71: 1-10",

610 # "151:48-60", or a bare trailing volume "116". Repeated to peel

611 # multiple chunks. Stops when only the journal name remains. We

612 # require the run to start with whitespace or punctuation so we

613 # don't eat the "2023" of "NeurIPS 2023" — the trailing-year regex

614 # below handles that case.

615 prev = None

616 while prev != name:

617 prev = name

618 name = re.sub(

619 r"[\s,;:]+\d+(?:\s*[:\-–]\s*\d+)?(?:\s*[-–]\s*\d+)?\s*$",

620 "",

621 name,

622 )

623 # Strip leftover trailing month name (no year) — happens when the

624 # year was stripped by another regex first.

625 name = re.sub(

626 rf",?\s*\b(?:{months})\b\.?\s*$",

627 "",

628 name,

629 flags=re.IGNORECASE,

630 )

631 # Strip leftover bare volume/page keyword at the end ("p", "pp",

632 # "vol", "vol.", "no", "no.") that survives when the number got

633 # truncated upstream by the search engine result preview.

634 name = re.sub(

635 r",?\s*\b(?:vol(?:ume)?|pp?|no)\b\.?\s*$",

636 "",

637 name,

638 flags=re.IGNORECASE,

639 )

640 # Strip empty / whitespace-only trailing parens — arXiv

641 # journal_ref fields sometimes end with "()" where the citation

642 # year was stripped upstream, e.g. "Physical Review Research ()".

643 name = re.sub(r"\s*$\s*$\s*$", "", name)

644 # Strip geographic qualifiers: "(London)", "(New York)", "(US)"

645 # Only strip parenthesized suffixes that contain no digits

646 # (preserves "NeurIPS (2023)" which is handled by the year regex)

647 name = re.sub(r"\s*$[^()0-9]+$\s*$", "", name)

648 # Strip trailing truncated volume/page markers: ", v", ", p",

649 # ", vol" — these appear when the search engine preview cut the

650 # citation mid-keyword ("Plasma Physics and Controlled Fusion,

651 # vol. 63, no. 8" → "Plasma Physics and Controlled Fusion, v").

652 name = re.sub(

653 r",\s*(?:v|vol|p|pp|no|n)\.?\s*$",

654 "",

655 name,

656 flags=re.IGNORECASE,

657 )

658 # Remove trailing punctuation and whitespace

659 name = re.sub(r"[,;.\s]+$", "", name)

660 # Strip a trailing 4-digit year/volume (conferences: "NeurIPS 2023"

661 # → "NeurIPS"). Comes after the punctuation strip so any trailing

662 # comma/period is already gone, and after the parenthesized-year

663 # regex above so we don't double-process "(2023)".

664 name = re.sub(r"\s+\d{4}\s*$", "", name)

665 # Normalize "&" → "and" for consistent matching

666 name = re.sub(r"\s*&\s*", " and ", name)

667 # Normalize internal whitespace

668 return re.sub(r"\s+", " ", name).strip()

669

670 def __llm_clean_journal_name(self, journal_name: str) -> Optional[str]:

671 """LLM-based fallback for canonicalizing unusual journal names.

672

673 The regex + JabRef abbreviation tiers handle the common cases

674 (volume/year/page stripping, well-known abbreviations like

675 "Phys. Rev. Lett." → "Physical Review Letters"). They cannot

676 handle locations ("ICML 2023, Honolulu"), unusual abbreviations

677 not in the JabRef list, or non-English title transliterations.

678

679 This is gated behind ``enable_llm_scoring`` so it never fires

680 unless the user opted into the Tier 4 LLM path. Called only as a

681 salvage step when bundled tiers all miss, so the LLM bill is

682 bounded by the number of *unrecognised* journals per query, not

683 every journal.

684

685 Args:

686 journal_name: A name that the regex tier could not match

687 against any bundled dataset.

688

689 Returns:

690 A canonicalised name from the LLM, or ``None`` if the call

691 failed or the response was empty.

692 """

693 prompt = (

694 f"Clean up the following journal or conference name:\n\n"

695 f'"{journal_name}"\n\n'

696 "Remove any references to volumes, pages, months, or years. "

697 "Expand common abbreviations. For conferences, remove "

698 "locations. Output only the clean name, no explanation."

699 )

700 try:

701 response = self.model.invoke(prompt)

702 content = getattr(response, "content", None) or response

703 cleaned = str(content).strip().strip('"').strip("'")

704 if not cleaned: 704 ↛ 705line 704 didn't jump to line 705 because the condition on line 704 was never true

705 return None

706 return cleaned

707 except (

708 ConnectionError,

709 TimeoutError,

710 ValueError,

711 ) as e:

712 # Network / service / parse failures are expected and

713 # recoverable — caller falls back to the regex-cleaned name.

714 # Surface at WARNING (not silent / not DEBUG) so they're

715 # visible during triage without flooding info-level logs.

716 # Log only the exception class name; the message can carry

717 # request-specific data (URLs, prompts) that doesn't belong

718 # in operational logs.

719 logger.warning(

720 f"LLM name cleaning failed for '{journal_name}' "

721 f"({type(e).__name__}); using regex-cleaned version"

722 )

723 return None

724

725 # ------------------------------------------------------------------

726 # Tier 4: LLM-based analysis (last resort)

727 # ------------------------------------------------------------------

728

729 def __analyze_journal_reputation(self, journal_name: str) -> int:

730 """Analyze journal reputation via 1 SearXNG search + 1 LLM call.

731

732 This is Tier 4 — the last-resort scoring path. Only used when

733 the journal is not found in bundled data (OpenAlex, DOAJ, predatory).

734 Uses a single web search for context, then a single LLM call to score.

735

736 Args:

737 journal_name: Cleaned journal name to research.

738

739 Returns:

740 Reputation score between 1 and 10.

741

742 Raises:

743 ValueError: If the LLM response cannot be parsed as a score.

744 """

745 logger.info(f"Tier 4: LLM analysis for journal '{journal_name}'...")

746

747 # Single SearXNG search for journal info. Serialize access to

748 # the shared SearXNG engine to prevent two threads from

749 # clobbering its instance state (_last_results_count,

750 # _search_results, rate tracker).

751 query = f'"{journal_name}" academic journal impact factor quartile'

752 with self.__engine_lock:

753 results = self.__engine.run(query)

754

755 # Extract snippets from search results

756 snippets = []

757 for r in results[:10]:

758 snippet = r.get("snippet", "") or r.get("content", "")

759 if snippet: 759 ↛ 757line 759 didn't jump to line 757 because the condition on line 759 was always true

760 snippets.append(snippet)

761 journal_info_text = "\n".join(snippets)

762

763 if not journal_info_text: 763 ↛ 764line 763 didn't jump to line 764 because the condition on line 763 was never true

764 logger.warning(

765 f"No SearXNG results for '{journal_name}' — "

766 f"cannot score via Tier 4"

767 )

768 raise ValueError(f"No search results for journal '{journal_name}'")

769

770 # Truncate to fit context

771 if len(journal_info_text) > self.__max_context: 771 ↛ 772line 771 didn't jump to line 772 because the condition on line 771 was never true

772 journal_info_text = journal_info_text[: self.__max_context] + "..."

773

774 # Single LLM call to score. Wording mirrors the long-standing

775 # original prompt — earlier code review flagged that arbitrary

776 # rewrites of this prompt have a real chance of regressing the

777 # Q1/Q2/Q3 calibration the rest of the code depends on.

778 prompt = f"""

779You are a research assistant helping to assess the reliability and

780reputability of scientific journals. A reputable journal should be

781peer-reviewed, not predatory, and high-impact. Please review the

782following information on the journal "{journal_name}" and output a

783reputability score between 1 and 10, where 1-3 is not reputable and

784probably predatory, 4-6 is reputable but low-impact (Q2 or Q3),

785and 7-10 is reputable Q1 journals. Only output the number, do not

786provide any explanation or other output.

787

788JOURNAL INFORMATION:

789

790{journal_info_text}

791"""

792

793 response_text = get_llm_response_text(self.model.invoke(prompt))

794 logger.debug(

795 f"Tier 4 LLM response for '{journal_name}': {response_text}"

796 )

797

798 match = re.search(r"\d+", response_text.strip())

799 if match is None:

800 logger.warning(

801 f"Failed to parse score from LLM response for "

802 f"'{journal_name}': {response_text!r}"

803 )

804 raise ValueError(

805 "Failed to parse reputation score from LLM response."

806 )

807

808 reputation_score = int(match.group())

809 if reputation_score not in VALID_QUALITY_SCORES:

810 # Scoring tiers emit {1,4,5,6,7,8,10}; LLM returning anything

811 # else is almost certainly prompt drift. Treat as a parse

812 # failure so the existing failure counter + circuit breaker

813 # observe it, rather than snapping to the nearest bucket and

814 # silently masking the problem.

815 logger.warning(

816 f"LLM returned out-of-set score {reputation_score} for "

817 f"'{journal_name}' (expected one of "

818 f"{sorted(VALID_QUALITY_SCORES)}); treating as parse failure"

819 )

820 raise ValueError(

821 f"LLM returned out-of-set score {reputation_score}."

822 )

823 return reputation_score

824

825 # ------------------------------------------------------------------

826 # Database operations

827 # ------------------------------------------------------------------

828

829 def __save_llm_score_to_db(self, *, name: str, quality: int) -> None:

830 """Cache a Tier 4 LLM score for future research runs.

831

832 Only Tier 4 (LLM) results are cached — Tiers 1–3 read directly

833 from the read-only reference DB on every scoring pass. The

834 lookup predicate filters on ``score_source == "llm"`` and the

835 current ``quality_model`` so stale scores from a superseded LLM

836 don't get served.

837

838 No-op during the preview filter phase (no thread context).

839 """

840 session_ctx = self.__db_session()

841 if session_ctx is None: 841 ↛ 843line 841 didn't jump to line 843 because the condition on line 841 was always true

842 return

843 try:

844 self._save_journal_to_db_inner(

845 session_ctx, name=name, quality=quality

846 )

847 except Exception:

848 # Score is still valid and returned to the caller — only

849 # the cache write failed.

850 logger.exception(

851 f"Failed to cache LLM score for '{name}' — "

852 f"score is still returned but won't be cached."

853 )

854

855 def _save_journal_to_db_inner(

856 self, session_ctx, *, name: str, quality: int

857 ) -> None:

858 """Race-safe upsert of the Tier 4 LLM cache row.

859

860 Mirrors the Paper upsert pattern: select-then-insert in a

861 savepoint, and on IntegrityError (a concurrent writer created

862 the row first) roll back the savepoint and re-fetch. This

863 prevents the pre-fix bug where two concurrent scorings of the

864 same journal collided on the UNIQUE(name) constraint and the

865 exception handler left the cache empty.

866 """

867 from sqlalchemy.exc import IntegrityError

868

869 now = int(time.time())

870 model_id = get_model_identifier(self.model)

871

872 with session_ctx as db_session:

873 journal = db_session.query(Journal).filter_by(name=name).first()

874 if journal is not None:

875 journal.quality = quality

876 journal.score_source = "llm"

877 journal.quality_analysis_time = now

878 journal.quality_model = model_id

879 # name_lower MUST go through normalize_name, not bare

880 # .lower(). Bare lowercase leaves U+2122 (TM), ligatures,

881 # fullwidth letters intact; NFKC collapses them. The

882 # migration backfill (0006:257) uses the same normalization;

883 # divergence produces silent cache misses and, on migration,

884 # UNIQUE violations that abort the upgrade.

885 journal.name_lower = normalize_name(name)

886 try:

887 db_session.commit()

888 except Exception:

889 db_session.rollback()

890 logger.warning(

891 f"Failed to update cached LLM score for '{name}'"

892 )

893 return

894

895 sp = db_session.begin_nested()

896 try:

897 journal = Journal(

898 name=name,

899 name_lower=normalize_name(name),

900 quality=quality,

901 score_source="llm",

902 quality_model=model_id,

903 quality_analysis_time=now,

904 )

905 db_session.add(journal)

906 db_session.flush()

907 sp.commit()

908 db_session.commit()

909 return

910 except IntegrityError:

911 sp.rollback()

912

913 # Competing writer inserted first; re-fetch and update.

914 journal = db_session.query(Journal).filter_by(name=name).first()

915 if journal is None:

916 # Genuinely unexpected — UNIQUE violation with no row.

917 logger.warning(

918 f"IntegrityError on Journal '{name}' insert but "

919 f"row not found on re-fetch; skipping cache write."

920 )

921 return

922 journal.quality = quality

923 journal.score_source = "llm"

924 journal.quality_analysis_time = now

925 journal.quality_model = model_id

926 journal.name_lower = normalize_name(name)

927 try:

928 db_session.commit()

929 except Exception:

930 db_session.rollback()

931 logger.warning(f"Failed to merge cached LLM score for '{name}'")

932

933 # ------------------------------------------------------------------

934 # Tiered scoring for a single journal

935 # ------------------------------------------------------------------

936

937 def __score_journal(

938 self, journal_name: str, result: Dict[str, Any]

939 ) -> tuple[int | None, str | None]:

940 """

941 Score a journal using the tiered approach.

942

943 Returns ``(score, source_tag)``:

944 - ``score`` is the 1-10 quality value, or ``None`` if the

945 journal is predatory (signal to auto-remove).

946 - ``source_tag`` identifies which tier produced the score:

947 ``"openalex"``, ``"doaj"``, ``"institution"``, ``"llm"``

948 (Tier 4 live scoring OR cache hit on a prior LLM row),

949 ``"conference"``, or ``"low_confidence"`` (no tier matched).

950 ``None`` when predatory.

951

952 The tag is attached to the result dict for rendering. Nothing

953 is frozen on the Paper row; the dashboard resolves current

954 quality live from ``journals.quality`` (Tier 4) or the bundled

955 reference DB (Tier 1-3) so a re-scored journal propagates

956 automatically.

957 """

958 dm = self.__data_manager

959

960 # Extract IDs from result for richer lookups

961 issn = result.get("issn")

962 openalex_sid = result.get("openalex_source_id")

963 publisher = result.get("publisher")

964

965 # --- Tier 1: Predatory check ---

966 is_pred, pred_source = dm.is_predatory(

967 journal_name=journal_name,

968 publisher_name=publisher,

969 )

970 if is_pred:

971 # Check whitelist override (avoids false positives)

972 if dm.is_whitelisted(issn=issn, name=journal_name):

973 logger.debug(

974 f"Tier 1: '{journal_name}' is on predatory list "

975 f"({pred_source}) but whitelisted — not removing"

976 )

977 else:

978 logger.warning(

979 f"Tier 1: PREDATORY — removing results from "

980 f"'{journal_name}' (source: {pred_source})"

981 )

982 return None, None # Signal auto-remove

983

984 # --- Tier 2: OpenAlex snapshot ---

985 oa_entry = dm.lookup_openalex(

986 source_id=openalex_sid, issn=issn, name=journal_name

987 )

988 if oa_entry:

989 h_idx = oa_entry.get("h_index")

990 oa_doaj = oa_entry.get("is_in_doaj", False)

991 oa_type = oa_entry.get("type", "journal")

992 oa_quartile = oa_entry.get("quartile")

993 score = dm.derive_quality_score(

994 h_index=h_idx,

995 quartile=oa_quartile,

996 is_in_doaj=oa_doaj,

997 source_type=oa_type,

998 )

999 if score is not None: 999 ↛ 1029line 999 didn't jump to line 1029 because the condition on line 999 was always true

1000 logger.debug(

1001 f"Tier 2 (OpenAlex): '{journal_name}' → "

1002 f"score {score}/10 "

1003 f"(quartile: {oa_quartile or '—'}, h-index: {h_idx})"

1004 )

1005 # Tier 2 results are NOT cached to the user DB — the

1006 # read-only reference DB is already a 100–300µs lookup,

1007 # so a second-level cache adds no value. Only Tier 4

1008 # (LLM) results are cached, below.

1009 # Preprint repositories (arxiv, biorxiv, ssrn, ...) get a

1010 # low Tier-2 floor because they aren't peer-reviewed. If

1011 # the authors are at a strong institution, lift the score

1012 # via the institution tier — taking max so a real venue

1013 # match (≥6) is never demoted. Only applies to repository

1014 # source types and only when score is weak (≤5).

1015 if oa_type == "repository" and score <= 5: 1015 ↛ 1016line 1015 didn't jump to line 1016 because the condition on line 1015 was never true

1016 affs = result.get("affiliations")

1017 if affs:

1018 inst = dm.score_from_affiliations(affs)

1019 if inst is not None and inst > score:

1020 logger.debug(

1021 f"Tier 2+3.5 (preprint lift): "

1022 f"'{journal_name}' {score} → {inst} via "

1023 f"institutions: {_format_affiliations(affs)}"

1024 )

1025 return inst, "institution"

1026 return score, "openalex"

1027

1028 # --- Tier 3: DOAJ ---

1029 if issn:

1030 doaj_entry = dm.lookup_doaj(issn=issn)

1031 if doaj_entry:

1032 score = dm.derive_quality_score(is_in_doaj=True)

1033 logger.debug(

1034 f"Tier 3 (DOAJ): '{journal_name}' → score {score}/10"

1035 )

1036 return score, "doaj"

1037

1038 # --- Tier 3.5: Institution lookup ---

1039 # When the venue tiers couldn't score the paper, fall back to

1040 # author affiliations. This is the *only* trust signal we have

1041 # for preprints with no journal_ref or for venues OpenAlex

1042 # doesn't index. Score is capped at 6 inside score_from_affiliations

1043 # so institution alone never beats a real venue match.

1044 affiliations = result.get("affiliations")

1045 if affiliations: 1045 ↛ 1046line 1045 didn't jump to line 1046 because the condition on line 1045 was never true

1046 inst_score = dm.score_from_affiliations(affiliations)

1047 if inst_score is not None:

1048 logger.debug(

1049 f"Tier 3.5 (Institution): '{journal_name}' → "

1050 f"score {inst_score}/10 from institutions: "

1051 f"{_format_affiliations(affiliations)}"

1052 )

1053 return inst_score, "institution"

1054

1055 # --- DB cache: check for cached LLM results before expensive tiers ---

1056 # Tiers 1-3 use bundled data (instant, no caching needed).

1057 # Only Tier 4 (LLM) results are expensive and worth caching.

1058 # The predicate filters on quality_model so scores from a

1059 # superseded LLM model miss the cache and re-score, and

1060 # `cached.quality` is validated against VALID_QUALITY_SCORES to

1061 # evict any pre-fix rows that stored an out-of-set value.

1062 session_ctx = self.__db_session()

1063 if session_ctx is not None:

1064 try:

1065 with session_ctx as session:

1066 cached = (

1067 session.query(Journal)

1068 .filter_by(name=journal_name)

1069 .filter(Journal.score_source == "llm")

1070 .filter(

1071 Journal.quality_model

1072 == get_model_identifier(self.model)

1073 )

1074 .first()

1075 )

1076 if cached is not None: 1076 ↛ 1111line 1076 didn't jump to line 1111

1077 is_fresh = (

1078 time.time() - cached.quality_analysis_time

1079 ) < self.__quality_reanalysis_period.total_seconds()

1080 is_valid = cached.quality in VALID_QUALITY_SCORES

1081

1082 if is_fresh and is_valid:

1083 logger.info(

1084 f"DB cache hit: '{journal_name}' → "

1085 f"score {cached.quality}/10 [cached LLM]"

1086 )

1087 # Cache hit on a Journal row written by the

1088 # LLM path — tag as "llm" so the dashboard

1089 # surfaces it as a Tier 4 verdict.

1090 return cached.quality, "llm"

1091 if is_fresh and not is_valid: 1091 ↛ 1111line 1091 didn't jump to line 1111

1092 logger.warning(

1093 f"Cached score {cached.quality} for "

1094 f"'{journal_name}' not in valid set "

1095 f"{sorted(VALID_QUALITY_SCORES)}; rescoring"

1096 )

1097 # Expired or invalid — fall through to re-evaluate

1098 except Exception:

1099 logger.exception(

1100 f"DB cache read failed for '{journal_name}', "

1101 f"continuing with LLM tiers"

1102 )

1103

1104 # --- Tier 4: LLM analysis (last resort) ---

1105 # Off by default — bundled data covers 217K+ sources and this tier

1106 # adds significant latency (1 SearXNG search + 1 LLM call per

1107 # unknown journal). Users opt in via the

1108 # `search.journal_reputation.enable_llm_scoring` setting. The

1109 # __searxng_available and consecutive-failures checks below remain

1110 # as runtime safety nets even when the user enabled it.

1111 from ...config.search_config import get_setting_from_snapshot

1112

1113 _enable_tier4 = bool(

1114 get_setting_from_snapshot(

1115 "search.journal_reputation.enable_llm_scoring",

1116 False,

1117 settings_snapshot=self.__settings_snapshot,

1118 )

1119 )

1120

1121 # Tier 3.6: LLM-based name cleanup salvage. Gated behind the same

1122 # opt-in flag as Tier 4. Asks the LLM to canonicalise the name

1123 # (handles abbreviations and locations the regex can't), then

1124 # retries the cheap bundled tiers. This costs one extra LLM call

1125 # per unknown journal but can avoid Tier 4's full SearXNG search.

1126 if _enable_tier4:

1127 relabeled = self.__llm_clean_journal_name(journal_name)

1128 if relabeled and relabeled != journal_name: 1128 ↛ 1157line 1128 didn't jump to line 1157 because the condition on line 1128 was always true

1129 logger.debug(

1130 f"Tier 3.6 (LLM cleanup): '{journal_name}' → "

1131 f"'{relabeled}', retrying bundled tiers"

1132 )

1133 oa_retry = dm.lookup_openalex(name=relabeled)

1134 if oa_retry:

1135 h_idx = oa_retry.get("h_index")

1136 oa_doaj = oa_retry.get("is_in_doaj", False)

1137 score = dm.derive_quality_score(

1138 h_index=h_idx,

1139 quartile=oa_retry.get("quartile"),

1140 is_in_doaj=oa_doaj,

1141 source_type=oa_retry.get("type", "journal"),

1142 )

1143 if score is not None: 1143 ↛ 1157line 1143 didn't jump to line 1157 because the condition on line 1143 was always true

1144 logger.info(

1145 f"Tier 3.6 (LLM cleanup → OpenAlex): "

1146 f"'{journal_name}' (as '{relabeled}') → "

1147 f"score {score}/10"

1148 )

1149 # Tier 3.6 is a Tier 2 retry under a cleaned

1150 # name; the result is effectively Tier 2 data

1151 # and is NOT cached in the user DB (reference

1152 # DB lookups are already instant). Tagged

1153 # "openalex" — only the NAME came from the LLM,

1154 # not the score.

1155 return score, "openalex"

1156

1157 if (

1158 _enable_tier4

1159 and self.__searxng_available

1160 and self.__searxng_failures() < 2

1161 ):

1162 try:

1163 quality = self.__analyze_journal_reputation(journal_name)

1164 self.__reset_searxng_failures()

1165 # There used to be a +1 "DOAJ Seal bonus" here. DOAJ

1166 # retired the Seal in April 2025, so the bonus could

1167 # only ever fire on stale pre-2025 data and was removed.

1168 self.__save_llm_score_to_db(name=journal_name, quality=quality)

1169 logger.debug(

1170 f"Tier 4 (LLM): '{journal_name}' → "

1171 f"score {quality}/10 "

1172 f"[via SearXNG + LLM analysis]"

1173 )

1174 return quality, "llm"

1175 except Exception:

1176 failures = self.__bump_searxng_failures()

1177 logger.exception(

1178 f"Tier 4 failed for '{journal_name}'. "

1179 f"Consecutive failures: {failures}"

1180 )

1181 if failures >= 2:

1182 logger.warning(

1183 "Tier 4 disabled for remaining journals in "

1184 "this batch (2 consecutive failures)."

1185 )

1186

1187 # --- Conference heuristic (for papers without DOI or OpenAlex match) ---

1188 # Guard: many high-tier journals start with "Proceedings of …"

1189 # (PNAS, Royal Society A/B, AMS, LMS, …). The bare `proceedings`

1190 # token in `_CONFERENCE_PATTERNS` would otherwise classify them

1191 # as Q3 conferences and throw away their real h-index. Skip the

1192 # heuristic for these — they fall through to the unknown-journal

1193 # score (3) and the user's threshold decides what to do.

1194 if journal_name.lower().lstrip().startswith("proceedings of "): 1194 ↛ 1195line 1194 didn't jump to line 1195 because the condition on line 1194 was never true

1195 logger.debug(

1196 f"Conference heuristic: skipped for '{journal_name}' "

1197 f"(starts with 'Proceedings of' — likely a journal, "

1198 f"not a conference)"

1199 )

1200 elif _is_likely_conference(journal_name): 1200 ↛ 1201line 1200 didn't jump to line 1201 because the condition on line 1200 was never true

1201 score = dm.derive_quality_score(source_type="conference")

1202 logger.debug(

1203 f"Conference heuristic: '{journal_name}' → "

1204 f"score {score}/10 (detected as conference by name pattern)"

1205 )

1206 return score, "conference"

1207

1208 # No tier could score this journal — neither OpenAlex/DOAJ

1209 # venue match nor Tier 3.5 institution salvage produced a

1210 # signal. Score it as low-confidence (3) so the default

1211 # threshold (4) actually filters it out. Distinct from

1212 # predatory (1) — these are merely unknown, not blacklisted.

1213 affs_for_log = result.get("affiliations") or []

1214 logger.debug(

1215 f"No scoring data for '{journal_name}' — flagging as "

1216 f"low-confidence (score 3); tried institutions: "

1217 f"{_format_affiliations(affs_for_log)}"

1218 )

1219 return 3, "low_confidence"

1220

1221 # ------------------------------------------------------------------

1222 # Main filter entry point

1223 # ------------------------------------------------------------------

1224

1225 def filter_results(

1226 self, results: List[Dict], query: str, **kwargs

1227 ) -> List[Dict]:

1228 """Filter results by journal quality, with deduplication."""

1229 logger.info(

1230 f"Journal filter: processing {len(results)} results "

1231 f"(threshold={self.__threshold})"

1232 )

1233 # Fail-soft during a fresh install: if the reference DB file

1234 # isn't on disk yet, don't score anything. Every journal would

1235 # otherwise fall through to the "no scoring data" branch and

1236 # be marked as score 3, which is semantically wrong — we don't

1237 # know the journal is unknown, we just haven't loaded the data

1238 # yet. Tag each result with the QUALITY_PENDING sentinel so the

1239 # report renderer can show the user a helpful note pointing

1240 # them at /metrics/journals.

1241 #

1242 # We probe the DB file path directly rather than calling

1243 # ``data_manager.available`` — the latter has side effects

1244 # (triggers `_ensure_engine`, which tries to lazy-build the DB

1245 # and can block on an in-flight download for several minutes).

1246 # A cached engine on the data manager means either the file

1247 # was successfully opened earlier, or a test fixture has

1248 # injected an in-memory DB — either way we're good to score.

1249 try:

1250 from ...config.paths import get_journal_data_directory

1251

1252 db_file = get_journal_data_directory() / "journal_quality.db"

1253 on_disk = db_file.exists() and db_file.stat().st_size > 0

1254 engine_cached = (

1255 getattr(self.__data_manager, "_engine", None) is not None

1256 )

1257 db_ready = on_disk or engine_cached

1258 logger.info(

1259 f"Journal filter: db_ready={db_ready} "

1260 f"(on_disk={on_disk}, engine_cached={engine_cached})"

1261 )

1262 except Exception:

1263 logger.exception(

1264 "Journal filter: db-ready probe raised; "

1265 "assuming DB not ready (pending)."

1266 )

1267 db_ready = False

1268 if not db_ready: 1268 ↛ 1269line 1268 didn't jump to line 1269 because the condition on line 1268 was never true

1269 from ...utilities.search_utilities import QUALITY_PENDING

1270

1271 # Fire-and-forget the download in a daemon thread so the

1272 # pending-marker copy ("by the time you check, it may

1273 # already be done") is actually true. Without this the

1274 # filter would just tag every search "pending" forever

1275 # and nobody would ever fetch the data unless the user

1276 # clicked Download manually. The 30-second TTL cache in

1277 # ensure_journal_data prevents multiple concurrent filter

1278 # workers from all racing to spawn the same download.

1279 #

1280 # Egress policy: skip the background fetch under

1281 # PRIVATE_ONLY or STRICT — those scopes shouldn't egress at

1282 # all, and the journal data sources (OpenAlex / DOAJ /

1283 # JabRef bulk downloads) are all public hosts.

1284 if self._should_skip_journal_fetch_for_scope():

1285 logger.bind(policy_audit=True).info(

1286 "journal-data background fetch skipped: egress scope "

1287 "forbids public fetches"

1288 )

1289 else:

1290 _start_background_journal_fetch()

1291

1292 # Respect exclude_non_published — results without a venue

1293 # still get dropped in pending mode, same as in the full

1294 # scoring path. Only venued results carry the marker.

1295 out = []

1296 tagged = 0

1297 dropped = 0

1298 for r in results:

1299 if r.get("journal_ref"):

1300 r.setdefault("journal_quality", QUALITY_PENDING)

1301 out.append(r)

1302 tagged += 1

1303 elif not self.__exclude_non_published:

1304 out.append(r)

1305 else:

1306 dropped += 1

1307 logger.warning(

1308 f"Journal filter: reference DB not yet built — "

1309 f"tagged {tagged} result(s) with QUALITY_PENDING, "

1310 f"kept {len(out) - tagged} venueless, "

1311 f"dropped {dropped} (exclude_non_published). "

1312 f"Background download triggered if not already running."

1313 )

1314 return out

1315

1316 # Initialize `filtered` outside the try so the predatory-safe

1317 # fallback in the except handler can always reference it even

1318 # if the crash happens before Pass-1 populates anything.

1319 filtered: list = []

1320

1321 try:

1322 # Reset the per-thread fail-fast counter for each batch.

1323 # The counter lives in `threading.local()` so concurrent

1324 # callers on the same filter instance don't clobber each

1325 # other (see Bug A3).

1326 self.__reset_searxng_failures()

1327

1328 # Pass 1: collect the richest metadata per journal (the result

1329 # with ISSN/source_id) so scoring uses the best available data.

1330 journal_best_result: Dict[str, Dict] = {}

1331 results_with_journals: list[tuple[Dict, str]] = []

1332

1333 def _handle_no_venue(result: Dict) -> None:

1334 """Institution-salvage then exclude_non_published policy.

1335

1336 If Tier 3.5 salvages a score from author affiliations,

1337 the result carries that numeric score. Otherwise we

1338 tag it with ``QUALITY_PREPRINT`` so the report renderer

1339 can show a "preprint — not in journal catalog" label

1340 instead of leaving the quality column ambiguously blank.

1341 """

1342 from ...utilities.search_utilities import QUALITY_PREPRINT

1343

1344 affs = result.get("affiliations")

1345 if affs: 1345 ↛ 1346line 1345 didn't jump to line 1346 because the condition on line 1345 was never true

1346 inst_score = self.__data_manager.score_from_affiliations(

1347 affs

1348 )

1349 if (

1350 inst_score is not None

1351 and inst_score >= self.__threshold

1352 ):

1353 result["journal_quality"] = inst_score

1354 logger.debug(

1355 f"Tier 3.5 (Institution, no venue): "

1356 f"'{result.get('title', '')[:60]}' → "

1357 f"score {inst_score}/10 from institutions: "

1358 f"{_format_affiliations(affs)}"

1359 )

1360 filtered.append(result)

1361 return

1362 if not self.__exclude_non_published:

1363 result.setdefault("journal_quality", QUALITY_PREPRINT)

1364 filtered.append(result)

1365

1366 # Per-batch cache for journal name cleaning — avoids redundant

1367 # regex + abbreviation DB lookups when multiple results come

1368 # from the same raw journal_ref string (common in OpenAlex

1369 # result batches).

1370 _name_cache: Dict[str, str] = {}

1371

1372 for result in results:

1373 journal_ref = result.get("journal_ref")

1374 # Strip whitespace-only refs — " " is truthy but has no

1375 # meaningful content for the filter.

1376 if isinstance(journal_ref, str):

1377 journal_ref = journal_ref.strip()

1378 if not journal_ref:

1379 _handle_no_venue(result)

1380 continue

1381

1382 # Use per-batch cache to skip redundant cleaning for

1383 # repeated raw journal_ref values in the same batch.

1384 clean_name = _name_cache.get(journal_ref)

1385 if clean_name is None: 1385 ↛ 1391line 1385 didn't jump to line 1391 because the condition on line 1385 was always true

1386 clean_name = self.__clean_journal_name(journal_ref)

1387 _name_cache[journal_ref] = clean_name

1388 # Cleanup can reduce a volume/page-only ref to "". Treat

1389 # that the same as "no venue" rather than bucketing all

1390 # affected results under a degenerate empty-string key.

1391 if not clean_name: 1391 ↛ 1392line 1391 didn't jump to line 1392 because the condition on line 1391 was never true

1392 _handle_no_venue(result)

1393 continue

1394

1395 results_with_journals.append((result, clean_name))

1396

1397 if clean_name not in journal_best_result:

1398 journal_best_result[clean_name] = result

1399 else:

1400 prev = journal_best_result[clean_name]

1401 if ( 1401 ↛ 1407line 1401 didn't jump to line 1407 because the condition on line 1401 was never true

1402 not prev.get("issn")

1403 and not prev.get("openalex_source_id")

1404 ) and (

1405 result.get("issn") or result.get("openalex_source_id")

1406 ):

1407 journal_best_result[clean_name] = result

1408

1409 # Pass 2: score each unique journal once, then filter

1410 journal_scores: Dict[str, tuple[int | None, str | None]] = {}

1411

1412 for result, clean_name in results_with_journals:

1413 if clean_name not in journal_scores:

1414 journal_scores[clean_name] = self.__score_journal(

1415 clean_name, journal_best_result[clean_name]

1416 )

1417

1418 score, source_tag = journal_scores[clean_name]

1419

1420 if score is None:

1421 # Predatory → auto-remove. Include original journal_ref

1422 # and URL in the log so false-positive reports can be

1423 # debugged without re-running the query.

1424 logger.warning(

1425 f"Auto-removed predatory: "

1426 f"title='{result.get('title', '')[:80]}' "

1427 f"journal_ref='{result.get('journal_ref', '')!r}' "

1428 f"cleaned='{clean_name}' "

1429 f"url={result.get('link') or result.get('url') or '—'}"

1430 )

1431 continue

1432

1433 if score >= self.__threshold:

1434 result["journal_quality"] = score

1435 # Cleaned name that keyed the successful score —

1436 # persisted on Paper.container_title so the dashboard

1437 # can GROUP BY it and enrich from the reference DB.

1438 result["journal_name_matched"] = clean_name

1439 # Source tag — propagates to the rendered quality

1440 # tag in the research output. Nothing is frozen on

1441 # the Paper row; the dashboard resolves current

1442 # quality live from journals.quality (Tier 4) or

1443 # the bundled reference DB (Tier 1-3).

1444 result["journal_quality_source"] = source_tag

1445 filtered.append(result)

1446

1447 predatory_count = sum(

1448 1 for s, _ in journal_scores.values() if s is None

1449 )

1450 passed_count = sum(

1451 1

1452 for s, _ in journal_scores.values()

1453 if s is not None and s >= self.__threshold

1454 )

1455 below_count = sum(

1456 1

1457 for s, _ in journal_scores.values()

1458 if s is not None and s < self.__threshold

1459 )

1460 logger.info(

1461 f"Journal quality filter: {len(results)} → "

1462 f"{len(filtered)} results | "

1463 f"{len(journal_scores)} unique journals scored | "

1464 f"{passed_count} passed, {below_count} below threshold, "

1465 f"{predatory_count} predatory removed"

1466 )

1467 return filtered

1468

1469 except Exception:

1470 # Safety net: a filter crash should not kill the entire search,

1471 # but it MUST NOT re-admit predatory journals either.

1472 # `filtered` is predatory-free by construction (the pass-2 loop

1473 # `continue`s on predatory scores). Returning `results` — the

1474 # original unfiltered list — would leak predatory sources that

1475 # Tier 1 had already caught. Prefer losing in-flight non-

1476 # predatory results over breaking the predatory-removal

1477 # safety contract. Logged at ERROR so the root cause surfaces.

1478 logger.exception(

1479 "Journal quality filtering failed — returning partial "

1480 "(predatory-free) results. This is a bug that should be "

1481 "investigated."

1482 )

1483 return filtered

Coverage for src/local_deep_research/advanced_search_system/filters/journal_reputation_filter.py: 69%

461 statements