Hunking Obsidian — export Obsidian File Recovery snapshots as unified diffs for LLM review

hunking_obsidian.md

Hunking Obsidian

Obsidian's File Recovery plugin snapshots every note you edit. This script reads those snapshots from disk and outputs unified diffs or hunks (inspired by hunk that makes it easier for large language model agents to work with code; this is just for your thoughts, notes, and journals).

The tool outputs one section per note, newest first, with a per-day edit table at the top.

The output is context for an LLM. You give it the diffs; it tells you what you've been working on, where load is concentrated, what's unresolved, and where you seem to be headed. This is a proxy signal — edit activity, not internal state — but it's enough for a useful weekly read.

Bulk syncs (iCloud or Obsidian Sync pulling another device's changes) are detected and excluded from the activity table so they don't inflate your edit counts or trigger false late-night flags.

Requires: macOS, Obsidian with File Recovery enabled (Settings → File Recovery), Python 3.10+, uv.

Run

# Export the past week as net diffs (one first→last diff per note)
./hunking_obsidian.py --since 7 --net --out week.md

# Export the past day, all consecutive diffs
./hunking_obsidian.py --since 1 --out today.md

# JSON output for programmatic use
./hunking_obsidian.py --since 7 --format json --out week.json

Flag	Default	Meaning
--since N	0 (all)	Last N days of snapshots. File Recovery keeps ~7 days by default.
--net	off	One first→last diff per note instead of every consecutive pair.
--full-below N	4000	Notes under N chars are shown in full instead of diffed.
--context N	3	Context lines per hunk. Hunks are annotated with the nearest Markdown heading.
--sync-threshold N	4	Notes sharing one timestamp ≥ N → treated as a bulk sync, excluded from metrics.
--no-metadata	—	Omit the per-day activity table.
--vault	—	Limit to one vault by substring of its path.

Example LLM prompt

Attach week.md and send:

Take the attached diff hunks from my Obsidian vault. Give me a summary of what has been on my mind this past week, any reliable psychometrics you can glean (edit volume, active hours, context switching across notes), and a list of open loops or things that look unresolved or stalled.

How it works

Obsidian stores snapshots in a Chromium IndexedDB (a LevelDB on disk). The script copies the database to a temp directory, reads it without any Obsidian involvement, and decodes all three value encodings Chromium uses (plain V8, inline Snappy-compressed, and external blob). It runs safely while Obsidian is open.

hunking_obsidian.py

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.10"
# dependencies = [
#   "ccl-chromium-reader @ git+https://github.com/cclgroupltd/ccl_chromium_reader.git@b51a01c913799f5af2735f964d4627275369e90e",
#   "ccl-simplesnappy    @ git+https://github.com/cclgroupltd/ccl_simplesnappy.git@3d085230baa8c46cf2090ebba29bf6e8eab31087",
# ]
# ///
"""
Hunking Obsidian — export Obsidian's local version history as unified diffs.

Obsidian's File Recovery plugin snapshots every note you edit. This script reads
those snapshots directly from disk and writes them as a Markdown file of diffs —
one section per note, newest first, with a per-day activity table at the top.
Feed the output to an LLM for a weekly review of what you worked on.

No plugins, Obsidian Sync, or GUI needed. Works on macOS while Obsidian is open.

Usage:
    ./hunking_obsidian.py --since 7 --net --out week.md
    ./hunking_obsidian.py --since 7 --format json --out week.json
    ./hunking_obsidian.py --help
"""

from __future__ import annotations
import argparse
import difflib
import io
import json
import os
import re
import shutil
import sys
import tempfile
import datetime as dt
from pathlib import Path
from collections import defaultdict

LEVELDB = "app_obsidian.md_0.indexeddb.leveldb"
BLOB = "app_obsidian.md_0.indexeddb.blob"
HEADING = re.compile(r"^\s{0,3}#{1,6}\s")
HUNK = re.compile(r"^@@ -\d+(?:,\d+)? \+(\d+)(?:,\d+)? @@")


def default_app_dir() -> Path:
    if sys.platform == "darwin":
        return Path.home() / "Library/Application Support/obsidian"
    if os.name == "nt":
        return Path(os.environ.get("APPDATA", Path.home())) / "obsidian"
    return Path.home() / ".config/obsidian"


def local(ms: float) -> dt.datetime:
    return dt.datetime.fromtimestamp(ms / 1000, dt.timezone.utc).astimezone()


def iso(ms: float) -> str:
    return local(ms).strftime("%Y-%m-%d %H:%M:%S")


def vault_id_for(app_dir: Path, needle: str) -> str | None:
    cfg = app_dir / "obsidian.json"
    if not cfg.exists():
        return None
    for vid, meta in json.loads(cfg.read_text()).get("vaults", {}).items():
        if needle.lower() in meta.get("path", "").lower():
            return vid
    return None


def load_snapshots(leveldb: Path, blob_dir: Path | None, vault_id: str | None):
    """Return ({path: [(ts_ms, text), ...] sorted asc}, unrecoverable_count)."""
    import ccl_simplesnappy
    from ccl_chromium_reader.ccl_chromium_indexeddb import (
        WrappedIndexDB,
        _le_varint_from_bytes,
    )
    from ccl_chromium_reader.serialization_formats import (
        ccl_v8_value_deserializer as v8d,
    )
    from ccl_chromium_reader.serialization_formats import (
        ccl_blink_value_deserializer as blinkd,
    )

    blink = blinkd.BlinkV8Deserializer()

    def parse_v8(body: bytes) -> dict | None:
        # V8 payload header is 0xff <version> 0x6f; scan past the variable-length Blink trailer.
        for i in range(min(len(body) - 2, 64)):
            if body[i] == 0xFF and body[i + 2] == 0x6F:
                try:
                    obj = v8d.Deserializer(
                        io.BytesIO(body[i:]), host_object_delegate=blink.read
                    ).read()
                    if (
                        isinstance(obj, dict)
                        and "path" in obj
                        and isinstance(obj.get("data"), str)
                    ):
                        return obj
                except Exception:
                    pass
        return None

    def decode(buf: bytes) -> dict | None:
        # Blink envelope: 0xff 0x11 <compression-byte> ...  0x02=Snappy, 0x01=external-blob-ref
        body = (
            ccl_simplesnappy.decompress(io.BytesIO(buf[3:]))
            if buf[2] == 0x02
            else buf[3:]
        )
        return parse_v8(body)

    db = WrappedIndexDB(
        str(leveldb), str(blob_dir) if blob_dir and blob_dir.exists() else None
    )
    snaps: dict[str, list] = defaultdict(list)
    unrecoverable = 0

    for dbid in db.database_ids:
        if not dbid.name.endswith("-backup"):
            continue
        if vault_id and not dbid.name.startswith(vault_id):
            continue

        wstore = db[dbid.dbid_no].get_object_store_by_name("backups")
        raw_db, store_id = wstore._raw_db, wstore.object_store_id
        deferred: list[tuple] = []

        for rec in raw_db.iterate_records(
            dbid.dbid_no,
            store_id,
            bad_deserializer_data_handler=lambda k, r: deferred.append((k, r)),
        ):
            v = rec.value
            if isinstance(v, dict) and "path" in v and isinstance(v.get("data"), str):
                snaps[v["path"]].append((float(v["ts"]), v["data"]))

        for key, raw in deferred:
            # Some records have a leading varint length-prefix before the Blink envelope.
            buf = (
                raw
                if raw[:2] == b"\xff\x11"
                else raw[len(_le_varint_from_bytes(raw)[1]) :]
            )
            obj = None

            if buf[:3] == b"\xff\x11\x02":
                obj = decode(buf)
            elif buf[:3] == b"\xff\x11\x01":
                # blob-ref layout: 3-byte header | varint(wrapped-size) | varint(blob-index)
                p = 3
                _, size_raw = _le_varint_from_bytes(buf[p:])
                p += len(size_raw)
                blob_idx, _ = _le_varint_from_bytes(buf[p:])
                try:
                    obj = decode(
                        raw_db.get_blob(
                            dbid.dbid_no, store_id, key.raw_key, blob_idx
                        ).read()
                    )
                except Exception:
                    pass  # blob has been GC'd — unrecoverable

            if obj:
                snaps[obj["path"]].append((float(obj["ts"]), obj["data"]))
            else:
                unrecoverable += 1

    for path in snaps:
        snaps[path].sort(key=lambda x: x[0])
    return snaps, unrecoverable


def sync_seconds(snaps: dict, threshold: int) -> set[int]:
    """Seconds where >= threshold distinct notes were captured — bulk sync, not live authoring."""
    by_sec: dict[int, set] = defaultdict(set)
    for path, versions in snaps.items():
        for ts, _ in versions:
            by_sec[int(ts // 1000)].add(path)
    return {sec for sec, paths in by_sec.items() if len(paths) >= threshold}


def heading_aware_diff(
    text_a: str, text_b: str, context: int, from_label: str, to_label: str
) -> str:
    lines_b = text_b.splitlines()
    out = []
    for line in difflib.unified_diff(
        text_a.splitlines(),
        lines_b,
        fromfile=from_label,
        tofile=to_label,
        lineterm="",
        n=context,
    ):
        if line.startswith("@@"):
            m = HUNK.match(line)
            if m:
                start = max(int(m.group(1)) - 1, 0)
                heading = next(
                    (
                        lines_b[i].strip()
                        for i in range(min(start, len(lines_b) - 1), -1, -1)
                        if HEADING.match(lines_b[i])
                    ),
                    "",
                )
                if heading:
                    line = f"{line} {heading}"
        out.append(line)
    return "\n".join(out)


def pair_volume(text_a: str, text_b: str) -> tuple[int, int]:
    add = rem = 0
    for line in difflib.unified_diff(
        text_a.splitlines(), text_b.splitlines(), lineterm="", n=0
    ):
        if line[:2] in ("--", "++") or line.startswith("@@"):
            continue
        if line.startswith("+"):
            add += len(line) - 1
        elif line.startswith("-"):
            rem += len(line) - 1
    return add, rem


def activity_table(snaps: dict, cutoff_ms: float, sync_secs: set[int]) -> str:
    days: dict[str, dict] = defaultdict(
        lambda: {"notes": set(), "edits": 0, "add": 0, "rem": 0, "hours": []}
    )
    for path, versions in snaps.items():
        for i, (ts, data) in enumerate(versions):
            if ts < cutoff_ms or int(ts // 1000) in sync_secs:
                continue
            d = days[local(ts).strftime("%Y-%m-%d")]
            d["notes"].add(path)
            d["edits"] += 1
            d["hours"].append(local(ts).hour)
            if i > 0:  # first-ever snapshot is a baseline capture, not volume authored that day
                add, rem = pair_volume(versions[i - 1][1], data)
                d["add"] += add
                d["rem"] += rem
    if not days:
        return "## Activity by day\n\n_no live (non-sync) edits in window._\n"
    rows = [
        "| Date | Notes | Edits | +chars | −chars | Active hrs | Late-night |",
        "|------|------:|------:|-------:|-------:|------------|:----------:|",
    ]
    for date in sorted(days):
        d = days[date]
        hrs = sorted(d["hours"])
        span = f"{hrs[0]:02d}:00–{hrs[-1]:02d}:59"
        late = "⚠️" if any(0 <= h < 6 for h in hrs) else ""
        rows.append(
            f"| {date} | {len(d['notes'])} | {d['edits']} | {d['add']} | {d['rem']} | {span} | {late} |"
        )
    return (
        "## Activity by day\n\n_(live edits only; bulk syncs excluded — see below)_\n\n"
        + "\n".join(rows)
        + "\n"
    )


def sync_summary(snaps: dict, cutoff_ms: float, sync_secs: set[int]) -> str:
    bursts: dict[int, set] = defaultdict(set)
    for path, versions in snaps.items():
        for ts, _ in versions:
            if ts >= cutoff_ms and int(ts // 1000) in sync_secs:
                bursts[int(ts // 1000)].add(path)
    if not bursts:
        return ""
    lines = [
        f"- {iso(sec * 1000)[:16]} — {len(paths)} notes captured together"
        for sec, paths in sorted(bursts.items())
    ]
    return (
        "## Sync events (bulk captures — excluded from metrics)\n\n"
        "_These notes changed elsewhere and synced in at one timestamp; their diffs "
        "still appear below but their timing is not real-time activity._\n\n"
        + "\n".join(lines)
        + "\n"
    )


def render(
    snaps: dict,
    cutoff_ms: float,
    context: int,
    net: bool,
    full_below: int,
    with_meta: bool,
    sync_secs: set[int],
) -> str:
    parts: list[str] = []
    total = sum(len(v) for v in snaps.values())
    parts.append(
        f"# Hunking Obsidian\n\n_generated {iso(dt.datetime.now().timestamp() * 1000)} · "
        f"{total} snapshots across {len(snaps)} notes_\n"
    )
    if with_meta:
        parts.append(activity_table(snaps, cutoff_ms, sync_secs))
        parts.append(sync_summary(snaps, cutoff_ms, sync_secs))

    ranked = sorted(
        ((p, v) for p, v in snaps.items() if v[-1][0] >= cutoff_ms),
        key=lambda kv: kv[1][-1][0],
        reverse=True,
    )

    for path, versions in ranked:
        in_win = [v for v in versions if v[0] >= cutoff_ms]
        pre = next((v for v in reversed(versions) if v[0] < cutoff_ms), None)
        head_ts, head_data = in_win[-1]
        parts.append(
            f"\n## {path}\n\n_{len(in_win)} edit(s) in window · newest {iso(head_ts)} · "
            f"{len(head_data)} chars_\n"
        )

        if len(head_data) < full_below:
            parts.append(
                "_current content:_\n\n```markdown\n" + head_data.rstrip() + "\n```\n"
            )
            continue

        chain = ([pre] + in_win) if pre else in_win
        if len(chain) < 2:
            parts.append("_single snapshot in window; no diff._\n")
            continue

        spans = [(chain[0], chain[-1])] if net else list(zip(chain, chain[1:]))
        for (ta, text_a), (tb, text_b) in spans:
            diff = heading_aware_diff(text_a, text_b, context, iso(ta), iso(tb))
            parts.append(
                f"### {iso(ta)} → {iso(tb)}\n\n```diff\n{diff or '(no textual change)'}\n```\n"
            )

    return "\n".join(parts)


def main() -> None:
    ap = argparse.ArgumentParser(description="Hunking Obsidian — LLM-ready version history.")
    ap.add_argument(
        "--app-dir", type=Path, help="Obsidian config dir (default: OS-specific)."
    )
    ap.add_argument("--leveldb", type=Path, help="Override path to the .leveldb dir.")
    ap.add_argument("--blob", type=Path, help="Override path to the .blob dir.")
    ap.add_argument("--vault", help="Limit to one vault by substring of its path.")
    ap.add_argument("--out", type=Path, default=Path("hunking-obsidian.md"))
    ap.add_argument("--format", choices=["md", "json"], default="md")
    ap.add_argument(
        "--since", type=int, default=0, help="Only the last N days (0 = all)."
    )
    ap.add_argument(
        "--context", type=int, default=3, help="Context lines per diff hunk."
    )
    ap.add_argument("--net", action="store_true", help="One first→last diff per note.")
    ap.add_argument(
        "--full-below",
        type=int,
        default=4000,
        help="Emit full content for notes whose newest version is under N chars.",
    )
    ap.add_argument(
        "--sync-threshold",
        type=int,
        default=4,
        help="Distinct notes in one second treated as a bulk sync (default 4).",
    )
    ap.add_argument(
        "--no-metadata",
        dest="metadata",
        action="store_false",
        help="Omit the per-day activity table.",
    )
    args = ap.parse_args()

    app_dir = args.app_dir or default_app_dir()
    leveldb = args.leveldb or app_dir / "IndexedDB" / LEVELDB
    blob_dir = args.blob or app_dir / "IndexedDB" / BLOB

    if not leveldb.exists():
        sys.exit(f"LevelDB not found: {leveldb}\nPass --app-dir or --leveldb.")

    vault_id = vault_id_for(app_dir, args.vault) if args.vault else None
    if args.vault and not vault_id:
        sys.exit(f"No vault matching {args.vault!r} in {app_dir / 'obsidian.json'}")

    with tempfile.TemporaryDirectory() as tmp:
        t = Path(tmp)
        shutil.copytree(leveldb, t / "l")
        if blob_dir.exists():
            shutil.copytree(blob_dir, t / "b")
        snaps, unrecoverable = load_snapshots(
            t / "l", t / "b" if blob_dir.exists() else None, vault_id
        )

    if not snaps:
        sys.exit("No snapshots found. Is the File recovery core plugin enabled?")

    cutoff_ms = (
        (dt.datetime.now() - dt.timedelta(days=args.since)).timestamp() * 1000
        if args.since
        else 0.0
    )
    gc_note = f" · {unrecoverable} unrecoverable (blob GC'd)" if unrecoverable else ""
    sync_secs = sync_seconds(snaps, args.sync_threshold)

    if args.format == "json":
        payload = {
            "generated": iso(dt.datetime.now().timestamp() * 1000),
            "since_days": args.since,
            "unrecoverable": unrecoverable,
            "snapshots": {
                p: [
                    {
                        "ts_ms": int(ts),
                        "ts": iso(ts),
                        "data": d,
                        "sync": int(ts // 1000) in sync_secs,
                    }
                    for ts, d in v
                    if ts >= cutoff_ms
                ]
                for p, v in snaps.items()
                if v[-1][0] >= cutoff_ms
            },
        }
        args.out.write_text(json.dumps(payload, indent=2, ensure_ascii=False))
    else:
        args.out.write_text(
            render(
                snaps,
                cutoff_ms,
                args.context,
                args.net,
                args.full_below,
                args.metadata,
                sync_secs,
            )
        )

    print(
        f"Wrote {args.out} — {len(snaps)} notes, since={args.since}d, net={args.net}, "
        f"sync_bursts={len(sync_secs)}{gc_note}."
    )


if __name__ == "__main__":
    main()