claude-session-to-obsidian

gistfile1.txt

#!/usr/bin/env python3
"""
Claude Code → Obsidian session export hook.

Dispatched by the Stop and SessionEnd hook events. Files are the source of
truth — the Obsidian app does not need to be running. One note per session_id,
re-rendered from the transcript JSONL on each call. Daily-note backlinks are
inserted once per (session, date) and never rewritten, so past days stay frozen.
"""

from __future__ import annotations

import fcntl
import json
import os
import re
import sys
import tempfile
from contextlib import contextmanager
from datetime import datetime
from pathlib import Path

VAULT = Path(os.environ.get(
    "OBSIDIAN_VAULT",
    str(Path.home() / "Dropbox" / "Obsidian" / "Obsidian Vault"),
))
SESSIONS_DIR = VAULT / "Claude Sessions"
# Daily notes live under Daily/ (matches Obsidian daily-notes plugin's folder setting).
DAILY_DIR = VAULT / "Daily"
STATE_DIR = Path.home() / ".claude" / "hooks" / "obsidian_state"
INDEX_PATH = STATE_DIR / "sessions.json"
LOCK_PATH = STATE_DIR / "lock"

DAILY_SECTIONS = ["## Claude Sessions", "## Work", "## Personal", "## Journal"]
SUMMARY_MAX = 120
SLUG_MAX = 60

NOISE_PATTERNS = [
    re.compile(r"<local-command-caveat>.*?</local-command-caveat>", re.DOTALL),
    re.compile(r"<local-command-stdout>.*?</local-command-stdout>", re.DOTALL),
    re.compile(r"<local-command-stderr>.*?</local-command-stderr>", re.DOTALL),
    re.compile(r"<system-reminder>.*?</system-reminder>", re.DOTALL),
    re.compile(r"<command-message>.*?</command-message>", re.DOTALL),
    re.compile(r"<user-prompt-submit-hook>.*?</user-prompt-submit-hook>", re.DOTALL),
]
COMMAND_NAME_RE = re.compile(r"<command-name>(.*?)</command-name>", re.DOTALL)
COMMAND_ARGS_RE = re.compile(r"<command-args>(.*?)</command-args>", re.DOTALL)


def parse_stdin() -> dict:
    try:
        return json.loads(sys.stdin.read())
    except (json.JSONDecodeError, ValueError):
        return {}


def find_transcript(hook_input: dict) -> str | None:
    tp = hook_input.get("transcript_path", "")
    if tp and os.path.exists(tp):
        return tp
    sid = hook_input.get("session_id", "")
    if not sid:
        return None
    projects = Path.home() / ".claude" / "projects"
    if projects.exists():
        for j in projects.rglob(f"{sid}.jsonl"):
            return str(j)
    return None


def parse_iso_ts(ts: str | int | float | None) -> datetime:
    if not ts:
        return datetime.now().astimezone()
    try:
        if isinstance(ts, (int, float)):
            return datetime.fromtimestamp(ts / 1000).astimezone()
        return datetime.fromisoformat(str(ts).replace("Z", "+00:00")).astimezone()
    except (ValueError, TypeError):
        return datetime.now().astimezone()


def clean_user_text(text: str) -> str:
    """Strip noise tags. If the message is a slash-command wrapper, render it
    as '> /name args' and drop everything else."""
    name_m = COMMAND_NAME_RE.search(text)
    if name_m:
        name = name_m.group(1).strip()
        args_m = COMMAND_ARGS_RE.search(text)
        args = args_m.group(1).strip() if args_m else ""
        args = re.sub(r"\s+", " ", args)
        return f"> {name} {args}".rstrip()
    for pat in NOISE_PATTERNS:
        text = pat.sub("", text)
    return text.strip()


def render_tool_call(item: dict) -> str:
    name = item.get("name", "tool")
    inp = item.get("input", {}) or {}
    primary = None
    for key in ("file_path", "path", "url", "command", "subject",
                "description", "query", "id", "pattern"):
        if key in inp and isinstance(inp[key], (str, int, float)):
            primary = (key, str(inp[key]))
            break
    if primary:
        k, v = primary
        v = v.replace("\n", " ").strip()
        if len(v) > 100:
            v = v[:97] + "..."
        return f"🔧 **{name}** — {k}: `{v}`"
    return f"🔧 **{name}**"


def render_assistant_content(content) -> str:
    if isinstance(content, str):
        return content.strip()
    if not isinstance(content, list):
        return ""
    out = []
    for item in content:
        if not isinstance(item, dict):
            continue
        t = item.get("type")
        if t == "text":
            txt = item.get("text", "").strip()
            if txt:
                out.append(txt)
        elif t == "tool_use":
            out.append(render_tool_call(item))
        # thinking, tool_result, etc. — skipped
    return "\n\n".join(out)


def render_user_content(content) -> str:
    if isinstance(content, str):
        return clean_user_text(content)
    if isinstance(content, list):
        parts = []
        for item in content:
            if isinstance(item, dict) and item.get("type") == "text":
                t = item.get("text", "").strip()
                if t:
                    parts.append(t)
        return clean_user_text("\n\n".join(parts))
    return ""


def parse_transcript(path: str) -> dict:
    messages: list[dict] = []
    session_id = None
    cwd = None
    first_ts = None
    last_ts = None

    with open(path) as f:
        for line in f:
            try:
                obj = json.loads(line)
            except json.JSONDecodeError:
                continue
            entry_type = obj.get("type")
            if entry_type not in ("user", "assistant"):
                continue
            if not session_id:
                session_id = obj.get("sessionId") or ""
            if not cwd:
                cwd = obj.get("cwd") or ""
            ts = obj.get("timestamp") or ""
            if ts and not first_ts:
                first_ts = ts
            if ts:
                last_ts = ts

            msg = obj.get("message", {})
            if isinstance(msg, str):
                try:
                    msg = json.loads(msg)
                except json.JSONDecodeError:
                    msg = {"role": entry_type, "content": msg}
            if not isinstance(msg, dict):
                continue

            role = msg.get("role", entry_type)
            content = msg.get("content", "")

            if role == "user":
                text = render_user_content(content)
            else:
                text = render_assistant_content(content)

            if text:
                messages.append({"role": role, "text": text, "ts": ts})

    return {
        "messages": messages,
        "session_id": session_id or Path(path).stem,
        "cwd": cwd or "",
        "first_ts": first_ts,
        "last_ts": last_ts,
    }


def generate_summary(messages: list[dict]) -> str:
    for m in messages:
        if m["role"] != "user":
            continue
        t = m["text"]
        if t.startswith("> /") or not t:
            continue
        t = re.sub(r"\s+", " ", t).strip()
        t = t.replace("[[", "(").replace("]]", ")")
        if not t:
            continue
        if len(t) > SUMMARY_MAX:
            t = t[:SUMMARY_MAX].rsplit(" ", 1)[0] + "..."
        return t
    return "Claude Code session"


def format_conversation(messages: list[dict]) -> str:
    out: list[str] = []
    for m in messages:
        label = "User" if m["role"] == "user" else "Claude"
        out.append(f"**{label}:**")
        out.append(m["text"])
        out.append("")
    return "\n".join(out).rstrip() + "\n"


def slugify(text: str) -> str:
    s = re.sub(r"[^\w\s-]", "", text).strip()
    s = re.sub(r"\s+", "-", s)
    return s[:SLUG_MAX] or "session"


def atomic_write(path: Path, content: str) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    fd, tmp = tempfile.mkstemp(dir=str(path.parent), prefix=".tmp_", suffix=path.suffix)
    try:
        with os.fdopen(fd, "w") as f:
            f.write(content)
        os.replace(tmp, path)
    except Exception:
        try:
            os.unlink(tmp)
        except OSError:
            pass
        raise


@contextmanager
def hook_lock():
    LOCK_PATH.parent.mkdir(parents=True, exist_ok=True)
    f = open(LOCK_PATH, "w")
    try:
        fcntl.flock(f.fileno(), fcntl.LOCK_EX)
        yield
    finally:
        try:
            fcntl.flock(f.fileno(), fcntl.LOCK_UN)
        finally:
            f.close()


def load_index() -> dict:
    if not INDEX_PATH.exists():
        return {}
    try:
        return json.loads(INDEX_PATH.read_text())
    except (json.JSONDecodeError, OSError):
        return {}


def save_index(idx: dict) -> None:
    atomic_write(INDEX_PATH, json.dumps(idx, indent=2, sort_keys=True))


def find_session_in_vault(session_id: str) -> Path | None:
    if not SESSIONS_DIR.exists():
        return None
    needle = f"session_id: {session_id}"
    for note in SESSIONS_DIR.rglob("*.md"):
        try:
            with open(note) as f:
                head = f.read(800)
        except OSError:
            continue
        if needle in head:
            return note
    return None


def ensure_daily_note(date_iso: str) -> Path:
    p = DAILY_DIR / f"{date_iso}.md"
    if not p.exists():
        atomic_write(p, "\n\n".join(DAILY_SECTIONS) + "\n")
        return p
    content = p.read_text()
    missing = [s for s in DAILY_SECTIONS if s not in content]
    if missing:
        atomic_write(p, content.rstrip() + "\n\n" + "\n\n".join(missing) + "\n")
    return p


def insert_session_backlink(date_iso: str, line: str) -> None:
    """Insert `line` at the end of the ## Claude Sessions section. No-op
    if the section already has a line for the same session note (checked
    by the wikilink target inside `line`)."""
    p = ensure_daily_note(date_iso)
    content = p.read_text()
    lines = content.split("\n")

    section_start = None
    section_end = len(lines)
    for i, l in enumerate(lines):
        if l.strip() == "## Claude Sessions":
            section_start = i
            for j in range(i + 1, len(lines)):
                if lines[j].startswith("## "):
                    section_end = j
                    break
            break
    if section_start is None:
        return

    m = re.search(r"\[\[([^|\]]+)", line)
    link_target = m.group(1) if m else None

    if link_target:
        marker = f"[[{link_target}"
        for j in range(section_start + 1, section_end):
            if marker in lines[j]:
                return  # already present — freeze it

    last_content = section_start
    for j in range(section_start + 1, section_end):
        if lines[j].strip():
            last_content = j
    insert_at = last_content + 1
    lines.insert(insert_at, line)
    atomic_write(p, "\n".join(lines))


def main() -> None:
    hook = parse_stdin()
    event = hook.get("hook_event_name", "")

    transcript_path = find_transcript(hook)
    if not transcript_path:
        return

    data = parse_transcript(transcript_path)
    messages = data["messages"]
    if not messages:
        return

    session_id = data["session_id"]
    project = os.path.basename(data["cwd"]) if data["cwd"] else "unknown"
    msg_count = len(messages)

    started_dt = parse_iso_ts(data["first_ts"])
    last_dt = parse_iso_ts(data["last_ts"]) if data["last_ts"] else datetime.now().astimezone()
    now = datetime.now().astimezone()
    today_iso = now.strftime("%Y-%m-%d")
    summary = generate_summary(messages)

    with hook_lock():
        idx = load_index()
        entry = idx.get(session_id) or {}
        note_relpath = entry.get("note_path")
        backlinks = list(entry.get("daily_backlinks", []))

        note_path = (VAULT / note_relpath) if note_relpath else None
        if note_path and not note_path.exists():
            found = find_session_in_vault(session_id)
            if found:
                note_path = found
                note_relpath = str(note_path.relative_to(VAULT))
            else:
                note_relpath = None
                note_path = None

        if not note_relpath:
            scan = find_session_in_vault(session_id)
            if scan:
                note_path = scan
                note_relpath = str(scan.relative_to(VAULT))

        if not note_relpath:
            year = started_dt.strftime("%Y")
            month = started_dt.strftime("%m")
            time_str = started_dt.strftime("%H%M")
            date_iso = started_dt.strftime("%Y-%m-%d")
            slug = slugify(summary)
            filename = f"{date_iso}-{time_str}-{slug}.md"
            note_relpath = f"Claude Sessions/{year}/{month}/{filename}"
            note_path = VAULT / note_relpath

        status = "completed" if event == "SessionEnd" else "active"
        date_dmy = started_dt.strftime("%d/%m/%Y")
        time_hm = started_dt.strftime("%H:%M")

        frontmatter = (
            "---\n"
            "type: claude-session\n"
            f"session_id: {session_id}\n"
            f"date: {date_dmy}\n"
            f"time: {time_hm}\n"
            f"project: {project}\n"
            f"message_count: {msg_count}\n"
            f"started_at: {started_dt.strftime('%Y-%m-%d %H:%M:%S')}\n"
            f"last_updated_at: {last_dt.strftime('%Y-%m-%d %H:%M:%S')}\n"
            "tags: []\n"
            f"status: {status}\n"
            f"transcript: {transcript_path}\n"
            "---\n"
        )
        body = (
            f"\n## Summary\n{summary}\n\n"
            f"## Conversation\n{format_conversation(messages)}"
        )
        atomic_write(note_path, frontmatter + body)

        note_relpath_no_ext = note_relpath[:-3] if note_relpath.endswith(".md") else note_relpath
        backlink_line = (
            f"- \U0001f916 **{now.strftime('%H:%M')}** — "
            f"[[{note_relpath_no_ext}|{summary}]] "
            f"({msg_count} messages, {project})"
        )
        insert_session_backlink(today_iso, backlink_line)
        if today_iso not in backlinks:
            backlinks.append(today_iso)

        idx[session_id] = {
            "note_path": note_relpath,
            "started_at": started_dt.strftime("%Y-%m-%d %H:%M:%S"),
            "last_updated_at": last_dt.strftime("%Y-%m-%d %H:%M:%S"),
            "message_count": msg_count,
            "daily_backlinks": backlinks,
            "status": status,
            "project": project,
        }
        save_index(idx)


if __name__ == "__main__":
    try:
        main()
    except Exception as exc:
        sys.stderr.write(f"obsidian_export hook error: {exc}\n")
        sys.exit(0)