"""
Journal reputation filter with tiered quality scoring.

Scores journals 1-10 and filters academic search results by quality.
Uses bundled bibliometric data for most journals; LLM analysis is an
opt-in last resort.  Predatory journals are auto-removed.

Scoring tiers (tried in order, first match wins):

  1. Predatory check — auto-removes blacklisted journals/publishers
                       (whitelist override prevents false positives)
  2. OpenAlex        — h-index, quartile, DOAJ from ~217K bundled sources;
                       preprint repos can be lifted via institution affiliations
  3. DOAJ            — quality floor (5) for listed OA journals
  3.5 Institutions   — author affiliation lookup when no venue matched
                       (capped at 6, never beats a real venue)

  --- DB cache check (only for cached LLM results from previous runs) ---

  3.6 LLM cleanup    — LLM canonicalises the name, then retries Tier 2
                       (opt-in via ``enable_llm_scoring``)
  4. LLM analysis    — SearXNG web search + LLM scoring (opt-in, expensive);
                       disabled after 2 consecutive failures
  Conference         — name-pattern heuristic for unmatched conferences

Unknown journals that no tier can score receive a low-confidence score (3).
"""

import re
import threading
import time
import unicodedata
from datetime import timedelta
from typing import Any, Dict, List, Optional

from langchain_core.language_models.chat_models import BaseChatModel
from loguru import logger
from sqlalchemy.orm import Session

from ...utilities.type_utils import unwrap_setting
from ...config.llm_config import get_llm
from ...constants import VALID_QUALITY_SCORES
from ...database.models import Journal
from ...database.session_context import get_user_db_session

# normalize_name applies NFKC + lower + strip — must match the
# migration backfill/dedupe expressions in 0006_journal_quality_system.py
# and the reference DB builder in journal_quality/db.py so name_lower
# is single-valued across every writer.
from ...journal_quality.db import get_db as get_journal_data_manager
from ...journal_quality.scoring import normalize_name
from ...security.egress.policy import PolicyDeniedError
from ...security.log_sanitizer import strip_control_chars
from ...utilities.llm_utils import get_model_identifier
from ...utilities.resource_utils import safe_close
from ...utilities.thread_context import get_search_context
from ...web_search_engines.search_engine_factory import create_search_engine
from .base_filter import BaseFilter
from ...constants import DEFAULT_SEARCH_TOOL


# Patterns that indicate a venue is a conference, not a journal.
# Used as a fallback when DOI enrichment and OpenAlex lookup both miss.
_CONFERENCE_PATTERNS = [
    re.compile(r"\b(?:proceedings|proc\.)\b", re.I),
    re.compile(r"\b(?:conference|conf\.)\b", re.I),
    re.compile(r"\b(?:symposium|symp\.)\b", re.I),
    re.compile(r"\bworkshop\b", re.I),
    re.compile(
        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"
    ),
]


def _is_likely_conference(name: str) -> bool:
    """Detect if a venue name is likely a conference based on common patterns."""
    return any(p.search(name) for p in _CONFERENCE_PATTERNS)


def _sanitize_name(name: str) -> str:
    """Sanitize a journal name for safe use in logs and LLM prompts.

    Args:
        name: Raw journal name string, potentially containing control
            characters, excessive length, or quotes.

    Returns:
        Sanitized string safe for use in logs and LLM prompts.
    """
    # Strip control + format characters. Covers C0/C1 (log injection),
    # bidi overrides (U+202A-E, U+2066-9), zero-width / joiner chars
    # (U+200B-F, U+2060-4, U+FEFF), Arabic letter mark, and digit-shape
    # controls — the comprehensive pattern audited in log_sanitizer,
    # not the narrow C0/C1-only regex we used to have here.
    name = strip_control_chars(name)
    # Normalize Unicode (prevents lookalike bypasses)
    name = unicodedata.normalize("NFKC", name)
    # Limit length (prevents resource exhaustion in prompts)
    if len(name) > 500:
        name = name[:500] + "..."
    # Strip quotes that could break prompt structure
    name = name.replace('"', "'")
    return name.strip()


def _format_affiliations(affiliations: list, max_n: int = 3) -> str:
    """Render an affiliation list as a compact human-readable string for
    log lines. Accepts the same shapes as ``score_from_affiliations``
    (plain strings or dicts with a ``name`` key) and truncates after
    ``max_n`` entries so a 20-author paper doesn't blow up the log.
    """
    if not affiliations:
        return "(none)"
    names: list[str] = []
    for aff in affiliations:
        if isinstance(aff, str):
            names.append(aff)
        elif isinstance(aff, dict):
            nm = aff.get("name") or aff.get("display_name")
            if nm:
                names.append(nm)
    if not names:
        return "(unknown)"
    shown = names[:max_n]
    suffix = "" if len(names) <= max_n else f" (+{len(names) - max_n} more)"
    return ", ".join(shown) + suffix


_bg_fetch_lock = threading.Lock()
_bg_fetch_thread: Optional[threading.Thread] = None


def _start_background_journal_fetch() -> None:
    """Kick off ``ensure_journal_data(auto_download=True)`` in a daemon
    thread the first time a search hits the pending path.

    The worker returns immediately if another thread is already in
    flight (``_bg_fetch_thread.is_alive()``) — so 30 concurrent filter
    workers can't each spawn their own download. The ``ensure_journal_data``
    TTL cache provides a second line of defence.

    Daemon thread so it doesn't block process exit.
    """
    global _bg_fetch_thread
    with _bg_fetch_lock:
        if _bg_fetch_thread is not None and _bg_fetch_thread.is_alive():
            logger.debug(
                "journal-data background fetch already in flight — "
                "not spawning a second thread"
            )
            return

        def _run():
            try:
                # Late import so this module doesn't pay the cost of
                # importing the downloader at its own import time.
                from ...journal_quality.downloader import (
                    ensure_journal_data,
                )

                logger.info(
                    "journal-data background fetch: starting "
                    "(triggered by filter pending path)"
                )
                ensure_journal_data(auto_download=True)
                logger.info("journal-data background fetch: done")
            except Exception:
                logger.exception(
                    "journal-data background fetch crashed — "
                    "next search will retry"
                )

        _bg_fetch_thread = threading.Thread(
            target=_run,
            name="journal-data-bg-fetch",
            daemon=True,
        )
        _bg_fetch_thread.start()


class JournalFilterError(Exception):
    """
    Custom exception for errors related to journal filtering.
    """


class JournalReputationFilter(BaseFilter):
    """
    A filter for academic results that considers the reputation of journals.

    Uses a tiered scoring approach: bundled data (OpenAlex, DOAJ, predatory
    lists) for most journals, with LLM-based analysis via SearXNG as a
    fallback for truly unknown journals.

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

    def __init__(
        self,
        model: BaseChatModel | None = None,
        reliability_threshold: int | None = None,
        max_context: int | None = None,
        exclude_non_published: bool | None = None,
        quality_reanalysis_period: timedelta | None = None,
        settings_snapshot: Dict[str, Any] | None = None,
    ):
        """Initialize the journal reputation filter.

        Args:
            model: The LLM model to use for Tier 4 analysis. If None,
                the default LLM from settings will be used.
            reliability_threshold: Minimum quality score (1-10) for a
                result to pass. Read from settings if not specified.
            max_context: Maximum characters of source content for LLM
                quality evaluation.
            exclude_non_published: If True, exclude results that don't
                have an associated journal publication reference.
            quality_reanalysis_period: Period after which cached journal
                quality assessments are refreshed.
            settings_snapshot: Settings snapshot for thread context.
        """
        super().__init__(model)

        self._owns_llm = self.model is None
        if self.model is None:
            # Forward the snapshot so the LLM PEP at llm_config.py:295
            # can evaluate ``llm.require_local_endpoint`` / egress scope.
            # Dropping it here previously bypassed the PEP entirely on
            # search engines that constructed this filter with no llm.
            self.model = get_llm(settings_snapshot=settings_snapshot)

        # Import here to avoid circular import
        from ...config.search_config import get_setting_from_snapshot

        self.__threshold = reliability_threshold
        if self.__threshold is None:
            self.__threshold = int(
                get_setting_from_snapshot(
                    "search.journal_reputation.threshold",
                    2,
                    settings_snapshot=settings_snapshot,
                )
            )
        self.__max_context = max_context
        if self.__max_context is None:
            self.__max_context = int(
                get_setting_from_snapshot(
                    "search.journal_reputation.max_context",
                    3000,
                    settings_snapshot=settings_snapshot,
                )
            )
        self.__exclude_non_published = exclude_non_published
        if self.__exclude_non_published is None:
            self.__exclude_non_published = bool(
                get_setting_from_snapshot(
                    "search.journal_reputation.exclude_non_published",
                    False,
                    settings_snapshot=settings_snapshot,
                )
            )
        self.__quality_reanalysis_period = quality_reanalysis_period
        if self.__quality_reanalysis_period is None:
            self.__quality_reanalysis_period = timedelta(
                days=int(
                    get_setting_from_snapshot(
                        "search.journal_reputation.reanalysis_period",
                        365,
                        settings_snapshot=settings_snapshot,
                    )
                )
            )

        self.__settings_snapshot = settings_snapshot

        # SearXNG for Tier 4 (LLM fallback). Not strictly required anymore
        # since bundled data covers most journals.
        self.__engine = create_search_engine(
            "searxng", llm=self.model, settings_snapshot=settings_snapshot
        )
        self.__searxng_available = self.__engine is not None and getattr(
            self.__engine, "is_available", False
        )
        if not self.__searxng_available:
            logger.info(
                "SearXNG not available — Tier 4 (LLM analysis) disabled. "
                "Bundled data tiers still active."
            )

        # Fail-fast counter for SearXNG failures within a batch.
        # Stored in `threading.local()` so concurrent `filter_results`
        # calls on the same cached filter instance (the parallel search
        # engine reuses instances across worker threads — see
        # concurrent engine threads) can't clobber each other's
        # counter. Per-thread state is reset at the top of every
        # `filter_results` invocation.
        self.__tls = threading.local()

        # Lock serializing access to the shared SearXNG engine for
        # Tier 4. BaseSearchEngine instances keep mutable bookkeeping
        # state (_last_results_count, _search_results, rate-limit
        # tracker) on self; two concurrent .run() calls on the same
        # instance would clobber that state. Tier 4 is rarely hit in
        # practice (requires enable_llm_scoring=True + SearXNG +
        # bundled-data miss), so the lock's contention cost is
        # negligible compared to the correctness guarantee.
        self.__engine_lock = threading.Lock()

        # Journal data manager (loads bundled datasets lazily)
        self.__data_manager = get_journal_data_manager()

    # ------------------------------------------------------------------
    # Thread-local fail-fast counter accessors
    # ------------------------------------------------------------------

    def __searxng_failures(self) -> int:
        return getattr(self.__tls, "searxng_failures", 0)

    def __reset_searxng_failures(self) -> None:
        self.__tls.searxng_failures = 0

    def __bump_searxng_failures(self) -> int:
        n = self.__searxng_failures() + 1
        self.__tls.searxng_failures = n
        return n

    def close(self) -> None:
        """Close the SearXNG engine and LLM client."""
        if hasattr(self, "_JournalReputationFilter__engine"):
            # allow_none=True: SearXNG is optional (Tier 4 only), so
            # __engine is None in the common no-SearXNG configuration.
            safe_close(self.__engine, "SearXNG engine", allow_none=True)
        if self._owns_llm:
            safe_close(self.model, "journal filter LLM")

    def _should_skip_journal_fetch_for_scope(self) -> bool:
        """Return True when the active egress scope forbids public
        fetches (PRIVATE_ONLY / STRICT). The journal-data sources
        (OpenAlex, DOAJ, JabRef) are all public, so under these scopes
        the daemon thread would burn cycles and (worse) attempt egress
        the user explicitly opted out of.
        """
        snapshot = self.__settings_snapshot
        if not snapshot:
            return False
        try:
            from ...security.egress.policy import (
                EgressScope,
                PolicyDeniedError,
                context_from_snapshot,
            )

            primary_raw = unwrap_setting(
                snapshot.get("search.tool", DEFAULT_SEARCH_TOOL)
            )
            ctx = context_from_snapshot(
                snapshot, primary_raw or DEFAULT_SEARCH_TOOL
            )
            return ctx.scope in (EgressScope.PRIVATE_ONLY, EgressScope.STRICT)
        except PolicyDeniedError:
            # Corrupt/unknown scope value — we cannot certify that a public
            # journal fetch is permitted, so SKIP the fetch (fail closed).
            # This matches the hardened sibling
            # notifications.manager._filter_urls_by_egress_policy, which also
            # refuses rather than proceeds when the policy is unevaluable.
            logger.bind(policy_audit=True).warning(
                "journal fetch skipped: egress policy unevaluable "
                "(corrupt scope) — failing closed"
            )
            return True
        except Exception:
            # Other errors (missing key, snapshot shape) — fail open; the
            # per-URL runtime check still gates each individual fetch.
            return False

    @classmethod
    def create_default(
        cls,
        model: BaseChatModel | None = None,
        *,
        engine_name: str,
        settings_snapshot: Dict[str, Any] | None = None,
    ) -> Optional["JournalReputationFilter"]:
        """Initializes a default configuration of the filter based on settings.

        SearXNG is not required — the filter works with bundled data alone.
        SearXNG enables the optional Tier 4 (LLM analysis) for journals
        not found in the bundled datasets.

        Args:
            model: Optional LLM model for Tier 4 analysis.
            engine_name: Search engine configuration key (e.g. "arxiv").
            settings_snapshot: Optional frozen settings dict.

        Returns:
            A configured JournalReputationFilter, or None if filtering
            is disabled in settings for this engine.
        """
        from ...config.search_config import get_setting_from_snapshot

        try:
            enabled = get_setting_from_snapshot(
                f"search.engine.web.{engine_name}.journal_reputation.enabled",
                True,
                settings_snapshot=settings_snapshot,
            )
            logger.info(
                f"Journal filter create_default: engine={engine_name}, "
                f"enabled={enabled} (type={type(enabled).__name__})"
            )
            if not bool(enabled):
                logger.info(
                    f"Journal filter disabled for {engine_name} in settings"
                )
                return None

            filt = JournalReputationFilter(
                model=model, settings_snapshot=settings_snapshot
            )
            logger.info(
                f"Journal filter created for {engine_name} — "
                f"threshold={filt._JournalReputationFilter__threshold}"
            )
            return filt
        except PolicyDeniedError:
            # The egress policy refused the LLM the filter needs. This
            # must propagate so the user sees a clear policy refusal —
            # silently returning None would let unfiltered results
            # through under ``llm.require_local_endpoint=True``.
            raise
        except Exception:
            # Any other failure — settings read, filter init — returns
            # None rather than silently defaulting to enabled. A separate
            # silent ``except Exception: enabled = True`` branch used to
            # wrap only the settings read, which hid legitimate
            # configuration errors; per CLAUDE.md, fallbacks have to be
            # explicit, not default-on.
            logger.exception(
                f"Failed to configure journal filter for {engine_name}; "
                "results will not be journal-quality filtered for this engine"
            )
            return None

    @staticmethod
    def __db_session() -> Session | None:
        """Get a database session using the current search context credentials.

        Returns:
            SQLAlchemy Session context manager for the user's database, or
            ``None`` if no search context is available (e.g. when called
            from the preview filter phase, before per-user thread context
            has been propagated). Callers should treat ``None`` as
            "skip the DB operation".
        """
        context = get_search_context()
        if context is None:
            return None
        username = context.get("username")
        password = context.get("user_password")
        return get_user_db_session(username=username, password=password)

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

    def __clean_journal_name(self, journal_name: str) -> str:
        """Clean journal name to normalize for deduplication and lookup.

        Deterministic regex-based cleaning only: strips volume / page /
        year / month references, then tries a JabRef abbreviation
        expansion. This method does NOT call the LLM — it is the cheap,
        instant cleaning path that every Tier runs through first.

        The separate ``__llm_clean_journal_name`` method is the LLM
        fallback, invoked later only as a salvage step when the bundled
        data tiers all miss and ``enable_llm_scoring`` is on.
        Abbreviations not in the JabRef list and location suffixes
        ("ICML 2023, Honolulu") only get LLM-cleaned on that salvage
        path; this method returns them unchanged.

        Args:
            journal_name: Raw journal name from search results.

        Returns:
            Cleaned, normalized journal name.
        """
        # Sanitize first (strips control chars, normalizes Unicode)
        journal_name = _sanitize_name(journal_name)
        # Regex handles volume/page/year stripping (instant)
        cleaned = self.__regex_clean_journal_name(journal_name)

        # Try JabRef abbreviation expansion (deterministic, instant)
        expanded = self.__data_manager.expand_abbreviation(cleaned)
        if expanded:
            logger.debug(f"Abbreviation expanded: '{cleaned}' → '{expanded}'")
            return expanded

        if cleaned != journal_name:
            logger.debug(
                f"Regex-cleaned journal name: '{journal_name}' → '{cleaned}'"
            )
        return cleaned

    @staticmethod
    def __regex_clean_journal_name(name: str) -> str:
        """
        Fast regex-based journal name normalization. Strips volume, issue,
        page, year, and month references. No LLM needed.
        """
        months = (
            "january|february|march|april|may|june|july|"
            "august|september|october|november|december|"
            "jan|feb|mar|apr|jun|jul|aug|sep|oct|nov|dec"
        )

        # Strip leading/trailing whitespace
        name = name.strip()

        # Strip a leading [bracketed-original-language] prefix that
        # MEDLINE uses for non-English journals:
        #   "[Rinsho ketsueki] The Japanese journal of clinical hematology"
        # → "The Japanese journal of clinical hematology"
        name = re.sub(r"^\[[^\]]+\]\s*", "", name)

        # Strip trailing publisher suffixes that some search engines glue
        # onto the journal name (e.g. "Information Fusion Elsevier" or
        # "Cell Press"). Conservative — only the handful of well-known
        # academic publishers, anchored at end of string with a leading
        # space so we don't eat them mid-name.
        name = re.sub(
            r"\s+(?:Elsevier|Springer|Wiley|Nature\s+Publishing|"
            r"Cell\s+Press|MDPI|Sage|Taylor\s*&?\s*Francis|"
            r"Oxford\s+University\s+Press|Cambridge\s+University\s+Press|"
            r"IEEE|ACM|Routledge|Frontiers)\s*$",
            "",
            name,
            flags=re.IGNORECASE,
        )

        # Strip a leading 4-digit year ("2015 Plasma Phys. ..." → "Plasma Phys. ...")
        name = re.sub(r"^(?:19|20)\d{2}\s+", "", name)

        # Strip a leading ordinal volume marker, e.g.
        # "31st Conference on Neural Information Processing Systems" → "Conference on …"
        # Without this the OpenAlex name lookup fails on most conference
        # entries because the canonical name has no ordinal prefix.
        name = re.sub(
            r"^\d+(?:st|nd|rd|th)\s+",
            "",
            name,
            flags=re.IGNORECASE,
        )

        # Remove month+year references FIRST (before bare year strip),
        # so "September 2023" is consumed as a unit and we don't leave
        # the month behind as an orphan word.
        name = re.sub(
            rf",?\s*\b(?:{months})\b\.?\s+(?:19|20)?\d{{2,4}}",
            "",
            name,
            flags=re.IGNORECASE,
        )

        # Remove volume/issue/page refs: "Vol. 12", "Issue 3", "pp. 100-200"
        name = re.sub(
            r",?\s*(?:vol(?:ume)?\.?\s*\d+|"
            r"issue\s*\d+|"
            r"no\.?\s*\d+|"
            r"pp?\.?\s*\d+[\s–-]*\d*|"
            r"pages?\s*\d+[\s–-]*\d*)",
            "",
            name,
            flags=re.IGNORECASE,
        )
        # Remove volume(issue) patterns: "141(5)" — and bare "(15)" issues.
        name = re.sub(r",?\s*\d+\(\d+\)", "", name)
        name = re.sub(r"\s*\(\d+\)\s*", " ", name)
        # Remove year references: "(2023)", ", 2023". Anchored to 19xx/20xx
        # so 4-digit page numbers like ", 1063" in "106335" aren't eaten.
        name = re.sub(r"\s*[\(,]\s*(?:19|20)\d{2}\b\s*\)?", "", name)
        # Remove bare trailing citation data: ", 95, 146802" style
        # Only strips when there's a comma before the first number
        # (preserves "NeurIPS 2023" where space-number is part of the name)
        name = re.sub(r",\s*\d[\d,\s]*$", "", name)
        # Strip trailing alphanumeric volume markers: "E48", "R569", "L102"
        # (single uppercase letter followed by digits at end of string).
        name = re.sub(r"\s+[A-Z]\d+\s*$", "", name)
        # Strip trailing volume/page debris like "170 266-275", "71: 1-10",
        # "151:48-60", or a bare trailing volume "116". Repeated to peel
        # multiple chunks. Stops when only the journal name remains. We
        # require the run to start with whitespace or punctuation so we
        # don't eat the "2023" of "NeurIPS 2023" — the trailing-year regex
        # below handles that case.
        prev = None
        while prev != name:
            prev = name
            name = re.sub(
                r"[\s,;:]+\d+(?:\s*[:\-–]\s*\d+)?(?:\s*[-–]\s*\d+)?\s*$",
                "",
                name,
            )
        # Strip leftover trailing month name (no year) — happens when the
        # year was stripped by another regex first.
        name = re.sub(
            rf",?\s*\b(?:{months})\b\.?\s*$",
            "",
            name,
            flags=re.IGNORECASE,
        )
        # Strip leftover bare volume/page keyword at the end ("p", "pp",
        # "vol", "vol.", "no", "no.") that survives when the number got
        # truncated upstream by the search engine result preview.
        name = re.sub(
            r",?\s*\b(?:vol(?:ume)?|pp?|no)\b\.?\s*$",
            "",
            name,
            flags=re.IGNORECASE,
        )
        # Strip empty / whitespace-only trailing parens — arXiv
        # journal_ref fields sometimes end with "()" where the citation
        # year was stripped upstream, e.g. "Physical Review Research ()".
        name = re.sub(r"\s*\(\s*\)\s*$", "", name)
        # Strip geographic qualifiers: "(London)", "(New York)", "(US)"
        # Only strip parenthesized suffixes that contain no digits
        # (preserves "NeurIPS (2023)" which is handled by the year regex)
        name = re.sub(r"\s*\([^()0-9]+\)\s*$", "", name)
        # Strip trailing truncated volume/page markers: ", v", ", p",
        # ", vol" — these appear when the search engine preview cut the
        # citation mid-keyword ("Plasma Physics and Controlled Fusion,
        # vol. 63, no. 8" → "Plasma Physics and Controlled Fusion, v").
        name = re.sub(
            r",\s*(?:v|vol|p|pp|no|n)\.?\s*$",
            "",
            name,
            flags=re.IGNORECASE,
        )
        # Remove trailing punctuation and whitespace
        name = re.sub(r"[,;.\s]+$", "", name)
        # Strip a trailing 4-digit year/volume (conferences: "NeurIPS 2023"
        # → "NeurIPS"). Comes after the punctuation strip so any trailing
        # comma/period is already gone, and after the parenthesized-year
        # regex above so we don't double-process "(2023)".
        name = re.sub(r"\s+\d{4}\s*$", "", name)
        # Normalize "&" → "and" for consistent matching
        name = re.sub(r"\s*&\s*", " and ", name)
        # Normalize internal whitespace
        return re.sub(r"\s+", " ", name).strip()

    def __llm_clean_journal_name(self, journal_name: str) -> Optional[str]:
        """LLM-based fallback for canonicalizing unusual journal names.

        The regex + JabRef abbreviation tiers handle the common cases
        (volume/year/page stripping, well-known abbreviations like
        "Phys. Rev. Lett." → "Physical Review Letters"). They cannot
        handle locations ("ICML 2023, Honolulu"), unusual abbreviations
        not in the JabRef list, or non-English title transliterations.

        This is gated behind ``enable_llm_scoring`` so it never fires
        unless the user opted into the Tier 4 LLM path. Called only as a
        salvage step when bundled tiers all miss, so the LLM bill is
        bounded by the number of *unrecognised* journals per query, not
        every journal.

        Args:
            journal_name: A name that the regex tier could not match
                against any bundled dataset.

        Returns:
            A canonicalised name from the LLM, or ``None`` if the call
            failed or the response was empty.
        """
        prompt = (
            f"Clean up the following journal or conference name:\n\n"
            f'"{journal_name}"\n\n'
            "Remove any references to volumes, pages, months, or years. "
            "Expand common abbreviations. For conferences, remove "
            "locations. Output only the clean name, no explanation."
        )
        try:
            response = self.model.invoke(prompt)
            content = getattr(response, "content", None) or response
            cleaned = str(content).strip().strip('"').strip("'")
            if not cleaned:
                return None
            return cleaned
        except (
            ConnectionError,
            TimeoutError,
            ValueError,
        ) as e:
            # Network / service / parse failures are expected and
            # recoverable — caller falls back to the regex-cleaned name.
            # Surface at WARNING (not silent / not DEBUG) so they're
            # visible during triage without flooding info-level logs.
            # Log only the exception class name; the message can carry
            # request-specific data (URLs, prompts) that doesn't belong
            # in operational logs.
            logger.warning(
                f"LLM name cleaning failed for '{journal_name}' "
                f"({type(e).__name__}); using regex-cleaned version"
            )
            return None

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

    def __analyze_journal_reputation(self, journal_name: str) -> int:
        """Analyze journal reputation via 1 SearXNG search + 1 LLM call.

        This is Tier 4 — the last-resort scoring path. Only used when
        the journal is not found in bundled data (OpenAlex, DOAJ, predatory).
        Uses a single web search for context, then a single LLM call to score.

        Args:
            journal_name: Cleaned journal name to research.

        Returns:
            Reputation score between 1 and 10.

        Raises:
            ValueError: If the LLM response cannot be parsed as a score.
        """
        logger.info(f"Tier 4: LLM analysis for journal '{journal_name}'...")

        # Single SearXNG search for journal info. Serialize access to
        # the shared SearXNG engine to prevent two threads from
        # clobbering its instance state (_last_results_count,
        # _search_results, rate tracker).
        query = f'"{journal_name}" academic journal impact factor quartile'
        with self.__engine_lock:
            results = self.__engine.run(query)

        # Extract snippets from search results
        snippets = []
        for r in results[:10]:
            snippet = r.get("snippet", "") or r.get("content", "")
            if snippet:
                snippets.append(snippet)
        journal_info_text = "\n".join(snippets)

        if not journal_info_text:
            logger.warning(
                f"No SearXNG results for '{journal_name}' — "
                f"cannot score via Tier 4"
            )
            raise ValueError(f"No search results for journal '{journal_name}'")

        # Truncate to fit context
        if len(journal_info_text) > self.__max_context:
            journal_info_text = journal_info_text[: self.__max_context] + "..."

        # Single LLM call to score. Wording mirrors the long-standing
        # original prompt — earlier code review flagged that arbitrary
        # rewrites of this prompt have a real chance of regressing the
        # Q1/Q2/Q3 calibration the rest of the code depends on.
        prompt = f"""
You are a research assistant helping to assess the reliability and
reputability of scientific journals. A reputable journal should be
peer-reviewed, not predatory, and high-impact. Please review the
following information on the journal "{journal_name}" and output a
reputability score between 1 and 10, where 1-3 is not reputable and
probably predatory, 4-6 is reputable but low-impact (Q2 or Q3),
and 7-10 is reputable Q1 journals. Only output the number, do not
provide any explanation or other output.

JOURNAL INFORMATION:

{journal_info_text}
"""

        response = self.model.invoke(prompt).content
        logger.debug(f"Tier 4 LLM response for '{journal_name}': {response}")

        match = re.search(r"\d+", response.strip())
        if match is None:
            logger.warning(
                f"Failed to parse score from LLM response for "
                f"'{journal_name}': {response!r}"
            )
            raise ValueError(
                "Failed to parse reputation score from LLM response."
            )

        reputation_score = int(match.group())
        if reputation_score not in VALID_QUALITY_SCORES:
            # Scoring tiers emit {1,4,5,6,7,8,10}; LLM returning anything
            # else is almost certainly prompt drift. Treat as a parse
            # failure so the existing failure counter + circuit breaker
            # observe it, rather than snapping to the nearest bucket and
            # silently masking the problem.
            logger.warning(
                f"LLM returned out-of-set score {reputation_score} for "
                f"'{journal_name}' (expected one of "
                f"{sorted(VALID_QUALITY_SCORES)}); treating as parse failure"
            )
            raise ValueError(
                f"LLM returned out-of-set score {reputation_score}."
            )
        return reputation_score

    # ------------------------------------------------------------------
    # Database operations
    # ------------------------------------------------------------------

    def __save_llm_score_to_db(self, *, name: str, quality: int) -> None:
        """Cache a Tier 4 LLM score for future research runs.

        Only Tier 4 (LLM) results are cached — Tiers 1–3 read directly
        from the read-only reference DB on every scoring pass. The
        lookup predicate filters on ``score_source == "llm"`` and the
        current ``quality_model`` so stale scores from a superseded LLM
        don't get served.

        No-op during the preview filter phase (no thread context).
        """
        session_ctx = self.__db_session()
        if session_ctx is None:
            return
        try:
            self._save_journal_to_db_inner(
                session_ctx, name=name, quality=quality
            )
        except Exception:
            # Score is still valid and returned to the caller — only
            # the cache write failed.
            logger.exception(
                f"Failed to cache LLM score for '{name}' — "
                f"score is still returned but won't be cached."
            )

    def _save_journal_to_db_inner(
        self, session_ctx, *, name: str, quality: int
    ) -> None:
        """Race-safe upsert of the Tier 4 LLM cache row.

        Mirrors the Paper upsert pattern: select-then-insert in a
        savepoint, and on IntegrityError (a concurrent writer created
        the row first) roll back the savepoint and re-fetch. This
        prevents the pre-fix bug where two concurrent scorings of the
        same journal collided on the UNIQUE(name) constraint and the
        exception handler left the cache empty.
        """
        from sqlalchemy.exc import IntegrityError

        now = int(time.time())
        model_id = get_model_identifier(self.model)

        with session_ctx as db_session:
            journal = db_session.query(Journal).filter_by(name=name).first()
            if journal is not None:
                journal.quality = quality
                journal.score_source = "llm"
                journal.quality_analysis_time = now
                journal.quality_model = model_id
                # name_lower MUST go through normalize_name, not bare
                # .lower(). Bare lowercase leaves U+2122 (TM), ligatures,
                # fullwidth letters intact; NFKC collapses them. The
                # migration backfill (0006:257) uses the same normalization;
                # divergence produces silent cache misses and, on migration,
                # UNIQUE violations that abort the upgrade.
                journal.name_lower = normalize_name(name)
                try:
                    db_session.commit()
                except Exception:
                    db_session.rollback()
                    logger.warning(
                        f"Failed to update cached LLM score for '{name}'"
                    )
                return

            sp = db_session.begin_nested()
            try:
                journal = Journal(
                    name=name,
                    name_lower=normalize_name(name),
                    quality=quality,
                    score_source="llm",
                    quality_model=model_id,
                    quality_analysis_time=now,
                )
                db_session.add(journal)
                db_session.flush()
                sp.commit()
                db_session.commit()
                return
            except IntegrityError:
                sp.rollback()

            # Competing writer inserted first; re-fetch and update.
            journal = db_session.query(Journal).filter_by(name=name).first()
            if journal is None:
                # Genuinely unexpected — UNIQUE violation with no row.
                logger.warning(
                    f"IntegrityError on Journal '{name}' insert but "
                    f"row not found on re-fetch; skipping cache write."
                )
                return
            journal.quality = quality
            journal.score_source = "llm"
            journal.quality_analysis_time = now
            journal.quality_model = model_id
            journal.name_lower = normalize_name(name)
            try:
                db_session.commit()
            except Exception:
                db_session.rollback()
                logger.warning(f"Failed to merge cached LLM score for '{name}'")

    # ------------------------------------------------------------------
    # Tiered scoring for a single journal
    # ------------------------------------------------------------------

    def __score_journal(
        self, journal_name: str, result: Dict[str, Any]
    ) -> tuple[int | None, str | None]:
        """
        Score a journal using the tiered approach.

        Returns ``(score, source_tag)``:
        - ``score`` is the 1-10 quality value, or ``None`` if the
          journal is predatory (signal to auto-remove).
        - ``source_tag`` identifies which tier produced the score:
          ``"openalex"``, ``"doaj"``, ``"institution"``, ``"llm"``
          (Tier 4 live scoring OR cache hit on a prior LLM row),
          ``"conference"``, or ``"low_confidence"`` (no tier matched).
          ``None`` when predatory.

        The tag is attached to the result dict for rendering. Nothing
        is frozen on the Paper row; the dashboard resolves current
        quality live from ``journals.quality`` (Tier 4) or the bundled
        reference DB (Tier 1-3) so a re-scored journal propagates
        automatically.
        """
        dm = self.__data_manager

        # Extract IDs from result for richer lookups
        issn = result.get("issn")
        openalex_sid = result.get("openalex_source_id")
        publisher = result.get("publisher")

        # --- Tier 1: Predatory check ---
        is_pred, pred_source = dm.is_predatory(
            journal_name=journal_name,
            publisher_name=publisher,
        )
        if is_pred:
            # Check whitelist override (avoids false positives)
            if dm.is_whitelisted(issn=issn, name=journal_name):
                logger.debug(
                    f"Tier 1: '{journal_name}' is on predatory list "
                    f"({pred_source}) but whitelisted — not removing"
                )
            else:
                logger.warning(
                    f"Tier 1: PREDATORY — removing results from "
                    f"'{journal_name}' (source: {pred_source})"
                )
                return None, None  # Signal auto-remove

        # --- Tier 2: OpenAlex snapshot ---
        oa_entry = dm.lookup_openalex(
            source_id=openalex_sid, issn=issn, name=journal_name
        )
        if oa_entry:
            h_idx = oa_entry.get("h_index")
            oa_doaj = oa_entry.get("is_in_doaj", False)
            oa_type = oa_entry.get("type", "journal")
            oa_quartile = oa_entry.get("quartile")
            score = dm.derive_quality_score(
                h_index=h_idx,
                quartile=oa_quartile,
                is_in_doaj=oa_doaj,
                source_type=oa_type,
            )
            if score is not None:
                logger.debug(
                    f"Tier 2 (OpenAlex): '{journal_name}' → "
                    f"score {score}/10 "
                    f"(quartile: {oa_quartile or '—'}, h-index: {h_idx})"
                )
                # Tier 2 results are NOT cached to the user DB — the
                # read-only reference DB is already a 100–300µs lookup,
                # so a second-level cache adds no value. Only Tier 4
                # (LLM) results are cached, below.
                # Preprint repositories (arxiv, biorxiv, ssrn, ...) get a
                # low Tier-2 floor because they aren't peer-reviewed. If
                # the authors are at a strong institution, lift the score
                # via the institution tier — taking max so a real venue
                # match (≥6) is never demoted. Only applies to repository
                # source types and only when score is weak (≤5).
                if oa_type == "repository" and score <= 5:
                    affs = result.get("affiliations")
                    if affs:
                        inst = dm.score_from_affiliations(affs)
                        if inst is not None and inst > score:
                            logger.debug(
                                f"Tier 2+3.5 (preprint lift): "
                                f"'{journal_name}' {score} → {inst} via "
                                f"institutions: {_format_affiliations(affs)}"
                            )
                            return inst, "institution"
                return score, "openalex"

        # --- Tier 3: DOAJ ---
        if issn:
            doaj_entry = dm.lookup_doaj(issn=issn)
            if doaj_entry:
                score = dm.derive_quality_score(is_in_doaj=True)
                logger.debug(
                    f"Tier 3 (DOAJ): '{journal_name}' → score {score}/10"
                )
                return score, "doaj"

        # --- Tier 3.5: Institution lookup ---
        # When the venue tiers couldn't score the paper, fall back to
        # author affiliations. This is the *only* trust signal we have
        # for preprints with no journal_ref or for venues OpenAlex
        # doesn't index. Score is capped at 6 inside score_from_affiliations
        # so institution alone never beats a real venue match.
        affiliations = result.get("affiliations")
        if affiliations:
            inst_score = dm.score_from_affiliations(affiliations)
            if inst_score is not None:
                logger.debug(
                    f"Tier 3.5 (Institution): '{journal_name}' → "
                    f"score {inst_score}/10 from institutions: "
                    f"{_format_affiliations(affiliations)}"
                )
                return inst_score, "institution"

        # --- DB cache: check for cached LLM results before expensive tiers ---
        # Tiers 1-3 use bundled data (instant, no caching needed).
        # Only Tier 4 (LLM) results are expensive and worth caching.
        # The predicate filters on quality_model so scores from a
        # superseded LLM model miss the cache and re-score, and
        # `cached.quality` is validated against VALID_QUALITY_SCORES to
        # evict any pre-fix rows that stored an out-of-set value.
        session_ctx = self.__db_session()
        if session_ctx is not None:
            try:
                with session_ctx as session:
                    cached = (
                        session.query(Journal)
                        .filter_by(name=journal_name)
                        .filter(Journal.score_source == "llm")
                        .filter(
                            Journal.quality_model
                            == get_model_identifier(self.model)
                        )
                        .first()
                    )
                    if cached is not None:
                        is_fresh = (
                            time.time() - cached.quality_analysis_time
                        ) < self.__quality_reanalysis_period.total_seconds()
                        is_valid = cached.quality in VALID_QUALITY_SCORES

                        if is_fresh and is_valid:
                            logger.info(
                                f"DB cache hit: '{journal_name}' → "
                                f"score {cached.quality}/10 [cached LLM]"
                            )
                            # Cache hit on a Journal row written by the
                            # LLM path — tag as "llm" so the dashboard
                            # surfaces it as a Tier 4 verdict.
                            return cached.quality, "llm"
                        if is_fresh and not is_valid:
                            logger.warning(
                                f"Cached score {cached.quality} for "
                                f"'{journal_name}' not in valid set "
                                f"{sorted(VALID_QUALITY_SCORES)}; rescoring"
                            )
                        # Expired or invalid — fall through to re-evaluate
            except Exception:
                logger.exception(
                    f"DB cache read failed for '{journal_name}', "
                    f"continuing with LLM tiers"
                )

        # --- Tier 4: LLM analysis (last resort) ---
        # Off by default — bundled data covers 217K+ sources and this tier
        # adds significant latency (1 SearXNG search + 1 LLM call per
        # unknown journal). Users opt in via the
        # `search.journal_reputation.enable_llm_scoring` setting. The
        # __searxng_available and consecutive-failures checks below remain
        # as runtime safety nets even when the user enabled it.
        from ...config.search_config import get_setting_from_snapshot

        _enable_tier4 = bool(
            get_setting_from_snapshot(
                "search.journal_reputation.enable_llm_scoring",
                False,
                settings_snapshot=self.__settings_snapshot,
            )
        )

        # Tier 3.6: LLM-based name cleanup salvage. Gated behind the same
        # opt-in flag as Tier 4. Asks the LLM to canonicalise the name
        # (handles abbreviations and locations the regex can't), then
        # retries the cheap bundled tiers. This costs one extra LLM call
        # per unknown journal but can avoid Tier 4's full SearXNG search.
        if _enable_tier4:
            relabeled = self.__llm_clean_journal_name(journal_name)
            if relabeled and relabeled != journal_name:
                logger.debug(
                    f"Tier 3.6 (LLM cleanup): '{journal_name}' → "
                    f"'{relabeled}', retrying bundled tiers"
                )
                oa_retry = dm.lookup_openalex(name=relabeled)
                if oa_retry:
                    h_idx = oa_retry.get("h_index")
                    oa_doaj = oa_retry.get("is_in_doaj", False)
                    score = dm.derive_quality_score(
                        h_index=h_idx,
                        quartile=oa_retry.get("quartile"),
                        is_in_doaj=oa_doaj,
                        source_type=oa_retry.get("type", "journal"),
                    )
                    if score is not None:
                        logger.info(
                            f"Tier 3.6 (LLM cleanup → OpenAlex): "
                            f"'{journal_name}' (as '{relabeled}') → "
                            f"score {score}/10"
                        )
                        # Tier 3.6 is a Tier 2 retry under a cleaned
                        # name; the result is effectively Tier 2 data
                        # and is NOT cached in the user DB (reference
                        # DB lookups are already instant). Tagged
                        # "openalex" — only the NAME came from the LLM,
                        # not the score.
                        return score, "openalex"

        if (
            _enable_tier4
            and self.__searxng_available
            and self.__searxng_failures() < 2
        ):
            try:
                quality = self.__analyze_journal_reputation(journal_name)
                self.__reset_searxng_failures()
                # There used to be a +1 "DOAJ Seal bonus" here. DOAJ
                # retired the Seal in April 2025, so the bonus could
                # only ever fire on stale pre-2025 data and was removed.
                self.__save_llm_score_to_db(name=journal_name, quality=quality)
                logger.debug(
                    f"Tier 4 (LLM): '{journal_name}' → "
                    f"score {quality}/10 "
                    f"[via SearXNG + LLM analysis]"
                )
                return quality, "llm"
            except Exception:
                failures = self.__bump_searxng_failures()
                logger.exception(
                    f"Tier 4 failed for '{journal_name}'. "
                    f"Consecutive failures: {failures}"
                )
                if failures >= 2:
                    logger.warning(
                        "Tier 4 disabled for remaining journals in "
                        "this batch (2 consecutive failures)."
                    )

        # --- Conference heuristic (for papers without DOI or OpenAlex match) ---
        # Guard: many high-tier journals start with "Proceedings of …"
        # (PNAS, Royal Society A/B, AMS, LMS, …). The bare `proceedings`
        # token in `_CONFERENCE_PATTERNS` would otherwise classify them
        # as Q3 conferences and throw away their real h-index. Skip the
        # heuristic for these — they fall through to the unknown-journal
        # score (3) and the user's threshold decides what to do.
        if journal_name.lower().lstrip().startswith("proceedings of "):
            logger.debug(
                f"Conference heuristic: skipped for '{journal_name}' "
                f"(starts with 'Proceedings of' — likely a journal, "
                f"not a conference)"
            )
        elif _is_likely_conference(journal_name):
            score = dm.derive_quality_score(source_type="conference")
            logger.debug(
                f"Conference heuristic: '{journal_name}' → "
                f"score {score}/10 (detected as conference by name pattern)"
            )
            return score, "conference"

        # No tier could score this journal — neither OpenAlex/DOAJ
        # venue match nor Tier 3.5 institution salvage produced a
        # signal. Score it as low-confidence (3) so the default
        # threshold (4) actually filters it out. Distinct from
        # predatory (1) — these are merely unknown, not blacklisted.
        affs_for_log = result.get("affiliations") or []
        logger.debug(
            f"No scoring data for '{journal_name}' — flagging as "
            f"low-confidence (score 3); tried institutions: "
            f"{_format_affiliations(affs_for_log)}"
        )
        return 3, "low_confidence"

    # ------------------------------------------------------------------
    # Main filter entry point
    # ------------------------------------------------------------------

    def filter_results(
        self, results: List[Dict], query: str, **kwargs
    ) -> List[Dict]:
        """Filter results by journal quality, with deduplication."""
        logger.info(
            f"Journal filter: processing {len(results)} results "
            f"(threshold={self.__threshold})"
        )
        # Fail-soft during a fresh install: if the reference DB file
        # isn't on disk yet, don't score anything. Every journal would
        # otherwise fall through to the "no scoring data" branch and
        # be marked as score 3, which is semantically wrong — we don't
        # know the journal is unknown, we just haven't loaded the data
        # yet. Tag each result with the QUALITY_PENDING sentinel so the
        # report renderer can show the user a helpful note pointing
        # them at /metrics/journals.
        #
        # We probe the DB file path directly rather than calling
        # ``data_manager.available`` — the latter has side effects
        # (triggers `_ensure_engine`, which tries to lazy-build the DB
        # and can block on an in-flight download for several minutes).
        # A cached engine on the data manager means either the file
        # was successfully opened earlier, or a test fixture has
        # injected an in-memory DB — either way we're good to score.
        try:
            from ...config.paths import get_journal_data_directory

            db_file = get_journal_data_directory() / "journal_quality.db"
            on_disk = db_file.exists() and db_file.stat().st_size > 0
            engine_cached = (
                getattr(self.__data_manager, "_engine", None) is not None
            )
            db_ready = on_disk or engine_cached
            logger.info(
                f"Journal filter: db_ready={db_ready} "
                f"(on_disk={on_disk}, engine_cached={engine_cached})"
            )
        except Exception:
            logger.exception(
                "Journal filter: db-ready probe raised; "
                "assuming DB not ready (pending)."
            )
            db_ready = False
        if not db_ready:
            from ...utilities.search_utilities import QUALITY_PENDING

            # Fire-and-forget the download in a daemon thread so the
            # pending-marker copy ("by the time you check, it may
            # already be done") is actually true. Without this the
            # filter would just tag every search "pending" forever
            # and nobody would ever fetch the data unless the user
            # clicked Download manually. The 30-second TTL cache in
            # ensure_journal_data prevents multiple concurrent filter
            # workers from all racing to spawn the same download.
            #
            # Egress policy: skip the background fetch under
            # PRIVATE_ONLY or STRICT — those scopes shouldn't egress at
            # all, and the journal data sources (OpenAlex / DOAJ /
            # JabRef bulk downloads) are all public hosts.
            if self._should_skip_journal_fetch_for_scope():
                logger.bind(policy_audit=True).info(
                    "journal-data background fetch skipped: egress scope "
                    "forbids public fetches"
                )
            else:
                _start_background_journal_fetch()

            # Respect exclude_non_published — results without a venue
            # still get dropped in pending mode, same as in the full
            # scoring path. Only venued results carry the marker.
            out = []
            tagged = 0
            dropped = 0
            for r in results:
                if r.get("journal_ref"):
                    r.setdefault("journal_quality", QUALITY_PENDING)
                    out.append(r)
                    tagged += 1
                elif not self.__exclude_non_published:
                    out.append(r)
                else:
                    dropped += 1
            logger.warning(
                f"Journal filter: reference DB not yet built — "
                f"tagged {tagged} result(s) with QUALITY_PENDING, "
                f"kept {len(out) - tagged} venueless, "
                f"dropped {dropped} (exclude_non_published). "
                f"Background download triggered if not already running."
            )
            return out

        # Initialize `filtered` outside the try so the predatory-safe
        # fallback in the except handler can always reference it even
        # if the crash happens before Pass-1 populates anything.
        filtered: list = []

        try:
            # Reset the per-thread fail-fast counter for each batch.
            # The counter lives in `threading.local()` so concurrent
            # callers on the same filter instance don't clobber each
            # other (see Bug A3).
            self.__reset_searxng_failures()

            # Pass 1: collect the richest metadata per journal (the result
            # with ISSN/source_id) so scoring uses the best available data.
            journal_best_result: Dict[str, Dict] = {}
            results_with_journals: list[tuple[Dict, str]] = []

            def _handle_no_venue(result: Dict) -> None:
                """Institution-salvage then exclude_non_published policy.

                If Tier 3.5 salvages a score from author affiliations,
                the result carries that numeric score. Otherwise we
                tag it with ``QUALITY_PREPRINT`` so the report renderer
                can show a "preprint — not in journal catalog" label
                instead of leaving the quality column ambiguously blank.
                """
                from ...utilities.search_utilities import QUALITY_PREPRINT

                affs = result.get("affiliations")
                if affs:
                    inst_score = self.__data_manager.score_from_affiliations(
                        affs
                    )
                    if (
                        inst_score is not None
                        and inst_score >= self.__threshold
                    ):
                        result["journal_quality"] = inst_score
                        logger.debug(
                            f"Tier 3.5 (Institution, no venue): "
                            f"'{result.get('title', '')[:60]}' → "
                            f"score {inst_score}/10 from institutions: "
                            f"{_format_affiliations(affs)}"
                        )
                        filtered.append(result)
                        return
                if not self.__exclude_non_published:
                    result.setdefault("journal_quality", QUALITY_PREPRINT)
                    filtered.append(result)

            # Per-batch cache for journal name cleaning — avoids redundant
            # regex + abbreviation DB lookups when multiple results come
            # from the same raw journal_ref string (common in OpenAlex
            # result batches).
            _name_cache: Dict[str, str] = {}

            for result in results:
                journal_ref = result.get("journal_ref")
                # Strip whitespace-only refs — " " is truthy but has no
                # meaningful content for the filter.
                if isinstance(journal_ref, str):
                    journal_ref = journal_ref.strip()
                if not journal_ref:
                    _handle_no_venue(result)
                    continue

                # Use per-batch cache to skip redundant cleaning for
                # repeated raw journal_ref values in the same batch.
                clean_name = _name_cache.get(journal_ref)
                if clean_name is None:
                    clean_name = self.__clean_journal_name(journal_ref)
                    _name_cache[journal_ref] = clean_name
                # Cleanup can reduce a volume/page-only ref to "". Treat
                # that the same as "no venue" rather than bucketing all
                # affected results under a degenerate empty-string key.
                if not clean_name:
                    _handle_no_venue(result)
                    continue

                results_with_journals.append((result, clean_name))

                if clean_name not in journal_best_result:
                    journal_best_result[clean_name] = result
                else:
                    prev = journal_best_result[clean_name]
                    if (
                        not prev.get("issn")
                        and not prev.get("openalex_source_id")
                    ) and (
                        result.get("issn") or result.get("openalex_source_id")
                    ):
                        journal_best_result[clean_name] = result

            # Pass 2: score each unique journal once, then filter
            journal_scores: Dict[str, tuple[int | None, str | None]] = {}

            for result, clean_name in results_with_journals:
                if clean_name not in journal_scores:
                    journal_scores[clean_name] = self.__score_journal(
                        clean_name, journal_best_result[clean_name]
                    )

                score, source_tag = journal_scores[clean_name]

                if score is None:
                    # Predatory → auto-remove. Include original journal_ref
                    # and URL in the log so false-positive reports can be
                    # debugged without re-running the query.
                    logger.warning(
                        f"Auto-removed predatory: "
                        f"title='{result.get('title', '')[:80]}' "
                        f"journal_ref='{result.get('journal_ref', '')!r}' "
                        f"cleaned='{clean_name}' "
                        f"url={result.get('link') or result.get('url') or '—'}"
                    )
                    continue

                if score >= self.__threshold:
                    result["journal_quality"] = score
                    # Cleaned name that keyed the successful score —
                    # persisted on Paper.container_title so the dashboard
                    # can GROUP BY it and enrich from the reference DB.
                    result["journal_name_matched"] = clean_name
                    # Source tag — propagates to the rendered quality
                    # tag in the research output. Nothing is frozen on
                    # the Paper row; the dashboard resolves current
                    # quality live from journals.quality (Tier 4) or
                    # the bundled reference DB (Tier 1-3).
                    result["journal_quality_source"] = source_tag
                    filtered.append(result)

            predatory_count = sum(
                1 for s, _ in journal_scores.values() if s is None
            )
            passed_count = sum(
                1
                for s, _ in journal_scores.values()
                if s is not None and s >= self.__threshold
            )
            below_count = sum(
                1
                for s, _ in journal_scores.values()
                if s is not None and s < self.__threshold
            )
            logger.info(
                f"Journal quality filter: {len(results)} → "
                f"{len(filtered)} results | "
                f"{len(journal_scores)} unique journals scored | "
                f"{passed_count} passed, {below_count} below threshold, "
                f"{predatory_count} predatory removed"
            )
            return filtered

        except Exception:
            # Safety net: a filter crash should not kill the entire search,
            # but it MUST NOT re-admit predatory journals either.
            # `filtered` is predatory-free by construction (the pass-2 loop
            # `continue`s on predatory scores). Returning `results` — the
            # original unfiltered list — would leak predatory sources that
            # Tier 1 had already caught. Prefer losing in-flight non-
            # predatory results over breaking the predatory-removal
            # safety contract. Logged at ERROR so the root cause surfaces.
            logger.exception(
                "Journal quality filtering failed — returning partial "
                "(predatory-free) results. This is a bug that should be "
                "investigated."
            )
            return filtered
