Importing existing knowledge

"""Import an Obsidian vault into a Living Brain.
Usage: python import_obsidian.py <vault folder> <brain id> [--dry-run]
Needs LIVING_BRAIN_API_KEY (and LIVING_BRAIN_SUBJECT_ID if your key isn't tied to one subject).
"""
import hashlib
import json
import os
import sys
import time
import urllib.error
import urllib.request
from pathlib import Path
API = "https://api.livingbrain.com/v1"
MAX_CHARS = 50_000  # the API's limit for one capture
BATCH_SIZE = 100  # the per-request limit on most plans
def split_note(text):
    """Split text longer than MAX_CHARS at paragraph breaks."""
    if len(text) <= MAX_CHARS:
        return [text]
    parts, current = [], ""
    for paragraph in text.split("\n\n"):
        if current and len(current) + len(paragraph) + 2 > MAX_CHARS:
            parts.append(current)
            current = ""
        current = f"{current}\n\n{paragraph}" if current else paragraph
        while len(current) > MAX_CHARS:  # a single paragraph over the limit
            parts.append(current[:MAX_CHARS])
            current = current[MAX_CHARS:]
    return parts + [current] if current else parts
def vault_items(vault):
    notes = (p for p in vault.rglob("*") if p.suffix.lower() == ".md" and p.is_file() and not p.is_symlink())
    for path in sorted(notes):
        relative = path.relative_to(vault)
        # Skip .obsidian (settings), .trash and any other hidden folder.
        if any(part.startswith(".") for part in relative.parts):
            continue
        try:
            text = path.read_bytes().decode("utf-8-sig")
        except UnicodeDecodeError:
            print(f"  skipped {relative}: not UTF-8 text")
            continue
        # Normalize Windows line endings so the paragraph split and hashes match on every OS.
        text = text.replace("\r\n", "\n").replace("\r", "\n").strip()
        if not text:
            continue
        name = relative.with_suffix("").as_posix()
        parts = split_note(text)
        for index, part in enumerate(parts, start=1):
            digest = hashlib.sha256(part.encode("utf-8")).hexdigest()[:16]
            suffix = f" (part {index} of {len(parts)})" if len(parts) > 1 else ""
            yield {
                "kind": "note",
                "content": part,
                "label": f"{name}{suffix}"[:256],
                # The brain skips a capture whose originRef it has already compiled.
                # The hash of the text means an edited note is captured again.
                "originRef": f"obsidian:{index}:{digest}:{vault.name}/{name}"[:512],
            }
def post_batch(brain_id, items):
    headers = {"x-api-key": os.environ["LIVING_BRAIN_API_KEY"], "content-type": "application/json"}
    if os.environ.get("LIVING_BRAIN_SUBJECT_ID"):
        headers["x-subject-id"] = os.environ["LIVING_BRAIN_SUBJECT_ID"]
    body = json.dumps({"items": items}).encode("utf-8")
    for attempt in range(5):
        request = urllib.request.Request(
            f"{API}/brains/{brain_id}/captures/batch", data=body, headers=headers, method="POST"
        )
        try:
            with urllib.request.urlopen(request, timeout=120) as response:
                return json.load(response)
        except urllib.error.HTTPError as error:
            # Retry rate limits and server errors; anything else needs fixing first.
            if (error.code != 429 and error.code < 500) or attempt == 4:
                raise SystemExit(f"HTTP {error.code}: {error.read().decode('utf-8', 'replace')}")
            retry_after = error.headers.get("retry-after", "")
            time.sleep(int(retry_after) if retry_after.isdigit() else 30)
        except (urllib.error.URLError, TimeoutError) as error:
            if attempt == 4:
                raise SystemExit(f"Network error: {error}")
            time.sleep(30)
def main():
    args = [arg for arg in sys.argv[1:] if arg != "--dry-run"]
    if len(args) != 2:
        raise SystemExit(__doc__)
    vault, brain_id = Path(os.path.abspath(os.path.expanduser(args[0]))), args[1]
    items = list(vault_items(vault))
    print(f"{len(items)} captures from {vault}")
    if "--dry-run" in sys.argv:
        for item in items:
            print(f"  {item['label']}  ({len(item['content']):,} characters)")
        return
    if not os.environ.get("LIVING_BRAIN_API_KEY"):
        raise SystemExit("Set LIVING_BRAIN_API_KEY first.")
    for start in range(0, len(items), BATCH_SIZE):
        batch = items[start : start + BATCH_SIZE]
        post_batch(brain_id, batch)
        print(f"  sent {start + len(batch)} of {len(items)}")
if __name__ == "__main__":
    main()