Function resolve_doi

quantmind/preprocess/fetch/doi.py:81–126

Look up a DOI on Crossref and return canonical metadata. Args: doi: A DOI string. Accepts canonical form (``10.NNNN/...``), ``doi:`` prefix, or ``https://doi.org/`` URL form. timeout: HTTP timeout in seconds. Returns: ``CrossrefMetadata`` populated from

(
    doi: str,
    *,
    timeout: float = 15.0,
)

async def resolve_doi(
    doi: str,
    *,
    timeout: float = 15.0,
) -> CrossrefMetadata:
    """Look up a DOI on Crossref and return canonical metadata.

    Args:
        doi: A DOI string. Accepts canonical form (``10.NNNN/...``),
            ``doi:`` prefix, or ``https://doi.org/`` URL form.
        timeout: HTTP timeout in seconds.

    Returns:
        ``CrossrefMetadata`` populated from the ``message`` block of
        Crossref's response.

    Raises:
        ValueError: If the DOI is malformed.
        httpx.HTTPStatusError: On 404 (DOI not registered) or 5xx.
    """
    normalized = _normalize_doi(doi)
    if not _DOI_PATTERN.match(normalized):
        raise ValueError(f"malformed DOI: {doi!r}")

    url = f"{CROSSREF_BASE_URL}/{normalized}"
    async with httpx.AsyncClient(
        timeout=timeout, follow_redirects=True
    ) as client:
        response = await client.get(url)
        response.raise_for_status()
        body = response.json()

    msg = body.get("message", {})
    titles = msg.get("title") or []
    container = msg.get("container-title") or []
    issued = msg.get("issued", {}).get("date-parts")

    return CrossrefMetadata(
        doi=normalized,
        title=titles[0] if titles else None,
        authors=_format_authors(msg.get("author")),
        journal=container[0] if container else None,
        publisher=msg.get("publisher"),
        published_at=_parse_crossref_date(issued),
        primary_url=msg.get("URL"),
    )