dedup

zocrates.discovery.dedup

In-memory deduplication and novelty filtering against local Zotero library and search batches.

logger module-attribute

logger = logging.getLogger(__name__)

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

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