#!/usr/bin/env python3
"""
Bibliographic metadata extraction cascade
File: klea_utils/biblio/extract.py
Copyright 2026 Ankur Sinha
Author: Ankur Sinha <sanjay DOT ankur AT gmail DOT com>
"""
import logging
import re
from pathlib import Path
from typing import Protocol
from .docling import extract_docling_structured, extract_layout_region
from .doi import BiblioRecord, normalize_doi
from .pdf import extract_pdf_info
from .regex import _scan_dois, extract_regex_metadata
logger = logging.getLogger(__name__)
[docs]
class Resolver(Protocol):
"""Protocol for objects that can resolve a DOI to a record.
:class:`~klea_utils.biblio.doi.DoiResolver` implements this; tests
and other callers may substitute any object with a compatible
``resolve`` method.
"""
[docs]
def resolve(self, doi: str) -> BiblioRecord | None:
"""Resolve *doi* to a record, or ``None`` on failure."""
def _resolve_record(
doi: str | None,
resolver: Resolver | None,
candidates: list[str] | None = None,
) -> BiblioRecord | None:
"""Resolve *doi* via *resolver*, falling back to *candidates*.
The primary *doi* is tried first. If it resolves to a record *with
authors* (a paper record), that is used. Otherwise -- or when the
primary is missing -- the remaining *candidates* are tried in order
until one yields a record with authors. If only author-less records
are found (e.g. a journal-level DOI like ``10.1073/pnas``), the first
resolved record is returned as a last resort, matching the old
behaviour.
Logs an informational message when a discovered DOI is not resolved
because no resolver was provided, and a warning when resolution is
attempted but fails.
:param doi: Primary discovered DOI, or ``None``
:param resolver: Resolver, or ``None``
:param candidates: Additional DOI candidates to try when the primary
does not resolve to a paper record
:returns: Resolved record, or ``None``
"""
if resolver is None:
if doi:
logger.info(
f"DOI {doi} discovered but DOI resolution skipped (no resolver)"
)
return None
ordered = []
if doi:
ordered.append(doi)
for candidate in candidates or []:
if candidate not in ordered:
ordered.append(candidate)
if not ordered:
return None
first_resolved: BiblioRecord | None = None
for candidate in ordered:
logger.debug(f"trying DOI candidate {candidate!r}")
record = resolver.resolve(candidate)
if record is None:
logger.warning(f"Could not resolve DOI {candidate} (see DOI resolver logs)")
continue
if first_resolved is None:
first_resolved = record
logger.info(
f"resolved DOI {candidate} via DOI services\n"
f"{record.title = }\n"
f"{record.authors = }\n"
f"{record.year = }"
)
if record.authors:
logger.info(
f"using DOI {candidate} (record has authors)\n"
f"{record.title = }\n"
f"{record.year = }"
)
return record
logger.warning(
"No DOI candidate resolved to a paper record; "
"falling back to the first resolved record"
)
return first_resolved
def _doi_candidates(full_text: str, docling_info: dict) -> list[str]:
"""Return deduplicated DOI candidates from text and docling URLs.
The primary DOI (from the labeled tiers) may be broken or a
journal-level stub; these additional candidates give the resolver
something to fall back to. They come from two sources:
- every ``10.`` match in the document text (:func:`_scan_dois`), and
- DOIs embedded in the docling ``urls`` hyperlinks.
:param full_text: Joined document text
:param docling_info: Docling tier output (may carry a ``urls`` list)
:returns: Deduplicated list of sanitized DOI strings
"""
candidates: list[str] = []
seen: set[str] = set()
text_candidates = _scan_dois(full_text)
url_candidates = [
candidate
for url in (docling_info.get("urls") or [])
for candidate in _scan_dois(url)
]
for candidate in text_candidates + url_candidates:
normalized = normalize_doi(candidate)
if normalized and normalized not in seen:
seen.add(normalized)
candidates.append(normalized)
return candidates
def _document_text(dl_doc) -> str:
"""Join the document's text items into a single string."""
return "\n".join(item.text for item in dl_doc.texts if item.text.strip())
def _pdf_fields(pdf_path: str | None) -> dict:
"""Extract and normalise the pdf-info tier, or ``{}``."""
if not pdf_path:
return {}
return _normalize_pdf_info(extract_pdf_info(pdf_path))
def _normalize_pdf_info(pdf_info: dict) -> dict:
"""Normalise PDF Info fields to the canonical metadata key set."""
result: dict = {}
if pdf_info.get("title"):
result["title"] = pdf_info["title"]
if pdf_info.get("author"):
result["authors"] = _split_terms(pdf_info["author"])
if pdf_info.get("keywords"):
result["keywords"] = _split_terms(pdf_info["keywords"])
if pdf_info.get("doi"):
result["doi"] = pdf_info["doi"]
if pdf_info.get("url"):
result["url"] = pdf_info["url"]
return result
def _discover_doi(*sources: dict) -> str | None:
"""Return the first DOI found across *sources*."""
for source in sources:
doi = source.get("doi")
if doi:
return doi
return None
def _record_fields(record: BiblioRecord) -> dict:
"""Convert a resolved record to canonical metadata fields.
The abstract is deliberately excluded (it is already part of the
chunked document).
"""
fields = {
"title": record.title,
"authors": record.authors,
"year": record.year,
"journal": record.journal,
"doi": record.doi,
}
return {key: value for key, value in fields.items() if value not in (None, [], "")}
def _merge_tiers(
tiers: list[tuple[str, dict]],
record: BiblioRecord | None,
pdf_fields: dict,
file_path: str,
) -> dict:
"""Merge the tier contributions into the final metadata dict.
*tiers* is an ordered ``(label, fields)`` list in **precedence
order, highest authority first** (``doi-service`` > ``pdf-info`` >
``docling``/``layout-regex`` > ``regex``). Each tier only fills
fields not already set (gap-fill), so the most authoritative tier
that has a value for a field wins. Also applies the filename-stem
title fallback and computes the ``_metadata_complete`` /
``_sources`` internal keys.
:param tiers: Ordered ``(label, fields)`` tier list, precedence order
:param record: Resolved DOI record, or ``None``
:param pdf_fields: Normalised pdf-info fields (used for the
completeness check)
:param file_path: Source file path (for the stem title fallback)
:returns: Flat metadata dict with the internal keys
"""
metadata: dict = {}
sources: list[str] = []
for label, fields in tiers:
_gap_fill(metadata, fields, label, sources)
if "title" not in metadata:
metadata["title"] = Path(file_path).stem
if record is not None:
complete = bool(record.title and record.authors and record.year)
else:
complete = bool(
pdf_fields.get("title")
and pdf_fields.get("authors")
and pdf_fields.get("keywords")
)
metadata["_metadata_complete"] = complete
metadata["_sources"] = sources
if not complete:
logger.warning(
f"metadata extraction incomplete for {file_path}; "
f"review the pre-populated template"
)
logger.info(
f"extracted metadata for {file_path}: "
f"doi={metadata.get('doi')!r} complete={complete} sources={sources}"
)
logger.debug(
f"metadata extraction done for {file_path = }\n"
f"{metadata = }\n"
f"{sources = }\n"
f"{complete = }"
)
return metadata
def _gap_fill(metadata: dict, source: dict, label: str, sources: list[str]) -> None:
"""Fill unset *metadata* fields from *source*, recording *label*.
Fields already present in *metadata* are never overwritten, which is
what gives the cascade its most-authoritative-first precedence.
"""
added = False
for key, value in source.items():
if key in metadata or not value:
continue
metadata[key] = value
added = True
if added and label not in sources:
sources.append(label)
def _split_terms(value: str) -> list[str]:
"""Split a comma/semicolon-separated term list."""
return [term.strip() for term in re.split(r"[,;]", value) if term.strip()]