Coverage for src/local_deep_research/journal_quality/data_sources/openalex.py: 89%
83 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +0000
1"""OpenAlex sources data source.
3Downloads the **bulk snapshot** of ~280K journals + conferences from
4``openalex.s3.amazonaws.com`` and writes a compact gzipped JSON file
5used by the in-memory tier-2 lookups.
7This is the public OpenAlex S3 snapshot — CC0 licensed, no auth, no
8rate limits, ~350 MB compressed across ~120 partition files. We stream
9each part, extract the few fields we need, and discard the rest. Total
10wall-clock is ~30–60 s on a normal connection.
12Why bulk snapshot instead of the REST API
13-----------------------------------------
15The previous implementation paginated ``/api/sources`` with
16``per_page=200`` and a 100 ms polite-rate-limiting sleep between
17requests. For ~280K records that meant ~1,400 HTTP requests + ~140 s
18of pure sleep, plus actual transfer time — wall-clock 5–10 minutes,
19and that's the time the user spends staring at the dashboard
20"Download Data" button. The S3 dump is the recommended bulk path per
21OpenAlex docs and finishes in well under a minute.
22"""
24from __future__ import annotations
26import gzip
27import json
28import time
29from pathlib import Path
31from loguru import logger
33from ...utilities.citation_normalizer import normalize_issn
34from .base import DataSource, atomic_replace
36from ._openalex_common import (
37 OPENALEX_S3_BASE,
38 iter_partitions,
39 validate_manifest_entries,
40)
42_OPENALEX_SOURCES_MANIFEST = (
43 f"{OPENALEX_S3_BASE}/data/jsonl/sources/manifest.json"
44)
47class SchemaDriftError(RuntimeError):
48 """OpenAlex renamed / removed a required field in the snapshot.
50 A row-count floor catches the case where the whole fetch collapses,
51 but not the case where every row loads but a key field (``h_index``,
52 ``cited_by_count``) is silently None. We refuse to overwrite the
53 existing snapshot in that case — better to keep the old data than
54 rebuild an all-None DB that would quietly reclassify every journal
55 into the "unknown" quality tier.
56 """
59class OpenAlexSource(DataSource):
60 key = "openalex" # gitleaks:allow
61 name = "OpenAlex"
62 url = "https://openalex.org"
63 dataset_url = (
64 "https://docs.openalex.org/download-all-data/openalex-snapshot"
65 )
66 license = "CC0 1.0"
67 license_url = "https://creativecommons.org/publicdomain/zero/1.0/"
68 description = (
69 "~280K journals and conferences with h-index, impact factor, "
70 "and publisher metadata"
71 )
72 # Deliberately NOT `downloader.REQUIRED_SNAPSHOT_FILENAME`: downloader.py
73 # imports this package (`from .data_sources import ALL_SOURCES,
74 # get_source`) before it defines that constant, and this is a class
75 # attribute (evaluated at class-definition / import time, not inside a
76 # function body), so even a lazy import wouldn't avoid the cycle.
77 filename = "openalex_sources.json.gz"
78 count_label = "OpenAlex sources"
79 auto_download = False # large; user opts in via dashboard
80 required = True # bulk-download fatal-on-failure
81 approx_size_mb = 13.0 # final compact output, NOT the raw snapshot
83 def fetch(self, data_dir: Path, progress_cb=None) -> int:
84 from ...security.safe_requests import (
85 safe_get_with_retries as safe_get,
86 )
88 # 1. Fetch the manifest. Tells us which partition files exist
89 # and how many records to expect — so we can give the user
90 # an accurate progress log instead of just dots.
91 logger.info(
92 f"Fetching OpenAlex sources manifest: {_OPENALEX_SOURCES_MANIFEST}"
93 )
94 # consume_body=True: the manifest is small but a serial
95 # bottleneck for the whole download. A body-read transient
96 # here aborts everything.
97 manifest_resp = safe_get(
98 _OPENALEX_SOURCES_MANIFEST,
99 timeout=30,
100 consume_body=True,
101 require_https=True,
102 )
103 manifest_resp.raise_for_status()
104 manifest = manifest_resp.json()
106 # OpenAlex's 2026-06 "standard-format" snapshot moved the data to
107 # ``data/jsonl/<entity>/`` and renamed the manifest's part list from
108 # ``entries`` to ``files``. Each file entry still has ``url``
109 # (s3://openalex/...) + ``meta.record_count`` / ``meta.content_length``.
110 entries = manifest.get("files", [])
111 total_records = sum(
112 e.get("meta", {}).get("record_count", 0) for e in entries
113 )
114 total_bytes = sum(
115 e.get("meta", {}).get("content_length", 0) for e in entries
116 )
117 logger.info(
118 f"OpenAlex sources snapshot: {len(entries)} parts, "
119 f"{total_records:,} records, "
120 f"{total_bytes / 1024 / 1024:.0f} MB compressed"
121 )
123 # Validate manifest URLs before fetching anything, so a
124 # compromised manifest cannot redirect fetches to an arbitrary
125 # host. Legitimate OpenAlex manifest entries always use the
126 # s3://openalex/ prefix.
127 validate_manifest_entries(entries, "OpenAlex sources")
129 # 2. Stream-process each part. The ``iter_partitions`` helper
130 # in ``_openalex_common`` owns the tmp-file lifecycle + the
131 # first-10 malformed-JSON suppression; we just consume
132 # records and track our own schema-drift counters.
133 type_map = {"journal": "j", "conference": "c"}
134 sources: dict = {}
135 # Raw parse counters feed the ``id``-rename drift check below —
136 # records without an ``id`` are silently skipped at the point of
137 # extraction, so we need to know the ratio to distinguish a
138 # collapsed fetch from a renamed identifier field.
139 parsed_records = 0
140 parsed_with_id = 0
141 start = time.time()
143 for idx, total_parts, records in iter_partitions(
144 entries,
145 data_dir,
146 file_prefix="openalex_sources",
147 label="OpenAlex sources",
148 safe_get=safe_get,
149 ):
150 for rec in records:
151 parsed_records += 1
152 src_id = (rec.get("id") or "").split("/")[-1]
153 if not src_id:
154 continue
155 parsed_with_id += 1
157 stats = rec.get("summary_stats") or {}
158 compact = {
159 "n": rec.get("display_name", ""),
160 "t": type_map.get(rec.get("type", ""), rec.get("type", "")),
161 "h": stats.get("h_index"),
162 "if": stats.get("2yr_mean_citedness"),
163 "cb": rec.get("cited_by_count"),
164 "p": rec.get("host_organization_name") or "",
165 "i": normalize_issn(rec.get("issn_l")),
166 }
167 if rec.get("is_in_doaj"): 167 ↛ 168line 167 didn't jump to line 168 because the condition on line 167 was never true
168 compact["d"] = 1
169 if rec.get("is_core"): 169 ↛ 170line 169 didn't jump to line 170 because the condition on line 169 was never true
170 compact["s"] = 1
171 sources[src_id] = compact
173 if (idx + 1) % 5 == 0 or idx == total_parts - 1: 173 ↛ 181line 173 didn't jump to line 181 because the condition on line 173 was always true
174 elapsed = time.time() - start
175 logger.info(
176 f"OpenAlex sources: processed {idx + 1}/{total_parts} "
177 f"parts ({len(sources):,} records, {elapsed:.0f}s)"
178 )
179 # Report on EVERY partition, not just every 5th — the UI
180 # needs smoother updates than the human-readable log.
181 if progress_cb is not None: 181 ↛ 182line 181 didn't jump to line 182 because the condition on line 181 was never true
182 try:
183 progress_cb(
184 idx + 1,
185 total_parts,
186 f"{len(sources):,} records",
187 )
188 except Exception:
189 logger.debug(
190 "OpenAlex progress_cb raised; continuing",
191 exc_info=True,
192 )
194 # ``id``-rename drift runs *before* the row-count floor: a
195 # renamed identifier makes every record drop out at parse time
196 # (src_id empty → continue), so ``sources`` is empty and the
197 # row-count floor would fire first with a generic RuntimeError
198 # that hides the actual cause.
199 if parsed_records >= 10_000 and parsed_with_id == 0:
200 raise SchemaDriftError(
201 f"OpenAlex snapshot parsed {parsed_records:,} records but "
202 "none carried an 'id' field — the source identifier may "
203 "have been renamed (e.g. to 'source_id'). Refusing to "
204 "overwrite existing data — please check "
205 "https://docs.openalex.org/download-all-data/"
206 "snapshot-data-format for schema changes."
207 )
209 # 3. Write the compact snapshot in the same shape the existing
210 # build pipeline expects (`{"s": {src_id: compact}}`).
211 # Sanity check: OpenAlex normally has ~280K sources. If the
212 # fetch silently returned a tiny subset (e.g., every partition
213 # returned an empty shard), refuse to overwrite existing data.
214 _MIN_OPENALEX_SOURCES = 10_000
215 if len(sources) < _MIN_OPENALEX_SOURCES: 215 ↛ 216line 215 didn't jump to line 216 because the condition on line 215 was never true
216 raise RuntimeError(
217 f"OpenAlex sources: suspiciously few records "
218 f"({len(sources):,} < {_MIN_OPENALEX_SOURCES:,}); "
219 "refusing to overwrite existing data"
220 )
222 # Field-level schema drift detection. The row-count floor above
223 # catches a collapsed fetch, but not a silent upstream rename.
224 # The journal-only sample avoids false-triggering on snapshots
225 # that skew to non-journal types (conferences, repositories,
226 # etc.) which legitimately lack ``h_index``.
227 _SCHEMA_SAMPLE_SIZE = 100
228 journal_sample = [r for r in sources.values() if r.get("t") == "j"][
229 :_SCHEMA_SAMPLE_SIZE
230 ]
231 if len(journal_sample) >= _SCHEMA_SAMPLE_SIZE:
232 has_hindex = any(r.get("h") is not None for r in journal_sample)
233 has_cited = any(r.get("cb") is not None for r in journal_sample)
234 if not has_hindex or not has_cited:
235 raise SchemaDriftError(
236 "OpenAlex snapshot appears to have renamed a required "
237 "field: "
238 f"h_index present in journal sample={has_hindex}, "
239 f"cited_by_count present in journal sample={has_cited}. "
240 "Refusing to overwrite existing data — please check "
241 "https://docs.openalex.org/download-all-data/"
242 "snapshot-data-format for schema changes."
243 )
244 else:
245 # The row-count floor above already refuses a collapsed
246 # fetch. This branch only fires in unusual cases (truncated
247 # test snapshot, snapshot with very few journal-typed
248 # sources). Log at info so operators see the drift check
249 # was bypassed — debug would be invisible at production
250 # log levels.
251 logger.info(
252 "OpenAlex schema-drift check skipped: "
253 f"only {len(journal_sample)} journal source(s) in sample "
254 f"(< {_SCHEMA_SAMPLE_SIZE} required)"
255 )
257 output = data_dir / self.filename
258 tmp = data_dir / f"{self.filename}.tmp"
259 with gzip.open(tmp, "wt", encoding="utf-8") as f:
260 json.dump({"s": sources}, f)
261 atomic_replace(tmp, output)
263 elapsed = time.time() - start
264 logger.info(
265 f"OpenAlex sources: saved {len(sources):,} sources in {elapsed:.0f}s"
266 )
267 return len(sources)