library_syncer

zocrates.rag.library_syncer ¶

Background ingestion worker.

Coordinates the document processing pipeline: ZoteroDB → DocumentProcessor → Chunker → VectorStore

CHUNK_MAX_CHARS `module-attribute` ¶

CHUNK_MAX_CHARS = int(os.getenv('CHUNK_MAX_CHARS', '4000'))

CHUNK_OVERLAP_CHARS `module-attribute` ¶

CHUNK_OVERLAP_CHARS = int(os.getenv('CHUNK_OVERLAP_CHARS', '200'))

DEFAULT_SYNC_AUTO_ABORT_FAILURES `module-attribute` ¶

DEFAULT_SYNC_AUTO_ABORT_FAILURES = 10

logger `module-attribute` ¶

logger = logging.getLogger(__name__)

DocumentProcessor ¶

DocumentProcessor(cache_dir, converters=None, document_converter_id=None, datalab_non_pdf_fallback='markitdown')

Converts attachments (PDF, HTML, etc.) to Markdown with local caching and image extraction. Manages cache persistence and delegates conversion to registered BaseDocumentConverters.

Initialize the DocumentProcessor.

Parameters:

Name	Type	Description	Default
`cache_dir`	`Path`	Directory used to store converted Markdown files and the `conversion_meta.db` metadata database. Created if it does not exist.	required
`converters`	`list[BaseDocumentConverter] \| None`	Optional list of document converters. If None, defaults to [DirectReadConverter(), MarkItDownConverter(), DoclingLocalConverter(), DatalabConverter()].	`None`
`document_converter_id`	`str \| None`	Global active converter ID from config. When set, `get_or_convert` prefers this converter on cache miss before falling back to the priority chain. If `None`, the priority chain is used directly.	`None`
`datalab_non_pdf_fallback`	`str`	Fallback converter ID when active converter is 'datalab' and content_type is not PDF ('markitdown' or 'docling_local').	`'markitdown'`

Source code in zocrates/rag/document_processor.py

def __init__(
        self,
        cache_dir: Path,
        converters: list[BaseDocumentConverter] | None = None,
        document_converter_id: str | None = None,
        datalab_non_pdf_fallback: str = "markitdown",
) -> None:
    """Initialize the DocumentProcessor.

    Args:
        cache_dir: Directory used to store converted Markdown files and the
            ``conversion_meta.db`` metadata database. Created if it does
            not exist.
        converters: Optional list of document converters. If None, defaults to
            [DirectReadConverter(), MarkItDownConverter(), DoclingLocalConverter(), DatalabConverter()].
        document_converter_id: Global active converter ID from config.
            When set, ``get_or_convert`` prefers this converter on cache miss
            before falling back to the priority chain. If ``None``, the
            priority chain is used directly.
        datalab_non_pdf_fallback: Fallback converter ID when active converter
            is 'datalab' and content_type is not PDF ('markitdown' or 'docling_local').
    """
    self._cache_dir = cache_dir
    self._cache_dir.mkdir(parents=True, exist_ok=True)
    self._document_converter_id = document_converter_id
    self._datalab_non_pdf_fallback = datalab_non_pdf_fallback
    if converters is None:
        self._converters: list[BaseDocumentConverter] = [
            DirectReadConverter(),
            MarkItDownConverter(),
            DoclingLocalConverter(),
            DatalabConverter(),
        ]
    else:
        self._converters = list(converters)

cache_dir `property` ¶

cache_dir

Return the root markdown cache directory Path.

datalab_non_pdf_fallback `property` `writable` ¶

datalab_non_pdf_fallback

Return the fallback converter ID for non-PDF items when Datalab is active.

clear_failure ¶

clear_failure(item_key, converter_id=None)

Clear failure record(s) for an item and optionally a specific converter.

Parameters:

Name	Type	Description	Default
`item_key`	`str`	Zotero item key.	required
`converter_id`	`str \| None`	Specific converter ID to clear, or None to clear all.	`None`

Source code in zocrates/rag/document_processor.py

def clear_failure(self, item_key: str, converter_id: str | None = None) -> None:
    """Clear failure record(s) for an item and optionally a specific converter.

    Args:
        item_key: Zotero item key.
        converter_id: Specific converter ID to clear, or None to clear all.
    """
    with _open_meta_db(self._cache_dir) as conn:
        if converter_id is not None:
            conn.execute(
                "DELETE FROM conversion_failures WHERE item_key = ? AND converter_id = ?",
                (item_key, converter_id),
            )
        else:
            conn.execute(
                "DELETE FROM conversion_failures WHERE item_key = ?",
                (item_key,),
            )
        conn.commit()

convert_with ¶

convert_with(item_key, converter_id, source_path, content_type, parent_item_type=None, title=None, collections=None, parent_item_key=None, authors=None, year=None, publication=None, doi=None, url=None, abstract=None, abstract_source=None, update_root_index=True)

Convert a document with a specific converter engine.

This is the explicit user-action path. No fallback -- if the specified converter doesn't support the content type, an error is raised. Any existing conversion for this item is destroyed and replaced.

The caller is responsible for re-indexing the vector store after this method returns.

Parameters:

Name	Type	Description	Default
`item_key`	`str`	Zotero item key.	required
`converter_id`	`str`	ID of the converter to use.	required
`source_path`	`Path`	Path to the source file on disk.	required
`content_type`	`str`	MIME type of the source file.	required
`parent_item_type`	`str \| None`	Zotero parent item type.	`None`
`title`	`str \| None`	Parent item title.	`None`
`collections`	`list[Any] \| None`	Parent item collections.	`None`
`parent_item_key`	`str \| None`	Parent Zotero item key string.	`None`
`authors`	`list[str] \| None`	Optional author names list.	`None`
`year`	`str \| int \| None`	Optional publication year.	`None`
`publication`	`str \| None`	Optional publication venue string.	`None`
`doi`	`str \| None`	Optional DOI string.	`None`
`url`	`str \| None`	Optional source URL string.	`None`
`abstract`	`str \| None`	Optional item abstract.	`None`
`abstract_source`	`str \| None`	Optional source of item abstract.	`None`
`update_root_index`	`bool`	If True (default), updates the root index.md.	`True`

Returns:

Type	Description
`str`	The converted Markdown text (with YAML frontmatter).

Raises:

Type	Description
`UnsupportedConversionError`	If the converter doesn't support the content type.
`KeyError`	If no converter with `converter_id` is registered.

Source code in zocrates/rag/document_processor.py

def convert_with(
        self,
        item_key: str,
        converter_id: str,
        source_path: Path,
        content_type: str,
        parent_item_type: str | None = None,
        title: str | None = None,
        collections: list[Any] | None = None,
        parent_item_key: str | None = None,
        authors: list[str] | None = None,
        year: str | int | None = None,
        publication: str | None = None,
        doi: str | None = None,
        url: str | None = None,
        abstract: str | None = None,
        abstract_source: str | None = None,
        update_root_index: bool = True,
) -> str:
    """Convert a document with a specific converter engine.

    This is the explicit user-action path. No fallback -- if the
    specified converter doesn't support the content type, an error is
    raised. Any existing conversion for this item is destroyed and
    replaced.

    The caller is responsible for re-indexing the vector store after
    this method returns.

    Args:
        item_key: Zotero item key.
        converter_id: ID of the converter to use.
        source_path: Path to the source file on disk.
        content_type: MIME type of the source file.
        parent_item_type: Zotero parent item type.
        title: Parent item title.
        collections: Parent item collections.
        parent_item_key: Parent Zotero item key string.
        authors: Optional author names list.
        year: Optional publication year.
        publication: Optional publication venue string.
        doi: Optional DOI string.
        url: Optional source URL string.
        abstract: Optional item abstract.
        abstract_source: Optional source of item abstract.
        update_root_index: If True (default), updates the root index.md.

    Returns:
        The converted Markdown text (with YAML frontmatter).

    Raises:
        UnsupportedConversionError: If the converter doesn't support
            the content type.
        KeyError: If no converter with ``converter_id`` is registered.
    """
    converter = self._get_converter_by_id(converter_id)
    if not converter.supports(content_type, source_path):
        # If requested converter is datalab (which is PDF-only), route non-PDFs to configured fallback
        fallback_conv = None
        if converter_id == "datalab":
            fallback_target = self._datalab_non_pdf_fallback or "markitdown"
            try:
                target_c = self._get_converter_by_id(fallback_target)
                if target_c.supports(content_type, source_path):
                    fallback_conv = target_c
                    logger.info(
                        "Converter '%s' does not support '%s'; routing to configured fallback '%s' for %s",
                        converter_id, content_type, target_c.converter_id, source_path.name,
                    )
            except KeyError:
                pass

            if fallback_conv is None:
                fallback_conv = self._resolve_converter(content_type, source_path)

        if fallback_conv is not None and fallback_conv.supports(content_type, source_path):
            converter = fallback_conv
            converter_id = fallback_conv.converter_id
        else:
            raise UnsupportedConversionError(
                f"Converter '{converter_id}' does not support "
                f"content type '{content_type}'"
            )

    # Convert into a temporary directory first so the existing bundle
    # is preserved if conversion fails.
    tmp_dir = self._cache_dir / f".tmp_{item_key}"
    if tmp_dir.exists():
        shutil.rmtree(tmp_dir)
    tmp_dir.mkdir(parents=True)

    current_hash = _file_sha256(source_path)

    try:
        markdown = converter.convert(
            item_key, source_path, content_type, tmp_dir
        )
    except Exception as exc:
        shutil.rmtree(tmp_dir, ignore_errors=True)
        self.record_failure(item_key, converter_id, str(exc), current_hash)
        raise RuntimeError(
            f"Converter '{converter_id}' failed for {source_path}: {exc}"
        ) from exc

    if markdown is None:
        shutil.rmtree(tmp_dir, ignore_errors=True)
        err_msg = f"Converter '{converter_id}' returned empty content for {source_path}"
        self.record_failure(item_key, converter_id, err_msg, current_hash)
        raise RuntimeError(err_msg)

    # Conversion succeeded -- destroy old bundle and swap in the new one
    item_dir = self._cache_dir / item_key
    if item_dir.exists():
        shutil.rmtree(item_dir)
    tmp_dir.rename(item_dir)

    # Replace any temporary directory references in the converted markdown
    markdown = _relativize_staging_markdown_links(markdown, tmp_dir)

    self.clear_failure(item_key, converter_id)
    if converter.converter_id != converter_id:
        self.clear_failure(item_key, converter.converter_id)

    self._save_to_cache(
        item_key, source_path, content_type, current_hash, markdown,
        converter.converter_id, converter.converter_version,
        parent_item_type=parent_item_type, title=title, collections=collections,
        parent_item_key=parent_item_key,
        authors=authors, year=year, publication=publication, doi=doi, url=url,
        abstract=abstract, abstract_source=abstract_source,
        update_root_index=update_root_index,
    )
    return self.read_cached(item_key) or markdown

delete_cached ¶

delete_cached(item_key, update_root_index=True)

Delete an item bundle from disk and conversion_meta DB.

Parameters:

Name	Type	Description	Default
`item_key`	`str`	Zotero item key to remove.	required
`update_root_index`	`bool`	If True (default), updates the root index.md catalog.	`True`

Returns:

Type	Description
`bool`	True if an item bundle or DB row was deleted, False if item was not cached.

Source code in zocrates/rag/document_processor.py

def delete_cached(self, item_key: str, update_root_index: bool = True) -> bool:
    """Delete an item bundle from disk and conversion_meta DB.

    Args:
        item_key: Zotero item key to remove.
        update_root_index: If True (default), updates the root index.md catalog.

    Returns:
        True if an item bundle or DB row was deleted, False if item was not cached.
    """
    deleted = False
    item_dir = self._cache_dir / item_key
    if item_dir.exists():
        shutil.rmtree(item_dir, ignore_errors=True)
        deleted = True

    affected_neighbors: list[str] = []
    with _open_meta_db(self._cache_dir) as conn:
        try:
            cur = conn.execute(
                """
                SELECT DISTINCT target_key FROM item_relations WHERE source_key = ?
                UNION
                SELECT DISTINCT source_key FROM item_relations WHERE target_key = ?
                """,
                (item_key, item_key),
            )
            affected_neighbors = [row[0] for row in cur.fetchall() if row[0] != item_key]
            conn.execute(
                "DELETE FROM item_relations WHERE source_key = ? OR target_key = ?",
                (item_key, item_key),
            )
        except sqlite3.OperationalError:
            pass

        cur = conn.execute(
            "DELETE FROM conversion_meta WHERE item_key = ?", (item_key,)
        )
        if cur.rowcount > 0:
            deleted = True
        conn.execute(
            "DELETE FROM conversion_failures WHERE item_key = ?", (item_key,)
        )
        conn.commit()

    # Regenerate item_relations.md for all affected neighbors
    for neighbor_key in affected_neighbors:
        try:
            regenerate_item_relations_md(neighbor_key, self._cache_dir)
        except Exception as exc:
            logger.warning(
                "Failed to regenerate item_relations.md for neighbor %s during deletion of %s: %s",
                neighbor_key, item_key, exc,
            )

    if deleted and update_root_index:
        generate_root_index_md(self._cache_dir)

    return deleted

ensure_parent_item_key ¶

ensure_parent_item_key(item_key, parent_item_key)

Ensure an item in conversion_meta has parent_item_key populated.

Source code in zocrates/rag/document_processor.py

def ensure_parent_item_key(self, item_key: str, parent_item_key: str) -> None:
    """Ensure an item in conversion_meta has parent_item_key populated."""
    if not item_key or not parent_item_key:
        return
    with _open_meta_db(self._cache_dir) as conn:
        conn.execute(
            """
            UPDATE conversion_meta
            SET parent_item_key = ?
            WHERE item_key = ? AND (parent_item_key IS NULL OR parent_item_key = '')
            """,
            (parent_item_key, item_key),
        )
        conn.commit()

get_all_meta_keys ¶

get_all_meta_keys()

Return the set of all unique item keys tracked in conversion_meta.

Source code in zocrates/rag/document_processor.py

def get_all_meta_keys(self) -> set[str]:
    """Return the set of all unique item keys tracked in conversion_meta."""
    try:
        with _open_meta_db(self._cache_dir) as conn:
            rows = conn.execute("SELECT DISTINCT item_key FROM conversion_meta").fetchall()
            return {row["item_key"] for row in rows if row["item_key"]}
    except Exception as exc:
        logger.error("Failed to query item keys from conversion_meta: %s", exc)
        return set()

get_all_metadata ¶

get_all_metadata()

Return metadata for all cached documents in a single fast query.

Returns:

Type	Description
`dict[str, dict[str, Any]]`	Dictionary mapping `item_key` to metadata dictionary.

Source code in zocrates/rag/document_processor.py

def get_all_metadata(self) -> dict[str, dict[str, Any]]:
    """Return metadata for all cached documents in a single fast query.

    Returns:
        Dictionary mapping ``item_key`` to metadata dictionary.
    """
    with _open_meta_db(self._cache_dir) as conn:
        rows = conn.execute("SELECT * FROM conversion_meta").fetchall()
        return {
            row["item_key"]: {
                "item_key": row["item_key"],
                "title": row["title"] if "title" in row.keys() else None,
                "converter_id": row["converter_id"],
                "converter_version": row["converter_version"],
                "source_path": row["source_path"],
                "content_type": row["content_type"],
                "parent_item_type": row["parent_item_type"],
                "converted_at": row["converted_at"],
                "indexed_at": row["indexed_at"],
                "source_hash": row["source_hash"],
                "char_count": row["char_count"],
                "parent_item_key": row["parent_item_key"],
                "parse_quality_score": row["parse_quality_score"],
                "checkpoint_id": row["checkpoint_id"],
                "abstract": row["abstract"] if "abstract" in row.keys() else None,
                "abstract_source": row["abstract_source"] if "abstract_source" in row.keys() else None,
            }
            for row in rows
        }

get_failures ¶

get_failures(item_key)

Retrieve recorded conversion failures for an item key.

Parameters:

Name	Type	Description	Default
`item_key`	`str`	Zotero item key.	required

Returns:

Type	Description
`dict[str, dict[str, Any]]`	Dict mapping converter_id -> failure details dict.

Source code in zocrates/rag/document_processor.py

def get_failures(self, item_key: str) -> dict[str, dict[str, Any]]:
    """Retrieve recorded conversion failures for an item key.

    Args:
        item_key: Zotero item key.

    Returns:
        Dict mapping converter_id -> failure details dict.
    """
    with _open_meta_db(self._cache_dir) as conn:
        rows = conn.execute(
            """
            SELECT converter_id, error_message, source_hash, failed_at
            FROM conversion_failures WHERE item_key = ?
            """,
            (item_key,),
        ).fetchall()
    return {
        row["converter_id"]: {
            "error_message": row["error_message"],
            "source_hash": row["source_hash"],
            "failed_at": row["failed_at"],
        }
        for row in rows
    }

get_metadata ¶

get_metadata(item_key)

Fetch metadata dictionary for an item key from conversion_meta.db.

Source code in zocrates/rag/document_processor.py

def get_metadata(self, item_key: str) -> dict | None:
    """Fetch metadata dictionary for an item key from conversion_meta.db."""
    with _open_meta_db(self._cache_dir) as conn:
        row = conn.execute(
            """
            SELECT item_key, title, converter_id, converter_version, source_path, content_type,
                   parent_item_type, converted_at, indexed_at, source_hash, char_count,
                   parent_item_key, parse_quality_score, checkpoint_id, abstract, abstract_source
            FROM conversion_meta WHERE item_key = ?
            """,
            (item_key,),
        ).fetchone()
    if not row:
        return None
    return {
        "item_key": row["item_key"],
        "title": row["title"],
        "converter_id": row["converter_id"],
        "converter_version": row["converter_version"],
        "source_path": row["source_path"],
        "content_type": row["content_type"],
        "parent_item_type": row["parent_item_type"],
        "converted_at": row["converted_at"],
        "indexed_at": row["indexed_at"],
        "source_hash": row["source_hash"],
        "char_count": row["char_count"],
        "parent_item_key": row["parent_item_key"],
        "parse_quality_score": row["parse_quality_score"],
        "checkpoint_id": row["checkpoint_id"],
        "abstract": row["abstract"] if "abstract" in row.keys() else None,
        "abstract_source": row["abstract_source"] if "abstract_source" in row.keys() else None,
    }

get_or_convert ¶

get_or_convert(item_key, source_path, content_type, parent_item_type=None, title=None, collections=None, parent_item_key=None, authors=None, year=None, publication=None, doi=None, url=None, abstract=None, abstract_source=None, update_root_index=True)

Return the Markdown for an attachment, converting if not already cached or if the source file has changed since the last conversion.

Parameters:

Name	Type	Description	Default
`item_key`	`str`	Zotero item key.	required
`source_path`	`Path`	Path to the source file on disk.	required
`content_type`	`str`	MIME type of the source file.	required
`parent_item_type`	`str \| None`	Zotero parent item type (e.g. 'journalArticle').	`None`
`title`	`str \| None`	Parent item title.	`None`
`collections`	`list[Any] \| None`	Parent item collections list.	`None`
`parent_item_key`	`str \| None`	Parent Zotero item key string.	`None`
`authors`	`list[str] \| None`	Optional author names list.	`None`
`year`	`str \| int \| None`	Optional publication year.	`None`
`publication`	`str \| None`	Optional publication venue string.	`None`
`doi`	`str \| None`	Optional DOI string.	`None`
`url`	`str \| None`	Optional source URL string.	`None`
`abstract`	`str \| None`	Optional abstract or summary text.	`None`
`abstract_source`	`str \| None`	Optional source of abstract ('zotero', 'generated', 'extractive').	`None`
`update_root_index`	`bool`	If True (default), updates the root index.md catalog.	`True`

Returns:

Type	Description
`str \| None`	The converted Markdown text with YAML frontmatter, or None on error.

Source code in zocrates/rag/document_processor.py

def get_or_convert(
        self,
        item_key: str,
        source_path: Path,
        content_type: str,
        parent_item_type: str | None = None,
        title: str | None = None,
        collections: list[Any] | None = None,
        parent_item_key: str | None = None,
        authors: list[str] | None = None,
        year: str | int | None = None,
        publication: str | None = None,
        doi: str | None = None,
        url: str | None = None,
        abstract: str | None = None,
        abstract_source: str | None = None,
        update_root_index: bool = True,
) -> str | None:
    """Return the Markdown for an attachment, converting if not already cached
    or if the source file has changed since the last conversion.

    Args:
        item_key: Zotero item key.
        source_path: Path to the source file on disk.
        content_type: MIME type of the source file.
        parent_item_type: Zotero parent item type (e.g. 'journalArticle').
        title: Parent item title.
        collections: Parent item collections list.
        parent_item_key: Parent Zotero item key string.
        authors: Optional author names list.
        year: Optional publication year.
        publication: Optional publication venue string.
        doi: Optional DOI string.
        url: Optional source URL string.
        abstract: Optional abstract or summary text.
        abstract_source: Optional source of abstract ('zotero', 'generated', 'extractive').
        update_root_index: If True (default), updates the root index.md catalog.

    Returns:
        The converted Markdown text with YAML frontmatter, or None on error.
    """
    if not source_path.exists():
        logger.warning("Attachment not found on disk: %s", source_path)
        return None

    current_hash = _file_sha256(source_path)
    cached_md = self._load_from_cache(item_key, current_hash)
    if cached_md is not None:
        return cached_md

    # Convert using active converter or priority chain fallback
    converter = self._resolve_converter(content_type, source_path)
    if converter is None:
        logger.warning("No suitable converter found for %s", source_path)
        return None

    markdown = self._run_converter(converter, item_key, source_path, content_type)
    if markdown is None:
        return None

    full_md = self._save_to_cache(
        item_key, source_path, content_type, current_hash, markdown,
        converter.converter_id, converter.converter_version,
        parent_item_type=parent_item_type, title=title, collections=collections,
        parent_item_key=parent_item_key,
        authors=authors, year=year, publication=publication, doi=doi, url=url,
        abstract=abstract, abstract_source=abstract_source,
        update_root_index=update_root_index,
    )
    return full_md

is_cached ¶

is_cached(item_key)

Return True if a converted Markdown file exists for item_key.

Does not validate freshness against the source file hash.

Parameters:

Name	Type	Description	Default
`item_key`	`str`	Zotero item key.	required

Returns:

Type	Description
`bool`	`True` if `<item_key>/document.md` exists, `False` otherwise.

Source code in zocrates/rag/document_processor.py

def is_cached(self, item_key: str) -> bool:
    """Return True if a converted Markdown file exists for item_key.

    Does not validate freshness against the source file hash.

    Args:
        item_key: Zotero item key.

    Returns:
        ``True`` if ``<item_key>/document.md`` exists, ``False`` otherwise.
    """
    md_path = self._cache_dir / item_key / "document.md"
    return md_path.exists()

list_cached_keys ¶

list_cached_keys()

Return the item keys of all documents currently in the Markdown cache.

Returns:

Type	Description
`list[str]`	List of item key strings (subdirectories containing `document.md`).

Source code in zocrates/rag/document_processor.py

def list_cached_keys(self) -> list[str]:
    """Return the item keys of all documents currently in the Markdown cache.

    Returns:
        List of item key strings (subdirectories containing ``document.md``).
    """
    return [
        p.parent.name
        for p in self._cache_dir.glob("*/document.md")
    ]

read_cached ¶

read_cached(item_key)

Read cached Markdown without triggering conversion.

Source code in zocrates/rag/document_processor.py

def read_cached(self, item_key: str) -> str | None:
    """Read cached Markdown without triggering conversion."""
    md_path = self._cache_dir / item_key / "document.md"
    if md_path.exists():
        return md_path.read_text(encoding="utf-8")
    return None

record_failure ¶

record_failure(item_key, converter_id, error_message, source_hash)

Record or update a failed conversion attempt for an item.

Parameters:

Name	Type	Description	Default
`item_key`	`str`	Zotero item key.	required
`converter_id`	`str`	Converter engine ID that failed.	required
`error_message`	`str`	Error description or exception string.	required
`source_hash`	`str`	SHA-256 hex digest of the source file at failure time.	required

Source code in zocrates/rag/document_processor.py

def record_failure(
    self, item_key: str, converter_id: str, error_message: str, source_hash: str
) -> None:
    """Record or update a failed conversion attempt for an item.

    Args:
        item_key: Zotero item key.
        converter_id: Converter engine ID that failed.
        error_message: Error description or exception string.
        source_hash: SHA-256 hex digest of the source file at failure time.
    """
    failed_at = datetime.now(UTC).isoformat()
    with _open_meta_db(self._cache_dir) as conn:
        conn.execute(
            """
            INSERT OR REPLACE INTO conversion_failures
                (item_key, converter_id, error_message, source_hash, failed_at)
            VALUES (?, ?, ?, ?, ?)
            """,
            (item_key, converter_id, error_message, source_hash, failed_at),
        )
        conn.commit()

regenerate_root_index ¶

regenerate_root_index()

Regenerate the root-level index.md catalog from conversion_meta.db.

Source code in zocrates/rag/document_processor.py

def regenerate_root_index(self) -> None:
    """Regenerate the root-level index.md catalog from conversion_meta.db."""
    generate_root_index_md(self._cache_dir)

register_converter ¶

register_converter(converter, prepend=True)

Register a new document converter strategy.

Parameters:

Name	Type	Description	Default
`converter`	`BaseDocumentConverter`	Instance of BaseDocumentConverter to register.	required
`prepend`	`bool`	If True (default), inserts the converter at the highest priority.	`True`

Source code in zocrates/rag/document_processor.py

def register_converter(
        self, converter: BaseDocumentConverter, prepend: bool = True
) -> None:
    """Register a new document converter strategy.

    Args:
        converter: Instance of BaseDocumentConverter to register.
        prepend: If True (default), inserts the converter at the highest priority.
    """
    if prepend:
        self._converters.insert(0, converter)
    else:
        self._converters.append(converter)

update_abstract ¶

update_abstract(item_key, abstract, abstract_source='zotero')

Update abstract and abstract_source in conversion_meta for an item.

Source code in zocrates/rag/document_processor.py

def update_abstract(
    self,
    item_key: str,
    abstract: str | None,
    abstract_source: str | None = "zotero",
) -> None:
    """Update abstract and abstract_source in conversion_meta for an item."""
    if not item_key:
        return
    with _open_meta_db(self._cache_dir) as conn:
        conn.execute(
            """
            UPDATE conversion_meta
            SET abstract = ?, abstract_source = ?
            WHERE item_key = ?
            """,
            (abstract, abstract_source, item_key),
        )
        conn.commit()

update_indexed_at ¶

update_indexed_at(item_key, timestamp=None)

Update the indexed_at timestamp for item_key in conversion_meta DB.

Parameters:

Name	Type	Description	Default
`item_key`	`str`	Zotero item key.	required
`timestamp`	`str \| None`	Optional ISO-8601 timestamp string. Defaults to current UTC time.	`None`

Source code in zocrates/rag/document_processor.py

def update_indexed_at(self, item_key: str, timestamp: str | None = None) -> None:
    """Update the indexed_at timestamp for item_key in conversion_meta DB.

    Args:
        item_key: Zotero item key.
        timestamp: Optional ISO-8601 timestamp string. Defaults to current UTC time.
    """
    ts = timestamp or datetime.now(UTC).isoformat()
    with _open_meta_db(self._cache_dir) as conn:
        conn.execute(
            "UPDATE conversion_meta SET indexed_at = ? WHERE item_key = ?",
            (ts, item_key),
        )
        conn.commit()

LibrarySyncer ¶

LibrarySyncer(zotero_data_dir, sqlite_path)

Coordinates RAG indexing pipeline.

Initialize the LibrarySyncer with all RAG pipeline components.

Parameters:

Name	Type	Description	Default
`zotero_data_dir`	`Path`	Root Zotero data directory containing `zotero.sqlite` and the `storage/` subdirectory.	required
`sqlite_path`	`Path`	Absolute path to `zotero.sqlite`.	required

Source code in zocrates/rag/library_syncer.py

def __init__(self, zotero_data_dir: Path, sqlite_path: Path) -> None:
    """Initialize the LibrarySyncer with all RAG pipeline components.

    Args:
        zotero_data_dir: Root Zotero data directory containing
            ``zotero.sqlite`` and the ``storage/`` subdirectory.
        sqlite_path: Absolute path to ``zotero.sqlite``.
    """
    self._zotero_data_dir = zotero_data_dir
    storage_dir = zotero_data_dir / "storage"

    self.db = ZoteroDB(sqlite_path, storage_dir)

    md_cache = get_corpus_dir(zotero_data_dir)
    active_converter = get_document_converter_id(zotero_data_dir)
    datalab_fallback = get_datalab_non_pdf_fallback(zotero_data_dir)
    self.processor = DocumentProcessor(
        md_cache,
        document_converter_id=active_converter,
        datalab_non_pdf_fallback=datalab_fallback,
    )

    embedder = get_embedder()
    self.store = VectorStore(
        chroma_dir=get_chroma_dir(zotero_data_dir),
        fts_db_path=get_fts_db_path(zotero_data_dir),
        embedder=embedder,
    )

    self._running = False
    self.active_context_keys = None

close ¶

close()

Tear down resources held by this syncer instance.

Source code in zocrates/rag/library_syncer.py

def close(self) -> None:
    """Tear down resources held by this syncer instance."""
    logger.info("Closing LibrarySyncer and releasing resources.")
    try:
        if hasattr(self, 'store') and hasattr(self.store, 'close'):
            self.store.close()
        elif hasattr(self, 'store') and hasattr(self.store, '_client'):
            # PersistentClient doesn't require explicit close, but we clear references
            del self.store._client
    except Exception as exc:
        logger.warning("Error during LibrarySyncer teardown: %s", exc)

full_sync `async` ¶

full_sync(progress_callback=None, cancel_check=None)

Scan every item, note, and standalone attachment in the library and ingest those not yet indexed.

Additive Phase 7 onboarding parameters (backward compatible):

Parameters:

Name	Type	Description	Default
`progress_callback`	`Callable[[dict], None] \| None`	Optional synchronous callback invoked at each item boundary with a progress dict (`stage`, `processed`, `skipped`, `errors`, `total`, `current_title`, `elapsed_ms`). Callback failures are logged and never abort the sync (fail-open).	`None`
`cancel_check`	`Callable[[], bool] \| None`	Optional cooperative-cancel predicate polled at the existing per-item yield points. When it returns True the sync finishes the current item and stops, skipping the tail graph-relations/orphan-purge phases. Resume semantics are untouched (`store.is_ingested` keying + SHA-256 conversion cache).	`None`

Source code in zocrates/rag/library_syncer.py

async def full_sync(
    self,
    progress_callback: Callable[[dict], None] | None = None,
    cancel_check: Callable[[], bool] | None = None,
) -> dict:
    """Scan every item, note, and standalone attachment in the library and ingest those not yet indexed.

    Additive Phase 7 onboarding parameters (backward compatible):

    Args:
        progress_callback: Optional synchronous callback invoked at each
            item boundary with a progress dict (``stage``, ``processed``,
            ``skipped``, ``errors``, ``total``, ``current_title``,
            ``elapsed_ms``). Callback failures are logged and never abort
            the sync (fail-open).
        cancel_check: Optional cooperative-cancel predicate polled at the
            existing per-item yield points. When it returns True the sync
            finishes the current item and stops, skipping the tail
            graph-relations/orphan-purge phases. Resume semantics are
            untouched (``store.is_ingested`` keying + SHA-256 conversion
            cache).
    """
    sync_started = time.perf_counter()

    def _report_progress(stage: str, current_title: str = "") -> None:
        """Invoke the progress callback with current sync progress metrics.

        Args:
            stage: Current stage of synchronization (e.g. 'syncing', 'indexing').
            current_title: Title of the document currently being processed.
        """
        if progress_callback is None:
            return
        try:
            progress_callback({
                "stage": stage,
                "processed": processed,
                "skipped": skipped,
                "errors": errors,
                "total": total,
                "current_title": current_title,
                "elapsed_ms": int((time.perf_counter() - sync_started) * 1000),
            })
        except Exception as exc:
            logger.warning("Sync progress callback failed: %s", exc)

    def _cancel_requested() -> bool:
        """Check if cooperative cancellation was requested for the sync job.

        Returns:
            True if cancellation was requested, False otherwise.
        """
        if cancel_check is None:
            return False
        try:
            return bool(cancel_check())
        except Exception as exc:
            logger.warning("Sync cancel_check failed: %s", exc)
            return False

    try:
        auto_abort_threshold = int(get_sync_auto_abort_failures(self._zotero_data_dir))
    except (ValueError, TypeError):
        auto_abort_threshold = DEFAULT_SYNC_AUTO_ABORT_FAILURES
    consecutive_failures = 0
    cancelled = False
    auto_aborted = False
    # Synchronize metadata for existing bundles modified in Zotero
    try:
        staleness_res = await asyncio.to_thread(self.sync_metadata_staleness)
        logger.info("Metadata staleness sync in full_sync complete: %s", staleness_res)
    except Exception as exc:
        logger.warning("Error during metadata staleness sync in full_sync: %s", exc)
        staleness_res = {"updated": 0}

    items = self.db.get_all_items()
    standalone_notes = self.db.get_standalone_notes()
    standalone_attachments = self.db.get_standalone_attachments()

    total = len(items) + len(standalone_notes) + len(standalone_attachments)
    processed = 0
    skipped = 0
    errors = 0
    # Emit an initial snapshot immediately so SSE subscribers get the
    # determinate total before the first (potentially multi-minute)
    # conversion finishes.
    _report_progress("starting")

    logger.info(
        "Starting full library sync: %d regular items, %d standalone notes, %d standalone attachments found.",
        len(items),
        len(standalone_notes),
        len(standalone_attachments),
    )

    # Auto-backfill missing parent_item_key values in conversion_meta.db
    all_atts = self.db.get_all_attachments()
    for att in all_atts:
        if att.parent_item_key:
            self.processor.ensure_parent_item_key(att.item_key, att.parent_item_key)

    for item in items:
        try:
            result = await self._ingest_item(
                item.item_id, item.item_key, item.title
            )
            if result == "processed":
                processed += 1
            else:
                skipped += 1
            consecutive_failures = 0
        except Exception as exc:
            logger.error(
                "Error ingesting item %s: %s", item.item_key, exc, exc_info=True
            )
            errors += 1
            consecutive_failures += 1
        await asyncio.sleep(0)
        _report_progress("items", item.title)
        if _cancel_requested():
            cancelled = True
            break
        if auto_abort_threshold > 0 and consecutive_failures >= auto_abort_threshold:
            logger.warning(
                "Auto-aborting full sync after %d consecutive failures (threshold: %d).",
                consecutive_failures,
                auto_abort_threshold,
            )
            auto_aborted = True
            break

    for note in standalone_notes:
        try:
            result = await self._ingest_standalone_note(note)
            if result == "processed":
                processed += 1
            else:
                skipped += 1
            consecutive_failures = 0
        except Exception as exc:
            logger.error(
                "Error ingesting standalone note %s: %s", note.note_key, exc, exc_info=True
            )
            errors += 1
            consecutive_failures += 1
        await asyncio.sleep(0)
        _report_progress("notes", getattr(note, "title", "") or "")
        if _cancel_requested():
            cancelled = True
            break
        if auto_abort_threshold > 0 and consecutive_failures >= auto_abort_threshold:
            logger.warning(
                "Auto-aborting full sync after %d consecutive failures (threshold: %d).",
                consecutive_failures,
                auto_abort_threshold,
            )
            auto_aborted = True
            break

    for att in standalone_attachments:
        try:
            result = await self._ingest_standalone_attachment(att)
            if result == "processed":
                processed += 1
            else:
                skipped += 1
            consecutive_failures = 0
        except Exception as exc:
            logger.error(
                "Error ingesting standalone attachment %s: %s", att.item_key, exc, exc_info=True
            )
            errors += 1
            consecutive_failures += 1
        await asyncio.sleep(0)
        _report_progress("attachments", getattr(att, "title", "") or "")
        if _cancel_requested():
            cancelled = True
            break
        if auto_abort_threshold > 0 and consecutive_failures >= auto_abort_threshold:
            logger.warning(
                "Auto-aborting full sync after %d consecutive failures (threshold: %d).",
                consecutive_failures,
                auto_abort_threshold,
            )
            auto_aborted = True
            break

    # Pass of manual relations and shared tags synchronization
    # (skipped on cooperative cancel / auto-abort — the next successful
    # /sync executes it; plan §6.3 tail-phase semantics)
    meta_db_path = self.processor._cache_dir / "conversion_meta.db"
    if meta_db_path.exists() and not cancelled and not auto_aborted:
        try:
            def _sync_graph_relations_worker() -> None:
                """Sync human-curated relations, shared tags, and regenerate bundle markdown."""
                from zocrates.rag.bundle_catalog import regenerate_item_relations_md
                from zocrates.config import get_graph_relations_config
                from zocrates.rag.document_processor import _open_meta_db
                from zocrates.rag.graph_engine import sync_manual_relations, sync_shared_tags

                cfg = get_graph_relations_config(self._zotero_data_dir)
                if not cfg.get("enabled", True):
                    return

                with _open_meta_db(self.processor._cache_dir) as meta_conn:
                    n_manual = 0
                    n_tags = 0
                    if cfg.get("include_human_curated", True):
                        n_manual = sync_manual_relations(meta_conn, self._zotero_data_dir)
                    if cfg.get("include_shared_tags", True):
                        n_tags = sync_shared_tags(meta_conn, self._zotero_data_dir)

                    if (n_manual > 0 or n_tags > 0) and cfg.get("sync_bundle_files", True):
                        rows = meta_conn.execute("SELECT item_key FROM conversion_meta").fetchall()
                        for r in rows:
                            try:
                                regenerate_item_relations_md(r["item_key"], self.processor._cache_dir)
                            except Exception as exc_inner:
                                logger.warning(
                                    "Failed to regenerate item_relations.md for %s during full_sync: %s",
                                    r["item_key"],
                                    exc_inner,
                                )

            await asyncio.to_thread(_sync_graph_relations_worker)
        except Exception as exc:
            logger.warning(
                "Error syncing manual relations or shared tags in full_sync: %s",
                exc,
            )

    # Orphan detection and cleanup (skipped on cooperative cancel /
    # auto-abort — same tail-phase semantics as above)
    orphaned_keys: set[str] = set()
    if not cancelled and not auto_aborted:
        active_keys: set[str] = set()
        for item in items:
            active_keys.add(item.item_key)
            for att in self.db.get_attachments(item.item_id):
                active_keys.add(att.item_key)
            for note in self.db.get_notes_for_item(item.item_id):
                active_keys.add(note.note_key)
        for note in standalone_notes:
            active_keys.add(note.note_key)
        for att in standalone_attachments:
            active_keys.add(att.item_key)

        all_known_keys = (
            set(self.processor.list_cached_keys())
            | self.processor.get_all_meta_keys()
            | self.store.get_all_item_keys()
        )
        if not active_keys:
            logger.warning("Skipping full sync orphan purge: no active items found in database")
        else:
            orphaned_keys = all_known_keys - active_keys

        if orphaned_keys:
            logger.info("Cleaning up %d orphaned cached items...", len(orphaned_keys))
            for o_key in orphaned_keys:
                self.purge_item(o_key, regenerate_index=False)

    summary = {
        "total": total,
        "processed": processed,
        "skipped": skipped,
        "errors": errors,
        "orphans_cleaned": len(orphaned_keys),
        "metadata_refreshed": staleness_res.get("updated", 0),
        "cancelled": cancelled,
        "auto_aborted": auto_aborted,
    }
    self.processor.regenerate_root_index()
    logger.info("Full sync complete: %s", summary)
    return summary

ingest_items `async` ¶

ingest_items(item_ids)

Force-ingest a list of item, note, or attachment IDs (e.g. newly added by Zotero).

Source code in zocrates/rag/library_syncer.py

async def ingest_items(self, item_ids: list[int]) -> dict:
    """Force-ingest a list of item, note, or attachment IDs (e.g. newly added by Zotero)."""
    items = self.db.get_items_by_ids(item_ids)
    processed = 0
    errors = 0

    for item in items:
        try:
            if item.item_type == "note":
                note = self.db.get_note_by_id(item.item_id)
                if note:
                    await self._ingest_standalone_note(note, force=True)
                    processed += 1
            elif item.item_type == "attachment":
                att = self.db.get_attachment_by_id(item.item_id)
                if att:
                    await self._ingest_standalone_attachment(att, force=True)
                    processed += 1
            else:
                await self._ingest_item(
                    item.item_id, item.item_key, item.title, force=True
                )
                processed += 1
        except Exception as exc:
            logger.error(
                "Error ingesting item %s: %s", item.item_key, exc, exc_info=True
            )
            errors += 1
        await asyncio.sleep(0)

    summary = {"requested": len(item_ids), "processed": processed, "errors": errors}
    self.processor.regenerate_root_index()
    return summary

purge_item ¶

purge_item(item_key, regenerate_index=True)

Atomically purge an item across all four datastores: disk bundle, metadata DB, vector/FTS store, and root index.

Source code in zocrates/rag/library_syncer.py

def purge_item(self, item_key: str, regenerate_index: bool = True) -> None:
    """Atomically purge an item across all four datastores: disk bundle, metadata DB, vector/FTS store, and root index."""
    if not item_key:
        return
    self.processor.delete_cached(item_key, update_root_index=False)
    self.store.delete_document(item_key)
    if regenerate_index:
        self.processor.regenerate_root_index()

sync_item `async` ¶

sync_item(item_id_or_key=None, item_key=None, document_title='', force=False)

Sync a single item, note, or attachment by ID or key.

Source code in zocrates/rag/library_syncer.py

async def sync_item(
    self,
    item_id_or_key: int | str | None = None,
    item_key: str | None = None,
    document_title: str = "",
    force: bool = False,
) -> str:
    """Sync a single item, note, or attachment by ID or key."""
    target_key: str | None = item_key
    target_id: int | None = None

    if isinstance(item_id_or_key, str):
        target_key = item_id_or_key
    elif isinstance(item_id_or_key, int):
        target_id = item_id_or_key

    if target_id is None and target_key is not None:
        item = self.db.get_item_by_key(target_key)
        if item:
            return await self._ingest_item(item.item_id, item.item_key, item.title, force=force)
        att = self.db.get_attachment_by_key(target_key)
        if att:
            return await self._ingest_standalone_attachment(att, force=force)
        note = self.db.get_note_by_key(target_key)
        if note:
            return await self._ingest_standalone_note(note, force=force)
        return "skipped"

    if target_id is not None:
        item = self.db.get_item(target_id)
        if item:
            return await self._ingest_item(item.item_id, item.item_key, item.title, force=force)
        att = self.db.get_attachment_by_id(target_id)
        if att:
            return await self._ingest_standalone_attachment(att, force=force)
        note = self.db.get_note_by_id(target_id)
        if note:
            return await self._ingest_standalone_note(note, force=force)

    return "skipped"

sync_metadata_staleness ¶

sync_metadata_staleness()

Scan cached bundles and synchronize frontmatter and FTS5 for any modified in Zotero.

Executes a <5ms batched staleness check comparing conversion_meta.db against zotero.sqlite. If items are stale, updates their YAML frontmatter, toc.json, chunks_fts Chunk 0, and Chroma metadata without re-running document converters or OCR engines.

Returns:

Type	Description
`dict[str, int]`	Summary dict containing counts of total, updated, skipped, and errored items.

Source code in zocrates/rag/library_syncer.py

def sync_metadata_staleness(self) -> dict[str, int]:
    """Scan cached bundles and synchronize frontmatter and FTS5 for any modified in Zotero.

    Executes a <5ms batched staleness check comparing conversion_meta.db against
    zotero.sqlite. If items are stale, updates their YAML frontmatter, toc.json,
    chunks_fts Chunk 0, and Chroma metadata without re-running document converters
    or OCR engines.

    Returns:
        Summary dict containing counts of total, updated, skipped, and errored items.
    """
    from zocrates.rag.frontmatter_refresher import sync_stale_metadata
    md_cache = get_corpus_dir(self._zotero_data_dir)
    res = sync_stale_metadata(
        zotero_dir=self._zotero_data_dir,
        store=self.store,
        cache_dir=md_cache,
    )

    # Sync-time janitor: purge headless or deleted items on backend startup
    try:
        all_known = (
            set(self.processor.list_cached_keys())
            | self.processor.get_all_meta_keys()
            | self.store.get_all_item_keys()
        )
        if all_known and self.db:
            active_keys = self.db.get_active_item_keys(list(all_known))
            if not active_keys:
                logger.warning("Skipping startup orphan purge: no active items found in database")
            else:
                orphans = all_known - active_keys
                if orphans:
                    logger.info("Cleaning up %d orphaned items during staleness sync...", len(orphans))
                    for o_key in orphans:
                        self.purge_item(o_key, regenerate_index=False)
                    self.processor.regenerate_root_index()
                    res["orphans_cleaned"] = len(orphans)
    except Exception as exc:
        logger.warning("Error checking orphans during staleness sync: %s", exc)

    return res

VectorStore ¶

VectorStore(chroma_dir, fts_db_path, embedder)

Hybrid vector + keyword search index for Zotero document chunks. Uses versioned Chroma collections per embedding model/dimension.

Initialize the VectorStore.

Parameters:

Name	Type	Description	Default
`chroma_dir`	`Path`	Directory for ChromaDB persistent storage.	required
`fts_db_path`	`Path`	Filesystem path for the SQLite FTS5 database.	required
`embedder`	`EmbedFn`	Callable that maps a list of strings to a list of embedding vectors and exposes `_model_name` and `dimension` attributes.	required

Source code in zocrates/rag/vector_store.py

def __init__(
        self,
        chroma_dir: Path,
        fts_db_path: Path,
        embedder: EmbedFn,
) -> None:
    """Initialize the VectorStore.

    Args:
        chroma_dir: Directory for ChromaDB persistent storage.
        fts_db_path: Filesystem path for the SQLite FTS5 database.
        embedder: Callable that maps a list of strings to a list of
            embedding vectors and exposes ``_model_name`` and
            ``dimension`` attributes.
    """
    self._rw_lock = threading.RLock()
    self._embedder = embedder
    self._fts_path = fts_db_path
    self._client = chromadb.PersistentClient(path=str(chroma_dir))

    # Determine model name and target embedding dimension
    model_name = getattr(embedder, "_model_name", "default")
    target_dim = getattr(embedder, "dimension", 768)

    # Generate versioned collection name
    self.active_collection_name = get_collection_name(model_name, target_dim)
    logger.info(
        "Initializing VectorStore with active collection: %s (model=%s, dim=%s)",
        self.active_collection_name,
        model_name,
        target_dim,
    )

    self._collection: Collection = self._client.get_or_create_collection(
        name=self.active_collection_name,
        metadata={
            "hnsw:space": "cosine",
            "model_name": str(model_name),
            "dimension": str(target_dim),
        },
    )

fts_path `property` ¶

fts_path

Return the path to the shared SQLite FTS database.

acquire_write_lock ¶

acquire_write_lock()

Return the store's RLock for external write operations (e.g., model swap).

Returns:

Type	Description
`Any`	The `threading.RLock` guarding Chroma and FTS access.

Source code in zocrates/rag/vector_store.py

def acquire_write_lock(self) -> Any:
    """Return the store's RLock for external write operations (e.g., model swap).

    Returns:
        The ``threading.RLock`` guarding Chroma and FTS access.
    """
    return self._rw_lock

close ¶

close()

Close the Chroma client and release underlying storage handles.

Source code in zocrates/rag/vector_store.py

def close(self) -> None:
    """Close the Chroma client and release underlying storage handles."""
    try:
        if hasattr(self._client, "close"):
            self._client.close()
    except Exception as exc:
        logger.warning("Error closing Chroma client: %s", exc)

delete_document ¶

delete_document(item_key)

Remove all chunks for the given item_key from both Chroma and FTS5.

Parameters:

Name	Type	Description	Default
`item_key`	`str`	Zotero item key identifying the document to delete.	required

Source code in zocrates/rag/vector_store.py

def delete_document(self, item_key: str) -> None:
    """Remove all chunks for the given item_key from both Chroma and FTS5.

    Args:
        item_key: Zotero item key identifying the document to delete.
    """
    self._delete_item(item_key)

get_all_chunk_ids_for_item ¶

get_all_chunk_ids_for_item(item_key)

Return all chunk IDs for an item key in chunk_index order.

Parameters:

Name	Type	Description	Default
`item_key`	`str`	Zotero item key.	required

Returns:

Type	Description
`list[str]`	List of chunk IDs.

Source code in zocrates/rag/vector_store.py

def get_all_chunk_ids_for_item(self, item_key: str) -> list[str]:
    """Return all chunk IDs for an item key in chunk_index order.

    Args:
        item_key: Zotero item key.

    Returns:
        List of chunk IDs.
    """
    try:
        with self._rw_lock:
            with _open_fts_db(self._fts_path) as conn:
                cursor = conn.execute(
                    "SELECT chunk_id FROM chunk_meta WHERE item_key = ? ORDER BY chunk_index ASC",
                    (item_key,),
                )
                return [row["chunk_id"] for row in cursor.fetchall()]
    except Exception as exc:
        logger.error("Failed to query all chunk IDs for %s: %s", item_key, exc)
        return []

get_all_ingested_item_keys ¶

get_all_ingested_item_keys()

Return the set of all unique item_key strings currently ingested in the active collection.

Returns:

Type	Description
`set[str]`	Set of unique item_key strings.

Source code in zocrates/rag/vector_store.py

def get_all_ingested_item_keys(self) -> set[str]:
    """Return the set of all unique item_key strings currently ingested in the active collection.

    Returns:
        Set of unique item_key strings.
    """
    with self._rw_lock:
        result = self._collection.get(include=["metadatas"])
        metadatas = result.get("metadatas") or []
        return {m["item_key"] for m in metadatas if m and "item_key" in m}

get_all_item_keys ¶

get_all_item_keys()

Return the set of all unique item keys indexed in the chunk_meta SQLite table.

Returns:

Type	Description
`set[str]`	Set of unique item_key strings.

Source code in zocrates/rag/vector_store.py

def get_all_item_keys(self) -> set[str]:
    """Return the set of all unique item keys indexed in the chunk_meta SQLite table.

    Returns:
        Set of unique item_key strings.
    """
    try:
        with _open_fts_db(self._fts_path) as conn:
            rows = conn.execute("SELECT DISTINCT item_key FROM chunk_meta").fetchall()
            return {row["item_key"] for row in rows if row["item_key"]}
    except Exception as exc:
        logger.error("Failed to query item keys from chunk_meta: %s", exc)
        return set()

get_intersecting_chunk_ids ¶

get_intersecting_chunk_ids(item_key, start_line, end_line)

Return chunk IDs for an item key that intersect the given line range [start_line, end_line].

Parameters:

Name	Type	Description	Default
`item_key`	`str`	Zotero item key.	required
`start_line`	`int`	1-indexed start line of the range.	required
`end_line`	`int`	1-indexed end line of the range.	required

Returns:

Type	Description
`list[str]`	List of chunk IDs in chunk_index order.

Source code in zocrates/rag/vector_store.py

def get_intersecting_chunk_ids(
    self, item_key: str, start_line: int, end_line: int
) -> list[str]:
    """Return chunk IDs for an item key that intersect the given line range [start_line, end_line].

    Args:
        item_key: Zotero item key.
        start_line: 1-indexed start line of the range.
        end_line: 1-indexed end line of the range.

    Returns:
        List of chunk IDs in chunk_index order.
    """
    try:
        with self._rw_lock:
            with _open_fts_db(self._fts_path) as conn:
                cursor = conn.execute(
                    """
                    SELECT chunk_id FROM chunk_meta
                    WHERE item_key = ?
                      AND line_start IS NOT NULL
                      AND line_end IS NOT NULL
                      AND line_start <= ?
                      AND line_end >= ?
                    ORDER BY chunk_index ASC
                    """,
                    (item_key, end_line, start_line),
                )
                return [row["chunk_id"] for row in cursor.fetchall()]
    except Exception as exc:
        logger.error(
            "Failed to query intersecting chunks for %s [%d-%d]: %s",
            item_key,
            start_line,
            end_line,
            exc,
        )
        return []

ingest_document ¶

ingest_document(item_key, item_id, chunks, extra_metadata=None)

Index all chunks from a single document into Chroma and FTS5.

Deletes any existing chunks for item_key before inserting, making the operation idempotent.

Parameters:

Name	Type	Description	Default
`item_key`	`str`	Unique Zotero item key used as the primary document ID.	required
`item_id`	`int`	Numeric Zotero item ID stored as chunk metadata.	required
`chunks`	`list[ChunkResult]`	Ordered list of `ChunkResult` objects to index.	required
`extra_metadata`	`dict[str, Any] \| None`	Additional key/value pairs merged into every chunk's Chroma metadata dict.	`None`

Source code in zocrates/rag/vector_store.py

def ingest_document(
        self,
        item_key: str,
        item_id: int,
        chunks: list[ChunkResult],
        extra_metadata: dict[str, Any] | None = None,
) -> None:
    """
    Index all chunks from a single document into Chroma and FTS5.

    Deletes any existing chunks for ``item_key`` before inserting, making
    the operation idempotent.

    Args:
        item_key: Unique Zotero item key used as the primary document ID.
        item_id: Numeric Zotero item ID stored as chunk metadata.
        chunks: Ordered list of ``ChunkResult`` objects to index.
        extra_metadata: Additional key/value pairs merged into every
            chunk's Chroma metadata dict.
    """
    if not chunks:
        return

    with self._rw_lock:
        self._delete_item(item_key)

        texts = [c.text for c in chunks]
        embeddings = self._embedder(texts, task="retrieval", prompt_name="document")
        if embeddings and len(embeddings[0]) > 0:
            self._ensure_collection_dimension(len(embeddings[0]))

        parent_item_key = (extra_metadata or {}).get("parent_item_key")
        ids = [f"{item_key}::{c.chunk_index}" for c in chunks]
        metadatas = []
        for c in chunks:
            m = {
                "item_key": item_key,
                "parent_item_key": parent_item_key or "",
                "item_id": item_id,
                "chunk_index": c.chunk_index,
                "chunk_type": c.chunk_type,
                "heading": c.heading_str or "",
                "document_title": c.document_title or "",
                "chunk_id": f"{item_key}::{c.chunk_index}",
                "line_start": getattr(c, "line_start", 1),
                "line_end": getattr(c, "line_end", 1),
            }
            if c.image_path:
                m["image_path"] = c.image_path
            if c.caption_source:
                m["caption_source"] = c.caption_source
            if c.vlm_model:
                m["vlm_model"] = c.vlm_model
            if extra_metadata:
                for k, v in extra_metadata.items():
                    if v is not None:
                        m[k] = v
            metadatas.append(m)

        try:
            # Chroma batch upsert
            self._collection.upsert(
                ids=ids,
                embeddings=embeddings,
                documents=texts,
                metadatas=metadatas,
            )

            # FTS5 idempotent delete-then-insert
            with _open_fts_db(self._fts_path) as conn:
                conn.execute("DELETE FROM chunks_fts WHERE item_key = ?", (item_key,))
                conn.execute("DELETE FROM chunk_meta WHERE item_key = ?", (item_key,))
                conn.executemany(
                    "INSERT INTO chunks_fts (chunk_id, item_key, chunk_text) VALUES (?, ?, ?)",
                    [(cid, item_key, text) for cid, text in zip(ids, texts, strict=False)],
                )
                conn.executemany(
                    """
                    INSERT INTO chunk_meta (
                        chunk_id, item_key, parent_item_key, item_id,
                        chunk_index, chunk_type, heading, image_path,
                        caption_source, vlm_model, line_start, line_end
                    ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                    """,
                    [
                        (
                            cid,
                            item_key,
                            parent_item_key,
                            item_id,
                            c.chunk_index,
                            c.chunk_type,
                            c.heading_str,
                            c.image_path,
                            c.caption_source,
                            c.vlm_model,
                            getattr(c, "line_start", 1),
                            getattr(c, "line_end", 1),
                        )
                        for cid, c in zip(ids, chunks, strict=False)
                    ],
                )
                conn.commit()

            logger.info("Ingested %d chunks for item_key=%s", len(chunks), item_key)
        finally:
            del texts, embeddings, ids, metadatas

is_ingested ¶

is_ingested(item_key)

Check whether any chunks exist for item_key in the active collection.

Parameters:

Name	Type	Description	Default
`item_key`	`str`	Zotero item key to look up.	required

Returns:

Type	Description
`bool`	`True` if at least one chunk is indexed, `False` otherwise.

Source code in zocrates/rag/vector_store.py

def is_ingested(self, item_key: str) -> bool:
    """Check whether any chunks exist for item_key in the active collection.

    Args:
        item_key: Zotero item key to look up.

    Returns:
        ``True`` if at least one chunk is indexed, ``False`` otherwise.
    """
    with self._rw_lock:
        result = self._collection.get(where={"item_key": item_key}, limit=1)
        return bool(result["ids"])

list_collections ¶

list_collections()

List all stored versioned Chroma collections with their metadata and chunk counts.

Returns:

Type	Description
`list[dict[str, Any]]`	List of dicts, each with keys `name`, `model_name`, `dimension`,
`list[dict[str, Any]]`	`count`, and `is_active`.

Source code in zocrates/rag/vector_store.py

def list_collections(self) -> list[dict[str, Any]]:
    """List all stored versioned Chroma collections with their metadata and chunk counts.

    Returns:
        List of dicts, each with keys ``name``, ``model_name``, ``dimension``,
        ``count``, and ``is_active``.
    """
    result: list[dict[str, Any]] = []
    try:
        raw_cols = self._client.list_collections()
        for col_entry in raw_cols:
            col_name = (
                col_entry
                if isinstance(col_entry, str)
                else getattr(col_entry, "name", str(col_entry))
            )
            try:
                if hasattr(col_entry, "count") and hasattr(col_entry, "metadata"):
                    col = col_entry
                else:
                    col = self._client.get_collection(name=col_name)
                count = col.count()
                meta = col.metadata or {}
                dim = meta.get("dimension")
                if not dim:
                    try:
                        sample = col.get(limit=1, include=["embeddings"])
                        emb = sample.get("embeddings") if sample else None
                        if emb is not None and len(emb) > 0:
                            dim = str(len(emb[0]))
                        else:
                            dim = "unknown"
                    except Exception:
                        dim = "unknown"
                result.append({
                    "name": col_name,
                    "model_name": meta.get("model_name", "unknown"),
                    "dimension": str(dim),
                    "count": count,
                    "is_active": (col_name == self.active_collection_name),
                })
            except Exception as inner_exc:
                logger.warning("Error inspecting collection %s: %s", col_name, inner_exc)
    except Exception as exc:
        logger.error("Failed to list collections: %s", exc)
    return result

reset_collection ¶

reset_collection(target='active')

Reset/clear vector store collection(s).

Parameters:

Name	Type	Description	Default
`target`	`str`	Which collection(s) to reset: - `"active"`: delete and recreate the active versioned collection. - `"all"`: delete all Chroma collections and the FTS SQLite database. - `<collection_name>`: delete a specific named collection.	`'active'`

Returns:

Type	Description
`dict[str, Any]`	Dict with keys `status`, `target`, `deleted_collections`,
`dict[str, Any]`	`fts_reset`, and `active_collection`.

Source code in zocrates/rag/vector_store.py

def reset_collection(self, target: str = "active") -> dict[str, Any]:
    """
    Reset/clear vector store collection(s).

    Args:
        target: Which collection(s) to reset:
            - ``"active"``: delete and recreate the active versioned collection.
            - ``"all"``: delete all Chroma collections and the FTS SQLite database.
            - ``<collection_name>``: delete a specific named collection.

    Returns:
        Dict with keys ``status``, ``target``, ``deleted_collections``,
        ``fts_reset``, and ``active_collection``.
    """
    with self._rw_lock:
        deleted_collections: list[str] = []
        fts_reset = False

        if target == "all":
            try:
                raw_cols = self._client.list_collections()
                for c in raw_cols:
                    c_name = c if isinstance(c, str) else getattr(c, "name", str(c))
                    self._client.delete_collection(name=c_name)
                    deleted_collections.append(c_name)
            except Exception as exc:
                logger.error("Failed to delete all Chroma collections: %s", exc)

            if self._fts_path.exists():
                try:
                    self._fts_path.unlink()
                    fts_reset = True
                    logger.info("Unlinked shared SQLite FTS database: %s", self._fts_path)
                except Exception as exc:
                    logger.error("Failed to unlink FTS database: %s", exc)

            model_name = getattr(self._embedder, "_model_name", "default")
            target_dim = getattr(self._embedder, "dimension", 768)
            collection_metadata = {
                "hnsw:space": "cosine",
                "model_name": str(model_name),
                "dimension": str(target_dim),
            }

            self._collection = self._client.get_or_create_collection(
                name=self.active_collection_name,
                metadata=collection_metadata,
            )
        elif target == "active":
            try:
                self._client.delete_collection(name=self.active_collection_name)
                deleted_collections.append(self.active_collection_name)
            except Exception as exc:
                logger.warning("Failed to delete active collection %s: %s", self.active_collection_name, exc)

            model_name = getattr(self._embedder, "_model_name", "default")
            target_dim = getattr(self._embedder, "dimension", 768)
            collection_metadata = {
                "hnsw:space": "cosine",
                "model_name": str(model_name),
                "dimension": str(target_dim),
            }

            self._collection = self._client.get_or_create_collection(
                name=self.active_collection_name,
                metadata=collection_metadata,
            )
        else:
            try:
                self._client.delete_collection(name=target)
                deleted_collections.append(target)
                if target == self.active_collection_name:
                    model_name = getattr(self._embedder, "_model_name", "default")
                    target_dim = getattr(self._embedder, "dimension", 768)
                    collection_metadata = {
                        "hnsw:space": "cosine",
                        "model_name": str(model_name),
                        "dimension": str(target_dim),
                    }
                    self._collection = self._client.get_or_create_collection(
                        name=self.active_collection_name,
                        metadata=collection_metadata,
                    )
            except Exception as exc:
                logger.error("Failed to delete specific collection %s: %s", target, exc)

        return {
            "status": "success",
            "target": target,
            "deleted_collections": deleted_collections,
            "fts_reset": fts_reset,
            "active_collection": self.active_collection_name,
        }

search ¶

search(query, top_k=RETRIEVAL_TOP_K, final_k=RETRIEVAL_FINAL_K, item_key_filter=None, chunk_type_filter=None, exclude_chunk_ids=None, reranker=None)

Hybrid (semantic + BM25) search with RRF fusion and raw score metrics.

Parameters:

Name	Type	Description	Default
`query`	`str`	Natural language query string.	required
`top_k`	`int`	Number of candidates retrieved from each index before fusion.	`RETRIEVAL_TOP_K`
`final_k`	`int`	Maximum number of results to return after fusion.	`RETRIEVAL_FINAL_K`
`item_key_filter`	`list[str] \| None`	If provided, restrict search to these Zotero item keys.	`None`
`chunk_type_filter`	`str \| list[str] \| None`	Optional chunk type or list of chunk types to filter by.	`None`
`exclude_chunk_ids`	`set[str] \| list[str] \| None`	Optional set or list of chunk IDs to exclude from results.	`None`
`reranker`	`Callable[[str, list[str]], list[float]] \| None`	Optional callable `(query, texts) -> scores` used to re-rank the fused results after retrieval.	`None`

Returns:

Type	Description
`list[SearchResult]`	List of `SearchResult` objects sorted by descending relevance score,
`list[SearchResult]`	capped at `final_k` entries.

Source code in zocrates/rag/vector_store.py

def search(
        self,
        query: str,
        top_k: int = RETRIEVAL_TOP_K,
        final_k: int = RETRIEVAL_FINAL_K,
        item_key_filter: list[str] | None = None,
        chunk_type_filter: str | list[str] | None = None,
        exclude_chunk_ids: set[str] | list[str] | None = None,
        reranker: Callable[[str, list[str]], list[float]] | None = None,
) -> list[SearchResult]:
    """
    Hybrid (semantic + BM25) search with RRF fusion and raw score metrics.

    Args:
        query: Natural language query string.
        top_k: Number of candidates retrieved from each index before fusion.
        final_k: Maximum number of results to return after fusion.
        item_key_filter: If provided, restrict search to these Zotero item keys.
        chunk_type_filter: Optional chunk type or list of chunk types to filter by.
        exclude_chunk_ids: Optional set or list of chunk IDs to exclude from results.
        reranker: Optional callable ``(query, texts) -> scores`` used to
            re-rank the fused results after retrieval.

    Returns:
        List of ``SearchResult`` objects sorted by descending relevance score,
        capped at ``final_k`` entries.
    """
    if not query or not query.strip():
        return []

    with self._rw_lock:
        semantic_results = self._semantic_search_with_scores(
            query, top_k, item_key_filter, chunk_type_filter, exclude_chunk_ids
        )
        keyword_results = self._keyword_search_with_ranks(
            query, top_k, item_key_filter, chunk_type_filter, exclude_chunk_ids
        )

    semantic_ids = [cid for cid, _ in semantic_results]
    keyword_ids = [cid for cid, _ in keyword_results]

    # Reciprocal Rank Fusion collapses duplicates natively
    fused = _rrf_fuse(semantic_ids, keyword_ids)[:final_k]

    if not fused:
        return []

    # Map details for explanation metrics
    semantic_scores = dict(semantic_results)
    keyword_ranks = dict(keyword_results)

    fused_ids = [cid for cid, _ in fused]
    chunk_data = self._fetch_chunks(fused_ids)

    results: list[SearchResult] = []
    for cid, score in fused:
        if cid not in chunk_data:
            continue
        data = chunk_data[cid]

        sem_score = semantic_scores.get(cid)
        kw_rank = keyword_ranks.get(cid)

        # Determine retrieval method
        if sem_score is not None and kw_rank is not None:
            method = "hybrid"
        elif sem_score is not None:
            method = "semantic"
        else:
            method = "keyword"

        results.append(
            SearchResult(
                chunk_text=data["text"],
                item_key=data["item_key"],
                item_id=data["item_id"],
                heading_path=data["heading"].split(" > ") if data["heading"] else [],
                chunk_index=data["chunk_index"],
                score=score,
                chunk_type=data.get("chunk_type", "text"),
                image_path=data.get("image_path"),
                caption_source=data.get("caption_source"),
                vlm_model=data.get("vlm_model"),
                raw_semantic_score=sem_score,
                raw_keyword_rank=kw_rank,
                retrieval_method=method,
                chunk_id=cid,
                line_start=data.get("line_start"),
                line_end=data.get("line_end"),
            )
        )

    # Optional reranking hook
    if reranker is not None:
        texts = [r.chunk_text for r in results]
        rerank_scores = reranker(query, texts)
        for r, s in zip(results, rerank_scores, strict=False):
            r.score = s
        results.sort(key=lambda r: r.score, reverse=True)

    return results

update_chunk_text_preserve_vector ¶

update_chunk_text_preserve_vector(chunk_id, new_text)

Update stored document text in Chroma while explicitly preserving existing vector.

Passing existing embeddings explicitly bypasses Chroma's internal embedding function, guaranteeing zero ML overhead, zero network roundtrips, and strict dimension alignment.

Parameters:

Name	Type	Description	Default
`chunk_id`	`str`	Canonical chunk identifier (e.g. `"{item_key}::0"`).	required
`new_text`	`str`	New document text for the chunk.	required

Returns:

Type	Description
`bool`	True if chunk was found, had embeddings, and was updated; False otherwise.

Source code in zocrates/rag/vector_store.py

def update_chunk_text_preserve_vector(self, chunk_id: str, new_text: str) -> bool:
    """Update stored document text in Chroma while explicitly preserving existing vector.

    Passing existing embeddings explicitly bypasses Chroma's internal embedding function,
    guaranteeing zero ML overhead, zero network roundtrips, and strict dimension alignment.

    Args:
        chunk_id: Canonical chunk identifier (e.g. ``"{item_key}::0"``).
        new_text: New document text for the chunk.

    Returns:
        True if chunk was found, had embeddings, and was updated; False otherwise.
    """
    if not chunk_id or new_text is None:
        return False

    with self._rw_lock:
        res = self._collection.get(ids=[chunk_id], include=["embeddings"])
        embs = res.get("embeddings")
        if embs is not None and len(embs) > 0 and len(embs[0]) > 0:
            self._collection.update(
                ids=[chunk_id],
                documents=[new_text],
                embeddings=[embs[0]],
            )
            logger.info("Updated chunk text preserving vector for chunk_id=%s", chunk_id)
            return True
        return False

update_metadata ¶

update_metadata(item_key, metadata)

Update metadata for all chunks belonging to item_key without re-embedding.

Parameters:

Name	Type	Description	Default
`item_key`	`str`	Zotero item key.	required
`metadata`	`dict[str, Any]`	Key/value pairs to merge into existing chunk metadatas.	required

Returns:

Type	Description
`bool`	True if chunks were found and updated, False if item was not found.

Source code in zocrates/rag/vector_store.py

def update_metadata(self, item_key: str, metadata: dict[str, Any]) -> bool:
    """Update metadata for all chunks belonging to item_key without re-embedding.

    Args:
        item_key: Zotero item key.
        metadata: Key/value pairs to merge into existing chunk metadatas.

    Returns:
        True if chunks were found and updated, False if item was not found.
    """
    if not metadata:
        return False

    with self._rw_lock:
        res = self._collection.get(where={"item_key": item_key})
        ids = res.get("ids", [])
        metadatas = res.get("metadatas", [])

        if not ids:
            return False

        clean_metadata = {k: v for k, v in metadata.items() if v is not None}
        if not clean_metadata:
            return False

        updated_metadatas = []
        for existing_meta in metadatas:
            m = dict(existing_meta) if existing_meta else {}
            m.update(clean_metadata)
            clean_m = {k: v for k, v in m.items() if v is not None}
            updated_metadatas.append(clean_m)

        self._collection.update(ids=ids, metadatas=updated_metadatas)

        # Update chunk_meta if heading is supplied in metadata
        if "heading" in metadata:
            with _open_fts_db(self._fts_path) as conn:
                conn.execute(
                    "UPDATE chunk_meta SET heading = ? WHERE item_key = ?",
                    (metadata["heading"], item_key),
                )
                conn.commit()

        logger.info("Updated metadata for %d chunks of item_key=%s", len(ids), item_key)
        return True

ZoteroAttachment `dataclass` ¶

ZoteroAttachment(item_id, item_key, parent_item_id, parent_item_key, content_type, path, title='')

An attachment (PDF, HTML, etc.) linked to a Zotero item.

ZoteroDB ¶

ZoteroDB(sqlite_path, storage_dir)

Read-only façade over zotero.sqlite.

Initialize ZoteroDB.

Parameters:

Name	Type	Description	Default
`sqlite_path`	`Path`	Absolute path to the `zotero.sqlite` database file.	required
`storage_dir`	`Path`	Absolute path to the Zotero `storage/` directory, used to resolve relative attachment paths.	required

Source code in zocrates/rag/zotero_db.py

def __init__(self, sqlite_path: Path, storage_dir: Path) -> None:
    """Initialize ZoteroDB.

    Args:
        sqlite_path: Absolute path to the ``zotero.sqlite`` database file.
        storage_dir: Absolute path to the Zotero ``storage/`` directory,
            used to resolve relative attachment paths.
    """
    self._sqlite_path = sqlite_path
    self._storage_dir = storage_dir  # <ZoteroDataDir>/storage/
    self._table_exists_cache: dict[str, bool] = {}

get_active_item_keys ¶

get_active_item_keys(item_keys)

Return a set of item keys that exist in the library and are not deleted or parent-deleted.

Parameters:

Name	Type	Description	Default
`item_keys`	`list[str]`	List of Zotero item keys (can include regular items, attachments, notes).	required

Returns:

Type	Description
`set[str]`	Set of active item keys.

Source code in zocrates/rag/zotero_db.py

def get_active_item_keys(self, item_keys: list[str]) -> set[str]:
    """Return a set of item keys that exist in the library and are not deleted or parent-deleted.

    Args:
        item_keys: List of Zotero item keys (can include regular items, attachments, notes).

    Returns:
        Set of active item keys.
    """
    if not item_keys:
        return set()

    unique_keys = list(dict.fromkeys(k for k in item_keys if k))
    if not unique_keys:
        return set()

    active: set[str] = set()
    batch_size = _SQL_BATCH_SIZE

    with _open_db(self._sqlite_path) as conn:
        has_deleted_items = self._table_exists(conn, "deletedItems")
        has_item_attachments = self._table_exists(conn, "itemAttachments")
        has_item_notes = self._table_exists(conn, "itemNotes")

        del_clause = ""
        if has_deleted_items:
            del_clause = "AND i.itemID NOT IN (SELECT itemID FROM deletedItems)"
            if has_item_attachments:
                del_clause += " AND (ia.parentItemID IS NULL OR ia.parentItemID NOT IN (SELECT itemID FROM deletedItems))"
            if has_item_notes:
                del_clause += " AND (n.parentItemID IS NULL OR n.parentItemID NOT IN (SELECT itemID FROM deletedItems))"

        join_attachments = (
            "LEFT JOIN itemAttachments ia ON i.itemID = ia.itemID"
            if has_item_attachments
            else ""
        )
        join_notes = (
            "LEFT JOIN itemNotes n ON i.itemID = n.itemID"
            if has_item_notes
            else ""
        )

        for i in range(0, len(unique_keys), batch_size):
            batch = unique_keys[i : i + batch_size]
            placeholders = ",".join("?" for _ in batch)
            sql = f"""
                SELECT i.key
                FROM items i
                {join_attachments}
                {join_notes}
                WHERE i.key IN ({placeholders})
                  {del_clause}
            """
            rows = conn.execute(sql, batch).fetchall()
            for row in rows:
                if row["key"]:
                    active.add(row["key"])

    return active

get_all_arxiv_ids ¶

get_all_arxiv_ids()

Return a set of all normalized arXiv IDs across active non-deleted items in the library.

Source code in zocrates/rag/zotero_db.py

def get_all_arxiv_ids(self) -> set[str]:
    """Return a set of all normalized arXiv IDs across active non-deleted items in the library."""
    with _open_db(self._sqlite_path) as conn:
        has_deleted_items = self._table_exists(conn, "deletedItems")
        del_clause = "AND i.itemID NOT IN (SELECT itemID FROM deletedItems)" if has_deleted_items else ""
        sql = f"""
            SELECT f.fieldName, idv.value
            FROM items i
            JOIN itemData id ON i.itemID = id.itemID
            JOIN fields f ON id.fieldID = f.fieldID
            JOIN itemDataValues idv ON id.valueID = idv.valueID
            WHERE f.fieldName IN ('extra', 'url', 'DOI')
              {del_clause}
        """
        rows = conn.execute(sql).fetchall()
        arxiv_ids: set[str] = set()
        for r in rows:
            field_name = r["fieldName"]
            val = r["value"]
            cand: str | None = None
            if field_name == "extra":
                cand = extract_arxiv_id(extra=val)
            elif field_name == "url":
                cand = extract_arxiv_id(url=val)
            elif field_name == "DOI":
                cand = extract_arxiv_id(doi=val)
            if cand:
                arxiv_ids.add(cand)
        return arxiv_ids

get_all_attachments ¶

get_all_attachments(content_types=('application/pdf', 'text/html', 'text/markdown', 'text/plain'))

Return all file attachments with a parent item across the library.

Parameters:

Name	Type	Description	Default
`content_types`	`tuple[str, ...]`	Tuple of MIME types to include.	`('application/pdf', 'text/html', 'text/markdown', 'text/plain')`

Returns:

Type	Description
`list[ZoteroAttachment]`	List of `ZoteroAttachment` objects with resolved filesystem paths.

Source code in zocrates/rag/zotero_db.py

def get_all_attachments(
    self,
    content_types: tuple[str, ...] = ("application/pdf", "text/html", "text/markdown", "text/plain"),
) -> list[ZoteroAttachment]:
    """Return all file attachments with a parent item across the library.

    Args:
        content_types: Tuple of MIME types to include.

    Returns:
        List of ``ZoteroAttachment`` objects with resolved filesystem paths.
    """
    sql = """
        SELECT
            i.itemID,
            i.key,
            ia.parentItemID,
            parentItem.key AS parentKey,
            COALESCE(ia.contentType, '') AS contentType,
            COALESCE(ia.path, '') AS path
        FROM items i
        JOIN itemAttachments ia ON i.itemID = ia.itemID
        LEFT JOIN items parentItem ON ia.parentItemID = parentItem.itemID
        WHERE i.itemID NOT IN (SELECT itemID FROM deletedItems)
          AND ia.parentItemID IS NOT NULL
    """
    with _open_db(self._sqlite_path) as conn:
        rows = conn.execute(sql).fetchall()

    attachments: list[ZoteroAttachment] = []
    for row in rows:
        if row["contentType"] not in content_types:
            continue
        resolved = self._resolve_attachment_path(row["key"], row["path"])
        attachments.append(
            ZoteroAttachment(
                item_id=row["itemID"],
                item_key=row["key"],
                parent_item_id=row["parentItemID"],
                parent_item_key=row["parentKey"],
                content_type=row["contentType"],
                path=resolved,
            )
        )
    return attachments

get_all_convertible_attachments ¶

get_all_convertible_attachments()

Return all valid document attachments in the library in a single fast query.

Source code in zocrates/rag/zotero_db.py

def get_all_convertible_attachments(self) -> list[dict[str, Any]]:
    """Return all valid document attachments in the library in a single fast query."""
    sql = """
        SELECT
            i.itemID AS attachment_id,
            i.key AS attachment_key,
            ia.parentItemID AS parent_item_id,
            parentItem.key AS parent_key,
            COALESCE(idvTitle.value, '') AS parent_title,
            COALESCE(ia.contentType, '') AS content_type,
            COALESCE(ia.path, '') AS raw_path
        FROM items i
        JOIN itemAttachments ia ON i.itemID = ia.itemID
        LEFT JOIN items parentItem ON ia.parentItemID = parentItem.itemID
        LEFT JOIN itemData idTitle ON parentItem.itemID = idTitle.itemID AND idTitle.fieldID = (
            SELECT fieldID FROM fields WHERE fieldName = 'title'
        )
        LEFT JOIN itemDataValues idvTitle ON idTitle.valueID = idvTitle.valueID
        WHERE i.itemID NOT IN (SELECT itemID FROM deletedItems)
          AND (parentItem.itemID IS NULL OR parentItem.itemID NOT IN (SELECT itemID FROM deletedItems))
        ORDER BY i.itemID
    """
    with _open_db(self._sqlite_path) as conn:
        rows = conn.execute(sql).fetchall()

    results: list[dict[str, Any]] = []
    for row in rows:
        resolved_path = self._resolve_attachment_path(row["attachment_key"], row["raw_path"])
        if resolved_path and resolved_path.exists():
            results.append({
                "attachment_id": row["attachment_id"],
                "attachment_key": row["attachment_key"],
                "parent_item_id": row["parent_item_id"],
                "parent_key": row["parent_key"] or "",
                "parent_title": row["parent_title"] or "",
                "content_type": row["content_type"],
                "path": resolved_path,
                "filename": resolved_path.name,
            })
    return results

get_all_dedup_records ¶

get_all_dedup_records()

Return lightweight records (title, year, creators, doi, arxiv_id) for active items for fuzzy dedup.

Source code in zocrates/rag/zotero_db.py

def get_all_dedup_records(self) -> list[dict[str, Any]]:
    """Return lightweight records (title, year, creators, doi, arxiv_id) for active items for fuzzy dedup."""
    with _open_db(self._sqlite_path) as conn:
        has_deleted_items = self._table_exists(conn, "deletedItems")
        del_clause = "AND i.itemID NOT IN (SELECT itemID FROM deletedItems)" if has_deleted_items else ""
        sql = f"""
            SELECT i.itemID, f.fieldName, idv.value
            FROM items i
            JOIN itemTypes it ON i.itemTypeID = it.itemTypeID
            JOIN itemData id ON i.itemID = id.itemID
            JOIN fields f ON id.fieldID = f.fieldID
            JOIN itemDataValues idv ON id.valueID = idv.valueID
            WHERE it.typeName NOT IN ('attachment', 'note')
              AND f.fieldName IN ('title', 'date', 'DOI', 'url', 'extra')
              {del_clause}
            ORDER BY i.itemID
        """
        rows = conn.execute(sql).fetchall()

        records_map: dict[int, dict[str, Any]] = {}
        for r in rows:
            item_id = r["itemID"]
            if item_id not in records_map:
                records_map[item_id] = {
                    "item_id": item_id,
                    "title": None,
                    "year": None,
                    "doi": None,
                    "url": None,
                    "extra": None,
                }
            field_name = r["fieldName"]
            val = r["value"]
            if field_name == "title":
                records_map[item_id]["title"] = val
            elif field_name == "date":
                y_str = extract_year(val)
                if y_str:
                    try:
                        records_map[item_id]["year"] = int(y_str)
                    except ValueError:
                        pass
            elif field_name == "DOI":
                records_map[item_id]["doi"] = normalize_doi(val)
            elif field_name == "url":
                records_map[item_id]["url"] = val
            elif field_name == "extra":
                records_map[item_id]["extra"] = val

        sql_creators = f"""
            SELECT ic.itemID, c.lastName
            FROM itemCreators ic
            JOIN creators c ON ic.creatorID = c.creatorID
            JOIN items i ON ic.itemID = i.itemID
            WHERE 1=1 {del_clause}
            ORDER BY ic.itemID, ic.orderIndex
        """
        c_rows = conn.execute(sql_creators).fetchall()
        creators_map: dict[int, list[str]] = {}
        for cr in c_rows:
            i_id = cr["itemID"]
            lname = (cr["lastName"] or "").strip()
            if lname:
                creators_map.setdefault(i_id, []).append(lname)

        results: list[dict[str, Any]] = []
        for item_id, rec in records_map.items():
            title = rec["title"]
            if not title:
                continue
            arxiv_id = extract_arxiv_id(extra=rec["extra"], url=rec["url"], doi=rec["doi"])
            results.append({
                "item_id": item_id,
                "title": title,
                "year": rec["year"],
                "doi": rec["doi"],
                "arxiv_id": arxiv_id,
                "creators": creators_map.get(item_id, []),
            })
        return results

get_all_dois ¶

get_all_dois()

Return a set of all normalized DOIs across active non-deleted items in the library.

Source code in zocrates/rag/zotero_db.py

def get_all_dois(self) -> set[str]:
    """Return a set of all normalized DOIs across active non-deleted items in the library."""
    with _open_db(self._sqlite_path) as conn:
        has_deleted_items = self._table_exists(conn, "deletedItems")
        del_clause = "AND i.itemID NOT IN (SELECT itemID FROM deletedItems)" if has_deleted_items else ""
        sql = f"""
            SELECT idv.value
            FROM items i
            JOIN itemData id ON i.itemID = id.itemID
            JOIN fields f ON id.fieldID = f.fieldID
            JOIN itemDataValues idv ON id.valueID = idv.valueID
            WHERE f.fieldName = 'DOI'
              {del_clause}
        """
        rows = conn.execute(sql).fetchall()
        dois: set[str] = set()
        for r in rows:
            norm = normalize_doi(r["value"])
            if norm:
                dois.add(norm)
        return dois

get_all_item_relations ¶

get_all_item_relations()

Return all explicit item relations as (source_key, target_key) pairs.

Source code in zocrates/rag/zotero_db.py

def get_all_item_relations(self) -> list[tuple[str, str]]:
    """Return all explicit item relations as (source_key, target_key) pairs."""
    with _open_db(self._sqlite_path) as conn:
        if not self._table_exists(conn, "itemRelations"):
            return []

        has_deleted_items = self._table_exists(conn, "deletedItems")
        del_clause = "WHERE i.itemID NOT IN (SELECT itemID FROM deletedItems)" if has_deleted_items else ""
        sql = f"""
            SELECT i.key AS source_key, ir.object AS target_uri
            FROM itemRelations ir
            JOIN items i ON ir.itemID = i.itemID
            {del_clause}
        """
        rows = conn.execute(sql).fetchall()

        del_keys: set[str] = set()
        if has_deleted_items and self._table_exists(conn, "items"):
            del_rows = conn.execute(
                "SELECT key FROM items WHERE itemID IN (SELECT itemID FROM deletedItems)"
            ).fetchall()
            del_keys = {r["key"] for r in del_rows if r["key"]}

        relations: list[tuple[str, str]] = []
        seen: set[tuple[str, str]] = set()
        for r in rows:
            src = str(r["source_key"] or "").strip()
            target_uri = str(r["target_uri"] or "").strip()
            if not src or "/items/" not in target_uri:
                continue
            target_key = target_uri.split("/items/")[-1].strip().split("/")[0].split("?")[0].split("#")[0]
            if target_key and target_key not in del_keys and target_key != src:
                pair = (src, target_key)
                if pair not in seen:
                    seen.add(pair)
                    relations.append(pair)

        return relations

get_all_items ¶

get_all_items()

Return all non-attachment, non-deleted items in the library.

Returns:

Type	Description
`list[ZoteroItem]`	List of fully populated `ZoteroItem` objects.

Source code in zocrates/rag/zotero_db.py

def get_all_items(self) -> list[ZoteroItem]:
    """Return all non-attachment, non-deleted items in the library.

    Returns:
        List of fully populated ``ZoteroItem`` objects.
    """
    with _open_db(self._sqlite_path) as conn:
        has_deleted = self._table_exists(conn, "deletedItems")
        del_clause = "AND i.itemID NOT IN (SELECT itemID FROM deletedItems)" if has_deleted else ""
        sql = f"""
            SELECT
                i.itemID,
                i.key,
                it.typeName
            FROM items i
            JOIN itemTypes it ON i.itemTypeID = it.itemTypeID
            WHERE it.typeName != 'attachment'
              AND it.typeName != 'note'
              {del_clause}
            ORDER BY i.itemID
        """
        rows = conn.execute(sql).fetchall()
        return self._build_items_batched(conn, rows)

get_all_notes ¶

get_all_notes()

Return all notes (standalone and child notes) in the library.

Returns:

Type	Description
`list[ZoteroNote]`	List of `ZoteroNote` objects ordered by `note_id`.

Source code in zocrates/rag/zotero_db.py

def get_all_notes(self) -> list[ZoteroNote]:
    """Return all notes (standalone and child notes) in the library.

    Returns:
        List of ``ZoteroNote`` objects ordered by ``note_id``.
    """
    sql = """
        SELECT
            i.itemID AS noteID,
            i.key AS noteKey,
            n.parentItemID,
            parentItem.key AS parentKey,
            COALESCE(n.note, '') AS noteHtml,
            COALESCE(n.title, '') AS noteTitle
        FROM items i
        JOIN itemTypes it ON i.itemTypeID = it.itemTypeID
        JOIN itemNotes n ON i.itemID = n.itemID
        LEFT JOIN items parentItem ON n.parentItemID = parentItem.itemID
        WHERE it.typeName = 'note'
          AND i.itemID NOT IN (SELECT itemID FROM deletedItems)
        ORDER BY i.itemID
    """
    with _open_db(self._sqlite_path) as conn:
        rows = conn.execute(sql).fetchall()

    notes: list[ZoteroNote] = []
    for row in rows:
        notes.append(
            ZoteroNote(
                note_id=row["noteID"],
                note_key=row["noteKey"],
                parent_item_id=row["parentItemID"],
                parent_item_key=row["parentKey"],
                title=row["noteTitle"],
                note_html=row["noteHtml"],
            )
        )
    return notes

get_all_titles ¶

get_all_titles()

Return a list of all titles across active non-deleted items in the library.

Source code in zocrates/rag/zotero_db.py

def get_all_titles(self) -> list[str]:
    """Return a list of all titles across active non-deleted items in the library."""
    with _open_db(self._sqlite_path) as conn:
        has_deleted_items = self._table_exists(conn, "deletedItems")
        del_clause = "AND i.itemID NOT IN (SELECT itemID FROM deletedItems)" if has_deleted_items else ""
        sql = f"""
            SELECT idv.value
            FROM items i
            JOIN itemData id ON i.itemID = id.itemID
            JOIN fields f ON id.fieldID = f.fieldID
            JOIN itemDataValues idv ON id.valueID = idv.valueID
            WHERE f.fieldName = 'title'
              {del_clause}
        """
        rows = conn.execute(sql).fetchall()
        titles: list[str] = []
        for r in rows:
            t = (r["value"] or "").strip()
            if t:
                titles.append(t)
        return titles

get_attachment_by_id ¶

get_attachment_by_id(item_id)

Fetch a single file attachment by its item ID.

Parameters:

Name	Type	Description	Default
`item_id`	`int`	Numeric Zotero item ID of the attachment.	required

Returns:

Type	Description
`ZoteroAttachment \| None`	`ZoteroAttachment` with a resolved path, or `None` if not found
`ZoteroAttachment \| None`	or deleted.

Source code in zocrates/rag/zotero_db.py

def get_attachment_by_id(
    self,
    item_id: int,
) -> ZoteroAttachment | None:
    """Fetch a single file attachment by its item ID.

    Args:
        item_id: Numeric Zotero item ID of the attachment.

    Returns:
        ``ZoteroAttachment`` with a resolved path, or ``None`` if not found
        or deleted.
    """
    sql = """
        SELECT
            i.itemID,
            i.key,
            ia.parentItemID,
            parentItem.key AS parentKey,
            COALESCE(ia.contentType, '') AS contentType,
            COALESCE(ia.path, '') AS path
        FROM items i
        JOIN itemAttachments ia ON i.itemID = ia.itemID
        LEFT JOIN items parentItem ON ia.parentItemID = parentItem.itemID
        WHERE i.itemID = ?
          AND i.itemID NOT IN (SELECT itemID FROM deletedItems)
    """
    with _open_db(self._sqlite_path) as conn:
        row = conn.execute(sql, (item_id,)).fetchone()
        if row is None:
            return None
        resolved = self._resolve_attachment_path(row["key"], row["path"])
        return ZoteroAttachment(
            item_id=row["itemID"],
            item_key=row["key"],
            parent_item_id=row["parentItemID"],
            parent_item_key=row["parentKey"],
            content_type=row["contentType"],
            path=resolved,
        )

get_attachment_by_key ¶

get_attachment_by_key(item_key)

Fetch a single file attachment by its item key or parent item key.

Parameters:

Name	Type	Description	Default
`item_key`	`str`	Zotero item key string (attachment key or parent item key).	required

Returns:

Type	Description
`ZoteroAttachment \| None`	`ZoteroAttachment` with a resolved path, or `None` if not found.

Source code in zocrates/rag/zotero_db.py

def get_attachment_by_key(
    self,
    item_key: str,
) -> ZoteroAttachment | None:
    """Fetch a single file attachment by its item key or parent item key.

    Args:
        item_key: Zotero item key string (attachment key or parent item key).

    Returns:
        ``ZoteroAttachment`` with a resolved path, or ``None`` if not found.
    """
    sql = """
        SELECT
            i.itemID,
            i.key,
            ia.parentItemID,
            parentItem.key AS parentKey,
            COALESCE(ia.contentType, '') AS contentType,
            COALESCE(ia.path, '') AS path
        FROM items i
        JOIN itemAttachments ia ON i.itemID = ia.itemID
        LEFT JOIN items parentItem ON ia.parentItemID = parentItem.itemID
        WHERE (i.key = ? OR parentItem.key = ?)
          AND i.itemID NOT IN (SELECT itemID FROM deletedItems)
        ORDER BY CASE WHEN ia.contentType = 'application/pdf' THEN 0 ELSE 1 END
    """
    with _open_db(self._sqlite_path) as conn:
        row = conn.execute(sql, (item_key, item_key)).fetchone()
        if row:
            resolved = self._resolve_attachment_path(row["key"], row["path"])
            return ZoteroAttachment(
                item_id=row["itemID"],
                item_key=row["key"],
                parent_item_id=row["parentItemID"],
                parent_item_key=row["parentKey"],
                content_type=row["contentType"],
                path=resolved,
            )

    item = self.get_item_by_key(item_key)
    if item:
        attachments = self.get_attachments(item.item_id)
        if attachments:
            return sorted(
                attachments,
                key=lambda a: 0 if a.content_type == "application/pdf" else 1,
            )[0]

    return None

get_attachments ¶

get_attachments(parent_item_id, content_types=('application/pdf', 'text/html', 'text/markdown', 'text/plain'))

Return file attachments for a given parent item.

Parameters:

Name	Type	Description	Default
`parent_item_id`	`int`	Numeric ID of the parent Zotero item.	required
`content_types`	`tuple[str, ...]`	Tuple of MIME types to include; defaults to PDF, HTML, Markdown, and plain text.	`('application/pdf', 'text/html', 'text/markdown', 'text/plain')`

Returns:

Type	Description
`list[ZoteroAttachment]`	List of `ZoteroAttachment` objects with resolved filesystem
`list[ZoteroAttachment]`	paths (`path` is `None` if the file cannot be resolved).
`list[ZoteroAttachment]`	The `title` field is set to the Zotero attachment title if
`list[ZoteroAttachment]`	available, otherwise the attachment filename.

Source code in zocrates/rag/zotero_db.py

def get_attachments(
    self,
    parent_item_id: int,
    content_types: tuple[str, ...] = ("application/pdf", "text/html", "text/markdown", "text/plain"),
) -> list[ZoteroAttachment]:
    """Return file attachments for a given parent item.

    Args:
        parent_item_id: Numeric ID of the parent Zotero item.
        content_types: Tuple of MIME types to include; defaults to PDF,
            HTML, Markdown, and plain text.

    Returns:
        List of ``ZoteroAttachment`` objects with resolved filesystem
        paths (``path`` is ``None`` if the file cannot be resolved).
        The ``title`` field is set to the Zotero attachment title if
        available, otherwise the attachment filename.
    """
    sql = """
        SELECT
            i.itemID,
            i.key,
            ia.parentItemID,
            parentItem.key AS parentKey,
            COALESCE(ia.contentType, '') AS contentType,
            COALESCE(ia.path, '') AS path,
            COALESCE(att_meta.att_title, '') AS attachmentTitle
        FROM items i
        JOIN itemAttachments ia ON i.itemID = ia.itemID
        LEFT JOIN items parentItem ON ia.parentItemID = parentItem.itemID
        LEFT JOIN (
            SELECT id2.itemID, idv2.value AS att_title
            FROM itemData id2
            JOIN fields f2 ON id2.fieldID = f2.fieldID
            JOIN itemDataValues idv2 ON id2.valueID = idv2.valueID
            WHERE f2.fieldName = 'title'
        ) att_meta ON i.itemID = att_meta.itemID
        WHERE ia.parentItemID = ?
          AND i.itemID NOT IN (SELECT itemID FROM deletedItems)
    """
    with _open_db(self._sqlite_path) as conn:
        rows = conn.execute(sql, (parent_item_id,)).fetchall()

    attachments: list[ZoteroAttachment] = []
    for row in rows:
        if row["contentType"] not in content_types:
            continue
        resolved = self._resolve_attachment_path(row["key"], row["path"])
        raw_title = (row["attachmentTitle"] or "").strip()
        filename = (resolved.name if resolved and resolved.name else "")
        if raw_title and filename and raw_title.lower() != filename.lower():
            att_title = f"{raw_title} ({filename})"
        elif raw_title:
            att_title = raw_title
        elif filename:
            att_title = filename
        else:
            att_title = "Attachment"
        attachments.append(
            ZoteroAttachment(
                item_id=row["itemID"],
                item_key=row["key"],
                parent_item_id=row["parentItemID"],
                parent_item_key=row["parentKey"],
                content_type=row["contentType"],
                path=resolved,
                title=att_title,
            )
        )
    return attachments

get_item ¶

get_item(item_id)

Fetch a single item with full metadata.

Parameters:

Name	Type	Description	Default
`item_id`	`int`	Numeric Zotero item ID.	required

Returns:

Type	Description
`ZoteroItem \| None`	Fully populated `ZoteroItem` (with creators, tags, collections,
`ZoteroItem \| None`	and relations), or `None` if not found or deleted.

Source code in zocrates/rag/zotero_db.py

def get_item(self, item_id: int) -> ZoteroItem | None:
    """Fetch a single item with full metadata.

    Args:
        item_id: Numeric Zotero item ID.

    Returns:
        Fully populated ``ZoteroItem`` (with creators, tags, collections,
        and relations), or ``None`` if not found or deleted.
    """
    with _open_db(self._sqlite_path) as conn:
        has_deleted = self._table_exists(conn, "deletedItems")
        del_clause = "AND i.itemID NOT IN (SELECT itemID FROM deletedItems)" if has_deleted else ""
        sql = f"""
            SELECT i.itemID, i.key, it.typeName
            FROM items i
            JOIN itemTypes it ON i.itemTypeID = it.itemTypeID
            WHERE i.itemID = ?
              {del_clause}
        """
        row = conn.execute(sql, (item_id,)).fetchone()
        if row is None:
            return None
        return self._build_item(conn, row)

get_item_by_key ¶

get_item_by_key(item_key)

Fetch a single item by its string key.

Parameters:

Name	Type	Description	Default
`item_key`	`str`	Zotero item key string (e.g. '89X2ELQ6').	required

Returns:

Type	Description
`ZoteroItem \| None`	Fully populated `ZoteroItem` or `None` if not found.

Source code in zocrates/rag/zotero_db.py

def get_item_by_key(self, item_key: str) -> ZoteroItem | None:
    """Fetch a single item by its string key.

    Args:
        item_key: Zotero item key string (e.g. '89X2ELQ6').

    Returns:
        Fully populated ``ZoteroItem`` or ``None`` if not found.
    """
    with _open_db(self._sqlite_path) as conn:
        has_deleted = self._table_exists(conn, "deletedItems")
        del_clause = "AND itemID NOT IN (SELECT itemID FROM deletedItems)" if has_deleted else ""
        sql = f"""
            SELECT itemID FROM items WHERE key = ? {del_clause}
        """
        row = conn.execute(sql, (item_key,)).fetchone()
        if row is None:
            return None
        return self.get_item(row["itemID"])

get_item_date_modified ¶

get_item_date_modified(item_key)

Fetch the last modified timestamp for an item as timezone-aware UTC datetime.

Parameters:

Name	Type	Description	Default
`item_key`	`str`	8-character Zotero alphanumeric key.	required

Returns:

Type	Description
`datetime \| None`	Timezone-aware UTC datetime, or None if item not found.

Source code in zocrates/rag/zotero_db.py

def get_item_date_modified(self, item_key: str) -> datetime | None:
    """Fetch the last modified timestamp for an item as timezone-aware UTC datetime.

    Args:
        item_key: 8-character Zotero alphanumeric key.

    Returns:
        Timezone-aware UTC datetime, or None if item not found.
    """
    sql = """
        SELECT COALESCE(clientDateModified, dateModified) AS modified
        FROM items
        WHERE key = ?
          AND itemID NOT IN (SELECT itemID FROM deletedItems)
    """
    with _open_db(self._sqlite_path) as conn:
        row = conn.execute(sql, (item_key,)).fetchone()
        if row and row["modified"]:
            return _parse_utc_datetime(row["modified"])
    return None

get_item_keys_in_collection ¶

get_item_keys_in_collection(collection_name, recursive=True)

Return the unique item keys for all items belonging to a collection name or key.

Parameters:

Name	Type	Description	Default
`collection_name`	`str`	Name or key of the collection.	required
`recursive`	`bool`	If True and parentCollectionID exists, includes items in all subcollections.	`True`

Returns:

Type	Description
`list[str]`	List of unique item keys.

Source code in zocrates/rag/zotero_db.py

def get_item_keys_in_collection(
    self, collection_name: str, recursive: bool = True
) -> list[str]:
    """Return the unique item keys for all items belonging to a collection name or key.

    Args:
        collection_name: Name or key of the collection.
        recursive: If True and parentCollectionID exists, includes items in all subcollections.

    Returns:
        List of unique item keys.
    """
    with _open_db(self._sqlite_path) as conn:
        has_deleted = self._table_exists(conn, "deletedItems")
        del_clause = "AND i.itemID NOT IN (SELECT itemID FROM deletedItems)" if has_deleted else ""

        col_columns = {c[1] for c in conn.execute("PRAGMA table_info(collections)").fetchall()}
        has_parent_col = "parentCollectionID" in col_columns

        if recursive and has_parent_col:
            sql = f"""
                WITH RECURSIVE subcols(collectionID) AS (
                    SELECT collectionID
                    FROM collections
                    WHERE collectionName = ? OR key = ?
                    UNION
                    SELECT c.collectionID
                    FROM collections c
                    JOIN subcols s ON c.parentCollectionID = s.collectionID
                )
                SELECT DISTINCT i.key
                FROM items i
                JOIN collectionItems ci ON i.itemID = ci.itemID
                JOIN subcols s ON ci.collectionID = s.collectionID
                WHERE 1=1 {del_clause}
            """
        else:
            sql = f"""
                SELECT DISTINCT i.key
                FROM items i
                JOIN collectionItems ci ON i.itemID = ci.itemID
                JOIN collections col ON ci.collectionID = col.collectionID
                WHERE (col.collectionName = ? OR col.key = ?)
                  {del_clause}
            """
        rows = conn.execute(sql, (collection_name, collection_name)).fetchall()
        return [row["key"] for row in rows]

get_item_relations ¶

get_item_relations(item_key_or_id)

Return target item keys related to the specified item.

Source code in zocrates/rag/zotero_db.py

def get_item_relations(self, item_key_or_id: str | int) -> list[str]:
    """Return target item keys related to the specified item."""
    with _open_db(self._sqlite_path) as conn:
        has_deleted_items = self._table_exists(conn, "deletedItems")
        item_id: int | None = None
        if isinstance(item_key_or_id, int):
            if has_deleted_items:
                del_row = conn.execute(
                    "SELECT 1 FROM deletedItems WHERE itemID = ?",
                    (item_key_or_id,),
                ).fetchone()
                if del_row:
                    return []
            item_id = item_key_or_id
        else:
            del_clause = "AND itemID NOT IN (SELECT itemID FROM deletedItems)" if has_deleted_items else ""
            row = conn.execute(
                f"SELECT itemID FROM items WHERE key = ? {del_clause}",
                (item_key_or_id,),
            ).fetchone()
            if not row:
                return []
            item_id = row["itemID"]

        return self._fetch_relations(conn, item_id)

get_item_tags ¶

get_item_tags(item_key_or_id, tag_type=None)

Return tag names for an item, optionally filtered by tag type (0=user, 1=automatic).

Source code in zocrates/rag/zotero_db.py

def get_item_tags(
    self, item_key_or_id: str | int, tag_type: int | None = None
) -> list[str]:
    """Return tag names for an item, optionally filtered by tag type (0=user, 1=automatic)."""
    with _open_db(self._sqlite_path) as conn:
        has_item_tags = self._table_exists(conn, "itemTags")
        if not has_item_tags:
            return []

        item_id: int | None = None
        has_deleted_items = self._table_exists(conn, "deletedItems")
        if isinstance(item_key_or_id, int):
            if has_deleted_items:
                del_row = conn.execute(
                    "SELECT 1 FROM deletedItems WHERE itemID = ?",
                    (item_key_or_id,),
                ).fetchone()
                if del_row:
                    return []
            item_id = item_key_or_id
        else:
            del_clause = "AND itemID NOT IN (SELECT itemID FROM deletedItems)" if has_deleted_items else ""
            row = conn.execute(
                f"SELECT itemID FROM items WHERE key = ? {del_clause}",
                (item_key_or_id,),
            ).fetchone()
            if not row:
                return []
            item_id = row["itemID"]

        params: list[Any] = [item_id]
        type_clause = ""
        if tag_type is not None:
            cols = {c[1] for c in conn.execute("PRAGMA table_info(itemTags)").fetchall()}
            if "type" in cols:
                type_clause = " AND it.type = ?"
                params.append(tag_type)

        sql = f"""
            SELECT DISTINCT t.name
            FROM itemTags it
            JOIN tags t ON it.tagID = t.tagID
            WHERE it.itemID = ?{type_clause}
            ORDER BY t.name
        """
        rows = conn.execute(sql, params).fetchall()
        if not rows:
            has_attachments_table = self._table_exists(conn, "itemAttachments")
            if has_attachments_table:
                p_row = conn.execute(
                    "SELECT parentItemID FROM itemAttachments WHERE itemID = ?",
                    (item_id,),
                ).fetchone()
                if p_row and p_row["parentItemID"]:
                    params[0] = p_row["parentItemID"]
                    rows = conn.execute(sql, params).fetchall()

        return [r["name"] for r in rows if r["name"]]

get_items_by_ids ¶

get_items_by_ids(item_ids)

Fetch multiple items by their numeric IDs.

Parameters:

Name	Type	Description	Default
`item_ids`	`list[int]`	List of numeric Zotero item IDs.	required

Returns:

Type	Description
`list[ZoteroItem]`	List of fully populated `ZoteroItem` objects (with creators,
`list[ZoteroItem]`	tags, and collections). Items not found or deleted are silently
`list[ZoteroItem]`	omitted.

Source code in zocrates/rag/zotero_db.py

def get_items_by_ids(self, item_ids: list[int]) -> list[ZoteroItem]:
    """Fetch multiple items by their numeric IDs.

    Args:
        item_ids: List of numeric Zotero item IDs.

    Returns:
        List of fully populated ``ZoteroItem`` objects (with creators,
        tags, and collections). Items not found or deleted are silently
        omitted.
    """
    if not item_ids:
        return []
    batch_size = _SQL_BATCH_SIZE
    with _open_db(self._sqlite_path) as conn:
        has_deleted = self._table_exists(conn, "deletedItems")
        del_clause = "AND i.itemID NOT IN (SELECT itemID FROM deletedItems)" if has_deleted else ""
        all_rows: list[sqlite3.Row] = []
        for i in range(0, len(item_ids), batch_size):
            batch_chunk = item_ids[i : i + batch_size]
            placeholders = ",".join("?" for _ in batch_chunk)
            sql = f"""
                SELECT i.itemID, i.key, it.typeName
                FROM items i
                JOIN itemTypes it ON i.itemTypeID = it.itemTypeID
                WHERE i.itemID IN ({placeholders})
                  {del_clause}
            """
            all_rows.extend(conn.execute(sql, batch_chunk).fetchall())
        return self._build_items_batched(conn, all_rows)

get_new_item_ids_since ¶

get_new_item_ids_since(last_seen_item_id)

Return item IDs created after last_seen_item_id for incremental sync.

Parameters:

Name	Type	Description	Default
`last_seen_item_id`	`int`	The highest item ID already processed.	required

Returns:

Type	Description
`list[int]`	Sorted list of numeric item IDs greater than `last_seen_item_id`
`list[int]`	that are not attachments, notes, or deleted items.

Source code in zocrates/rag/zotero_db.py

def get_new_item_ids_since(self, last_seen_item_id: int) -> list[int]:
    """Return item IDs created after last_seen_item_id for incremental sync.

    Args:
        last_seen_item_id: The highest item ID already processed.

    Returns:
        Sorted list of numeric item IDs greater than ``last_seen_item_id``
        that are not attachments, notes, or deleted items.
    """
    sql = """
        SELECT i.itemID
        FROM items i
        JOIN itemTypes it ON i.itemTypeID = it.itemTypeID
        WHERE it.typeName != 'attachment'
          AND it.typeName != 'note'
          AND i.itemID > ?
          AND i.itemID NOT IN (SELECT itemID FROM deletedItems)
        ORDER BY i.itemID
    """
    with _open_db(self._sqlite_path) as conn:
        rows = conn.execute(sql, (last_seen_item_id,)).fetchall()
    return [row["itemID"] for row in rows]

get_note_by_id ¶

get_note_by_id(note_id)

Fetch a single note by note item ID.

Source code in zocrates/rag/zotero_db.py

def get_note_by_id(self, note_id: int) -> ZoteroNote | None:
    """Fetch a single note by note item ID."""
    sql = """
        SELECT
            i.itemID AS noteID,
            i.key AS noteKey,
            n.parentItemID,
            parentItem.key AS parentKey,
            COALESCE(n.note, '') AS noteHtml,
            COALESCE(n.title, '') AS noteTitle
        FROM items i
        JOIN itemTypes it ON i.itemTypeID = it.itemTypeID
        JOIN itemNotes n ON i.itemID = n.itemID
        LEFT JOIN items parentItem ON n.parentItemID = parentItem.itemID
        WHERE i.itemID = ?
          AND i.itemID NOT IN (SELECT itemID FROM deletedItems)
    """
    with _open_db(self._sqlite_path) as conn:
        row = conn.execute(sql, (note_id,)).fetchone()
        if row is None:
            return None
        return ZoteroNote(
            note_id=row["noteID"],
            note_key=row["noteKey"],
            parent_item_id=row["parentItemID"],
            parent_item_key=row["parentKey"],
            title=row["noteTitle"],
            note_html=row["noteHtml"],
        )

get_note_by_key ¶

get_note_by_key(note_key)

Fetch a single note by its Zotero item key.

Parameters:

Name	Type	Description	Default
`note_key`	`str`	The alphanumeric Zotero item key of the note.	required

Returns:

Type	Description
`ZoteroNote \| None`	A `ZoteroNote` if found, or `None` if the key does not exist
`ZoteroNote \| None`	or the item has been deleted.

Source code in zocrates/rag/zotero_db.py

def get_note_by_key(self, note_key: str) -> "ZoteroNote | None":
    """Fetch a single note by its Zotero item key.

    Args:
        note_key: The alphanumeric Zotero item key of the note.

    Returns:
        A ``ZoteroNote`` if found, or ``None`` if the key does not exist
        or the item has been deleted.
    """
    sql = """
        SELECT
            i.itemID AS noteID,
            i.key AS noteKey,
            n.parentItemID,
            parentItem.key AS parentKey,
            COALESCE(n.note, '') AS noteHtml,
            COALESCE(n.title, '') AS noteTitle
        FROM items i
        JOIN itemTypes it ON i.itemTypeID = it.itemTypeID
        JOIN itemNotes n ON i.itemID = n.itemID
        LEFT JOIN items parentItem ON n.parentItemID = parentItem.itemID
        WHERE i.key = ?
          AND i.itemID NOT IN (SELECT itemID FROM deletedItems)
    """
    with _open_db(self._sqlite_path) as conn:
        row = conn.execute(sql, (note_key,)).fetchone()
        if row is None:
            return None
        return ZoteroNote(
            note_id=row["noteID"],
            note_key=row["noteKey"],
            parent_item_id=row["parentItemID"],
            parent_item_key=row["parentKey"],
            title=row["noteTitle"],
            note_html=row["noteHtml"],
        )

get_notes_for_item ¶

get_notes_for_item(parent_item_id)

Return child notes attached to a specific parent item.

Parameters:

Name	Type	Description	Default
`parent_item_id`	`int`	Numeric ID of the parent Zotero item.	required

Returns:

Type	Description
`list[ZoteroNote]`	List of `ZoteroNote` objects whose `parent_item_id` matches.

Source code in zocrates/rag/zotero_db.py

def get_notes_for_item(self, parent_item_id: int) -> list[ZoteroNote]:
    """Return child notes attached to a specific parent item.

    Args:
        parent_item_id: Numeric ID of the parent Zotero item.

    Returns:
        List of ``ZoteroNote`` objects whose ``parent_item_id`` matches.
    """
    sql = """
        SELECT
            i.itemID AS noteID,
            i.key AS noteKey,
            n.parentItemID,
            parentItem.key AS parentKey,
            COALESCE(n.note, '') AS noteHtml,
            COALESCE(n.title, '') AS noteTitle
        FROM items i
        JOIN itemTypes it ON i.itemTypeID = it.itemTypeID
        JOIN itemNotes n ON i.itemID = n.itemID
        LEFT JOIN items parentItem ON n.parentItemID = parentItem.itemID
        WHERE it.typeName = 'note'
          AND n.parentItemID = ?
          AND i.itemID NOT IN (SELECT itemID FROM deletedItems)
        ORDER BY i.itemID
    """
    with _open_db(self._sqlite_path) as conn:
        rows = conn.execute(sql, (parent_item_id,)).fetchall()

    notes: list[ZoteroNote] = []
    for row in rows:
        notes.append(
            ZoteroNote(
                note_id=row["noteID"],
                note_key=row["noteKey"],
                parent_item_id=row["parentItemID"],
                parent_item_key=row["parentKey"],
                title=row["noteTitle"],
                note_html=row["noteHtml"],
            )
        )
    return notes

get_stale_item_keys ¶

get_stale_item_keys(converted_map, key_to_atts)

Identify cached items whose Zotero metadata has been modified since conversion.

Parameters:

Name	Type	Description	Default
`converted_map`	`dict[str, str]`	Mapping of attachment/note item_key -> converted_at ISO string.	required
`key_to_atts`	`dict[str, list[str]]`	Mapping of query key (parent_item_key or direct item_key) to list of associated attachment/note item_keys.	required

Returns:

Type	Description
`list[str]`	Sorted, unique list of attachment/note item_keys whose metadata is stale.

Source code in zocrates/rag/zotero_db.py

def get_stale_item_keys(
    self,
    converted_map: dict[str, str],
    key_to_atts: dict[str, list[str]],
) -> list[str]:
    """Identify cached items whose Zotero metadata has been modified since conversion.

    Args:
        converted_map: Mapping of attachment/note item_key -> converted_at ISO string.
        key_to_atts: Mapping of query key (parent_item_key or direct item_key)
            to list of associated attachment/note item_keys.

    Returns:
        Sorted, unique list of attachment/note item_keys whose metadata is stale.
    """
    if not key_to_atts or not converted_map:
        return []

    stale_item_keys: set[str] = set()
    query_keys = list(key_to_atts.keys())
    batch_size = 500

    with _open_db(self._sqlite_path) as conn:
        for i in range(0, len(query_keys), batch_size):
            batch = query_keys[i : i + batch_size]
            placeholders = ",".join("?" for _ in batch)
            sql = f"""
                SELECT key, COALESCE(clientDateModified, dateModified) AS modified
                FROM items
                WHERE key IN ({placeholders})
                  AND itemID NOT IN (SELECT itemID FROM deletedItems)
            """
            rows = conn.execute(sql, batch).fetchall()
            for row in rows:
                q_key = row["key"]
                mod_str = row["modified"]
                if not mod_str:
                    continue
                mod_dt = _parse_utc_datetime(mod_str)
                if not mod_dt:
                    continue

                for att_key in key_to_atts.get(q_key, []):
                    conv_str = converted_map.get(att_key)
                    if not conv_str:
                        continue
                    conv_dt = _parse_utc_datetime(conv_str)
                    if not conv_dt:
                        continue

                    if mod_dt > conv_dt:
                        stale_item_keys.add(att_key)

    return sorted(stale_item_keys)

get_standalone_attachments ¶

get_standalone_attachments(content_types=('application/pdf', 'text/html', 'text/markdown', 'text/plain'))

Return standalone file attachments (attachments without a parent item).

Parameters:

Name	Type	Description	Default
`content_types`	`tuple[str, ...]`	Tuple of MIME types to include.	`('application/pdf', 'text/html', 'text/markdown', 'text/plain')`

Returns:

Type	Description
`list[ZoteroAttachment]`	List of `ZoteroAttachment` objects whose `parent_item_id` is `None`.

Source code in zocrates/rag/zotero_db.py

def get_standalone_attachments(
    self,
    content_types: tuple[str, ...] = ("application/pdf", "text/html", "text/markdown", "text/plain"),
) -> list[ZoteroAttachment]:
    """Return standalone file attachments (attachments without a parent item).

    Args:
        content_types: Tuple of MIME types to include.

    Returns:
        List of ``ZoteroAttachment`` objects whose ``parent_item_id`` is ``None``.
    """
    sql = """
        SELECT
            i.itemID,
            i.key,
            ia.parentItemID,
            parentItem.key AS parentKey,
            COALESCE(ia.contentType, '') AS contentType,
            COALESCE(ia.path, '') AS path
        FROM items i
        JOIN itemAttachments ia ON i.itemID = ia.itemID
        LEFT JOIN items parentItem ON ia.parentItemID = parentItem.itemID
        WHERE ia.parentItemID IS NULL
          AND i.itemID NOT IN (SELECT itemID FROM deletedItems)
    """
    with _open_db(self._sqlite_path) as conn:
        rows = conn.execute(sql).fetchall()

    attachments: list[ZoteroAttachment] = []
    for row in rows:
        if row["contentType"] not in content_types:
            continue
        resolved = self._resolve_attachment_path(row["key"], row["path"])
        attachments.append(
            ZoteroAttachment(
                item_id=row["itemID"],
                item_key=row["key"],
                parent_item_id=row["parentItemID"],
                parent_item_key=row["parentKey"],
                content_type=row["contentType"],
                path=resolved,
            )
        )
    return attachments

get_standalone_notes ¶

get_standalone_notes()

Return top-level standalone notes (notes without a parent item).

Returns:

Type	Description
`list[ZoteroNote]`	List of `ZoteroNote` objects whose `parent_item_id` is `None`.

Source code in zocrates/rag/zotero_db.py

def get_standalone_notes(self) -> list[ZoteroNote]:
    """Return top-level standalone notes (notes without a parent item).

    Returns:
        List of ``ZoteroNote`` objects whose ``parent_item_id`` is ``None``.
    """
    sql = """
        SELECT
            i.itemID AS noteID,
            i.key AS noteKey,
            n.parentItemID,
            parentItem.key AS parentKey,
            COALESCE(n.note, '') AS noteHtml,
            COALESCE(n.title, '') AS noteTitle
        FROM items i
        JOIN itemTypes it ON i.itemTypeID = it.itemTypeID
        JOIN itemNotes n ON i.itemID = n.itemID
        LEFT JOIN items parentItem ON n.parentItemID = parentItem.itemID
        WHERE it.typeName = 'note'
          AND n.parentItemID IS NULL
          AND i.itemID NOT IN (SELECT itemID FROM deletedItems)
        ORDER BY i.itemID
    """
    with _open_db(self._sqlite_path) as conn:
        rows = conn.execute(sql).fetchall()

    notes: list[ZoteroNote] = []
    for row in rows:
        notes.append(
            ZoteroNote(
                note_id=row["noteID"],
                note_key=row["noteKey"],
                parent_item_id=row["parentItemID"],
                parent_item_key=row["parentKey"],
                title=row["noteTitle"],
                note_html=row["noteHtml"],
            )
        )
    return notes

invalidate_schema_cache ¶

invalidate_schema_cache()

Clear cached table existence checks to re-probe schema on next access.

Source code in zocrates/rag/zotero_db.py

def invalidate_schema_cache(self) -> None:
    """Clear cached table existence checks to re-probe schema on next access."""
    self._table_exists_cache.clear()

is_item_active ¶

is_item_active(item_key)

Check if a single item key exists and is not deleted (or parent-deleted).

Parameters:

Name	Type	Description	Default
`item_key`	`str`	Zotero item key.	required

Returns:

Type	Description
`bool`	True if the item exists and is active, False otherwise.

Source code in zocrates/rag/zotero_db.py

def is_item_active(self, item_key: str) -> bool:
    """Check if a single item key exists and is not deleted (or parent-deleted).

    Args:
        item_key: Zotero item key.

    Returns:
        True if the item exists and is active, False otherwise.
    """
    if not item_key:
        return False
    return item_key in self.get_active_item_keys([item_key])

ZoteroNote `dataclass` ¶

ZoteroNote(note_id, note_key, parent_item_id, parent_item_key, title, note_html)

A Zotero note item (standalone or attached to a parent item).

chunk_markdown ¶

chunk_markdown(markdown, max_chars=CHUNK_MAX_CHARS, overlap_chars=CHUNK_OVERLAP_CHARS, document_title='')

Chunk a Markdown document into semantically coherent pieces.

Splits by ATX headers (H1–H3) first, isolates figures and tables as standalone atomic chunks with outer

tags stripped, stitches interrupted narrative prose across figure boundaries, and recursively splits oversized narrative prose by paragraph and sentence up to max_chars.

Preserves document order via anchor interleaving.

Parameters:

Name	Type	Description	Default
`markdown`	`str`	Raw Markdown string to chunk.	required
`max_chars`	`int`	Maximum character count per chunk. Defaults to `CHUNK_MAX_CHARS` from config.	`CHUNK_MAX_CHARS`
`overlap_chars`	`int`	Characters repeated at chunk boundaries for context continuity. Defaults to `CHUNK_OVERLAP_CHARS` from config.	`CHUNK_OVERLAP_CHARS`
`document_title`	`str`	Zotero item title attached to every `ChunkResult`.	`''`

Returns:

Type	Description
`list[ChunkResult]`	Ordered list of `ChunkResult` objects, one per non-empty sub-chunk.

Source code in zocrates/rag/chunker.py

def chunk_markdown(
    markdown: str,
    max_chars: int = CHUNK_MAX_CHARS,
    overlap_chars: int = CHUNK_OVERLAP_CHARS,
    document_title: str = "",
) -> list[ChunkResult]:
    """
    Chunk a Markdown document into semantically coherent pieces.

    Splits by ATX headers (H1–H3) first, isolates figures and tables as
    standalone atomic chunks with outer <figure> tags stripped, stitches
    interrupted narrative prose across figure boundaries, and recursively
    splits oversized narrative prose by paragraph and sentence up to ``max_chars``.

    Preserves document order via anchor interleaving.

    Args:
        markdown: Raw Markdown string to chunk.
        max_chars: Maximum character count per chunk. Defaults to
            ``CHUNK_MAX_CHARS`` from config.
        overlap_chars: Characters repeated at chunk boundaries for context
            continuity. Defaults to ``CHUNK_OVERLAP_CHARS`` from config.
        document_title: Zotero item title attached to every ``ChunkResult``.

    Returns:
        Ordered list of ``ChunkResult`` objects, one per non-empty sub-chunk.
    """
    # 0. Build line index on verbatim raw markdown before any masking or splitting
    line_index = LineIndex(markdown)

    fm_block, remaining_body = _extract_leading_frontmatter(markdown)
    chunks: list[ChunkResult] = []
    idx = 0

    if fm_block:
        fm_match = _FRONTMATTER_RE.match(markdown)
        if fm_match:
            fm_raw = fm_match.group(0)
            fm_end_line = len(fm_raw.rstrip("\r\n").splitlines())
        else:
            fm_end_line = len(fm_block.rstrip("\r\n").splitlines())

        chunks.append(
            ChunkResult(
                text=fm_block,
                heading_path=[],
                chunk_index=idx,
                document_title=document_title,
                chunk_type="frontmatter",
                line_start=1,
                line_end=max(1, fm_end_line),
            )
        )
        idx += 1

    sections = _split_by_headers(remaining_body)
    doc_search_cursor = 0

    for heading_path, body in sections:
        if not body.strip():
            continue

        # 1. Mask fenced code blocks before scanning for figures
        code_masked_body, code_masks = _mask_code_blocks(body)

        # 2. Extract all <figure> blocks and record their anchor positions
        extracted_figures: list[_ExtractedFigure] = []
        for match in _FIGURE_RE.finditer(code_masked_body):
            raw_fig_tag = match.group(0)
            anchor_pos = match.start()

            # Unmask any code placeholders inside figure if present
            unmasked_fig_tag = _unmask_code_blocks(raw_fig_tag, code_masks)
            orig_verbatim_fig_tag = unmasked_fig_tag

            # Determine type from attributes
            if re.search(r'data-type=["\']table["\']', unmasked_fig_tag, re.IGNORECASE):
                fig_type = "table"
            else:
                fig_type = "figure"

            # Extract image path if present
            image_path = None
            if fig_type == "figure":
                md_img = _MD_IMG_RE.search(unmasked_fig_tag)
                html_img = _HTML_IMG_RE.search(unmasked_fig_tag)
                if md_img:
                    image_path = _normalize_image_path(md_img.group(2))
                elif html_img:
                    image_path = _normalize_image_path(html_img.group(1))

            # Extract synthetic caption and determine caption source & model
            caption_source = None
            vlm_model = None
            synth_match = _SYNTH_CAPTION_RE.search(unmasked_fig_tag)
            if synth_match:
                attrs = synth_match.group(1)
                synth_text = synth_match.group(2).strip()

                src_attr = re.search(r'data-source=[\"\']([^\"\']+)[\"\']', attrs, re.IGNORECASE)
                if src_attr:
                    caption_source = src_attr.group(1)
                else:
                    if any(cue in synth_text for cue in ("**Type**:", "Type:", "**Key Elements**:", "**Takeaway**:", "Takeaway:")):
                        caption_source = "vlm"
                    else:
                        caption_source = "datalab"

                model_attr = re.search(r'data-model=[\"\']([^\"\']+)[\"\']', attrs, re.IGNORECASE)
                if model_attr:
                    vlm_model = model_attr.group(1)

                # Format synthetic caption into clean Markdown [Visual Summary]
                unmasked_fig_tag = _SYNTH_CAPTION_RE.sub(
                    lambda m: f"\n\n[Visual Summary]:\n{m.group(2).strip()}", unmasked_fig_tag
                )

            # Determine figure start/end line in the source markdown using the verbatim tag
            fig_start_char = markdown.find(orig_verbatim_fig_tag, doc_search_cursor)
            if fig_start_char == -1:
                fig_start_char = markdown.find(orig_verbatim_fig_tag)
            if fig_start_char != -1:
                fig_end_char = fig_start_char + len(orig_verbatim_fig_tag)
                fig_l_start, fig_l_end = line_index.span_to_lines(fig_start_char, fig_end_char)
            else:
                fig_l_start = 1
                fig_l_end = 1
            fig_l_end = max(fig_l_start, fig_l_end)

            # Strip all HTML wrapper tags from the figure chunk text
            inner_content = re.sub(
                r"^<figure\b[^>]*>", "", unmasked_fig_tag, flags=re.IGNORECASE
            ).strip()
            inner_content = re.sub(
                r"</figure>$", "", inner_content, flags=re.IGNORECASE
            ).strip()
            inner_content = re.sub(
                r"</?figcaption\b[^>]*>", "", inner_content, flags=re.IGNORECASE
            ).strip()
            inner_content = re.sub(
                r"</?synthetic_caption\b[^>]*>", "", inner_content, flags=re.IGNORECASE
            ).strip()

            if inner_content:
                clean_inner = re.sub(r"\n{3,}", "\n\n", inner_content).strip()
                extracted_figures.append(
                    _ExtractedFigure(
                        anchor=anchor_pos,
                        text=clean_inner,
                        chunk_type=fig_type,
                        image_path=image_path if fig_type == "figure" else None,
                        caption_source=caption_source if fig_type == "figure" else None,
                        vlm_model=vlm_model if fig_type == "figure" else None,
                        line_start=fig_l_start,
                        line_end=fig_l_end,
                    )
                )

        # 3. Stitch narrative prose by replacing figures with empty string and collapsing redundant blank lines
        stitched_masked_body = _FIGURE_RE.sub("", code_masked_body)
        stitched_masked_body = re.sub(r"\n{3,}", "\n\n", stitched_masked_body)
        stitched_body = _unmask_code_blocks(stitched_masked_body, code_masks).strip()

        # 4. Split stitched narrative prose into size-bounded sub-chunks
        prose_sub_chunks: list[str] = []
        if stitched_body:
            prose_sub_chunks = _recursive_split(stitched_body, max_chars, overlap_chars)

        # 5. Interleave prose chunks and figure chunks to preserve document flow
        pending_figs = list(extracted_figures)
        search_cursor = 0
        is_first_prose = True

        for sub in prose_sub_chunks:
            sub_str = re.sub(r"\n{3,}", "\n\n", sub.strip()).strip()
            if not sub_str:
                continue

            # Locate position of this prose chunk in code_masked_body
            non_empty_lines = [l.strip() for l in sub_str.splitlines() if l.strip()]
            first_line = non_empty_lines[0][:40] if non_empty_lines else ""
            last_line = non_empty_lines[-1][-40:] if non_empty_lines else ""

            prose_start = (
                code_masked_body.find(first_line, search_cursor)
                if first_line
                else search_cursor
            )
            if prose_start == -1:
                prose_start = search_cursor
            else:
                search_cursor = prose_start

            prose_end = (
                code_masked_body.find(last_line, search_cursor)
                if last_line
                else search_cursor
            )
            if prose_end == -1:
                prose_end = prose_start + len(sub_str)
            else:
                prose_end += len(last_line)
                search_cursor = prose_end

            # Map prose start/end lines in verbatim markdown
            p_start = (
                markdown.find(first_line, doc_search_cursor)
                if first_line
                else doc_search_cursor
            )
            if p_start == -1 and first_line:
                p_start = markdown.find(first_line)
            if p_start == -1:
                p_start = doc_search_cursor

            p_end = (
                markdown.find(last_line, p_start)
                if last_line
                else p_start
            )
            if p_end == -1 and last_line:
                p_end = markdown.find(last_line)
            if p_end == -1:
                p_end = p_start + len(sub_str)
            else:
                p_end = p_end + len(last_line)

            p_line_start, p_line_end = line_index.span_to_lines(p_start, p_end)

            # If this is the first prose chunk under an ATX heading, encompass the heading line
            if is_first_prose and heading_path:
                h_matches = list(
                    re.finditer(
                        r"^(#{1,6})\s+" + re.escape(heading_path[-1]),
                        markdown[:p_start],
                        re.MULTILINE,
                    )
                )
                if h_matches:
                    h_line = line_index.char_to_line(h_matches[-1].start())
                    p_line_start = min(p_line_start, h_line)
                is_first_prose = False

            p_line_end = max(p_line_start, p_line_end)
            doc_search_cursor = max(doc_search_cursor, p_start + 1)

            # Emit figures that appeared before or at the start of this prose chunk
            while pending_figs and pending_figs[0].anchor <= prose_start:
                fig = pending_figs.pop(0)
                chunks.append(
                    ChunkResult(
                        text=fig.text,
                        heading_path=heading_path,
                        chunk_index=idx,
                        document_title=document_title,
                        chunk_type=fig.chunk_type,
                        image_path=fig.image_path,
                        caption_source=fig.caption_source,
                        vlm_model=fig.vlm_model,
                        line_start=fig.line_start,
                        line_end=fig.line_end,
                    )
                )
                idx += 1

            # Emit the prose chunk
            prose_type = determine_chunk_type(sub_str)
            prose_img_path = None
            prose_cap_source = None
            if prose_type == "figure":
                md_img = _MD_IMG_RE.search(sub_str)
                html_img = _HTML_IMG_RE.search(sub_str)
                if md_img:
                    prose_img_path = _normalize_image_path(md_img.group(2))
                elif html_img:
                    prose_img_path = _normalize_image_path(html_img.group(1))

            chunks.append(
                ChunkResult(
                    text=sub_str,
                    heading_path=heading_path,
                    chunk_index=idx,
                    document_title=document_title,
                    chunk_type=prose_type,
                    image_path=prose_img_path,
                    caption_source=prose_cap_source,
                    line_start=p_line_start,
                    line_end=p_line_end,
                )
            )
            idx += 1

            # Emit figures that appeared within this prose chunk span
            while pending_figs and pending_figs[0].anchor <= prose_end:
                fig = pending_figs.pop(0)
                chunks.append(
                    ChunkResult(
                        text=fig.text,
                        heading_path=heading_path,
                        chunk_index=idx,
                        document_title=document_title,
                        chunk_type=fig.chunk_type,
                        image_path=fig.image_path,
                        caption_source=fig.caption_source,
                        vlm_model=fig.vlm_model,
                        line_start=fig.line_start,
                        line_end=fig.line_end,
                    )
                )
                idx += 1

        # Emit any trailing figures in the section
        while pending_figs:
            fig = pending_figs.pop(0)
            chunks.append(
                ChunkResult(
                    text=fig.text,
                    heading_path=heading_path,
                    chunk_index=idx,
                    document_title=document_title,
                    chunk_type=fig.chunk_type,
                    image_path=fig.image_path,
                    caption_source=fig.caption_source,
                    vlm_model=fig.vlm_model,
                    line_start=fig.line_start,
                    line_end=fig.line_end,
                )
            )
            idx += 1

    return chunks

convert_note_html_to_markdown ¶

convert_note_html_to_markdown(html_content)

Convert Zotero note HTML content to clean, structured Markdown.

Handles headings (h1–h6), paragraphs, ordered/unordered lists, blockquotes, pre/code blocks, tables, inline strong/em/code, anchors, and horizontal rules. Falls back to basic regex stripping if BeautifulSoup is unavailable.

Source code in zocrates/rag/document_processor.py

def convert_note_html_to_markdown(html_content: str) -> str:
    """Convert Zotero note HTML content to clean, structured Markdown.

    Handles headings (h1–h6), paragraphs, ordered/unordered lists, blockquotes,
    pre/code blocks, tables, inline strong/em/code, anchors, and horizontal
    rules.  Falls back to basic regex stripping if BeautifulSoup is unavailable.
    """
    if not html_content or not html_content.strip():
        return ""

    import html as html_module

    try:
        from bs4 import BeautifulSoup, NavigableString, Tag

        soup = BeautifulSoup(html_content, "html.parser")

        def _node_to_md(node: Any) -> str:  # noqa: ANN401
            """Recursively convert a BeautifulSoup DOM node into markdown text.

            Args:
                node: NavigableString or Tag DOM node to convert.

            Returns:
                Markdown formatted string corresponding to the node.
            """
            if isinstance(node, NavigableString):
                return html_module.unescape(str(node))
            if not isinstance(node, Tag):
                return ""

            tag = node.name.lower() if node.name else ""
            children_md = "".join(_node_to_md(c) for c in node.children)

            if tag in ("h1", "h2", "h3", "h4", "h5", "h6"):
                level = int(tag[1])
                return f"\n\n{'#' * level} {children_md.strip()}\n\n"
            if tag == "p":
                return f"\n\n{children_md.strip()}\n\n"
            if tag == "br":
                return "\n"
            if tag in ("strong", "b"):
                inner = children_md.strip()
                return f"**{inner}**" if inner else ""
            if tag in ("em", "i"):
                inner = children_md.strip()
                return f"*{inner}*" if inner else ""
            if tag == "code":
                parent_tag = node.parent.name.lower() if node.parent and node.parent.name else ""
                if parent_tag == "pre":
                    return children_md  # <pre> wrapper handles fencing
                return f"`{children_md.strip()}`"
            if tag == "pre":
                inner = node.get_text()
                return f"\n\n```\n{inner.rstrip()}\n```\n\n"
            if tag == "blockquote":
                inner = children_md.strip()
                quoted = "\n".join(f"> {line}" for line in inner.splitlines())
                return f"\n\n{quoted}\n\n"
            if tag == "ul":
                return f"\n\n{children_md.strip()}\n\n"
            if tag == "ol":
                items_md: list[str] = []
                counter = 1
                for child in node.children:
                    if isinstance(child, Tag) and child.name.lower() == "li":
                        items_md.append(f"{counter}. {''.join(_node_to_md(gc) for gc in child.children).strip()}")
                        counter += 1
                return "\n\n" + "\n".join(items_md) + "\n\n"
            if tag == "li":
                parent_tag = node.parent.name.lower() if node.parent and node.parent.name else ""
                if parent_tag == "ol":
                    return children_md.strip()  # ol handler above owns numbering
                return f"- {children_md.strip()}\n"
            if tag == "table":
                rows: list[str] = []
                for tr in node.find_all("tr"):
                    cells = [
                        cell.get_text().strip().replace("|", "\\|")
                        for cell in tr.find_all(["th", "td"])
                    ]
                    if cells:
                        rows.append("| " + " | ".join(cells) + " |")
                if not rows:
                    return children_md
                col_count = rows[0].count("|") - 1
                sep = "| " + " | ".join(["---"] * col_count) + " |"
                rows.insert(1, sep)
                return "\n\n" + "\n".join(rows) + "\n\n"
            if tag == "a":
                href = node.get("href", "")
                label = children_md.strip()
                return f"[{label}]({href})" if href else label
            if tag in ("hr",):
                return "\n\n---\n\n"
            # Pass-through containers
            return children_md

        result = _node_to_md(soup)

    except Exception:
        result = re.sub(r"<br\s*/?>", "\n", html_content, flags=re.IGNORECASE)
        result = re.sub(r"</p>", "\n\n", result, flags=re.IGNORECASE)
        result = re.sub(r"</li>", "\n", result, flags=re.IGNORECASE)
        result = re.sub(r"<[^>]+>", "", result)
        result = html_module.unescape(result)

    lines = [line.rstrip() for line in result.splitlines()]
    clean_text = "\n".join(lines).strip()
    return re.sub(r"\n{3,}", "\n\n", clean_text)

get_chroma_dir ¶

get_chroma_dir(zotero_data_dir)

Return (and create if needed) the ChromaDB storage directory.

Parameters:

Name	Type	Description	Default
`zotero_data_dir`	`Path`	Root Zotero data directory.	required

Returns:

Type	Description
`Path`	Path to `<zocrates_data>/chroma/`, created if absent.

Source code in zocrates/config.py

def get_chroma_dir(zotero_data_dir: Path) -> Path:
    """Return (and create if needed) the ChromaDB storage directory.

    Args:
        zotero_data_dir: Root Zotero data directory.

    Returns:
        Path to ``<zocrates_data>/chroma/``, created if absent.
    """
    zocrates_dir = get_zocrates_data_dir(zotero_data_dir)
    chroma_dir = zocrates_dir / "chroma"
    chroma_dir.mkdir(parents=True, exist_ok=True)
    return chroma_dir

get_corpus_dir ¶

get_corpus_dir(zotero_data_dir)

Return (and create if needed) the document corpus directory.

Parameters:

Name	Type	Description	Default
`zotero_data_dir`	`Path`	Root Zotero data directory.	required

Returns:

Type	Description
`Path`	Path to `<zocrates_data>/corpus/`, created if absent.

Source code in zocrates/config.py

def get_corpus_dir(zotero_data_dir: Path) -> Path:
    """Return (and create if needed) the document corpus directory.

    Args:
        zotero_data_dir: Root Zotero data directory.

    Returns:
        Path to ``<zocrates_data>/corpus/``, created if absent.
    """
    zocrates_dir = get_zocrates_data_dir(zotero_data_dir)
    legacy_dir = zocrates_dir / "markdown_cache"
    corpus_dir = zocrates_dir / "corpus"
    if legacy_dir.is_dir() and not corpus_dir.exists():
        try:
            legacy_dir.rename(corpus_dir)
            logger.info("Migrated legacy markdown_cache to %s", corpus_dir)
        except OSError as exc:
            logger.warning(
                "Failed to automatically migrate legacy markdown_cache to corpus: %s. "
                "Please rename '%s' to '%s' manually.",
                exc,
                legacy_dir,
                corpus_dir,
            )
    elif legacy_dir.is_dir() and corpus_dir.is_dir():
        logger.debug("Both corpus/ and legacy markdown_cache/ exist. Using %s.", corpus_dir)
    corpus_dir.mkdir(parents=True, exist_ok=True)
    return corpus_dir

get_datalab_non_pdf_fallback ¶

get_datalab_non_pdf_fallback(zotero_data_dir=None, cfg=None)

Return the fallback converter ID for non-PDF items when Datalab is active ('markitdown' or 'docling_local').

Checks DATALAB_NON_PDF_FALLBACK environment variable first; if unset, falls back to datalab_non_pdf_fallback in zocrates_config.json (defaulting to "markitdown").

Parameters:

Name	Type	Description	Default
`zotero_data_dir`	`Path \| None`	Path to the Zotero data directory.	`None`
`cfg`	`dict[str, Any] \| None`	Optional in-memory config dict. If provided, bypasses disk read.	`None`

Returns:

Type	Description
`str`	Fallback converter ID string (`"markitdown"` or `"docling_local"`).

Source code in zocrates/config.py

def get_datalab_non_pdf_fallback(
    zotero_data_dir: Path | None = None,
    cfg: dict[str, Any] | None = None,
) -> str:
    """Return the fallback converter ID for non-PDF items when Datalab is active ('markitdown' or 'docling_local').

    Checks ``DATALAB_NON_PDF_FALLBACK`` environment variable first;
    if unset, falls back to ``datalab_non_pdf_fallback`` in ``zocrates_config.json``
    (defaulting to ``"markitdown"``).

    Args:
        zotero_data_dir: Path to the Zotero data directory.
        cfg: Optional in-memory config dict. If provided, bypasses disk read.

    Returns:
        Fallback converter ID string (``"markitdown"`` or ``"docling_local"``).
    """
    env_val = os.getenv("DATALAB_NON_PDF_FALLBACK", "")
    if env_val:
        return env_val.lower()
    if cfg is None:
        cfg = load_zocrates_config(zotero_data_dir)
    return cfg.get("datalab_non_pdf_fallback", "markitdown").lower()

get_document_converter_id ¶

get_document_converter_id(zotero_data_dir=None, cfg=None)

Return the document converter ID from the Zocrates config.

Checks DOCUMENT_CONVERTER_ID environment variable first; if unset, falls back to document_converter_id in zocrates_config.json (defaulting to "markitdown").

Parameters:

Name	Type	Description	Default
`zotero_data_dir`	`Path \| None`	Path to the Zotero data directory.	`None`
`cfg`	`dict[str, Any] \| None`	Optional in-memory config dict. If provided, bypasses disk read.	`None`

Returns:

Type	Description
`str`	Converter ID string (e.g., `"markitdown"`, `"docling_local"`).

Source code in zocrates/config.py

def get_document_converter_id(
    zotero_data_dir: Path | None = None,
    cfg: dict[str, Any] | None = None,
) -> str:
    """Return the document converter ID from the Zocrates config.

    Checks ``DOCUMENT_CONVERTER_ID`` environment variable first;
    if unset, falls back to ``document_converter_id`` in ``zocrates_config.json``
    (defaulting to ``"markitdown"``).

    Args:
        zotero_data_dir: Path to the Zotero data directory.
        cfg: Optional in-memory config dict. If provided, bypasses disk read.

    Returns:
        Converter ID string (e.g., ``"markitdown"``, ``"docling_local"``).
    """
    env_val = os.getenv("DOCUMENT_CONVERTER_ID", "")
    if env_val:
        return env_val
    if cfg is None:
        cfg = load_zocrates_config(zotero_data_dir)
    return cfg.get("document_converter_id", "markitdown")

get_embedder ¶

get_embedder(mode=None)

Return the configured embedding function.

Reads the active mode from the EMBEDDING_MODE environment variable, falling back to the persisted config file value, then to "local".

Parameters:

Name	Type	Description	Default
`mode`	`str \| None`	Override for the embedding mode (`"local"` or `"cloud"`). If `None`, the mode is determined from the environment and the Zocrates config file.	`None`

Returns:

Type	Description
`LocalEmbedder \| OpenRouterEmbedder`	A `LocalEmbedder` for CPU-based Jina embeddings, or an
`LocalEmbedder \| OpenRouterEmbedder`	`OpenRouterEmbedder` for OpenRouter cloud embeddings.

Source code in zocrates/rag/embeddings.py

def get_embedder(
    mode: str | None = None,
) -> LocalEmbedder | OpenRouterEmbedder:
    """Return the configured embedding function.

    Reads the active mode from the ``EMBEDDING_MODE`` environment variable,
    falling back to the persisted config file value, then to ``"local"``.

    Args:
        mode: Override for the embedding mode (``"local"`` or ``"cloud"``).
            If ``None``, the mode is determined from the environment and
            the Zocrates config file.

    Returns:
        A ``LocalEmbedder`` for CPU-based Jina embeddings, or an
        ``OpenRouterEmbedder`` for OpenRouter cloud embeddings.
    """
    from zocrates.config import load_zocrates_config

    cfg = load_zocrates_config()
    active_mode = mode or os.getenv(
        "EMBEDDING_MODE",
        cfg.get("embedding_provider", "local"),
    )

    if active_mode == "cloud":
        logger.info("Using cloud embedding mode (OpenRouter).")
        return OpenRouterEmbedder()

    logger.info("Using local CPU embedding mode with model jinaai/jina-embeddings-v5-text-nano (768d).")
    return LocalEmbedder()

get_fts_db_path ¶

get_fts_db_path(zotero_data_dir)

Return the path to the FTS5 SQLite database.

Parameters:

Name	Type	Description	Default
`zotero_data_dir`	`Path`	Root Zotero data directory.	required

Returns:

Type	Description
`Path`	Path to `<zocrates_data>/fts_index.db`.

Source code in zocrates/config.py

def get_fts_db_path(zotero_data_dir: Path) -> Path:
    """Return the path to the FTS5 SQLite database.

    Args:
        zotero_data_dir: Root Zotero data directory.

    Returns:
        Path to ``<zocrates_data>/fts_index.db``.
    """
    zocrates_dir = get_zocrates_data_dir(zotero_data_dir)
    return zocrates_dir / "fts_index.db"

get_sync_auto_abort_failures ¶

get_sync_auto_abort_failures(zotero_data_dir=None, cfg=None)

Return the consecutive-failure auto-abort threshold for full-library sync jobs.

Reads sync_auto_abort_failures from zocrates_config.json (defaulting to DEFAULT_SYNC_AUTO_ABORT_FAILURES). Values <= 0 disable auto-abort.

Parameters:

Name	Type	Description	Default
`zotero_data_dir`	`Path \| None`	Path to the Zotero data directory.	`None`
`cfg`	`dict[str, Any] \| None`	Optional in-memory config dict. If provided, bypasses disk read.	`None`

Source code in zocrates/config.py

def get_sync_auto_abort_failures(
    zotero_data_dir: Path | None = None,
    cfg: dict[str, Any] | None = None,
) -> int:
    """Return the consecutive-failure auto-abort threshold for full-library sync jobs.

    Reads ``sync_auto_abort_failures`` from ``zocrates_config.json`` (defaulting
    to ``DEFAULT_SYNC_AUTO_ABORT_FAILURES``). Values <= 0 disable auto-abort.

    Args:
        zotero_data_dir: Path to the Zotero data directory.
        cfg: Optional in-memory config dict. If provided, bypasses disk read.
    """
    if cfg is None:
        cfg = load_zocrates_config(zotero_data_dir)
    try:
        return int(cfg.get("sync_auto_abort_failures", DEFAULT_SYNC_AUTO_ABORT_FAILURES))
    except (ValueError, TypeError):
        return DEFAULT_SYNC_AUTO_ABORT_FAILURES

materialize_note_bundle ¶

materialize_note_bundle(note, syncer, cache_dir=None)

Create or refresh an OKF bundle for a Zotero note in the markdown cache.

The bundle is placed at <cache_dir>/<note.note_key>/ and contains document.md, toc.json, and toc.md. Staleness is detected via a SHA-256 hash of the raw HTML stored in conversion_meta.db.

Summary policy: never calls an LLM. - Notes < 100 plain-text words → no abstract. - Notes ≥ 100 plain-text words → deterministic extractive lead paragraph (first non-heading paragraph, capped at 300 characters).

Thread-safe: uses a process-wide lock and an atomic directory swap (write to .tmp_{key} then rename).

Parameters:

Name	Type	Description	Default
`note`	`Any`	A `ZoteroNote` dataclass instance.	required
`syncer`	`Any`	The active `LibrarySyncer` instance (provides `processor`).	required
`cache_dir`	`Path \| None`	Override for the cache directory; defaults to `syncer.processor._cache_dir`.	`None`

Returns:

Type	Description
`tuple[Path, str]`	Tuple of `(note_dir, note.note_key)`.

Source code in zocrates/rag/document_processor.py

def materialize_note_bundle(
    note: Any,
    syncer: Any,
    cache_dir: Path | None = None,
) -> tuple[Path, str]:
    """Create or refresh an OKF bundle for a Zotero note in the markdown cache.

    The bundle is placed at ``<cache_dir>/<note.note_key>/`` and contains
    ``document.md``, ``toc.json``, and ``toc.md``.  Staleness is detected via
    a SHA-256 hash of the raw HTML stored in ``conversion_meta.db``.

    Summary policy: **never calls an LLM**.
    - Notes < 100 plain-text words → no abstract.
    - Notes ≥ 100 plain-text words → deterministic extractive lead paragraph
      (first non-heading paragraph, capped at 300 characters).

    Thread-safe: uses a process-wide lock and an atomic directory swap
    (write to ``.tmp_{key}`` then rename).

    Args:
        note: A ``ZoteroNote`` dataclass instance.
        syncer: The active ``LibrarySyncer`` instance (provides ``processor``).
        cache_dir: Override for the cache directory; defaults to
            ``syncer.processor._cache_dir``.

    Returns:
        Tuple of ``(note_dir, note.note_key)``.
    """
    effective_cache_dir: Path = cache_dir if cache_dir is not None else syncer.processor._cache_dir
    note_dir = effective_cache_dir / note.note_key
    current_hash = hashlib.sha256(note.note_html.encode("utf-8")).hexdigest()

    # Fast path: bundle exists and HTML hasn't changed (checked outside lock).
    if note_dir.exists() and (note_dir / "document.md").exists():
        existing_meta = syncer.processor.get_metadata(note.note_key)
        if existing_meta and existing_meta.get("source_hash") == current_hash:
            return note_dir, note.note_key

    with _note_bundle_lock:
        # Re-check inside lock (double-checked locking pattern).
        if note_dir.exists() and (note_dir / "document.md").exists():
            existing_meta = syncer.processor.get_metadata(note.note_key)
            if existing_meta and existing_meta.get("source_hash") == current_hash:
                return note_dir, note.note_key

        title = (note.title or "Note").strip()
        markdown = convert_note_html_to_markdown(note.note_html)

        # Plain-text word count (strip all HTML tags for accurate count).
        plain_text = re.sub(r"<[^>]+>", " ", note.note_html)
        word_count = len(plain_text.split())

        # Strictly extractive summary — no LLM calls, ever.
        effective_abstract: str | None = None
        effective_abstract_source: str = "none"
        if word_count >= 100:
            paragraphs = [p.strip() for p in re.split(r"\n{2,}", markdown) if p.strip()]
            lead = ""
            for p in paragraphs:
                if not p.startswith("#"):  # skip heading lines
                    lead = p
                    break
            if lead:
                effective_abstract = lead[:300] + ("\u2026" if len(lead) > 300 else "")
                effective_abstract_source = "extractive"

        # Build YAML frontmatter.
        parent_key = note.parent_item_key or ""
        fm_lines = [
            "---",
            f'title: "{title}"',
            f'item_key: "{note.note_key}"',
            f'parent_item_key: "{parent_key}"',
            "type: note",
            "source:",
            f'  path: "zotero://note/{note.note_key}"',
            f'  hash: "{current_hash}"',
            '  content_type: "text/note"',
            "---",
        ]
        frontmatter = "\n".join(fm_lines) + "\n"
        full_doc = frontmatter + f"\n# {title}\n\n" + markdown

        # Write to a temp directory then atomically rename.
        tmp_dir = effective_cache_dir / f".tmp_{note.note_key}"
        if tmp_dir.exists():
            shutil.rmtree(tmp_dir, ignore_errors=True)
        tmp_dir.mkdir(parents=True, exist_ok=True)

        (tmp_dir / "document.md").write_text(full_doc, encoding="utf-8")

        try:
            from zocrates.rag.toc_generator import generate_toc, save_toc_files  # local import to avoid cycles

            toc_json, toc_md = generate_toc(
                document_md=full_doc,
                title=title,
                abstract=effective_abstract,
                abstract_source=effective_abstract_source,
                item_key=note.note_key,
                parent_item_key=note.parent_item_key,
            )
            save_toc_files(tmp_dir, toc_json, toc_md)
        except Exception as exc:
            logger.warning("ToC generation failed for note %s: %s", note.note_key, exc)

        # Generate item_relations.md and index.md inside staging directory before atomic rename
        try:
            regenerate_item_relations_md(
                item_key=note.note_key,
                cache_dir=effective_cache_dir,
                target_bundle_dir=tmp_dir,
            )
        except Exception as exc:
            logger.warning("Failed to generate item_relations.md for note %s: %s", note.note_key, exc)
        generate_item_index_md(
            item_dir=tmp_dir,
            item_key=note.note_key,
            title=title,
            converter_id="zotero_note_converter",
            converter_version="1.0",
        )

        # Atomic swap.
        if note_dir.exists():
            shutil.rmtree(note_dir, ignore_errors=True)
        tmp_dir.rename(note_dir)

        # Persist conversion metadata.
        existing_keywords = None
        with _open_meta_db(effective_cache_dir) as conn:
            try:
                existing_row = conn.execute(
                    "SELECT keywords FROM conversion_meta WHERE item_key = ?",
                    (note.note_key,),
                ).fetchone()
                if existing_row and existing_row["keywords"]:
                    existing_keywords = existing_row["keywords"]
            except sqlite3.OperationalError:
                pass

            converted_at = datetime.now(UTC).isoformat()
            conn.execute(
                """
                INSERT OR REPLACE INTO conversion_meta
                    (item_key, title, converter_id, converter_version, source_path,
                     content_type, parent_item_type, converted_at, source_hash, char_count,
                     parent_item_key, parse_quality_score, checkpoint_id, abstract, abstract_source, keywords)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                """,
                (
                    note.note_key,
                    title,
                    "zotero_note_converter",
                    "1.0",
                    f"zotero://note/{note.note_key}",
                    "text/note",
                    "note",
                    converted_at,
                    current_hash,
                    len(full_doc),
                    note.parent_item_key,
                    None,  # parse_quality_score
                    None,  # checkpoint_id
                    effective_abstract,
                    effective_abstract_source,
                    existing_keywords,
                ),
            )

        return note_dir, note.note_key

library_syncer

zocrates.rag.library_syncer ¶

CHUNK_MAX_CHARS module-attribute ¶

CHUNK_OVERLAP_CHARS module-attribute ¶

DEFAULT_SYNC_AUTO_ABORT_FAILURES module-attribute ¶

logger module-attribute ¶

DocumentProcessor ¶

cache_dir property ¶

datalab_non_pdf_fallback property writable ¶

clear_failure ¶

convert_with ¶

delete_cached ¶

ensure_parent_item_key ¶

get_all_meta_keys ¶

get_all_metadata ¶

get_failures ¶

get_metadata ¶

get_or_convert ¶

is_cached ¶

list_cached_keys ¶

read_cached ¶

record_failure ¶

regenerate_root_index ¶

register_converter ¶

update_abstract ¶

update_indexed_at ¶

LibrarySyncer ¶

close ¶

full_sync async ¶

ingest_items async ¶

purge_item ¶

sync_item async ¶

sync_metadata_staleness ¶

VectorStore ¶

fts_path property ¶

acquire_write_lock ¶

close ¶

delete_document ¶

get_all_chunk_ids_for_item ¶

get_all_ingested_item_keys ¶

get_all_item_keys ¶

get_intersecting_chunk_ids ¶

ingest_document ¶

is_ingested ¶

list_collections ¶

reset_collection ¶

search ¶

update_chunk_text_preserve_vector ¶

update_metadata ¶

ZoteroAttachment dataclass ¶

ZoteroDB ¶

get_active_item_keys ¶

get_all_arxiv_ids ¶

get_all_attachments ¶

get_all_convertible_attachments ¶

get_all_dedup_records ¶

get_all_dois ¶

get_all_item_relations ¶

get_all_items ¶

get_all_notes ¶

get_all_titles ¶

get_attachment_by_id ¶

get_attachment_by_key ¶

get_attachments ¶

get_item ¶

get_item_by_key ¶

get_item_date_modified ¶

get_item_keys_in_collection ¶

get_item_relations ¶

get_item_tags ¶

get_items_by_ids ¶

get_new_item_ids_since ¶

get_note_by_id ¶

get_note_by_key ¶

get_notes_for_item ¶

get_stale_item_keys ¶

get_standalone_attachments ¶

get_standalone_notes ¶

invalidate_schema_cache ¶

is_item_active ¶

CHUNK_MAX_CHARS `module-attribute` ¶

CHUNK_OVERLAP_CHARS `module-attribute` ¶

DEFAULT_SYNC_AUTO_ABORT_FAILURES `module-attribute` ¶

logger `module-attribute` ¶

cache_dir `property` ¶

datalab_non_pdf_fallback `property` `writable` ¶

full_sync `async` ¶

ingest_items `async` ¶

sync_item `async` ¶

fts_path `property` ¶

ZoteroAttachment `dataclass` ¶

ZoteroNote `dataclass` ¶