pyeuropepmc.processing.extensionsfrom pyeuropepmc.processing.extensions import *
Pydantic helpers are available only when pydantic is installed:
try:
from pyeuropepmc.processing.extensions import (
PydanticModelGenerator,
dataclass_to_pydantic,
)
except ImportError:
# Pydantic not installed
pass
ContentBlockTypeclass ContentBlockType(str, Enum):
PARAGRAPH = "paragraph"
LIST = "list"
FORMULA = "formula"
FIGURE_REF = "figure_ref"
TABLE_REF = "table_ref"
CODE = "code"
BOXED_TEXT = "boxed_text"
HEADING = "heading"
FIGURE = "figure"
TABLE = "table"
MATHML = "mathml"
PEER_REVIEW = "peer_review"
UNKNOWN_BLOCK = "unknown_block"
ContentBlock@dataclass
class ContentBlock:
type: ContentBlockType
text: str = ""
items: list[str] = field(default_factory=list)
list_type: str = ""
label: str = ""
target_id: str = ""
language: str = ""
tex: str = ""
mathml: str = ""
caption: str = ""
uri: str = ""
jats_tag: str = ""
metadata: dict[str, Any] = field(default_factory=dict)
Factory Methods:
ContentBlock.paragraph(text: str) -> ContentBlockContentBlock.heading(text: str) -> ContentBlockContentBlock.list_block(items: list[str], list_type: str = "unordered") -> ContentBlockContentBlock.formula(tex: str, label: str = "") -> ContentBlockContentBlock.figure_ref(target_id: str, label: str = "") -> ContentBlockContentBlock.table_ref(target_id: str, label: str = "") -> ContentBlockContentBlock.code(text: str, language: str = "") -> ContentBlockContentBlock.boxed_text(text: str) -> ContentBlockContentBlock.figure(label: str, caption: str, uri: str = "", target_id: str = "") -> ContentBlockContentBlock.table_block(label: str, caption: str, text: str = "") -> ContentBlockContentBlock.unknown_block(jats_tag: str, text: str = "") -> ContentBlockClass Variables:
LIST_TYPES: ClassVar[set[ContentBlockType]] — Block types with list-like structureRICH_TYPES: ClassVar[set[ContentBlockType]] — Block types with rich structured dataMethods:
to_dict() -> dict[str, Any] — Serialize to dict, omitting empty fieldsStructuredSection@dataclass
class StructuredSection:
title: str
content: list[ContentBlock] = field(default_factory=list)
section_type: str = "body"
Methods:
to_dict() -> dict[str, Any] — Serialize to dictContentBlockExtractorclass ContentBlockExtractor(BaseParser):
def __init__(
self,
root: ET.Element | None = None,
config: ElementPatterns | None = None,
)
def extract_sections() -> list[StructuredSection]
LXMLParserclass LXMLParser:
def __init__(self, **kwargs)
def parse(xml_content: str) -> ET.Element
@staticmethod
def enable_for(parser: FullTextXMLParser) -> None
is_lxml_availabledef is_lxml_available() -> bool
PeerReviewTypeclass PeerReviewType(str, Enum):
DECISION_LETTER = "decision-letter"
REFEREE_REPORT = "referee-report"
EDITOR_REPORT = "editor-report"
REVIEWER_REPORT = "reviewer-report"
REBUTTAL = "rebuttal"
AUTHOR_RESPONSE = "author-response"
APPROVAL = "approval"
OTHER = "other"
PeerReviewMaterial@dataclass
class PeerReviewMaterial:
review_type: PeerReviewType
content: str
date: str = ""
author: str = ""
title: str = ""
PeerReviewSet@dataclass
class PeerReviewSet:
revision_round: int
reviews: list[PeerReviewMaterial] = field(default_factory=list)
article_type: str = ""
PeerReviewExtractorclass PeerReviewExtractor(BaseParser):
def __init__(
self,
root: ET.Element | None = None,
config: ElementPatterns | None = None,
)
def extract_all() -> list[PeerReviewSet]
def extract_by_type(review_type: PeerReviewType) -> list[PeerReviewMaterial]
MathMLConverterclass MathMLConverter:
def __init__(self)
def convert(mathml_str: str) -> str
def convert_element(element: ET.Element) -> str
# Namespace for MathML elements
namespaces: dict[str, str]
# Mapping of 80+ named entities
ENTITY_MAP: ClassVar[dict[str, str]]
ValidationFinding@dataclass
class ValidationFinding:
severity: str # "error", "warning", "info"
category: str # e.g., "AUTHORS", "FUNDING"
message: str
element: ET.Element | None = None
ValidationReport@dataclass
class ValidationReport:
compliance_score: float # 0.0 to 1.0
findings: list[ValidationFinding] = field(default_factory=list)
total_checks: int = 0
passed_checks: int = 0
JATS4RValidatorclass JATS4RValidator(BaseParser):
def __init__(
self,
root: ET.Element | None = None,
config: ElementPatterns | None = None,
)
def validate() -> ValidationReport
Validation Categories (accessible via class attributes):
CHECK_AUTHORS, CHECK_AFFILIATIONS, CHECK_ABSTRACTSCHECK_FUNDING, CHECK_CITATIONS, CHECK_DATA_AVAILABILITYProcessingResult@dataclass
class ProcessingResult:
index: int
success: bool
metadata: dict | None = None
error: str | None = None
duration: float = 0.0
BatchResult@dataclass
class BatchResult:
results: list[ProcessingResult]
errors: list[str]
total_time: float
total_count: int
BatchProcessorclass BatchProcessor:
def __init__(
self,
rate_per_second: float = 5.0,
max_workers: int = 4,
on_progress: Callable[[int, int], None] | None = None,
on_error: Callable[[str], None] | None = None,
)
def process(xml_strings: list[str]) -> BatchResult
def process_files(file_paths: list[str]) -> BatchResult
def process_directory(
directory: str, pattern: str = "*.xml"
) -> BatchResult
AssetTypeclass AssetType(str, Enum):
FIGURE = "figure"
GRAPHIC = "graphic"
SUPPLEMENTARY = "supplementary"
MEDIA = "media"
TABLE = "table"
AssetRef@dataclass
class AssetRef:
type: AssetType
uri: str
label: str = ""
caption: str = ""
target_id: str = ""
AssetFetchPolicy@dataclass
class AssetFetchPolicy:
output_dir: str = "assets"
overwrite: bool = False
timeout: int = 30
ImageFetcherclass ImageFetcher(BaseParser):
def __init__(
self,
root: ET.Element | None = None,
config: ElementPatterns | None = None,
)
def extract_assets() -> list[AssetRef]
def download_assets(
assets: list[AssetRef],
policy: AssetFetchPolicy | None = None,
) -> list[Path]
ResolvedReference@dataclass
class ResolvedReference:
label: str
doi: str = ""
pmid: str = ""
pmcid: str = ""
title: str = ""
authors: str = ""
source: str = ""
year: str = ""
volume: str = ""
pages: str = ""
cited_by_count: int = 0
resolved: bool = False
ReferenceResolverclass ReferenceResolver:
def __init__(
self,
cache_size: int = 1000,
rate_per_second: float = 5.0,
)
def resolve_references(parser: FullTextXMLParser) -> list[ResolvedReference]
def resolve_by_doi(doi: str) -> ResolvedReference | None
def resolve_by_pmid(pmid: str) -> ResolvedReference | None
def resolve_by_title(title: str) -> ResolvedReference | None
def clear_cache()
dataclass_to_pydanticdef dataclass_to_pydantic(
dataclass: type,
*,
field_mapping: dict[str, str] | None = None,
field_overrides: dict[str, Any] | None = None,
) -> type
PydanticModelGeneratorclass PydanticModelGenerator:
def __init__(self)
def generate_model(name: str, sample_data: dict) -> type
def add_validator(name: str, validator_func: Callable)
def add_field_description(field: str, description: str)
parse_xml_filedef parse_xml_file(path: str | Path) -> FullTextXMLParser
parse_xml_directorydef parse_xml_directory(
path: str | Path,
pattern: str = "*.xml",
) -> list[FullTextXMLParser]
extract_article_id_from_xmldef extract_article_id_from_xml(xml_content: str) -> dict[str, str]
# Returns: {"pmcid": "...", "pmid": "...", "doi": "..."}
LocalXMLProcessorclass LocalXMLProcessor:
@staticmethod
def parse_file(path: str | Path) -> FullTextXMLParser
@staticmethod
def parse_string(xml_content: str) -> FullTextXMLParser
@staticmethod
def write_markdown(xml_content: str, output_path: str | Path)
@staticmethod
def write_plaintext(xml_content: str, output_path: str | Path)
@staticmethod
def extract_id(xml_content: str) -> dict[str, str]
@staticmethod
def batch_process_files(
file_paths: list[str | Path],
rate_per_second: float = 5.0,
) -> BatchResult
New methods on FullTextXMLParser:
get_full_text_sections_structureddef get_full_text_sections_structured() -> list[StructuredSection]
Returns body sections as typed content blocks. Uses ContentBlockExtractor internally. Raises ParsingError if no content has been parsed.