zettelkasten.ingest

zettelkasten.ingest

Source ingestion: the canonical front door for getting a paper into the zettelkasten.

A single resolver, :func:ingest, accepts a Zotero key, a local file path, or a title and returns the extracted full text, saved annotations, a stable content_hash (sha256 of the original file's bytes), and provenance (zotero_key / source_path). The hash gives every source a stable identity independent of Zotero, lets create_source dedup re-ingested files, and keys a disposable full-text cache so the reference is reproducible offline.

Full source text is never committed: it is cached under <ANGELO_DIR>/zettel/fulltext/<content_hash>.txt (gitignored, like the kglite embedding caches), while the committed _meta.yaml keeps the pointer (content_hash + source_path/zotero_key) needed to re-derive it.

sources_dir

sources_dir() -> Path

Convenience inbox directory for local source files.

Override with ZK_SOURCES_DIR; defaults to sources/ at the workspace root. Paths passed to :func:ingest may be absolute, relative to the cwd, or a bare filename resolved against this directory.

Source code in zettelkasten/ingest.py

def sources_dir() -> Path:
    """Convenience inbox directory for local source files.

    Override with ``ZK_SOURCES_DIR``; defaults to ``sources/`` at the workspace
    root. Paths passed to :func:`ingest` may be absolute, relative to the cwd, or
    a bare filename resolved against this directory.
    """
    from zettelkasten.graph import _anchor

    env = os.environ.get("ZK_SOURCES_DIR", "").strip()
    return _anchor(Path(env)) if env else _anchor(Path("sources"))

compute_content_hash

compute_content_hash(file_path: Path) -> str

sha256 of a file's raw bytes — the source's stable identity.

Source code in zettelkasten/ingest.py

def compute_content_hash(file_path: Path) -> str:
    """sha256 of a file's raw bytes — the source's stable identity."""
    h = hashlib.sha256()
    with open(file_path, "rb") as fh:
        for chunk in iter(lambda: fh.read(1 << 20), b""):
            h.update(chunk)
    return h.hexdigest()

fulltext_cache_path

fulltext_cache_path(content_hash: str) -> Path

Path to the cached full-text file for a content hash.

Source code in zettelkasten/ingest.py

def fulltext_cache_path(content_hash: str) -> Path:
    """Path to the cached full-text file for a content hash."""
    return _fulltext_dir() / f"{content_hash}.txt"

cache_fulltext

cache_fulltext(content_hash: str, text: str) -> Path | None

Write extracted text to the content-addressed cache.

Best-effort: returns the path on success, None if hashing/text is missing or the write fails (the cache is disposable, so a failure is non-fatal).

Source code in zettelkasten/ingest.py

def cache_fulltext(content_hash: str, text: str) -> Path | None:
    """Write extracted text to the content-addressed cache.

    Best-effort: returns the path on success, ``None`` if hashing/text is missing
    or the write fails (the cache is disposable, so a failure is non-fatal).
    """
    if not content_hash or not text:
        return None
    try:
        cache_dir = _fulltext_dir()
        cache_dir.mkdir(parents=True, exist_ok=True)
        path = cache_dir / f"{content_hash}.txt"
        path.write_text(text, encoding="utf-8")
        return path
    except OSError:
        return None

read_cached_fulltext

read_cached_fulltext(content_hash: str) -> str | None

Return cached full text for a content hash, or None if not cached.

Source code in zettelkasten/ingest.py

def read_cached_fulltext(content_hash: str) -> str | None:
    """Return cached full text for a content hash, or ``None`` if not cached."""
    if not content_hash:
        return None
    path = fulltext_cache_path(content_hash)
    if not path.is_file():
        return None
    try:
        return path.read_text(encoding="utf-8")
    except OSError:
        return None

page_index_path

page_index_path(content_hash: str) -> Path

Path to the disposable page/outline sidecar for a content hash.

Source code in zettelkasten/ingest.py

def page_index_path(content_hash: str) -> Path:
    """Path to the disposable page/outline sidecar for a content hash."""
    return _fulltext_dir() / f"{content_hash}.pages.json"

cache_page_index

cache_page_index(content_hash: str, *, num_pages: int, total_chars: int, page_offsets: list[int], outline: list[dict]) -> Path | None

Write the page-offset + chapter-outline sidecar beside the full-text cache.

Disposable and content-addressed like the full-text cache itself: it lets :func:~zettelkasten.server.get_fulltext slice by page range and lets the coordinator snap window boundaries to chapter starts. Best-effort -- returns None if the hash is missing or the write fails.

Source code in zettelkasten/ingest.py

def cache_page_index(
    content_hash: str,
    *,
    num_pages: int,
    total_chars: int,
    page_offsets: list[int],
    outline: list[dict],
) -> Path | None:
    """Write the page-offset + chapter-outline sidecar beside the full-text cache.

    Disposable and content-addressed like the full-text cache itself: it lets
    :func:`~zettelkasten.server.get_fulltext` slice by page range and lets the
    coordinator snap window boundaries to chapter starts. Best-effort -- returns
    ``None`` if the hash is missing or the write fails.
    """
    if not content_hash:
        return None
    try:
        import json

        cache_dir = _fulltext_dir()
        cache_dir.mkdir(parents=True, exist_ok=True)
        path = cache_dir / f"{content_hash}.pages.json"
        path.write_text(
            json.dumps({
                "num_pages": num_pages,
                "total_chars": total_chars,
                "page_offsets": page_offsets,
                "outline": outline,
            }),
            encoding="utf-8",
        )
        return path
    except OSError:
        return None

read_page_index

read_page_index(content_hash: str) -> dict | None

Return the cached page/outline sidecar for a content hash, or None.

Shape: {num_pages, total_chars, page_offsets, outline} where each outline entry is {title, page, char_offset}.

Source code in zettelkasten/ingest.py

def read_page_index(content_hash: str) -> dict | None:
    """Return the cached page/outline sidecar for a content hash, or ``None``.

    Shape: ``{num_pages, total_chars, page_offsets, outline}`` where each
    ``outline`` entry is ``{title, page, char_offset}``.
    """
    if not content_hash:
        return None
    path = page_index_path(content_hash)
    if not path.is_file():
        return None
    try:
        import json

        return json.loads(path.read_text(encoding="utf-8"))
    except (OSError, ValueError):
        return None

extract_text

extract_text(file_path: Path, max_chars: int = DEFAULT_MAX_CHARS) -> tuple[str, int, str]

Extract text from a local file. Returns (text, num_pages, warning).

PDFs go through the shared pypdf extractor; recognised text suffixes are read directly; anything else is read as UTF-8 best-effort. num_pages is 0 for non-PDF inputs.

Source code in zettelkasten/ingest.py

def extract_text(file_path: Path, max_chars: int = DEFAULT_MAX_CHARS) -> tuple[str, int, str]:
    """Extract text from a local file. Returns (text, num_pages, warning).

    PDFs go through the shared pypdf extractor; recognised text suffixes are read
    directly; anything else is read as UTF-8 best-effort. ``num_pages`` is 0 for
    non-PDF inputs.
    """
    suffix = file_path.suffix.lower()
    if suffix == ".pdf":
        from zettelkasten.zotero import _extract_pdf_text

        return _extract_pdf_text(file_path, max_chars)
    if suffix in _TEXT_SUFFIXES or suffix == "":
        try:
            text = file_path.read_text(encoding="utf-8", errors="replace")
        except OSError as exc:
            return "", 0, f"Could not read {file_path}: {exc}"
        warning = ""
        if len(text) > max_chars:
            text = text[:max_chars]
            warning = f"Text truncated to {max_chars} characters."
        return text, 0, warning
    return (
        "",
        0,
        f"Unsupported file type {suffix!r}; read the file at 'source_path' directly.",
    )

extract_text_full

extract_text_full(file_path: Path) -> tuple[str, int, str, list[int], list[dict]]

Extract the WHOLE file plus a page/char index and chapter outline.

The full-extract counterpart to :func:extract_text: it never truncates, so the disposable cache can hold a complete book. PDFs go through the shared page-aware extractor (page_offsets + outline); text files are a single page (offset [0], no outline); unsupported types return empty.

Returns (text, num_pages, warning, page_offsets, outline).

Source code in zettelkasten/ingest.py

def extract_text_full(
    file_path: Path,
) -> tuple[str, int, str, list[int], list[dict]]:
    """Extract the WHOLE file plus a page/char index and chapter outline.

    The full-extract counterpart to :func:`extract_text`: it never truncates, so
    the disposable cache can hold a complete book. PDFs go through the shared
    page-aware extractor (``page_offsets`` + ``outline``); text files are a single
    page (offset ``[0]``, no outline); unsupported types return empty.

    Returns ``(text, num_pages, warning, page_offsets, outline)``.
    """
    suffix = file_path.suffix.lower()
    if suffix == ".pdf":
        from zettelkasten.zotero import _extract_pdf_text_full

        return _extract_pdf_text_full(file_path)
    if suffix in _TEXT_SUFFIXES or suffix == "":
        try:
            text = file_path.read_text(encoding="utf-8", errors="replace")
        except OSError as exc:
            return "", 0, f"Could not read {file_path}: {exc}", [], []
        return text, 0, "", [0], []
    return (
        "",
        0,
        f"Unsupported file type {suffix!r}; read the file at 'source_path' directly.",
        [],
        [],
    )

ingest

ingest(key: str = '', path: str = '', title: str = '', max_chars: int = DEFAULT_MAX_CHARS, cache_full: bool = False) -> dict[str, Any]

Resolve a source to full text + a stable content hash + provenance.

Exactly one of key (Zotero item key), path (local file), or title (looked up in Zotero) should be provided. title resolves to a Zotero key when there is a single unambiguous match; otherwise the candidate list is returned for the caller to disambiguate.

Returns a dict with text, annotations, content_hash, source_path, optional zotero_key, num_pages, and any text_warning. The extracted text is also written to the disposable cache keyed by content_hash. On failure returns {"error": ...}.

When cache_full is set the WHOLE document is extracted and cached (plus a <hash>.pages.json page/outline sidecar), while the returned text stays capped at max_chars for the caller's token budget. This is the precondition for book-scale extraction: a later chapter is only addressable if the cache holds it. The result then also carries total_chars and outline.

Source code in zettelkasten/ingest.py

def ingest(
    key: str = "",
    path: str = "",
    title: str = "",
    max_chars: int = DEFAULT_MAX_CHARS,
    cache_full: bool = False,
) -> dict[str, Any]:
    """Resolve a source to full text + a stable content hash + provenance.

    Exactly one of ``key`` (Zotero item key), ``path`` (local file), or ``title``
    (looked up in Zotero) should be provided. ``title`` resolves to a Zotero key
    when there is a single unambiguous match; otherwise the candidate list is
    returned for the caller to disambiguate.

    Returns a dict with ``text``, ``annotations``, ``content_hash``,
    ``source_path``, optional ``zotero_key``, ``num_pages``, and any
    ``text_warning``. The extracted text is also written to the disposable cache
    keyed by ``content_hash``. On failure returns ``{"error": ...}``.

    When ``cache_full`` is set the WHOLE document is extracted and cached (plus a
    ``<hash>.pages.json`` page/outline sidecar), while the returned ``text`` stays
    capped at ``max_chars`` for the caller's token budget. This is the precondition
    for book-scale extraction: a later chapter is only addressable if the cache
    holds it. The result then also carries ``total_chars`` and ``outline``.
    """
    key = (key or "").strip()
    path = (path or "").strip()
    title = (title or "").strip()

    provided = [bool(key), bool(path), bool(title)]
    if sum(provided) == 0:
        return {"error": "Provide one of: key (Zotero), path (local file), or title.", "type": "BadRequest"}
    if sum(provided) > 1:
        return {"error": "Provide only one of key, path, or title.", "type": "BadRequest"}

    if title:
        return _ingest_by_title(title, max_chars, cache_full=cache_full)
    if key:
        return _ingest_by_zotero_key(key, max_chars, cache_full=cache_full)
    return _ingest_by_path(path, max_chars, cache_full=cache_full)