anti-plagiarism/scripts/parsers/pmc.py

"""Парсер 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, ProgressCallback

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,
        progress_cb: ProgressCallback | None = None,
    ) -> list[dict[str, Any]]:
        """
        Получить статьи из PMC Open Access Subset.

        Args:
            query: Поисковый запрос
            limit: Максимальное количество документов
            year_from: Год публикации от
            year_to: Год публикации до
            progress_cb: см. base.ProgressCallback

        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))
                if progress_cb and not progress_cb(min(len(results), limit)):
                    logger.info(f"PMC: выборка остановлена по запросу (получено {len(results)})")
                    break
                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 + полный текст из одной <article>."""
        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"),
        }