From cf4e3d8ffed368eae8932df440d145c48c252029 Mon Sep 17 00:00:00 2001 From: caufieldjh Date: Thu, 9 Jul 2026 14:34:53 -0400 Subject: [PATCH 1/2] feat(pmid): retrieve PublicationTypeList from PubMed metadata MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes #56. Surface PubMed publication types (MeSH publication-type descriptors, e.g. "Case Reports", "Clinical Trial", "Review" — see https://www.nlm.nih.gov/mesh/pubtypes.html) end-to-end. - ReferenceContent gains a publication_types field. - PMIDSource fetches the article XML once and parses both MeSH terms and publication types from it (no extra NCBI round-trip): _fetch_pubmed_xml plus pure parsers _parse_mesh_terms / _parse_publication_types. - Persisted to and reloaded from the disk cache frontmatter. - Exposed in lookup json/yaml/text output. Tests written first: XML parse (present/absent), fetch() population with mocked Entrez.efetch, cache round-trip, plus doctests on the parsers. Co-Authored-By: Claude Opus 4.8 --- src/linkml_reference_validator/cli/lookup.py | 3 + .../etl/reference_fetcher.py | 13 +++ .../etl/sources/pmid.py | 84 ++++++++++++++++--- src/linkml_reference_validator/models.py | 4 + tests/test_reference_fetcher.py | 17 ++++ tests/test_sources.py | 84 +++++++++++++++++++ 6 files changed, 194 insertions(+), 11 deletions(-) diff --git a/src/linkml_reference_validator/cli/lookup.py b/src/linkml_reference_validator/cli/lookup.py index a0e1d14..2d60202 100644 --- a/src/linkml_reference_validator/cli/lookup.py +++ b/src/linkml_reference_validator/cli/lookup.py @@ -71,6 +71,7 @@ def _reference_to_dict(reference: ReferenceContent) -> dict: "year": reference.year, "doi": reference.doi, "keywords": reference.keywords, + "publication_types": reference.publication_types, "content_type": reference.content_type, "content": reference.content, } @@ -185,6 +186,8 @@ def _format_as_text(reference: ReferenceContent) -> str: lines.append(f"DOI: {reference.doi}") if reference.keywords: lines.append(f"Keywords: {', '.join(reference.keywords)}") + if reference.publication_types: + lines.append(f"Publication types: {', '.join(reference.publication_types)}") lines.append(f"Content type: {reference.content_type}") if reference.supplementary_files: lines.append("") diff --git a/src/linkml_reference_validator/etl/reference_fetcher.py b/src/linkml_reference_validator/etl/reference_fetcher.py index a09d14e..b07b3c0 100644 --- a/src/linkml_reference_validator/etl/reference_fetcher.py +++ b/src/linkml_reference_validator/etl/reference_fetcher.py @@ -491,6 +491,10 @@ def _save_to_disk(self, reference: ReferenceContent) -> None: lines.append("keywords:") for keyword in reference.keywords: lines.append(f"- {self._quote_yaml_value(keyword)}") + if reference.publication_types: + lines.append("publication_types:") + for publication_type in reference.publication_types: + lines.append(f"- {self._quote_yaml_value(publication_type)}") lines.append(f"content_type: {reference.content_type}") if reference.is_preprint is not None: lines.append(f"is_preprint: {str(reference.is_preprint).lower()}") @@ -631,6 +635,14 @@ def _load_markdown_format( else: keywords = None + publication_types = frontmatter.get("publication_types") + if publication_types and isinstance(publication_types, list): + publication_types = publication_types + elif publication_types: + publication_types = [publication_types] + else: + publication_types = None + # Parse supplementary files supplementary_files = self._parse_supplementary_files( frontmatter.get("supplementary_files") @@ -650,6 +662,7 @@ def _load_markdown_format( year=str(frontmatter.get("year")) if frontmatter.get("year") else None, doi=frontmatter.get("doi"), keywords=keywords, + publication_types=publication_types, supplementary_files=supplementary_files, metadata=metadata, full_text_provider=frontmatter.get("full_text_provider"), diff --git a/src/linkml_reference_validator/etl/sources/pmid.py b/src/linkml_reference_validator/etl/sources/pmid.py index e2c8ffd..0a7fd5a 100644 --- a/src/linkml_reference_validator/etl/sources/pmid.py +++ b/src/linkml_reference_validator/etl/sources/pmid.py @@ -138,7 +138,11 @@ def fetch( doi = str(record_dict.get("DOI", "")) if record_dict.get("DOI") else "" abstract = self._fetch_abstract(pmid, config) - keywords = self._fetch_mesh_terms(pmid, config) + article_xml = self._fetch_pubmed_xml(pmid, config) + keywords = self._parse_mesh_terms(article_xml) if article_xml else None + publication_types = ( + self._parse_publication_types(article_xml) if article_xml else None + ) content: Optional[str] = abstract content_type = "abstract_only" if abstract else "unavailable" @@ -164,6 +168,7 @@ def fetch( year=year, doi=doi, keywords=keywords, + publication_types=publication_types, metadata=metadata, ) @@ -207,22 +212,20 @@ def _fetch_abstract( return None - def _fetch_mesh_terms( + def _fetch_pubmed_xml( self, pmid: str, config: ReferenceValidationConfig - ) -> Optional[list[str]]: - """Fetch MeSH terms for a PMID from PubMed XML. + ) -> Optional[BeautifulSoup]: + """Fetch and parse the PubMed article XML for a PMID. + + A single efetch call backs both MeSH terms and publication types, so we + don't round-trip to NCBI twice for the same document. Args: pmid: PubMed ID config: Configuration for rate limiting Returns: - List of MeSH terms if available - - Examples: - >>> source = PMIDSource() - >>> # Would return MeSH terms like: - >>> # ['Adaptation, Physiological/genetics', 'Climate Change', ...] + Parsed BeautifulSoup document, or None if nothing was returned """ time.sleep(config.rate_limit_delay) @@ -234,7 +237,28 @@ def _fetch_mesh_terms( if isinstance(xml_content, bytes): xml_content = xml_content.decode("utf-8") - soup = BeautifulSoup(xml_content, "xml") + if not xml_content: + return None + + return BeautifulSoup(xml_content, "xml") + + def _parse_mesh_terms(self, soup: BeautifulSoup) -> Optional[list[str]]: + """Parse MeSH terms from a PubMed article XML document. + + Args: + soup: Parsed PubMed article XML + + Returns: + List of MeSH terms if available + + Examples: + >>> from bs4 import BeautifulSoup + >>> xml = ''' + ... Climate Change + ... ''' + >>> PMIDSource()._parse_mesh_terms(BeautifulSoup(xml, "xml")) + ['Climate Change'] + """ mesh_list = soup.find("MeshHeadingList") if not mesh_list: @@ -254,6 +278,44 @@ def _fetch_mesh_terms( return terms if terms else None + def _parse_publication_types( + self, soup: BeautifulSoup + ) -> Optional[list[str]]: + """Parse PublicationTypeList from a PubMed article XML document. + + Publication types are MeSH publication-type descriptors + (https://www.nlm.nih.gov/mesh/pubtypes.html) that classify the source, + e.g. "Case Reports", "Clinical Trial", "Review". Nearly every record + carries the generic "Journal Article" type; it is retained as-is. + + Args: + soup: Parsed PubMed article XML + + Returns: + List of publication type labels if available + + Examples: + >>> from bs4 import BeautifulSoup + >>> xml = ''' + ... Journal Article + ... Case Reports + ... ''' + >>> PMIDSource()._parse_publication_types(BeautifulSoup(xml, "xml")) + ['Journal Article', 'Case Reports'] + """ + type_list = soup.find("PublicationTypeList") + + if not type_list: + return None + + types = [ + text + for pt in type_list.find_all("PublicationType") + if (text := pt.get_text().strip()) + ] + + return types if types else None + def _fetch_pmc_fulltext( self, pmid: str, config: ReferenceValidationConfig ) -> tuple[Optional[str], str]: diff --git a/src/linkml_reference_validator/models.py b/src/linkml_reference_validator/models.py index b2734ec..682247e 100644 --- a/src/linkml_reference_validator/models.py +++ b/src/linkml_reference_validator/models.py @@ -708,6 +708,10 @@ class ReferenceContent: year: Optional[str] = None doi: Optional[str] = None keywords: Optional[list[str]] = None # MeSH terms, subjects, tags + # PubMed PublicationTypeList values (https://www.nlm.nih.gov/mesh/pubtypes.html), + # e.g. ["Journal Article", "Case Reports", "Clinical Trial"]. These classify the + # source so downstream KBs can reason about evidence type. None when unknown. + publication_types: Optional[list[str]] = None supplementary_files: Optional[list[SupplementaryFile]] = None metadata: dict = field(default_factory=dict) full_text_provider: Optional[str] = None diff --git a/tests/test_reference_fetcher.py b/tests/test_reference_fetcher.py index 1d2ede2..d0a7e37 100644 --- a/tests/test_reference_fetcher.py +++ b/tests/test_reference_fetcher.py @@ -140,6 +140,23 @@ def test_save_and_load_preprint_metadata(fetcher, tmp_path): assert loaded.peer_review_status == "preprint" +def test_save_and_load_publication_types(fetcher, tmp_path): + """Publication types round-trip through the disk cache frontmatter.""" + ref = ReferenceContent( + reference_id="PMID:12345678", + title="An illustrative case", + content="Case description.", + content_type="abstract_only", + publication_types=["Journal Article", "Case Reports"], + ) + + fetcher._save_to_disk(ref) + loaded = fetcher._load_from_disk("PMID:12345678") + + assert loaded is not None + assert loaded.publication_types == ["Journal Article", "Case Reports"] + + def test_save_and_load_non_preprint_leaves_status_unset(fetcher, tmp_path): """A record with no preprint status must not gain one via the cache.""" ref = ReferenceContent( diff --git a/tests/test_sources.py b/tests/test_sources.py index d79517b..e6c0f93 100644 --- a/tests/test_sources.py +++ b/tests/test_sources.py @@ -318,6 +318,90 @@ def test_get_pmcid_handles_entrez_error( assert result is None handle.close.assert_called_once() + # --- Publication types (issue #56) ----------------------------------- + + _ARTICLE_XML = """ + + + +
+ An illustrative case + + Journal Article + Case Reports + Randomized Controlled Trial + +
+ + + Calcimycin + + + + + + """ + + def test_parse_publication_types(self, source): + """Should extract every PublicationType value from article XML.""" + from bs4 import BeautifulSoup + + soup = BeautifulSoup(self._ARTICLE_XML, "xml") + types = source._parse_publication_types(soup) + + assert types == [ + "Journal Article", + "Case Reports", + "Randomized Controlled Trial", + ] + + def test_parse_publication_types_absent(self, source): + """Should return None when there is no PublicationTypeList.""" + from bs4 import BeautifulSoup + + soup = BeautifulSoup( + "", "xml" + ) + + assert source._parse_publication_types(soup) is None + + @patch("linkml_reference_validator.etl.sources.pmid.Entrez.efetch") + @patch("linkml_reference_validator.etl.sources.pmid.Entrez.esummary") + @patch("linkml_reference_validator.etl.sources.pmid.Entrez.read") + def test_fetch_populates_publication_types( + self, mock_read, mock_esummary, mock_efetch, source, config + ): + """fetch() should surface publication types on the ReferenceContent.""" + mock_read.return_value = [ + { + "Title": "An illustrative case", + "AuthorList": ["Smith J"], + "Source": "J Test", + "PubDate": "2020 Jan", + "DOI": "10.1/x", + } + ] + + def _efetch(*args, **kwargs): + handle = MagicMock() + if kwargs.get("rettype") == "abstract": + handle.read.return_value = "x" * 100 + else: + handle.read.return_value = self._ARTICLE_XML + return handle + + mock_efetch.side_effect = _efetch + + ref = source.fetch("12345678", config) + + assert ref is not None + assert ref.publication_types == [ + "Journal Article", + "Case Reports", + "Randomized Controlled Trial", + ] + assert ref.keywords == ["Calcimycin"] + class TestDOISource: """Tests for DOISource (refactored from ReferenceFetcher).""" From 9caecc971c78771867ee139f93d9635489f205f8 Mon Sep 17 00:00:00 2001 From: caufieldjh Date: Thu, 9 Jul 2026 14:49:38 -0400 Subject: [PATCH 2/2] =?UTF-8?q?refactor(cache):=20address=20PR=20review=20?= =?UTF-8?q?=E2=80=94=20collapse=20list-coercion=20+=20edge=20tests?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Collapse the repeated authors/keywords/publication_types frontmatter list-coercion into a shared _as_optional_list helper (review note 1). - Add empty-list edge-case tests for _parse_publication_types and _parse_mesh_terms, covering the `return x if x else None` branch (note 3). Deferring note 2 (folding the abstract into the single article-XML fetch): the text-mode abstract fetch returns a richer formatted citation block than the raw AbstractText XML nodes and feeds supporting-text validation, so the swap is a behavior change rather than a mechanical cleanup — left as a follow-up, as the reviewer flagged it deferrable. Co-Authored-By: Claude Opus 4.8 --- .../etl/reference_fetcher.py | 49 ++++++++++--------- tests/test_sources.py | 18 +++++++ 2 files changed, 43 insertions(+), 24 deletions(-) diff --git a/src/linkml_reference_validator/etl/reference_fetcher.py b/src/linkml_reference_validator/etl/reference_fetcher.py index b07b3c0..ce4a2c3 100644 --- a/src/linkml_reference_validator/etl/reference_fetcher.py +++ b/src/linkml_reference_validator/etl/reference_fetcher.py @@ -7,7 +7,7 @@ import logging import re from pathlib import Path -from typing import Optional +from typing import Any, Optional from ruamel.yaml import YAML # type: ignore @@ -596,6 +596,25 @@ def _load_from_disk(self, reference_id: str) -> Optional[ReferenceContent]: else: return self._load_legacy_format(content_text, reference_id) + @staticmethod + def _as_optional_list(value: Any) -> Optional[list]: + """Normalise a frontmatter value into an optional list. + + YAML may parse a single-item field as a scalar rather than a list; + this coerces such values back to a list and maps empties to None. + + Examples: + >>> ReferenceFetcher._as_optional_list(["a", "b"]) + ['a', 'b'] + >>> ReferenceFetcher._as_optional_list("solo") + ['solo'] + >>> ReferenceFetcher._as_optional_list(None) is None + True + """ + if not value: + return None + return value if isinstance(value, list) else [value] + def _load_markdown_format( self, content_text: str, reference_id: str ) -> Optional[ReferenceContent]: @@ -619,29 +638,11 @@ def _load_markdown_format( content = self._extract_content_from_markdown(body) - authors = frontmatter.get("authors") - if authors and isinstance(authors, list): - authors = authors - elif authors: - authors = [authors] - else: - authors = None - - keywords = frontmatter.get("keywords") - if keywords and isinstance(keywords, list): - keywords = keywords - elif keywords: - keywords = [keywords] - else: - keywords = None - - publication_types = frontmatter.get("publication_types") - if publication_types and isinstance(publication_types, list): - publication_types = publication_types - elif publication_types: - publication_types = [publication_types] - else: - publication_types = None + authors = self._as_optional_list(frontmatter.get("authors")) + keywords = self._as_optional_list(frontmatter.get("keywords")) + publication_types = self._as_optional_list( + frontmatter.get("publication_types") + ) # Parse supplementary files supplementary_files = self._parse_supplementary_files( diff --git a/tests/test_sources.py b/tests/test_sources.py index e6c0f93..0fc6a53 100644 --- a/tests/test_sources.py +++ b/tests/test_sources.py @@ -365,6 +365,24 @@ def test_parse_publication_types_absent(self, source): assert source._parse_publication_types(soup) is None + def test_parse_publication_types_empty_list(self, source): + """A present-but-empty PublicationTypeList should yield None.""" + from bs4 import BeautifulSoup + + soup = BeautifulSoup( + "", "xml" + ) + + assert source._parse_publication_types(soup) is None + + def test_parse_mesh_terms_empty_list(self, source): + """A present-but-empty MeshHeadingList should yield None.""" + from bs4 import BeautifulSoup + + soup = BeautifulSoup("", "xml") + + assert source._parse_mesh_terms(soup) is None + @patch("linkml_reference_validator.etl.sources.pmid.Entrez.efetch") @patch("linkml_reference_validator.etl.sources.pmid.Entrez.esummary") @patch("linkml_reference_validator.etl.sources.pmid.Entrez.read")