document_processor

zocrates.rag.document_processor

Document processor: converts PDF/HTML attachments to Markdown and caches them.

Supports pluggable converter strategies via BaseDocumentConverter. Default converters include: - DirectReadConverter: Fast UTF-8 reader for .md and .txt files. - MarkItDownConverter: Uses MarkItDown for document conversion with page-by-page PDF image extraction via system utility pdfimages.

The cache directory lives in

/zocrates-data/corpus/

Each item gets an item-centric bundle directory: corpus//document.md -- converted Markdown corpus//images/ -- extracted images corpus/conversion_meta.db -- SQLite metadata (authoritative)

CONVERTER_REGISTRY module-attribute

CONVERTER_REGISTRY = {'direct_read': DirectReadConverter, 'markitdown': MarkItDownConverter, 'docling_local': DoclingLocalConverter, 'datalab': DatalabConverter}

_INITIALIZED_META_DBS module-attribute

_INITIALIZED_META_DBS = set()

_META_INIT_LOCK module-attribute

_META_INIT_LOCK = threading.Lock()

_META_SCHEMA module-attribute

_META_SCHEMA = '\nCREATE TABLE IF NOT EXISTS conversion_meta (\n    item_key            TEXT PRIMARY KEY,\n    title               TEXT,\n    converter_id        TEXT NOT NULL,\n    converter_version   TEXT NOT NULL,\n    source_path         TEXT NOT NULL,\n    content_type        TEXT NOT NULL,\n    parent_item_type    TEXT,\n    converted_at        TEXT NOT NULL,\n    indexed_at          TEXT,\n    source_hash         TEXT NOT NULL,   -- SHA-256 of the source file\n    char_count          INTEGER NOT NULL,\n    parent_item_key     TEXT,\n    parse_quality_score REAL,\n    checkpoint_id       TEXT,\n    abstract            TEXT,\n    abstract_source     TEXT,\n    keywords            TEXT,\n    updated_at          TEXT\n);\n\nCREATE TABLE IF NOT EXISTS conversion_failures (\n    item_key       TEXT NOT NULL,\n    converter_id   TEXT NOT NULL,\n    error_message  TEXT NOT NULL,\n    source_hash    TEXT NOT NULL,\n    failed_at      TEXT NOT NULL,\n    PRIMARY KEY (item_key, converter_id)\n);\n\nCREATE TABLE IF NOT EXISTS item_relations (\n    source_key      TEXT NOT NULL,\n    target_key      TEXT NOT NULL,\n    score           REAL NOT NULL,\n    relation_type   TEXT NOT NULL,\n    match_evidence  TEXT,\n    updated_at      TEXT NOT NULL,\n    PRIMARY KEY (source_key, target_key, relation_type)\n);\nCREATE INDEX IF NOT EXISTS idx_relations_source_score ON item_relations (source_key, score DESC);\nCREATE INDEX IF NOT EXISTS idx_relations_target ON item_relations (target_key);\n'

_note_bundle_lock module-attribute

_note_bundle_lock = threading.Lock()

logger module-attribute

logger = logging.getLogger(__name__)

BaseDocumentConverter

Bases: ABC

Abstract base class for document conversion engines.

converter_id abstractmethod property

converter_id

Unique identifier for the converter strategy (e.g., 'markitdown', 'direct_read').

converter_version property

converter_version

Version string for cache invalidation. Override in subclasses when needed.

is_remote property

is_remote

Indicate whether this converter relies on an external/remote network service.

convert abstractmethod

convert(item_key, source_path, content_type, cache_dir)

Convert source document to Markdown text.

Parameters:

Name	Type	Description	Default
item_key	str	Zotero item key used for identification.	required
source_path	Path	Path to the source file on disk.	required
content_type	str	MIME type of the source file.	required
cache_dir	Path	Item-level bundle directory (corpus/<item_key>/). Converters should write extracted assets (images, tables) into subdirectories relative to this path (e.g., cache_dir / "images/").	required

Returns:

Type	Description
str | None	Converted Markdown text, or None if conversion failed or produced empty content.

Source code in zocrates/rag/document_converters.py

@abstractmethod
def convert(
    self,
    item_key: str,
    source_path: Path,
    content_type: str,
    cache_dir: Path,
) -> str | None:
    """Convert source document to Markdown text.

    Args:
        item_key: Zotero item key used for identification.
        source_path: Path to the source file on disk.
        content_type: MIME type of the source file.
        cache_dir: Item-level bundle directory (``corpus/<item_key>/``).
            Converters should write extracted assets (images, tables) into
            subdirectories relative to this path (e.g., ``cache_dir / "images/"``).

    Returns:
        Converted Markdown text, or None if conversion failed or produced empty content.
    """

supports abstractmethod

supports(content_type, source_path)

Return True if this converter can process the given document.

Parameters:

Name	Type	Description	Default
content_type	str	MIME type of the document (e.g., 'application/pdf').	required
source_path	Path	Path to the source file on disk.	required

Source code in zocrates/rag/document_converters.py

@abstractmethod
def supports(self, content_type: str, source_path: Path) -> bool:
    """Return True if this converter can process the given document.

    Args:
        content_type: MIME type of the document (e.g., 'application/pdf').
        source_path: Path to the source file on disk.
    """

DatalabConverter

DatalabConverter()

Bases: BaseDocumentConverter

Remote document conversion engine via Datalab OCR API.

Initialize the remote Datalab converter instance.

Source code in zocrates/rag/document_converters.py

def __init__(self) -> None:
    """Initialize the remote Datalab converter instance."""
    self._metadata: dict[str, dict[str, Any]] = {}

converter_id property

converter_id

Unique identifier for remote Datalab OCR converter strategy.

converter_version property

converter_version

Version string of the installed Datalab SDK.

is_remote property

is_remote

Indicate whether conversion relies on a remote cloud service.

convert

convert(item_key, source_path, content_type, cache_dir)

Convert document via Datalab API with adaptive retry and asset unpacking.

Includes an extension safeguard for valid PDFs lacking a .pdf extension.

Parameters:

Name	Type	Description	Default
item_key	str	Zotero item key.	required
source_path	Path	Source file path on disk.	required
content_type	str	MIME type string.	required
cache_dir	Path	Bundle directory (or temporary staging directory).	required

Returns:

Type	Description
str | None	Converted Markdown string with tags, or None on failure.

Source code in zocrates/rag/document_converters.py

def convert(
    self,
    item_key: str,
    source_path: Path,
    content_type: str,
    cache_dir: Path,
) -> str | None:
    """Convert document via Datalab API with adaptive retry and asset unpacking.

    Includes an extension safeguard for valid PDFs lacking a .pdf extension.

    Args:
        item_key: Zotero item key.
        source_path: Source file path on disk.
        content_type: MIME type string.
        cache_dir: Bundle directory (or temporary staging directory).

    Returns:
        Converted Markdown string with <figure> tags, or None on failure.
    """
    import tempfile

    if source_path.suffix.lower() != ".pdf":
        with tempfile.TemporaryDirectory() as tmpdir:
            staged_path = Path(tmpdir) / f"{source_path.stem}.pdf"
            shutil.copy2(source_path, staged_path)
            return self._convert_internal(item_key, staged_path, content_type, cache_dir)
    else:
        return self._convert_internal(item_key, source_path, content_type, cache_dir)

get_metadata

get_metadata(item_key)

Return the latest conversion metadata for an item key.

Parameters:

Name	Type	Description	Default
item_key	str	Zotero item key.	required

Returns:

Type	Description
dict[str, Any]	Dictionary containing parse_quality_score, checkpoint_id, etc.

Source code in zocrates/rag/document_converters.py

def get_metadata(self, item_key: str) -> dict[str, Any]:
    """Return the latest conversion metadata for an item key.

    Args:
        item_key: Zotero item key.

    Returns:
        Dictionary containing parse_quality_score, checkpoint_id, etc.
    """
    return self._metadata.get(item_key, {})

supports

supports(content_type, source_path)

Check if document format is a PDF supported by Datalab.

Strictly limits conversion to PDF documents only.

Parameters:

Name	Type	Description	Default
content_type	str	MIME type string.	required
source_path	Path	File path on disk.	required

Returns:

Type	Description
bool	True if file is a PDF, False otherwise.

Source code in zocrates/rag/document_converters.py

def supports(self, content_type: str, source_path: Path) -> bool:
    """Check if document format is a PDF supported by Datalab.

    Strictly limits conversion to PDF documents only.

    Args:
        content_type: MIME type string.
        source_path: File path on disk.

    Returns:
        True if file is a PDF, False otherwise.
    """
    return (
        content_type == "application/pdf"
        or source_path.suffix.lower() == ".pdf"
        or _is_pdf_file(source_path)
    )

DirectReadConverter

Bases: BaseDocumentConverter

Direct UTF-8 reader for Markdown and plain text files (<1ms bypass).

converter_id property

converter_id

Unique identifier for direct UTF-8 file reading strategy.

convert

convert(item_key, source_path, content_type, cache_dir)

Read Markdown or text content directly from disk.

Parameters:

Name	Type	Description	Default
item_key	str	Zotero item key.	required
source_path	Path	Path to the source file.	required
content_type	str	MIME type of the file.	required
cache_dir	Path	Item-level bundle directory (unused by this converter).	required

Returns:

Type	Description
str | None	UTF-8 content string, or None if empty.

Source code in zocrates/rag/document_converters.py

def convert(
    self,
    item_key: str,
    source_path: Path,
    content_type: str,
    cache_dir: Path,
) -> str | None:
    """Read Markdown or text content directly from disk.

    Args:
        item_key: Zotero item key.
        source_path: Path to the source file.
        content_type: MIME type of the file.
        cache_dir: Item-level bundle directory (unused by this converter).

    Returns:
        UTF-8 content string, or None if empty.
    """
    text = source_path.read_text(encoding="utf-8", errors="replace")
    if not text or not text.strip():
        logger.warning("Empty Markdown/Text file for %s", source_path)
        return None
    return text

supports

supports(content_type, source_path)

Check if content type or file extension corresponds to plain text or Markdown.

Parameters:

Name	Type	Description	Default
content_type	str	MIME type string.	required
source_path	Path	File path on disk.	required

Returns:

Type	Description
bool	True if file is plain text or Markdown, False otherwise.

Source code in zocrates/rag/document_converters.py

def supports(self, content_type: str, source_path: Path) -> bool:
    """Check if content type or file extension corresponds to plain text or Markdown.

    Args:
        content_type: MIME type string.
        source_path: File path on disk.

    Returns:
        True if file is plain text or Markdown, False otherwise.
    """
    return (
        content_type in ("text/markdown", "text/plain")
        or source_path.suffix.lower() in (".md", ".txt", ".markdown")
    )

DoclingLocalConverter

DoclingLocalConverter(converter_instance=None)

Bases: BaseDocumentConverter

Document converter using IBM Docling for deep layout analysis and table extraction.

Initialize the DoclingLocalConverter strategy.

Parameters:

Name	Type	Description	Default
converter_instance	Any | None	Optional pre-configured Docling DocumentConverter instance.	None

Source code in zocrates/rag/document_converters.py

def __init__(self, converter_instance: Any | None = None) -> None:
    """Initialize the DoclingLocalConverter strategy.

    Args:
        converter_instance: Optional pre-configured Docling DocumentConverter instance.
    """
    self._converter = converter_instance
    self._cpu_converter = None

converter_id property

converter_id

Unique identifier for local IBM Docling converter strategy.

converter_version property

converter_version

Return the installed docling package version with mode indicator.

convert

convert(item_key, source_path, content_type, cache_dir)

Convert document using Docling and save AST, images, and tables to bundle.

Parameters:

Name	Type	Description	Default
item_key	str	Zotero item key.	required
source_path	Path	Path to the source file on disk.	required
content_type	str	MIME type of the file.	required
cache_dir	Path	Item-level bundle directory (corpus/<item_key>/).	required

Returns:

Type	Description
str | None	Converted Markdown string, or None if conversion failed or returned empty text.

Source code in zocrates/rag/document_converters.py

def convert(
    self,
    item_key: str,
    source_path: Path,
    content_type: str,
    cache_dir: Path,
) -> str | None:
    """Convert document using Docling and save AST, images, and tables to bundle.

    Args:
        item_key: Zotero item key.
        source_path: Path to the source file on disk.
        content_type: MIME type of the file.
        cache_dir: Item-level bundle directory (``corpus/<item_key>/``).

    Returns:
        Converted Markdown string, or None if conversion failed or returned empty text.
    """
    doc = None
    try:
        conv = self._get_converter()
        conv_result = conv.convert(str(source_path))
        doc = conv_result.document
    except (ImportError, Exception) as exc:
        if _is_vc_redist_dll_failure(exc):
            logger.error(
                "Missing Microsoft Visual C++ Redistributable dependency during Docling conversion: %s",
                exc,
            )
            raise RuntimeError(VC_REDIST_WINDOWS_MSG) from exc
        raw_exc_str = str(exc)
        exc_str = raw_exc_str.lower()
        is_oom = "out of memory" in exc_str or "cuda" in exc_str or "alloc" in exc_str
        if is_oom:
            logger.warning(
                "Docling failed with device/OOM error on %s (%s). Attempting CPU fallback...",
                source_path.name,
                exc,
            )
            import gc
            gc.collect()
            try:
                import torch
                if torch.cuda.is_available():
                    torch.cuda.empty_cache()
                    if hasattr(torch.cuda, "ipc_collect"):
                        torch.cuda.ipc_collect()
            except Exception:
                pass

            try:
                cpu_conv = self._get_converter(force_cpu=True)
                conv_result = cpu_conv.convert(str(source_path))
                doc = conv_result.document
            except (ImportError, Exception) as cpu_exc:
                if _is_vc_redist_dll_failure(cpu_exc):
                    logger.error(
                        "Missing Microsoft Visual C++ Redistributable dependency during Docling CPU fallback: %s",
                        cpu_exc,
                    )
                    raise RuntimeError(VC_REDIST_WINDOWS_MSG) from cpu_exc
                logger.error("Docling CPU fallback failed for %s: %s", source_path, cpu_exc)
                gc.collect()
                try:
                    import torch
                    if torch.cuda.is_available():
                        torch.cuda.empty_cache()
                except Exception:
                    pass
                return None
        else:
            logger.error("Docling conversion failed for %s: %s", source_path, exc)
            return None

    if doc is None:
        logger.warning("Docling returned empty document for %s", source_path)
        return None

    # 1. Export Markdown
    try:
        markdown = doc.export_to_markdown()
    except Exception as exc:
        logger.error("Docling markdown export failed for %s: %s", source_path, exc)
        return None

    if not markdown or not markdown.strip():
        logger.warning("Docling produced empty markdown for %s", source_path)
        return None

    # 2. Save full Docling document JSON AST to item bundle
    try:
        json_path = cache_dir / "docling_document.json"
        if hasattr(doc, "save_as_json"):
            doc.save_as_json(json_path)
        elif hasattr(doc, "model_dump_json"):
            json_path.write_text(doc.model_dump_json(), encoding="utf-8")
    except Exception as exc:
        logger.warning("Failed to save docling_document.json for %s: %s", item_key, exc)

    # 3. Extract and save pictures
    saved_images: list[tuple[int, Path]] = []
    alt_map: dict[int, str] = {}
    try:
        pictures = getattr(doc, "pictures", [])
        images_dir = cache_dir / "images"
        if images_dir.exists():
            shutil.rmtree(images_dir)
        if pictures:
            images_dir.mkdir(parents=True, exist_ok=True)
            for idx, pic in enumerate(pictures):
                try:
                    img = pic.get_image(doc) if hasattr(pic, "get_image") else None
                    if img is not None:
                        img_path = images_dir / f"picture_{idx:03d}.png"
                        img.save(img_path, format="PNG")
                        saved_images.append((idx, img_path.resolve()))
                except Exception as p_exc:
                    logger.debug(
                        "Could not extract image %d for %s: %s", idx, item_key, p_exc
                    )

        # If doc.pictures yielded no images (standard for HTML in Docling),
        # check if the source is HTML and extract embedded/local HTML images:
        is_html = (
            content_type in ("text/html", "application/xhtml+xml")
            or source_path.suffix.lower() in (".html", ".htm", ".xhtml")
        )
        if not saved_images and is_html:
            html_images = _extract_html_images(source_path, images_dir)
            saved_images = [(idx, img_p) for idx, img_p, _ in html_images]
            alt_map = {idx: alt for idx, _, alt in html_images if alt}
    except Exception as exc:
        logger.warning("Failed extracting pictures for %s: %s", item_key, exc)

    # Inject extracted image links into markdown (replacing <!-- image --> placeholders)
    canonical_dir = (
        cache_dir.parent / item_key
        if cache_dir.name.startswith(".tmp_")
        else cache_dir
    )
    markdown = _inject_docling_images(
        markdown, doc, saved_images, canonical_dir=canonical_dir, alt_map=alt_map
    )

    # 4. Extract and save tables (CSV & HTML)
    saved_tables: list[tuple[int, Path | None, Path | None]] = []
    try:
        tables = getattr(doc, "tables", [])
        tables_dir = cache_dir / "tables"
        if tables_dir.exists():
            shutil.rmtree(tables_dir)
        if tables:
            tables_dir.mkdir(parents=True, exist_ok=True)
            for idx, table in enumerate(tables):
                csv_path = None
                html_path = None
                try:
                    if hasattr(table, "export_to_dataframe"):
                        df = table.export_to_dataframe(doc=doc)
                        if df is not None and not df.empty:
                            csv_path = tables_dir / f"table_{idx:03d}.csv"
                            df.to_csv(csv_path, index=False)
                    if hasattr(table, "export_to_html"):
                        html_content = table.export_to_html(doc=doc)
                        if html_content:
                            html_path = tables_dir / f"table_{idx:03d}.html"
                            html_path.write_text(html_content, encoding="utf-8")
                    saved_tables.append((idx, csv_path, html_path))
                except Exception as t_exc:
                    logger.debug(
                        "Could not extract table %d for %s: %s", idx, item_key, t_exc
                    )
    except Exception as exc:
        logger.warning("Failed extracting tables for %s: %s", item_key, exc)

    # Wrap tables in <figure data-type="table"> blocks
    markdown = _inject_docling_tables(markdown, doc, saved_tables)

    # Release AST references and collect garbage to prevent memory retention
    del doc
    del conv_result
    import gc
    gc.collect()
    try:
        import torch
        if torch.cuda.is_available():
            torch.cuda.empty_cache()
    except Exception:
        pass

    return markdown

supports

supports(content_type, source_path)

Check if content type or file extension is supported by Docling.

Supports PDF, HTML, DOCX, PPTX, XLSX, etc.

Source code in zocrates/rag/document_converters.py

def supports(self, content_type: str, source_path: Path) -> bool:
    """Check if content type or file extension is supported by Docling.

    Supports PDF, HTML, DOCX, PPTX, XLSX, etc.
    """
    suffix = source_path.suffix.lower()
    supported_types = {
        "application/pdf",
        "text/html",
        "application/xhtml+xml",
        "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
        "application/vnd.openxmlformats-officedocument.presentationml.presentation",
        "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
    }
    supported_extensions = {
        ".pdf",
        ".html",
        ".htm",
        ".xhtml",
        ".docx",
        ".pptx",
        ".xlsx",
        ".asciidoc",
    }
    return content_type in supported_types or suffix in supported_extensions

DocumentProcessor

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

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

Initialize the DocumentProcessor.

Parameters:

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

Source code in zocrates/rag/document_processor.py

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

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

cache_dir property

cache_dir

Return the root markdown cache directory Path.

datalab_non_pdf_fallback property writable

datalab_non_pdf_fallback

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

clear_failure

clear_failure(item_key, converter_id=None)

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

Parameters:

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

Source code in zocrates/rag/document_processor.py

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

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

convert_with

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

Convert a document with a specific converter engine.

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

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

Parameters:

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

Returns:

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

Raises:

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

Source code in zocrates/rag/document_processor.py

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

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

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

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

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

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

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

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

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

    current_hash = _file_sha256(source_path)

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

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

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

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

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

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

delete_cached

delete_cached(item_key, update_root_index=True)

Delete an item bundle from disk and conversion_meta DB.

Parameters:

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

Returns:

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

Source code in zocrates/rag/document_processor.py

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

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

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

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

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

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

    if deleted and update_root_index:
        generate_root_index_md(self._cache_dir)

    return deleted

ensure_parent_item_key

ensure_parent_item_key(item_key, parent_item_key)

Ensure an item in conversion_meta has parent_item_key populated.

Source code in zocrates/rag/document_processor.py

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

get_all_meta_keys

get_all_meta_keys()

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

Source code in zocrates/rag/document_processor.py

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

get_all_metadata

get_all_metadata()

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

Returns:

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

Source code in zocrates/rag/document_processor.py

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

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

get_failures

get_failures(item_key)

Retrieve recorded conversion failures for an item key.

Parameters:

Name	Type	Description	Default
item_key	str	Zotero item key.	required

Returns:

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

Source code in zocrates/rag/document_processor.py

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

    Args:
        item_key: Zotero item key.

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

get_metadata

get_metadata(item_key)

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

Source code in zocrates/rag/document_processor.py

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

get_or_convert

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

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

Parameters:

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

Returns:

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

Source code in zocrates/rag/document_processor.py

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

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

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

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

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

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

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

is_cached

is_cached(item_key)

Return True if a converted Markdown file exists for item_key.

Does not validate freshness against the source file hash.

Parameters:

Name	Type	Description	Default
item_key	str	Zotero item key.	required

Returns:

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

Source code in zocrates/rag/document_processor.py

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

    Does not validate freshness against the source file hash.

    Args:
        item_key: Zotero item key.

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

list_cached_keys

list_cached_keys()

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

Returns:

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

Source code in zocrates/rag/document_processor.py

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

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

read_cached

read_cached(item_key)

Read cached Markdown without triggering conversion.

Source code in zocrates/rag/document_processor.py

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

record_failure

record_failure(item_key, converter_id, error_message, source_hash)

Record or update a failed conversion attempt for an item.

Parameters:

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

Source code in zocrates/rag/document_processor.py

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

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

regenerate_root_index

regenerate_root_index()

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

Source code in zocrates/rag/document_processor.py

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

register_converter

register_converter(converter, prepend=True)

Register a new document converter strategy.

Parameters:

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

Source code in zocrates/rag/document_processor.py

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

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

update_abstract

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

Update abstract and abstract_source in conversion_meta for an item.

Source code in zocrates/rag/document_processor.py

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

update_indexed_at

update_indexed_at(item_key, timestamp=None)

Update the indexed_at timestamp for item_key in conversion_meta DB.

Parameters:

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

Source code in zocrates/rag/document_processor.py

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

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

MarkItDownConverter

MarkItDownConverter(markitdown_instance=None)

Bases: BaseDocumentConverter

Document converter using Microsoft MarkItDown and pdfimages for embedded images.

Initialize the MarkItDownConverter strategy.

Parameters:

Name	Type	Description	Default
markitdown_instance	Any | None	Optional pre-configured MarkItDown instance.	None

Source code in zocrates/rag/document_converters.py

def __init__(self, markitdown_instance: Any | None = None) -> None:
    """Initialize the MarkItDownConverter strategy.

    Args:
        markitdown_instance: Optional pre-configured MarkItDown instance.
    """
    self._md = markitdown_instance

converter_id property

converter_id

Unique identifier for Microsoft MarkItDown converter strategy.

converter_version property

converter_version

Return the installed MarkItDown package version.

convert

convert(item_key, source_path, content_type, cache_dir)

Convert document using MarkItDown and inject extracted PDF images if applicable.

Parameters:

Name	Type	Description	Default
item_key	str	Zotero item key.	required
source_path	Path	Path to the source file on disk.	required
content_type	str	MIME type of the file.	required
cache_dir	Path	Item-level bundle directory (corpus/<item_key>/).	required

Returns:

Type	Description
str | None	Converted Markdown string, or None if conversion failed or returned empty text.

Source code in zocrates/rag/document_converters.py

def convert(
    self,
    item_key: str,
    source_path: Path,
    content_type: str,
    cache_dir: Path,
) -> str | None:
    """Convert document using MarkItDown and inject extracted PDF images if applicable.

    Args:
        item_key: Zotero item key.
        source_path: Path to the source file on disk.
        content_type: MIME type of the file.
        cache_dir: Item-level bundle directory (``corpus/<item_key>/``).

    Returns:
        Converted Markdown string, or None if conversion failed or returned empty text.
    """
    try:
        if self._md is None:
            markitdown_cls = getattr(sys.modules[__name__], "MarkItDown")
            self._md = markitdown_cls()
        result = self._md.convert(str(source_path))
    except (ImportError, Exception) as exc:
        if _is_vc_redist_dll_failure(exc):
            logger.error(
                "Missing Microsoft Visual C++ Redistributable dependency during MarkItDown conversion: %s",
                exc,
            )
            raise RuntimeError(VC_REDIST_WINDOWS_MSG) from exc
        raise

    text = result.text_content
    if not text or not text.strip():
        logger.warning("MarkItDown returned empty content for %s", source_path)
        return None

    # For PDFs, extract and inject images
    if content_type == "application/pdf":
        images_dir = cache_dir / "images"
        if images_dir.exists():
            shutil.rmtree(images_dir)

        images_by_page = _extract_pdf_images(source_path, images_dir)
        text = _embed_images_in_markdown(text, images_by_page)

    # For HTML documents, extract embedded data URIs / local images and wrap in <figure> blocks
    is_html = (
        content_type in ("text/html", "application/xhtml+xml")
        or source_path.suffix.lower() in (".html", ".htm", ".xhtml")
    )
    if is_html:
        images_dir = cache_dir / "images"
        if images_dir.exists():
            shutil.rmtree(images_dir)

        html_images = _extract_html_images(source_path, images_dir)
        canonical_dir = (
            cache_dir.parent / item_key
            if cache_dir.name.startswith(".tmp_")
            else cache_dir
        )
        text = _inject_html_images_into_markdown(
            text, html_images, canonical_dir=canonical_dir
        )

    return text

supports

supports(content_type, source_path)

MarkItDown serves as the default fallback converter for all supported documents.

Parameters:

Name	Type	Description	Default
content_type	str	MIME type of the document.	required
source_path	Path	Path to the source file on disk.	required

Returns:

Type	Description
bool	Always returns True.

Source code in zocrates/rag/document_converters.py

def supports(self, content_type: str, source_path: Path) -> bool:
    """MarkItDown serves as the default fallback converter for all supported documents.

    Args:
        content_type: MIME type of the document.
        source_path: Path to the source file on disk.

    Returns:
        Always returns True.
    """
    return True

UnsupportedConversionError

Bases: Exception

Raised when a converter does not support the given content type.

_file_sha256

_file_sha256(path)

Compute the SHA-256 hex digest of a file.

Reads in 64 KiB chunks to keep memory usage constant for large PDFs.

Parameters:

Name	Type	Description	Default
path	Path	Absolute path to the file to hash.	required

Returns:

Type	Description
str	Lowercase hex string of the SHA-256 digest.

Source code in zocrates/rag/document_processor.py

def _file_sha256(path: Path) -> str:
    """Compute the SHA-256 hex digest of a file.

    Reads in 64 KiB chunks to keep memory usage constant for large PDFs.

    Args:
        path: Absolute path to the file to hash.

    Returns:
        Lowercase hex string of the SHA-256 digest.
    """
    h = hashlib.sha256()
    with path.open("rb") as f:
        for chunk in iter(lambda: f.read(65536), b""):
            h.update(chunk)
    return h.hexdigest()

_migrate_storage_paths_to_relative

_migrate_storage_paths_to_relative(db_path, conn=None)

Migrate legacy absolute source_path entries to storage-relative paths.

Idempotent and atomic. Upgrades PRAGMA user_version from 0 to 1. Creates a pre-migration snapshot backup if not already present.

Source code in zocrates/rag/document_processor.py

def _migrate_storage_paths_to_relative(
    db_path: Path, conn: sqlite3.Connection | None = None
) -> None:
    """Migrate legacy absolute source_path entries to storage-relative paths.

    Idempotent and atomic. Upgrades PRAGMA user_version from 0 to 1.
    Creates a pre-migration snapshot backup if not already present.
    """
    close_conn = False
    if conn is None:
        conn = sqlite3.connect(str(db_path))
        conn.row_factory = sqlite3.Row
        close_conn = True

    try:
        conn.execute("PRAGMA busy_timeout = 5000;")
        cursor = conn.execute("PRAGMA user_version;")
        version = cursor.fetchone()[0]
        if version >= 1:
            return

        # Pre-migration snapshot backup
        bak_path = db_path.with_name(f"{db_path.name}.bak-pre-relativization")
        if not bak_path.exists() and db_path.exists():
            try:
                bck = sqlite3.connect(str(bak_path))
                conn.backup(bck)
                bck.close()
            except Exception:
                try:
                    shutil.copy2(db_path, bak_path)
                except Exception as exc:
                    logger.warning("Could not create pre-migration backup %s: %s", bak_path, exc)

        cursor = conn.execute(
            "SELECT item_key, source_path FROM conversion_meta "
            "WHERE source_path NOT LIKE 'storage/%' AND source_path NOT LIKE 'zotero://note/%';"
        )
        rows = cursor.fetchall()
        updates = []
        pattern = re.compile(
            r"(?:^|[/\\])storage[/\\](?P<key>[A-Za-z0-9_-]+)(?:[/\\](?P<filename>.+))?$"
        )

        for row in rows:
            key = row["item_key"]
            src = row["source_path"] or ""
            unquoted_src = urllib.parse.unquote(src)
            m = pattern.search(unquoted_src) or pattern.search(src)
            # Key-equality guard: only rewrite when captured key matches the row's item_key
            if m and m.group("key") == key:
                match_key = m.group("key")
                match_file = (m.group("filename") or "").replace("\\", "/").rstrip("/")
                rel_path = (
                    f"storage/{match_key}/{match_file}"
                    if match_file
                    else f"storage/{match_key}"
                )
                updates.append((rel_path, key))

        conn.execute("BEGIN IMMEDIATE;")
        if updates:
            conn.executemany(
                "UPDATE conversion_meta SET source_path = ? WHERE item_key = ?;",
                updates,
            )
        conn.execute("PRAGMA user_version = 1;")
        conn.commit()
    finally:
        if close_conn:
            conn.close()

_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

_relativize_staging_markdown_links

_relativize_staging_markdown_links(markdown, tmp_dir)

Rewrite any temporary staging directory paths/URIs to bundle-relative images/ links.

Source code in zocrates/rag/document_processor.py

def _relativize_staging_markdown_links(markdown: str, tmp_dir: Path) -> str:
    """Rewrite any temporary staging directory paths/URIs to bundle-relative images/ links."""
    tmp_uri = tmp_dir.as_uri()
    tmp_str = str(tmp_dir)
    tmp_posix = tmp_dir.as_posix()

    for prefix in (tmp_uri, tmp_str, tmp_posix):
        markdown = markdown.replace(f"{prefix}/images/", "images/")
        markdown = markdown.replace(f"{prefix}/", "")

    def _sanitize_img(m: re.Match) -> str:
        """Sanitize file:// URL image paths to relative images/<name> format.

        Args:
            m: Regex match object containing the raw image filename.

        Returns:
            Relative markdown image path.
        """
        return f"images/{urllib.parse.unquote(m.group(1))}"

    markdown = re.sub(
        r"file://[^\s\)\"']+?[/\\]images[/\\]([^\s\)\"']+)",
        _sanitize_img,
        markdown,
    )
    return markdown

_sanitize_cached_bundle_image_links

_sanitize_cached_bundle_image_links(cache_dir)

Sanitize legacy file:///.../images/ absolute links in cached document.md files.

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

Parameters:

Name	Type	Description	Default
cache_dir	Path	Root corpus directory.	required

Returns:

Type	Description
int	Count of sanitized document bundles.

Source code in zocrates/rag/document_processor.py

def _sanitize_cached_bundle_image_links(cache_dir: Path) -> int:
    """Sanitize legacy file:///.../images/ absolute links in cached document.md files.

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

    Args:
        cache_dir: Root corpus directory.

    Returns:
        Count of sanitized document bundles.
    """
    if not cache_dir.exists():
        return 0

    sanitized_count = 0
    for item_dir in cache_dir.iterdir():
        if not item_dir.is_dir() or item_dir.name.startswith("."):
            continue
        doc_file = item_dir / "document.md"
        if not doc_file.exists():
            continue

        try:
            content = doc_file.read_text(encoding="utf-8")
        except Exception as exc:
            logger.debug("Could not read %s for image sanitization: %s", doc_file, exc)
            continue

        new_content = sanitize_image_links_in_markdown(content)
        if new_content == content:
            continue

        old_lines = content.splitlines(keepends=True)
        new_lines = new_content.splitlines(keepends=True)

        if len(old_lines) != len(new_lines):
            logger.warning(
                "Line count mismatch when sanitizing %s (%d vs %d lines); aborting edit.",
                doc_file,
                len(old_lines),
                len(new_lines),
            )
            continue

        # Atomic intra-line replacement
        tmp_file = item_dir / "document.md.tmp"
        try:
            tmp_file.write_text(new_content, encoding="utf-8")
            tmp_file.replace(doc_file)
            sanitized_count += 1
        except Exception as exc:
            logger.warning("Failed writing sanitized %s: %s", doc_file, exc)
            if tmp_file.exists():
                tmp_file.unlink(missing_ok=True)

    return sanitized_count

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"

convert_note_html_to_markdown

convert_note_html_to_markdown(html_content)

Convert Zotero note HTML content to clean, structured Markdown.

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

Source code in zocrates/rag/document_processor.py

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

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

    import html as html_module

    try:
        from bs4 import BeautifulSoup, NavigableString, Tag

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

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

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

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

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

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

        result = _node_to_md(soup)

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

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

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)

materialize_note_bundle

materialize_note_bundle(note, syncer, cache_dir=None)

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

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

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

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

Parameters:

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

Returns:

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

Source code in zocrates/rag/document_processor.py

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

        return note_dir, note.note_key

normalize_to_storage_relative

normalize_to_storage_relative(source_path, zotero_profile_dir=None)

Normalize a source document path to storage-relative format or verbatim URI.

Normalizes storage paths to 'storage//' (or 'storage/'). Virtual note URIs (zotero://note/...) and remote URIs are preserved verbatim. External linked files outside storage retain their normalized absolute POSIX paths.

Parameters:

Name	Type	Description	Default
source_path	str | Path | None	File path on disk, URI, or storage path string.	required
zotero_profile_dir	Path | None	Optional root Zotero profile directory.	None

Returns:

Type	Description
str	Normalized storage-relative path string, URI, or absolute POSIX path.

Source code in zocrates/rag/bundle_catalog.py

def normalize_to_storage_relative(
    source_path: str | Path | None,
    zotero_profile_dir: Path | None = None,
) -> str:
    """Normalize a source document path to storage-relative format or verbatim URI.

    Normalizes storage paths to 'storage/<item_key>/<filename>' (or 'storage/<item_key>').
    Virtual note URIs (zotero://note/...) and remote URIs are preserved verbatim.
    External linked files outside storage retain their normalized absolute POSIX paths.

    Args:
        source_path: File path on disk, URI, or storage path string.
        zotero_profile_dir: Optional root Zotero profile directory.

    Returns:
        Normalized storage-relative path string, URI, or absolute POSIX path.
    """
    if not source_path or not str(source_path).strip():
        return ""

    str_path = str(source_path).strip()
    if str_path.startswith("zotero://note/"):
        return str_path

    if "://" in str_path:
        if str_path.startswith("file://"):
            str_path = urllib.parse.unquote(str_path[7:])
            if str_path.startswith("/") and len(str_path) > 2 and str_path[2] == ":":
                str_path = str_path[1:]
        elif not str_path.startswith("zotero:"):
            return str_path

    # Decode any remaining percent-encoding (e.g. %20)
    str_path = urllib.parse.unquote(str_path)

    # Check if path is within zotero_profile_dir / "storage"
    if zotero_profile_dir is not None:
        try:
            storage_dir = (zotero_profile_dir / "storage").resolve()
            resolved = Path(str_path).resolve()
            if resolved.is_relative_to(storage_dir):
                rel = resolved.relative_to(zotero_profile_dir)
                return Path(rel).as_posix()
        except (ValueError, OSError):
            pass

    # Regex heuristic matching storage/<key>/<filename> or storage/<key>
    m = re.search(r"(?:^|[/\\])storage[/\\]([A-Za-z0-9_-]+)(?:[/\\](.+))?$", str_path)
    if m:
        key = m.group(1)
        rest = m.group(2)
        if rest:
            clean_rest = rest.replace("\\", "/").rstrip("/")
            return f"storage/{key}/{clean_rest}"
        return f"storage/{key}"

    # External linked file outside storage: retain normalized absolute POSIX path
    return Path(str_path).as_posix()

regenerate_item_relations_md

regenerate_item_relations_md(item_key, cache_dir, target_bundle_dir=None)

Regenerate item_relations.md and refresh index.md for a given bundle.

Queries item_relations from conversion_meta.db for source_key = item_key. Also queries conversion_meta for target titles (fallback to target_key if not found). Target directory is target_bundle_dir if provided, else cache_dir / item_key. If target directory does not exist, returns None. Writes relations markdown atomically using .tmp_item_relations.md and os.replace (P1-4 pattern). Refreshes index.md catalog with the new/updated item_relations.md link.

Parameters:

Name	Type	Description	Default
item_key	str	Zotero item key.	required
cache_dir	Path	Markdown cache directory containing conversion_meta.db.	required
target_bundle_dir	Path | None	Optional bundle directory override (e.g. staging dir).	None

Returns:

Type	Description
Path | None	Path to the materialized item_relations.md, or None if target directory does not exist.

Source code in zocrates/rag/bundle_catalog.py

def regenerate_item_relations_md(
    item_key: str,
    cache_dir: Path,
    target_bundle_dir: Path | None = None,
) -> Path | None:
    """Regenerate item_relations.md and refresh index.md for a given bundle.

    Queries item_relations from conversion_meta.db for source_key = item_key.
    Also queries conversion_meta for target titles (fallback to target_key if not found).
    Target directory is target_bundle_dir if provided, else cache_dir / item_key.
    If target directory does not exist, returns None.
    Writes relations markdown atomically using .tmp_item_relations.md and os.replace (P1-4 pattern).
    Refreshes index.md catalog with the new/updated item_relations.md link.

    Args:
        item_key: Zotero item key.
        cache_dir: Markdown cache directory containing conversion_meta.db.
        target_bundle_dir: Optional bundle directory override (e.g. staging dir).

    Returns:
        Path to the materialized item_relations.md, or None if target directory does not exist.
    """
    bundle_dir = target_bundle_dir if target_bundle_dir is not None else (cache_dir / item_key)
    if not bundle_dir.exists():
        return None

    db_path = cache_dir / "conversion_meta.db"
    relations: list[dict[str, Any]] = []
    titles: dict[str, str] = {}
    self_title: str | None = None
    self_converter_id = "unknown"
    self_converter_version = "1.0"

    if db_path.exists():
        try:
            with contextlib.closing(sqlite3.connect(str(db_path), timeout=5.0)) as conn:
                conn.row_factory = sqlite3.Row
                conn.execute("PRAGMA busy_timeout = 5000;")
                try:
                    rows = conn.execute(
                        """
                        SELECT source_key, target_key, score, relation_type, match_evidence, updated_at
                        FROM item_relations
                        WHERE source_key = ?
                        ORDER BY score DESC, target_key ASC
                        """,
                        (item_key,),
                    ).fetchall()
                    relations = [dict(r) for r in rows]
                except sqlite3.OperationalError:
                    relations = []

                target_keys = {r["target_key"] for r in relations if r.get("target_key")}
                if target_keys:
                    placeholders = ",".join("?" for _ in target_keys)
                    try:
                        meta_rows = conn.execute(
                            f"SELECT item_key, title FROM conversion_meta WHERE item_key IN ({placeholders})",
                            tuple(target_keys),
                        ).fetchall()
                        for mr in meta_rows:
                            if mr["title"]:
                                titles[mr["item_key"]] = mr["title"]
                    except sqlite3.OperationalError:
                        pass

                try:
                    self_row = conn.execute(
                        "SELECT title, converter_id, converter_version FROM conversion_meta WHERE item_key = ?",
                        (item_key,),
                    ).fetchone()
                    if self_row:
                        self_title = self_row["title"]
                        self_converter_id = self_row["converter_id"] or self_converter_id
                        self_converter_version = self_row["converter_version"] or self_converter_version
                except sqlite3.OperationalError:
                    pass
        except Exception as exc:
            logger.warning("Failed to query conversion_meta.db for relations of %s: %s", item_key, exc)

    if not self_title and (bundle_dir / "document.md").exists():
        try:
            with (bundle_dir / "document.md").open("r", encoding="utf-8") as f:
                for _ in range(30):
                    line = f.readline()
                    if not line:
                        break
                    m = re.match(r"^title:\s*['\"]?(.*?)['\"]?\s*$", line.strip(), re.IGNORECASE)
                    if m:
                        self_title = m.group(1).strip()
                        break
        except Exception:
            pass

    content = generate_item_relations_md(item_key=item_key, relations=relations, titles=titles, cache_dir=cache_dir)

    target_path = bundle_dir / "item_relations.md"
    tmp_path = bundle_dir / f".tmp_{os.getpid()}_{threading.get_ident()}_{uuid.uuid4().hex[:8]}_item_relations.md"
    try:
        tmp_path.write_text(content, encoding="utf-8")
        os.replace(tmp_path, target_path)
    except Exception as exc:
        if tmp_path.exists():
            tmp_path.unlink(missing_ok=True)
        raise exc

    # Refresh index.md catalog with the new/updated item_relations.md link
    generate_item_index_md(
        item_dir=bundle_dir,
        item_key=item_key,
        title=self_title,
        converter_id=self_converter_id,
        converter_version=self_converter_version,
    )

    return target_path

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)

strip_existing_frontmatter

strip_existing_frontmatter(markdown)

Strip leading YAML frontmatter block if present.

Parameters:

Name	Type	Description	Default
markdown	str	Input Markdown string.	required

Returns:

Type	Description
str	Markdown content with leading YAML frontmatter removed.

Source code in zocrates/rag/bundle_catalog.py

def strip_existing_frontmatter(markdown: str) -> str:
    """Strip leading YAML frontmatter block if present.

    Args:
        markdown: Input Markdown string.

    Returns:
        Markdown content with leading YAML frontmatter removed.
    """
    m = _FRONTMATTER_RE.match(markdown)
    if m:
        return markdown[m.end() :].lstrip("\r\n")
    return markdown

wrap_standalone_markdown_tables

wrap_standalone_markdown_tables(markdown_text)

Wrap standalone GFM Markdown tables not already enclosed in

tags.

Parameters:

Name	Type	Description	Default
markdown_text	str	Raw or converted Markdown text.	required

Returns:

Type	Description
str	Markdown with standalone tables enclosed in containers.

Source code in zocrates/rag/document_converters.py

def wrap_standalone_markdown_tables(markdown_text: str) -> str:
    """Wrap standalone GFM Markdown tables not already enclosed in <figure> tags.

    Args:
        markdown_text: Raw or converted Markdown text.

    Returns:
        Markdown with standalone tables enclosed in <figure data-type="table"> containers.
    """
    if not markdown_text:
        return markdown_text

    table_pattern = re.compile(
        r"(?:^|\n)((?:\|[^\n]+\|\r?\n)\|[\s\-:|]+\|(?:\r?\n\|[^\n]+\|)+)",
        re.MULTILINE,
    )

    def _replacer(match: re.Match) -> str:
        """Wrap an unencapsulated markdown table in semantic figure tags.

        Args:
            match: Regex match object containing the detected markdown table.

        Returns:
            Wrapped table string or original match if already inside a figure block.
        """
        matched_table = match.group(1).strip()
        start_pos = match.start()
        prefix = markdown_text[:start_pos]
        last_open = prefix.rfind("<figure")
        last_close = prefix.rfind("</figure>")
        if last_open > last_close:
            return match.group(0)

        return f'\n\n<figure data-type="table">\n\n{matched_table}\n\n</figure>\n\n'

    return table_pattern.sub(_replacer, markdown_text)