Index

zocrates.discovery

Scholarly literature discovery and batch import package for Zocrates.

__all__ module-attribute

__all__ = ['CandidatePaper', 'StagedItem', 'DiscoveryEngine', 'get_default_discovery_engine', 'Deduplicator', 'OpenAlexProvider', 'get_staging_root', 'cleanup_stale_staging_directories', 'download_and_stage_pdf', 'normalize_doi', 'normalize_arxiv_id', 'normalize_title']

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]

StagedItem dataclass

StagedItem(candidate_id, staged_path, metadata=dict())

Represents a downloaded or staged item ready for Zotero batch import.

from_dict classmethod

from_dict(data)

Deserialize StagedItem from a dictionary.

Source code in zocrates/discovery/models.py

@classmethod
def from_dict(cls, data: dict[str, Any]) -> "StagedItem":
    """Deserialize StagedItem from a dictionary."""
    return cls(
        candidate_id=str(data.get("candidate_id") or ""),
        staged_path=str(data.get("staged_path") or ""),
        metadata=dict(data.get("metadata") or {}),
    )

to_dict

to_dict()

Serialize StagedItem to a plain dictionary.

Source code in zocrates/discovery/models.py

def to_dict(self) -> dict[str, Any]:
    """Serialize StagedItem to a plain dictionary."""
    return {
        "candidate_id": self.candidate_id,
        "staged_path": self.staged_path,
        "metadata": dict(self.metadata),
    }

cleanup_stale_staging_directories

cleanup_stale_staging_directories(max_age_seconds=7200)

Remove staging subdirectories older than max_age_seconds (default: 2 hours).

Parameters:

Name	Type	Description	Default
max_age_seconds	int	Age threshold in seconds.	7200

Returns:

Type	Description
int	Number of directories removed.

Source code in zocrates/discovery/staging.py

def cleanup_stale_staging_directories(max_age_seconds: int = 7200) -> int:
    """
    Remove staging subdirectories older than `max_age_seconds` (default: 2 hours).

    Args:
        max_age_seconds: Age threshold in seconds.

    Returns:
        Number of directories removed.
    """
    root = get_staging_root()
    if not root.exists():
        return 0

    removed = 0
    now = time.time()
    for item in root.iterdir():
        if item.is_dir():
            try:
                mtime = item.stat().st_mtime
                if (now - mtime) > max_age_seconds:
                    shutil.rmtree(item, ignore_errors=True)
                    removed += 1
            except Exception as e:
                logger.debug("Failed removing stale staging directory %s: %s", item, e)
    return removed

download_and_stage_pdf async

download_and_stage_pdf(candidate, staging_dir, client=None)

Download and stage an Open Access PDF for a candidate paper.

Enforces

• SSRF Protection: strictly requires https:// URL scheme.
• Safe Filename: sanitizes candidate_id against directory traversal.
• Directory Containment: ensures target path is strictly within staging_dir.
• Magic Byte Verification: first 5 bytes must equal b"%PDF-".
• Streaming Byte Cap: aborts if cumulative bytes exceed MAX_PDF_BYTES (100 MB).

Parameters:

Name	Type	Description	Default
candidate	CandidatePaper	CandidatePaper instance.	required
staging_dir	Path	Target batch directory Path.	required
client	AsyncClient | None	Optional existing httpx.AsyncClient for connection reuse.	None

Returns:

Type	Description
str | None	Absolute file path string on success, or None on failure/non-PDF.

Source code in zocrates/discovery/staging.py

async def download_and_stage_pdf(
    candidate: CandidatePaper,
    staging_dir: Path,
    client: httpx.AsyncClient | None = None,
) -> str | None:
    """
    Download and stage an Open Access PDF for a candidate paper.

    Enforces:
      - SSRF Protection: strictly requires https:// URL scheme.
      - Safe Filename: sanitizes candidate_id against directory traversal.
      - Directory Containment: ensures target path is strictly within staging_dir.
      - Magic Byte Verification: first 5 bytes must equal b"%PDF-".
      - Streaming Byte Cap: aborts if cumulative bytes exceed MAX_PDF_BYTES (100 MB).

    Args:
        candidate: CandidatePaper instance.
        staging_dir: Target batch directory Path.
        client: Optional existing httpx.AsyncClient for connection reuse.

    Returns:
        Absolute file path string on success, or None on failure/non-PDF.
    """
    urls_to_try: list[str] = []
    if candidate.oa_pdf_url and isinstance(candidate.oa_pdf_url, str):
        urls_to_try.append(candidate.oa_pdf_url.strip())
    for alt in getattr(candidate, "oa_pdf_urls", []):
        if alt and isinstance(alt, str):
            clean_alt = alt.strip()
            if clean_alt and clean_alt not in urls_to_try:
                urls_to_try.append(clean_alt)

    if not urls_to_try:
        return None

    raw_id = candidate.candidate_id.strip() if candidate.candidate_id else "candidate"
    safe_id = re.sub(r"[^a-zA-Z0-9_\-]", "_", raw_id)
    if not safe_id:
        safe_id = "candidate"
    id_hash = hashlib.sha1(raw_id.encode("utf-8")).hexdigest()[:8]
    filename = f"{safe_id}_{id_hash}.pdf"

    resolved_staging_dir = Path(staging_dir).resolve()
    dest_path = (resolved_staging_dir / filename).resolve()

    try:
        if not dest_path.is_relative_to(resolved_staging_dir) or dest_path == resolved_staging_dir:
            logger.warning("Path traversal attempt in filename %r", filename)
            return None
    except AttributeError:
        if resolved_staging_dir not in dest_path.parents:
            return None

    resolved_staging_dir.mkdir(parents=True, exist_ok=True)

    async def _try_download(c: httpx.AsyncClient) -> str | None:
        """Attempt downloading candidate PDF across candidate URLs sequentially.

        Args:
            c: Asynchronous HTTP client instance to use for downloads.

        Returns:
            Absolute file path string if downloaded successfully, or None.
        """
        for raw_url in urls_to_try:
            url = raw_url
            if url.lower().startswith("http://"):
                url = "https://" + url[7:]
            elif not url.lower().startswith("https://"):
                logger.warning(
                    "SSRF Protection: Rejected non-https URL %r for candidate %s",
                    url,
                    candidate.candidate_id,
                )
                continue

            res = await _stream_to_file(c, url, dest_path)
            if res and Path(res).exists() and Path(res).stat().st_size > 0:
                return res
        return None

    if client is not None:
        return await _try_download(client)
    else:
        async with httpx.AsyncClient(timeout=DEFAULT_DOWNLOAD_TIMEOUT, follow_redirects=True) as own_client:
            return await _try_download(own_client)

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,
    )

get_staging_root

get_staging_root()

Return the root directory for temporary discovery batch staging.

Prefers /staging/ (which is within user home, accessible to sandboxed Flatpak Zotero instances having filesystem=home permissions), falling back to ~/.zocrates/staging, and finally system temp.

Source code in zocrates/discovery/staging.py

def get_staging_root() -> Path:
    """Return the root directory for temporary discovery batch staging.

    Prefers <zocrates-data>/staging/ (which is within user home, accessible to sandboxed
    Flatpak Zotero instances having filesystem=home permissions), falling back to
    ~/.zocrates/staging, and finally system temp.
    """
    try:
        from zocrates.config import get_zocrates_data_dir

        root = get_zocrates_data_dir() / "staging"
        root.mkdir(parents=True, exist_ok=True)
        return root
    except Exception:
        pass

    try:
        root = Path.home() / ".zocrates" / "staging"
        root.mkdir(parents=True, exist_ok=True)
        return root
    except Exception:
        pass

    root = Path(tempfile.gettempdir()) / "zocrates_staging"
    root.mkdir(parents=True, exist_ok=True)
    return root

normalize_arxiv_id

normalize_arxiv_id(arxiv_id)

Normalize an arXiv identifier by stripping URL prefixes, 'arxiv:' tags, and version suffixes.

Source code in zocrates/rag/zotero_db.py

def normalize_arxiv_id(arxiv_id: str | None) -> str | None:
    """Normalize an arXiv identifier by stripping URL prefixes, 'arxiv:' tags, and version suffixes."""
    if not arxiv_id:
        return None
    s = arxiv_id.strip()
    url_m = re.search(
        r"(?i)arxiv\.org\/(?:abs|pdf)\/([a-zA-Z\-]+(?:\.[a-zA-Z\-]+)?\/\d{7}|\d{4}\.\d{4,5}(?:v\d+)?)",
        s,
    )
    if url_m:
        s = url_m.group(1)
    else:
        doi_m = re.search(
            r"(?i)10\.48550\/arxiv\.([a-zA-Z\-]+(?:\.[a-zA-Z\-]+)?\/\d{7}|\d{4}\.\d{4,5}(?:v\d+)?)",
            s,
        )
        if doi_m:
            s = doi_m.group(1)
        else:
            if s.lower().startswith("arxiv:"):
                s = s[6:].strip()

    s = re.sub(r"(?i)v\d+$", "", s).strip()
    s = re.sub(r"(?i)\.pdf$", "", s).strip().rstrip(".")
    s = s.lower()
    if ARXIV_ID_PATTERN.match(s):
        return s
    return None

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

normalize_title

normalize_title(title)

Normalize paper title for fuzzy comparison by lowercasing, stripping punctuation, and normalizing whitespace.

Parameters:

Name	Type	Description	Default
title	str | None	Raw title string.	required

Returns:

Type	Description
str	Clean, lowercase, punctuation-free string.

Source code in zocrates/discovery/dedup.py

def normalize_title(title: str | None) -> str:
    """
    Normalize paper title for fuzzy comparison by lowercasing,
    stripping punctuation, and normalizing whitespace.

    Args:
        title: Raw title string.

    Returns:
        Clean, lowercase, punctuation-free string.
    """
    if not title:
        return ""
    # Lowercase
    s = title.lower()
    # Replace punctuation symbols with whitespace
    s = re.sub(r"[^\w\s]", " ", s)
    # Collapse multiple whitespaces
    s = re.sub(r"\s+", " ", s).strip()
    return s