zettelkasten.import_vault

zettelkasten.import_vault

Bulk import of an Obsidian / markdown vault into a zettelkasten box.

This is the ONE untrusted-input surface in the WS5 export/import plan: every byte of every file is treated as DATA, never code. The module therefore layers several guards over the parse:

• Path containment — source_dir is resolved to a realpath and every candidate file is confirmed to stay under that root (rejecting .. traversal AND symlink escape).
• Resource caps — a per-file byte cap and a total file-count cap (dir-bomb guard) are enforced before any file is read.
• Safe parsing — frontmatter is parsed with :func:yaml.safe_load only (never yaml.load); note content is never exec/eval/import-ed; no XML is parsed, so there is no XXE surface.
• Id sanitization — generated note ids are validated with :func:memory.safety.validate_id so a crafted title can never produce a path separator / control-char id.

Semantics:

• Each .md → a :class:~zettelkasten.graph.Note. Frontmatter supplies type/tags/aliases/status; the H1 (or frontmatter title / filename) supplies the title; the remaining prose is the body.
• [[wikilinks]] → :class:~zettelkasten.graph.Link. A Dataview typed line relation:: <RELATION> [[target]] maps <RELATION> to a :data:~zettelkasten.graph.VALID_RELATIONS value; an UNKNOWN relation is NEVER dropped — it falls back to related with a warning. A bare wikilink defaults to related. Optional (confidence:: <float>), (direction:: <dir>) and (graph:: <box>) subfields on a typed line round-trip a link's confidence/direction/graph (emitted by :mod:zettelkasten.export). A (graph:: …) marker names a cross-graph target, so the importer keeps the raw target and does NOT resolve it against local notes (a same-id/same-title local note is a DIFFERENT entity). Prose is never silently dropped: a relation:: line that is NOT a valid typed wikilink (e.g. relation:: see chapter 3) is kept in the body verbatim, and a typed line carrying trailing prose (contradicts [[Beta]] but only in bear markets) still captures the contradicts edge (relation NOT downgraded to related) while KEEPING the line in the body, with a warning.
• Imports are IDEMPOTENT along two axes:
• Identity — when a file's frontmatter carries an id (a vault WE exported), a re-import into the SAME box dedups on note identity: if that id already names a note, the import is a no-op (no duplicate mint). The id is UNTRUSTED and validated (:func:memory.safety.validate_id) before use, and is only ever used to mint a NEW note — never to overwrite an existing one.
• Content — otherwise each note's content_hash (the file's sha256, via :func:zettelkasten.ingest.compute_content_hash) gates re-import: a file whose hash already exists in the box is skipped.
• Imported notes get epistemic_status = "inferred" (they carry no PDF/quote grounding).
• propose_only (wired to ZK_PROPOSE_ONLY) returns a DRY-RUN manifest and writes NOTHING to the canonical store.

import_vault

import_vault(source_dir: str | Path, graph: ZettelGraph, *, propose_only: bool = False, max_file_bytes: int = DEFAULT_MAX_FILE_BYTES, max_files: int = DEFAULT_MAX_FILES) -> dict

Import an Obsidian/markdown vault at source_dir into graph.

Parameters:

Name	Type	Description	Default
source_dir	str | Path	the vault root. Resolved to a realpath; must be an existing directory. Files that escape this root (traversal/symlink) are skipped.	required
graph	ZettelGraph	a loaded :class:~zettelkasten.graph.ZettelGraph to import into (used for dedup and, unless propose_only, for saving). Its existing notes' content_hash values gate idempotent re-import.	required
propose_only	bool	when true, return a DRY-RUN manifest and write NOTHING to the canonical store (wired to ZK_PROPOSE_ONLY).	False
max_file_bytes	int	per-file size cap; larger files are skipped with a warning.	DEFAULT_MAX_FILE_BYTES
max_files	int	total .md count cap (dir-bomb guard); exceeding it raises ValueError.	DEFAULT_MAX_FILES

Returns:

Type	Description
dict	A manifest dict: {created, skipped, links, warnings, dry_run} where
dict	created lists {id, title} (or would-create entries in dry-run).

Source code in zettelkasten/import_vault.py

def import_vault(
    source_dir: str | Path,
    graph: ZettelGraph,
    *,
    propose_only: bool = False,
    max_file_bytes: int = DEFAULT_MAX_FILE_BYTES,
    max_files: int = DEFAULT_MAX_FILES,
) -> dict:
    """Import an Obsidian/markdown vault at ``source_dir`` into ``graph``.

    Args:
        source_dir: the vault root. Resolved to a realpath; must be an existing
            directory. Files that escape this root (traversal/symlink) are
            skipped.
        graph: a loaded :class:`~zettelkasten.graph.ZettelGraph` to import into
            (used for dedup and, unless ``propose_only``, for saving). Its
            existing notes' ``content_hash`` values gate idempotent re-import.
        propose_only: when true, return a DRY-RUN manifest and write NOTHING to
            the canonical store (wired to ``ZK_PROPOSE_ONLY``).
        max_file_bytes: per-file size cap; larger files are skipped with a
            warning.
        max_files: total ``.md`` count cap (dir-bomb guard); exceeding it raises
            ``ValueError``.

    Returns:
        A manifest dict: ``{created, skipped, links, warnings, dry_run}`` where
        ``created`` lists ``{id, title}`` (or would-create entries in dry-run).
    """
    root = Path(source_dir).expanduser().resolve()
    if not root.is_dir():
        raise ValueError(f"source_dir is not a directory: {source_dir!r}")

    files, warnings = _collect_markdown_files(root, max_files)

    # Existing content hashes gate idempotent re-import.
    existing_hashes: set[str] = set()
    for note in graph.notes.values():
        if isinstance(note.source, dict):
            h = note.source.get("content_hash")
            if h:
                existing_hashes.add(str(h))

    # ── Pass 1: parse + stage notes (dedup on content_hash) ──────────────────
    staged: list[dict] = []
    skipped: list[dict] = []
    assigned_ids: set[str] = set(graph.notes.keys())
    # Frontmatter ids claimed by files EARLIER in this same batch. Used to tell a
    # benign idempotent re-import (id matches a note already in the box → quiet
    # no-op) apart from two DIFFERENT files in one vault claiming the same id
    # (ambiguous content loss → warn loudly).
    staged_fm_ids: set[str] = set()
    title_index: dict[str, str] = {}
    alias_index: dict[str, str] = {}
    for note in graph.notes.values():
        title_index[note.title.lower()] = note.id
        for alias in note.aliases:
            alias_index[alias.lower()] = note.id

    for path in files:
        try:
            size = path.stat().st_size
        except OSError as exc:
            warnings.append(f"skipped '{path.name}': {exc}")
            continue
        if size > max_file_bytes:
            warnings.append(
                f"skipped '{path.name}': {size} bytes exceeds the "
                f"{max_file_bytes}-byte per-file cap"
            )
            continue
        content_hash = compute_content_hash(path)
        if content_hash in existing_hashes:
            skipped.append({"file": path.name, "reason": "duplicate", "content_hash": content_hash})
            continue

        try:
            text = path.read_text(encoding="utf-8", errors="replace")
        except OSError as exc:
            warnings.append(f"skipped '{path.name}': {exc}")
            continue

        parsed = _parse_markdown_file(text, path.stem)
        for w in parsed.pop("warnings"):
            warnings.append(f"{path.name}: {w}")

        # Honor a frontmatter ``id`` (e.g. from a vault WE exported) so a
        # re-import dedups on note IDENTITY, not the file's sha256 (export
        # rewrites bytes, so the hash never matches). The id is UNTRUSTED input:
        # sanitize it before any use, and only ever use it to MINT a brand-new
        # note — never to overwrite an existing one (no malicious clobber).
        fm_id = _sanitize_frontmatter_id(parsed.pop("fm_id", ""), path.name, warnings)
        if fm_id and fm_id in staged_fm_ids:
            # A DIFFERENT file earlier in THIS vault already claimed this id (its
            # content_hash differed, so it wasn't a byte-dup). Honoring the id
            # would silently discard this file's distinct content — that is
            # ambiguous data loss, not the benign idempotent case, so warn.
            warnings.append(
                f"{path.name}: frontmatter id {fm_id!r} already claimed by another "
                f"file in this import; skipped (its content was NOT imported)"
            )
            skipped.append({"file": path.name, "reason": "duplicate-id", "id": fm_id})
            continue
        if fm_id and _id_exists(fm_id, assigned_ids, graph):
            # A note with this identity already exists in the box: benign
            # idempotent no-op (e.g. re-importing an exported vault), quiet.
            skipped.append({"file": path.name, "reason": "exists", "id": fm_id})
            continue

        note_id = fm_id or _unique_id(parsed["title"], assigned_ids, graph)
        assigned_ids.add(note_id)
        if fm_id:
            staged_fm_ids.add(fm_id)
        # Guard against a duplicate WITHIN this import batch too.
        existing_hashes.add(content_hash)
        parsed["id"] = note_id
        parsed["content_hash"] = content_hash
        parsed["file"] = path.name
        staged.append(parsed)
        title_index.setdefault(parsed["title"].lower(), note_id)
        for alias in parsed["aliases"]:
            alias_index.setdefault(alias.lower(), note_id)

    # ── Pass 2: resolve links + materialize Note objects ─────────────────────
    def _resolve(ref: str) -> str:
        r = ref.strip()
        if r in assigned_ids:
            return r
        by_title = title_index.get(r.lower())
        if by_title:
            return by_title
        by_alias = alias_index.get(r.lower())
        if by_alias:
            return by_alias
        # Unresolved: keep the raw reference as a (dangling) target rather than
        # drop the edge — the store tolerates unresolved link targets.
        return r

    notes: list[Note] = []
    link_count = 0
    for item in staged:
        links = []
        for raw in item["raw_links"]:
            graph_ref = raw.get("graph") or ""
            # A cross-graph link (carrying an explicit ``graph`` marker) names an
            # entity in ANOTHER box; it must NOT be resolved against local notes —
            # a same-id/same-title local note is a DIFFERENT entity, so resolving
            # would silently re-target the edge. Keep its raw target + marker.
            target = raw["target"] if graph_ref else _resolve(raw["target"])
            links.append(
                Link(
                    target=target,
                    relation=raw["relation"],
                    confidence=raw.get("confidence"),
                    direction=raw.get("direction") or "outgoing",
                    graph=graph_ref,
                )
            )
        link_count += len(links)
        notes.append(
            Note(
                id=item["id"],
                title=item["title"],
                type=item["type"],
                source={"content_hash": item["content_hash"], "import_file": item["file"]},
                body=item["body"],
                tags=item["tags"],
                links=links,
                aliases=item["aliases"],
                status=item["status"],
                epistemic_status="inferred",
            )
        )

    created = [{"id": n.id, "title": n.title, "links": len(n.links)} for n in notes]

    if propose_only:
        return {
            "dry_run": True,
            "source_dir": str(root),
            "created": created,
            "skipped": skipped,
            "links": link_count,
            "warnings": warnings,
        }

    for note in notes:
        graph.save_note(note)

    return {
        "dry_run": False,
        "source_dir": str(root),
        "created": created,
        "skipped": skipped,
        "links": link_count,
        "warnings": warnings,
    }