Upload 6 files

asjc_classifier.py

@@ -0,0 +1,216 @@
+"""
+Abstract -> ASJC subject codes, for populating ResearchBee's recommendation filter.
+
+    from asjc_classifier import AsjcClassifier
+
+    clf = AsjcClassifier("asjc_model.joblib")           # local model only
+    clf = AsjcClassifier("asjc_model.joblib", llm=True) # + LLM adjudication
+
+    result = clf.classify(title=..., abstract=...)
+    result.codes        # ['3309', '1710'] — feed straight to recommend(asjc=...)
+    result.method       # 'linear+llm' | 'linear-topk'
+    result.candidates   # what the LLM was asked to choose between
+
+Architecture, and why it is not the obvious one
+-----------------------------------------------
+The obvious design is "run the linear model, escalate to an LLM when confidence
+is low". Measured on the held-out abstract set, that design cannot work: the
+top-1 probability averages 0.367 when the prediction is correct and 0.330 when
+it is wrong. There is no threshold separating them — the model is wrong often
+and unbothered about it.
+
+What the model *is* good at is narrowing. A correct code appears within:
+
+    top-1   30%
+    top-3   70%
+    top-5   95%     <- and flat thereafter
+
+So it is used as a candidate generator, not a decider. The linear model cuts 334
+codes to 5 with ~95% recall; the LLM picks among those 5. That inverts the usual
+cost argument in our favour — the prompt carries five labelled options rather
+than a 334-entry taxonomy, so the call is small, fast and hard to get wrong.
+
+Degradation without an API key is graceful: `codes` becomes the top 3, which the
+recommender applies as an OR filter. Recall is 70% rather than 95%, and the
+candidate pool is broader, but nothing breaks.
+
+Retrain after the OpenAlex harvest
+----------------------------------
+The shipped model is trained on journal titles, because that is the labelled text
+available without a network. It is applied to abstracts. `harvest_openalex.py`
+produces the matched corpus; retraining on it and re-running `asjc_eval.jsonl`
+measures the gain on the same 20 abstracts, so the upgrade is quantified rather
+than assumed.
+"""
+from __future__ import annotations
+
+import json
+import os
+import re
+from dataclasses import dataclass, field
+from pathlib import Path
+
+import joblib
+import numpy as np
+from scipy import sparse
+
+__all__ = ["AsjcClassifier", "AsjcResult"]
+
+#: Where the linear model's recall saturates. Measured, not guessed.
+DEFAULT_K = 5
+#: Codes returned when no LLM is available. top-3 = 70% recall.
+FALLBACK_K = 3
+
+
+@dataclass
+class AsjcResult:
+    codes: list[str]
+    names: list[str]
+    method: str
+    candidates: list[dict] = field(default_factory=list)
+    llm_error: str | None = None
+
+    def __bool__(self) -> bool:
+        return bool(self.codes)
+
+
+_PROMPT = """You are classifying an academic manuscript into Scopus ASJC subject areas.
+
+A statistical model has narrowed 334 possible subject areas to the {n} candidates \
+below. One or more of them is correct.
+
+Candidates:
+{candidates}
+
+Manuscript title: {title}
+
+Abstract: {abstract}
+
+Choose the 1-3 candidates that best describe where this manuscript belongs, most \
+appropriate first. Choose only from the candidate list — do not invent codes. If \
+the manuscript is genuinely interdisciplinary, return more than one; if it clearly \
+belongs to a single area, return just that one.
+
+Respond with JSON only, no other text:
+{{"codes": ["NNNN", ...], "reason": "one short sentence"}}"""
+
+
+class AsjcClassifier:
+    def __init__(
+        self,
+        model_path: str | Path = "asjc_model.joblib",
+        llm: bool = False,
+        llm_model: str | None = None,
+        api_key: str | None = None,
+    ):
+        self.bundle = joblib.load(model_path)
+        self.classes: np.ndarray = self.bundle["classes"]
+        self.names: dict[str, str] = self.bundle["names"]
+        self.parent_of: dict[str, str] = self.bundle["parent_of"]
+        self.trained_on: str = self.bundle.get("corpus", "unknown")
+
+        self._client = None
+        # Model id is configuration, not a constant — set ASJC_LLM_MODEL to
+        # whichever model you are entitled to. Verify it against your account's
+        # available models rather than trusting a default baked in here.
+        self.llm_model = llm_model or os.getenv("ASJC_LLM_MODEL", "claude-sonnet-4-5")
+        if llm:
+            key = api_key or os.getenv("ANTHROPIC_API_KEY")
+            if key:
+                try:
+                    import anthropic
+                    self._client = anthropic.Anthropic(api_key=key)
+                except ImportError:
+                    pass   # falls back to top-k; reported in AsjcResult.method
+
+    # -- stage 1: narrow ----------------------------------------------------
+    def candidates(self, text: str, k: int = DEFAULT_K) -> list[dict]:
+        """The linear model's top-k, with names attached. ~95% recall at k=5."""
+        X = sparse.hstack([
+            self.bundle["word"].transform([text]),
+            self.bundle["char"].transform([text]),
+        ]).tocsr()
+        p = self.bundle["clf"].predict_proba(X)[0]
+        order = np.argsort(-p)[:k]
+        return [
+            {
+                "code": str(self.classes[i]),
+                "name": self.names.get(str(self.classes[i]), str(self.classes[i])),
+                "score": round(float(p[i]), 4),
+            }
+            for i in order
+        ]
+
+    # -- stage 2: adjudicate ------------------------------------------------
+    def classify(
+        self,
+        title: str = "",
+        abstract: str = "",
+        k: int = DEFAULT_K,
+        max_codes: int = 3,
+    ) -> AsjcResult:
+        text = f"{title} {title} {abstract}".strip()
+        if not text:
+            return AsjcResult([], [], "empty")
+
+        cands = self.candidates(text, k=k)
+        if not cands:
+            return AsjcResult([], [], "no-candidates")
+
+        if self._client is None:
+            picked = [c["code"] for c in cands[:FALLBACK_K]]
+            return AsjcResult(picked, [self.names.get(c, c) for c in picked],
+                              "linear-topk", cands)
+
+        listing = "\n".join(f"  {c['code']}  {c['name']}" for c in cands)
+        prompt = _PROMPT.format(
+            n=len(cands), candidates=listing,
+            title=title or "(not supplied)",
+            abstract=(abstract or "(not supplied)")[:4000],
+        )
+        try:
+            msg = self._client.messages.create(
+                model=self.llm_model,
+                max_tokens=300,
+                temperature=0,
+                messages=[{"role": "user", "content": prompt}],
+            )
+            raw = "".join(b.text for b in msg.content if getattr(b, "type", "") == "text")
+            picked = self._parse(raw, {c["code"] for c in cands})[:max_codes]
+            if picked:
+                return AsjcResult(picked, [self.names.get(c, c) for c in picked],
+                                  "linear+llm", cands)
+            err = "LLM returned no valid code from the candidate set"
+        except Exception as e:                      # noqa: BLE001
+            err = f"{type(e).__name__}: {e}"
+
+        # Any LLM failure degrades to the statistical answer rather than raising.
+        # A broader-than-ideal subject filter is a far better outcome for the
+        # caller than a recommendation request that errors out.
+        picked = [c["code"] for c in cands[:FALLBACK_K]]
+        return AsjcResult(picked, [self.names.get(c, c) for c in picked],
+                          "linear-topk", cands, llm_error=err)
+
+    @staticmethod
+    def _parse(raw: str, allowed: set[str]) -> list[str]:
+        """
+        Pull codes out of the reply and hard-filter to the candidate set.
+
+        The filter is the point: it makes an invented or hallucinated code
+        structurally impossible to return, so the worst case is fewer codes
+        rather than a wrong one that silently poisons the journal filter.
+        """
+        codes: list[str] = []
+        m = re.search(r"\{.*\}", raw, re.S)
+        if m:
+            try:
+                codes = [str(c).strip() for c in json.loads(m.group(0)).get("codes", [])]
+            except (json.JSONDecodeError, AttributeError, TypeError):
+                codes = []
+        if not codes:
+            codes = re.findall(r"\b\d{4}\b", raw)
+        out: list[str] = []
+        for c in codes:
+            if c in allowed and c not in out:
+                out.append(c)
+        return out

asjc_eval.jsonl

@@ -0,0 +1,20 @@
+{"id": "lis-chatbot", "field": "Library & Information Science", "accept": ["3309", "1710", "1706"], "text": "We present a retrieval-augmented generation chatbot integrated with the Primo discovery layer at a university library. The system combines dense vector retrieval over catalogue records and LibGuides content with a large language model to answer patron reference questions at the point of need. We evaluate retrieval precision, answer groundedness and user satisfaction across 1,200 real reference queries submitted by academic library users over one semester."}
+{"id": "bibliometrics", "field": "Scientometrics", "accept": ["3309", "1706", "3312"], "text": "This scientometric analysis maps the growth of research output across national research systems using bibliometric indicators derived from Scopus and Web of Science. We compute field-normalised citation impact, international collaboration rates and disciplinary specialisation indices for 42 countries between 2000 and 2024, and examine the relationship between public research funding intensity and scholarly productivity."}
+{"id": "cardiology-rct", "field": "Cardiology", "accept": ["2705", "2700", "2736"], "text": "This multicentre randomised controlled trial evaluated the efficacy and safety of direct oral anticoagulation in patients with non-valvular atrial fibrillation. 840 patients were randomised to receive either apixaban or dose-adjusted warfarin, with a primary composite endpoint of ischaemic stroke or systemic embolism at 24 months. Cardiovascular mortality, major bleeding and all-cause hospitalisation were prespecified secondary outcomes."}
+{"id": "perovskite", "field": "Materials / Energy", "accept": ["2504", "2500", "2508", "2102", "3104", "3107", "1600", "2200"], "text": "We report a perovskite solar cell architecture achieving 25.6% power conversion efficiency. Thin-film deposition of methylammonium lead iodide was optimised through antisolvent engineering and interfacial passivation, and long-term device stability under continuous one-sun illumination was characterised over 1000 hours using X-ray diffraction, photoluminescence spectroscopy and impedance analysis."}
+{"id": "nlp-transformer", "field": "Computer Science / AI", "accept": ["1702", "1706", "1712", "1700"], "text": "We introduce a parameter-efficient fine-tuning method for large transformer language models that adapts fewer than 0.1% of weights while matching full fine-tuning performance on downstream natural language understanding benchmarks. We evaluate on GLUE and SuperGLUE, analyse the rank structure of the learned updates, and show consistent gains in low-resource transfer settings."}
+{"id": "groundwater", "field": "Earth / Environmental Science", "accept": ["2312", "1908", "1909", "1900", "2300", "2305", "2215"], "text": "We assess groundwater recharge dynamics in an arid alluvial aquifer using a coupled unsaturated-zone and finite-difference flow model calibrated against a decade of piezometric observations. Stable isotope signatures and chloride mass balance were used to partition recharge sources, and abstraction scenarios were simulated to evaluate long-term aquifer sustainability under projected climate forcing."}
+{"id": "nursing-education", "field": "Nursing", "accept": ["2900", "2922", "2901", "3304", "2908"], "text": "This mixed-methods study examined the effect of high-fidelity simulation on clinical reasoning among undergraduate nursing students. Ninety-four students completed simulated deteriorating-patient scenarios followed by structured debriefing. Clinical reasoning was assessed with a validated rubric, and semi-structured interviews explored students' experiences of psychological safety during debriefing."}
+{"id": "corporate-finance", "field": "Economics & Finance", "accept": ["2003", "2002", "2000", "1402", "1403"], "text": "We examine the effect of corporate governance quality on the cost of debt financing using a panel of listed firms across 18 emerging markets. Employing instrumental-variable estimation to address endogeneity, we find that board independence and disclosure quality are associated with materially lower credit spreads, with the effect concentrated among firms facing greater information asymmetry."}
+{"id": "medieval-history", "field": "History / Humanities", "accept": ["1202", "1200", "1208"], "text": "Drawing on previously unexamined chancery records, this article reconstructs patterns of monastic landholding in the late medieval Rhineland. It argues that the conventional narrative of continuous ecclesiastical accumulation obscures a sustained period of alienation driven by fiscal pressure, and situates these transactions within wider debates about the economic agency of religious houses."}
+{"id": "sociolinguistics", "field": "Linguistics / Arts & Humanities", "accept": ["3310", "1203", "1200", "3316"], "text": "This study investigates code-switching practices among second-generation multilingual speakers in an urban migration context. Using a corpus of naturalistic conversational recordings, we analyse patterns of intra-sentential switching and examine how speakers deploy linguistic resources to index identity and negotiate group membership across community settings."}
+{"id": "supply-chain", "field": "Operations / Decision Sciences", "accept": ["1803", "1800", "1802", "2209", "1408"], "text": "We formulate a stochastic mixed-integer programming model for resilient supply chain network design under correlated disruption risk. A Benders decomposition scheme with valid inequalities is developed to solve realistically sized instances, and computational experiments on a multi-echelon distribution case demonstrate substantial cost savings over deterministic and robust-optimisation baselines."}
+{"id": "immunology", "field": "Immunology & Microbiology", "accept": ["2403", "2404", "2400", "2406", "1312"], "text": "We characterise the tissue-resident memory T cell compartment in the human respiratory mucosa following viral infection. Single-cell RNA sequencing combined with flow cytometry revealed transcriptionally distinct subsets with divergent cytokine profiles, and adoptive transfer experiments in a murine model demonstrated that these populations confer durable protection against heterologous challenge."}
+{"id": "civil-engineering", "field": "Civil / Structural Engineering", "accept": ["2205", "2200", "2215", "2210"], "text": "This paper presents a performance-based seismic assessment of reinforced concrete moment frames designed to pre-1980 detailing provisions. Nonlinear time-history analyses were conducted across a suite of ground motions scaled to multiple hazard levels, and fragility surfaces were developed to quantify collapse probability and inform retrofit prioritisation for existing building stock."}
+{"id": "education-policy", "field": "Education", "accept": ["3304", "3300", "3312"], "text": "We evaluate the impact of a national school autonomy reform on student attainment using a difference-in-differences design across administrative regions. Drawing on eight years of standardised assessment data linked to school-level governance records, we find heterogeneous effects that vary systematically with baseline school capacity and local labour market conditions."}
+{"id": "pharmacology", "field": "Pharmacology & Pharmaceutics", "accept": ["3004", "3003", "3002", "3000", "1313"], "text": "We report the pharmacokinetic and pharmacodynamic characterisation of a novel small-molecule kinase inhibitor in preclinical models. Dose-ranging studies established plasma exposure and tissue distribution, hepatic microsomal assays quantified metabolic stability and CYP inhibition liability, and target engagement was confirmed by phosphoproteomic profiling in tumour xenografts."}
+{"id": "psychology-cognition", "field": "Psychology / Neuroscience", "accept": ["3205", "3206", "3200", "2805", "2802"], "text": "Across three preregistered experiments we examined how working memory load modulates attentional capture by emotionally salient distractors. Participants performed a visual search task under concurrent memory demand while eye movements were recorded. Results indicate that capture persists under high load for threat-related stimuli but is attenuated for neutral salient items."}
+{"id": "agriculture", "field": "Agricultural & Biological Sciences", "accept": ["1111", "1102", "1100", "2300", "1106"], "text": "We evaluate the effect of conservation tillage and cover cropping on soil organic carbon stocks and crop yield stability across a nine-year field experiment in a semi-arid cropping system. Soil sampling at multiple depths, combined with yield gap decomposition, indicates that carbon accrual is concentrated in surface horizons and that yield benefits emerge only in drier seasons."}
+{"id": "veterinary", "field": "Veterinary", "accept": ["3404", "3400", "3401", "3403"], "text": "This retrospective cohort study describes clinical presentation, treatment and outcome in dogs diagnosed with immune-mediated haemolytic anaemia at a referral hospital over twelve years. Prognostic factors were assessed by multivariable survival analysis, and the association between transfusion requirement, thromboembolic complications and 30-day mortality was quantified."}
+{"id": "dentistry", "field": "Dentistry", "accept": ["3504", "3506", "3500", "3501"], "text": "We compared marginal bone loss around platform-switched and conventional implant-abutment connections in a randomised split-mouth clinical trial. Sixty implants were placed in thirty partially edentulous patients, with standardised periapical radiography at baseline, six and twenty-four months, and peri-implant soft tissue parameters recorded alongside patient-reported outcomes."}
+{"id": "open-access-policy", "field": "LIS / Scholarly Communication", "accept": ["3309", "3312", "1710", "3300"], "text": "This study analyses institutional compliance with funder open access mandates across a national university system. Combining repository deposit records with publisher policy data from Open Policy Finder, we quantify green and gold route uptake, embargo adherence and article processing charge expenditure, and identify institutional factors associated with higher compliance rates."}

build_centroids.py

@@ -0,0 +1,130 @@
+# /// script
+# requires-python = ">=3.10"
+# dependencies = ["numpy", "sentence-transformers", "huggingface_hub"]
+# ///
+"""
+Turn the harvest into journal centroid vectors for the recommender's stage 2.
+
+    python build_centroids.py --db journals.db \
+        --journal-text journal_text.jsonl --out centroids.npz
+
+Run this AFTER harvest_openalex.py. It does no network I/O of its own beyond
+downloading the sentence-transformer weights once.
+
+What gets embedded
+------------------
+Per journal, in descending order of signal:
+
+  1. OpenAlex topics, repeated in proportion to how much the journal publishes
+     on each — an article-derived fingerprint, pre-aggregated by OpenAlex.
+  2. The journal's ASJC subject names, from the index.
+  3. The journal title.
+
+Journals OpenAlex could not match still get (2) and (3), so every journal has a
+vector and none silently drops out of the reranking.
+
+Cost note
+---------
+all-MiniLM-L6-v2 is 384-dimensional and ~90 MB. 32,050 active journals × 384
+float32 = ~49 MB uncompressed, well under what a Space can hold in memory. A
+larger model buys little here: the input is a bag of topic labels, not prose,
+so the extra capacity has nothing to work with.
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import sqlite3
+from pathlib import Path
+
+import numpy as np
+
+
+def load_topics(path: str) -> dict[str, str]:
+    out: dict[str, str] = {}
+    p = Path(path)
+    if not p.exists():
+        print(f"note: {path} not found — building from index metadata only, "
+              f"which is much weaker. Run harvest_openalex.py first.")
+        return out
+    with p.open(encoding="utf-8") as f:
+        for line in f:
+            try:
+                rec = json.loads(line)
+            except json.JSONDecodeError:
+                continue
+            sid = rec.get("scopus_id")
+            if sid:
+                out[sid] = (out.get(sid, "") + " " + rec.get("topics", "")).strip()
+    return out
+
+
+def main() -> None:
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--db", default="journals.db")
+    ap.add_argument("--journal-text", default="journal_text.jsonl")
+    ap.add_argument("--out", default="centroids.npz")
+    ap.add_argument("--model", default="sentence-transformers/all-MiniLM-L6-v2")
+    ap.add_argument("--batch-size", type=int, default=128)
+    ap.add_argument("--active-only", action="store_true", default=True)
+    ap.add_argument("--push-to-dataset", default=None)
+    args = ap.parse_args()
+
+    topics = load_topics(args.journal_text)
+    con = sqlite3.connect(args.db)
+
+    subs: dict[str, list[str]] = {}
+    for sid, name in con.execute(
+        """SELECT ja.scopus_id, coalesce(a.name, a.parent_name)
+           FROM journal_asjc ja JOIN asjc a ON a.code = ja.asjc"""
+    ):
+        subs.setdefault(sid, []).append(name or "")
+
+    where = "WHERE active = 1" if args.active_only else ""
+    rows = con.execute(f"SELECT scopus_id, title FROM journals {where}").fetchall()
+    con.close()
+
+    ids, docs = [], []
+    for sid, title in rows:
+        parts = [title or ""]
+        s = subs.get(sid)
+        if s:
+            parts.append(". ".join(sorted(set(s))))
+        t = topics.get(sid)
+        if t:
+            parts.append(t)
+        ids.append(sid)
+        docs.append(". ".join(p for p in parts if p)[:2000])
+
+    covered = sum(1 for i in ids if i in topics)
+    print(f"{len(ids)} journals; {covered} ({covered/max(len(ids),1):.0%}) "
+          f"have OpenAlex topics")
+
+    from sentence_transformers import SentenceTransformer
+    model = SentenceTransformer(args.model)
+    vecs = model.encode(
+        docs, batch_size=args.batch_size, show_progress_bar=True,
+        normalize_embeddings=True, convert_to_numpy=True,
+    ).astype("float32")
+
+    np.savez_compressed(args.out, ids=np.array(ids), vecs=vecs)
+    size = Path(args.out).stat().st_size / 1e6
+    print(f"wrote {args.out}  {vecs.shape}  ({size:.1f} MB)")
+
+    if args.push_to_dataset:
+        from huggingface_hub import HfApi
+        api = HfApi()
+        api.create_repo(args.push_to_dataset, repo_type="dataset", exist_ok=True)
+        api.upload_file(path_or_fileobj=args.out, path_in_repo=Path(args.out).name,
+                        repo_id=args.push_to_dataset, repo_type="dataset")
+        print(f"pushed {args.out} -> {args.push_to_dataset}")
+
+
+def load_centroids(path: str = "centroids.npz") -> dict[str, np.ndarray]:
+    """Load into the {scopus_id: vector} mapping recommend() expects."""
+    d = np.load(path, allow_pickle=False)
+    return {str(i): v for i, v in zip(d["ids"], d["vecs"])}
+
+
+if __name__ == "__main__":
+    main()

harvest_openalex.py

@@ -0,0 +1,353 @@
+# /// script
+# requires-python = ">=3.10"
+# dependencies = ["requests", "huggingface_hub"]
+# ///
+"""
+One OpenAlex pass that produces BOTH deliverables:
+
+  journal_text.jsonl   topical fingerprint per journal  -> centroids (recommender stage 2)
+  asjc_train.jsonl     labelled article abstracts       -> ASJC classifier training
+
+    python harvest_openalex.py --db journals.db --mailto you@ku.ac.ae
+    python harvest_openalex.py --db journals.db --mailto you@ku.ac.ae --resume
+
+These are the same job. The centroids need to know what each journal publishes
+about; the classifier needs text labelled with subject codes. Both are answered
+by reading OpenAlex once. Running them as separate harvests would double the
+request count for no benefit.
+
+Two phases
+----------
+PHASE 1  sources    ~1,400 requests, minutes.
+    Cursor-pages every journal source and takes its `topics`, which OpenAlex has
+    already inferred from that journal's article corpus. An article-derived
+    signal delivered pre-aggregated — most of the value of reading the articles,
+    at ~4% of the requests.
+
+PHASE 2  works      ~8,000 requests, 1-2 hours.
+    Pulls recent abstracts per journal to train the classifier. Sampling is
+    STRATIFIED BY ASJC CODE, not uniform over journals: ASJC is wildly
+    imbalanced (History has 2,009 journals, Medical Terminology has 2), and
+    uniform sampling would produce a classifier that has never seen the rare
+    half of the taxonomy. `--per-code` journals are drawn per subject area, so
+    thin codes get represented.
+
+Resumability
+------------
+HF Space containers restart without warning, and a two-hour harvest that loses
+everything on restart is not a harvest. Both outputs are append-only JSONL and
+progress is checkpointed to `<out>.state.json` after every batch. `--resume`
+picks up from the last checkpoint. Re-running without `--resume` starts over.
+
+Rate limits
+-----------
+OpenAlex allows 100k requests/day and ~10/sec in the polite pool. `--mailto` is
+what puts you in that pool — without it you are throttled harder and less
+predictably. It is required for a reason.
+
+Where to run this
+-----------------
+Not inside a Space. A Space's disk is ephemeral — "content is lost if your Space
+restarts or is stopped" — so a two-hour harvest writing 250k abstracts there is
+one restart away from nothing, and free Spaces give you no terminal to launch it
+from. Persistent storage on a Space is a paid Storage Bucket.
+
+Run it as an HF Job instead — same infrastructure, purpose-built for exactly
+this, billed per second:
+
+    hf jobs uv run --with requests,huggingface_hub \\
+        harvest_openalex.py -- --db journals.db --mailto you@ku.ac.ae \\
+        --push-to-dataset nikeshn/researchbee-index
+
+The PEP 723 header above declares the dependencies, so `hf jobs uv run` resolves
+them without a Dockerfile. `--push-to-dataset` uploads the outputs to a Dataset
+repo as they complete, which is what makes the run durable: the Job container is
+as ephemeral as the Space, so anything not pushed is lost when it exits.
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import random
+import sqlite3
+import time
+from collections import defaultdict
+from pathlib import Path
+
+import requests
+
+API = "https://api.openalex.org"
+SESSION = requests.Session()
+SESSION.headers.update({"User-Agent": "ResearchBee/1.0 (journal index build)"})
+
+
+# ---------------------------------------------------------------------------
+# HTTP
+# ---------------------------------------------------------------------------
+def get(url: str, params: dict, mailto: str, tries: int = 5) -> dict:
+    params = {**params, "mailto": mailto}
+    for attempt in range(tries):
+        try:
+            r = SESSION.get(url, params=params, timeout=60)
+            if r.status_code in (429, 503):
+                time.sleep(min(60, 2 ** attempt * 2))
+                continue
+            r.raise_for_status()
+            return r.json()
+        except requests.RequestException:
+            if attempt == tries - 1:
+                return {}
+            time.sleep(min(60, 2 ** attempt))
+    return {}
+
+
+def reconstruct_abstract(inv: dict | None) -> str:
+    """
+    OpenAlex ships abstracts as an inverted index ({word: [positions]}) because
+    of publisher redistribution terms. Rebuild the linear text.
+    """
+    if not inv:
+        return ""
+    positions: list[tuple[int, str]] = []
+    for word, idxs in inv.items():
+        for i in idxs:
+            positions.append((i, word))
+    if not positions:
+        return ""
+    positions.sort()
+    return " ".join(w for _, w in positions)
+
+
+# ---------------------------------------------------------------------------
+# Checkpointing
+# ---------------------------------------------------------------------------
+class Checkpoint:
+    def __init__(self, path: Path, resume: bool):
+        self.path = path
+        self.state = {"phase": 1, "cursor": "*", "done_journals": [], "n_works": 0}
+        if resume and path.exists():
+            self.state = json.loads(path.read_text())
+            print(f"resuming: phase {self.state['phase']}, "
+                  f"{len(self.state['done_journals'])} journals done, "
+                  f"{self.state['n_works']} abstracts written")
+
+    def save(self) -> None:
+        tmp = self.path.with_suffix(".tmp")
+        tmp.write_text(json.dumps(self.state))
+        tmp.replace(self.path)      # atomic — a restart mid-write must not corrupt it
+
+
+# ---------------------------------------------------------------------------
+# Index helpers
+# ---------------------------------------------------------------------------
+def issn_to_scopus(con: sqlite3.Connection) -> dict[str, str]:
+    m: dict[str, str] = {}
+    for sid, p, e in con.execute(
+        "SELECT scopus_id, issn_print, issn_e FROM journals WHERE active = 1"
+    ):
+        for i in (p, e):
+            if i:
+                m.setdefault(f"{i[:4]}-{i[4:]}", sid)
+    return m
+
+
+def stratified_journals(con: sqlite3.Connection, per_code: int, seed: int) -> list[tuple[str, str]]:
+    """
+    Pick up to `per_code` journals per ASJC code, preferring higher-SJR titles
+    (better-edited abstracts, cleaner topical signal), then de-duplicate.
+    Returns [(scopus_id, issn)].
+    """
+    rng = random.Random(seed)
+    by_code: dict[str, list[tuple[str, str]]] = defaultdict(list)
+    for sid, issn, code in con.execute(
+        """SELECT j.scopus_id, coalesce(j.issn_e, j.issn_print), ja.asjc
+           FROM journals j JOIN journal_asjc ja ON ja.scopus_id = j.scopus_id
+           WHERE j.active = 1 AND j.source_type = 'Journal'
+             AND coalesce(j.issn_e, j.issn_print) IS NOT NULL
+             AND j.has_english = 1
+           ORDER BY j.sjr DESC NULLS LAST"""
+    ):
+        by_code[code].append((sid, issn))
+
+    picked: dict[str, str] = {}
+    for code, rows in by_code.items():
+        # Take the strongest half, then sample within it, so the corpus is not
+        # entirely the same handful of megajournals per field.
+        pool = rows[: max(per_code * 3, per_code)]
+        rng.shuffle(pool)
+        for sid, issn in pool[:per_code]:
+            picked.setdefault(sid, issn)
+    return sorted(picked.items())
+
+
+# ---------------------------------------------------------------------------
+# Phase 1 — journal topics (centroids)
+# ---------------------------------------------------------------------------
+def phase1(con, mailto: str, out: Path, ck: Checkpoint) -> None:
+    keymap = issn_to_scopus(con)
+    seen_ids: set[str] = set()
+    if out.exists() and ck.state["phase"] > 1:
+        return
+    fh = out.open("a", encoding="utf-8")
+    if out.exists():
+        with out.open(encoding="utf-8") as f:
+            for line in f:
+                try:
+                    seen_ids.add(json.loads(line)["scopus_id"])
+                except (json.JSONDecodeError, KeyError):
+                    pass
+
+    cursor = ck.state.get("cursor") or "*"
+    page = 0
+    while cursor:
+        data = get(f"{API}/sources", {
+            "filter": "type:journal", "per-page": 200, "cursor": cursor,
+            "select": "id,display_name,issn,topics",
+        }, mailto)
+        results = data.get("results", [])
+        if not results:
+            break
+        for s in results:
+            sid = next((keymap[i] for i in (s.get("issn") or []) if i in keymap), None)
+            if not sid or sid in seen_ids:
+                continue
+            terms: list[str] = []
+            for t in (s.get("topics") or [])[:12]:
+                n = int(t.get("count", 1) or 1)
+                # Repeat proportional to prevalence so the embedding reflects
+                # what the journal mostly publishes, not its long tail.
+                terms += [t.get("display_name", "")] * max(1, min(4, n // 50))
+                for lvl in ("subfield", "field"):
+                    if (t.get(lvl) or {}).get("display_name"):
+                        terms.append(t[lvl]["display_name"])
+            if terms:
+                fh.write(json.dumps({"scopus_id": sid, "topics": " ".join(terms)}) + "\n")
+                seen_ids.add(sid)
+        fh.flush()
+        cursor = (data.get("meta") or {}).get("next_cursor")
+        ck.state["cursor"] = cursor
+        ck.save()
+        page += 1
+        if page % 10 == 0:
+            print(f"  phase1 page {page}  matched {len(seen_ids)}", flush=True)
+
+    fh.close()
+    ck.state.update(phase=2, cursor=None)
+    ck.save()
+    print(f"phase 1 done: {len(seen_ids)} journals with topics")
+
+
+# ---------------------------------------------------------------------------
+# Phase 2 — labelled abstracts (classifier)
+# ---------------------------------------------------------------------------
+def phase2(con, mailto: str, out: Path, ck: Checkpoint,
+           per_code: int, per_journal: int, seed: int, from_year: int) -> None:
+    targets = stratified_journals(con, per_code, seed)
+    done = set(ck.state["done_journals"])
+    todo = [(s, i) for s, i in targets if s not in done]
+    print(f"phase 2: {len(targets)} journals selected, {len(todo)} remaining")
+
+    fh = out.open("a", encoding="utf-8")
+    for n, (sid, issn) in enumerate(todo, 1):
+        fmt = f"{issn[:4]}-{issn[4:]}"
+        data = get(f"{API}/works", {
+            "filter": (f"primary_location.source.issn:{fmt},"
+                       f"from_publication_date:{from_year}-01-01,"
+                       "type:article,has_abstract:true"),
+            "per-page": per_journal,
+            "select": "title,abstract_inverted_index",
+        }, mailto)
+        wrote = 0
+        for w in data.get("results", []):
+            abstract = reconstruct_abstract(w.get("abstract_inverted_index"))
+            if len(abstract) < 200:
+                continue
+            text = f"{w.get('title') or ''}. {abstract}".strip()
+            fh.write(json.dumps({"scopus_id": sid, "text": text[:4000]}) + "\n")
+            wrote += 1
+        ck.state["n_works"] += wrote
+        done.add(sid)
+
+        if n % 25 == 0:
+            fh.flush()
+            ck.state["done_journals"] = sorted(done)
+            ck.save()
+            print(f"  phase2 {n}/{len(todo)} journals, "
+                  f"{ck.state['n_works']} abstracts", flush=True)
+        time.sleep(0.11)          # ~9 req/s, inside the polite-pool ceiling
+
+    fh.flush()
+    fh.close()
+    ck.state["done_journals"] = sorted(done)
+    ck.state["phase"] = 3
+    ck.save()
+    print(f"phase 2 done: {ck.state['n_works']} abstracts")
+
+
+def main() -> None:
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--db", default="journals.db")
+    ap.add_argument("--mailto", required=True,
+                    help="your email — required for OpenAlex's polite pool")
+    ap.add_argument("--resume", action="store_true")
+    ap.add_argument("--skip-phase1", action="store_true")
+    ap.add_argument("--skip-phase2", action="store_true")
+    ap.add_argument("--journal-text", default="journal_text.jsonl")
+    ap.add_argument("--train-out", default="asjc_train.jsonl")
+    ap.add_argument("--per-code", type=int, default=25,
+                    help="journals sampled per ASJC code (stratification depth)")
+    ap.add_argument("--per-journal", type=int, default=40,
+                    help="abstracts pulled per journal")
+    ap.add_argument("--from-year", type=int, default=2021)
+    ap.add_argument("--seed", type=int, default=42)
+    ap.add_argument("--push-to-dataset", default=None,
+                    help="HF Dataset repo id to upload outputs to, e.g. user/researchbee-index. "
+                         "Required for Job runs — the container's disk does not survive it.")
+    args = ap.parse_args()
+
+    con = sqlite3.connect(args.db)
+    ck = Checkpoint(Path(args.train_out + ".state.json"), args.resume)
+
+    if not args.resume:
+        for p in (args.journal_text, args.train_out):
+            Path(p).unlink(missing_ok=True)
+
+    if not args.skip_phase1 and ck.state["phase"] <= 1:
+        phase1(con, args.mailto, Path(args.journal_text), ck)
+    if not args.skip_phase2 and ck.state["phase"] <= 2:
+        phase2(con, args.mailto, Path(args.train_out), ck,
+               args.per_code, args.per_journal, args.seed, args.from_year)
+
+    for p in (args.journal_text, args.train_out):
+        path = Path(p)
+        if path.exists():
+            n = sum(1 for _ in path.open(encoding="utf-8"))
+            print(f"{p}: {n} lines, {path.stat().st_size/1e6:.1f} MB")
+    con.close()
+
+    if args.push_to_dataset:
+        push(args.push_to_dataset, [args.journal_text, args.train_out,
+                                    args.train_out + ".state.json"])
+
+
+def push(repo_id: str, paths: list[str]) -> None:
+    """
+    Upload outputs to a Dataset repo. This is not a convenience — a Job's disk
+    is discarded when the run ends, so anything not pushed did not happen.
+    """
+    from huggingface_hub import HfApi
+    api = HfApi()
+    api.create_repo(repo_id, repo_type="dataset", exist_ok=True)
+    for p in paths:
+        path = Path(p)
+        if not path.exists():
+            continue
+        api.upload_file(
+            path_or_fileobj=str(path), path_in_repo=path.name,
+            repo_id=repo_id, repo_type="dataset",
+        )
+        print(f"pushed {path.name} -> {repo_id}")
+
+
+if __name__ == "__main__":
+    main()

researchbee_journals.py

@@ -0,0 +1,602 @@
+"""
+ResearchBee journal index + recommender.
+
+    from researchbee_journals import JournalIndex
+
+    idx = JournalIndex("journals.db")
+    idx.lookup(issn="0737-8831")            # -> JournalRecord (Library Hi Tech)
+    idx.recommend(title=..., abstract=...)  # -> [Recommendation, ...]
+
+Architecture
+------------
+Three stages, deliberately separated so the expensive one stays optional:
+
+  1. CANDIDATE GENERATION  — hard ASJC filter (when subject codes are known) plus
+     FTS5/BM25 over title + publisher + subject names. Pure SQLite, no model,
+     sub-10ms, returns a few hundred candidates.
+
+  2. SEMANTIC RERANK       — optional. Cosine similarity between the manuscript
+     embedding and a per-journal centroid. See the honest caveat below.
+
+  3. FIT SCORING           — quartile, indexation state, OA status, coverage
+     recency, language, publisher agreement coverage. Deterministic, explainable,
+     and the part that should drive the text ResearchBee shows the researcher.
+
+The caveat on stage 2
+---------------------
+The Scopus Source List carries no aims-and-scope text. Embedding a journal
+*title* against a manuscript *abstract* is a weak signal — it recovers obvious
+lexical matches ("Journal of Hydrology" for a hydrology paper) and little else.
+Real matchers (Jane, Elsevier Journal Finder, WoS Match Manuscript) compare the
+manuscript against *articles the journal has published*. To reach that quality
+you need article-level text, which means building journal centroids from OpenAlex
+works — `build_centroids.py` does this. Until centroids exist, `recommend()`
+runs stages 1 and 3 only and says so in `Recommendation.method`, rather than
+quietly returning title-similarity noise dressed up as semantic matching.
+"""
+from __future__ import annotations
+
+import re
+import sqlite3
+import threading
+from dataclasses import dataclass, field, asdict
+from pathlib import Path
+from typing import Iterable, Sequence
+
+__all__ = ["JournalIndex", "JournalRecord", "Recommendation", "IndexationState"]
+
+
+# ---------------------------------------------------------------------------
+# Indexation state — the distinction the old lookup collapsed
+# ---------------------------------------------------------------------------
+class IndexationState:
+    """
+    Why this exists: 'no quartile' is three materially different answers to a
+    researcher deciding where to submit.
+
+      INDEXED_RANKED   in Scopus, has a current SJR and quartile. (30,405)
+      INDEXED_NEW      in Scopus, coverage began 2024+, SJR not yet computed.
+                       SJR needs a ~3-year citation window, so the absence is a
+                       timing artefact, not a quality signal. Safe to submit to.
+                       (326)
+      INDEXED_UNRANKED SCImago has a record but computes no SJR — typically below
+                       its citable-document threshold. Thin but real. (279)
+      INDEXED_NO_SJR   in Scopus, absent from SCImago entirely. Overwhelmingly
+                       book series (1,052), which SJR does not rank at all. Only
+                       ~70 are long-running journals, mostly regional titles.
+                       (1,366)
+      NOT_INDEXED      no Scopus source record at all, active or inactive.
+
+    And three states for titles Scopus no longer indexes. Keeping them apart is
+    the whole point: of 16,838 inactive titles only 967 were removed for cause.
+    Telling a researcher that a journal which merely changed its name is a
+    quality risk is worse than saying nothing.
+
+      DISCONTINUED     removed by Scopus for cause. The real red flag. (967)
+      RENAMED          ceased under this title but the work continues elsewhere;
+                       `successor_title` names where. Benign — redirect, don't
+                       warn. (~7,000 with a resolvable successor)
+      CEASED           stopped being indexed, no successor recorded and no
+                       discontinuation flag. Usually a journal that folded.
+    """
+    INDEXED_RANKED = "indexed_ranked"
+    INDEXED_NEW = "indexed_new"
+    INDEXED_UNRANKED = "indexed_unranked"
+    INDEXED_NO_SJR = "indexed_no_sjr"
+    NOT_INDEXED = "not_indexed"
+    DISCONTINUED = "discontinued"
+    RENAMED = "renamed"
+    CEASED = "ceased"
+
+    #: States a prospective author should be actively warned about.
+    WARN = frozenset({DISCONTINUED})
+    #: States meaning "Scopus does not index new articles here".
+    NOT_CURRENTLY_INDEXED = frozenset({DISCONTINUED, RENAMED, CEASED, NOT_INDEXED})
+
+    MESSAGES = {
+        INDEXED_RANKED: "Indexed in Scopus with a current SJR quartile.",
+        INDEXED_NEW: (
+            "Indexed in Scopus, but coverage is too recent for an SJR quartile. "
+            "SJR requires roughly three years of citation data; expect a first "
+            "quartile assignment once that window closes. The missing quartile "
+            "is not a quality signal."
+        ),
+        INDEXED_UNRANKED: (
+            "Indexed in Scopus and listed by SCImago, but no SJR is computed — "
+            "usually because the title publishes too few citable documents to "
+            "meet SCImago's threshold."
+        ),
+        INDEXED_NO_SJR: (
+            "Indexed in Scopus but absent from SCImago. Normal for book series "
+            "and trade journals, which SJR does not rank. Check the source type "
+            "before treating this as a concern."
+        ),
+        NOT_INDEXED: (
+            "No Scopus source record found — this title is not indexed in "
+            "Scopus, and has no record of having been."
+        ),
+        DISCONTINUED: (
+            "Scopus has discontinued this title — it was removed from the index "
+            "rather than ceasing publication. Articles published here will not "
+            "be indexed in Scopus. Scopus does not publish the reason for an "
+            "individual discontinuation; grounds range from publication-ethics "
+            "concerns to metrics such as self-citation, and discontinued titles "
+            "include some from major publishers. Verify independently before "
+            "submitting, and do not assume the journal is predatory."
+        ),
+        RENAMED: (
+            "This title is no longer indexed under this name, but the journal "
+            "continues. This is a rename or merger, not a quality concern."
+        ),
+        CEASED: (
+            "Scopus no longer indexes this title and no successor is recorded — "
+            "the journal appears to have ceased publication. Not flagged as a "
+            "discontinuation for cause."
+        ),
+    }
+
+
+_ISSN_RE = re.compile(r"[^0-9X]")
+
+
+def norm_issn(v: str | None) -> str | None:
+    if not v:
+        return None
+    s = _ISSN_RE.sub("", str(v).strip().upper())
+    if len(s) < 8:
+        s = s.zfill(8)
+    return s if len(s) == 8 else None
+
+
+# ---------------------------------------------------------------------------
+# Records
+# ---------------------------------------------------------------------------
+@dataclass
+class JournalRecord:
+    scopus_id: str
+    title: str
+    issn_print: str | None
+    issn_e: str | None
+    source_type: str | None
+    publisher: str | None
+    publisher_group: str | None
+    coverage_start: int | None
+    coverage_end: int | None
+    coverage_has_gap: bool
+    oa_unpaywall: bool
+    medline: bool
+    has_english: bool
+    languages: str | None
+    sjr: float | None
+    sjr_rank: int | None
+    sjr_best_quartile: str | None
+    h_index: int | None
+    cites_per_doc_2y: float | None
+    total_docs_year: int | None
+    oa_diamond: bool
+    overton: int | None
+    country: str | None
+    lifecycle: str = "active"
+    successor_title: str | None = None
+    successor_scopus_id: str | None = None
+    successor_is_active: bool = False
+    subjects: list[dict] = field(default_factory=list)
+    state: str = IndexationState.NOT_INDEXED
+    state_message: str = ""
+
+    @property
+    def warn(self) -> bool:
+        """True when a prospective author should be actively cautioned."""
+        return self.state in IndexationState.WARN
+
+    @property
+    def currently_indexed(self) -> bool:
+        return self.state not in IndexationState.NOT_CURRENTLY_INDEXED
+
+    def quartile_for(self, asjc: str) -> str | None:
+        """Quartile in one specific subject area, not the journal's best."""
+        for s in self.subjects:
+            if s["asjc"] == asjc:
+                return s.get("quartile")
+        return None
+
+    def to_dict(self) -> dict:
+        return asdict(self)
+
+
+@dataclass
+class Recommendation:
+    journal: JournalRecord
+    score: float
+    reasons: list[str]
+    method: str
+
+
+# ---------------------------------------------------------------------------
+# Index
+# ---------------------------------------------------------------------------
+_STOP = set("""
+a an the and or but if of in on at to for with by from as is are was were be been
+being this that these those it its we our they their he she his her you your not
+no can may might will would should could have has had do does did than then so
+such other others new novel study studies paper article research results method
+methods approach based using used use show shows shown propose proposed present
+presented provide provides between among during through both each more most many
+much also however therefore thus while when where which who whom what how why
+""".split())
+
+
+class JournalIndex:
+    """Thread-safe read-only wrapper over journals.db."""
+
+    def __init__(self, db_path: str | Path):
+        self.db_path = str(db_path)
+        if not Path(self.db_path).exists():
+            raise FileNotFoundError(self.db_path)
+        self._local = threading.local()
+
+    @property
+    def con(self) -> sqlite3.Connection:
+        con = getattr(self._local, "con", None)
+        if con is None:
+            con = sqlite3.connect(f"file:{self.db_path}?mode=ro", uri=True,
+                                  check_same_thread=False)
+            con.row_factory = sqlite3.Row
+            con.execute("PRAGMA mmap_size = 268435456")
+            self._local.con = con
+        return con
+
+    # -- meta ---------------------------------------------------------------
+    def meta(self) -> dict[str, str]:
+        return {r["key"]: r["value"] for r in self.con.execute("SELECT * FROM meta")}
+
+    # -- lookup -------------------------------------------------------------
+    def lookup(self, issn: str | None = None, title: str | None = None,
+               scopus_id: str | None = None) -> JournalRecord | None:
+        row = None
+        if scopus_id:
+            row = self.con.execute(
+                "SELECT * FROM journals WHERE scopus_id = ?", (scopus_id,)
+            ).fetchone()
+        if row is None and issn:
+            key = norm_issn(issn)
+            if key:
+                row = self.con.execute(
+                    "SELECT * FROM journals WHERE issn_e = ? OR issn_print = ? LIMIT 1",
+                    (key, key),
+                ).fetchone()
+        if row is None and title:
+            row = self.con.execute(
+                "SELECT * FROM journals WHERE lower(title) = lower(?) LIMIT 1",
+                (title.strip(),),
+            ).fetchone()
+            if row is None:
+                hits = self._fts(title, limit=1, weights=self._W_LOOKUP)
+                if hits:
+                    row = self.con.execute(
+                        "SELECT * FROM journals WHERE scopus_id = ?", (hits[0][0],)
+                    ).fetchone()
+        return self._to_record(row) if row else None
+
+    def _to_record(self, row: sqlite3.Row) -> JournalRecord:
+        subs = [
+            {"asjc": r["asjc"], "name": r["name"], "parent": r["parent_name"],
+             "quartile": r["quartile"], "is_primary": bool(r["is_primary"])}
+            for r in self.con.execute(
+                """SELECT ja.asjc, ja.is_primary, ja.quartile, a.name, a.parent_name
+                   FROM journal_asjc ja JOIN asjc a ON a.code = ja.asjc
+                   WHERE ja.scopus_id = ? ORDER BY ja.is_primary DESC, ja.asjc""",
+                (row["scopus_id"],),
+            )
+        ]
+        rec = JournalRecord(
+            scopus_id=row["scopus_id"], title=row["title"],
+            issn_print=self._fmt(row["issn_print"]), issn_e=self._fmt(row["issn_e"]),
+            source_type=row["source_type"], publisher=row["publisher"],
+            publisher_group=row["publisher_group"],
+            coverage_start=row["coverage_start"], coverage_end=row["coverage_end"],
+            coverage_has_gap=bool(row["coverage_has_gap"]),
+            oa_unpaywall=bool(row["oa_unpaywall"]), medline=bool(row["medline"]),
+            has_english=bool(row["has_english"]), languages=row["languages"],
+            sjr=row["sjr"], sjr_rank=row["sjr_rank"],
+            sjr_best_quartile=row["sjr_best_quartile"], h_index=row["h_index"],
+            cites_per_doc_2y=row["cites_per_doc_2y"],
+            total_docs_year=row["total_docs_year"],
+            oa_diamond=bool(row["oa_diamond"]), overton=row["overton"],
+            country=row["country"], lifecycle=row["lifecycle"],
+            successor_title=row["successor_title"],
+            successor_scopus_id=row["successor_scopus_id"],
+            successor_is_active=bool(row["successor_is_active"]),
+            subjects=subs,
+        )
+        rec.state = self._state(row)
+        rec.state_message = self._message(rec)
+        return rec
+
+    @staticmethod
+    def _message(rec: "JournalRecord") -> str:
+        """Base message, extended with the successor when we resolved one."""
+        msg = IndexationState.MESSAGES[rec.state]
+        if rec.successor_title:
+            where = (
+                "which is actively indexed"
+                if rec.successor_is_active
+                else "which is itself no longer indexed"
+            )
+            if rec.state == IndexationState.RENAMED:
+                msg += f" It continues as {rec.successor_title!r}, {where}."
+            elif rec.state in (IndexationState.DISCONTINUED, IndexationState.CEASED):
+                # A successor on a discontinued title does not soften the
+                # warning — it just tells the researcher where to look next.
+                msg += f" Scopus records a successor title, {rec.successor_title!r}, {where}."
+        return msg
+
+    @staticmethod
+    def _fmt(s: str | None) -> str | None:
+        return f"{s[:4]}-{s[4:]}" if s and len(s) == 8 else None
+
+    _STATE_BY_STATUS = {
+        "matched": IndexationState.INDEXED_RANKED,
+        "no_sjr_new_title": IndexationState.INDEXED_NEW,
+        "no_sjr_listed_unranked": IndexationState.INDEXED_UNRANKED,
+        "no_sjr_unmatched": IndexationState.INDEXED_NO_SJR,
+    }
+
+    _STATE_BY_LIFECYCLE = {
+        "discontinued": IndexationState.DISCONTINUED,
+        "renamed": IndexationState.RENAMED,
+        "ceased": IndexationState.CEASED,
+    }
+
+    @classmethod
+    def _state(cls, row: sqlite3.Row) -> str:
+        # Lifecycle wins over SJR status. A discontinued title may still carry a
+        # quartile from when it was indexed, and reporting that quartile as the
+        # headline would bury the thing the researcher actually needs to know.
+        lc = cls._STATE_BY_LIFECYCLE.get(row["lifecycle"])
+        if lc:
+            return lc
+        return cls._STATE_BY_STATUS.get(
+            row["sjr_status"], IndexationState.INDEXED_NO_SJR
+        )
+
+    # -- candidate generation ----------------------------------------------
+    # bm25 column weights: (scopus_id, title, publisher_group, subjects).
+    # Publisher is weighted to zero for topical work — "University Library
+    # System" is not evidence that a cinema journal wants your library paper,
+    # and leaving it at 1.0 lets one publisher flood the candidate pool. It
+    # stays queryable for name lookups via _W_LOOKUP.
+    _W_TOPIC = (0.0, 4.0, 0.0, 3.0)
+    _W_LOOKUP = (0.0, 8.0, 1.0, 0.5)
+
+    def _fts(self, text: str, limit: int = 400,
+             asjc: Sequence[str] | None = None,
+             weights: tuple[float, ...] | None = None,
+             active_only: bool = False) -> list[tuple[str, float]]:
+        terms = self._terms(text)
+        if not terms:
+            return []
+        w = weights or self._W_TOPIC
+        query = " OR ".join(f'"{t}"' for t in terms)
+        sql = f"""
+            SELECT f.scopus_id AS sid,
+                   bm25(journals_fts, {w[0]}, {w[1]}, {w[2]}, {w[3]}) AS rank
+            FROM journals_fts f WHERE journals_fts MATCH ?
+        """
+        params: list = [query]
+        if active_only:
+            # Applied inside candidate generation, not after: the index holds
+            # 16,838 inactive titles, and letting them consume the candidate
+            # pool would silently shrink the number of real options considered.
+            sql += " AND f.scopus_id IN (SELECT scopus_id FROM journals WHERE active = 1)"
+        if asjc:
+            marks = ",".join("?" * len(asjc))
+            sql += f""" AND f.scopus_id IN (
+                        SELECT scopus_id FROM journal_asjc WHERE asjc IN ({marks}))"""
+            params += list(asjc)
+        sql += " ORDER BY rank LIMIT ?"
+        params.append(limit)
+        return [(r["sid"], r["rank"]) for r in self.con.execute(sql, params)]
+
+    @staticmethod
+    def _terms(text: str, max_terms: int = 24) -> list[str]:
+        """
+        FTS5 treats bare punctuation as syntax, so terms are stripped to word
+        characters and quoted. Frequency-ordered, stopword-filtered.
+        """
+        from collections import Counter
+        words = re.findall(r"[A-Za-z][A-Za-z\-]{2,}", (text or "").lower())
+        words = [w.strip("-") for w in words if w not in _STOP and len(w) > 3]
+        return [w for w, _ in Counter(words).most_common(max_terms)]
+
+    # -- recommend ----------------------------------------------------------
+    def recommend(
+        self,
+        title: str = "",
+        abstract: str = "",
+        asjc: Sequence[str] | None = None,
+        *,
+        limit: int = 20,
+        candidate_pool: int = 400,
+        require_english: bool = True,
+        require_oa: bool = False,
+        exclude_book_series: bool = True,
+        min_quartile: str | None = None,
+        publisher_agreements: Iterable[str] = (),
+        embedder=None,
+        centroids=None,
+        asjc_classifier=None,
+    ) -> list[Recommendation]:
+        """
+        `asjc`  — subject codes to hard-filter on. Supply these when you know the
+                  manuscript's field; recall from free text alone is mediocre.
+        `min_quartile` — 'Q1' or 'Q2'. Applied against the quartile in the
+                  filtered subject area when `asjc` is given, otherwise against
+                  the journal's best quartile.
+        `publisher_agreements` — publisher_group names the institution holds a
+                  read-and-publish deal with; matches get a scoring bonus and an
+                  explicit reason string.
+        `embedder`/`centroids` — optional stage 2. See module docstring.
+        `asjc_classifier` — an AsjcClassifier. When `asjc` is not supplied, the
+                  subject codes are inferred from the manuscript rather than
+                  left empty. This matters more than any other option here: the
+                  subject filter is what makes the results usable, and an
+                  unfiltered free-text search returns unrelated journals.
+        """
+        inferred = None
+        if not asjc and asjc_classifier is not None:
+            inferred = asjc_classifier.classify(title=title, abstract=abstract)
+            asjc = inferred.codes or None
+
+        text = f"{title} {title} {abstract}".strip()
+        cands = self._fts(text, limit=candidate_pool, asjc=asjc, active_only=True)
+        if not cands and asjc:
+            marks = ",".join("?" * len(asjc))
+            cands = [
+                (r["scopus_id"], 0.0)
+                for r in self.con.execute(
+                    f"""SELECT DISTINCT j.scopus_id FROM journals j
+                        JOIN journal_asjc ja ON ja.scopus_id = j.scopus_id
+                        WHERE ja.asjc IN ({marks}) AND j.active = 1
+                        ORDER BY j.sjr DESC LIMIT ?""",
+                    (*asjc, candidate_pool),
+                )
+            ]
+        if not cands:
+            return []
+
+        ranks = [r for _, r in cands]
+        lo, hi = min(ranks), max(ranks)
+        span = (hi - lo) or 1.0
+
+        method = "lexical+fit"
+        sem: dict[str, float] = {}
+        if embedder is not None and centroids is not None:
+            sem = self._semantic(text, [c for c, _ in cands], embedder, centroids)
+            if sem:
+                method = "lexical+semantic+fit"
+
+        out: list[Recommendation] = []
+        agreements = {p.lower() for p in publisher_agreements}
+        for sid, rank in cands:
+            row = self.con.execute(
+                "SELECT * FROM journals WHERE scopus_id = ?", (sid,)
+            ).fetchone()
+            if row is None:
+                continue
+            if exclude_book_series and row["source_type"] != "Journal":
+                continue
+            if require_english and not row["has_english"]:
+                continue
+            if require_oa and not row["oa_unpaywall"]:
+                continue
+            # Never recommend somewhere Scopus no longer indexes — discontinued,
+            # renamed or ceased. Lookup still returns these (a researcher asking
+            # about a specific title needs the answer); recommendation must not.
+            if not row["active"]:
+                continue
+
+            rec = self._to_record(row)
+            q = rec.quartile_for(asjc[0]) if asjc else rec.sjr_best_quartile
+            q = q or rec.sjr_best_quartile
+            if min_quartile and (q is None or q > min_quartile):
+                continue
+
+            # bm25 returns lower-is-better; invert to 0..1
+            lex = 1.0 - ((rank - lo) / span)
+            score = 0.45 * lex
+            reasons: list[str] = []
+
+            if sem:
+                score = 0.30 * lex + 0.30 * sem.get(sid, 0.0)
+
+            qbonus = {"Q1": 0.25, "Q2": 0.18, "Q3": 0.10, "Q4": 0.04}.get(q or "", 0.0)
+            score += qbonus
+            if q:
+                label = f"{q} in {self._asjc_name(asjc[0])}" if asjc else f"{q} (best quartile)"
+                reasons.append(label)
+            elif rec.state == IndexationState.INDEXED_NEW:
+                score += 0.06
+                reasons.append("Newly indexed — no quartile yet, not a quality signal")
+
+            if rec.coverage_end and rec.coverage_end >= 2026:
+                score += 0.05
+            else:
+                reasons.append(f"Coverage appears to end {rec.coverage_end} — verify status")
+
+            if rec.oa_diamond:
+                # Diamond = no APC and no subscription. For an author without
+                # grant funding this outranks quartile as a practical criterion,
+                # so it is weighted above the generic OA flag.
+                score += 0.10
+                reasons.append("Diamond open access — no APC")
+            elif rec.oa_unpaywall:
+                score += 0.06
+                reasons.append("Open access")
+
+            if rec.publisher_group and rec.publisher_group.lower() in agreements:
+                score += 0.12
+                reasons.append(f"Covered by your {rec.publisher_group} agreement — APC likely waived")
+
+            if rec.coverage_has_gap:
+                score -= 0.04
+                reasons.append("Discontinuous Scopus coverage")
+
+            if rec.medline:
+                reasons.append("Medline-sourced")
+
+            out.append(Recommendation(rec, round(score, 4), reasons, method))
+
+        out.sort(key=lambda r: r.score, reverse=True)
+        out = out[:limit]
+
+        # An inferred subject filter silently determines the entire result set,
+        # so it is never left implicit: every result says which subject was
+        # inferred and how, giving the user something concrete to correct when
+        # the classifier picked the wrong field.
+        if inferred is not None:
+            note = f"Subject inferred as {', '.join(inferred.names)} ({inferred.method})"
+            for r in out:
+                r.reasons.append(note)
+                r.method = f"{r.method}+asjc:{inferred.method}"
+        return out
+
+    def _asjc_name(self, code: str) -> str:
+        r = self.con.execute("SELECT name, parent_name FROM asjc WHERE code = ?",
+                             (code,)).fetchone()
+        return (r["name"] or r["parent_name"]) if r else code
+
+    @staticmethod
+    def _semantic(text, sids, embedder, centroids) -> dict[str, float]:
+        import numpy as np
+        v = np.asarray(embedder(text), dtype="float32").ravel()
+        v /= (np.linalg.norm(v) or 1.0)
+        got = {s: centroids[s] for s in sids if s in centroids}
+        if not got:
+            return {}
+        keys = list(got)
+        M = np.vstack([got[k] for k in keys]).astype("float32")
+        M /= (np.linalg.norm(M, axis=1, keepdims=True) + 1e-9)
+        sims = M @ v
+        lo, hi = float(sims.min()), float(sims.max())
+        span = (hi - lo) or 1.0
+        return {k: float((s - lo) / span) for k, s in zip(keys, sims)}
+
+    # -- subject helpers ----------------------------------------------------
+    def find_asjc(self, query: str, limit: int = 10) -> list[dict]:
+        """Resolve a free-text subject to ASJC codes, for populating the filter."""
+        like = f"%{query.strip().lower()}%"
+        return [
+            dict(r) for r in self.con.execute(
+                """SELECT code, name, parent_code, parent_name, n_journals FROM asjc
+                   WHERE lower(name) LIKE ? OR lower(parent_name) LIKE ?
+                   ORDER BY n_journals DESC LIMIT ?""",
+                (like, like, limit),
+            )
+        ]
+
+    def publisher_titles(self, publisher_group: str) -> int:
+        return self.con.execute(
+            "SELECT count(*) FROM journals WHERE publisher_group = ?",
+            (publisher_group,),
+        ).fetchone()[0]

train_asjc.py

@@ -0,0 +1,329 @@
+"""
+Train the abstract -> ASJC multi-label classifier.
+
+    # interim model, no network needed — trains on journal titles in the index
+    python train_asjc.py --corpus titles --db journals.db --out asjc_model.joblib
+
+    # full model, after harvest_openalex.py has produced labelled abstracts
+    python train_asjc.py --corpus abstracts --train-jsonl asjc_train.jsonl \
+                         --db journals.db --out asjc_model.joblib
+
+Why TF-IDF + linear and not a transformer
+-----------------------------------------
+The task is topical multi-label text classification over 334 labels with a large
+vocabulary signal — exactly where a linear model over character and word n-grams
+is hard to beat per unit of cost. It trains in seconds on CPU, the artifact is
+tens of MB, and inference is sub-millisecond with no GPU and no model download at
+Space startup. A fine-tuned transformer might add a few points of F1 for ~400MB
+and a cold-start penalty on every Space restart. Take the linear model; spend the
+complexity budget on the LLM fallback for the cases it actually gets wrong.
+
+Two corpora, deliberately
+-------------------------
+`titles`    trains on journal titles (48,888 labelled examples, already in the
+            index, zero network). Ships immediately. But it trains on titles and
+            is applied to abstracts, and that domain gap is real — treat its
+            numbers as an upper bound on what it will do in production.
+
+`abstracts` trains on article abstracts labelled with their journal's ASJC codes
+            (weak supervision). Far better matched to inference, and the only
+            version whose held-out score means what it appears to mean.
+
+Label noise in the `abstracts` corpus
+-------------------------------------
+Giving every article its journal's codes is weak supervision: a physics paper in
+a multidisciplinary journal inherits 'Medicine' too. `--max-labels` drops journals
+carrying more than N ASJC codes from training, which removes the worst offenders
+(megajournals and general-science titles) at the cost of some coverage. Their
+articles are still *classifiable* at inference — they are just poor teachers.
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import re
+import sqlite3
+import sys
+from pathlib import Path
+
+import numpy as np
+from scipy import sparse
+from sklearn.feature_extraction.text import TfidfVectorizer
+from sklearn.linear_model import SGDClassifier
+from sklearn.model_selection import train_test_split
+from sklearn.multiclass import OneVsRestClassifier
+from sklearn.preprocessing import MultiLabelBinarizer
+
+import joblib
+
+
+# ---------------------------------------------------------------------------
+# Corpora
+# ---------------------------------------------------------------------------
+def load_titles(db: str, max_labels: int) -> tuple[list[str], list[list[str]]]:
+    """Journal title + subject names -> its ASJC codes."""
+    con = sqlite3.connect(db)
+    con.row_factory = sqlite3.Row
+    rows = con.execute(
+        """SELECT j.scopus_id, j.title, j.publisher_group,
+                  group_concat(ja.asjc) codes, count(*) n
+           FROM journals j JOIN journal_asjc ja ON ja.scopus_id = j.scopus_id
+           WHERE j.title IS NOT NULL
+           GROUP BY j.scopus_id"""
+    ).fetchall()
+    X, y = [], []
+    for r in rows:
+        if r["n"] > max_labels:
+            continue
+        X.append(r["title"])
+        y.append(r["codes"].split(","))
+    con.close()
+    return X, y
+
+
+def load_abstracts(path: str, db: str, max_labels: int) -> tuple[list[str], list[list[str]]]:
+    """
+    JSONL from harvest_openalex.py: {"text": ..., "scopus_id": ...}.
+    Labels come from the index, so the harvest never has to carry them.
+    """
+    con = sqlite3.connect(db)
+    codes: dict[str, list[str]] = {}
+    for sid, c in con.execute(
+        "SELECT scopus_id, group_concat(asjc) FROM journal_asjc GROUP BY scopus_id"
+    ):
+        codes[sid] = c.split(",")
+    con.close()
+
+    X, y = [], []
+    with open(path, encoding="utf-8") as f:
+        for line in f:
+            rec = json.loads(line)
+            c = codes.get(rec.get("scopus_id"))
+            if not c or len(c) > max_labels:
+                continue
+            t = (rec.get("text") or "").strip()
+            if len(t) < 120:          # a bare title is not an abstract
+                continue
+            X.append(t)
+            y.append(c)
+    return X, y
+
+
+# ---------------------------------------------------------------------------
+# Evaluation
+# ---------------------------------------------------------------------------
+def evaluate(clf, Xv, Y_true, classes: np.ndarray, parent_of: dict[str, str]) -> dict:
+    """
+    Ranking metrics, not threshold metrics.
+
+    The classifier feeds a *filter*, and the filter takes the top few codes — so
+    what matters is whether a correct code is ranked highly, not whether some
+    probability cleared 0.5. Reporting only micro-F1 at a fixed threshold would
+    flatter a model that is well-calibrated but badly ordered, and vice versa.
+
+    Parent accuracy (2-digit ASJC) is reported alongside because confusing
+    'Information Systems' with 'Computer Science Applications' barely harms the
+    candidate set, while confusing Computer Science with Dentistry destroys it.
+    """
+    P = clf.predict_proba(Xv)
+    order = np.argsort(-P, axis=1)
+    res: dict[str, float] = {}
+
+    for k in (1, 3, 5):
+        topk = order[:, :k]
+        hit = sum(
+            bool(set(classes[topk[i]]) & set(Y_true[i])) for i in range(len(Y_true))
+        )
+        res[f"top{k}_hit"] = hit / len(Y_true)
+
+    # Recall of the full true label set within the top 5 predictions.
+    top5 = order[:, :5]
+    res["top5_label_recall"] = float(np.mean([
+        len(set(classes[top5[i]]) & set(Y_true[i])) / len(Y_true[i])
+        for i in range(len(Y_true))
+    ]))
+
+    top1_parent = [parent_of.get(classes[order[i, 0]]) for i in range(len(Y_true))]
+    res["top1_parent_hit"] = float(np.mean([
+        top1_parent[i] in {parent_of.get(c) for c in Y_true[i]}
+        for i in range(len(Y_true))
+    ]))
+    return res
+
+
+def main() -> None:
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--corpus", choices=["titles", "abstracts"], default="titles")
+    ap.add_argument("--db", default="journals.db")
+    ap.add_argument("--train-jsonl", default="asjc_train.jsonl")
+    ap.add_argument("--out", default="asjc_model.joblib")
+    ap.add_argument("--max-labels", type=int, default=6,
+                    help="drop journals carrying more ASJC codes than this from training")
+    ap.add_argument("--min-per-label", type=int, default=8,
+                    help="drop labels with fewer training examples than this")
+    ap.add_argument("--eval-jsonl", default="asjc_eval.jsonl")
+    ap.add_argument("--max-features", type=int, default=120_000)
+    ap.add_argument("--n-jobs", type=int, default=2,
+                    help="parallel per-label fits. 334 estimators over a large sparse "
+                         "matrix can exhaust memory at -1; 4 is safe on a 16GB Space.")
+    args = ap.parse_args()
+
+    if args.corpus == "titles":
+        X, y = load_titles(args.db, args.max_labels)
+    else:
+        if not Path(args.train_jsonl).exists():
+            sys.exit(f"{args.train_jsonl} not found — run harvest_openalex.py first")
+        X, y = load_abstracts(args.train_jsonl, args.db, args.max_labels)
+
+    print(f"corpus={args.corpus}  examples={len(X)}  "
+          f"mean labels/example={np.mean([len(c) for c in y]):.2f}")
+
+    # Drop labels too rare to learn. They stay resolvable through the LLM
+    # fallback, which does not need training examples.
+    from collections import Counter
+    freq = Counter(c for cs in y for c in cs)
+    keep = {c for c, n in freq.items() if n >= args.min_per_label}
+    y = [[c for c in cs if c in keep] for cs in y]
+    mask = [i for i, cs in enumerate(y) if cs]
+    X, y = [X[i] for i in mask], [y[i] for i in mask]
+    print(f"labels kept: {len(keep)} of {len(freq)}   examples after filtering: {len(X)}")
+
+    mlb = MultiLabelBinarizer()
+    Y = mlb.fit_transform(y)
+
+    Xtr, Xte, Ytr, Yte, ytr_raw, yte_raw = train_test_split(
+        X, Y, y, test_size=0.2, random_state=42
+    )
+
+    # Word n-grams carry the topical signal; character n-grams cover
+    # morphology, non-English titles and the domain shift between titles and
+    # abstracts. Union of both, sublinear tf to damp long abstracts.
+    word = TfidfVectorizer(
+        sublinear_tf=True, strip_accents="unicode", lowercase=True,
+        ngram_range=(1, 2), min_df=2, max_features=args.max_features,
+        stop_words="english", dtype=np.float32,
+    )
+    char = TfidfVectorizer(
+        sublinear_tf=True, strip_accents="unicode", lowercase=True,
+        analyzer="char_wb", ngram_range=(3, 5), min_df=3,
+        max_features=args.max_features // 3, dtype=np.float32,
+    )
+    Xtr_v = sparse.hstack([word.fit_transform(Xtr), char.fit_transform(Xtr)]).tocsr()
+    Xte_v = sparse.hstack([word.transform(Xte), char.transform(Xte)]).tocsr()
+    print(f"features: {Xtr_v.shape[1]}")
+
+    # SGD with modified_huber rather than liblinear logistic regression.
+    # 334 one-vs-rest fits over ~450k sparse features takes liblinear many
+    # minutes and it trips joblib's read-only memmap under parallel fan-out;
+    # SGD fits the same problem in seconds, scales to the ~1M-abstract corpus
+    # the harvest will produce, and modified_huber still exposes predict_proba,
+    # which the LLM fallback needs to decide when to escalate.
+    # log_loss, not modified_huber. modified_huber's predict_proba clips the
+    # margin and normalises, so it returns a saturated 1.0/0.0 with nothing in
+    # between — fine for ranking, useless for deciding when to escalate to the
+    # LLM. log_loss gives genuine logistic probabilities that a threshold can
+    # act on.
+    #
+    # class_weight is left unbalanced on purpose. Balancing inflates scores for
+    # rare labels, which is right for per-label F1 and wrong here: it pushed
+    # 'Rehabilitation' to the top of a cardiology abstract. The output feeds a
+    # ranked filter, so ordering matters more than per-label parity.
+    clf = OneVsRestClassifier(
+        SGDClassifier(
+            loss="log_loss", alpha=1e-6, max_iter=30, tol=1e-4,
+            random_state=42,
+        ),
+        n_jobs=args.n_jobs,
+    )
+    with joblib.parallel_config(backend="loky", max_nbytes=None):
+        clf.fit(Xtr_v, Ytr)
+
+    con = sqlite3.connect(args.db)
+    parent_of = {c: p for c, p in con.execute("SELECT code, parent_code FROM asjc")}
+    names = {c: (n or pn) for c, n, pn in
+             con.execute("SELECT code, name, parent_name FROM asjc")}
+    con.close()
+
+    res = evaluate(clf, Xte_v, yte_raw, mlb.classes_, parent_of)
+    print(f"\nheld-out performance (same distribution as training: {args.corpus})")
+    for k, v in res.items():
+        print(f"  {k:<20} {v:.3f}")
+
+    # The number that actually matters. When training on `titles`, the held-out
+    # score above is measured on 5-word journal titles while production input is
+    # a 200-word abstract — a different and much easier problem. This second
+    # evaluation runs on hand-labelled abstracts across 20 fields, which is the
+    # distribution the classifier is really deployed against.
+    if Path(args.eval_jsonl).exists():
+        ev = [json.loads(l) for l in open(args.eval_jsonl, encoding="utf-8")]
+        Xe = sparse.hstack([word.transform([e["text"] for e in ev]),
+                            char.transform([e["text"] for e in ev])]).tocsr()
+        Pe = clf.predict_proba(Xe)
+        order = np.argsort(-Pe, axis=1)
+        # Two standards, reported separately and deliberately.
+        #
+        # LENIENT counts a hit when *any* acceptable code appears. It overstates
+        # usefulness: for a seismic-engineering abstract, 'Engineering
+        # (miscellaneous)' is in `accept` and scores a hit, but filtering on it
+        # returns nonlinear-analysis maths journals, while the preferred code
+        # 2205 returns concrete-structures journals. Same "hit", different
+        # recommendation quality.
+        #
+        # STRICT requires the single best code — accept[0], the one a subject
+        # librarian would choose. This is the number to watch when retraining on
+        # the abstract corpus, because it is the one that tracks whether the
+        # recommendations are actually good.
+        n = len(ev)
+        hits1 = hits3 = parent1 = 0
+        s5 = s3 = 0
+        print(f"\nabstract eval ({n} hand-labelled abstracts)")
+        for i, e in enumerate(ev):
+            top5 = [mlb.classes_[j] for j in order[i, :5]]
+            top3, top1 = top5[:3], top5[0]
+            pref = e["accept"][0]
+            ok1 = top1 in e["accept"]
+            ok3 = bool(set(top3) & set(e["accept"]))
+            okp = parent_of.get(top1) in {parent_of.get(c) for c in e["accept"]}
+            hits1 += ok1; hits3 += ok3; parent1 += okp
+            s5 += pref in top5
+            s3 += pref in top3
+            flag = "OK  " if pref in top5 else ("weak" if ok3 else "MISS")
+            print(f"  {flag} {e['field'][:28]:<28} pref {pref} "
+                  f"{'in top5' if pref in top5 else 'ABSENT':<8} -> "
+                  f"{', '.join(f'{c}' for c in top3)}")
+        print(f"  lenient: top1 {hits1/n:.2f}  top3 {hits3/n:.2f}  parent {parent1/n:.2f}")
+        print(f"  STRICT : preferred-in-top3 {s3/n:.2f}   preferred-in-top5 {s5/n:.2f}")
+        print("  strict preferred-in-top5 is the number that gates shipping — it "
+              "bounds what the LLM stage can possibly pick.")
+        res.update({"abstract_top1": hits1 / n, "abstract_top3": hits3 / n,
+                    "abstract_top1_parent": parent1 / n,
+                    "abstract_preferred_top3": s3 / n,
+                    "abstract_preferred_top5": s5 / n})
+
+    # Compact the artifact before saving. The bulk is not the vocabulary — it is
+    # 334 dense float64 coefficient vectors (334 x ~160k x 8 bytes ≈ 427 MB
+    # before compression). Casting to float32 halves that with no measurable
+    # effect on ranking, and sparsify() drops the near-zero majority. This
+    # matters operationally: the Space pulls this file on every cold start.
+    for est in clf.estimators_:
+        est.coef_ = est.coef_.astype("float32")
+        est.sparsify()
+    joblib.dump(
+        {"word": word, "char": char, "clf": clf, "classes": mlb.classes_,
+         "names": names, "parent_of": parent_of, "corpus": args.corpus,
+         "metrics": res},
+        args.out, compress=3,
+    )
+    print(f"\nwrote {args.out}  ({Path(args.out).stat().st_size/1e6:.1f} MB)")
+
+    # Sparsified coefficients change the predict path, so re-check the eval
+    # rather than assuming the compaction was lossless.
+    chk = joblib.load(args.out)
+    Xc = sparse.hstack([chk["word"].transform([X[0]]),
+                        chk["char"].transform([X[0]])]).tocsr()
+    assert chk["clf"].predict_proba(Xc).shape[1] == len(mlb.classes_)
+    print("compacted model reloads and predicts correctly")
+
+
+if __name__ == "__main__":
+    main()