frontmatter_refresher

zocrates.rag.frontmatter_refresher

Frontmatter refresher and metadata synchronization engine for Zocrates.

Synchronizes OKF v0.2 YAML frontmatter in cached Markdown bundles with the latest bibliographic metadata from Zotero (titles, authors, year, publication, DOI, URL) without triggering PDF/HTML reconversion, OCR, or embedding models.

_FRONTMATTER_RE module-attribute

_FRONTMATTER_RE = re.compile('^---\\s*\\r?\\n([\\s\\S]*?)\\r?\\n---\\s*(?:\\r?\\n|$)')

_REFRESH_LOCK module-attribute

_REFRESH_LOCK = threading.Lock()

logger module-attribute

logger = logging.getLogger(__name__)

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

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

_open_fts_db

_open_fts_db(path)

Open (or create) the FTS5 and chunk metadata SQLite databases.

Creates the chunks_fts FTS5 virtual table (trigram tokenizer) and the chunk_meta table if they do not already exist. Migrates missing columns on chunk_meta if opened on a legacy database.

Parameters:

Name	Type	Description	Default
path	Path	Filesystem path to the SQLite database file.	required

Returns:

Type	Description
Connection	An open sqlite3.Connection with row_factory set to
Connection	sqlite3.Row.

Source code in zocrates/rag/vector_store.py

def _open_fts_db(path: Path) -> sqlite3.Connection:
    """Open (or create) the FTS5 and chunk metadata SQLite databases.

    Creates the ``chunks_fts`` FTS5 virtual table (trigram tokenizer) and
    the ``chunk_meta`` table if they do not already exist. Migrates missing
    columns on ``chunk_meta`` if opened on a legacy database.

    Args:
        path: Filesystem path to the SQLite database file.

    Returns:
        An open ``sqlite3.Connection`` with ``row_factory`` set to
        ``sqlite3.Row``.
    """
    conn = sqlite3.connect(str(path), timeout=5.0)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA journal_mode=WAL;")
    conn.execute("PRAGMA busy_timeout = 5000;")

    if path not in _INITIALIZED_FTS_DBS:
        with _FTS_INIT_LOCK:
            if path not in _INITIALIZED_FTS_DBS:
                conn.execute(_FTS_SCHEMA)
                conn.execute(_META_SCHEMA)

                # Check and migrate columns on chunk_meta
                existing_cols = {
                    row["name"] for row in conn.execute("PRAGMA table_info(chunk_meta)").fetchall()
                }
                for col_name in ("image_path", "caption_source", "vlm_model", "line_start", "line_end"):
                    if col_name not in existing_cols:
                        col_type = "INTEGER" if col_name in ("line_start", "line_end") else "TEXT"
                        conn.execute(f"ALTER TABLE chunk_meta ADD COLUMN {col_name} {col_type}")

                conn.commit()
                _INITIALIZED_FTS_DBS.add(path)

    return conn

_open_meta_db

_open_meta_db(cache_dir)

Open (or create) the conversion metadata SQLite database.

Creates the conversion_meta table if it does not already exist. Enables WAL mode for concurrent readers.

Parameters:

Name	Type	Description	Default
cache_dir	Path	Directory containing the conversion_meta.db file.	required

Returns:

Type	Description
Connection	An open sqlite3.Connection with row_factory set to
Connection	sqlite3.Row.

Source code in zocrates/rag/document_processor.py

def _open_meta_db(cache_dir: Path) -> sqlite3.Connection:
    """Open (or create) the conversion metadata SQLite database.

    Creates the ``conversion_meta`` table if it does not already exist.
    Enables WAL mode for concurrent readers.

    Args:
        cache_dir: Directory containing the ``conversion_meta.db`` file.

    Returns:
        An open ``sqlite3.Connection`` with ``row_factory`` set to
        ``sqlite3.Row``.
    """
    cache_dir.mkdir(parents=True, exist_ok=True)
    meta_path = cache_dir / "conversion_meta.db"
    conn = sqlite3.connect(str(meta_path), timeout=5.0)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA busy_timeout = 5000;")

    if cache_dir not in _INITIALIZED_META_DBS:
        with _META_INIT_LOCK:
            if cache_dir not in _INITIALIZED_META_DBS:
                conn.execute("PRAGMA journal_mode=WAL;")
                conn.executescript(_META_SCHEMA)
                # Migrate columns on pre-existing databases if needed
                existing_cols = {
                    row["name"]
                    for row in conn.execute("PRAGMA table_info(conversion_meta)").fetchall()
                }
                if "title" not in existing_cols:
                    try:
                        conn.execute("ALTER TABLE conversion_meta ADD COLUMN title TEXT;")
                    except sqlite3.OperationalError:
                        pass
                if "parse_quality_score" not in existing_cols:
                    try:
                        conn.execute("ALTER TABLE conversion_meta ADD COLUMN parse_quality_score REAL;")
                    except sqlite3.OperationalError:
                        pass
                if "checkpoint_id" not in existing_cols:
                    try:
                        conn.execute("ALTER TABLE conversion_meta ADD COLUMN checkpoint_id TEXT;")
                    except sqlite3.OperationalError:
                        pass
                if "abstract" not in existing_cols:
                    try:
                        conn.execute("ALTER TABLE conversion_meta ADD COLUMN abstract TEXT;")
                    except sqlite3.OperationalError:
                        pass
                if "abstract_source" not in existing_cols:
                    try:
                        conn.execute("ALTER TABLE conversion_meta ADD COLUMN abstract_source TEXT;")
                    except sqlite3.OperationalError:
                        pass
                if "updated_at" not in existing_cols:
                    try:
                        conn.execute("ALTER TABLE conversion_meta ADD COLUMN updated_at TEXT;")
                    except sqlite3.OperationalError:
                        pass
                if "keywords" not in existing_cols:
                    try:
                        conn.execute("ALTER TABLE conversion_meta ADD COLUMN keywords TEXT;")
                    except sqlite3.OperationalError:
                        pass
                try:
                    conn.execute(
                        """
                        CREATE TABLE IF NOT EXISTS item_relations (
                            source_key      TEXT NOT NULL,
                            target_key      TEXT NOT NULL,
                            score           REAL NOT NULL,
                            relation_type   TEXT NOT NULL,
                            match_evidence  TEXT,
                            updated_at      TEXT NOT NULL,
                            PRIMARY KEY (source_key, target_key, relation_type)
                        );
                        """
                    )
                    conn.execute("CREATE INDEX IF NOT EXISTS idx_relations_source_score ON item_relations (source_key, score DESC);")
                    conn.execute("CREATE INDEX IF NOT EXISTS idx_relations_target ON item_relations (target_key);")
                except sqlite3.OperationalError:
                    pass
                conn.commit()

                # Migrate storage paths to relative and sanitize legacy holdover bundles
                meta_path = cache_dir / "conversion_meta.db"
                version_row = conn.execute("PRAGMA user_version;").fetchone()
                current_version = version_row[0] if version_row else 0
                if current_version < 1:
                    _migrate_storage_paths_to_relative(meta_path, conn=conn)
                    _sanitize_cached_bundle_image_links(cache_dir)

                _INITIALIZED_META_DBS.add(cache_dir)

    return conn

build_frontmatter

build_frontmatter(item_key, parent_item_type, title, collections, source_path, source_hash, content_type, converter_id, converter_version, converted_at, parent_item_key=None, parse_quality_score=None, checkpoint_id=None, authors=None, year=None, publication=None, doi=None, url=None, keywords=None, bundle_dir=None)

Build OKF-0.2 YAML frontmatter string.

Args:
    item_key: Zotero item key.
    parent_item_type: Zotero item type (e.g., 'journalArticle'). Defaults to 'document'.
    title: Title of the parent item.
    collections: List of ZoteroCollection objects or dicts with name and key/collection_key.
    source_path: Absolute path to the source file on disk.
    source_hash: SHA-256 hex digest of the source file.
    content_type: MIME type of the source file.
    converter_id: Converter engine ID.
    converter_version: Converter engine version string.
    converted_at: ISO-8601 timestamp string.
    parent_item_key: Parent item key if item is an attachment.
    parse_quality_score: Optional parse quality score (e.g. 0.0 to 5.0 from Datalab).
    checkpoint_id: Optional converter checkpoint ID.
    authors: Optional list of creator/author name strings.
    year: Optional publication year (int or string).
    publication: Optional publication venue or publisher string.
    doi: Optional normalized DOI string.
    url: Optional source URL string.
    bundle_dir: Optional bundle directory (e.g. cache_dir / item_key). If provided,
        source.path is computed as a relative POSIX path from bundle_dir.

Returns:
    Formatted YAML frontmatter block ending with '---

'.

Source code in zocrates/rag/bundle_catalog.py

def build_frontmatter(
    item_key: str,
    parent_item_type: str | None,
    title: str | None,
    collections: list[Any] | None,
    source_path: Path | str,
    source_hash: str,
    content_type: str,
    converter_id: str,
    converter_version: str,
    converted_at: str,
    parent_item_key: str | None = None,
    parse_quality_score: float | None = None,
    checkpoint_id: 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,
    keywords: list[str] | None = None,
    bundle_dir: Path | None = None,
) -> str:
    """Build OKF-0.2 YAML frontmatter string.

    Args:
        item_key: Zotero item key.
        parent_item_type: Zotero item type (e.g., 'journalArticle'). Defaults to 'document'.
        title: Title of the parent item.
        collections: List of ZoteroCollection objects or dicts with name and key/collection_key.
        source_path: Absolute path to the source file on disk.
        source_hash: SHA-256 hex digest of the source file.
        content_type: MIME type of the source file.
        converter_id: Converter engine ID.
        converter_version: Converter engine version string.
        converted_at: ISO-8601 timestamp string.
        parent_item_key: Parent item key if item is an attachment.
        parse_quality_score: Optional parse quality score (e.g. 0.0 to 5.0 from Datalab).
        checkpoint_id: Optional converter checkpoint ID.
        authors: Optional list of creator/author name strings.
        year: Optional publication year (int or string).
        publication: Optional publication venue or publisher string.
        doi: Optional normalized DOI string.
        url: Optional source URL string.
        bundle_dir: Optional bundle directory (e.g. cache_dir / item_key). If provided,
            source.path is computed as a relative POSIX path from bundle_dir.

    Returns:
        Formatted YAML frontmatter block ending with '---\n'.
    """
    formatted_collections: list[dict[str, str]] = []
    if collections:
        for c in collections:
            if isinstance(c, dict):
                c_name = c.get("name", "")
                c_key = c.get("collection_key") or c.get("key", "")
                formatted_collections.append({"name": c_name, "collection_key": c_key})
            elif hasattr(c, "name") and hasattr(c, "key"):
                formatted_collections.append({"name": c.name, "collection_key": c.key})

    provenance_data: dict[str, Any] = {
        "converter_id": converter_id,
        "converter_version": converter_version,
        "converted_at": converted_at,
    }
    if parse_quality_score is not None:
        provenance_data["parse_quality_score"] = parse_quality_score
    if checkpoint_id is not None:
        provenance_data["checkpoint_id"] = checkpoint_id

    data: dict[str, Any] = {
        "type": parent_item_type or "document",
        "okf_version": "0.2",
        "item_key": item_key,
        "parent_item_key": parent_item_key,
        "title": title or "",
    }

    if authors:
        cleaned_authors = [a.strip() for a in authors if isinstance(a, str) and a.strip()]
        if cleaned_authors:
            data["authors"] = cleaned_authors

    if year is not None:
        if isinstance(year, int):
            data["year"] = year
        elif isinstance(year, str) and year.strip():
            if year.strip().isdigit():
                data["year"] = int(year.strip())
            else:
                data["year"] = year.strip()

    if publication and isinstance(publication, str) and publication.strip():
        data["publication"] = publication.strip()

    if doi and isinstance(doi, str) and doi.strip():
        data["doi"] = doi.strip()

    if url and isinstance(url, str) and url.strip():
        data["url"] = url.strip()

    if keywords:
        cleaned_keywords = [k.strip() for k in keywords if isinstance(k, str) and k.strip()]
        if cleaned_keywords:
            data["keywords"] = cleaned_keywords

    data["member_of_collections"] = formatted_collections
    rel_path = (
        _compute_relative_source_path(source_path, bundle_dir)
        if bundle_dir is not None
        else str(source_path)
    )
    data["source"] = {
        "path": rel_path,
        "hash": source_hash,
        "content_type": content_type,
    }
    data["provenance"] = provenance_data
    yaml_str = yaml.dump(data, sort_keys=False, allow_unicode=True).strip()
    return f"---\n{yaml_str}\n---\n"

generate_item_index_md

generate_item_index_md(item_dir, item_key, title, converter_id, converter_version)

Generate item-level index.md catalog inside item_dir.

Catalogs main document as well as all associated image and table assets.

Parameters:

Name	Type	Description	Default
item_dir	Path	Bundle directory (corpus//).	required
item_key	str	Zotero item key.	required
title	str | None	Title of the parent item.	required
converter_id	str	Converter engine ID.	required
converter_version	str	Converter engine version string.	required

Source code in zocrates/rag/bundle_catalog.py

def generate_item_index_md(
    item_dir: Path,
    item_key: str,
    title: str | None,
    converter_id: str,
    converter_version: str,
) -> None:
    """Generate item-level index.md catalog inside item_dir.

    Catalogs main document as well as all associated image and table assets.

    Args:
        item_dir: Bundle directory (corpus/<item_key>/).
        item_key: Zotero item key.
        title: Title of the parent item.
        converter_id: Converter engine ID.
        converter_version: Converter engine version string.
    """
    doc_path = item_dir / "document.md"
    doc_ref = (
        f"- **Main Document**: [{doc_path.name}](./{doc_path.name})"
        if doc_path.exists()
        else "- **Main Document**: None"
    )

    toc_md_path = item_dir / "toc.md"
    toc_json_path = item_dir / "toc.json"
    toc_ref = (
        f"- **Table of Contents**: [{toc_md_path.name}](./{toc_md_path.name}) ([JSON](./{toc_json_path.name}))"
        if toc_md_path.exists()
        else None
    )

    rel_md_path = item_dir / "item_relations.md"
    rel_ref = (
        "- **Related Items**: [item_relations.md](./item_relations.md)"
        if rel_md_path.exists()
        else None
    )

    image_extensions = {".png", ".jpg", ".jpeg", ".webp", ".svg"}
    images_dir = item_dir / "images"
    image_links: list[str] = []
    if images_dir.exists() and images_dir.is_dir():
        for img_path in sorted(images_dir.iterdir()):
            if img_path.is_file() and img_path.suffix.lower() in image_extensions:
                rel = img_path.relative_to(item_dir).as_posix()
                image_links.append(f"- [{rel}](./{rel})")

    table_extensions = {".csv", ".html", ".md", ".json"}
    tables_dir = item_dir / "tables"
    table_links: list[str] = []
    if tables_dir.exists() and tables_dir.is_dir():
        for tbl_path in sorted(tables_dir.iterdir()):
            if tbl_path.is_file() and tbl_path.suffix.lower() in table_extensions:
                rel = tbl_path.relative_to(item_dir).as_posix()
                table_links.append(f"- [{rel}](./{rel})")

    lines = [
        f"# Item Catalog: {item_key}",
        "",
        f"- **Title**: {title or 'Untitled'}",
        f"- **Converter**: {converter_id} (v{converter_version})",
        doc_ref,
    ]
    if toc_ref:
        lines.append(toc_ref)
    if rel_ref:
        lines.append(rel_ref)
    lines.extend(["", "## Assets"])

    if not image_links and not table_links:
        lines.append("None")
    else:
        if image_links:
            lines.extend(["", "### Images", *image_links])
        if table_links:
            lines.extend(["", "### Tables", *table_links])

    lines.append("")
    index_path = item_dir / "index.md"
    index_path.write_text("\n".join(lines), encoding="utf-8")

generate_root_index_md

generate_root_index_md(cache_dir)

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

Parameters:

Name	Type	Description	Default
cache_dir	Path	Corpus directory containing conversion_meta.db.	required

Source code in zocrates/rag/bundle_catalog.py

def generate_root_index_md(cache_dir: Path) -> None:
    """Regenerate root-level index.md catalog from conversion_meta.db.

    Args:
        cache_dir: Corpus directory containing conversion_meta.db.
    """
    db_path = cache_dir / "conversion_meta.db"
    if not db_path.exists():
        return

    lines = []
    with contextlib.closing(sqlite3.connect(str(db_path), timeout=5.0)) as conn:
        conn.row_factory = sqlite3.Row
        conn.execute("PRAGMA busy_timeout = 5000;")
        rows = conn.execute(
            "SELECT item_key, title, parent_item_type, converter_id, converter_version, "
            "converted_at, source_path, char_count "
            "FROM conversion_meta ORDER BY converted_at DESC"
        ).fetchall()

        lines = [
            "# Zocrates Knowledge Base Catalog",
            "",
            f"Total Converted Documents: {len(rows)}",
            "",
            "| Item Key | Title | Type | Converter | Converted At | Source Path | Chars |",
            "| :--- | :--- | :--- | :--- | :--- | :--- | :--- |",
        ]

        title_pattern = re.compile(r"^title:\s*['\"]?(.*?)['\"]?\s*$", re.IGNORECASE)

        for r in rows:
            key = r["item_key"]
            raw_title = r["title"]

            # If title is missing in DB, attempt fast scan from document.md frontmatter
            if not raw_title or not str(raw_title).strip():
                doc_file = cache_dir / key / "document.md"
                if doc_file.exists():
                    try:
                        with doc_file.open("r", encoding="utf-8") as f:
                            for _ in range(30):
                                line = f.readline()
                                if not line:
                                    break
                                m = title_pattern.match(line.strip())
                                if m and m.group(1).strip():
                                    raw_title = m.group(1).strip()
                                    # Lazily backfill to DB
                                    conn.execute(
                                        "UPDATE conversion_meta SET title = ? WHERE item_key = ?",
                                        (raw_title, key),
                                    )
                                    conn.commit()
                                    break
                    except Exception:
                        pass

            if not raw_title or not str(raw_title).strip():
                raw_title = (r["source_path"] and Path(r["source_path"]).stem) or "Untitled"

            clean_title = (
                str(raw_title)
                .replace("|", "\\|")
                .replace("\r", " ")
                .replace("\n", " ")
                .strip()
            )
            p_type = r["parent_item_type"] or "document"
            conv = f"{r['converter_id']} (v{r['converter_version']})"
            conv_at = r["converted_at"]
            stored_path = (r["source_path"] or "").strip()
            if stored_path.startswith("zotero://note/"):
                rel_source = f"[Zotero Note]({stored_path})"
            elif stored_path.startswith("storage/"):
                rel_source = f"[Source Document](../../{stored_path})"
            elif stored_path.startswith("../../storage/"):
                rel_source = f"[Source Document]({stored_path})"
            elif stored_path:
                rel_source = f"[Source Document]({stored_path})"
            else:
                rel_source = ""
            count = r["char_count"]
            lines.append(
                f"| [{key}](./{key}/document.md) | {clean_title} | {p_type} | {conv} | {conv_at} | {rel_source} | {count} |"
            )

    lines.append("")
    root_index = cache_dir / "index.md"
    root_index.write_text("\n".join(lines), encoding="utf-8")

    # Keep AGENTS.md synchronized with root catalog
    generate_agents_md(cache_dir)

generate_toc

generate_toc(document_md, title=None, abstract=None, abstract_source='zotero', item_key='', parent_item_key=None)

Generate dual-format Table of Contents from a converted Markdown document.

Parameters:

Name	Type	Description	Default
document_md	str	Full converted Markdown text (including YAML frontmatter).	required
title	str | None	Document title.	None
abstract	str | None	Abstract or summary text.	None
abstract_source	str	Source of abstract ('zotero', 'generated', 'extractive').	'zotero'
item_key	str	Zotero item key.	''
parent_item_key	str | None	Parent Zotero item key if item is an attachment.	None

Returns:

Type	Description
tuple[dict[str, Any], str]	A 2-tuple of (toc_json_dict, toc_md_text).

Source code in zocrates/rag/toc_generator.py

def generate_toc(
    document_md: str,
    title: str | None = None,
    abstract: str | None = None,
    abstract_source: str = "zotero",
    item_key: str = "",
    parent_item_key: str | None = None,
) -> tuple[dict[str, Any], str]:
    """Generate dual-format Table of Contents from a converted Markdown document.

    Args:
        document_md: Full converted Markdown text (including YAML frontmatter).
        title: Document title.
        abstract: Abstract or summary text.
        abstract_source: Source of abstract ('zotero', 'generated', 'extractive').
        item_key: Zotero item key.
        parent_item_key: Parent Zotero item key if item is an attachment.

    Returns:
        A 2-tuple of `(toc_json_dict, toc_md_text)`.
    """
    line_index = LineIndex(document_md)
    lines = document_md.splitlines(keepends=True)
    if not lines:
        lines = [""]

    total_lines = line_index.total_lines
    total_words = len(document_md.split())

    # -----------------------------------------------------------------------
    # 1. Frontmatter Detection (s0)
    # -----------------------------------------------------------------------
    sections: list[dict[str, Any]] = []
    fm_match = _FRONTMATTER_RE.match(document_md)
    fm_end_line = 0

    if fm_match:
        fm_raw = fm_match.group(0)
        fm_end_line = len(fm_raw.rstrip("\r\n").splitlines())
        fm_content = fm_match.group(1)
        if parent_item_key is None:
            try:
                fm_data = yaml.safe_load(fm_content)
                if isinstance(fm_data, dict) and "parent_item_key" in fm_data:
                    parent_item_key = fm_data.get("parent_item_key")
            except Exception:
                pass
        sections.append(
            {
                "id": "s0",
                "heading": "Frontmatter",
                "level": 0,
                "start_line": 1,
                "end_line": fm_end_line,
                "word_count": len(fm_content.split()),
            }
        )

    # -----------------------------------------------------------------------
    # 2. Section Extraction (s1, s2, ...)
    # -----------------------------------------------------------------------
    # Mask code blocks during heading discovery
    in_code_fence = False
    fence_char = ""
    discovered_headings: list[dict[str, Any]] = []

    for line_idx, line in enumerate(lines, start=1):
        # Only parse headings after frontmatter
        if line_idx <= fm_end_line:
            continue

        stripped_line = line.strip()
        if not in_code_fence:
            if stripped_line.startswith(("```", "~~~")):
                in_code_fence = True
                fence_char = stripped_line[:3]
                continue
        else:
            if stripped_line.startswith(fence_char):
                in_code_fence = False
            continue

        h_match = _HEADING_LINE_RE.match(line)
        if h_match:
            hashes = h_match.group(2)
            level = len(hashes)
            raw_title = h_match.group(3)
            clean_title = _clean_text_inline(raw_title)
            if clean_title:
                discovered_headings.append(
                    {
                        "heading": clean_title,
                        "level": level,
                        "start_line": line_idx,
                    }
                )

    if discovered_headings:
        for idx, h in enumerate(discovered_headings):
            # Section IDs start at s1 if no frontmatter (s0 is reserved for Frontmatter)
            s_id = f"s{len(sections) if fm_match else len(sections) + 1}"
            start_line = h["start_line"]
            if idx + 1 < len(discovered_headings):
                end_line = discovered_headings[idx + 1]["start_line"] - 1
            else:
                end_line = total_lines

            # Calculate word count in this section's slice
            section_slice = "".join(lines[start_line - 1 : end_line])
            word_count = len(section_slice.split())

            sections.append(
                {
                    "id": s_id,
                    "heading": h["heading"],
                    "level": h["level"],
                    "start_line": start_line,
                    "end_line": max(start_line, end_line),
                    "word_count": word_count,
                }
            )
    else:
        # Fallback for documents with no ATX headings (only if there is non-frontmatter body)
        if not fm_match or fm_end_line < total_lines:
            body_start = fm_end_line + 1 if (fm_match and fm_end_line < total_lines) else 1
            body_slice = "".join(lines[body_start - 1 : total_lines])
            s_id = f"s{len(sections) if fm_match else len(sections) + 1}"
            sections.append(
                {
                    "id": s_id,
                    "heading": "Document Body",
                    "level": 1,
                    "start_line": body_start,
                    "end_line": total_lines,
                    "word_count": len(body_slice.split()),
                }
            )

    # -----------------------------------------------------------------------
    # 3. Figure Extraction (f1, f2, ...)
    # -----------------------------------------------------------------------
    figures: list[dict[str, Any]] = []
    figure_spans: list[tuple[int, int]] = []

    # 3a. Match semantic figure blocks
    for m in _FIGURE_BLOCK_RE.finditer(document_md):
        attrs = m.group(1)
        body = m.group(2)
        start_char, end_char = m.span()
        start_line, end_line = line_index.span_to_lines(start_char, end_char)

        # Check if figure is an image or table
        is_table = 'data-type="table"' in attrs.lower() or 'data-type=\'table\'' in attrs.lower()
        if is_table:
            continue

        figure_spans.append((start_char, end_char))

        # Extract image path
        path = ""
        img_match = _STANDALONE_IMG_RE.search(body)
        if img_match:
            path = img_match.group(2)
        else:
            raw_p_match = re.search(r'(?:src=["\']|images/)([^\s"\'\)]+)', body)
            if raw_p_match:
                p_candidate = raw_p_match.group(1)
                path = p_candidate if p_candidate.startswith("images/") else f"images/{Path(p_candidate).name}"

        # Resolve caption using hierarchy: author caption -> synthetic caption -> alt text -> fallback
        resolved_caption = ""
        for line in body.splitlines():
            line_str = line.strip()
            if line_str.startswith("!["):
                continue
            cap_line_match = _FIGURE_CAPTION_LINE_RE.match(line_str)
            if cap_line_match:
                label = cap_line_match.group(1).strip()
                rest = cap_line_match.group(2).strip()
                prefix = f"Figure {label}" if not label.lower().startswith("fig") else label
                resolved_caption = f"{prefix} {rest}".strip() if label else rest
                break

        if not resolved_caption:
            synth_match = _SYNTH_CAPTION_RE.search(body)
            if synth_match:
                resolved_caption = synth_match.group(1).strip()

        if not resolved_caption and img_match:
            alt = img_match.group(1).strip()
            # Ignore generic alt texts
            if alt and not re.match(r"^(?:figure|image|picture)(?:\s*\d+)?(?:\s*\(page\s*\d+\))?$", alt, re.IGNORECASE):
                resolved_caption = alt

        fig_id = f"f{len(figures) + 1}"
        if not resolved_caption:
            resolved_caption = f"Figure {len(figures) + 1}"

        figures.append(
            {
                "id": fig_id,
                "caption": _truncate_caption(resolved_caption),
                "path": path,
                "start_line": start_line,
                "end_line": max(start_line, end_line),
            }
        )

    # 3b. Match standalone markdown images outside <figure> blocks
    for m in _STANDALONE_IMG_RE.finditer(document_md):
        start_char, end_char = m.span()
        # Check if inside any already-matched figure block
        inside_figure = any(s <= start_char and end_char <= e for s, e in figure_spans)
        if inside_figure:
            continue

        start_line, end_line = line_index.span_to_lines(start_char, end_char)
        alt = m.group(1).strip()
        path = m.group(2).strip()

        fig_id = f"f{len(figures) + 1}"
        caption = alt if (alt and not re.match(r"^(?:figure|image|picture)(?:\s*\d+)?$", alt, re.IGNORECASE)) else f"Figure {len(figures) + 1}"

        figures.append(
            {
                "id": fig_id,
                "caption": _truncate_caption(caption),
                "path": path,
                "start_line": start_line,
                "end_line": max(start_line, end_line),
            }
        )

    # -----------------------------------------------------------------------
    # 4. Table Extraction (t1, t2, ...)
    # -----------------------------------------------------------------------
    tables: list[dict[str, Any]] = []
    table_spans: list[tuple[int, int]] = []

    # 4a. Match semantic table blocks
    for m in _FIGURE_BLOCK_RE.finditer(document_md):
        attrs = m.group(1)
        body = m.group(2)
        is_table = 'data-type="table"' in attrs.lower() or 'data-type=\'table\'' in attrs.lower()
        if not is_table:
            continue

        start_char, end_char = m.span()
        start_line, end_line = line_index.span_to_lines(start_char, end_char)
        table_spans.append((start_char, end_char))

        # Extract table path
        path = None
        csv_match = re.search(r'data-csv=["\']([^"\']+)["\']', attrs)
        html_match = re.search(r'data-html=["\']([^"\']+)["\']', attrs)
        comment_match = re.search(r'<!--\s*artifacts:\s*(?:csv=([^\s]+)|html=([^\s]+))', body)
        link_match = re.search(r'\[(?:CSV|HTML)\]\((tables\/[^\)]+)\)', body)

        if csv_match:
            path = csv_match.group(1)
        elif html_match:
            path = html_match.group(1)
        elif comment_match:
            path = comment_match.group(1) or comment_match.group(2)
        elif link_match:
            path = link_match.group(1)

        # Resolve caption
        resolved_caption = ""
        for line in body.splitlines():
            line_str = line.strip()
            if line_str.startswith("![") or line_str.startswith("|"):
                continue
            cap_line_match = _TABLE_CAPTION_LINE_RE.match(line_str)
            if cap_line_match:
                label = cap_line_match.group(1).strip()
                rest = cap_line_match.group(2).strip()
                prefix = f"Table {label}" if not label.lower().startswith("tab") else label
                resolved_caption = f"{prefix} {rest}".strip() if label else rest
                break

        tbl_id = f"t{len(tables) + 1}"
        if not resolved_caption:
            resolved_caption = f"Table {len(tables) + 1}"

        tables.append(
            {
                "id": tbl_id,
                "caption": _truncate_caption(resolved_caption),
                "path": path,
                "start_line": start_line,
                "end_line": max(start_line, end_line),
            }
        )

    # 4b. Match standalone GFM Markdown tables outside <figure> blocks
    for m in _GFM_TABLE_RE.finditer(document_md):
        start_char, end_char = m.span()
        inside_table_figure = any(s <= start_char and end_char <= e for s, e in table_spans)
        if inside_table_figure:
            continue

        start_line, end_line = line_index.span_to_lines(start_char, end_char)

        # Check line above for table caption
        caption = ""
        if start_line > 1:
            prev_line = lines[start_line - 2].strip()
            if not prev_line.startswith("!["):
                cap_match = _TABLE_CAPTION_LINE_RE.match(prev_line)
                if cap_match:
                    caption = _clean_text_inline(prev_line)

        tbl_id = f"t{len(tables) + 1}"
        if not caption:
            caption = f"Table {len(tables) + 1}"

        tables.append(
            {
                "id": tbl_id,
                "caption": _truncate_caption(caption),
                "path": None,
                "start_line": start_line,
                "end_line": max(start_line, end_line),
            }
        )

    # -----------------------------------------------------------------------
    # 5. Build toc.json structure
    # -----------------------------------------------------------------------
    clean_abstract = (abstract or "").strip()
    toc_json: dict[str, Any] = {
        "version": 1,
        "item_key": item_key,
        "parent_item_key": parent_item_key,
        "title": title or "",
        "abstract": clean_abstract,
        "abstract_source": abstract_source,
        "total_lines": total_lines,
        "total_words": total_words,
        "sections": sections,
        "figures": figures,
        "tables": tables,
    }

    # -----------------------------------------------------------------------
    # 6. Render toc.md
    # -----------------------------------------------------------------------
    md_lines: list[str] = [
        f"# {title or 'Document'}",
        "",
        "## Abstract",
        clean_abstract if clean_abstract else "No abstract or summary available.",
        "",
        "## Table of Contents",
    ]

    for sec in sections:
        s_id = sec["id"]
        heading = sec["heading"]
        level = sec["level"]
        words = sec["word_count"]

        if level == 0:
            indent = ""
            heading_str = heading
        elif level <= 2:
            indent = ""
            heading_str = f"{'#' * level} {heading}"
        else:
            indent = "  " * (level - 2)
            heading_str = f"{'#' * level} {heading}"

        md_lines.append(f"{indent}- [{s_id}] {heading_str} (~{words} words)")

    md_lines.append("")
    md_lines.append("## Figures")
    if figures:
        for fig in figures:
            f_id = fig["id"]
            cap = fig["caption"]
            p = fig.get("path")
            ref = f" ({p})" if p else ""
            md_lines.append(f"- [{f_id}] {cap}{ref}")
    else:
        md_lines.append("None")

    md_lines.append("")
    md_lines.append("## Tables")
    if tables:
        for tbl in tables:
            t_id = tbl["id"]
            cap = tbl["caption"]
            p = tbl.get("path")
            ref = f" ({p})" if p else ""
            md_lines.append(f"- [{t_id}] {cap}{ref}")
    else:
        md_lines.append("None")

    md_lines.append("")
    toc_md = "\n".join(md_lines)

    return toc_json, toc_md

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

refresh_bundle_frontmatter

refresh_bundle_frontmatter(zotero_dir, target_keys=None, dry_run=False, sanitize_images=False, store=None, cache_dir=None)

Refresh YAML frontmatter and associated artifacts across target or all cached bundles.

Parameters:

Name	Type	Description	Default
zotero_dir	Path	Root Zotero profile directory containing zotero.sqlite and storage/.	required
target_keys	list[str] | None	Optional list of attachment or note item keys to refresh. If None, scans all cached bundles in conversion_meta.db.	None
dry_run	bool	If True, previews changes without writing to disk or database.	False
sanitize_images	bool	If True, converts legacy file:/// URIs in document.md to images/.	False
store	VectorStore | None	Optional existing VectorStore instance for FTS5 and Chroma updates.	None
cache_dir	Path | None	Optional explicit corpus directory. Defaults to zotero_dir / "zocrates-data" / "corpus".	None

Returns:

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

Source code in zocrates/rag/frontmatter_refresher.py

def refresh_bundle_frontmatter(
    zotero_dir: Path,
    target_keys: list[str] | None = None,
    dry_run: bool = False,
    sanitize_images: bool = False,
    store: VectorStore | None = None,
    cache_dir: Path | None = None,
) -> dict[str, int]:
    """Refresh YAML frontmatter and associated artifacts across target or all cached bundles.

    Args:
        zotero_dir: Root Zotero profile directory containing zotero.sqlite and storage/.
        target_keys: Optional list of attachment or note item keys to refresh.
            If None, scans all cached bundles in conversion_meta.db.
        dry_run: If True, previews changes without writing to disk or database.
        sanitize_images: If True, converts legacy file:/// URIs in document.md to images/.
        store: Optional existing VectorStore instance for FTS5 and Chroma updates.
        cache_dir: Optional explicit corpus directory. Defaults to
            zotero_dir / "zocrates-data" / "corpus".

    Returns:
        Summary dict containing counts of total, updated, skipped, and errored items.
    """
    if cache_dir is None:
        cache_dir = get_corpus_dir(zotero_dir)

    sqlite_path = zotero_dir / "zotero.sqlite"
    storage_dir = zotero_dir / "storage"
    meta_db_path = cache_dir / "conversion_meta.db"

    if not sqlite_path.exists():
        logger.error("Zotero database not found at %s", sqlite_path)
        return {"total": 0, "updated": 0, "skipped": 0, "errored": 0}

    if not meta_db_path.exists():
        logger.info("No conversion_meta.db found at %s; nothing to refresh", meta_db_path)
        return {"total": 0, "updated": 0, "skipped": 0, "errored": 0}

    db = ZoteroDB(sqlite_path, storage_dir)

    with _open_meta_db(cache_dir) as meta_conn:
        if target_keys:
            placeholders = ",".join("?" for _ in target_keys)
            rows = meta_conn.execute(
                f"SELECT * FROM conversion_meta WHERE item_key IN ({placeholders})",
                target_keys,
            ).fetchall()
        else:
            rows = meta_conn.execute(
                "SELECT * FROM conversion_meta ORDER BY item_key"
            ).fetchall()

    if not rows:
        return {"total": 0, "updated": 0, "skipped": 0, "errored": 0}

    fts_path = store.fts_path if store is not None else get_fts_db_path(zotero_dir)
    fts_available = fts_path.exists()

    total = len(rows)
    updated = 0
    skipped = 0
    errored = 0
    noop_skipped_keys: list[str] = []

    with _REFRESH_LOCK:
        for r in rows:
            item_key = r["item_key"]
            if r["content_type"] == "text/note":
                skipped += 1
                continue
            parent_key = r["parent_item_key"]
            bundle_dir = cache_dir / item_key
            doc_path = bundle_dir / "document.md"

            if not doc_path.exists():
                logger.warning("[%s] document.md not found on disk; skipping.", item_key)
                skipped += 1
                continue

            # Query fresh metadata from Zotero DB with single retry resilience
            parent_item = None
            try:
                if parent_key:
                    parent_item = db.get_item_by_key(parent_key)
                if not parent_item:
                    parent_item = db.get_item_by_key(item_key)
            except Exception as exc:
                time.sleep(0.25)
                try:
                    if parent_key:
                        parent_item = db.get_item_by_key(parent_key)
                    if not parent_item:
                        parent_item = db.get_item_by_key(item_key)
                except Exception as retry_exc:
                    logger.error(
                        "Failed to read Zotero metadata for item_key=%s: %s",
                        item_key,
                        retry_exc,
                    )
                    errored += 1
                    continue

            if parent_item is None:
                logger.warning(
                    "Item %s (parent %s) no longer found in Zotero library; skipping frontmatter refresh.",
                    item_key,
                    parent_key,
                )
                skipped += 1
                continue

            col_keys = r.keys()
            title = (
                parent_item.title
                if parent_item and parent_item.title
                else (r["title"] if "title" in col_keys and r["title"] else "")
            )
            p_type = parent_item.item_type if parent_item else r["parent_item_type"]
            colls = parent_item.collections if parent_item else []
            authors = parent_item.creators if parent_item else None
            year = parent_item.year if parent_item else None
            pub = parent_item.publication if parent_item else None
            doi = parent_item.doi if parent_item else None
            url = parent_item.url if parent_item else None
            abstract = (
                parent_item.abstract
                if parent_item and parent_item.abstract
                else (r["abstract"] if "abstract" in col_keys else None)
            )
            abstract_source = (
                "zotero"
                if (parent_item and parent_item.abstract)
                else (
                    r["abstract_source"]
                    if "abstract_source" in col_keys and r["abstract_source"]
                    else "extractive"
                )
            )

            raw_content = doc_path.read_text(encoding="utf-8")
            fm_match = _FRONTMATTER_RE.match(raw_content)
            if fm_match:
                old_fm_raw = fm_match.group(0)
                old_fm_lines = len(old_fm_raw.rstrip("\r\n").splitlines())
                body = raw_content[fm_match.end() :]
            else:
                old_fm_raw = ""
                old_fm_lines = 0
                body = raw_content

            if sanitize_images:
                body = sanitize_image_links_in_markdown(body)

            stored_path = r["source_path"] or ""
            resolved_src = resolve_source_path(stored_path, zotero_dir)
            fm_source_path = (
                stored_path
                if stored_path.startswith("storage/") or stored_path.startswith("zotero://note/")
                else (Path(resolved_src) if resolved_src else Path(stored_path))
            )

            keywords: list[str] | None = None
            if "keywords" in col_keys and r["keywords"]:
                try:
                    raw_kw = json.loads(r["keywords"])
                    if isinstance(raw_kw, list):
                        keywords = [str(k) for k in raw_kw if k]
                    elif isinstance(raw_kw, str):
                        keywords = [raw_kw]
                except Exception:
                    keywords = [k.strip() for k in str(r["keywords"]).split(",") if k.strip()]

            if keywords is None and fm_match:
                try:
                    parsed_yaml = yaml.safe_load(fm_match.group(1))
                    if isinstance(parsed_yaml, dict) and "keywords" in parsed_yaml:
                        yk = parsed_yaml["keywords"]
                        if isinstance(yk, list):
                            keywords = [str(k) for k in yk if k]
                        elif isinstance(yk, str):
                            keywords = [yk]
                except Exception:
                    pass

            new_fm = build_frontmatter(
                item_key=item_key,
                parent_item_type=p_type,
                title=title,
                collections=colls,
                source_path=fm_source_path,
                source_hash=r["source_hash"],
                content_type=r["content_type"],
                converter_id=r["converter_id"],
                converter_version=r["converter_version"],
                converted_at=r["converted_at"],
                parent_item_key=parent_key,
                parse_quality_score=r["parse_quality_score"]
                if "parse_quality_score" in col_keys
                else None,
                checkpoint_id=r["checkpoint_id"]
                if "checkpoint_id" in col_keys
                else None,
                authors=authors,
                year=year,
                publication=pub,
                doi=doi,
                url=url,
                keywords=keywords,
                bundle_dir=bundle_dir,
            )

            # Deterministic No-Op check
            new_fm_normalized = new_fm.replace("\r\n", "\n")
            old_fm_normalized = old_fm_raw.replace("\r\n", "\n")
            body_changed = (
                sanitize_images
                and fm_match
                and body != raw_content[fm_match.end() :]
            )

            # Self-healing check: verify if toc.json exists and matches current frontmatter line count
            toc_json_path = bundle_dir / "toc.json"
            toc_out_of_sync = False
            if not toc_json_path.exists():
                toc_out_of_sync = True
            else:
                try:
                    with open(toc_json_path, encoding="utf-8") as tf:
                        cached_toc = json.load(tf)
                    secs = cached_toc.get("sections", [])
                    if secs and secs[0].get("id") == "s0":
                        if secs[0].get("end_line") != old_fm_lines:
                            toc_out_of_sync = True
                except Exception:
                    toc_out_of_sync = True

            # Self-healing check: verify if chunk_meta has drifted Chunk 0 line boundary
            chunk_meta_out_of_sync = False
            if fts_available:
                try:
                    with _open_fts_db(fts_path) as conn:
                        cm_row = conn.execute(
                            "SELECT chunk_type, line_end FROM chunk_meta WHERE item_key = ? AND chunk_index = 0",
                            (item_key,),
                        ).fetchone()
                        if cm_row and cm_row["chunk_type"] == "frontmatter":
                            if cm_row["line_end"] is not None and cm_row["line_end"] != old_fm_lines:
                                chunk_meta_out_of_sync = True
                except Exception:
                    pass

            if (
                new_fm_normalized == old_fm_normalized
                and not body_changed
                and not toc_out_of_sync
                and not chunk_meta_out_of_sync
            ):
                if not dry_run:
                    noop_skipped_keys.append(item_key)
                skipped += 1
                continue

            if dry_run:
                updated += 1
                continue

            new_fm_lines = len(new_fm.rstrip("\r\n").splitlines())
            delta_l = new_fm_lines - old_fm_lines
            full_markdown = new_fm + body

            # 1. Crash-safe atomic write of document.md
            tmp_doc = bundle_dir / "document.md.tmp"
            try:
                with open(tmp_doc, "w", encoding="utf-8") as f:
                    f.write(full_markdown)
                    f.flush()
                    os.fsync(f.fileno())
                os.replace(tmp_doc, doc_path)
            except Exception as write_exc:
                logger.error("Failed atomic write for %s: %s", doc_path, write_exc)
                if tmp_doc.exists():
                    tmp_doc.unlink(missing_ok=True)
                errored += 1
                continue

            # 2. Synchronize ToC
            try:
                toc_json, toc_md = generate_toc(
                    document_md=full_markdown,
                    title=title,
                    abstract=abstract,
                    abstract_source=abstract_source,
                    item_key=item_key,
                    parent_item_key=parent_key,
                )
                save_toc_files(bundle_dir, toc_json, toc_md)
            except Exception as toc_exc:
                logger.warning("Failed to regenerate ToC for %s: %s", item_key, toc_exc)

            # 3. Synchronize chunk_meta and chunks_fts in SQLite FTS5
            is_fm_chunk = True
            fts_succeeded = True
            if fts_available:
                try:
                    with _open_fts_db(fts_path) as conn:
                        c0_row = conn.execute(
                            "SELECT chunk_type, line_end FROM chunk_meta WHERE item_key = ? AND chunk_index = 0",
                            (item_key,),
                        ).fetchone()
                        is_fm_chunk = (c0_row is None) or (c0_row["chunk_type"] == "frontmatter")

                        if c0_row and c0_row["chunk_type"] == "frontmatter" and c0_row["line_end"] is not None:
                            shift_amount = new_fm_lines - c0_row["line_end"]
                        else:
                            shift_amount = delta_l

                        if is_fm_chunk:
                            # Update Chunk 0 line boundary
                            conn.execute(
                                "UPDATE chunk_meta SET line_end = ? WHERE item_key = ? AND chunk_index = 0",
                                (new_fm_lines, item_key),
                            )
                            # Idempotently replace Chunk 0 text in chunks_fts
                            conn.execute(
                                "DELETE FROM chunks_fts WHERE chunk_id = ?",
                                (f"{item_key}::0",),
                            )
                            conn.execute(
                                "INSERT INTO chunks_fts (chunk_id, item_key, chunk_text) VALUES (?, ?, ?)",
                                (f"{item_key}::0", item_key, new_fm),
                            )

                        # Shift body chunks by shift_amount (derived from stored Chunk 0 line_end for self-healing)
                        if shift_amount != 0:
                            conn.execute(
                                "UPDATE chunk_meta SET line_start = line_start + ?, line_end = line_end + ? "
                                "WHERE item_key = ? AND chunk_index > 0",
                                (shift_amount, shift_amount, item_key),
                            )
                        conn.commit()
                except Exception as fts_exc:
                    fts_succeeded = False
                    logger.warning(
                        "Failed to update FTS5/chunk_meta for %s: %s",
                        item_key,
                        fts_exc,
                    )

            # 4. Chroma Chunk 0 Text & Title Propagation (Zero Re-Embedding)
            if store is not None:
                try:
                    if is_fm_chunk:
                        store.update_chunk_text_preserve_vector(f"{item_key}::0", new_fm)
                    store.update_metadata(item_key, {"document_title": title})
                except Exception as chroma_exc:
                    logger.warning(
                        "Failed to update Chroma metadata for %s: %s",
                        item_key,
                        chroma_exc,
                    )

            # 5. Update conversion_meta.db (only record updated_at if FTS succeeded)
            new_char_count = len(full_markdown)
            now_ts = datetime.now(timezone.utc).isoformat()
            try:
                with _open_meta_db(cache_dir) as meta_conn:
                    if fts_succeeded:
                        meta_conn.execute(
                            "UPDATE conversion_meta SET title = ?, char_count = ?, updated_at = ? WHERE item_key = ?",
                            (title, new_char_count, now_ts, item_key),
                        )
                    else:
                        meta_conn.execute(
                            "UPDATE conversion_meta SET title = ?, char_count = ? WHERE item_key = ?",
                            (title, new_char_count, item_key),
                        )
                    if abstract:
                        meta_conn.execute(
                            "UPDATE conversion_meta SET abstract = ? WHERE item_key = ? AND (abstract IS NULL OR abstract = '')",
                            (abstract, item_key),
                        )
                        if "abstract_source" in col_keys:
                            meta_conn.execute(
                                "UPDATE conversion_meta SET abstract_source = ? WHERE item_key = ? AND (abstract_source IS NULL OR abstract_source = '')",
                                (abstract_source, item_key),
                            )
                    if keywords and "keywords" in col_keys and not r["keywords"]:
                        meta_conn.execute(
                            "UPDATE conversion_meta SET keywords = ? WHERE item_key = ? AND (keywords IS NULL OR keywords = '')",
                            (json.dumps(keywords), item_key),
                        )
                    meta_conn.commit()
            except Exception as meta_exc:
                logger.warning(
                    "Failed to update conversion_meta for %s: %s",
                    item_key,
                    meta_exc,
                )

            # 6. Item-level catalog
            try:
                generate_item_index_md(
                    bundle_dir,
                    item_key,
                    title,
                    r["converter_id"],
                    r["converter_version"],
                )
            except Exception as idx_exc:
                logger.warning(
                    "Failed to regenerate index.md for %s: %s",
                    item_key,
                    idx_exc,
                )

            updated += 1

        # Batch update updated_at for no-op skipped items in a single transaction
        if noop_skipped_keys and not dry_run:
            try:
                now_ts = datetime.now(timezone.utc).isoformat()
                with _open_meta_db(cache_dir) as meta_conn:
                    meta_conn.executemany(
                        "UPDATE conversion_meta SET updated_at = ? WHERE item_key = ?",
                        [(now_ts, k) for k in noop_skipped_keys],
                    )
                    meta_conn.commit()
            except Exception as batch_exc:
                logger.debug("Failed to batch update updated_at for skipped items: %s", batch_exc)

    # 7. Batched root catalog regeneration
    if not dry_run and updated > 0:
        try:
            generate_root_index_md(cache_dir)
        except Exception as root_exc:
            logger.warning("Failed to regenerate root index.md: %s", root_exc)

    return {"total": total, "updated": updated, "skipped": skipped, "errored": errored}

resolve_source_path

resolve_source_path(stored_path, zotero_profile_dir)

Resolve a stored source_path to a local physical Path or verbatim URI.

Parameters:

Name	Type	Description	Default
stored_path	str | Path | None	Storage-relative path ('storage/...'), note URI, or absolute path.	required
zotero_profile_dir	Path	Root Zotero profile directory.	required

Returns:

Type	Description
Path | str	Resolved physical Path on disk or URI string.

Source code in zocrates/rag/bundle_catalog.py

def resolve_source_path(
    stored_path: str | Path | None,
    zotero_profile_dir: Path,
) -> Path | str:
    """Resolve a stored source_path to a local physical Path or verbatim URI.

    Args:
        stored_path: Storage-relative path ('storage/...'), note URI, or absolute path.
        zotero_profile_dir: Root Zotero profile directory.

    Returns:
        Resolved physical Path on disk or URI string.
    """
    if not stored_path or not str(stored_path).strip():
        return ""

    str_stored = str(stored_path).strip()
    if str_stored.startswith("zotero://note/") or ("file:" not in str_stored and "://" in str_stored):
        return str_stored

    if str_stored.startswith("storage/"):
        unquoted = urllib.parse.unquote(str_stored)
        return (zotero_profile_dir / unquoted).resolve()

    p = Path(str_stored)
    if p.exists():
        return p.resolve()

    # Fallback heuristic: check if storage/<key>/<filename> exists in local zotero_profile_dir
    m = re.search(
        r"(?:^|[/\\])storage[/\\](?P<key>[A-Za-z0-9_-]+)(?:[/\\](?P<file>.+))?$",
        str_stored,
    )
    if m:
        key = m.group("key")
        filename = urllib.parse.unquote(m.group("file") or "")
        candidate = (
            (zotero_profile_dir / "storage" / key / filename)
            if filename
            else (zotero_profile_dir / "storage" / key)
        )
        if candidate.exists():
            return candidate.resolve()

    return p

sanitize_image_links_in_markdown

sanitize_image_links_in_markdown(text)

Sanitize legacy file:///.../images/ absolute links in markdown text.

Ensures strict line-count invariance to preserve ToC and chunk line offsets.

Parameters:

Name	Type	Description	Default
text	str	Raw markdown text.	required

Returns:

Type	Description
str	Markdown text with file:///.../images/ rewritten to images/.

Source code in zocrates/rag/document_processor.py

def sanitize_image_links_in_markdown(text: str) -> str:
    """Sanitize legacy file:///.../images/ absolute links in markdown text.

    Ensures strict line-count invariance to preserve ToC and chunk line offsets.

    Args:
        text: Raw markdown text.

    Returns:
        Markdown text with file:///.../images/<name> rewritten to images/<name>.
    """
    if "file://" not in text or "images/" not in text:
        return text
    pattern = re.compile(r"file://[^\s\)\"']+?[/\\]images[/\\]([^\s\)\"']+)")
    return pattern.sub(lambda m: f"images/{urllib.parse.unquote(m.group(1))}", text)

save_toc_files

save_toc_files(bundle_dir, toc_json, toc_md)

Safely persist toc.json and toc.md inside bundle_dir using atomic writes.

Writes to temporary files within the target bundle directory and renames via os.replace to prevent partial reads or corrupt states.

Parameters:

Name	Type	Description	Default
bundle_dir	Path	OKF document bundle directory (e.g. corpus//).	required
toc_json	dict[str, Any]	Dictionary representation of toc.json.	required
toc_md	str	String contents for toc.md.	required

Source code in zocrates/rag/toc_generator.py

def save_toc_files(bundle_dir: Path, toc_json: dict[str, Any], toc_md: str) -> None:
    """Safely persist toc.json and toc.md inside bundle_dir using atomic writes.

    Writes to temporary files within the target bundle directory and renames
    via os.replace to prevent partial reads or corrupt states.

    Args:
        bundle_dir: OKF document bundle directory (e.g. corpus/<item_key>/).
        toc_json: Dictionary representation of toc.json.
        toc_md: String contents for toc.md.
    """
    bundle_dir.mkdir(parents=True, exist_ok=True)

    toc_md_path = bundle_dir / "toc.md"
    toc_json_path = bundle_dir / "toc.json"

    # Write toc.md atomically first
    tmp_md = tempfile.NamedTemporaryFile(
        mode="w",
        encoding="utf-8",
        dir=bundle_dir,
        delete=False,
        prefix="toc_md_",
        suffix=".tmp",
    )
    try:
        tmp_md.write(toc_md)
        tmp_md.flush()
        os.fsync(tmp_md.fileno())
    finally:
        tmp_md.close()
    os.replace(tmp_md.name, toc_md_path)

    # Write toc.json atomically second
    tmp_json = tempfile.NamedTemporaryFile(
        mode="w",
        encoding="utf-8",
        dir=bundle_dir,
        delete=False,
        prefix="toc_json_",
        suffix=".tmp",
    )
    try:
        tmp_json.write(json.dumps(toc_json, indent=2, ensure_ascii=False) + "\n")
        tmp_json.flush()
        os.fsync(tmp_json.fileno())
    finally:
        tmp_json.close()
    os.replace(tmp_json.name, toc_json_path)

sync_stale_metadata

sync_stale_metadata(zotero_dir, dry_run=False, sanitize_images=False, store=None, cache_dir=None)

Scan all cached items and refresh any whose Zotero metadata has been modified.

Performs a lightweight timestamp comparison (<5ms) between conversion_meta.db and zotero.sqlite. Only items whose source metadata was modified since conversion are dispatched to refresh_bundle_frontmatter.

Parameters:

Name	Type	Description	Default
zotero_dir	Path	Root Zotero profile directory.	required
dry_run	bool	If True, previews changes without writing to disk or database.	False
sanitize_images	bool	If True, sanitizes legacy file:/// image links.	False
store	VectorStore | None	Optional existing VectorStore instance for Chroma and FTS updates.	None
cache_dir	Path | None	Optional explicit markdown cache directory.	None

Returns:

Type	Description
dict[str, int]	Summary dict containing counts of total, updated, skipped, and errored items.
dict[str, int]	When stale items are found, total reflects the count of stale items dispatched
dict[str, int]	to refresh_bundle_frontmatter; when no items are stale, total and skipped reflect
dict[str, int]	the full count of checked cached bundles.

Source code in zocrates/rag/frontmatter_refresher.py

def sync_stale_metadata(
    zotero_dir: Path,
    dry_run: bool = False,
    sanitize_images: bool = False,
    store: VectorStore | None = None,
    cache_dir: Path | None = None,
) -> dict[str, int]:
    """Scan all cached items and refresh any whose Zotero metadata has been modified.

    Performs a lightweight timestamp comparison (<5ms) between conversion_meta.db
    and zotero.sqlite. Only items whose source metadata was modified since
    conversion are dispatched to refresh_bundle_frontmatter.

    Args:
        zotero_dir: Root Zotero profile directory.
        dry_run: If True, previews changes without writing to disk or database.
        sanitize_images: If True, sanitizes legacy file:/// image links.
        store: Optional existing VectorStore instance for Chroma and FTS updates.
        cache_dir: Optional explicit markdown cache directory.

    Returns:
        Summary dict containing counts of total, updated, skipped, and errored items.
        When stale items are found, total reflects the count of stale items dispatched
        to refresh_bundle_frontmatter; when no items are stale, total and skipped reflect
        the full count of checked cached bundles.
    """
    if cache_dir is None:
        cache_dir = get_corpus_dir(zotero_dir)

    meta_db = cache_dir / "conversion_meta.db"
    if not meta_db.exists():
        return {"total": 0, "updated": 0, "skipped": 0, "errored": 0}

    with _open_meta_db(cache_dir) as conn:
        cols = {
            row["name"]
            for row in conn.execute("PRAGMA table_info(conversion_meta)").fetchall()
        }
        if "updated_at" in cols:
            rows = conn.execute(
                "SELECT item_key, parent_item_key, COALESCE(updated_at, converted_at) AS converted_at "
                "FROM conversion_meta"
            ).fetchall()
        else:
            rows = conn.execute(
                "SELECT item_key, parent_item_key, converted_at FROM conversion_meta"
            ).fetchall()

    if not rows:
        return {"total": 0, "updated": 0, "skipped": 0, "errored": 0}

    key_to_atts: dict[str, list[str]] = {}
    converted_map: dict[str, str] = {}
    for r in rows:
        target_key = r["parent_item_key"] or r["item_key"]
        key_to_atts.setdefault(target_key, []).append(r["item_key"])
        converted_map[r["item_key"]] = r["converted_at"]

    sqlite_path = zotero_dir / "zotero.sqlite"
    storage_dir = zotero_dir / "storage"
    if not sqlite_path.exists():
        return {"total": len(rows), "updated": 0, "skipped": len(rows), "errored": 0}

    db = ZoteroDB(sqlite_path, storage_dir)
    stale_keys = db.get_stale_item_keys(converted_map, key_to_atts)

    if not stale_keys:
        return {"total": len(rows), "updated": 0, "skipped": len(rows), "errored": 0}

    return refresh_bundle_frontmatter(
        zotero_dir=zotero_dir,
        target_keys=stale_keys,
        dry_run=dry_run,
        sanitize_images=sanitize_images,
        store=store,
        cache_dir=cache_dir,
    )