feat(pmid): retrieve PublicationTypeList from PubMed metadata

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("")

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"),

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 = '''<MeshHeadingList><MeshHeading>
+            ...   <DescriptorName>Climate Change</DescriptorName>
+            ... </MeshHeading></MeshHeadingList>'''
+            >>> 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 = '''<PublicationTypeList>
+            ...   <PublicationType UI="D016428">Journal Article</PublicationType>
+            ...   <PublicationType UI="D002363">Case Reports</PublicationType>
+            ... </PublicationTypeList>'''
+            >>> 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]:

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

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(

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 = """<?xml version="1.0"?>
+    <PubmedArticleSet>
+      <PubmedArticle>
+        <MedlineCitation>
+          <Article>
+            <ArticleTitle>An illustrative case</ArticleTitle>
+            <PublicationTypeList>
+              <PublicationType UI="D016428">Journal Article</PublicationType>
+              <PublicationType UI="D002363">Case Reports</PublicationType>
+              <PublicationType UI="D016449">Randomized Controlled Trial</PublicationType>
+            </PublicationTypeList>
+          </Article>
+          <MeshHeadingList>
+            <MeshHeading>
+              <DescriptorName UI="D000001">Calcimycin</DescriptorName>
+            </MeshHeading>
+          </MeshHeadingList>
+        </MedlineCitation>
+      </PubmedArticle>
+    </PubmedArticleSet>
+    """
+
+    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(
+            "<PubmedArticleSet><PubmedArticle/></PubmedArticleSet>", "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(
+            "<PublicationTypeList></PublicationTypeList>", "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("<MeshHeadingList></MeshHeadingList>", "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)."""