Index

zocrates.discovery.providers

Scholarly literature discovery providers and resolvers.

DEFAULT_POLITE_EMAIL module-attribute

DEFAULT_POLITE_EMAIL = 'zocrates-dev@proton.me'

DEFAULT_USER_AGENT module-attribute

DEFAULT_USER_AGENT = f'Zocrates/0.2.0 (mailto:{DEFAULT_POLITE_EMAIL})'

__all__ module-attribute

__all__ = ['DEFAULT_POLITE_EMAIL', 'DEFAULT_USER_AGENT', 'SearchProvider', 'TokenBucketRateLimiter', 'make_user_agent', 'OpenAlexProvider', 'UnpaywallResolver', 'CrossrefValidator']

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

OpenAlexProvider

OpenAlexProvider(email=DEFAULT_POLITE_EMAIL, rate_limiter=None, timeout=15.0, api_key=None)

Bases: SearchProvider

Primary search provider backed by the OpenAlex REST API. Indexes 250M+ scholarly works, arXiv preprints, and open-access metadata.

Initialize the OpenAlex search provider.

Parameters:

Name	Type	Description	Default
email	str	Contact email for polite-pool API requests.	DEFAULT_POLITE_EMAIL
rate_limiter	TokenBucketRateLimiter | None	Token bucket rate limiter instance.	None
timeout	float	HTTP request timeout in seconds.	15.0
api_key	str | None	Optional OpenAlex premium API key.	None

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

def __init__(
    self,
    email: str = DEFAULT_POLITE_EMAIL,
    rate_limiter: TokenBucketRateLimiter | None = None,
    timeout: float = 15.0,
    api_key: str | None = None,
) -> None:
    """Initialize the OpenAlex search provider.

    Args:
        email: Contact email for polite-pool API requests.
        rate_limiter: Token bucket rate limiter instance.
        timeout: HTTP request timeout in seconds.
        api_key: Optional OpenAlex premium API key.
    """
    self.email = email
    self.rate_limiter = rate_limiter or TokenBucketRateLimiter(rate=10.0, capacity=10.0)
    self.timeout = timeout
    self.api_key = api_key.strip() if api_key and isinstance(api_key, str) else None

extract_arxiv_id_from_work staticmethod

extract_arxiv_id_from_work(work)

Extract canonical arXiv ID without version suffix from an OpenAlex work record.

Parameters:

Name	Type	Description	Default
work	dict[str, Any]	OpenAlex work dictionary.	required

Returns:

Type	Description
str | None	Canonical lowercase arXiv ID without 'v' suffix (e.g. '2305.12345'), or None.

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

@staticmethod
def extract_arxiv_id_from_work(work: dict[str, Any]) -> str | None:
    """
    Extract canonical arXiv ID without version suffix from an OpenAlex work record.

    Args:
        work: OpenAlex work dictionary.

    Returns:
        Canonical lowercase arXiv ID without 'v' suffix (e.g. '2305.12345'), or None.
    """
    # 1. Check locations landing_page_url and pdf_url
    locations = work.get("locations") or []
    primary_loc = work.get("primary_location") or {}
    candidate_locs: list[dict[str, Any]] = [primary_loc]
    if isinstance(locations, list):
        candidate_locs.extend(locations)

    for loc in candidate_locs:
        if not isinstance(loc, dict):
            continue
        for url_key in ("landing_page_url", "pdf_url"):
            val = loc.get(url_key)
            if val and isinstance(val, str):
                cand = normalize_arxiv_id(val)
                if cand:
                    return cand

    # 2. Check DataCite DOIs matching 10.48550/arxiv.(...)
    doi = work.get("doi")
    if doi and isinstance(doi, str):
        cand_doi = normalize_arxiv_id(doi)
        if cand_doi:
            return cand_doi

    # 3. Check ids dictionary if present
    ids = work.get("ids") or {}
    if isinstance(ids, dict):
        arxiv_val = ids.get("arxiv")
        if arxiv_val and isinstance(arxiv_val, str):
            cand_ids = normalize_arxiv_id(arxiv_val)
            if cand_ids:
                return cand_ids

    return None

reconstruct_abstract staticmethod

reconstruct_abstract(inverted_index)

Reconstruct plain-text abstract from an OpenAlex abstract_inverted_index dictionary.

Parameters:

Name	Type	Description	Default
inverted_index	dict[str, list[int]] | None	Mapping of words to their token position indices.	required

Returns:

Type	Description
str | None	Coherent, human-readable abstract string, or None if empty/invalid.

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

@staticmethod
def reconstruct_abstract(inverted_index: dict[str, list[int]] | None) -> str | None:
    """
    Reconstruct plain-text abstract from an OpenAlex abstract_inverted_index dictionary.

    Args:
        inverted_index: Mapping of words to their token position indices.

    Returns:
        Coherent, human-readable abstract string, or None if empty/invalid.
    """
    if not inverted_index or not isinstance(inverted_index, dict):
        return None

    positions: list[tuple[int, str]] = []
    for word, pos_list in inverted_index.items():
        if not isinstance(pos_list, list):
            continue
        for pos in pos_list:
            if isinstance(pos, int):
                positions.append((pos, word))

    if not positions:
        return None

    positions.sort(key=lambda x: x[0])
    return " ".join(word for _, word in positions).strip() or None

resolve_identifier async

resolve_identifier(identifier)

Resolve a paper by DOI, arXiv ID, or OpenAlex ID.

Parameters:

Name	Type	Description	Default
identifier	str	DOI, arXiv ID, or OpenAlex ID.	required

Returns:

Type	Description
CandidatePaper | None	Resolved CandidatePaper, or None if not found.

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

async def resolve_identifier(self, identifier: str) -> CandidatePaper | None:
    """
    Resolve a paper by DOI, arXiv ID, or OpenAlex ID.

    Args:
        identifier: DOI, arXiv ID, or OpenAlex ID.

    Returns:
        Resolved CandidatePaper, or None if not found.
    """
    ident = identifier.strip() if identifier else ""
    if not ident:
        return None

    # 1. OpenAlex ID (e.g. "https://openalex.org/W..." or "W123456789")
    m_oa = re.search(r"\b(W\d+)\b", ident, re.IGNORECASE)
    if m_oa:
        oa_id = m_oa.group(1).upper()
        url = f"{OPENALEX_WORKS_URL}/{oa_id}"
        params = {"select": OPENALEX_SELECT_FIELDS}
        data = await self._get_with_retry(url, params=params)
        if data and isinstance(data, dict) and "id" in data:
            return self._parse_work(data)

    # 2. DOI resolution
    clean_doi = normalize_doi(ident)
    if clean_doi and clean_doi.startswith("10.") and "/" in clean_doi:
        # If it's a DataCite arXiv DOI, extract the arXiv ID as fallback
        arxiv_from_doi = normalize_arxiv_id(clean_doi)

        # Query OpenAlex with canonical DOI filter
        params = {
            "filter": f"doi:https://doi.org/{clean_doi}",
            "select": OPENALEX_SELECT_FIELDS,
            "per-page": 1,
        }
        data = await self._get_with_retry(OPENALEX_WORKS_URL, params=params)
        if data and isinstance(data, dict):
            results = data.get("results") or []
            if isinstance(results, list) and results:
                return self._parse_work(results[0])

        # If DOI query didn't find it but it's an arXiv DOI, fall through to arXiv lookup
        if arxiv_from_doi:
            ident = arxiv_from_doi

    # 3. arXiv ID resolution
    clean_arxiv = normalize_arxiv_id(ident)
    if clean_arxiv:
        # First try DataCite DOI filter
        datacite_doi = f"https://doi.org/10.48550/arxiv.{clean_arxiv}"
        params = {
            "filter": f"doi:{datacite_doi}",
            "select": OPENALEX_SELECT_FIELDS,
            "per-page": 1,
        }
        data = await self._get_with_retry(OPENALEX_WORKS_URL, params=params)
        if data and isinstance(data, dict):
            results = data.get("results") or []
            if isinstance(results, list) and results:
                return self._parse_work(results[0])

        # Secondary search for landing page or arxiv ID text
        params_search = {
            "search": clean_arxiv,
            "select": OPENALEX_SELECT_FIELDS,
            "per-page": 5,
        }
        data_search = await self._get_with_retry(OPENALEX_WORKS_URL, params=params_search)
        if data_search and isinstance(data_search, dict):
            results = data_search.get("results") or []
            if isinstance(results, list):
                for w in results:
                    if isinstance(w, dict) and self.extract_arxiv_id_from_work(w) == clean_arxiv:
                        return self._parse_work(w)

    return None

search async

search(query, limit=10, year_from=None, year_to=None, is_oa_only=False)

Execute free-text or conceptual search against OpenAlex works.

Parameters:

Name	Type	Description	Default
query	str	User search query string.	required
limit	int	Maximum candidate works to return.	10
year_from	int | None	Optional minimum publication year.	None
year_to	int | None	Optional maximum publication year.	None
is_oa_only	bool	Restrict to open-access papers.	False

Returns:

Type	Description
list[CandidatePaper]	List of parsed CandidatePaper instances.

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

async def search(
    self,
    query: str,
    limit: int = 10,
    year_from: int | None = None,
    year_to: int | None = None,
    is_oa_only: bool = False,
) -> list[CandidatePaper]:
    """
    Execute free-text or conceptual search against OpenAlex works.

    Args:
        query: User search query string.
        limit: Maximum candidate works to return.
        year_from: Optional minimum publication year.
        year_to: Optional maximum publication year.
        is_oa_only: Restrict to open-access papers.

    Returns:
        List of parsed CandidatePaper instances.
    """
    q = query.strip() if query else ""
    if not q:
        return []

    # If query looks like an exact DOI or arXiv ID, delegate directly to resolve_identifier
    clean_doi = normalize_doi(q)
    if clean_doi and ("/" in clean_doi) and clean_doi.startswith("10."):
        single = await self.resolve_identifier(clean_doi)
        return [single] if single else []

    clean_arxiv = normalize_arxiv_id(q)
    if clean_arxiv and re.match(r"^(\d{4}\.\d{4,5}|[a-zA-Z\-]+(?:\.[a-zA-Z\-]+)?\/\d{7})$", clean_arxiv):
        single = await self.resolve_identifier(clean_arxiv)
        return [single] if single else []

    per_page = min(max(1, limit), 100)
    params: dict[str, Any] = {
        "search": q,
        "per-page": per_page,
        "select": OPENALEX_SELECT_FIELDS,
    }

    # Build filter expressions
    filters: list[str] = []
    if is_oa_only:
        filters.append("is_oa:true")

    if year_from is not None and year_to is not None:
        if year_from == year_to:
            filters.append(f"publication_year:{year_from}")
        else:
            filters.append(f"publication_year:{year_from}-{year_to}")
    elif year_from is not None:
        filters.append(f"publication_year:>{year_from - 1}")
    elif year_to is not None:
        filters.append(f"publication_year:<{year_to + 1}")

    if filters:
        params["filter"] = ",".join(filters)

    data = await self._get_with_retry(OPENALEX_WORKS_URL, params=params)
    if not data:
        return []

    results = data.get("results") or []
    candidates: list[CandidatePaper] = []
    if isinstance(results, list):
        for work in results:
            if isinstance(work, dict):
                candidates.append(self._parse_work(work))

    return candidates[:limit]

SearchProvider

Bases: ABC

Abstract base class for scholarly literature search providers.

resolve_identifier abstractmethod async

resolve_identifier(identifier)

Resolve a single scholarly paper by DOI, arXiv ID, or provider-specific ID.

Parameters:

Name	Type	Description	Default
identifier	str	DOI, arXiv ID, or provider ID string.	required

Returns:

Type	Description
CandidatePaper | None	CandidatePaper if resolved, or None if not found.

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

@abstractmethod
async def resolve_identifier(self, identifier: str) -> CandidatePaper | None:
    """
    Resolve a single scholarly paper by DOI, arXiv ID, or provider-specific ID.

    Args:
        identifier: DOI, arXiv ID, or provider ID string.

    Returns:
        CandidatePaper if resolved, or None if not found.
    """
    ...

search abstractmethod async

search(query, limit=10, year_from=None, year_to=None, is_oa_only=False)

Search for scholarly literature matching the query and filters.

Parameters:

Name	Type	Description	Default
query	str	Free-text search terms or conceptual query.	required
limit	int	Maximum number of candidate papers to return.	10
year_from	int | None	Optional minimum publication year (inclusive).	None
year_to	int | None	Optional maximum publication year (inclusive).	None
is_oa_only	bool	When True, restrict results to open-access works.	False

Returns:

Type	Description
list[CandidatePaper]	List of CandidatePaper instances.

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

@abstractmethod
async def search(
    self,
    query: str,
    limit: int = 10,
    year_from: int | None = None,
    year_to: int | None = None,
    is_oa_only: bool = False,
) -> list[CandidatePaper]:
    """
    Search for scholarly literature matching the query and filters.

    Args:
        query: Free-text search terms or conceptual query.
        limit: Maximum number of candidate papers to return.
        year_from: Optional minimum publication year (inclusive).
        year_to: Optional maximum publication year (inclusive).
        is_oa_only: When True, restrict results to open-access works.

    Returns:
        List of CandidatePaper instances.
    """
    ...

TokenBucketRateLimiter

TokenBucketRateLimiter(rate=10.0, capacity=10.0)

Token-bucket rate limiter for respectful, throttled API consumption. Thread/asyncio safe.

Parameters:

Name	Type	Description	Default
rate	float	Refill rate in tokens per second.	10.0
capacity	float	Maximum burst capacity in tokens.	10.0

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

def __init__(self, rate: float = 10.0, capacity: float = 10.0) -> None:
    """
    Args:
        rate: Refill rate in tokens per second.
        capacity: Maximum burst capacity in tokens.
    """
    self.rate = max(0.1, float(rate))
    self.capacity = max(1.0, float(capacity))
    self.tokens = self.capacity
    self.last_update = time.monotonic()
    self._lock = asyncio.Lock()

acquire async

acquire(tokens=1.0)

Acquire tokens from the bucket, sleeping if necessary until replenished.

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

async def acquire(self, tokens: float = 1.0) -> None:
    """Acquire `tokens` from the bucket, sleeping if necessary until replenished."""
    async with self._lock:
        now = time.monotonic()
        elapsed = now - self.last_update
        self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)
        self.last_update = now

        if self.tokens < tokens:
            needed = tokens - self.tokens
            wait_time = needed / self.rate
            await asyncio.sleep(wait_time)
            self.tokens = 0.0
            self.last_update = time.monotonic()
        else:
            self.tokens -= tokens

UnpaywallResolver

UnpaywallResolver(email=DEFAULT_POLITE_EMAIL, timeout=5.0)

Resolves open-access fulltext PDF URLs via the Unpaywall API.

Initialize the Unpaywall resolver.

Parameters:

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

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

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

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

resolve_pdf async

resolve_pdf(doi, email=None)

Query Unpaywall for a legal, direct Open Access PDF URL for the given DOI.

Parameters:

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

Returns:

Type	Description
str | None	Direct PDF URL string if available and verified, or None.

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

async def resolve_pdf(self, doi: str, email: str | None = None) -> str | None:
    """
    Query Unpaywall for a legal, direct Open Access PDF URL for the given DOI.

    Args:
        doi: Digital Object Identifier (canonical or raw).
        email: User or developer email for polite pool identification.

    Returns:
        Direct PDF URL string if available and verified, or None.
    """
    clean_doi = normalize_doi(doi)
    if not clean_doi:
        return None

    polite_email = email or self.email
    url = f"https://api.unpaywall.org/v2/{clean_doi}"
    params = {"email": polite_email}
    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, params=params, headers=headers)
            if resp.status_code != 200:
                logger.debug("Unpaywall returned status %d for DOI %s", resp.status_code, clean_doi)
                return None
            data: dict[str, Any] = resp.json()

        best_oa = data.get("best_oa_location")
        if isinstance(best_oa, dict):
            pdf_url = best_oa.get("url_for_pdf")
            if pdf_url and isinstance(pdf_url, str) and pdf_url.strip():
                return pdf_url.strip()
            # Fallback to landing URL if it directly points to a PDF
            url_candidate = best_oa.get("url")
            if url_candidate and isinstance(url_candidate, str) and url_candidate.strip().lower().endswith(".pdf"):
                return url_candidate.strip()

        # Secondary check: inspect any oa_locations for url_for_pdf
        oa_locations = data.get("oa_locations")
        if isinstance(oa_locations, list):
            for loc in oa_locations:
                if isinstance(loc, dict):
                    loc_pdf = loc.get("url_for_pdf")
                    if loc_pdf and isinstance(loc_pdf, str) and loc_pdf.strip():
                        return loc_pdf.strip()

        return None
    except Exception as e:
        logger.debug("Unpaywall resolution 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})"