crossref

zocrates.discovery.providers.crossref

Crossref DOI metadata validator and resolver.

DEFAULT_POLITE_EMAIL module-attribute

DEFAULT_POLITE_EMAIL = 'zocrates-dev@proton.me'

logger module-attribute

logger = logging.getLogger(__name__)

CrossrefValidator

CrossrefValidator(email=DEFAULT_POLITE_EMAIL, timeout=5.0)

Validates DOI metadata and queries bibliographic records via Crossref REST API.

Initialize the Crossref validator.

Parameters:

Name	Type	Description	Default
email	str	Contact email used for polite-pool API identification.	DEFAULT_POLITE_EMAIL
timeout	float	HTTP request timeout in seconds.	5.0

Source code in zocrates/discovery/providers/crossref.py

def __init__(self, email: str = DEFAULT_POLITE_EMAIL, timeout: float = 5.0) -> None:
    """Initialize the Crossref validator.

    Args:
        email: Contact email used for polite-pool API identification.
        timeout: HTTP request timeout in seconds.
    """
    self.email = email
    self.timeout = timeout

validate_doi async

validate_doi(doi, email=None)

Validate and retrieve Crossref work metadata for a given DOI.

Parameters:

Name	Type	Description	Default
doi	str	Digital Object Identifier.	required
email	str | None	User or developer email for polite pool identification.	None

Returns:

Type	Description
dict[str, Any] | None	Dictionary containing Crossref 'message' payload if valid, or None.

Source code in zocrates/discovery/providers/crossref.py

async def validate_doi(self, doi: str, email: str | None = None) -> dict[str, Any] | None:
    """
    Validate and retrieve Crossref work metadata for a given DOI.

    Args:
        doi: Digital Object Identifier.
        email: User or developer email for polite pool identification.

    Returns:
        Dictionary containing Crossref 'message' payload if valid, or None.
    """
    clean_doi = normalize_doi(doi)
    if not clean_doi:
        return None

    polite_email = email or self.email
    url = f"https://api.crossref.org/works/{clean_doi}"
    headers = {
        "User-Agent": make_user_agent(polite_email),
        "Accept": "application/json",
    }

    try:
        async with httpx.AsyncClient(timeout=self.timeout) as client:
            resp = await client.get(url, headers=headers)
            if resp.status_code != 200:
                logger.debug("Crossref returned status %d for DOI %s", resp.status_code, clean_doi)
                return None
            data: dict[str, Any] = resp.json()
            message = data.get("message")
            return message if isinstance(message, dict) else None
    except Exception as e:
        logger.debug("Crossref validation failed for DOI %s: %s", clean_doi, e)
        return None

make_user_agent

make_user_agent(email=DEFAULT_POLITE_EMAIL)

Build a respectful User-Agent header string adhering to API etiquette.

Source code in zocrates/discovery/providers/base.py

def make_user_agent(email: str = DEFAULT_POLITE_EMAIL) -> str:
    """Build a respectful User-Agent header string adhering to API etiquette."""
    polite_email = email.strip() if email and email.strip() else DEFAULT_POLITE_EMAIL
    return f"Zocrates/0.2.0 (mailto:{polite_email})"

normalize_doi

normalize_doi(doi)

Normalize DOI by stripping standard URI and resolver prefixes and lowercasing.

Source code in zocrates/rag/zotero_db.py

def normalize_doi(doi: str | None) -> str | None:
    """Normalize DOI by stripping standard URI and resolver prefixes and lowercasing."""
    if not doi:
        return None
    d = doi.strip()
    prefixes = (
        "https://doi.org/",
        "http://doi.org/",
        "https://dx.doi.org/",
        "http://dx.doi.org/",
        "doi.org/",
        "doi:",
    )
    for prefix in prefixes:
        if d.lower().startswith(prefix):
            d = d[len(prefix) :].strip()
    d = d.strip().rstrip("/")
    return d.lower() if d else None