diff --git a/scripts/parsers/pmc.py b/scripts/parsers/pmc.py new file mode 100644 index 0000000..acd0af8 --- /dev/null +++ b/scripts/parsers/pmc.py @@ -0,0 +1,215 @@ +"""Парсер PubMed Central (PMC) — NCBI E-utilities. + +PMC Open Access Subset — крупнейший биомедицинский открытый архив с реальным +полным текстом статей (не только аннотацией). API бесплатный, ключ не нужен +(рекомендуется для повышения лимита с 3 до 10 запросов/сек — NCBI_API_KEY). +Документация: https://www.ncbi.nlm.nih.gov/books/NBK25501/ + +Два запроса на пачку: esearch (ID) → efetch (полные JATS XML статьи, откуда +разом достаём метаданные + abstract + body — тело статьи, реальный полный +текст, а не аннотация или OCR-фрагмент). +""" + +import logging +import os +import time +import xml.etree.ElementTree as ET +from typing import Any + +import httpx +from base import BaseParser + +logger = logging.getLogger(__name__) + +EUTILS = "https://eutils.ncbi.nlm.nih.gov/entrez/eutils" +BATCH_SIZE = 20 +# 3 запроса/сек без ключа, 10/сек с ключом (NCBI_API_KEY в окружении) +RATE_LIMIT_DELAY = 0.12 if os.environ.get("NCBI_API_KEY") else 0.35 + + +class PMCParser(BaseParser): + """Парсер PubMed Central через NCBI E-utilities.""" + + source_name = "pmc" + + def __init__(self) -> None: + super().__init__() + self.api_key = os.environ.get("NCBI_API_KEY") + self.client = httpx.Client( + headers={"User-Agent": "AcademicHelper/1.0 (noreply@jze9.ru)"}, + timeout=30.0, + ) + + def _params(self, **extra: Any) -> dict[str, Any]: + p = dict(extra) + if self.api_key: + p["api_key"] = self.api_key + return p + + def fetch( + self, + query: str = "", + limit: int = 500, + year_from: int | None = None, + year_to: int | None = None, + ) -> list[dict[str, Any]]: + """ + Получить статьи из PMC Open Access Subset. + + Args: + query: Поисковый запрос + limit: Максимальное количество документов + year_from: Год публикации от + year_to: Год публикации до + + Returns: + Список сырых словарей (уже с извлечёнными метаданными + full_text) + """ + term = f"{query} AND open access[filter]" if query else "open access[filter]" + if year_from or year_to: + lo = year_from or 1900 + hi = year_to or 3000 + term += f' AND ("{lo}"[PDAT] : "{hi}"[PDAT])' + + ids = self._search_ids(term, limit) + if not ids: + return [] + + results: list[dict[str, Any]] = [] + for i in range(0, len(ids), BATCH_SIZE): + batch = ids[i : i + BATCH_SIZE] + try: + response = self.client.get( + f"{EUTILS}/efetch.fcgi", + params=self._params(db="pmc", id=",".join(batch), rettype="full", retmode="xml"), + ) + response.raise_for_status() + results.extend(self._parse_articles(response.text)) + time.sleep(RATE_LIMIT_DELAY) + except httpx.HTTPStatusError as e: + logger.error(f"PMC efetch HTTP ошибка: {e.response.status_code}") + if e.response.status_code == 429: + time.sleep(5) + continue + except Exception as e: + logger.error(f"Ошибка efetch PMC (батч {i}): {e}") + continue + + return results[:limit] + + def _search_ids(self, term: str, limit: int) -> list[str]: + """Собрать PMC ID постранично через esearch.""" + ids: list[str] = [] + retstart = 0 + page = min(200, limit) + + while len(ids) < limit: + try: + response = self.client.get( + f"{EUTILS}/esearch.fcgi", + params=self._params( + db="pmc", term=term, retstart=retstart, + retmax=min(page, limit - len(ids)), retmode="json", + ), + ) + response.raise_for_status() + page_ids = response.json().get("esearchresult", {}).get("idlist", []) + if not page_ids: + break + ids.extend(page_ids) + retstart += len(page_ids) + time.sleep(RATE_LIMIT_DELAY) + except httpx.HTTPStatusError as e: + logger.error(f"PMC esearch HTTP ошибка: {e.response.status_code}") + break + except Exception as e: + logger.error(f"Ошибка esearch PMC: {e}") + break + + return ids[:limit] + + def _parse_articles(self, xml_text: str) -> list[dict[str, Any]]: + """Разобрать JATS XML (пачка статей из efetch) в плоские словари.""" + try: + root = ET.fromstring(xml_text) + except ET.ParseError as e: + logger.error(f"Ошибка парсинга XML PMC: {e}") + return [] + + return [self._parse_article(art) for art in root.findall("article")] + + @staticmethod + def _text(el: ET.Element | None) -> str | None: + """Склеить весь текст элемента, включая вложенные теги форматирования.""" + return "".join(el.itertext()).strip() if el is not None else None + + def _parse_article(self, article: ET.Element) -> dict[str, Any]: + """Извлечь метаданные + abstract + полный текст из одной
.""" + am = article.find(".//article-meta") + if am is None: + return {} + + pmcaid = doi = None + for aid in am.findall("article-id"): + if aid.get("pub-id-type") == "pmcaid": + pmcaid = aid.text + elif aid.get("pub-id-type") == "doi": + doi = aid.text + + authors = [] + for c in am.findall(".//contrib-group/contrib[@contrib-type='author']"): + surname = c.find(".//surname") + given = c.find(".//given-names") + if surname is not None and surname.text: + authors.append({ + "last_name": surname.text.strip(), + "first_name": (given.text or "").strip() if given is not None else "", + }) + + year = None + for y in am.findall(".//pub-date/year"): + if y.text and y.text.isdigit(): + year = int(y.text) + break + + body = article.find("body") + full_text = None + if body is not None: + paragraphs = [self._text(p) for p in body.findall(".//p")] + full_text = " ".join(p for p in paragraphs if p) or None + + return { + "id": pmcaid, + "doi": doi, + "title": self._text(am.find(".//title-group/article-title")), + "journal": self._text(article.find(".//journal-meta/journal-title-group/journal-title")), + "authors": authors, + "year": year, + "abstract": self._text(am.find(".//abstract")), + "full_text": full_text, + } + + def transform(self, raw: dict[str, Any]) -> dict[str, Any]: + """Преобразовать статью PMC в унифицированный формат.""" + ext_id = raw.get("id") + if not ext_id: + return {} + + authors = self.normalize_authors(raw.get("authors", [])) + + return { + "source": self.source_name, + "ext_id": f"pmc:{ext_id}", + "doi": raw.get("doi"), + "title": (raw.get("title") or "").strip() or None, + "authors": authors, + "year": raw.get("year"), + "lang": "en", # PMC — практически полностью англоязычный корпус + "journal": raw.get("journal"), + "volume": None, + "issue": None, + "pages": None, + "abstract": raw.get("abstract"), + "url": f"https://www.ncbi.nlm.nih.gov/pmc/articles/PMC{ext_id}/", + "full_text": raw.get("full_text"), + } diff --git a/scripts/parsers/tests/test_pmc.py b/scripts/parsers/tests/test_pmc.py new file mode 100644 index 0000000..eb09ea1 --- /dev/null +++ b/scripts/parsers/tests/test_pmc.py @@ -0,0 +1,88 @@ +"""Юнит-тесты парсера PubMed Central (PMC) — чистая логика (без сети). + +_parse_article работает с реальными xml.etree.ElementTree узлами (JATS XML из +efetch), поэтому тесты строят минимальные JATS-фрагменты, а не мокают HTTP. +""" + +import xml.etree.ElementTree as ET + +from pmc import PMCParser + +ARTICLE_XML = """ +
+ + + Journal of Testing + + + 1234567 + 10.1000/test.123 + A Study of Testing + + + IvanovIvan + + + NotAnAuthorX + + + 2024 +

This is the abstract text.

 + + + +

First paragraph of the body.

+

Second paragraph with a ref inline.

+ +
+""" + + +def _article() -> ET.Element: + return ET.fromstring(ARTICLE_XML) + + +def test_parse_article_extracts_all_fields(): + raw = PMCParser()._parse_article(_article()) + assert raw["id"] == "1234567" + assert raw["doi"] == "10.1000/test.123" + assert raw["title"] == "A Study of Testing" # вложенный склеен + assert raw["journal"] == "Journal of Testing" + assert raw["year"] == 2024 + assert raw["abstract"] == "This is the abstract text." + assert raw["full_text"] == "First paragraph of the body. Second paragraph with a ref inline." + + +def test_parse_article_only_includes_authors_not_editors(): + raw = PMCParser()._parse_article(_article()) + assert raw["authors"] == [{"last_name": "Ivanov", "first_name": "Ivan"}] + + +def test_parse_article_without_article_meta_is_empty(): + assert PMCParser()._parse_article(ET.fromstring("
")) == {} + + +def test_transform_maps_to_unified_schema(): + raw = PMCParser()._parse_article(_article()) + t = PMCParser().transform(raw) + assert t["source"] == "pmc" + assert t["ext_id"] == "pmc:1234567" + assert t["lang"] == "en" + assert t["url"] == "https://www.ncbi.nlm.nih.gov/pmc/articles/PMC1234567/" + assert t["authors"][0]["last_name"] == "Ivanov" + assert t["full_text"].startswith("First paragraph") + + +def test_transform_empty_without_id(): + assert PMCParser().transform({"title": "x"}) == {} + + +def test_parse_articles_batch(): + xml = f"{ARTICLE_XML}{ARTICLE_XML}" + parsed = PMCParser()._parse_articles(xml) + assert len(parsed) == 2 + assert all(p["id"] == "1234567" for p in parsed) + + +def test_parse_articles_malformed_xml_returns_empty(): + assert PMCParser()._parse_articles(" None: ) +def run_pmc(args: argparse.Namespace, output_dir: Path) -> None: + from pmc import PMCParser + + parser = PMCParser() + parser.run( + output_dir=output_dir, + query=args.query, + limit=args.limit, + year_from=args.year_from, + year_to=args.year_to, + ) + + def run_all(args: argparse.Namespace, output_dir: Path) -> None: """Запустить все парсеры последовательно.""" logger.info("Запуск всех парсеров...") run_openalex(args, output_dir) run_cyberleninka(args, output_dir) run_arxiv(args, output_dir) + run_pmc(args, output_dir) PARSERS = { "openalex": run_openalex, "cyberleninka": run_cyberleninka, "arxiv": run_arxiv, + "pmc": run_pmc, "all": run_all, } diff --git a/services/worker-indexer/app/tasks/index.py b/services/worker-indexer/app/tasks/index.py index 3aa02ec..4518e58 100644 --- a/services/worker-indexer/app/tasks/index.py +++ b/services/worker-indexer/app/tasks/index.py @@ -501,6 +501,14 @@ def run_parser(source_id: int) -> dict[str, Any]: "limit": cfg["limit"], "year_from": cfg.get("year_from"), } + elif stype == "pmc": + from pmc import PMCParser as P + fetch_kwargs = { + "query": cfg.get("query") or "", + "limit": cfg["limit"], + "year_from": cfg.get("year_from"), + "year_to": cfg.get("year_to"), + } else: raise ValueError(f"неизвестный тип источника: {stype}")