diff --git a/datasets/amfv_datasets/scraping/LICENSE_NOTES.md b/datasets/amfv_datasets/scraping/LICENSE_NOTES.md new file mode 100644 index 0000000..a98b822 --- /dev/null +++ b/datasets/amfv_datasets/scraping/LICENSE_NOTES.md @@ -0,0 +1,30 @@ +# Source licensing notes for scraped corpora + +## WHO (World Health Organization) + +Publications on [who.int](https://www.who.int/publications) published since +November 2016 are licensed under **Creative Commons Attribution-NonCommercial- +ShareAlike 3.0 IGO** (CC BY-NC-SA 3.0 IGO). + +- **Non-commercial use and adaptation** are permitted. +- **Attribution** to WHO is required. +- **Share-alike**: derivatives must use the same or a similar licence. + +Pre-2017 publications were not reissued under this licence. Use +`metadata.publication_date` to filter when building corpora. + +### Suggested attribution + +> © World Health Organization {year}. *{publication title}*. +> Licensed under CC BY-NC-SA 3.0 IGO. +> https://creativecommons.org/licenses/by-nc-sa/3.0/igo/ + +Each scraped document also records `metadata.license` and +`metadata.attribution`. + +### Milestone 1 content scope + +The WHO scraper collects the HTML **Overview** section from each publication +landing page (`metadata.content_scope = "overview"`). Full guideline text is +typically available only as a linked PDF (`metadata.download_url`). PDF +extraction is intentionally deferred. diff --git a/datasets/amfv_datasets/scraping/__init__.py b/datasets/amfv_datasets/scraping/__init__.py index cd7cdb0..110f609 100644 --- a/datasets/amfv_datasets/scraping/__init__.py +++ b/datasets/amfv_datasets/scraping/__init__.py @@ -27,6 +27,16 @@ scrape_guideline, scrape_nice, ) +from amfv_datasets.scraping.who import ( + WhoFetchError, + WhoListingPage, + WhoPublicationRef, + build_publication_text, + list_publications, + publication_ref_from_url, + scrape_publication, + scrape_who, +) __all__ = [ "GuidanceRef", @@ -39,8 +49,12 @@ "ScrapedDocument", "ScraperSource", "USER_AGENT", + "WhoFetchError", + "WhoListingPage", + "WhoPublicationRef", "absolute_unique_urls", "build_guideline_text", + "build_publication_text", "clean_text", "document_title", "default_client", @@ -48,7 +62,11 @@ "guidance_ref_from_url", "html_to_markdown", "list_published_guidance", + "list_publications", + "publication_ref_from_url", "scrape_guideline", "scrape_listing_documents", "scrape_nice", + "scrape_publication", + "scrape_who", ] diff --git a/datasets/amfv_datasets/scraping/cli.py b/datasets/amfv_datasets/scraping/cli.py index e8aece3..59e69bd 100644 --- a/datasets/amfv_datasets/scraping/cli.py +++ b/datasets/amfv_datasets/scraping/cli.py @@ -27,6 +27,7 @@ from amfv_datasets.scraping.base import ScrapedDocument, ScrapeRun from amfv_datasets.scraping.html import LinkMode from amfv_datasets.scraping.nice import scrape_nice +from amfv_datasets.scraping.who import scrape_who class ScraperSource(StrEnum): @@ -34,6 +35,7 @@ class ScraperSource(StrEnum): ALL = "all" NICE = "nice" + WHO = "who" class OutputFormat(StrEnum): @@ -72,6 +74,8 @@ def scrape_documents( match selected_source: case ScraperSource.NICE: return scrape_nice(documents=documents, link_mode=link_mode, url=url) + case ScraperSource.WHO: + return scrape_who(documents=documents, link_mode=link_mode, url=url) case ScraperSource.ALL: raise AssertionError("expanded source cannot be all") raise AssertionError(f"unsupported source: {source}") @@ -125,7 +129,7 @@ def write_markdown_files(documents: Iterable[ScrapedDocument], output_path: Path def _expand_source(source: ScraperSource) -> tuple[ScraperSource, ...]: if source is ScraperSource.ALL: - return (ScraperSource.NICE,) + return (ScraperSource.NICE, ScraperSource.WHO) return (source,) diff --git a/datasets/amfv_datasets/scraping/who.py b/datasets/amfv_datasets/scraping/who.py new file mode 100644 index 0000000..a241251 --- /dev/null +++ b/datasets/amfv_datasets/scraping/who.py @@ -0,0 +1,379 @@ +"""Scrape WHO guideline publications into normalized markdown documents. + +WHO (World Health Organization) publishes guidelines primarily as PDFs hosted on +iris.who.int. The publication landing pages expose a short HTML Overview section +plus bibliographic metadata. For Milestone 1 we scrape that Overview text rather +than PDFs, because PDF extraction loses reading order and interleaves +headers/footers. Full-text PDF extraction is intentionally deferred. + +Discovery uses WHO's Sitefinity OData publications hub API, filtered to the +Guidelines publishing office. Extraction fetches each publication HTML page and +converts the Overview block to markdown. + +Attribution: +The publishing-office filter UUID and PDF-first approach in Meditron's WHO +scraper (epfLLM/meditron, gap-replay/guidelines/scrapers/scrapers.py) informed +discovery scope. Source license: Apache License 2.0. +""" + +from __future__ import annotations + +import json +import logging +import re +from collections.abc import Iterable +from dataclasses import dataclass +from typing import Any +from urllib.parse import urlparse + +import httpx +from lxml import html as lxml_html + +from amfv_datasets.scraping.base import ( + ScrapedDocument, + ScrapeError, + ScrapeRun, + default_client, + scrape_listing_documents, +) +from amfv_datasets.scraping.html import LinkMode, clean_text, document_title, html_to_markdown + +BASE_URL = "https://www.who.int" +GUIDELINES_LISTING_URL = f"{BASE_URL}/publications/who-guidelines" +GUIDELINES_PUBLISHING_OFFICE = "c09761c0-ab8e-4cfa-9744-99509c4d306b" +SF_SITE = "15210d59-ad60-47ff-a542-7ed76645f0c7" +PUBLICATIONS_API_PATH = "/api/hubs/publications" +LISTING_PAGE_SIZE = 25 +DOCUMENT_DELAY_SECONDS = 5.0 +WHO_LICENSE = "CC BY-NC-SA 3.0 IGO" +WHO_ATTRIBUTION = "© World Health Organization. Licensed under CC BY-NC-SA 3.0 IGO." +CONTENT_SCOPE = "overview" + +logger = logging.getLogger(__name__) + +_PUBLICATION_PATH_RE = re.compile( + r"^/publications/i/item/(?P[^/?#]+)(?:/|$)", + re.IGNORECASE, +) +_ISBN_RE = re.compile(r"ISBN:\s*([\d\-]+)", re.IGNORECASE) + +class WhoFetchError(ScrapeError): + """Raised when a WHO publication cannot be fetched or parsed.""" + + +@dataclass(frozen=True) +class WhoPublicationRef: + """A WHO guideline publication reference from the listing.""" + + publication_id: str + title: str + page_url: str + publication_date: str | None = None + tag: str | None = None + download_url: str | None = None + + +@dataclass(frozen=True) +class WhoListingPage: + """A page of WHO guideline publication references.""" + + refs: list[WhoPublicationRef] + total: int | None + + +def publication_ref_from_url(url: str) -> WhoPublicationRef: + """Parse a WHO publication URL into a canonical publication reference. + + Args: + url: WHO publication URL to parse. + """ + parsed = urlparse(url.strip()) + if parsed.scheme not in {"http", "https"} or parsed.netloc.lower() not in { + "www.who.int", + "who.int", + }: + raise WhoFetchError(f"Enter a WHO publication URL from who.int; got {url!r}") + + match = _PUBLICATION_PATH_RE.match(parsed.path.rstrip("/") + "/") + if not match: + raise WhoFetchError( + f"Enter a URL like https://www.who.int/publications/i/item/9789240121805; got {url!r}" + ) + + publication_id = match.group("publication_id") + return WhoPublicationRef( + publication_id=publication_id, + title=publication_id, + page_url=_page_url(publication_id=publication_id), + ) + + +def _page_url(*, publication_id: str) -> str: + return f"{BASE_URL}/publications/i/item/{publication_id}" + + +def _publications_api_params(*, page: int) -> dict[str, str]: + skip = (page - 1) * LISTING_PAGE_SIZE + return { + "sf_site": SF_SITE, + "sf_provider": "OpenAccessProvider", + "sf_culture": "en", + "$orderby": "PublicationDateAndTime desc", + "$select": "Title,ItemDefaultUrl,FormatedDate,Tag,DownloadUrl", + "$filter": f"publishingoffices/any(s:s eq {GUIDELINES_PUBLISHING_OFFICE})", + "$top": str(LISTING_PAGE_SIZE), + "$skip": str(skip), + "$count": "true", + } + + +def _publication_id_from_item_url(item_default_url: str) -> str: + return item_default_url.strip("/").split("/")[-1] + + +def _parse_api_listing(payload: dict[str, Any]) -> WhoListingPage: + """Parse a WHO publications OData response into listing refs.""" + try: + items = payload["value"] + except KeyError as exc: + raise WhoFetchError("Could not parse WHO publications API JSON") from exc + + refs: list[WhoPublicationRef] = [] + for item in items: + item_default_url = (item.get("ItemDefaultUrl") or "").strip() + title = (item.get("Title") or "").strip() + if not item_default_url or not title: + continue + publication_id = _publication_id_from_item_url(item_default_url) + refs.append( + WhoPublicationRef( + publication_id=publication_id, + title=title, + page_url=_page_url(publication_id=publication_id), + publication_date=(item.get("FormatedDate") or None), + tag=(item.get("Tag") or None), + download_url=(item.get("DownloadUrl") or None), + ) + ) + + total = payload.get("@odata.count") + return WhoListingPage(refs=refs, total=total if isinstance(total, int) else None) + + +def _parse_html_listing(html_text: str) -> WhoListingPage: + """Parse server-rendered WHO guideline cards from the listing page.""" + doc = lxml_html.fromstring(html_text) + refs: list[WhoPublicationRef] = [] + seen: set[str] = set() + for link in doc.xpath("//a[contains(@href,'/publications/i/item/')]"): + href = link.get("href", "").split("?")[0] + match = _PUBLICATION_PATH_RE.match(href.rstrip("/") + "/") + if not match or href in seen: + continue + seen.add(href) + publication_id = match.group("publication_id") + title = clean_text(link.xpath("string(.)")) or publication_id + refs.append( + WhoPublicationRef( + publication_id=publication_id, + title=title, + page_url=_page_url(publication_id=publication_id), + ) + ) + return WhoListingPage(refs=refs, total=None) + + +def list_publications(client: httpx.Client, page: int = 1) -> WhoListingPage: + """Return one page of WHO guideline publication refs. + + Args: + client: HTTP client used to fetch the publications API. + page: Listing page number (default: 1). + """ + response = client.get(f"{BASE_URL}{PUBLICATIONS_API_PATH}", params=_publications_api_params(page=page)) + if response.status_code >= 400: + logger.warning("WHO publications API returned %s; falling back to HTML listing", response.status_code) + listing = client.get(GUIDELINES_LISTING_URL) + listing.raise_for_status() + return _parse_html_listing(listing.text) + + try: + payload = response.json() + except json.JSONDecodeError as exc: + raise WhoFetchError("Could not decode WHO publications API JSON") from exc + return _parse_api_listing(payload) + + +def _overview_html(section: lxml_html.HtmlElement) -> str: + """Return HTML for the Overview block within a publication section.""" + overview_headings = section.xpath(".//h3[normalize-space()='Overview']") + if not overview_headings: + return lxml_html.tostring(section, encoding="unicode") + + heading = overview_headings[0] + parts = [lxml_html.tostring(heading, encoding="unicode")] + sibling = heading.getnext() + while sibling is not None and sibling.tag.lower() not in {"h2", "h3"}: + parts.append(lxml_html.tostring(sibling, encoding="unicode")) + sibling = sibling.getnext() + return "".join(parts) + + +def _page_isbn(html_text: str) -> str | None: + doc = lxml_html.fromstring(html_text) + for node in doc.xpath("//*[contains(normalize-space(.), 'ISBN')]"): + match = _ISBN_RE.search(node.text_content()) + if match: + return match.group(1) + return None + + +def _page_publication_date(doc: lxml_html.HtmlElement) -> str | None: + values = doc.xpath("//*[contains(@class,'dynamic-content__date')]/text()") + return clean_text(values[0]) if values else None + + +def _page_tag(doc: lxml_html.HtmlElement) -> str | None: + values = doc.xpath("//*[contains(@class,'dynamic-content__tag')]/text()") + text = clean_text(" ".join(values)) + return text.lstrip("| ").strip() if text else None + + +def _page_download_url(doc: lxml_html.HtmlElement) -> str | None: + for href in doc.xpath("//a[contains(@href,'iris.who.int')]/@href"): + if "bitstreams" in href: + return href.split("?")[0] + return None + + +def build_publication_text( + client: httpx.Client, + ref: WhoPublicationRef, + *, + link_mode: LinkMode = LinkMode.KEEP, +) -> tuple[str, int, str, dict[str, Any]]: + """Scrape a publication Overview into markdown text and bibliographic metadata. + + Args: + client: HTTP client used to fetch the publication page. + ref: WHO publication reference to scrape. + link_mode: Whether links are kept as markdown links or stripped to their + visible text (default: LinkMode.KEEP). + """ + response = client.get(ref.page_url) + response.raise_for_status() + html_text = response.text + doc = lxml_html.fromstring(html_text) + + sections = doc.xpath("//section[contains(@class,'dynamic-content__section')]") + if not sections: + raise WhoFetchError(f"No publication content section for '{ref.publication_id}'") + + title = ( + ref.title + if ref.title != ref.publication_id + else document_title(html_text, fallback=ref.publication_id) + ) + overview_html = _overview_html(sections[0]) + markdown = html_to_markdown(overview_html, link_mode=link_mode, base_url=BASE_URL).strip() + if not markdown: + raise WhoFetchError(f"No readable Overview content for '{ref.publication_id}'") + + extra_metadata: dict[str, Any] = { + "publication_date": ref.publication_date or _page_publication_date(doc), + "tag": ref.tag or _page_tag(doc), + "isbn": _page_isbn(html_text), + "download_url": ref.download_url or _page_download_url(doc), + "content_scope": CONTENT_SCOPE, + "license": WHO_LICENSE, + "attribution": WHO_ATTRIBUTION, + "listing_category": "who-guidelines", + } + return markdown, 1, title, extra_metadata + + +def scrape_publication( + client: httpx.Client, + ref: WhoPublicationRef, + *, + link_mode: LinkMode = LinkMode.KEEP, +) -> ScrapedDocument: + """Scrape a WHO publication into a normalized document. + + Args: + client: HTTP client used to fetch the publication page. + ref: WHO publication reference to scrape. + link_mode: Whether links are kept as markdown links or stripped to their + visible text (default: LinkMode.KEEP). + """ + content, section_count, title, metadata = build_publication_text(client, ref, link_mode=link_mode) + return ScrapedDocument( + source="who", + external_id=f"who-{ref.publication_id}", + title=title, + url=ref.page_url, + content=content, + section_count=section_count, + metadata={ + "publication_id": ref.publication_id, + **metadata, + }, + ) + + +def scrape_who( + *, + documents: int | None, + link_mode: LinkMode = LinkMode.KEEP, + url: str | None = None, +) -> ScrapeRun: + """Scrape WHO documents from a URL or guideline listing pages. + + Args: + documents: Number of documents to scrape. Ignored when `url` is set. + When unset, WHO listing pages are fetched until a page returns no + items (default: None). + link_mode: Whether links are kept as markdown links or stripped to their + visible text (default: LinkMode.KEEP). + url: WHO publication URL to scrape as a single document (default: None). + """ + if url is not None: + + def scrape_url() -> Iterable[ScrapedDocument]: + with default_client() as client: + yield scrape_publication(client, publication_ref_from_url(url), link_mode=link_mode) + + return ScrapeRun(documents=scrape_url(), total=1) + + with default_client() as client: + first_page = list_publications(client, page=1) + total = first_page.total if documents is None or first_page.total is None else min(documents, first_page.total) + return ScrapeRun( + total=total, + documents=scrape_listing_documents( + documents=documents, + client_factory=default_client, + first_page_items=first_page.refs, + list_page=lambda client, page: list_publications(client, page).refs, + scrape_item=lambda client, ref: scrape_publication(client, ref, link_mode=link_mode), + document_delay_seconds=DOCUMENT_DELAY_SECONDS, + ), + ) + + +__all__ = [ + "BASE_URL", + "CONTENT_SCOPE", + "DOCUMENT_DELAY_SECONDS", + "GUIDELINES_LISTING_URL", + "WHO_ATTRIBUTION", + "WHO_LICENSE", + "WhoFetchError", + "WhoListingPage", + "WhoPublicationRef", + "build_publication_text", + "list_publications", + "publication_ref_from_url", + "scrape_publication", + "scrape_who", +] diff --git a/datasets/test/fixtures/who_listing_api.json b/datasets/test/fixtures/who_listing_api.json new file mode 100644 index 0000000..a4392d3 --- /dev/null +++ b/datasets/test/fixtures/who_listing_api.json @@ -0,0 +1,20 @@ +{ + "@odata.context": "https://www.who.int/api/hubs/$metadata#publications(Title,ItemDefaultUrl,FormatedDate,Tag,DownloadUrl)", + "@odata.count": 356, + "value": [ + { + "ItemDefaultUrl": "/9789240121805", + "Title": "Guidelines for the prevention of bloodstream infections and other infections associated with the use of intravascular catheters: part 2: central venous catheters", + "FormatedDate": "28 May 2026", + "Tag": "Guideline", + "DownloadUrl": "https://iris.who.int/server/api/core/bitstreams/f750f24d-c0c2-425c-85fd-ec310d2ce994/content" + }, + { + "ItemDefaultUrl": "/9789240121744", + "Title": "WHO guideline for screening and treatment of cervical pre-cancer lesions for cervical cancer prevention", + "FormatedDate": "8 May 2026", + "Tag": "Guideline", + "DownloadUrl": "https://iris.who.int/server/api/core/bitstreams/32214b73-0e95-4243-9e83-617948510dcd/content" + } + ] +} diff --git a/datasets/test/fixtures/who_publication_overview.html b/datasets/test/fixtures/who_publication_overview.html new file mode 100644 index 0000000..b6f16f2 --- /dev/null +++ b/datasets/test/fixtures/who_publication_overview.html @@ -0,0 +1,17 @@ + +
+

Guidelines for the prevention of bloodstream infections and other infections associated with the use of intravascular catheters: part 2: central venous catheters

+
+
28 May 2026
+
 | Guideline
+
+
+ Download (1.2 MB) +

Overview

+

These WHO guidelines provide evidence-based recommendations for the prevention of bloodstream infections (BSIs) and other infections associated with the use of central venous catheters (CVCs) across health care settings.

+

WHO Team

+

Infection Prevention and Control (IPC)

+

ISBN: 978-92-4-012180-5

+
+
+ diff --git a/datasets/test/test_scraping_who.py b/datasets/test/test_scraping_who.py new file mode 100644 index 0000000..d3db37b --- /dev/null +++ b/datasets/test/test_scraping_who.py @@ -0,0 +1,155 @@ +"""Tests for WHO scraping helpers.""" + +import json +from pathlib import Path + +import httpx +import pytest + +from amfv_datasets.scraping.html import LinkMode +from amfv_datasets.scraping.who import ( + BASE_URL, + WHO_LICENSE, + WhoFetchError, + WhoListingPage, + WhoPublicationRef, + build_publication_text, + list_publications, + publication_ref_from_url, + scrape_publication, +) + +_FIXTURES = Path(__file__).parent / "fixtures" +_CVC_GUIDELINE_TITLE = ( + "Guidelines for the prevention of bloodstream infections and other infections " + "associated with the use of intravascular catheters: part 2: central venous catheters" +) +_CERVICAL_GUIDELINE_TITLE = ( + "WHO guideline for screening and treatment of cervical pre-cancer lesions " + "for cervical cancer prevention" +) + + +def test_list_publications_parses_api_listing() -> None: + """WHO listing payloads are parsed from the publications OData API.""" + payload = json.loads((_FIXTURES / "who_listing_api.json").read_text(encoding="utf-8")) + + def handler(request: httpx.Request) -> httpx.Response: + assert request.url.path == "/api/hubs/publications" + assert request.url.params["$filter"] == "publishingoffices/any(s:s eq c09761c0-ab8e-4cfa-9744-99509c4d306b)" + assert request.url.params["$skip"] == "0" + return httpx.Response(200, json=payload) + + client = httpx.Client(transport=httpx.MockTransport(handler), base_url=BASE_URL) + + listing_page = list_publications(client) + + assert listing_page == WhoListingPage( + total=356, + refs=[ + WhoPublicationRef( + publication_id="9789240121805", + title=_CVC_GUIDELINE_TITLE, + page_url="https://www.who.int/publications/i/item/9789240121805", + publication_date="28 May 2026", + tag="Guideline", + download_url="https://iris.who.int/server/api/core/bitstreams/f750f24d-c0c2-425c-85fd-ec310d2ce994/content", + ), + WhoPublicationRef( + publication_id="9789240121744", + title=_CERVICAL_GUIDELINE_TITLE, + page_url="https://www.who.int/publications/i/item/9789240121744", + publication_date="8 May 2026", + tag="Guideline", + download_url="https://iris.who.int/server/api/core/bitstreams/32214b73-0e95-4243-9e83-617948510dcd/content", + ), + ], + ) + + +def test_publication_ref_from_url_normalizes_publication_url() -> None: + """WHO publication URLs are normalized to canonical refs.""" + assert publication_ref_from_url("https://www.who.int/publications/i/item/9789240121805") == WhoPublicationRef( + publication_id="9789240121805", + title="9789240121805", + page_url="https://www.who.int/publications/i/item/9789240121805", + ) + + +def test_build_publication_text_scrapes_overview() -> None: + """Publication pages are converted into Overview markdown.""" + html = (_FIXTURES / "who_publication_overview.html").read_text(encoding="utf-8") + page_url = "https://www.who.int/publications/i/item/9789240121805" + + def handler(request: httpx.Request) -> httpx.Response: + assert str(request.url) == page_url + return httpx.Response(200, text=html) + + client = httpx.Client(transport=httpx.MockTransport(handler)) + ref = WhoPublicationRef( + publication_id="9789240121805", + title=_CVC_GUIDELINE_TITLE, + page_url=page_url, + publication_date="28 May 2026", + tag="Guideline", + download_url="https://iris.who.int/server/api/core/bitstreams/f750f24d-c0c2-425c-85fd-ec310d2ce994/content", + ) + + content, section_count, title, metadata = build_publication_text(client, ref) + + assert title.startswith("Guidelines for the prevention of bloodstream infections") + assert section_count == 1 + assert "### Overview" in content + assert "central venous catheters (CVCs)" in content + assert "WHO Team" not in content + assert metadata["publication_date"] == "28 May 2026" + assert metadata["tag"] == "Guideline" + assert metadata["isbn"] == "978-92-4-012180-5" + assert metadata["content_scope"] == "overview" + assert metadata["license"] == WHO_LICENSE + assert metadata["listing_category"] == "who-guidelines" + + +def test_scrape_publication_builds_scraped_document() -> None: + """A WHO publication ref is normalized into a ScrapedDocument.""" + html = (_FIXTURES / "who_publication_overview.html").read_text(encoding="utf-8") + page_url = "https://www.who.int/publications/i/item/9789240121805" + + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(200, text=html) + + client = httpx.Client(transport=httpx.MockTransport(handler)) + ref = WhoPublicationRef( + publication_id="9789240121805", + title=_CVC_GUIDELINE_TITLE, + page_url=page_url, + ) + + document = scrape_publication(client, ref, link_mode=LinkMode.STRIP) + + assert document.source == "who" + assert document.external_id == "who-9789240121805" + assert document.url == page_url + assert document.section_count == 1 + assert document.content.strip() + assert document.metadata["publication_id"] == "9789240121805" + assert document.metadata["content_scope"] == "overview" + + +def test_build_publication_text_raises_when_overview_missing() -> None: + """Missing publication markup raises WhoFetchError.""" + html = "

No publication section here.

" + page_url = "https://www.who.int/publications/i/item/9789240121805" + + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(200, text=html) + + client = httpx.Client(transport=httpx.MockTransport(handler)) + ref = WhoPublicationRef( + publication_id="9789240121805", + title="Example guideline", + page_url=page_url, + ) + + with pytest.raises(WhoFetchError, match="No publication content section"): + build_publication_text(client, ref)