Read Apple Notes directly from its SQLite store

Splits the original single-file exporter into a package and adds a query
layer, so notes can be searched and read live instead of only exported.

Reading the Core Data store rather than driving Notes.app through AppleScript
is what makes full-text search practical. Measured on an 818-note store,
osascript needs ~0.325s per note body (`first note whose id is` is an
unindexed linear scan), so searching everything costs minutes; decoding every
gzipped protobuf body here takes 0.11s total. No cache, so results are always
current.

- store.py   read-only SQLite access, folder tree, attachments
- parser.py  gzip + protobuf decoding and Markdown rendering (moved verbatim)
- search.py  search, recency listing, single-note reads
- export.py  Markdown output and the incremental manifest

Three correctness details found while building this:

- Open with mode=ro, never immutable=1. The immutable flag makes SQLite skip
  the -wal file and silently serve a stale snapshot; the live store routinely
  carries megabytes of un-checkpointed WAL.
- Recently Deleted is ZFOLDERTYPE = 1, and notes in it keep
  ZMARKEDFORDELETION = 0, so the existing filter still surfaced them. Matching
  the folder type is also language-independent. Now excluded by default.
- Creation timestamps live in whichever of ZCREATIONDATE..3 is populated, so
  every candidate column is read.

Export no longer writes a CLAUDE.md into the output directory unless asked
with --write-claude-md.

tests/test_parity.py keeps the original suite byte-for-byte as a regression
check on the split; test_search.py covers the new query layer against a
synthetic store. 48 tests pass.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

src/apple_notes_cli/__init__.py

@@ -0,0 +1,3 @@
+"""Read Apple Notes directly from its local Core Data store."""
+
+__version__ = "0.1.0"

src/apple_notes_cli/__main__.py

@@ -0,0 +1,4 @@
+from .cli import main
+
+if __name__ == "__main__":
+    raise SystemExit(main())

src/apple_notes_cli/cli.py

@@ -0,0 +1,278 @@
+"""Command line entry point."""
+
+from __future__ import annotations
+
+import argparse
+import datetime as dt
+import json
+import sys
+from pathlib import Path
+
+from . import search as search_mod
+from .store import DEFAULT_NOTES_DIR, connect, default_database
+
+EPILOG = """\
+examples:
+  apple-notes recent --since 2d
+  apple-notes search "OpenTelemetry" --json
+  apple-notes search "TODO|FIXME" --regex --folder Work
+  apple-notes show 8554
+  apple-notes export ~/Downloads/notes --incremental
+"""
+
+
+def add_common(parser: argparse.ArgumentParser) -> None:
+    parser.add_argument(
+        "--notes-dir",
+        type=Path,
+        default=DEFAULT_NOTES_DIR,
+        help=f"Apple Notes group container (default: {DEFAULT_NOTES_DIR})",
+    )
+    parser.add_argument(
+        "--database",
+        type=Path,
+        help="explicit NoteStore.sqlite path, overriding --notes-dir",
+    )
+    parser.add_argument(
+        "--include-password-protected",
+        action="store_true",
+        help="include locked notes (their bodies stay encrypted and unreadable)",
+    )
+    parser.add_argument(
+        "--include-trashed",
+        action="store_true",
+        help="include notes sitting in Recently Deleted",
+    )
+
+
+def add_filters(parser: argparse.ArgumentParser) -> None:
+    parser.add_argument(
+        "--since",
+        help="only notes modified since this time (30m, 6h, 2d, 1w, or 2026-09-01)",
+    )
+    parser.add_argument("--folder", help="restrict to folder paths containing this text")
+    parser.add_argument("--json", action="store_true", help="emit JSON instead of text")
+
+
+def local_time(value: str | None) -> str:
+    """Render a stored UTC timestamp in the viewer's own timezone.
+
+    Timestamps are kept as UTC ISO-8601 for machine consumers (``--json``,
+    export frontmatter); only human-facing output is localized.
+    """
+    if not value:
+        return "?"
+    try:
+        return dt.datetime.fromisoformat(value).astimezone().strftime("%Y-%m-%d %H:%M")
+    except ValueError:
+        return value
+
+
+def open_store(args: argparse.Namespace):
+    database = args.database or default_database(args.notes_dir)
+    return connect(database)
+
+
+def resolve_since(args: argparse.Namespace):
+    if not getattr(args, "since", None):
+        return None
+    return search_mod.parse_since(args.since)
+
+
+def cmd_recent(args: argparse.Namespace) -> int:
+    with open_store(args) as conn:
+        notes = search_mod.recent(
+            conn,
+            limit=args.limit,
+            since=resolve_since(args),
+            folder=args.folder,
+            include_password_protected=args.include_password_protected,
+            include_trashed=args.include_trashed,
+        )
+    if args.json:
+        print(json.dumps([n.__dict__ for n in notes], ensure_ascii=False, indent=2))
+        return 0
+    if not notes:
+        print("no notes matched")
+        return 0
+    for note in notes:
+        print(f"{note.pk:>7}  {local_time(note.modified):<16}  {note.folder}/{note.title}")
+    return 0
+
+
+def cmd_search(args: argparse.Namespace) -> int:
+    with open_store(args) as conn:
+        hits = search_mod.search(
+            conn,
+            args.query,
+            regex=args.regex,
+            case_sensitive=args.case_sensitive,
+            limit=args.limit,
+            since=resolve_since(args),
+            folder=args.folder,
+            include_password_protected=args.include_password_protected,
+            include_trashed=args.include_trashed,
+            snippet_width=args.snippet_width,
+            max_snippets=0 if args.no_snippets else args.max_snippets,
+            search_titles=not args.body_only,
+        )
+    if args.json:
+        payload = [
+            {
+                **hit.note.__dict__,
+                "match_count": hit.match_count,
+                "snippets": hit.snippets,
+            }
+            for hit in hits
+        ]
+        print(json.dumps(payload, ensure_ascii=False, indent=2))
+        return 0
+    if not hits:
+        print(f"no notes matched {args.query!r}")
+        return 1
+    for hit in hits:
+        note = hit.note
+        print(
+            f"{note.pk:>7}  {local_time(note.modified):<16}  "
+            f"{note.folder}/{note.title}  ({hit.match_count}x)"
+        )
+        for snippet in hit.snippets:
+            print(f"         {snippet}")
+    return 0
+
+
+def cmd_show(args: argparse.Namespace) -> int:
+    with open_store(args) as conn:
+        try:
+            note, body = search_mod.show(conn, args.id, markdown=not args.plain)
+        except KeyError as exc:
+            print(str(exc), file=sys.stderr)
+            return 1
+    if args.json:
+        print(json.dumps({**note.__dict__, "body": body}, ensure_ascii=False, indent=2))
+        return 0
+    print(f"# {note.title}\n")
+    print(f"- id: {note.pk}")
+    print(f"- folder: {note.folder}")
+    print(f"- created: {local_time(note.created)}")
+    print(f"- modified: {local_time(note.modified)}\n")
+    print(body)
+    return 0
+
+
+def cmd_folders(args: argparse.Namespace) -> int:
+    with open_store(args) as conn:
+        rows = search_mod.folders(conn, include_trashed=args.include_trashed)
+    if args.json:
+        print(json.dumps([{"folder": f, "notes": c} for f, c in rows], ensure_ascii=False, indent=2))
+        return 0
+    for path, count in rows:
+        print(f"{count:>6}  {path}")
+    return 0
+
+
+def cmd_export(args: argparse.Namespace) -> int:
+    from .export import export_notes
+
+    return export_notes(args)
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(
+        prog="apple-notes",
+        description="Search, read, and export Apple Notes by reading the local database directly.",
+        epilog=EPILOG,
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+    )
+    sub = parser.add_subparsers(dest="command", required=True)
+
+    p_recent = sub.add_parser("recent", help="list recently modified notes")
+    p_recent.add_argument("-n", "--limit", type=int, default=20, help="max notes (default: 20)")
+    add_filters(p_recent)
+    add_common(p_recent)
+    p_recent.set_defaults(func=cmd_recent)
+
+    p_search = sub.add_parser("search", help="full-text search across note bodies")
+    p_search.add_argument("query", help="text to find, or a regex with --regex")
+    p_search.add_argument("-n", "--limit", type=int, help="max results (default: all)")
+    p_search.add_argument("--regex", action="store_true", help="treat query as a regex")
+    p_search.add_argument(
+        "--case-sensitive", action="store_true", help="match case exactly"
+    )
+    p_search.add_argument(
+        "--body-only", action="store_true", help="do not match against titles"
+    )
+    p_search.add_argument(
+        "--max-snippets", type=int, default=3, help="context snippets per note (default: 3)"
+    )
+    p_search.add_argument(
+        "--snippet-width", type=int, default=60, help="context chars each side (default: 60)"
+    )
+    p_search.add_argument("--no-snippets", action="store_true", help="omit context snippets")
+    add_filters(p_search)
+    add_common(p_search)
+    p_search.set_defaults(func=cmd_search)
+
+    p_show = sub.add_parser("show", help="print one note by id")
+    p_show.add_argument("id", type=int, help="note id, as shown by search/recent")
+    p_show.add_argument(
+        "--plain", action="store_true", help="raw extracted text instead of Markdown"
+    )
+    p_show.add_argument("--json", action="store_true", help="emit JSON instead of text")
+    add_common(p_show)
+    p_show.set_defaults(func=cmd_show)
+
+    p_folders = sub.add_parser("folders", help="list folders with note counts")
+    p_folders.add_argument("--json", action="store_true", help="emit JSON instead of text")
+    add_common(p_folders)
+    p_folders.set_defaults(func=cmd_folders)
+
+    p_export = sub.add_parser("export", help="export every note to Markdown files")
+    p_export.add_argument(
+        "output", nargs="?", type=Path, default=Path.cwd() / "exported-notes",
+        help="output directory (default: ./exported-notes)",
+    )
+    p_export.add_argument("--limit", type=int, help="export at most this many notes")
+    p_export.add_argument("--dry-run", action="store_true", help="report without writing")
+    p_export.add_argument(
+        "--incremental", action="store_true",
+        help="skip unchanged notes and prune deleted ones via the manifest",
+    )
+    p_export.add_argument(
+        "--keep-stale", action="store_true",
+        help="with --incremental, do not delete exports of removed notes",
+    )
+    p_export.add_argument(
+        "--no-attachments", action="store_true", help="do not copy attachment files"
+    )
+    p_export.add_argument(
+        "--write-claude-md", action="store_true",
+        help="also drop a CLAUDE.md into the output directory describing the layout",
+    )
+    p_export.add_argument(
+        "--no-snapshot", action="store_true",
+        help="read the live database instead of copying it to a temporary snapshot first",
+    )
+    add_common(p_export)
+    p_export.set_defaults(func=cmd_export)
+
+    return parser
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = build_parser()
+    args = parser.parse_args(argv)
+    try:
+        return args.func(args)
+    except FileNotFoundError as exc:
+        print(str(exc), file=sys.stderr)
+        return 2
+    except ValueError as exc:
+        print(f"error: {exc}", file=sys.stderr)
+        return 2
+    except BrokenPipeError:
+        return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

src/apple_notes_cli/export.py

@@ -0,0 +1,528 @@
+"""Write notes to disk as Markdown, with incremental-manifest bookkeeping."""
+
+from __future__ import annotations
+
+import argparse
+import contextlib
+import datetime as dt
+import hashlib
+import json
+import os
+import shutil
+import subprocess
+import sqlite3
+import sys
+from pathlib import Path
+
+from .models import ExportResult, Note
+from .store import copy_live_database, iter_notes, parse_iso_timestamp
+from .utils import sanitize_filename
+
+MANIFEST_FILENAME = ".apple_notes_export_manifest.json"
+MANIFEST_VERSION = 1
+IGNORED_PRUNE_FILENAMES = {".DS_Store"}
+
+
+def set_macos_creation_time(path: Path, created: dt.datetime) -> None:
+    setfile = shutil.which("SetFile")
+    if not setfile:
+        return
+    local = created.astimezone()
+    formatted = local.strftime("%m/%d/%Y %H:%M:%S")
+    with contextlib.suppress(OSError, subprocess.SubprocessError):
+        subprocess.run(
+            [setfile, "-d", formatted, str(path)],
+            check=False,
+            stdout=subprocess.DEVNULL,
+            stderr=subprocess.DEVNULL,
+        )
+
+
+def apply_file_timestamps(path: Path, *, created: str | None, modified: str | None) -> None:
+    created_dt = parse_iso_timestamp(created)
+    modified_dt = parse_iso_timestamp(modified) or created_dt
+    if modified_dt is not None:
+        timestamp = modified_dt.timestamp()
+        with contextlib.suppress(OSError):
+            os.utime(path, (timestamp, timestamp))
+    if created_dt is not None and sys.platform == "darwin":
+        set_macos_creation_time(path, created_dt)
+
+
+
+
+def yaml_scalar(value: str | int | None) -> str:
+    if value is None:
+        return '""'
+    if isinstance(value, int):
+        return str(value)
+    return json.dumps(value, ensure_ascii=False)
+
+
+
+
+def unique_path(path: Path) -> Path:
+    if not path.exists():
+        return path
+    stem = path.stem
+    suffix = path.suffix
+    parent = path.parent
+    counter = 2
+    while True:
+        candidate = parent / f"{stem}-{counter}{suffix}"
+        if not candidate.exists():
+            return candidate
+        counter += 1
+
+
+def note_filename_stem(title: str, pk: int) -> str:
+    return f"{sanitize_filename(title)}-{pk}"
+
+
+
+
+def build_markdown(note: Note) -> str:
+    lines = [
+        "---",
+        f"title: {yaml_scalar(note.title)}",
+        f"id: {note.pk}",
+        f"apple_notes_id: {yaml_scalar(note.apple_notes_id)}",
+        f"created: {yaml_scalar(note.created)}",
+        f"modified: {yaml_scalar(note.modified)}",
+        f"folder: {yaml_scalar(note.folder_path)}",
+    ]
+
+    if note.attachments:
+        lines.append("attachments:")
+        for attachment in note.attachments:
+            lines.append(f"  - filename: {yaml_scalar(attachment.output_name)}")
+            lines.append(f"    identifier: {yaml_scalar(attachment.identifier)}")
+            lines.append(f"    type: {yaml_scalar(attachment.type_uti)}")
+    else:
+        lines.append("attachments: []")
+
+    if note.links:
+        lines.append("links:")
+        for url in note.links:
+            lines.append(f"  - url: {yaml_scalar(url)}")
+            lines.append(f"    title: {yaml_scalar(url)}")
+    else:
+        lines.append("links: []")
+
+    lines.append(f"source: notes://showNote?identifier={note.apple_notes_id}")
+    lines.append("---")
+    lines.append("")
+    lines.append(note.body.rstrip())
+    lines.append("")
+    return "\n".join(lines)
+
+
+def file_fingerprint(path: Path | None) -> dict[str, int | str] | None:
+    if path is None or not path.exists():
+        return None
+    stat = path.stat()
+    return {
+        "path": str(path),
+        "size": stat.st_size,
+        "mtime_ns": stat.st_mtime_ns,
+    }
+
+
+def note_signature(note: Note) -> str:
+    payload = {
+        "markdown": build_markdown(note),
+        "attachments": [
+            {
+                "identifier": attachment.identifier,
+                "type": attachment.type_uti,
+                "output_name": attachment.output_name,
+                "source": file_fingerprint(attachment.source_path),
+            }
+            for attachment in note.attachments
+        ],
+    }
+    encoded = json.dumps(payload, ensure_ascii=False, sort_keys=True).encode("utf-8")
+    return hashlib.sha256(encoded).hexdigest()
+
+
+def path_relative_to_output(output_root: Path, path: Path) -> str:
+    return str(path.resolve().relative_to(output_root.resolve()))
+
+
+def manifest_file(output_root: Path) -> Path:
+    return output_root / MANIFEST_FILENAME
+
+
+def load_manifest(output_root: Path) -> dict:
+    path = manifest_file(output_root)
+    if not path.exists():
+        return {"version": MANIFEST_VERSION, "notes": {}}
+    try:
+        data = json.loads(path.read_text(encoding="utf-8"))
+    except json.JSONDecodeError:
+        return {"version": MANIFEST_VERSION, "notes": {}}
+    if not isinstance(data, dict) or not isinstance(data.get("notes"), dict):
+        return {"version": MANIFEST_VERSION, "notes": {}}
+    return data
+
+
+def manifest_entry(output_root: Path, note: Note, result: ExportResult, signature: str) -> dict:
+    return {
+        "id": note.pk,
+        "apple_notes_id": note.apple_notes_id,
+        "title": note.title,
+        "folder": note.folder_path,
+        "modified": note.modified,
+        "signature": signature,
+        "path": path_relative_to_output(output_root, result.note_path),
+        "attachments": [path_relative_to_output(output_root, path) for path in result.attachment_paths],
+    }
+
+
+def safe_output_path(output_root: Path, relative_path: str) -> Path | None:
+    if not relative_path or os.path.isabs(relative_path):
+        return None
+    root = output_root.resolve()
+    path = (root / relative_path).resolve()
+    try:
+        path.relative_to(root)
+    except ValueError:
+        return None
+    return path
+
+
+def manifest_paths_exist(output_root: Path, entry: dict) -> bool:
+    paths = [entry.get("path"), *entry.get("attachments", [])]
+    for relative_path in paths:
+        if not isinstance(relative_path, str):
+            return False
+        path = safe_output_path(output_root, relative_path)
+        if path is None or not path.exists():
+            return False
+    return True
+
+
+def path_has_mtime(path: Path, modified: str | None) -> bool:
+    modified_dt = parse_iso_timestamp(modified)
+    if modified_dt is None:
+        return True
+    with contextlib.suppress(OSError):
+        return abs(path.stat().st_mtime - modified_dt.timestamp()) < 1.0
+    return False
+
+
+def manifest_entry_timestamps_match(output_root: Path, entry: dict) -> bool:
+    modified = entry.get("modified")
+    if not isinstance(modified, str):
+        return True
+    paths = [entry.get("path"), *entry.get("attachments", [])]
+    for relative_path in paths:
+        if not isinstance(relative_path, str):
+            return False
+        path = safe_output_path(output_root, relative_path)
+        if path is None or not path.exists() or not path_has_mtime(path, modified):
+            return False
+    return True
+
+
+def manifest_referenced_paths(notes: dict[str, dict]) -> set[str]:
+    paths: set[str] = set()
+    for entry in notes.values():
+        if not isinstance(entry, dict):
+            continue
+        note_path = entry.get("path")
+        if isinstance(note_path, str):
+            paths.add(note_path)
+        for attachment_path in entry.get("attachments", []):
+            if isinstance(attachment_path, str):
+                paths.add(attachment_path)
+    return paths
+
+
+def prune_empty_dirs(output_root: Path, start: Path) -> None:
+    root = output_root.resolve()
+    current = start.resolve()
+    while current != root:
+        try:
+            current.relative_to(root)
+        except ValueError:
+            return
+        entries = list(current.iterdir()) if current.exists() else []
+        if entries and all(entry.name in IGNORED_PRUNE_FILENAMES and entry.is_file() for entry in entries):
+            for entry in entries:
+                entry.unlink()
+        try:
+            current.rmdir()
+        except OSError:
+            return
+        current = current.parent
+
+
+def prune_all_empty_dirs(output_root: Path) -> None:
+    if not output_root.exists():
+        return
+    directories = [path for path in output_root.rglob("*") if path.is_dir()]
+    directories.sort(key=lambda path: len(path.relative_to(output_root).parts), reverse=True)
+    for directory in directories:
+        prune_empty_dirs(output_root, directory)
+
+
+def remove_manifest_entry_paths(output_root: Path, entry: dict, keep: set[str] | None = None) -> int:
+    keep = keep or set()
+    deleted = 0
+    paths = [entry.get("path"), *entry.get("attachments", [])]
+    for relative_path in paths:
+        if not isinstance(relative_path, str) or relative_path in keep:
+            continue
+        path = safe_output_path(output_root, relative_path)
+        if path is None or not path.exists() or not path.is_file():
+            continue
+        path.unlink()
+        deleted += 1
+        prune_empty_dirs(output_root, path.parent)
+    return deleted
+
+
+def exported_note_attachment_names(markdown: str) -> list[str]:
+    lines = markdown.splitlines()
+    if not lines or lines[0] != "---":
+        return []
+
+    names: list[str] = []
+    for line in lines[1:]:
+        if line == "---":
+            break
+        stripped = line.strip()
+        if not stripped.startswith("- filename: "):
+            continue
+        raw = stripped.removeprefix("- filename: ").strip()
+        with contextlib.suppress(json.JSONDecodeError):
+            value = json.loads(raw)
+            if isinstance(value, str) and value:
+                names.append(value)
+    return names
+
+
+def is_exported_note_markdown(path: Path) -> bool:
+    try:
+        text = path.read_text(encoding="utf-8", errors="replace")
+    except OSError:
+        return False
+    return text.startswith("---\n") and "\nsource: notes://showNote?identifier=" in text.split("\n---", 1)[0]
+
+
+def remove_orphaned_export_files(output_root: Path, referenced_paths: set[str]) -> int:
+    if not output_root.exists():
+        return 0
+
+    deleted = 0
+    for note_path in sorted(output_root.rglob("*.md")):
+        relative_note_path = path_relative_to_output(output_root, note_path)
+        if relative_note_path in referenced_paths or not is_exported_note_markdown(note_path):
+            continue
+
+        text = note_path.read_text(encoding="utf-8", errors="replace")
+        attachment_names = exported_note_attachment_names(text)
+        note_path.unlink()
+        deleted += 1
+
+        for attachment_name in attachment_names:
+            attachment_path = safe_output_path(output_root, str(Path(relative_note_path).parent / attachment_name))
+            if attachment_path is None or not attachment_path.exists() or not attachment_path.is_file():
+                continue
+            relative_attachment_path = path_relative_to_output(output_root, attachment_path)
+            if relative_attachment_path in referenced_paths:
+                continue
+            attachment_path.unlink()
+            deleted += 1
+
+        prune_empty_dirs(output_root, note_path.parent)
+    return deleted
+
+
+def write_manifest(output_root: Path, notes: dict[str, dict]) -> None:
+    output_root.mkdir(parents=True, exist_ok=True)
+    payload = {
+        "version": MANIFEST_VERSION,
+        "generated_at": dt.datetime.now(dt.timezone.utc).isoformat(),
+        "notes": notes,
+    }
+    manifest_file(output_root).write_text(
+        json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True) + "\n",
+        encoding="utf-8",
+    )
+
+
+
+
+def expected_note_path(output_root: Path, note: Note) -> Path:
+    filename = f"{note_filename_stem(note.title, note.pk)}.md"
+    return output_root / Path(note.folder_path) / filename
+
+
+def write_note(output_root: Path, note: Note, copy_attachments: bool) -> ExportResult:
+    note_dir = output_root / Path(note.folder_path)
+    note_dir.mkdir(parents=True, exist_ok=True)
+
+    attachment_paths: list[Path] = []
+    if copy_attachments:
+        used_paths: set[str] = set()
+        for attachment in note.attachments:
+            if attachment.source_path is None:
+                continue
+            destination = note_dir / attachment.output_name
+            destination.parent.mkdir(parents=True, exist_ok=True)
+            if attachment.output_name in used_paths:
+                destination = unique_path(destination)
+            used_paths.add(path_relative_to_output(note_dir, destination))
+            shutil.copy2(attachment.source_path, destination)
+            apply_file_timestamps(destination, created=note.created, modified=note.modified)
+            attachment_paths.append(destination)
+
+    path = expected_note_path(output_root, note)
+    path.write_text(build_markdown(note), encoding="utf-8")
+    apply_file_timestamps(path, created=note.created, modified=note.modified)
+    return ExportResult(note_path=path, attachment_paths=attachment_paths)
+
+
+def write_claude_files(output_root: Path, exported_count: int) -> None:
+    output_root.mkdir(parents=True, exist_ok=True)
+    generated_at = dt.datetime.now().strftime("%Y-%m-%d, %H:%M")
+    content = f"""# Apple Notes Export
+
+This directory contains Apple Notes exported as markdown files.
+
+## Structure
+
+- Each Notes folder is represented as a subdirectory.
+- Each note is saved as `{{note-title}}-{{id}}.md`.
+- Attachments are saved in `attachments/{{id}}/` next to the markdown file.
+- Frontmatter includes title, Apple Notes ID, dates, folder, links, attachments, and the `notes://` source URL.
+
+## Working With Notes
+
+1. Preserve `source` links when editing or referencing notes.
+2. Keep the `attachments/` folder next to the markdown files that reference it.
+3. Keep folder structure intact when moving exported notes.
+
+---
+Generated by `apple-notes export` • Notes exported: {exported_count} • Last updated: {generated_at}
+"""
+    (output_root / "CLAUDE.md").write_text(content, encoding="utf-8")
+
+
+def export_notes(args: argparse.Namespace) -> int:
+    notes_dir = args.notes_dir.expanduser().resolve()
+    db_path = args.database.expanduser().resolve() if args.database else notes_dir / "NoteStore.sqlite"
+    if not db_path.exists():
+        raise FileNotFoundError(f"Notes database not found: {db_path}")
+
+    temp_db: Path | None = None
+    if args.no_snapshot:
+        read_db = db_path
+    else:
+        temp_db = copy_live_database(db_path)
+        read_db = temp_db
+
+    output_root = args.output.expanduser().resolve()
+    old_manifest = load_manifest(output_root) if args.incremental else {"notes": {}}
+    old_notes: dict[str, dict] = old_manifest.get("notes", {})
+    current_manifest_notes: dict[str, dict] = {}
+    exported = 0
+    written = 0
+    skipped = 0
+    deleted_notes = 0
+    deleted_files = 0
+    try:
+        conn = sqlite3.connect(read_db)
+        conn.row_factory = sqlite3.Row
+        try:
+            notes = iter_notes(
+                conn,
+                notes_dir,
+                include_password_protected=args.include_password_protected,
+                limit=args.limit,
+                include_trashed=getattr(args, "include_trashed", False),
+            )
+            for note in notes:
+                exported += 1
+                key = str(note.pk)
+                signature = note_signature(note)
+                expected_relpath = path_relative_to_output(output_root, expected_note_path(output_root, note))
+                old_entry = old_notes.get(key)
+                unchanged = (
+                    args.incremental
+                    and isinstance(old_entry, dict)
+                    and old_entry.get("signature") == signature
+                    and old_entry.get("path") == expected_relpath
+                    and manifest_paths_exist(output_root, old_entry)
+                    and manifest_entry_timestamps_match(output_root, old_entry)
+                )
+
+                if unchanged:
+                    skipped += 1
+                    current_manifest_notes[key] = old_entry
+                    continue
+
+                written += 1
+                if args.dry_run:
+                    current_manifest_notes[key] = {
+                        "id": note.pk,
+                        "apple_notes_id": note.apple_notes_id,
+                        "title": note.title,
+                        "folder": note.folder_path,
+                        "modified": note.modified,
+                        "signature": signature,
+                        "path": expected_relpath,
+                        "attachments": [],
+                    }
+                    continue
+                if args.incremental and isinstance(old_entry, dict):
+                    deleted_files += remove_manifest_entry_paths(
+                        output_root,
+                        old_entry,
+                        keep={expected_relpath},
+                    )
+                result = write_note(output_root, note, copy_attachments=not args.no_attachments)
+                new_entry = manifest_entry(output_root, note, result, signature)
+                current_manifest_notes[key] = new_entry
+                if args.incremental and isinstance(old_entry, dict):
+                    new_paths = {new_entry["path"], *new_entry.get("attachments", [])}
+                    deleted_files += remove_manifest_entry_paths(output_root, old_entry, keep=new_paths)
+        finally:
+            conn.close()
+    finally:
+        if temp_db is not None:
+            temp_db.unlink(missing_ok=True)
+
+    if args.incremental and not args.keep_stale and args.limit is None:
+        stale_keys = set(old_notes) - set(current_manifest_notes)
+        deleted_notes = len(stale_keys)
+        if not args.dry_run:
+            for key in stale_keys:
+                entry = old_notes.get(key)
+                if isinstance(entry, dict):
+                    deleted_files += remove_manifest_entry_paths(output_root, entry)
+
+    if not args.dry_run:
+        if args.incremental and not args.keep_stale and args.limit is None:
+            deleted_files += remove_orphaned_export_files(
+                output_root,
+                manifest_referenced_paths(current_manifest_notes),
+            )
+        if getattr(args, "write_claude_md", False):
+            write_claude_files(output_root, exported)
+        write_manifest(output_root, current_manifest_notes)
+        if args.incremental:
+            prune_all_empty_dirs(output_root)
+    print(f"exported_notes={exported}")
+    if args.incremental:
+        print(f"notes_written={written}")
+        print(f"notes_skipped={skipped}")
+        print(f"notes_deleted={deleted_notes}")
+        print(f"files_deleted={deleted_files}")
+    if not args.dry_run:
+        print(f"output={output_root}")
+    return 0
+
+

src/apple_notes_cli/models.py

@@ -0,0 +1,72 @@
+"""Dataclasses shared across the parser, store, and exporter."""
+
+from __future__ import annotations
+
+import dataclasses
+from pathlib import Path
+
+
+@dataclasses.dataclass(frozen=True)
+class ProtoField:
+    number: int
+    wire_type: int
+    value: int | bytes
+
+
+@dataclasses.dataclass(frozen=True)
+class Attachment:
+    identifier: str
+    type_uti: str | None
+    source_path: Path | None
+    output_name: str
+
+    @property
+    def markdown_reference(self) -> str:
+        reference = self.output_name
+        alt = Path(reference).name.replace("[", "\\[").replace("]", "\\]")
+        if reference.lower().endswith((".png", ".jpg", ".jpeg", ".gif", ".webp")):
+            return f"![{alt}]({reference})"
+        return f"[{alt}]({reference})"
+
+
+@dataclasses.dataclass(frozen=True)
+class Note:
+    pk: int
+    title: str
+    apple_notes_id: str
+    folder_path: str
+    created: str | None
+    modified: str | None
+    body: str
+    attachments: list[Attachment]
+    links: list[str]
+
+
+@dataclasses.dataclass(frozen=True)
+class ExportResult:
+    note_path: Path
+    attachment_paths: list[Path]
+
+
+@dataclasses.dataclass(frozen=True)
+class EmbeddedTable:
+    identifier: str
+    markdown: str
+
+
+@dataclasses.dataclass(frozen=True)
+class TextRun:
+    start: int
+    end: int
+    style_type: int | None
+    indent: int
+    block_quote: int | None
+    checklist_done: int | None
+    font_weight: int | None
+    underlined: int | None
+    strikethrough: int | None
+    superscript: int | None
+    link: str | None
+    attachment_identifier: str | None
+    attachment_type: str | None
+

src/apple_notes_cli/parser.py

@@ -0,0 +1,780 @@
+"""Pure decoding of the Apple Notes protobuf payload into Markdown.
+
+Nothing in here touches SQLite or the filesystem: every function takes bytes
+or text and returns text. Lifted verbatim from the original exporter.
+"""
+
+from __future__ import annotations
+
+import contextlib
+import gzip
+import re
+from typing import Iterator
+
+from .models import ProtoField, TextRun
+
+URL_RE = re.compile(r"https?://[^\s<>()\[\]\"'，。！？；、]+")
+STYLE_TYPE_TITLE = 0
+STYLE_TYPE_HEADING = 1
+STYLE_TYPE_SUBHEADING = 2
+STYLE_TYPE_MONOSPACED = 4
+STYLE_TYPE_DOTTED_LIST = 100
+STYLE_TYPE_DASHED_LIST = 101
+STYLE_TYPE_NUMBERED_LIST = 102
+STYLE_TYPE_CHECKBOX = 103
+STYLE_TYPE_BLOCK_QUOTE = 1
+FONT_TYPE_BOLD = 1
+FONT_TYPE_ITALIC = 2
+FONT_TYPE_BOLD_ITALIC = 3
+
+
+def read_varint(data: bytes, offset: int) -> tuple[int, int]:
+    value = 0
+    shift = 0
+    while offset < len(data):
+        byte = data[offset]
+        offset += 1
+        value |= (byte & 0x7F) << shift
+        if byte & 0x80 == 0:
+            return value, offset
+        shift += 7
+        if shift > 70:
+            raise ValueError("protobuf varint is too large")
+    raise ValueError("unexpected end of protobuf varint")
+
+
+def iter_proto_fields(data: bytes) -> Iterator[ProtoField]:
+    offset = 0
+    while offset < len(data):
+        key, offset = read_varint(data, offset)
+        number = key >> 3
+        wire_type = key & 0x07
+
+        if wire_type == 0:
+            value, offset = read_varint(data, offset)
+            yield ProtoField(number, wire_type, value)
+        elif wire_type == 1:
+            if offset + 8 > len(data):
+                raise ValueError("unexpected end of fixed64 field")
+            yield ProtoField(number, wire_type, data[offset : offset + 8])
+            offset += 8
+        elif wire_type == 2:
+            length, offset = read_varint(data, offset)
+            end = offset + length
+            if end > len(data):
+                raise ValueError("unexpected end of length-delimited field")
+            yield ProtoField(number, wire_type, data[offset:end])
+            offset = end
+        elif wire_type == 5:
+            if offset + 4 > len(data):
+                raise ValueError("unexpected end of fixed32 field")
+            yield ProtoField(number, wire_type, data[offset : offset + 4])
+            offset += 4
+        else:
+            raise ValueError(f"unsupported protobuf wire type: {wire_type}")
+
+
+def first_bytes_field(data: bytes, number: int) -> bytes:
+    for field in iter_proto_fields(data):
+        if field.number == number and field.wire_type == 2 and isinstance(field.value, bytes):
+            return field.value
+    raise ValueError(f"protobuf field {number} was not found")
+
+
+def first_varint_field(data: bytes, number: int) -> int | None:
+    for field in iter_proto_fields(data):
+        if field.number == number and field.wire_type == 0 and isinstance(field.value, int):
+            return field.value
+    return None
+
+
+def first_string_field(data: bytes, number: int) -> str | None:
+    for field in iter_proto_fields(data):
+        if field.number != number or field.wire_type != 2 or not isinstance(field.value, bytes):
+            continue
+        with contextlib.suppress(UnicodeDecodeError):
+            return field.value.decode("utf-8")
+    return None
+
+
+def note_content_from_zdata(zdata: bytes) -> bytes:
+    if zdata.startswith(b"\x1f\x8b"):
+        zdata = gzip.decompress(zdata)
+    document = first_bytes_field(zdata, 2)
+    return first_bytes_field(document, 3)
+
+
+def find_largest_utf8_field(data: bytes, depth: int = 0) -> str:
+    best = ""
+    if depth > 5:
+        return best
+    with contextlib.suppress(ValueError):
+        for field in iter_proto_fields(data):
+            if field.wire_type != 2 or not isinstance(field.value, bytes):
+                continue
+            with contextlib.suppress(UnicodeDecodeError):
+                text = field.value.decode("utf-8")
+                printable = sum(ch.isprintable() or ch in "\n\r\t" for ch in text)
+                if text and printable / max(len(text), 1) > 0.9 and len(text) > len(best):
+                    best = text
+            nested = find_largest_utf8_field(field.value, depth + 1)
+            if len(nested) > len(best):
+                best = nested
+    return best
+
+
+def extract_note_text(zdata: bytes) -> str:
+    try:
+        content = note_content_from_zdata(zdata)
+        text = first_bytes_field(content, 2).decode("utf-8")
+    except (UnicodeDecodeError, ValueError, OSError):
+        text = find_largest_utf8_field(zdata)
+
+    return normalize_note_text(text)
+
+
+def parse_paragraph_style(data: bytes) -> tuple[int | None, int, int | None, int | None]:
+    style_type = first_varint_field(data, 1)
+    indent = first_varint_field(data, 4) or 0
+    block_quote = first_varint_field(data, 8)
+    checklist_done = None
+    for field in iter_proto_fields(data):
+        if field.number != 5 or field.wire_type != 2 or not isinstance(field.value, bytes):
+            continue
+        checklist_done = first_varint_field(field.value, 2)
+        break
+    return style_type, indent, block_quote, checklist_done
+
+
+def parse_text_runs(content: bytes) -> list[TextRun]:
+    runs: list[TextRun] = []
+    offset = 0
+    for field in iter_proto_fields(content):
+        if field.number != 5 or field.wire_type != 2 or not isinstance(field.value, bytes):
+            continue
+
+        length = first_varint_field(field.value, 1)
+        if length is None:
+            continue
+
+        style_type = None
+        indent = 0
+        block_quote = None
+        checklist_done = None
+        font_weight = first_varint_field(field.value, 5)
+        underlined = first_varint_field(field.value, 6)
+        strikethrough = first_varint_field(field.value, 7)
+        superscript = first_varint_field(field.value, 8)
+        link = first_string_field(field.value, 9)
+        attachment_identifier = None
+        attachment_type = None
+        for run_field in iter_proto_fields(field.value):
+            if run_field.wire_type != 2 or not isinstance(run_field.value, bytes):
+                continue
+            if run_field.number == 2:
+                style_type, indent, block_quote, checklist_done = parse_paragraph_style(run_field.value)
+            elif run_field.number == 12:
+                attachment_identifier = first_string_field(run_field.value, 1)
+                attachment_type = first_string_field(run_field.value, 2)
+
+        runs.append(
+            TextRun(
+                offset,
+                offset + length,
+                style_type,
+                indent,
+                block_quote,
+                checklist_done,
+                font_weight,
+                underlined,
+                strikethrough,
+                superscript,
+                link,
+                attachment_identifier,
+                attachment_type,
+            )
+        )
+        offset += length
+    return runs
+
+
+def run_at_offset(runs: list[TextRun], offset: int) -> TextRun | None:
+    for run in runs:
+        if run.start <= offset < run.end:
+            return run
+    return None
+
+
+def line_ranges(text: str) -> list[tuple[int, str]]:
+    ranges: list[tuple[int, str]] = []
+    offset = 0
+    for line in text.split("\n"):
+        ranges.append((offset, line))
+        offset += len(line) + 1
+    while ranges and ranges[0][1] == "":
+        ranges.pop(0)
+    while ranges and ranges[-1][1] == "":
+        ranges.pop()
+    return ranges
+
+
+def overlapping_runs(runs: list[TextRun], start: int, end: int) -> list[TextRun]:
+    return [run for run in runs if run.start < end and run.end > start]
+
+
+def block_style_key(run: TextRun) -> tuple[int | None, int, int | None, int | None]:
+    return (run.style_type, run.indent, run.block_quote, run.checklist_done)
+
+
+def dominant_block_run(runs: list[TextRun], start: int, end: int) -> TextRun | None:
+    scores: dict[tuple[int | None, int, int | None, int | None], tuple[int, int, TextRun]] = {}
+    for run in overlapping_runs(runs, start, end):
+        overlap = min(end, run.end) - max(start, run.start)
+        if overlap <= 0:
+            continue
+        key = block_style_key(run)
+        total, _last_start, _representative = scores.get(key, (0, run.start, run))
+        scores[key] = (total + overlap, run.start, run)
+    if not scores:
+        return None
+    return max(scores.values(), key=lambda item: (item[0], item[1]))[2]
+
+
+def apply_inline_markdown(text: str, run: TextRun, object_replacements: dict[str, str] | None = None) -> str:
+    if not text:
+        return text
+
+    match = re.match(r"^(\s*)(.*?)(\s*)$", text.replace("\x00", "\u2400"), re.DOTALL)
+    if not match:
+        return text
+    leading, result, trailing = match.groups()
+    if not result:
+        return leading + trailing
+
+    if "\ufffc" in result and run.attachment_identifier and object_replacements:
+        replacement = object_replacements.get(run.attachment_identifier)
+        if replacement:
+            result = result.replace("\ufffc", replacement)
+
+    if run.link:
+        label = result.replace("[", "\\[").replace("]", "\\]")
+        url = run.link.replace(")", "%29")
+        result = f"[{label}]({url})"
+
+    if run.font_weight == FONT_TYPE_BOLD:
+        result = f"**{result}**"
+    elif run.font_weight == FONT_TYPE_ITALIC:
+        result = f"*{result}*"
+    elif run.font_weight == FONT_TYPE_BOLD_ITALIC:
+        result = f"***{result}***"
+
+    if run.strikethrough == 1:
+        result = f"~~{result}~~"
+    if run.underlined == 1:
+        result = f"<u>{result}</u>"
+    if run.superscript == 1:
+        result = f"<sup>{result}</sup>"
+    elif run.superscript == -1:
+        result = f"<sub>{result}</sub>"
+    return leading + result + trailing
+
+
+def add_spacing_after_bold_labels(text: str) -> str:
+    return re.sub(r"(\*\*[^*\n]{1,40}[：:]\*\*)(?=\S)", r"\1 ", text)
+
+
+def fence_for_lines(lines: list[str]) -> str:
+    longest = max((len(match.group(0)) for line in lines for match in re.finditer(r"`+", line)), default=0)
+    return "`" * max(3, longest + 1)
+
+
+def should_preserve_plain_block(lines: list[str]) -> bool:
+    if len(lines) < 2:
+        return False
+    if any(is_markdown_block_line(line) or has_markdown_inline_markup(line) for line in lines):
+        return False
+    if any(line.startswith("\t") for line in lines):
+        return True
+    return len(lines) >= 3 and any(line.strip() == "↓" for line in lines)
+
+
+def has_markdown_inline_markup(line: str) -> bool:
+    return "](" in line or "<u>" in line or "</u>" in line
+
+
+def is_markdown_block_line(line: str) -> bool:
+    return bool(re.match(r"^\s*(#{1,6}\s|[-*+]\s|\d+\.\s|>\s|```|\|)", line))
+
+
+def replace_leading_tabs(line: str) -> str:
+    tabs = len(line) - len(line.lstrip("\t"))
+    if tabs == 0:
+        return line
+    return "&emsp;" * tabs + line[tabs:]
+
+
+def with_hard_breaks(lines: list[str]) -> list[str]:
+    rendered: list[str] = []
+    for index, line in enumerate(lines):
+        line = replace_leading_tabs(line)
+        next_line = lines[index + 1] if index + 1 < len(lines) else None
+        if next_line is not None and not is_markdown_block_line(line) and not is_markdown_block_line(next_line):
+            rendered.append(line.rstrip() + "  ")
+        else:
+            rendered.append(line)
+    return rendered
+
+
+def preserve_plain_structure_blocks(text: str) -> str:
+    groups: list[list[str]] = []
+    current: list[str] = []
+    for line in text.split("\n"):
+        if line == "":
+            if current:
+                groups.append(current)
+                current = []
+            groups.append([])
+        else:
+            current.append(line)
+    if current:
+        groups.append(current)
+
+    rendered: list[str] = []
+    for group in groups:
+        if not group:
+            rendered.append("")
+            continue
+        if should_preserve_plain_block(group):
+            fence = fence_for_lines(group)
+            rendered.extend([fence, *group, fence])
+        else:
+            rendered.extend(with_hard_breaks(group))
+    return "\n".join(rendered)
+
+
+def inline_style_key(run: TextRun) -> tuple[int | None, int | None, int | None, int | None, str | None, str | None]:
+    return (
+        run.font_weight,
+        run.underlined,
+        run.strikethrough,
+        run.superscript,
+        run.link,
+        run.attachment_identifier,
+    )
+
+
+def is_plain_inline_key(key: tuple | None) -> bool:
+    return key is None or all(value is None for value in key)
+
+
+def is_plain_segment(key: tuple | None, text: str) -> bool:
+    return is_plain_inline_key(key) and len(text) == 1 and bool(text.strip())
+
+
+def merge_leading_style_fragments(
+    segments: list[tuple[tuple | None, TextRun | None, str]],
+) -> list[tuple[tuple | None, TextRun | None, str]]:
+    merged: list[tuple[tuple | None, TextRun | None, str]] = []
+    index = 0
+    while index < len(segments):
+        key, run, text = segments[index]
+        if (
+            is_plain_segment(key, text)
+            and index + 1 < len(segments)
+            and not is_plain_inline_key(segments[index + 1][0])
+            and segments[index + 1][2]
+            and not segments[index + 1][2][0].isspace()
+        ):
+            next_key, next_run, next_text = segments[index + 1]
+            merged.append((next_key, next_run, text + next_text))
+            index += 2
+            continue
+        merged.append((key, run, text))
+        index += 1
+    return merged
+
+
+def render_inline_markdown(
+    text: str,
+    runs: list[TextRun],
+    start: int,
+    object_replacements: dict[str, str] | None = None,
+) -> str:
+    if not text:
+        return text
+
+    end = start + len(text)
+    segments: list[tuple[tuple | None, TextRun | None, str]] = []
+    cursor = start
+    for run in overlapping_runs(runs, start, end):
+        if run.start > cursor:
+            segments.append((None, None, text[cursor - start : run.start - start]))
+        segment_start = max(cursor, run.start)
+        segment_end = min(end, run.end)
+        if segment_start < segment_end:
+            segment = text[segment_start - start : segment_end - start]
+            key = inline_style_key(run)
+            if segments and segments[-1][0] == key:
+                previous_key, previous_run, previous_text = segments[-1]
+                segments[-1] = (previous_key, previous_run, previous_text + segment)
+            else:
+                segments.append((key, run, segment))
+        cursor = max(cursor, segment_end)
+    if cursor < end:
+        segments.append((None, None, text[cursor - start :]))
+    segments = merge_leading_style_fragments(segments)
+
+    parts: list[str] = []
+    for _key, run, segment in segments:
+        if run is None:
+            parts.append(segment)
+        else:
+            parts.append(apply_inline_markdown(segment, run, object_replacements))
+    return add_spacing_after_bold_labels("".join(parts))
+
+
+def line_prefix(style_type: int | None, run: TextRun | None, line_number: int) -> tuple[str, bool]:
+    indent = "  " * (run.indent if run else 0)
+    if style_type == STYLE_TYPE_CHECKBOX:
+        checked = "x" if run and run.checklist_done == 1 else " "
+        return f"{indent}- [{checked}] ", True
+    if style_type in {STYLE_TYPE_DOTTED_LIST, STYLE_TYPE_DASHED_LIST}:
+        return f"{indent}- ", True
+    if style_type == STYLE_TYPE_NUMBERED_LIST:
+        return f"{indent}{line_number}. ", True
+    if run and run.block_quote == STYLE_TYPE_BLOCK_QUOTE:
+        return "> " * max(run.indent, 1), False
+    if run and run.indent > 0:
+        return "> " * run.indent, False
+    if style_type == STYLE_TYPE_TITLE:
+        return "### ", True
+    if style_type == STYLE_TYPE_HEADING:
+        return "#### ", True
+    if style_type == STYLE_TYPE_SUBHEADING:
+        return "##### ", True
+    if style_type == STYLE_TYPE_MONOSPACED:
+        return "    ", False
+    return "", False
+
+
+def render_note_text_as_markdown(
+    text: str,
+    runs: list[TextRun],
+    object_replacements: dict[str, str] | None = None,
+) -> str:
+    if not runs:
+        return preserve_plain_structure_blocks(normalize_note_text(text))
+
+    rendered: list[str] = []
+    ordered_numbers: dict[int, int] = {}
+    for start, line in line_ranges(text):
+        if not line:
+            rendered.append("")
+            ordered_numbers.clear()
+            continue
+
+        first_content = len(line) - len(line.lstrip())
+        line_start = start + first_content
+        line_end = start + len(line)
+        run = dominant_block_run(runs, line_start, line_end)
+        style_type = run.style_type if run else None
+        if style_type in {STYLE_TYPE_TITLE, STYLE_TYPE_HEADING, STYLE_TYPE_SUBHEADING} and len(line.strip()) > 80:
+            style_type = None
+        if style_type == STYLE_TYPE_NUMBERED_LIST:
+            indent = run.indent if run else 0
+            line_number = ordered_numbers.get(indent, 1)
+            prefix, strip_line = line_prefix(style_type, run, line_number)
+            ordered_numbers[indent] = line_number + 1
+            for deeper_indent in [key for key in ordered_numbers if key > indent]:
+                del ordered_numbers[deeper_indent]
+        else:
+            prefix, strip_line = line_prefix(style_type, run, 1)
+            if style_type not in {STYLE_TYPE_CHECKBOX, STYLE_TYPE_DOTTED_LIST, STYLE_TYPE_DASHED_LIST}:
+                ordered_numbers.clear()
+
+        content = line.lstrip() if strip_line else line
+        content_start = start + (first_content if strip_line else 0)
+        rendered.append(prefix + render_inline_markdown(content, runs, content_start, object_replacements))
+    return preserve_plain_structure_blocks("\n".join(rendered))
+
+
+def extract_note_markdown(zdata: bytes, object_replacements: dict[str, str] | None = None) -> str:
+    try:
+        content = note_content_from_zdata(zdata)
+        text = first_bytes_field(content, 2).decode("utf-8")
+        runs = parse_text_runs(content)
+        return render_note_text_as_markdown(text, runs, object_replacements)
+    except (UnicodeDecodeError, ValueError, OSError):
+        return extract_note_text(zdata)
+
+
+def normalize_note_text(text: str) -> str:
+    text = text.replace("\r\n", "\n").replace("\r", "\n")
+    text = text.replace("\u2028", "\n").replace("\u2029", "\n")
+    return text.strip("\n")
+
+
+
+
+def object_id(data: bytes) -> dict[str, int | str]:
+    result: dict[str, int | str] = {}
+    unsigned = first_varint_field(data, 2)
+    string = first_string_field(data, 4)
+    index = first_varint_field(data, 6)
+    if unsigned is not None:
+        result["unsigned"] = unsigned
+    if string is not None:
+        result["string"] = string
+    if index is not None:
+        result["index"] = index
+    return result
+
+
+def custom_map(data: bytes) -> tuple[int | None, list[tuple[int, dict[str, int | str]]]]:
+    map_type = first_varint_field(data, 1)
+    entries: list[tuple[int, dict[str, int | str]]] = []
+    for field in iter_proto_fields(data):
+        if field.number != 3 or field.wire_type != 2 or not isinstance(field.value, bytes):
+            continue
+        key = first_varint_field(field.value, 1)
+        value = None
+        for entry_field in iter_proto_fields(field.value):
+            if entry_field.number == 2 and entry_field.wire_type == 2 and isinstance(entry_field.value, bytes):
+                value = object_id(entry_field.value)
+                break
+        if key is not None and value is not None:
+            entries.append((key, value))
+    return map_type, entries
+
+
+def dictionary_elements(data: bytes) -> list[tuple[dict[str, int | str], dict[str, int | str]]]:
+    elements: list[tuple[dict[str, int | str], dict[str, int | str]]] = []
+    for field in iter_proto_fields(data):
+        if field.number != 1 or field.wire_type != 2 or not isinstance(field.value, bytes):
+            continue
+        key = None
+        value = None
+        for element_field in iter_proto_fields(field.value):
+            if element_field.wire_type != 2 or not isinstance(element_field.value, bytes):
+                continue
+            if element_field.number == 1:
+                key = object_id(element_field.value)
+            elif element_field.number == 2:
+                value = object_id(element_field.value)
+        if key is not None and value is not None:
+            elements.append((key, value))
+    return elements
+
+
+def entry_field(entry: bytes, number: int) -> bytes | None:
+    for field in iter_proto_fields(entry):
+        if field.number == number and field.wire_type == 2 and isinstance(field.value, bytes):
+            return field.value
+    return None
+
+
+def entry_custom_map(entry: bytes) -> tuple[int | None, list[tuple[int, dict[str, int | str]]]] | None:
+    data = entry_field(entry, 13)
+    if data is None:
+        return None
+    return custom_map(data)
+
+
+def entry_type(entry: bytes, type_items: list[str]) -> str | None:
+    decoded = entry_custom_map(entry)
+    if decoded is None:
+        return None
+    type_index, _entries = decoded
+    if type_index is None or type_index >= len(type_items):
+        return None
+    return type_items[type_index]
+
+
+def target_uuid_index(entry: bytes) -> int | None:
+    decoded = entry_custom_map(entry)
+    if decoded is None:
+        return None
+    _type_index, entries = decoded
+    if not entries:
+        return None
+    value = entries[0][1]
+    unsigned = value.get("unsigned")
+    return unsigned if isinstance(unsigned, int) else None
+
+
+def ordered_set_parts(entry: bytes) -> tuple[list[bytes], list[tuple[dict[str, int | str], dict[str, int | str]]]]:
+    ordered_set = entry_field(entry, 16)
+    if ordered_set is None:
+        return [], []
+
+    ordering = entry_field(ordered_set, 1)
+    if ordering is None:
+        return [], []
+
+    attachments: list[bytes] = []
+    elements: list[tuple[dict[str, int | str], dict[str, int | str]]] = []
+    for field in iter_proto_fields(ordering):
+        if field.wire_type != 2 or not isinstance(field.value, bytes):
+            continue
+        if field.number == 1:
+            for array_field in iter_proto_fields(field.value):
+                if array_field.number != 2 or array_field.wire_type != 2 or not isinstance(array_field.value, bytes):
+                    continue
+                for attachment_field in iter_proto_fields(array_field.value):
+                    if attachment_field.number == 2 and attachment_field.wire_type == 2 and isinstance(
+                        attachment_field.value, bytes
+                    ):
+                        attachments.append(attachment_field.value)
+        elif field.number == 2:
+            elements = dictionary_elements(field.value)
+    return attachments, elements
+
+
+def index_from_object_id(value: dict[str, int | str]) -> int | None:
+    index = value.get("index")
+    return index if isinstance(index, int) else None
+
+
+def table_axis_indices(entry: bytes, table_objects: list[bytes], uuid_items: list[bytes]) -> dict[int, int]:
+    attachments, elements = ordered_set_parts(entry)
+    indices: dict[int, int] = {}
+    for position, uuid in enumerate(attachments):
+        with contextlib.suppress(ValueError):
+            indices[uuid_items.index(uuid)] = position
+
+    for key, value in elements:
+        key_index = index_from_object_id(key)
+        value_index = index_from_object_id(value)
+        if key_index is None or value_index is None:
+            continue
+        if key_index >= len(table_objects) or value_index >= len(table_objects):
+            continue
+        key_uuid = target_uuid_index(table_objects[key_index])
+        value_uuid = target_uuid_index(table_objects[value_index])
+        if key_uuid in indices and value_uuid is not None:
+            indices[value_uuid] = indices[key_uuid]
+    return indices
+
+
+def note_message_to_markdown(note_message: bytes) -> str:
+    text = first_bytes_field(note_message, 2).decode("utf-8")
+    return render_note_text_as_markdown(text, parse_text_runs(note_message)).strip()
+
+
+def markdown_table_cell(value: str) -> str:
+    value = value.replace("|", "\\|")
+    value = re.sub(r"\n{2,}", "<br><br>", value)
+    return value.replace("\n", "<br>").strip()
+
+
+def markdown_table(rows: list[list[str]]) -> str:
+    rows = [[markdown_table_cell(cell) for cell in row] for row in rows if row]
+    if not rows:
+        return ""
+    width = max(len(row) for row in rows)
+    normalized = [row + [""] * (width - len(row)) for row in rows]
+    header = normalized[0]
+    separator = ["---"] * width
+    lines = [
+        "| " + " | ".join(header) + " |",
+        "| " + " | ".join(separator) + " |",
+    ]
+    for row in normalized[1:]:
+        lines.append("| " + " | ".join(row) + " |")
+    return "\n".join(lines)
+
+
+def extract_markdown_table(mergeable_data: bytes) -> str:
+    if mergeable_data.startswith(b"\x1f\x8b"):
+        mergeable_data = gzip.decompress(mergeable_data)
+
+    data = first_bytes_field(first_bytes_field(mergeable_data, 2), 3)
+    table_objects: list[bytes] = []
+    key_items: list[str] = []
+    type_items: list[str] = []
+    uuid_items: list[bytes] = []
+    for field in iter_proto_fields(data):
+        if field.number == 3 and field.wire_type == 2 and isinstance(field.value, bytes):
+            table_objects.append(field.value)
+        elif field.number == 4 and field.wire_type == 2 and isinstance(field.value, bytes):
+            key_items.append(field.value.decode("utf-8"))
+        elif field.number == 5 and field.wire_type == 2 and isinstance(field.value, bytes):
+            type_items.append(field.value.decode("utf-8"))
+        elif field.number == 6 and field.wire_type == 2 and isinstance(field.value, bytes):
+            uuid_items.append(field.value)
+
+    for table_entry in table_objects:
+        if entry_type(table_entry, type_items) != "com.apple.notes.ICTable":
+            continue
+
+        decoded = entry_custom_map(table_entry)
+        if decoded is None:
+            continue
+        _type_index, map_entries = decoded
+        row_indices: dict[int, int] = {}
+        column_indices: dict[int, int] = {}
+        cell_columns_entry: bytes | None = None
+
+        for key_index, value in map_entries:
+            if key_index >= len(key_items):
+                continue
+            object_index = index_from_object_id(value)
+            if object_index is None or object_index >= len(table_objects):
+                continue
+            key = key_items[key_index]
+            if key == "crRows":
+                row_indices = table_axis_indices(table_objects[object_index], table_objects, uuid_items)
+            elif key == "crColumns":
+                column_indices = table_axis_indices(table_objects[object_index], table_objects, uuid_items)
+            elif key == "cellColumns":
+                cell_columns_entry = table_objects[object_index]
+
+        if not row_indices or not column_indices or cell_columns_entry is None:
+            continue
+
+        rows = [["" for _ in range(max(column_indices.values()) + 1)] for _ in range(max(row_indices.values()) + 1)]
+        dictionary = entry_field(cell_columns_entry, 6)
+        if dictionary is None:
+            continue
+        for column_key, column_value in dictionary_elements(dictionary):
+            column_object_index = index_from_object_id(column_key)
+            row_dictionary_index = index_from_object_id(column_value)
+            if column_object_index is None or row_dictionary_index is None:
+                continue
+            if column_object_index >= len(table_objects) or row_dictionary_index >= len(table_objects):
+                continue
+            column_uuid = target_uuid_index(table_objects[column_object_index])
+            if column_uuid not in column_indices:
+                continue
+            row_dictionary = entry_field(table_objects[row_dictionary_index], 6)
+            if row_dictionary is None:
+                continue
+            for row_key, row_value in dictionary_elements(row_dictionary):
+                row_object_index = index_from_object_id(row_key)
+                cell_index = index_from_object_id(row_value)
+                if row_object_index is None or cell_index is None:
+                    continue
+                if row_object_index >= len(table_objects) or cell_index >= len(table_objects):
+                    continue
+                row_uuid = target_uuid_index(table_objects[row_object_index])
+                if row_uuid not in row_indices:
+                    continue
+                note_message = entry_field(table_objects[cell_index], 10)
+                if note_message is None:
+                    continue
+                rows[row_indices[row_uuid]][column_indices[column_uuid]] = note_message_to_markdown(note_message)
+        table = markdown_table(rows)
+        if table:
+            return table
+    return ""
+
+
+def summary_table_fallback(summary: str | None) -> str:
+    if not summary:
+        return ""
+    lines = [line for line in normalize_note_text(summary).splitlines()]
+    if not lines:
+        return ""
+    return "\n".join(f"> {line}" if line else ">" for line in lines)
+
+

src/apple_notes_cli/search.py

@@ -0,0 +1,295 @@
+"""Search and browse notes straight from the live database.
+
+Apple's store keeps note bodies as gzipped protobuf in ``ZICNOTEDATA.ZDATA``,
+and there is no FTS index anywhere in the schema, so full-text search means
+inflating every body and matching in memory. That sounds expensive and isn't:
+a ~800-note store decodes end to end in well under a second, which is why this
+does no caching and always reports the current state of the database.
+"""
+
+from __future__ import annotations
+
+import contextlib
+import dataclasses
+import datetime as dt
+import re
+import sqlite3
+
+from .parser import extract_note_markdown, extract_note_text
+from .store import apple_date, query_folder_paths, trashed_folder_pks
+
+SINCE_RE = re.compile(r"^(\d+)([smhdw])$")
+SINCE_UNITS = {"s": "seconds", "m": "minutes", "h": "hours", "d": "days", "w": "weeks"}
+APPLE_EPOCH_OFFSET = 978307200.0
+
+
+@dataclasses.dataclass(frozen=True)
+class NoteSummary:
+    """A note's metadata, read without inflating its body."""
+
+    pk: int
+    title: str
+    folder: str
+    created: str | None
+    modified: str | None
+
+
+@dataclasses.dataclass(frozen=True)
+class SearchHit:
+    note: NoteSummary
+    match_count: int
+    snippets: list[str]
+
+
+def parse_since(value: str) -> dt.datetime:
+    """Accept a relative offset (``30m``, ``2d``, ``1w``) or an ISO date."""
+    match = SINCE_RE.match(value.strip())
+    if match:
+        amount, unit = int(match.group(1)), match.group(2)
+        delta = dt.timedelta(**{SINCE_UNITS[unit]: amount})
+        return dt.datetime.now(dt.timezone.utc) - delta
+    text = value.strip()
+    try:
+        parsed = dt.datetime.fromisoformat(text)
+    except ValueError as exc:
+        raise ValueError(
+            f"cannot read {value!r} as a time; use 30m/6h/2d/1w or 2026-09-01"
+        ) from exc
+    if parsed.tzinfo is None:
+        parsed = parsed.astimezone()
+    return parsed
+
+
+def _to_apple_time(moment: dt.datetime) -> float:
+    return moment.timestamp() - APPLE_EPOCH_OFFSET
+
+
+def _build_query(
+    *,
+    since: dt.datetime | None,
+    folder: str | None,
+    include_password_protected: bool,
+    with_body: bool,
+    newest_first: bool,
+    limit: int | None,
+    trashed: set[int],
+) -> tuple[str, list]:
+    columns = [
+        "n.Z_PK",
+        "n.ZTITLE1",
+        "n.ZFOLDER",
+        # Which creation-date column is populated varies by how and when the
+        # note was created, so every candidate is read and the first non-null
+        # one wins. Matches the exporter's behaviour.
+        "n.ZCREATIONDATE",
+        "n.ZCREATIONDATE1",
+        "n.ZCREATIONDATE2",
+        "n.ZCREATIONDATE3",
+        "n.ZMODIFICATIONDATE1",
+    ]
+    if with_body:
+        columns.append("d.ZDATA")
+    where = ["d.ZDATA IS NOT NULL", "IFNULL(n.ZMARKEDFORDELETION, 0) = 0"]
+    params: list = []
+    if not include_password_protected:
+        where.append("IFNULL(n.ZISPASSWORDPROTECTED, 0) = 0")
+    if trashed:
+        placeholders = ", ".join(str(int(pk)) for pk in sorted(trashed))
+        where.append(f"IFNULL(n.ZFOLDER, 0) NOT IN ({placeholders})")
+    if since is not None:
+        where.append("n.ZMODIFICATIONDATE1 >= ?")
+        params.append(_to_apple_time(since))
+    order = "DESC" if newest_first else "ASC"
+    sql = f"""
+        SELECT {", ".join(columns)}
+        FROM ZICCLOUDSYNCINGOBJECT n
+        JOIN ZICNOTEDATA d ON n.Z_PK = d.ZNOTE
+        WHERE {" AND ".join(where)}
+        ORDER BY n.ZMODIFICATIONDATE1 {order}, n.Z_PK {order}
+    """
+    # A folder filter cannot go in SQL: folder paths are assembled in Python
+    # from the parent chain, so limiting rows here would drop nested matches.
+    if limit is not None and folder is None:
+        sql += " LIMIT ?"
+        params.append(limit)
+    return sql, params
+
+
+def _created(row: sqlite3.Row) -> str | None:
+    for column in (
+        "ZCREATIONDATE1",
+        "ZCREATIONDATE",
+        "ZCREATIONDATE2",
+        "ZCREATIONDATE3",
+    ):
+        with contextlib.suppress(IndexError):
+            stamp = apple_date(row[column])
+            if stamp:
+                return stamp
+    return None
+
+
+def _summary(row: sqlite3.Row, folder_paths: dict[int, str]) -> NoteSummary:
+    folder = folder_paths.get(row["ZFOLDER"], "Notes") if row["ZFOLDER"] else "Notes"
+    return NoteSummary(
+        pk=int(row["Z_PK"]),
+        title=(row["ZTITLE1"] or "Untitled").strip() or "Untitled",
+        folder=folder,
+        created=_created(row),
+        modified=apple_date(row["ZMODIFICATIONDATE1"]),
+    )
+
+
+def _folder_matches(candidate: str, wanted: str) -> bool:
+    return wanted.lower() in candidate.lower()
+
+
+def recent(
+    conn: sqlite3.Connection,
+    *,
+    limit: int | None = 20,
+    since: dt.datetime | None = None,
+    folder: str | None = None,
+    include_password_protected: bool = False,
+    include_trashed: bool = False,
+) -> list[NoteSummary]:
+    """List notes newest-first. Reads metadata only — no body inflation."""
+    folder_paths = query_folder_paths(conn)
+    sql, params = _build_query(
+        since=since,
+        folder=folder,
+        include_password_protected=include_password_protected,
+        with_body=False,
+        newest_first=True,
+        limit=limit,
+        trashed=set() if include_trashed else trashed_folder_pks(conn),
+    )
+    results = []
+    for row in conn.execute(sql, params):
+        summary = _summary(row, folder_paths)
+        if folder and not _folder_matches(summary.folder, folder):
+            continue
+        results.append(summary)
+        if limit is not None and len(results) >= limit:
+            break
+    return results
+
+
+def _snippet(text: str, start: int, end: int, width: int) -> str:
+    left = max(0, start - width)
+    right = min(len(text), end + width)
+    fragment = text[left:right].replace("\n", " ").strip()
+    fragment = re.sub(r"\s+", " ", fragment)
+    prefix = "…" if left > 0 else ""
+    suffix = "…" if right < len(text) else ""
+    return f"{prefix}{fragment}{suffix}"
+
+
+def search(
+    conn: sqlite3.Connection,
+    query: str,
+    *,
+    regex: bool = False,
+    case_sensitive: bool = False,
+    limit: int | None = None,
+    since: dt.datetime | None = None,
+    folder: str | None = None,
+    include_password_protected: bool = False,
+    include_trashed: bool = False,
+    snippet_width: int = 60,
+    max_snippets: int = 3,
+    search_titles: bool = True,
+) -> list[SearchHit]:
+    """Full-text search across every note body.
+
+    Every candidate body is inflated and matched in memory; results are ranked
+    by match count, then by recency.
+    """
+    flags = 0 if case_sensitive else re.IGNORECASE
+    pattern = re.compile(query if regex else re.escape(query), flags)
+
+    folder_paths = query_folder_paths(conn)
+    sql, params = _build_query(
+        since=since,
+        folder=folder,
+        include_password_protected=include_password_protected,
+        with_body=True,
+        newest_first=True,
+        limit=None,
+        trashed=set() if include_trashed else trashed_folder_pks(conn),
+    )
+
+    hits: list[SearchHit] = []
+    for row in conn.execute(sql, params):
+        summary = _summary(row, folder_paths)
+        if folder and not _folder_matches(summary.folder, folder):
+            continue
+        try:
+            body = extract_note_text(row["ZDATA"])
+        except Exception:
+            # A single unreadable note must not abort the whole search.
+            continue
+        haystack = f"{summary.title}\n{body}" if search_titles else body
+        matches = list(pattern.finditer(haystack))
+        if not matches:
+            continue
+        snippets = [
+            _snippet(haystack, m.start(), m.end(), snippet_width)
+            for m in matches[:max_snippets]
+        ]
+        hits.append(SearchHit(note=summary, match_count=len(matches), snippets=snippets))
+
+    # ISO-8601 strings sort chronologically, so one pass ranks by match count
+    # and breaks ties by recency.
+    hits.sort(key=lambda h: (h.match_count, h.note.modified or ""), reverse=True)
+    return hits[:limit] if limit is not None else hits
+
+
+def show(
+    conn: sqlite3.Connection,
+    pk: int,
+    *,
+    markdown: bool = True,
+) -> tuple[NoteSummary, str]:
+    """Fetch one note's metadata and body by primary key."""
+    row = conn.execute(
+        """
+        SELECT n.Z_PK, n.ZTITLE1, n.ZFOLDER, n.ZCREATIONDATE,
+               n.ZCREATIONDATE1, n.ZCREATIONDATE2, n.ZCREATIONDATE3,
+               n.ZMODIFICATIONDATE1, d.ZDATA
+        FROM ZICCLOUDSYNCINGOBJECT n
+        JOIN ZICNOTEDATA d ON n.Z_PK = d.ZNOTE
+        WHERE n.Z_PK = ? AND d.ZDATA IS NOT NULL
+        """,
+        (pk,),
+    ).fetchone()
+    if row is None:
+        raise KeyError(f"no note with id {pk}")
+    summary = _summary(row, query_folder_paths(conn))
+    body = (
+        extract_note_markdown(row["ZDATA"]) if markdown else extract_note_text(row["ZDATA"])
+    )
+    return summary, body
+
+
+def folders(
+    conn: sqlite3.Connection, *, include_trashed: bool = False
+) -> list[tuple[str, int]]:
+    """Every folder path with its note count, ordered by path."""
+    folder_paths = query_folder_paths(conn)
+    trashed = set() if include_trashed else trashed_folder_pks(conn)
+    counts: dict[str, int] = {}
+    rows = conn.execute(
+        """
+        SELECT n.ZFOLDER
+        FROM ZICCLOUDSYNCINGOBJECT n
+        JOIN ZICNOTEDATA d ON n.Z_PK = d.ZNOTE
+        WHERE d.ZDATA IS NOT NULL AND IFNULL(n.ZMARKEDFORDELETION, 0) = 0
+        """
+    )
+    for row in rows:
+        if row["ZFOLDER"] and int(row["ZFOLDER"]) in trashed:
+            continue
+        path = folder_paths.get(row["ZFOLDER"], "Notes") if row["ZFOLDER"] else "Notes"
+        counts[path] = counts.get(path, 0) + 1
+    return sorted(counts.items())

src/apple_notes_cli/store.py

@@ -0,0 +1,395 @@
+"""Read-only access to the Apple Notes Core Data store.
+
+Two ways in:
+
+* ``connect()`` opens the live database with ``mode=ro``. SQLite's WAL mode
+  supports concurrent readers, so this is safe while Notes.app is running and
+  it always sees the newest committed data.
+* ``snapshot()`` copies the database via ``backup()`` first. Slower, but gives
+  a stable point-in-time view for long operations like a full export.
+
+Never open the store with ``immutable=1``: that flag makes SQLite skip the
+``-wal`` file entirely, silently serving a stale snapshot that is missing the
+most recent notes.
+"""
+
+from __future__ import annotations
+
+import contextlib
+import dataclasses
+import datetime as dt
+import sqlite3
+import tempfile
+from pathlib import Path
+from typing import Iterator
+
+from .models import Attachment, EmbeddedTable, Note
+from .parser import (
+    URL_RE,
+    extract_markdown_table,
+    extract_note_markdown,
+    summary_table_fallback,
+)
+from .utils import sanitize_filename
+
+APPLE_EPOCH = dt.datetime(2001, 1, 1, tzinfo=dt.timezone.utc)
+DEFAULT_NOTES_DIR = Path.home() / "Library/Group Containers/group.com.apple.notes"
+
+
+def default_database(notes_dir: Path | None = None) -> Path:
+    return (notes_dir or DEFAULT_NOTES_DIR) / "NoteStore.sqlite"
+
+
+def connect(database: Path) -> sqlite3.Connection:
+    """Open the live store read-only. Sees WAL content; does not block Notes.app."""
+    if not database.exists():
+        raise FileNotFoundError(
+            f"Apple Notes database not found at {database}.\n"
+            "Notes.app must have been launched at least once on this machine."
+        )
+    conn = sqlite3.connect(f"file:{database}?mode=ro", uri=True)
+    conn.row_factory = sqlite3.Row
+    return conn
+
+
+def apple_date(value: float | int | None) -> str | None:
+    if value is None:
+        return None
+    try:
+        numeric = float(value)
+    except (TypeError, ValueError):
+        return None
+    if numeric <= 0:
+        return None
+    return (APPLE_EPOCH + dt.timedelta(seconds=numeric)).isoformat()
+
+
+def parse_iso_timestamp(value: str | None) -> dt.datetime | None:
+    if not value:
+        return None
+    with contextlib.suppress(ValueError):
+        parsed = dt.datetime.fromisoformat(value.replace("Z", "+00:00"))
+        if parsed.tzinfo is None:
+            parsed = parsed.replace(tzinfo=dt.timezone.utc)
+        return parsed
+    return None
+
+
+
+
+def query_folder_paths(conn: sqlite3.Connection) -> dict[int, str]:
+    rows = conn.execute(
+        """
+        SELECT Z_PK, ZTITLE2, ZPARENT
+        FROM ZICCLOUDSYNCINGOBJECT
+        WHERE ZTITLE2 IS NOT NULL
+        """
+    ).fetchall()
+    titles = {int(row["Z_PK"]): row["ZTITLE2"] or "Notes" for row in rows}
+    parents = {int(row["Z_PK"]): row["ZPARENT"] for row in rows}
+    cache: dict[int, str] = {}
+
+    def build(folder_id: int, seen: set[int] | None = None) -> str:
+        if folder_id in cache:
+            return cache[folder_id]
+        if seen is None:
+            seen = set()
+        if folder_id in seen:
+            return sanitize_filename(titles.get(folder_id, "Notes"))
+        seen.add(folder_id)
+        title = sanitize_filename(titles.get(folder_id, "Notes"))
+        parent = parents.get(folder_id)
+        if parent is None or int(parent) not in titles:
+            cache[folder_id] = title
+        else:
+            cache[folder_id] = str(Path(build(int(parent), seen)) / title)
+        return cache[folder_id]
+
+    for folder_id in titles:
+        build(folder_id)
+    return cache
+
+
+def trashed_folder_pks(conn: sqlite3.Connection) -> set[int]:
+    """Folder primary keys for the trash, plus anything nested inside it.
+
+    ``ZFOLDERTYPE == 1`` marks Recently Deleted independently of the UI
+    language, so this beats matching a localized folder name. Notes sitting in
+    the trash keep ``ZMARKEDFORDELETION = 0``, which is why filtering on that
+    column alone still surfaces them.
+    """
+    rows = conn.execute(
+        """
+        SELECT Z_PK, ZPARENT, ZFOLDERTYPE
+        FROM ZICCLOUDSYNCINGOBJECT
+        WHERE ZTITLE2 IS NOT NULL
+        """
+    ).fetchall()
+    parents = {int(row["Z_PK"]): row["ZPARENT"] for row in rows}
+    roots = {int(row["Z_PK"]) for row in rows if row["ZFOLDERTYPE"] == 1}
+    trashed = set(roots)
+    for folder_pk in parents:
+        chain: set[int] = set()
+        current: int | None = folder_pk
+        while current is not None and current not in chain:
+            chain.add(current)
+            if current in roots:
+                trashed |= chain
+                break
+            parent = parents.get(current)
+            current = int(parent) if parent is not None else None
+    return trashed
+
+
+def find_media_file(notes_dir: Path, media_id: str | None, generation: str | None, filename: str | None) -> Path | None:
+    if not media_id:
+        return None
+    accounts_dir = notes_dir / "Accounts"
+    if not accounts_dir.exists():
+        return None
+    for account_dir in accounts_dir.iterdir():
+        media_dir = account_dir / "Media" / media_id
+        if not media_dir.exists():
+            continue
+        candidates: list[Path] = []
+        if generation and filename:
+            candidates.append(media_dir / generation / filename)
+        if filename:
+            candidates.extend(media_dir.glob(f"*/{filename}"))
+        candidates.extend(path for path in media_dir.glob("*/*") if path.is_file())
+        for candidate in candidates:
+            if candidate.is_file():
+                return candidate
+    return None
+
+
+
+def query_embedded_tables(conn: sqlite3.Connection, note_pk: int) -> list[EmbeddedTable]:
+    rows = conn.execute(
+        """
+        SELECT
+            ZIDENTIFIER AS table_identifier,
+            ZMERGEABLEDATA1 AS mergeable_data,
+            ZMERGEABLEDATA AS legacy_mergeable_data,
+            ZSUMMARY AS summary
+        FROM ZICCLOUDSYNCINGOBJECT
+        WHERE ZNOTE = ?
+          AND ZTYPEUTI = 'com.apple.notes.table'
+        ORDER BY Z_PK
+        """,
+        (note_pk,),
+    ).fetchall()
+
+    tables: list[EmbeddedTable] = []
+    for row in rows:
+        markdown = ""
+        data = row["mergeable_data"] or row["legacy_mergeable_data"]
+        if data:
+            with contextlib.suppress(ValueError, OSError, UnicodeDecodeError):
+                markdown = extract_markdown_table(data)
+        if not markdown:
+            markdown = summary_table_fallback(row["summary"])
+        if markdown:
+            tables.append(EmbeddedTable(identifier=row["table_identifier"] or "", markdown=markdown))
+    return tables
+
+
+def query_attachments(conn: sqlite3.Connection, notes_dir: Path, note_pk: int) -> list[Attachment]:
+    rows = conn.execute(
+        """
+        SELECT
+            a.ZIDENTIFIER AS attachment_identifier,
+            a.ZTYPEUTI AS attachment_type,
+            m.ZIDENTIFIER AS media_identifier,
+            m.ZFILENAME AS media_filename,
+            m.ZGENERATION1 AS media_generation
+        FROM ZICCLOUDSYNCINGOBJECT a
+        LEFT JOIN ZICCLOUDSYNCINGOBJECT m ON a.ZMEDIA = m.Z_PK
+        WHERE a.ZNOTE = ?
+          AND a.ZMEDIA IS NOT NULL
+        ORDER BY a.Z_PK
+        """,
+        (note_pk,),
+    ).fetchall()
+
+    attachments: list[Attachment] = []
+    for row in rows:
+        attachment_id = row["attachment_identifier"] or row["media_identifier"] or "attachment"
+        source = find_media_file(
+            notes_dir,
+            row["media_identifier"],
+            row["media_generation"],
+            row["media_filename"],
+        )
+        suffix = ""
+        if row["media_filename"]:
+            suffix = Path(row["media_filename"]).suffix
+        output_name = sanitize_filename(f"{attachment_id.lower()}{suffix}", fallback="attachment")
+        attachments.append(
+            Attachment(
+                identifier=attachment_id,
+                type_uti=row["attachment_type"],
+                source_path=source,
+                output_name=output_name,
+            )
+        )
+    return attachments
+
+
+def attachment_reference_path(note_id: int, output_name: str) -> str:
+    return str(Path("attachments") / str(note_id) / output_name)
+
+
+def nest_attachments_under_note_id(attachments: list[Attachment], note_id: int) -> list[Attachment]:
+    return [
+        dataclasses.replace(
+            attachment,
+            output_name=attachment_reference_path(note_id, Path(attachment.output_name).name),
+        )
+        for attachment in attachments
+    ]
+
+
+def replace_attachment_placeholders(body: str, attachments: list[Attachment]) -> str:
+    result = body
+    inserted = 0
+    for attachment in attachments:
+        if "\ufffc" not in result:
+            break
+        result = result.replace("\ufffc", attachment.markdown_reference, 1)
+        inserted += 1
+
+    remaining = attachments[inserted:]
+    if remaining:
+        lines = ["", "## Attachments", ""]
+        lines.extend(f"- {attachment.markdown_reference}" for attachment in remaining)
+        result = result.rstrip() + "\n" + "\n".join(lines)
+    return result
+
+
+def replace_table_placeholders(body: str, tables: list[EmbeddedTable]) -> str:
+    result = body
+    inserted = 0
+    for table in tables:
+        if "\ufffc" not in result:
+            break
+        result = result.replace("\ufffc", "\n\n" + table.markdown + "\n\n", 1)
+        inserted += 1
+
+    remaining = tables[inserted:]
+    if remaining:
+        lines = ["", "## Tables", ""]
+        for table in remaining:
+            lines.extend([table.markdown, ""])
+        result = result.rstrip() + "\n" + "\n".join(lines)
+    return result
+
+
+def extract_links(body: str) -> list[str]:
+    seen: set[str] = set()
+    links: list[str] = []
+    for match in URL_RE.finditer(body):
+        url = match.group(0).rstrip(".,;:!?)]}")
+        if url and url not in seen:
+            seen.add(url)
+            links.append(url)
+    return links
+
+
+
+
+def copy_live_database(source: Path) -> Path:
+    temp = tempfile.NamedTemporaryFile(prefix="apple-notes-", suffix=".sqlite", delete=False)
+    temp_path = Path(temp.name)
+    temp.close()
+
+    src = sqlite3.connect(f"file:{source}?mode=ro", uri=True)
+    try:
+        dst = sqlite3.connect(temp_path)
+        try:
+            src.backup(dst)
+        finally:
+            dst.close()
+    finally:
+        src.close()
+    return temp_path
+
+
+def iter_notes(
+    conn: sqlite3.Connection,
+    notes_dir: Path,
+    include_password_protected: bool = False,
+    limit: int | None = None,
+    include_trashed: bool = False,
+) -> Iterator[Note]:
+    folder_paths = query_folder_paths(conn)
+    where = [
+        "d.ZDATA IS NOT NULL",
+        "IFNULL(n.ZMARKEDFORDELETION, 0) = 0",
+    ]
+    if not include_password_protected:
+        where.append("IFNULL(n.ZISPASSWORDPROTECTED, 0) = 0")
+    if not include_trashed:
+        trashed = trashed_folder_pks(conn)
+        if trashed:
+            placeholders = ", ".join(str(int(pk)) for pk in sorted(trashed))
+            where.append(f"IFNULL(n.ZFOLDER, 0) NOT IN ({placeholders})")
+    sql = f"""
+        SELECT
+            n.Z_PK,
+            n.ZTITLE1,
+            n.ZFOLDER,
+            n.ZIDENTIFIER,
+            n.ZCREATIONDATE,
+            n.ZCREATIONDATE1,
+            n.ZCREATIONDATE2,
+            n.ZCREATIONDATE3,
+            n.ZMODIFICATIONDATE1,
+            d.ZDATA
+        FROM ZICCLOUDSYNCINGOBJECT n
+        JOIN ZICNOTEDATA d ON n.Z_PK = d.ZNOTE
+        WHERE {" AND ".join(where)}
+        ORDER BY n.ZMODIFICATIONDATE1 ASC, n.Z_PK ASC
+    """
+    if limit is not None:
+        sql += " LIMIT ?"
+        rows = conn.execute(sql, (limit,))
+    else:
+        rows = conn.execute(sql)
+
+    for row in rows:
+        pk = int(row["Z_PK"])
+        tables = query_embedded_tables(conn, pk)
+        body = extract_note_markdown(row["ZDATA"])
+        body = replace_table_placeholders(body, tables)
+        attachments = query_attachments(conn, notes_dir, pk)
+        title = note_title(row["ZTITLE1"], body)
+        attachments = nest_attachments_under_note_id(attachments, pk)
+        body = replace_attachment_placeholders(body, attachments)
+        folder_path = folder_paths.get(row["ZFOLDER"], "Notes") if row["ZFOLDER"] else "Notes"
+        yield Note(
+            pk=pk,
+            title=title,
+            apple_notes_id=row["ZIDENTIFIER"] or str(pk),
+            folder_path=folder_path,
+            created=apple_date(row["ZCREATIONDATE1"])
+            or apple_date(row["ZCREATIONDATE"])
+            or apple_date(row["ZCREATIONDATE2"])
+            or apple_date(row["ZCREATIONDATE3"]),
+            modified=apple_date(row["ZMODIFICATIONDATE1"]),
+            body=body,
+            attachments=attachments,
+            links=extract_links(body),
+        )
+
+
+
+def note_title(row_title: str | None, body: str) -> str:
+    if row_title and row_title.strip():
+        return row_title.strip()
+    for line in body.splitlines():
+        if line.strip():
+            return line.strip()
+    return "Untitled"
+
+

src/apple_notes_cli/utils.py

@@ -0,0 +1,20 @@
+"""Filename helpers shared by the store and the exporter."""
+
+from __future__ import annotations
+
+import re
+
+FORBIDDEN_FILENAME_CHARS = re.compile(r"[/\\:*?\"<>|\x00-\x1f]")
+
+
+def sanitize_filename(value: str, fallback: str = "Untitled", max_length: int = 180) -> str:
+    value = value.replace("\n", " ").replace("\r", " ").strip()
+    value = FORBIDDEN_FILENAME_CHARS.sub("-", value)
+    value = re.sub(r"\s+", " ", value).strip(" .")
+    if not value:
+        value = fallback
+    if len(value) > max_length:
+        value = value[:max_length].rstrip(" .")
+    return value or fallback
+
+