Add PubMed metadata source

plugins/pubmed_metadata/README.md

@@ -0,0 +1,68 @@
+# PubMed metadata
+
+This source ingests the RENCI `pubmed2db` snapshot published at
+<https://stars.renci.org/var/babel_outputs/pubmed2db/2026jun30/>. The pinned
+release contains 16 gzip-compressed NDJSON shards (about 16.85 GB compressed).
+
+Each upstream record is stored under a `pubmed` source key in a standalone
+PubMed index:
+
+```json
+{
+  "_id": "PMID:12345678",
+  "pubmed": {
+    "journal": {
+      "name": "Example Journal",
+      "abbr": "Example J"
+    },
+    "title": "Example title",
+    "vol": "1",
+    "iss": "2",
+    "pub_date": "2026-06-30",
+    "abstract": "Example abstract"
+  }
+}
+```
+
+The parser streams each compressed shard without materializing it in memory. It
+requires the exact ten-field upstream schema, string values, valid
+`PMID:<digits>` identifiers, valid UTF-8, and valid gzip/NDJSON input. A
+malformed record fails the upload with the shard and line number rather than
+producing a partial or silently altered document. The default storage also
+treats duplicate IDs as an error.
+
+The parser converts NLM month abbreviations to numbers and preserves the
+available publication-date precision: `YYYY-MM-DD` when all parts exist,
+`YYYY-MM` when the day is missing, and `YYYY` when only the year exists. It
+omits `pub_date` when the year is absent. Elasticsearch maps all three forms as
+a date; partial dates sort at the beginning of their represented period.
+
+Downloads and uploads are each capped at four concurrent shards. For
+compatibility with pending.api's BioThings Hub 0.12.x dependency, the uploader
+partitions the 16 shards among four workers instead of relying only on the
+per-source concurrency setting introduced in BioThings Hub 1.x. Abstracts are
+retained in Elasticsearch `_source` but are not indexed or sortable. The other
+metadata fields are indexed. `title` supports full-text matching and relevance
+scoring but is not sortable. `journal.name` uses its `.raw` keyword subfield
+when sorting; the keyword and date fields are directly sortable. Because this
+is a very large source, the uploader retains only one previous MongoDB source
+collection instead of the BioThings default of ten.
+
+Build `pubmed_metadata` by itself into a versioned `pubmed_*` Elasticsearch
+index. After validating the index, point the stable `annotator-pubmed` alias to
+it. NodeAnnotator routes `PMID:` identifiers to that alias.
+
+For subsequent releases, move the alias from the previous index to the newly
+validated index in one atomic Elasticsearch alias update. Keep the previous
+index temporarily for rollback and remove it separately after validation.
+Build configuration and alias state are deployment state and are not stored in
+this repository.
+
+This is a full snapshot rather than an incremental feed. DOI and PMC identifiers
+are not present in this export. To adopt a newer snapshot, update `RELEASE` in
+`static.py` and verify that its shard count and schema are unchanged.
+
+The export is produced by
+[TranslatorSRI/pubmed2db](https://github.com/TranslatorSRI/pubmed2db) from NLM
+PubMed data. Downstream use must follow the
+[NLM PubMed terms and conditions](https://ftp.ncbi.nlm.nih.gov/pubmed/baseline/README.txt).

plugins/pubmed_metadata/__init__.py

@@ -0,0 +1,3 @@
+# flake8: noqa F401
+from .dumper import PubMedMetadataDumper
+from .uploader import PubMedMetadataUploader

plugins/pubmed_metadata/dumper.py

@@ -0,0 +1,26 @@
+"""Dumper for the RENCI PubMed metadata export."""
+
+from pathlib import Path
+
+from biothings import config
+from biothings.hub.dataload.dumper import LastModifiedHTTPDumper
+
+from .static import PUBMED_METADATA_URLS, RELEASE
+
+
+class PubMedMetadataDumper(LastModifiedHTTPDumper):
+    """Download the pinned PubMed metadata snapshot in bounded parallelism."""
+
+    SRC_NAME = "pubmed_metadata"
+    SRC_ROOT_FOLDER = Path(config.DATA_ARCHIVE_ROOT) / SRC_NAME
+    SRC_URLS = list(PUBMED_METADATA_URLS)
+
+    ARCHIVE = True
+    AUTO_UPLOAD = True
+    MAX_PARALLEL_DUMP = 4
+    RESOLVE_FILENAME = False
+    SCHEDULE = None
+
+    def set_release(self) -> None:
+        """Use the date encoded in the pinned upstream directory."""
+        self.release = RELEASE

plugins/pubmed_metadata/mapping.py

@@ -0,0 +1,46 @@
+"""Elasticsearch mapping for transformed PubMed metadata documents."""
+
+from copy import deepcopy
+
+
+PUBMED_DATE_FORMAT = "strict_date||strict_year_month||strict_year"
+MAX_SORTABLE_TEXT_LENGTH = 8191
+
+
+def _sortable_text() -> dict:
+    return {
+        "type": "text",
+        "fields": {
+            "raw": {
+                "type": "keyword",
+                "ignore_above": MAX_SORTABLE_TEXT_LENGTH,
+            }
+        },
+    }
+
+
+PUBMED_METADATA_MAPPING = {
+    "pubmed": {
+        "properties": {
+            "journal": {
+                "properties": {
+                    "name": _sortable_text(),
+                    "abbr": {"type": "keyword"},
+                },
+            },
+            "title": {"type": "text"},
+            "vol": {"type": "keyword"},
+            "iss": {"type": "keyword"},
+            "pub_date": {
+                "type": "date",
+                "format": PUBMED_DATE_FORMAT,
+            },
+            "abstract": {"type": "text", "index": False},
+        },
+    }
+}
+
+
+def get_pubmed_metadata_mapping() -> dict:
+    """Return an independent mapping because Hub consumers may mutate it."""
+    return deepcopy(PUBMED_METADATA_MAPPING)

plugins/pubmed_metadata/parser.py

@@ -0,0 +1,179 @@
+"""Streaming parser and validation for PubMed metadata NDJSON shards."""
+
+import calendar
+import gzip
+import json
+import re
+from datetime import date
+from pathlib import Path
+from typing import Iterator
+
+
+EXPECTED_RECORD_FIELDS = (
+    "id",
+    "journal_name",
+    "journal_abbrev",
+    "article_title",
+    "volume",
+    "issue",
+    "pub_year",
+    "pub_month",
+    "pub_day",
+    "abstract",
+)
+PMID_PATTERN = re.compile(r"^PMID:[1-9][0-9]*$")
+YEAR_PATTERN = re.compile(r"^[0-9]{4}$")
+NUMERIC_DATE_PART_PATTERN = re.compile(r"^[0-9]{1,2}$")
+MONTH_NUMBERS = {
+    month_abbreviation.lower(): month_number
+    for month_number, month_abbreviation in enumerate(calendar.month_abbr)
+    if month_abbreviation
+}
+
+
+class PubMedMetadataValidationError(ValueError):
+    """Raised when an input shard or record does not match the contract."""
+
+
+def _location(source: str, line_number: int | None) -> str:
+    if line_number is None:
+        return source
+    return f"{source}:{line_number}"
+
+
+def _parse_month(value: str, location: str) -> int:
+    if NUMERIC_DATE_PART_PATTERN.fullmatch(value):
+        month = int(value)
+    else:
+        month = MONTH_NUMBERS.get(value.lower(), 0)
+
+    if not 1 <= month <= 12:
+        raise PubMedMetadataValidationError(f"{location}: invalid publication month {value!r}")
+    return month
+
+
+def _build_pub_date(record: dict, location: str) -> str | None:
+    year_value = record["pub_year"]
+    month_value = record["pub_month"]
+    day_value = record["pub_day"]
+
+    if not year_value:
+        if month_value or day_value:
+            raise PubMedMetadataValidationError(f"{location}: publication month/day requires a year")
+        return None
+
+    if YEAR_PATTERN.fullmatch(year_value) is None or int(year_value) == 0:
+        raise PubMedMetadataValidationError(f"{location}: invalid publication year {year_value!r}")
+
+    if not month_value:
+        if day_value:
+            raise PubMedMetadataValidationError(f"{location}: publication day requires a month")
+        return year_value
+
+    month = _parse_month(month_value, location)
+    year_month = f"{year_value}-{month:02d}"
+    if not day_value:
+        return year_month
+
+    if NUMERIC_DATE_PART_PATTERN.fullmatch(day_value) is None:
+        raise PubMedMetadataValidationError(f"{location}: invalid publication day {day_value!r}")
+
+    day = int(day_value)
+    try:
+        publication_date = date(int(year_value), month, day)
+    except ValueError as error:
+        raise PubMedMetadataValidationError(
+            f"{location}: invalid publication date {year_value!r}/{month_value!r}/{day_value!r}"
+        ) from error
+    return publication_date.isoformat()
+
+
+def transform_pubmed_metadata_record(
+    record: object,
+    *,
+    source: str = "<record>",
+    line_number: int | None = None,
+) -> dict:
+    """Validate and namespace one upstream PubMed record."""
+    location = _location(source, line_number)
+    if not isinstance(record, dict):
+        raise PubMedMetadataValidationError(
+            f"{location}: expected a JSON object, got {type(record).__name__}"
+        )
+
+    expected_fields = set(EXPECTED_RECORD_FIELDS)
+    actual_fields = set(record)
+    missing_fields = sorted(expected_fields - actual_fields)
+    extra_fields = sorted(str(field) for field in actual_fields - expected_fields)
+    if missing_fields or extra_fields:
+        details = []
+        if missing_fields:
+            details.append(f"missing fields: {', '.join(missing_fields)}")
+        if extra_fields:
+            details.append(f"unexpected fields: {', '.join(extra_fields)}")
+        raise PubMedMetadataValidationError(f"{location}: {'; '.join(details)}")
+
+    non_string_fields = sorted(
+        field for field in EXPECTED_RECORD_FIELDS if not isinstance(record[field], str)
+    )
+    if non_string_fields:
+        raise PubMedMetadataValidationError(
+            f"{location}: fields must contain strings: {', '.join(non_string_fields)}"
+        )
+
+    pubmed_id = record["id"]
+    if PMID_PATTERN.fullmatch(pubmed_id) is None:
+        raise PubMedMetadataValidationError(f"{location}: invalid PubMed identifier {pubmed_id!r}")
+
+    pubmed = {
+        "journal": {
+            "name": record["journal_name"],
+            "abbr": record["journal_abbrev"],
+        },
+        "title": record["article_title"],
+        "vol": record["volume"],
+        "iss": record["issue"],
+        "abstract": record["abstract"],
+    }
+    pub_date = _build_pub_date(record, location)
+    if pub_date is not None:
+        pubmed["pub_date"] = pub_date
+
+    return {
+        "_id": pubmed_id,
+        "pubmed": pubmed,
+    }
+
+
+def iter_pubmed_metadata_documents(data_path: str | Path) -> Iterator[dict]:
+    """Yield validated documents from a gzip-compressed NDJSON shard."""
+    path = Path(data_path)
+    try:
+        with gzip.open(
+            path,
+            mode="rt",
+            encoding="utf-8",
+            errors="strict",
+            newline="",
+        ) as input_file:
+            for line_number, line in enumerate(input_file, start=1):
+                if not line.strip():
+                    raise PubMedMetadataValidationError(
+                        f"{path}:{line_number}: blank lines are not allowed"
+                    )
+                try:
+                    record = json.loads(line)
+                except json.JSONDecodeError as error:
+                    raise PubMedMetadataValidationError(
+                        f"{path}:{line_number}: invalid JSON: {error.msg}"
+                    ) from error
+
+                yield transform_pubmed_metadata_record(
+                    record,
+                    source=str(path),
+                    line_number=line_number,
+                )
+    except PubMedMetadataValidationError:
+        raise
+    except (EOFError, OSError, UnicodeError) as error:
+        raise PubMedMetadataValidationError(f"{path}: unable to read gzip stream: {error}") from error

plugins/pubmed_metadata/static.py

@@ -0,0 +1,13 @@
+"""Release-specific constants for the PubMed metadata source."""
+
+RELEASE = "2026jun30"
+BASE_URL = f"https://stars.renci.org/var/babel_outputs/pubmed2db/{RELEASE}/"
+SHARD_COUNT = 16
+
+PUBMED_METADATA_FILES = tuple(
+    f"pubmed_metadata_{index:05d}.ndjson.gz" for index in range(SHARD_COUNT)
+)
+PUBMED_METADATA_URLS = tuple(f"{BASE_URL}{filename}" for filename in PUBMED_METADATA_FILES)
+
+NLM_TERMS_URL = "https://ftp.ncbi.nlm.nih.gov/pubmed/baseline/README.txt"
+PUBMED2DB_URL = "https://github.com/TranslatorSRI/pubmed2db"

plugins/pubmed_metadata/uploader.py

@@ -0,0 +1,56 @@
+"""Uploader for the RENCI PubMed metadata export."""
+
+from pathlib import Path
+
+from biothings.hub.dataload.uploader import ParallelizedSourceUploader
+
+from .mapping import get_pubmed_metadata_mapping
+from .parser import iter_pubmed_metadata_documents
+from .static import BASE_URL, NLM_TERMS_URL, PUBMED2DB_URL, PUBMED_METADATA_FILES
+
+
+class PubMedMetadataUploader(ParallelizedSourceUploader):
+    """Stream each upstream shard into one source collection."""
+
+    name = "pubmed_metadata"
+    MAX_PARALLEL_UPLOAD = 4
+    keep_archive = 1
+    __metadata__ = {
+        "src_meta": {
+            "url": BASE_URL,
+            "license": "NLM PubMed Terms and Conditions",
+            "license_url": NLM_TERMS_URL,
+            "description": (
+                "PubMed citation metadata exported for Translator by "
+                f"{PUBMED2DB_URL}"
+            ),
+        }
+    }
+
+    def jobs(self) -> list[tuple[tuple[str, ...]]]:
+        data_folder = Path(self.data_folder)
+        shard_paths = [data_folder / filename for filename in PUBMED_METADATA_FILES]
+        missing_paths = [path.name for path in shard_paths if not path.is_file()]
+        if missing_paths:
+            raise FileNotFoundError(
+                "PubMed metadata upload requires all 16 shards; missing: "
+                + ", ".join(missing_paths)
+            )
+
+        # pending.api still uses BioThings Hub 0.12.x, whose parallel uploader
+        # does not enforce MAX_PARALLEL_UPLOAD. Partitioning the shards into
+        # four jobs preserves the source's intended concurrency on both the
+        # 0.12.x and 1.x Hub implementations.
+        shard_groups = [
+            tuple(str(path) for path in shard_paths[group_index:: self.MAX_PARALLEL_UPLOAD])
+            for group_index in range(self.MAX_PARALLEL_UPLOAD)
+        ]
+        return [(group,) for group in shard_groups]
+
+    def load_data(self, data_paths: tuple[str, ...]):
+        for data_path in data_paths:
+            yield from iter_pubmed_metadata_documents(data_path)
+
+    @classmethod
+    def get_mapping(cls) -> dict:
+        return get_pubmed_metadata_mapping()