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..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
@@ -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()}")
@@ -592,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]:
@@ -615,21 +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
+ 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(
@@ -650,6 +663,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..0fc6a53 100644
--- a/tests/test_sources.py
+++ b/tests/test_sources.py
@@ -318,6 +318,108 @@ 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
+
+ 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")
+ 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)."""