engine

zocrates.discovery.engine

Discovery Engine orchestrating search providers, deduplication, abstract backfill, and OA PDF resolution.

DEFAULT_POLITE_EMAIL module-attribute

DEFAULT_POLITE_EMAIL = 'zocrates-dev@proton.me'

logger module-attribute

logger = logging.getLogger(__name__)

AbstractFallback

AbstractFallback(email=DEFAULT_POLITE_EMAIL, timeout=2.0)

Manages 3-tier opportunistic fallback for paper abstracts

• Tier 1: Direct OpenAlex abstract (present in CandidatePaper).
• Tier 2: Semantic Scholar REST API (and NCBI PubMed E-utilities for biomedical papers) with 2.0s timeout.
• Tier 3: Clean fallback to OpenAlex keywords and primary topic.

Initialize the abstract fallback resolver.

Parameters:

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

Source code in zocrates/discovery/abstract_fallback.py

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

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

backfill_abstract async

backfill_abstract(candidate)

Backfill missing abstract and TLDR on a candidate paper using opportunistic fallbacks.

Parameters:

Name	Type	Description	Default
candidate	CandidatePaper	CandidatePaper instance.	required

Returns:

Type	Description
CandidatePaper	CandidatePaper with abstract/tldr backfilled if resolved.

Source code in zocrates/discovery/abstract_fallback.py

async def backfill_abstract(self, candidate: CandidatePaper) -> CandidatePaper:
    """
    Backfill missing abstract and TLDR on a candidate paper using opportunistic fallbacks.

    Args:
        candidate: CandidatePaper instance.

    Returns:
        CandidatePaper with abstract/tldr backfilled if resolved.
    """
    # Tier 1: Candidate already has an abstract
    if candidate.abstract and candidate.abstract.strip():
        return candidate

    # Tier 2: Try Semantic Scholar
    doi_clean = normalize_doi(candidate.doi)
    arxiv_clean = normalize_arxiv_id(candidate.arxiv_id)

    s2_identifier: str | None = None
    if doi_clean:
        s2_identifier = f"DOI:{doi_clean}"
    elif arxiv_clean:
        s2_identifier = f"ARXIV:{arxiv_clean}"

    if s2_identifier:
        abs_text, tldr_text = await self._fetch_semantic_scholar(s2_identifier)
        if abs_text:
            candidate.abstract = abs_text
        if tldr_text:
            candidate.tldr = tldr_text

    # If abstract still missing and DOI present, try PubMed for biomedical papers
    if not candidate.abstract and doi_clean:
        is_biomed = False
        if candidate.primary_topic and any(
            term in candidate.primary_topic.lower()
            for term in ("bio", "medicine", "health", "clinical", "pharm", "immun", "neuro")
        ):
            is_biomed = True

        if is_biomed:
            pm_abstract = await self._fetch_pubmed_abstract(doi_clean)
            if pm_abstract:
                candidate.abstract = pm_abstract

    # Tier 3: If no abstract available, candidate cleanly retains keywords and primary_topic
    return candidate

backfill_batch async

backfill_batch(candidates)

Backfill missing abstracts concurrently across a batch of candidates.

Parameters:

Name	Type	Description	Default
candidates	list[CandidatePaper]	List of CandidatePaper instances.	required

Returns:

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

Source code in zocrates/discovery/abstract_fallback.py

async def backfill_batch(self, candidates: list[CandidatePaper]) -> list[CandidatePaper]:
    """
    Backfill missing abstracts concurrently across a batch of candidates.

    Args:
        candidates: List of CandidatePaper instances.

    Returns:
        List of updated CandidatePaper instances.
    """
    tasks = [self.backfill_abstract(c) for c in candidates]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    updated: list[CandidatePaper] = []
    for orig, res in zip(candidates, results, strict=False):
        if isinstance(res, CandidatePaper):
            updated.append(res)
        else:
            updated.append(orig)
    return updated

CandidatePaper dataclass

CandidatePaper(candidate_id, title, authors=list(), year=None, venue=None, doi=None, arxiv_id=None, openalex_id='', abstract=None, tldr=None, is_oa=False, oa_pdf_url=None, oa_pdf_urls=list(), citation_count=0, primary_topic=None, keywords=list(), in_library=False)

Represents a discovered scholarly paper candidate.

from_dict classmethod

from_dict(data)

Deserialize CandidatePaper from a dictionary.

Source code in zocrates/discovery/models.py

@classmethod
def from_dict(cls, data: dict[str, Any]) -> "CandidatePaper":
    """Deserialize CandidatePaper from a dictionary."""
    citation_count_val = data.get("citation_count", 0)
    try:
        citation_count = int(citation_count_val) if citation_count_val is not None else 0
    except (ValueError, TypeError):
        citation_count = 0

    year_val = data.get("year")
    year: int | None = None
    if year_val is not None:
        try:
            year = int(year_val)
        except (ValueError, TypeError):
            year = None

    raw_authors = data.get("authors")
    if isinstance(raw_authors, list):
        authors = [str(a).strip() for a in raw_authors if a and str(a).strip()]
    elif isinstance(raw_authors, str) and raw_authors.strip():
        authors = [raw_authors.strip()]
    else:
        authors = []

    raw_keywords = data.get("keywords")
    if isinstance(raw_keywords, list):
        keywords = [str(k).strip() for k in raw_keywords if k and str(k).strip()]
    elif isinstance(raw_keywords, str) and raw_keywords.strip():
        keywords = [raw_keywords.strip()]
    else:
        keywords = []

    def _clean_opt_str(val: Any) -> str | None:
        """Clean and trim an optional string or scalar value.

        Args:
            val: Input value to convert to a stripped string.

        Returns:
            Stripped string, or None if empty or invalid.
        """
        if val is not None and isinstance(val, (str, int, float)):
            s = str(val).strip()
            return s if s else None
        return None

    raw_urls = data.get("oa_pdf_urls")
    oa_pdf_urls: list[str] = []
    if isinstance(raw_urls, list):
        oa_pdf_urls = [str(u).strip() for u in raw_urls if u and str(u).strip()]
    primary_pdf = _clean_opt_str(data.get("oa_pdf_url"))
    if primary_pdf and primary_pdf not in oa_pdf_urls:
        oa_pdf_urls.insert(0, primary_pdf)

    return cls(
        candidate_id=str(data.get("candidate_id") or ""),
        title=str(data.get("title") or "").strip(),
        authors=authors,
        year=year,
        venue=_clean_opt_str(data.get("venue")),
        doi=_clean_opt_str(data.get("doi")),
        arxiv_id=_clean_opt_str(data.get("arxiv_id")),
        openalex_id=str(data.get("openalex_id") or ""),
        abstract=_clean_opt_str(data.get("abstract")),
        tldr=_clean_opt_str(data.get("tldr")),
        is_oa=bool(data.get("is_oa", False)),
        oa_pdf_url=primary_pdf,
        oa_pdf_urls=oa_pdf_urls,
        citation_count=citation_count,
        primary_topic=_clean_opt_str(data.get("primary_topic")),
        keywords=keywords,
        in_library=bool(data.get("in_library", False)),
    )

to_dict

to_dict()

Serialize CandidatePaper to a plain dictionary.

Source code in zocrates/discovery/models.py

def to_dict(self) -> dict[str, Any]:
    """Serialize CandidatePaper to a plain dictionary."""
    return {
        "candidate_id": self.candidate_id,
        "title": self.title,
        "authors": list(self.authors),
        "year": self.year,
        "venue": self.venue,
        "doi": self.doi,
        "arxiv_id": self.arxiv_id,
        "openalex_id": self.openalex_id,
        "abstract": self.abstract,
        "tldr": self.tldr,
        "is_oa": self.is_oa,
        "oa_pdf_url": self.oa_pdf_url,
        "oa_pdf_urls": list(self.oa_pdf_urls),
        "citation_count": self.citation_count,
        "primary_topic": self.primary_topic,
        "keywords": list(self.keywords),
        "in_library": self.in_library,
    }

Deduplicator

Deduplicator(zotero_db=None, existing_dois=None, existing_arxiv_ids=None, existing_records=None, existing_titles=None)

Enforces strict novelty against the user's local Zotero library and removes duplicates within search result batches.

Initialize Deduplicator with library state from ZoteroDB or explicit collections.

Parameters:

Name	Type	Description	Default
zotero_db	Any | None	Optional ZoteroDB instance to query live library records.	None
existing_dois	set[str] | None	Pre-populated set of normalized DOIs.	None
existing_arxiv_ids	set[str] | None	Pre-populated set of normalized arXiv IDs.	None
existing_records	list[dict[str, Any]] | None	Pre-populated list of library record dicts (with title, year, creators).	None
existing_titles	list[str] | None	Pre-populated list of raw or normalized titles.	None

Source code in zocrates/discovery/dedup.py

def __init__(
    self,
    zotero_db: Any | None = None,
    existing_dois: set[str] | None = None,
    existing_arxiv_ids: set[str] | None = None,
    existing_records: list[dict[str, Any]] | None = None,
    existing_titles: list[str] | None = None,
) -> None:
    """
    Initialize Deduplicator with library state from ZoteroDB or explicit collections.

    Args:
        zotero_db: Optional ZoteroDB instance to query live library records.
        existing_dois: Pre-populated set of normalized DOIs.
        existing_arxiv_ids: Pre-populated set of normalized arXiv IDs.
        existing_records: Pre-populated list of library record dicts (with title, year, creators).
        existing_titles: Pre-populated list of raw or normalized titles.
    """
    self.existing_dois: set[str] = set(existing_dois) if existing_dois else set()
    self.existing_arxiv_ids: set[str] = set(existing_arxiv_ids) if existing_arxiv_ids else set()
    self.existing_records: list[dict[str, Any]] = list(existing_records) if existing_records else []

    if existing_titles:
        for t in existing_titles:
            norm_t = normalize_title(t)
            if norm_t:
                self.existing_records.append({
                    "title": t,
                    "normalized_title": norm_t,
                    "year": None,
                    "creators": [],
                })

    # Load from ZoteroDB if provided
    if zotero_db is not None:
        self._load_from_zotero_db(zotero_db)

    # Precompute normalized titles on existing records
    for rec in self.existing_records:
        if "normalized_title" not in rec and rec.get("title"):
            rec["normalized_title"] = normalize_title(rec["title"])

filter_novel

filter_novel(candidates)

Filter candidate papers enforcing strict novelty against the library and deduplicating across the current batch.

Parameters:

Name	Type	Description	Default
candidates	list[CandidatePaper]	List of CandidatePaper instances.	required

Returns:

Type	Description
list[CandidatePaper]	List of strictly novel CandidatePaper instances, deduplicated intra-batch.

Source code in zocrates/discovery/dedup.py

def filter_novel(self, candidates: list[CandidatePaper]) -> list[CandidatePaper]:
    """
    Filter candidate papers enforcing strict novelty against the library
    and deduplicating across the current batch.

    Args:
        candidates: List of CandidatePaper instances.

    Returns:
        List of strictly novel CandidatePaper instances, deduplicated intra-batch.
    """
    novel: list[CandidatePaper] = []
    seen_dois: set[str] = set()
    seen_arxiv_ids: set[str] = set()
    seen_titles: set[str] = set()

    for c in candidates:
        # 1. Strict novelty check against existing library
        if self.is_in_library(c):
            continue

        # 2. Intra-batch deduplication
        c_doi = normalize_doi(c.doi)
        c_arxiv = normalize_arxiv_id(c.arxiv_id)
        c_title = normalize_title(c.title)

        if c_doi and c_doi in seen_dois:
            continue
        if c_arxiv and c_arxiv in seen_arxiv_ids:
            continue
        if c_title and c_title in seen_titles:
            continue

        if c_doi:
            seen_dois.add(c_doi)
        if c_arxiv:
            seen_arxiv_ids.add(c_arxiv)
        if c_title:
            seen_titles.add(c_title)

        novel.append(c)

    return novel

is_in_library

is_in_library(candidate)

Check whether candidate paper already exists in the user's library.

Evaluates

1. DOI exact match.
2. arXiv ID exact match.
3. Exact normalized title match.
4. Fuzzy title similarity (>= 0.92) with matching year or author.

Parameters:

Name	Type	Description	Default
candidate	CandidatePaper	CandidatePaper instance.	required

Returns:

Type	Description
bool	True if the paper exists in the library, False otherwise.

Source code in zocrates/discovery/dedup.py

def is_in_library(self, candidate: CandidatePaper) -> bool:
    """
    Check whether candidate paper already exists in the user's library.

    Evaluates:
      1. DOI exact match.
      2. arXiv ID exact match.
      3. Exact normalized title match.
      4. Fuzzy title similarity (>= 0.92) with matching year or author.

    Args:
        candidate: CandidatePaper instance.

    Returns:
        True if the paper exists in the library, False otherwise.
    """
    # 1. Exact DOI match
    if candidate.doi:
        cand_doi = normalize_doi(candidate.doi)
        if cand_doi and cand_doi in self.existing_dois:
            return True

    # 2. Exact arXiv ID match
    if candidate.arxiv_id:
        cand_arxiv = normalize_arxiv_id(candidate.arxiv_id)
        if cand_arxiv and cand_arxiv in self.existing_arxiv_ids:
            return True

    # 3. Title matching (exact and fuzzy)
    cand_norm = normalize_title(candidate.title)
    if not cand_norm or len(cand_norm) < 4:
        return False

    cand_tokens = set(cand_norm.split())

    for rec in self.existing_records:
        rec_norm = rec.get("normalized_title") or normalize_title(rec.get("title"))
        if not rec_norm:
            continue

        # Exact normalized title match
        if cand_norm == rec_norm:
            return True

        rec_tokens = set(rec_norm.split())
        token_intersection = cand_tokens.intersection(rec_tokens)
        token_jaccard = (
            len(token_intersection) / len(cand_tokens.union(rec_tokens))
            if cand_tokens or rec_tokens
            else 0.0
        )
        min_token_len = min(len(cand_tokens), len(rec_tokens))
        token_containment = (
            len(token_intersection) / min_token_len
            if min_token_len > 0
            else 0.0
        )

        similarity = difflib.SequenceMatcher(None, cand_norm, rec_norm).ratio()
        if similarity >= 0.98:
            return True

        if (
            similarity >= 0.88
            or token_jaccard >= 0.80
            or (token_containment >= 0.85 and min_token_len >= 4)
        ):
            cand_year = candidate.year
            rec_year = rec.get("year")
            year_match = False
            if cand_year is not None and rec_year is not None:
                if abs(cand_year - rec_year) <= 1:
                    year_match = True
            elif cand_year is None or rec_year is None:
                year_match = True

            author_match = False
            cand_authors = candidate.authors
            rec_creators = rec.get("creators") or []
            if cand_authors and rec_creators:
                cand_surnames = {a.split()[-1].lower() for a in cand_authors if a.strip()}
                rec_surnames = {
                    (c.split(",")[0] if "," in c else c.split()[-1]).strip().lower()
                    for c in rec_creators
                    if c.strip()
                }
                if cand_surnames.intersection(rec_surnames):
                    author_match = True

            if (
                (year_match and author_match)
                or (similarity >= 0.92 and (year_match or author_match))
                or (token_containment >= 0.95 and (year_match or author_match))
            ):
                return True

    return False

DiscoveryEngine

DiscoveryEngine(provider=None, deduplicator=None, abstract_fallback=None, unpaywall=None, email=DEFAULT_POLITE_EMAIL)

Coordinates scholarly search across providers, enforces strict novelty against the local Zotero library, backfills abstracts, and resolves Open Access PDFs.

Parameters:

Name	Type	Description	Default
provider	SearchProvider | None	SearchProvider instance (defaults to OpenAlexProvider).	None
deduplicator	Deduplicator | None	Deduplicator instance with loaded library state.	None
abstract_fallback	AbstractFallback | None	AbstractFallback instance for missing abstracts.	None
unpaywall	UnpaywallResolver | None	UnpaywallResolver instance for OA PDF links.	None
email	str	Polite pool email for upstream academic APIs.	DEFAULT_POLITE_EMAIL

Source code in zocrates/discovery/engine.py

def __init__(
    self,
    provider: SearchProvider | None = None,
    deduplicator: Deduplicator | None = None,
    abstract_fallback: AbstractFallback | None = None,
    unpaywall: UnpaywallResolver | None = None,
    email: str = DEFAULT_POLITE_EMAIL,
) -> None:
    """
    Args:
        provider: SearchProvider instance (defaults to OpenAlexProvider).
        deduplicator: Deduplicator instance with loaded library state.
        abstract_fallback: AbstractFallback instance for missing abstracts.
        unpaywall: UnpaywallResolver instance for OA PDF links.
        email: Polite pool email for upstream academic APIs.
    """
    self.email = email
    self.provider = provider or OpenAlexProvider(email=email)
    self.deduplicator = deduplicator or Deduplicator()
    self.abstract_fallback = abstract_fallback or AbstractFallback(email=email)
    self.unpaywall = unpaywall or UnpaywallResolver(email=email)

resolve_identifier async

resolve_identifier(identifier)

Resolve a single paper by DOI, arXiv ID, or OpenAlex ID and check its library presence.

Parameters:

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

Returns:

Type	Description
tuple[CandidatePaper | None, bool]	Tuple of (CandidatePaper or None, in_library boolean).

Source code in zocrates/discovery/engine.py

async def resolve_identifier(self, identifier: str) -> tuple[CandidatePaper | None, bool]:
    """
    Resolve a single paper by DOI, arXiv ID, or OpenAlex ID and check its library presence.

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

    Returns:
        Tuple of (CandidatePaper or None, in_library boolean).
    """
    ident = identifier.strip() if identifier else ""
    if not ident:
        return None, False

    try:
        candidate = await self.provider.resolve_identifier(ident)
    except Exception as e:
        logger.warning("Failed resolving identifier %r: %s", ident, e)
        return None, False

    if candidate is None:
        return None, False

    # Check library presence
    in_library = self.deduplicator.is_in_library(candidate)
    candidate.in_library = in_library

    # Only perform external network lookups if paper is not already in the library
    if not in_library:
        # Backfill abstract if missing
        if not candidate.abstract:
            try:
                candidate = await self.abstract_fallback.backfill_abstract(candidate)
            except Exception as e:
                logger.debug("Abstract backfill failed for %r: %s", ident, e)

        # Resolve OA PDF if missing
        if not candidate.oa_pdf_url and candidate.doi:
            try:
                pdf_url = await self.unpaywall.resolve_pdf(candidate.doi)
                if pdf_url:
                    candidate.oa_pdf_url = pdf_url
                    candidate.is_oa = True
            except Exception as e:
                logger.debug("Unpaywall PDF resolution failed for %r: %s", ident, e)

    return candidate, in_library

search async

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

Search for scholarly literature, filtering out papers already in the library and enriching top results with abstracts and OA PDF URLs.

Parameters:

Name	Type	Description	Default
query	str	Free-text search terms or conceptual query.	required
limit	int	Target number of novel candidate papers to return.	10
year_from	int | None	Optional starting publication year.	None
year_to	int | None	Optional ending publication year.	None
is_oa_only	bool	Restrict results to open-access papers.	False

Returns:

Type	Description
list[CandidatePaper]	List of novel CandidatePaper instances (at most limit).

Source code in zocrates/discovery/engine.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]:
    """
    Search for scholarly literature, filtering out papers already in the library
    and enriching top results with abstracts and OA PDF URLs.

    Args:
        query: Free-text search terms or conceptual query.
        limit: Target number of novel candidate papers to return.
        year_from: Optional starting publication year.
        year_to: Optional ending publication year.
        is_oa_only: Restrict results to open-access papers.

    Returns:
        List of novel CandidatePaper instances (at most `limit`).
    """
    q = query.strip() if query else ""
    if not q:
        return []

    # Over-fetch from provider to account for library deduplication
    fetch_limit = min(max(limit * 2, 20), 100)
    try:
        raw_candidates = await self.provider.search(
            query=q,
            limit=fetch_limit,
            year_from=year_from,
            year_to=year_to,
            is_oa_only=is_oa_only,
        )
    except Exception as e:
        logger.warning("Provider search failed for query %r: %s", q, e)
        return []

    if not raw_candidates:
        return []

    # Filter strictly novel papers against local library and batch duplicates
    novel_candidates = self.deduplicator.filter_novel(raw_candidates)

    # Slice to requested limit
    selected = novel_candidates[:limit]
    if not selected:
        return []

    # Opportunistic abstract backfill for selected candidates lacking abstracts
    try:
        selected = await self.abstract_fallback.backfill_batch(selected)
    except Exception as e:
        logger.debug("Abstract backfill encountered error: %s", e)

    # Resolve OA PDF URLs via Unpaywall for OA candidates missing direct PDF URLs
    async def _resolve_oa_pdf(c: CandidatePaper) -> None:
        """Resolve Open Access PDF URL via Unpaywall for a candidate paper.

        Args:
            c: CandidatePaper to inspect and update with resolved PDF URL.
        """
        if not c.oa_pdf_url and c.doi:
            try:
                pdf_url = await self.unpaywall.resolve_pdf(c.doi)
                if pdf_url:
                    c.oa_pdf_url = pdf_url
                    c.is_oa = True
            except Exception as ex:
                logger.debug("Unpaywall resolution failed for DOI %s: %s", c.doi, ex)

    pdf_tasks = [_resolve_oa_pdf(c) for c in selected if not c.oa_pdf_url and c.doi]
    if pdf_tasks:
        await asyncio.gather(*pdf_tasks, return_exceptions=True)

    return selected

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.
    """
    ...

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

get_default_discovery_engine

get_default_discovery_engine(zotero_data_dir=None, polite_email=None, openalex_api_key=None)

Instantiate a configured DiscoveryEngine with local Zotero deduplication.

Source code in zocrates/discovery/engine.py

def get_default_discovery_engine(
    zotero_data_dir: Path | str | None = None,
    polite_email: str | None = None,
    openalex_api_key: str | None = None,
) -> DiscoveryEngine:
    """Instantiate a configured DiscoveryEngine with local Zotero deduplication."""
    from zocrates.discovery.abstract_fallback import AbstractFallback
    from zocrates.discovery.dedup import Deduplicator
    from zocrates.discovery.providers.base import DEFAULT_POLITE_EMAIL
    from zocrates.discovery.providers.openalex import OpenAlexProvider
    from zocrates.discovery.providers.unpaywall import UnpaywallResolver
    from zocrates.config import get_openalex_api_key, get_scholar_polite_email, resolve_zotero_data_dir

    resolved_dir = zotero_data_dir or resolve_zotero_data_dir()
    path_dir = Path(resolved_dir) if resolved_dir else None
    email = polite_email
    if not email:
        try:
            email = get_scholar_polite_email(path_dir)
        except Exception:
            pass
    if not email:
        email = DEFAULT_POLITE_EMAIL

    api_key = openalex_api_key
    if not api_key:
        try:
            api_key = get_openalex_api_key(path_dir)
        except Exception:
            pass

    zotero_db = None
    if resolved_dir:
        sqlite_path = Path(resolved_dir) / "zotero.sqlite"
        storage_dir = Path(resolved_dir) / "storage"
        if sqlite_path.exists():
            try:
                from zocrates.rag.zotero_db import ZoteroDB

                zotero_db = ZoteroDB(sqlite_path=sqlite_path, storage_dir=storage_dir)
            except Exception as e:
                logger.debug("Could not initialize ZoteroDB for discovery deduplication: %s", e)

    deduplicator = Deduplicator(zotero_db=zotero_db)
    provider = OpenAlexProvider(email=email, api_key=api_key)
    abstract_fallback = AbstractFallback(email=email)
    unpaywall = UnpaywallResolver(email=email)

    return DiscoveryEngine(
        provider=provider,
        deduplicator=deduplicator,
        abstract_fallback=abstract_fallback,
        unpaywall=unpaywall,
        email=email,
    )