Function convert_document

openkb/converter.py:142–247

Convert a document and integrate it into the knowledge base. Steps: 1. Hash-check — skip if already known. 2. Copy source to ``raw/``. 3. If PDF and page count >= threshold → return :attr:`ConvertResult.is_long_doc`. 4. If ``.md`` — read, process relative images, save to ``wiki/

(
    src: Path,
    kb_dir: Path,
    *,
    staging_dir: Path | None = None,
)

def convert_document(
    src: Path,
    kb_dir: Path,
    *,
    staging_dir: Path | None = None,
) -> ConvertResult:
    """Convert a document and integrate it into the knowledge base.

    Steps:
    1. Hash-check — skip if already known.
    2. Copy source to ``raw/``.
    3. If PDF and page count >= threshold → return :attr:`ConvertResult.is_long_doc`.
    4. If ``.md`` — read, process relative images, save to ``wiki/sources/``.
    5. Otherwise — run MarkItDown, extract base64 images, save to ``wiki/sources/``.
    6. Register hash in the registry.
    """
    with kb_ingest_lock(kb_dir / ".openkb"):
        # ------------------------------------------------------------------
        # Load config & state
        # ------------------------------------------------------------------
        openkb_dir = kb_dir / ".openkb"
        config = load_config(openkb_dir / "config.yaml")
        threshold: int = config.get("pageindex_threshold", 20)
        artifact_root = staging_dir if staging_dir is not None else kb_dir
        registry = HashRegistry(openkb_dir / "hashes.json")

        # ------------------------------------------------------------------
        # 1. Hash check + identity resolution
        # ------------------------------------------------------------------
        file_hash = HashRegistry.hash_file(src)
        if registry.is_known(file_hash):
            logger.info("Skipping already-known file: %s", src.name)
            stored = registry.get(file_hash) or {}
            return ConvertResult(
                skipped=True,
                file_hash=file_hash,
                doc_name=stored.get("doc_name") or Path(stored.get("name", src.name)).stem,
            )
        doc_name = resolve_doc_name(
            src,
            kb_dir,
            registry,
            persist_legacy=staging_dir is None,
        )

        # ------------------------------------------------------------------
        # 2. Copy to raw/
        # ------------------------------------------------------------------
        raw_dir = artifact_root / "raw"
        raw_dir.mkdir(parents=True, exist_ok=True)
        if staging_dir is None and src.resolve().is_relative_to(raw_dir.resolve()):
            # Watch mode: the file already lives in raw/ — don't copy/rename.
            raw_dest = src
        else:
            raw_dest = raw_dir / f"{doc_name}{src.suffix.lower()}"
            shutil.copy2(src, raw_dest)

        # ------------------------------------------------------------------