abstract_fallback

zocrates.discovery.abstract_fallback ¶

Opportunistic 3-tier abstract backfill cascade for scholarly candidates.

DEFAULT_POLITE_EMAIL `module-attribute` ¶

DEFAULT_POLITE_EMAIL = 'zocrates-dev@proton.me'

NCBI_EFETCH_URL `module-attribute` ¶

NCBI_EFETCH_URL = 'https://eutils.ncbi.nlm.nih.gov/entrez/eutils/efetch.fcgi'

NCBI_ESEARCH_URL `module-attribute` ¶

NCBI_ESEARCH_URL = 'https://eutils.ncbi.nlm.nih.gov/entrez/eutils/esearch.fcgi'

SEMANTIC_SCHOLAR_URL `module-attribute` ¶

SEMANTIC_SCHOLAR_URL = 'https://api.semanticscholar.org/graph/v1/paper'

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

make_user_agent ¶

make_user_agent(email=DEFAULT_POLITE_EMAIL)

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

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

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

normalize_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

abstract_fallback

zocrates.discovery.abstract_fallback ¶

DEFAULT_POLITE_EMAIL module-attribute ¶

NCBI_EFETCH_URL module-attribute ¶

NCBI_ESEARCH_URL module-attribute ¶

SEMANTIC_SCHOLAR_URL module-attribute ¶

logger module-attribute ¶

AbstractFallback ¶

backfill_abstract async ¶

backfill_batch async ¶

CandidatePaper dataclass ¶

from_dict classmethod ¶

to_dict ¶

make_user_agent ¶

normalize_arxiv_id ¶

normalize_doi ¶

DEFAULT_POLITE_EMAIL `module-attribute` ¶

NCBI_EFETCH_URL `module-attribute` ¶

NCBI_ESEARCH_URL `module-attribute` ¶

SEMANTIC_SCHOLAR_URL `module-attribute` ¶

logger `module-attribute` ¶

backfill_abstract `async` ¶

backfill_batch `async` ¶

CandidatePaper `dataclass` ¶

from_dict `classmethod` ¶