commit a8615b3c459495812e7ebb895a28d8a5973072bb Author: ingjieye Date: Wed Sep 9 16:31:17 2026 +0800 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 diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml new file mode 100644 index 0000000..b2d475c --- /dev/null +++ b/.github/workflows/tests.yml @@ -0,0 +1,19 @@ +name: tests + +on: + push: + branches: [main] + pull_request: + +jobs: + test: + runs-on: macos-latest + strategy: + matrix: + python-version: ["3.10", "3.13"] + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python-version }} + - run: python -m unittest discover -s tests -v diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..173890e --- /dev/null +++ b/.gitignore @@ -0,0 +1,8 @@ +__pycache__/ +*.py[cod] +.venv/ +dist/ +build/ +*.egg-info/ +.DS_Store +uv.lock diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..f0eeaef --- /dev/null +++ b/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2026 ingjieye + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/README.md b/README.md new file mode 100644 index 0000000..c179c4d --- /dev/null +++ b/README.md @@ -0,0 +1,173 @@ +# apple-notes-cli + +Search, read, and export Apple Notes from the terminal by reading the local +Notes database directly — no AppleScript, no UI automation, no cloud API. + +```console +$ apple-notes search "kubernetes" -n 3 + 4821 2026-09-08 14:02 Work/Infra/Cluster upgrade notes (7x) + …drain the node before the kubernetes control plane restarts… + 3907 2026-08-30 09:15 Work/Runbooks/On-call (2x) + …kubernetes events are the fastest way to see why a pod is pending… + +$ apple-notes recent --since 2d + 4830 2026-09-09 15:59 Work/Meeting notes + 4829 2026-09-09 13:54 Personal/Reading list +``` + +## Why not AppleScript + +The usual approach is `osascript` against Notes.app. It works, but the cost +lands squarely on search. Measured on a real 818-note store: + +| Operation | AppleScript | apple-notes-cli | +| --- | --- | --- | +| Enumerate folders and count notes | 1.85 s | — | +| Fetch one note body | 0.325 s | — | +| **Full-text search every note** | **~266 s** (extrapolated) | **0.11 s** | + +`first note whose id is "…"` is a linear scan inside Notes.app with no index +behind it, so reading every body costs a third of a second each. That is why +AppleScript-based tools either search titles only or cache aggressively and +serve you stale results. Reading the SQLite store sidesteps the whole problem: +one query, then in-memory decoding of every body. + +## Install + +Requires macOS and Python 3.10+. + +```bash +uv tool install git+https://github.com/ingjieye/apple-notes-cli +# or +pipx install git+https://github.com/ingjieye/apple-notes-cli +``` + +Run it straight from a clone without installing: + +```bash +git clone https://github.com/ingjieye/apple-notes-cli && cd apple-notes-cli +uv run python -m apple_notes_cli recent +``` + +macOS may ask you to grant your terminal access to the Notes container the +first time. Nothing here needs Full Disk Access on its own. + +## Commands + +``` +apple-notes search full-text search across every note body +apple-notes recent list recently modified notes +apple-notes show print one note as Markdown +apple-notes folders list folders with note counts +apple-notes export export everything to Markdown files +``` + +### search + +```bash +apple-notes search "OpenTelemetry" +apple-notes search "TODO|FIXME" --regex +apple-notes search "invoice" --folder Finance --since 1w +apple-notes search "postgres" --json | jq -r '.[].title' +apple-notes search "draft" --body-only --no-snippets +``` + +Results rank by match count, then by recency. Titles are searched alongside +bodies unless you pass `--body-only`. `--json` emits structured records, which +is the useful shape for piping into an LLM or another tool. + +### recent + +```bash +apple-notes recent -n 50 +apple-notes recent --since 6h --folder Work +``` + +Metadata only — bodies are never inflated, so this stays fast regardless of +store size. + +### export + +```bash +apple-notes export ~/Downloads/notes +apple-notes export ~/Downloads/notes --incremental +``` + +One Markdown file per note, mirroring your folder tree, with YAML frontmatter +(title, ids, dates, folder, links, attachments, and a `notes://` link back to +the note). Attachments are copied next to the Markdown under +`attachments//`. `--incremental` keeps a manifest so unchanged notes are +skipped and exports of deleted notes are pruned. + +## How it works + +``` +NoteStore.sqlite + └── ZICNOTEDATA.ZDATA gzip → protobuf → text runs → Markdown + └── ZICCLOUDSYNCINGOBJECT titles, dates, folder tree, attachments +``` + +| Module | Role | +| --- | --- | +| `store.py` | read-only SQLite access, folder tree, attachments | +| `parser.py` | gzip + protobuf decoding, Markdown rendering | +| `search.py` | search, recency listing, single-note reads | +| `export.py` | Markdown output and the incremental manifest | + +Three details are worth knowing if you build something similar: + +**Open the database with `mode=ro`, never `immutable=1`.** The `immutable` +flag makes SQLite skip the `-wal` file, which silently serves a stale snapshot +missing your most recent notes. On a live store the `-wal` file routinely holds +megabytes of un-checkpointed data. WAL mode allows concurrent readers, so +`mode=ro` is safe while Notes.app is running and never blocks it. Export uses +SQLite's `backup()` API instead, for a stable point-in-time copy. + +**There is no FTS index.** The schema has no full-text tables, and bodies are +gzipped protobuf blobs, so `WHERE body LIKE …` is not available. Search means +inflating every candidate body and matching in memory. At ~800 notes that is +about 30 ms, so this tool does no caching and always reflects the live store. + +**Recently Deleted is `ZFOLDERTYPE = 1`.** Notes in the trash keep +`ZMARKEDFORDELETION = 0`, so filtering on that column alone still surfaces +them. Matching on the folder type is also language-independent, unlike +comparing against a localized "Recently Deleted" string. They are excluded by +default; pass `--include-trashed` to include them. + +## Exit codes + +`search` follows the `grep` convention: `0` when something matched, `1` when +nothing did, `2` on a bad argument or an unreadable database. That makes it +usable in a conditional: + +```bash +if apple-notes search "release checklist" --no-snippets >/dev/null; then + echo "found it" +fi +``` + +## Reading is all it does + +This tool only reads. It never writes to `NoteStore.sqlite` — modifying the +store directly would corrupt CloudKit's sync state and risk losing notes. If +you need to create or edit notes, drive Notes.app through AppleScript for the +write path and use this for reading. + +Password-protected notes are listed but their bodies stay encrypted and +unreadable; they are skipped unless you pass `--include-password-protected`. + +## Development + +```bash +uv run python -m unittest discover -s tests +``` + +`tests/test_parity.py` is the original single-file exporter's suite, kept +unchanged apart from neutralized example URLs, so it guards the protobuf +decoding against regressions. +`tests/test_search.py` builds a synthetic Core Data store to cover the query +layer, trash handling, and time filters. + +## License + +MIT diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..6fd3b91 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,33 @@ +[project] +name = "apple-notes-cli" +version = "0.1.0" +description = "Search, read, and export Apple Notes by reading the local database directly — no AppleScript, no UI automation." +readme = "README.md" +requires-python = ">=3.10" +license = { text = "MIT" } +authors = [{ name = "ingjieye" }] +keywords = ["apple-notes", "macos", "notes", "cli", "search", "markdown", "export"] +classifiers = [ + "Development Status :: 4 - Beta", + "Environment :: MacOS X", + "Intended Audience :: End Users/Desktop", + "License :: OSI Approved :: MIT License", + "Operating System :: MacOS :: MacOS X", + "Programming Language :: Python :: 3", + "Topic :: Utilities", +] +dependencies = [] + +[project.urls] +Homepage = "https://github.com/ingjieye/apple-notes-cli" +Issues = "https://github.com/ingjieye/apple-notes-cli/issues" + +[project.scripts] +apple-notes = "apple_notes_cli.cli:main" + +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[tool.hatch.build.targets.wheel] +packages = ["src/apple_notes_cli"] diff --git a/src/apple_notes_cli/__init__.py b/src/apple_notes_cli/__init__.py new file mode 100644 index 0000000..e2f520a --- /dev/null +++ b/src/apple_notes_cli/__init__.py @@ -0,0 +1,3 @@ +"""Read Apple Notes directly from its local Core Data store.""" + +__version__ = "0.1.0" diff --git a/src/apple_notes_cli/__main__.py b/src/apple_notes_cli/__main__.py new file mode 100644 index 0000000..bfdcd0c --- /dev/null +++ b/src/apple_notes_cli/__main__.py @@ -0,0 +1,4 @@ +from .cli import main + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/apple_notes_cli/cli.py b/src/apple_notes_cli/cli.py new file mode 100644 index 0000000..f79fd6e --- /dev/null +++ b/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()) diff --git a/src/apple_notes_cli/export.py b/src/apple_notes_cli/export.py new file mode 100644 index 0000000..46d54fb --- /dev/null +++ b/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 + + diff --git a/src/apple_notes_cli/models.py b/src/apple_notes_cli/models.py new file mode 100644 index 0000000..b397614 --- /dev/null +++ b/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 + diff --git a/src/apple_notes_cli/parser.py b/src/apple_notes_cli/parser.py new file mode 100644 index 0000000..8a38cbe --- /dev/null +++ b/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"{result}" + if run.superscript == 1: + result = f"{result}" + elif run.superscript == -1: + result = f"{result}" + 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 "" in line or "" 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 " " * 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,}", "

", value) + return value.replace("\n", "
").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) + + diff --git a/src/apple_notes_cli/search.py b/src/apple_notes_cli/search.py new file mode 100644 index 0000000..6e52a6a --- /dev/null +++ b/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()) diff --git a/src/apple_notes_cli/store.py b/src/apple_notes_cli/store.py new file mode 100644 index 0000000..9fbd973 --- /dev/null +++ b/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" + + diff --git a/src/apple_notes_cli/utils.py b/src/apple_notes_cli/utils.py new file mode 100644 index 0000000..85fb40e --- /dev/null +++ b/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 + + diff --git a/tests/test_parity.py b/tests/test_parity.py new file mode 100644 index 0000000..6eaa6a9 --- /dev/null +++ b/tests/test_parity.py @@ -0,0 +1,541 @@ +from __future__ import annotations + +import gzip +import tempfile +import types +import unittest +import datetime as dt +from pathlib import Path + + +from apple_notes_cli import export, models, parser, store, utils + +# These tests were written against the original single-file exporter. The code +# is now split across modules, so this flattens the public names back into one +# namespace and the assertions below stay byte-for-byte identical -- which is +# exactly what makes them useful as a regression check on the split. +export_apple_notes = types.SimpleNamespace() +for _module in (models, parser, utils, store, export): + for _name in dir(_module): + if not _name.startswith("_"): + setattr(export_apple_notes, _name, getattr(_module, _name)) + + +def varint(value: int) -> bytes: + out = bytearray() + while True: + byte = value & 0x7F + value >>= 7 + if value: + out.append(byte | 0x80) + else: + out.append(byte) + return bytes(out) + + +def field(number: int, wire_type: int, payload: bytes | int) -> bytes: + key = varint((number << 3) | wire_type) + if wire_type == 0: + assert isinstance(payload, int) + return key + varint(payload) + assert isinstance(payload, bytes) + return key + varint(len(payload)) + payload + + +def note_payload(text: str, attribute_runs: list[bytes] | None = None) -> bytes: + content = field(2, 2, text.encode("utf-8")) + for attribute_run in attribute_runs or []: + content += field(5, 2, attribute_run) + document = field(1, 0, 0) + field(2, 0, 0) + field(3, 2, content) + top = field(1, 0, 0) + field(2, 2, document) + return gzip.compress(top) + + +def checklist_attribute_run(length: int, done: int, indent: int = 0) -> bytes: + checklist = field(1, 2, b"uuid") + field(2, 0, done) + paragraph_style = field(1, 0, export_apple_notes.STYLE_TYPE_CHECKBOX) + if indent: + paragraph_style += field(4, 0, indent) + paragraph_style += field(5, 2, checklist) + return field(1, 0, length) + field(2, 2, paragraph_style) + + +def paragraph_attribute_run( + length: int, + style_type: int | None, + indent: int = 0, + block_quote: bool = False, +) -> bytes: + paragraph_style = b"" + if style_type is not None: + paragraph_style += field(1, 0, style_type) + if indent: + paragraph_style += field(4, 0, indent) + if block_quote: + paragraph_style += field(8, 0, export_apple_notes.STYLE_TYPE_BLOCK_QUOTE) + return field(1, 0, length) + field(2, 2, paragraph_style) + + +def inline_attribute_run( + length: int, + *, + font_weight: int | None = None, + underlined: int | None = None, + strikethrough: int | None = None, + superscript: int | None = None, + link: str | None = None, +) -> bytes: + payload = field(1, 0, length) + if font_weight is not None: + payload += field(5, 0, font_weight) + if underlined is not None: + payload += field(6, 0, underlined) + if strikethrough is not None: + payload += field(7, 0, strikethrough) + if superscript is not None: + payload += field(8, 0, superscript) + if link is not None: + payload += field(9, 2, link.encode("utf-8")) + return payload + + +class AppleNotesExportTests(unittest.TestCase): + def test_extract_note_text_from_gzipped_notestore_proto(self): + payload = note_payload("Title\n\nbody line") + + self.assertEqual(export_apple_notes.extract_note_text(payload), "Title\n\nbody line") + + def test_extract_note_markdown_renders_checklists(self): + payload = note_payload( + "\nTask\nDone\nNested\n", + [ + checklist_attribute_run(1, 0), + checklist_attribute_run(5, 0), + checklist_attribute_run(5, 1), + checklist_attribute_run(7, 0, indent=1), + ], + ) + + self.assertEqual( + export_apple_notes.extract_note_markdown(payload), + "- [ ] Task\n- [x] Done\n - [ ] Nested", + ) + + def test_extract_note_markdown_renders_block_styles(self): + payload = note_payload( + "\nTitle\nHeading\nSub\nCode\nQuote\nBullet\nOne\nTwo\n", + [ + paragraph_attribute_run(1, export_apple_notes.STYLE_TYPE_TITLE), + paragraph_attribute_run(6, export_apple_notes.STYLE_TYPE_TITLE), + paragraph_attribute_run(8, export_apple_notes.STYLE_TYPE_HEADING), + paragraph_attribute_run(4, export_apple_notes.STYLE_TYPE_SUBHEADING), + paragraph_attribute_run(5, export_apple_notes.STYLE_TYPE_MONOSPACED), + paragraph_attribute_run(6, None, indent=1, block_quote=True), + paragraph_attribute_run(7, export_apple_notes.STYLE_TYPE_DOTTED_LIST), + paragraph_attribute_run(4, export_apple_notes.STYLE_TYPE_NUMBERED_LIST), + paragraph_attribute_run(4, export_apple_notes.STYLE_TYPE_NUMBERED_LIST), + ], + ) + + self.assertEqual( + export_apple_notes.extract_note_markdown(payload), + "### Title\n#### Heading\n##### Sub\n Code\n> Quote\n- Bullet\n1. One\n2. Two", + ) + + def test_extract_note_markdown_does_not_wrap_trailing_space_in_bold(self): + payload = note_payload( + "产量： 约占全球咖啡产量", + [ + inline_attribute_run(4, font_weight=export_apple_notes.FONT_TYPE_BOLD), + ], + ) + + self.assertEqual( + export_apple_notes.extract_note_markdown(payload), + "**产量：** 约占全球咖啡产量", + ) + + def test_extract_note_markdown_adds_space_after_bold_label(self): + payload = note_payload( + "产量：约占全球咖啡产量", + [ + inline_attribute_run(3, font_weight=export_apple_notes.FONT_TYPE_BOLD), + ], + ) + + self.assertEqual( + export_apple_notes.extract_note_markdown(payload), + "**产量：** 约占全球咖啡产量", + ) + + def test_extract_note_markdown_preserves_tab_indented_plain_blocks(self): + payload = note_payload("RC Account\n\t0:N WABA\n\t\t1:N Phone") + + self.assertEqual( + export_apple_notes.extract_note_markdown(payload), + "```\nRC Account\n\t0:N WABA\n\t\t1:N Phone\n```", + ) + + def test_extract_note_markdown_preserves_arrow_flow_blocks(self): + payload = note_payload("Meta\n↓\nTIG -> Verification\n↓\nNECA") + + self.assertEqual( + export_apple_notes.extract_note_markdown(payload), + "```\nMeta\n↓\nTIG -> Verification\n↓\nNECA\n```", + ) + + def test_extract_note_markdown_preserves_plain_line_breaks(self): + payload = note_payload("SW: https://example.com\nBUS: test/test\nLogInsight: http://log") + + self.assertEqual( + export_apple_notes.extract_note_markdown(payload), + "SW: https://example.com \nBUS: test/test \nLogInsight: http://log", + ) + + def test_extract_note_markdown_does_not_fence_mixed_markdown_blocks(self): + payload = note_payload("Heading\n[https://example.com](https://example.com)\n\tIndented") + + self.assertEqual( + export_apple_notes.extract_note_markdown(payload), + "Heading \n[https://example.com](https://example.com) \n Indented", + ) + + def test_extract_note_markdown_merges_leading_plain_style_fragment(self): + payload = note_payload( + "主要产地： 巴西", + [ + paragraph_attribute_run(1, None), + inline_attribute_run(4, font_weight=export_apple_notes.FONT_TYPE_BOLD), + ], + ) + + self.assertEqual( + export_apple_notes.extract_note_markdown(payload), + "**主要产地：** 巴西", + ) + + def test_extract_note_markdown_uses_dominant_block_style(self): + payload = note_payload( + "产量：约", + [ + paragraph_attribute_run(1, export_apple_notes.STYLE_TYPE_HEADING), + paragraph_attribute_run(4, export_apple_notes.STYLE_TYPE_DOTTED_LIST), + ], + ) + + self.assertEqual(export_apple_notes.extract_note_markdown(payload), "- 产量：约") + + def test_extract_note_markdown_preserves_numbering_across_nested_bullets(self): + payload = note_payload( + "南美洲\n巴西豆\n中美洲\n哥斯达\n非洲\n埃塞", + [ + paragraph_attribute_run(1, export_apple_notes.STYLE_TYPE_HEADING), + paragraph_attribute_run(3, export_apple_notes.STYLE_TYPE_NUMBERED_LIST), + paragraph_attribute_run(1, export_apple_notes.STYLE_TYPE_NUMBERED_LIST), + paragraph_attribute_run(3, export_apple_notes.STYLE_TYPE_DOTTED_LIST, indent=1), + paragraph_attribute_run(1, export_apple_notes.STYLE_TYPE_DOTTED_LIST, indent=1), + paragraph_attribute_run(3, export_apple_notes.STYLE_TYPE_NUMBERED_LIST), + paragraph_attribute_run(1, export_apple_notes.STYLE_TYPE_NUMBERED_LIST), + paragraph_attribute_run(2, export_apple_notes.STYLE_TYPE_DOTTED_LIST, indent=1), + paragraph_attribute_run(1, export_apple_notes.STYLE_TYPE_DOTTED_LIST, indent=1), + paragraph_attribute_run(2, export_apple_notes.STYLE_TYPE_NUMBERED_LIST), + paragraph_attribute_run(1, export_apple_notes.STYLE_TYPE_NUMBERED_LIST), + paragraph_attribute_run(2, export_apple_notes.STYLE_TYPE_DOTTED_LIST, indent=1), + ], + ) + + self.assertEqual( + export_apple_notes.extract_note_markdown(payload), + "1. 南美洲\n - 巴西豆\n2. 中美洲\n - 哥斯达\n3. 非洲\n - 埃塞", + ) + + def test_extract_note_markdown_renders_inline_styles(self): + payload = note_payload( + "BoldItalicBothUnderStrikeSupLink", + [ + inline_attribute_run(4, font_weight=export_apple_notes.FONT_TYPE_BOLD), + inline_attribute_run(6, font_weight=export_apple_notes.FONT_TYPE_ITALIC), + inline_attribute_run(4, font_weight=export_apple_notes.FONT_TYPE_BOLD_ITALIC), + inline_attribute_run(5, underlined=1), + inline_attribute_run(6, strikethrough=1), + inline_attribute_run(3, superscript=1), + inline_attribute_run(4, link="https://example.com"), + ], + ) + + self.assertEqual( + export_apple_notes.extract_note_markdown(payload), + "**Bold***Italic****Both***Under~~Strike~~Sup[Link](https://example.com)", + ) + + def test_markdown_table_escapes_cells(self): + table = export_apple_notes.markdown_table([["A|B", "C"], ["one\ntwo", "three"]]) + + self.assertEqual(table, "| A\\|B | C |\n| --- | --- |\n| one
two | three |") + + def test_replace_table_placeholders_in_order(self): + tables = [ + export_apple_notes.EmbeddedTable("T1", "| A |\n| --- |\n| 1 |"), + export_apple_notes.EmbeddedTable("T2", "| B |\n| --- |\n| 2 |"), + ] + + body = export_apple_notes.replace_table_placeholders("before\n\ufffc\nmiddle\n\ufffc\nafter", tables) + + self.assertIn("| A |\n| --- |\n| 1 |", body) + self.assertIn("| B |\n| --- |\n| 2 |", body) + + def test_extract_links_matches_urls_before_chinese_punctuation(self): + body = "https://wiki.example.com/a 账号\nhttps://git.example.com/x?ref_type=heads，/ai" + + self.assertEqual( + export_apple_notes.extract_links(body), + [ + "https://wiki.example.com/a", + "https://git.example.com/x?ref_type=heads", + ], + ) + + def test_sanitize_filename_matches_alto_style_for_url_like_titles(self): + title = "https://platform.example.com/external-channels" + + self.assertEqual( + export_apple_notes.sanitize_filename(title), + "https---platform.example.com-external-channels", + ) + + def test_build_markdown_frontmatter_shape(self): + note = export_apple_notes.Note( + pk=42, + title="Example", + apple_notes_id="ABC-123", + folder_path="Notes", + created="2026-06-10T00:00:00+00:00", + modified="2026-06-10T01:00:00+00:00", + body="Example\n\nbody", + attachments=[], + links=["https://example.com"], + ) + + markdown = export_apple_notes.build_markdown(note) + + self.assertTrue(markdown.startswith("---\n")) + self.assertIn("title: \"Example\"\n", markdown) + self.assertIn("id: 42\n", markdown) + self.assertIn("apple_notes_id: \"ABC-123\"\n", markdown) + self.assertIn("attachments: []\n", markdown) + self.assertIn("source: notes://showNote?identifier=ABC-123\n", markdown) + self.assertTrue(markdown.endswith("Example\n\nbody\n")) + + def test_note_signature_changes_when_body_changes(self): + note = export_apple_notes.Note( + pk=1, + title="Example", + apple_notes_id="A", + folder_path="Notes", + created=None, + modified=None, + body="old", + attachments=[], + links=[], + ) + changed = export_apple_notes.dataclasses.replace(note, body="new") + + self.assertNotEqual(export_apple_notes.note_signature(note), export_apple_notes.note_signature(changed)) + + def test_manifest_roundtrip_and_safe_delete(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + note_file = root / "Notes" / "Example-1.md" + attachment_file = root / "Notes" / "image.png" + outside_file = root.parent / "outside-apple-notes-export-test.txt" + note_file.parent.mkdir() + note_file.write_text("note", encoding="utf-8") + attachment_file.write_text("attachment", encoding="utf-8") + outside_file.write_text("outside", encoding="utf-8") + self.addCleanup(lambda: outside_file.unlink(missing_ok=True)) + + entry = { + "path": "Notes/Example-1.md", + "attachments": ["Notes/image.png", "../outside-apple-notes-export-test.txt"], + } + + deleted = export_apple_notes.remove_manifest_entry_paths(root, entry) + + self.assertEqual(deleted, 2) + self.assertFalse(note_file.exists()) + self.assertFalse(attachment_file.exists()) + self.assertTrue(outside_file.exists()) + + def test_write_and_load_manifest(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + export_apple_notes.write_manifest(root, {"1": {"path": "Notes/Example-1.md"}}) + + manifest = export_apple_notes.load_manifest(root) + + self.assertEqual(manifest["version"], export_apple_notes.MANIFEST_VERSION) + self.assertEqual(manifest["notes"]["1"]["path"], "Notes/Example-1.md") + + def test_prune_empty_dirs_removes_ds_store_only_directories(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + old_dir = root / "Old Folder" / "Nested" + old_dir.mkdir(parents=True) + (old_dir / ".DS_Store").write_text("metadata", encoding="utf-8") + + export_apple_notes.prune_all_empty_dirs(root) + + self.assertFalse((root / "Old Folder").exists()) + + def test_prune_empty_dirs_keeps_directories_with_regular_files(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + kept_dir = root / "Old Folder" + kept_dir.mkdir() + (kept_dir / ".DS_Store").write_text("metadata", encoding="utf-8") + (kept_dir / "manual.md").write_text("manual", encoding="utf-8") + + export_apple_notes.prune_all_empty_dirs(root) + + self.assertTrue(kept_dir.exists()) + self.assertTrue((kept_dir / "manual.md").exists()) + + def test_remove_orphaned_export_files_deletes_stale_note_and_attachment(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + old_dir = root / "10 - 工作" / "12 - RC" + old_dir.mkdir(parents=True) + stale_note = old_dir / "Example-1.md" + stale_attachment = old_dir / "image.png" + stale_note.write_text( + "\n".join([ + "---", + 'title: "Example"', + "id: 1", + 'apple_notes_id: "A"', + 'folder: "10 - 工作/12 - RC"', + "attachments:", + ' - filename: "image.png"', + ' identifier: "ATTACHMENT"', + ' type: "public.png"', + "links: []", + "source: notes://showNote?identifier=A", + "---", + "", + "body", + "", + ]), + encoding="utf-8", + ) + stale_attachment.write_text("attachment", encoding="utf-8") + + deleted = export_apple_notes.remove_orphaned_export_files(root, set()) + + self.assertEqual(deleted, 2) + self.assertFalse((root / "10 - 工作").exists()) + + def test_remove_orphaned_export_files_keeps_manual_markdown(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + old_dir = root / "10 - 工作" + old_dir.mkdir() + manual = old_dir / "manual.md" + manual.write_text("# manual\n", encoding="utf-8") + + deleted = export_apple_notes.remove_orphaned_export_files(root, set()) + + self.assertEqual(deleted, 0) + self.assertTrue(manual.exists()) + + def test_write_note_applies_modified_time_to_markdown(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + modified = "2026-06-10T01:00:00+00:00" + note = export_apple_notes.Note( + pk=7, + title="Timestamped", + apple_notes_id="T", + folder_path="Notes", + created="2026-06-09T01:00:00+00:00", + modified=modified, + body="body", + attachments=[], + links=[], + ) + + result = export_apple_notes.write_note(root, note, copy_attachments=True) + + expected = dt.datetime.fromisoformat(modified).timestamp() + self.assertAlmostEqual(result.note_path.stat().st_mtime, expected, delta=1.0) + + def test_manifest_entry_timestamps_match_detects_stale_mtime(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + note_path = root / "Notes" / "Example-1.md" + note_path.parent.mkdir() + note_path.write_text("note", encoding="utf-8") + modified = "2026-06-10T01:00:00+00:00" + + self.assertFalse( + export_apple_notes.manifest_entry_timestamps_match( + root, + {"path": "Notes/Example-1.md", "attachments": [], "modified": modified}, + ) + ) + + export_apple_notes.apply_file_timestamps(note_path, created=None, modified=modified) + + self.assertTrue( + export_apple_notes.manifest_entry_timestamps_match( + root, + {"path": "Notes/Example-1.md", "attachments": [], "modified": modified}, + ) + ) + + def test_attachment_references_are_nested_under_note_id(self): + attachment = export_apple_notes.Attachment( + identifier="ATTACHMENT", + type_uti="public.png", + source_path=None, + output_name="image.png", + ) + + nested = export_apple_notes.nest_attachments_under_note_id([attachment], 1) + body = export_apple_notes.replace_attachment_placeholders("before\n\ufffc\nafter", nested) + + self.assertEqual(nested[0].output_name, "attachments/1/image.png") + self.assertIn("![image.png](attachments/1/image.png)", body) + + def test_write_note_copies_attachments_to_same_named_folder(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + source = root / "source.png" + source.write_text("attachment", encoding="utf-8") + attachment = export_apple_notes.Attachment( + identifier="ATTACHMENT", + type_uti="public.png", + source_path=source, + output_name="attachments/7/image.png", + ) + note = export_apple_notes.Note( + pk=7, + title="Timestamped", + apple_notes_id="T", + folder_path="Notes", + created=None, + modified=None, + body="![image.png](attachments/7/image.png)", + attachments=[attachment], + links=[], + ) + + result = export_apple_notes.write_note(root, note, copy_attachments=True) + markdown = result.note_path.read_text(encoding="utf-8") + + self.assertTrue((root / "Notes" / "Timestamped-7.md").exists()) + self.assertTrue((root / "Notes" / "attachments" / "7" / "image.png").exists()) + self.assertIn('filename: "attachments/7/image.png"', markdown) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_search.py b/tests/test_search.py new file mode 100644 index 0000000..898914e --- /dev/null +++ b/tests/test_search.py @@ -0,0 +1,251 @@ +"""Tests for the query layer, built on a synthetic Core Data store.""" + +from __future__ import annotations + +import datetime as dt +import gzip +import sqlite3 +import tempfile +import unittest +from pathlib import Path + +from apple_notes_cli import search as search_mod +from apple_notes_cli.store import trashed_folder_pks + +APPLE_EPOCH_OFFSET = 978307200.0 + + +def varint(value: int) -> bytes: + out = bytearray() + while True: + byte = value & 0x7F + value >>= 7 + if value: + out.append(byte | 0x80) + else: + out.append(byte) + return bytes(out) + + +def field(number: int, wire_type: int, payload: bytes | int) -> bytes: + key = varint((number << 3) | wire_type) + if wire_type == 0: + return key + varint(payload) + return key + varint(len(payload)) + payload + + +def note_payload(text: str) -> bytes: + content = field(2, 2, text.encode("utf-8")) + document = field(1, 0, 0) + field(2, 0, 0) + field(3, 2, content) + return gzip.compress(field(1, 0, 0) + field(2, 2, document)) + + +def apple_time(moment: dt.datetime) -> float: + return moment.timestamp() - APPLE_EPOCH_OFFSET + + +SCHEMA = """ +CREATE TABLE ZICCLOUDSYNCINGOBJECT ( + Z_PK INTEGER PRIMARY KEY, + ZTITLE1 VARCHAR, + ZTITLE2 VARCHAR, + ZFOLDER INTEGER, + ZPARENT INTEGER, + ZFOLDERTYPE INTEGER, + ZMARKEDFORDELETION INTEGER, + ZISPASSWORDPROTECTED INTEGER, + ZIDENTIFIER VARCHAR, + ZCREATIONDATE TIMESTAMP, + ZCREATIONDATE1 TIMESTAMP, + ZCREATIONDATE2 TIMESTAMP, + ZCREATIONDATE3 TIMESTAMP, + ZMODIFICATIONDATE1 TIMESTAMP, + ZNOTEDATA INTEGER +); +CREATE TABLE ZICNOTEDATA ( + Z_PK INTEGER PRIMARY KEY, + ZNOTE INTEGER, + ZDATA BLOB +); +""" + + +class QueryLayerTests(unittest.TestCase): + def setUp(self) -> None: + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + self.db_path = Path(self.tmp.name) / "NoteStore.sqlite" + self.conn = sqlite3.connect(self.db_path) + self.conn.row_factory = sqlite3.Row + self.conn.executescript(SCHEMA) + + now = dt.datetime.now(dt.timezone.utc) + # Folders: 10 = Work, 11 = Work/Deep, 20 = trash, 21 = a folder nested + # inside the trash (Notes does not create these itself, but the parent + # walk has to survive one). + self._folder(10, "Work", None, 0) + self._folder(11, "Deep", 10, 0) + self._folder(20, "Recently Deleted", None, 1) + self._folder(21, "Nested", 20, 0) + + self._note(1, "Alpha", 10, "telemetry is here", now - dt.timedelta(hours=1)) + self._note(2, "Beta", 11, "telemetry telemetry twice", now - dt.timedelta(days=5)) + self._note(3, "Gamma", 10, "nothing to see", now - dt.timedelta(minutes=10)) + self._note(4, "Trashed", 20, "telemetry in the bin", now - dt.timedelta(minutes=5)) + self._note(5, "NestedTrash", 21, "telemetry nested", now - dt.timedelta(minutes=5)) + self._note(6, "Locked", 10, "telemetry locked", now, protected=1) + self._note(7, "Deleted", 10, "telemetry deleted", now, marked=1) + self.conn.commit() + + def _folder(self, pk: int, title: str, parent: int | None, folder_type: int) -> None: + self.conn.execute( + "INSERT INTO ZICCLOUDSYNCINGOBJECT (Z_PK, ZTITLE2, ZPARENT, ZFOLDERTYPE)" + " VALUES (?, ?, ?, ?)", + (pk, title, parent, folder_type), + ) + + def _note( + self, + pk: int, + title: str, + folder: int, + body: str, + modified: dt.datetime, + *, + protected: int = 0, + marked: int = 0, + ) -> None: + note_pk = 1000 + pk + self.conn.execute( + "INSERT INTO ZICCLOUDSYNCINGOBJECT (Z_PK, ZTITLE1, ZFOLDER, ZMARKEDFORDELETION," + " ZISPASSWORDPROTECTED, ZIDENTIFIER, ZCREATIONDATE1, ZMODIFICATIONDATE1, ZNOTEDATA)" + " VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", + ( + note_pk, + title, + folder, + marked, + protected, + f"uuid-{pk}", + apple_time(modified), + apple_time(modified), + pk, + ), + ) + self.conn.execute( + "INSERT INTO ZICNOTEDATA (Z_PK, ZNOTE, ZDATA) VALUES (?, ?, ?)", + (pk, note_pk, note_payload(body)), + ) + + # --- trash handling ------------------------------------------------- + + def test_trashed_folder_pks_includes_nested_folders(self): + self.assertEqual(trashed_folder_pks(self.conn), {20, 21}) + + def test_search_excludes_trashed_notes_by_default(self): + titles = {hit.note.title for hit in search_mod.search(self.conn, "telemetry")} + self.assertEqual(titles, {"Alpha", "Beta"}) + + def test_search_can_opt_into_trashed_notes(self): + titles = { + hit.note.title + for hit in search_mod.search(self.conn, "telemetry", include_trashed=True) + } + self.assertEqual(titles, {"Alpha", "Beta", "Trashed", "NestedTrash"}) + + def test_recent_excludes_trashed_notes(self): + self.assertEqual( + [n.title for n in search_mod.recent(self.conn)], ["Gamma", "Alpha", "Beta"] + ) + + def test_locked_and_deleted_notes_are_excluded(self): + titles = { + hit.note.title + for hit in search_mod.search(self.conn, "telemetry", include_trashed=True) + } + self.assertNotIn("Locked", titles) + self.assertNotIn("Deleted", titles) + + # --- search behaviour ----------------------------------------------- + + def test_hits_rank_by_match_count(self): + hits = search_mod.search(self.conn, "telemetry") + self.assertEqual([h.note.title for h in hits], ["Beta", "Alpha"]) + self.assertEqual([h.match_count for h in hits], [2, 1]) + + def test_title_is_searched_unless_body_only(self): + self.assertEqual(len(search_mod.search(self.conn, "Gamma")), 1) + self.assertEqual( + len(search_mod.search(self.conn, "Gamma", search_titles=False)), 0 + ) + + def test_regex_search(self): + hits = search_mod.search(self.conn, r"tele\w+try", regex=True) + self.assertEqual({h.note.title for h in hits}, {"Alpha", "Beta"}) + + def test_plain_query_is_not_treated_as_regex(self): + self.assertEqual(search_mod.search(self.conn, "tele.metry"), []) + + def test_case_sensitivity(self): + self.assertEqual(len(search_mod.search(self.conn, "TELEMETRY")), 2) + self.assertEqual( + len(search_mod.search(self.conn, "TELEMETRY", case_sensitive=True)), 0 + ) + + def test_since_filters_older_notes(self): + since = dt.datetime.now(dt.timezone.utc) - dt.timedelta(days=1) + titles = {h.note.title for h in search_mod.search(self.conn, "telemetry", since=since)} + self.assertEqual(titles, {"Alpha"}) + + def test_folder_filter_matches_nested_paths(self): + hits = search_mod.search(self.conn, "telemetry", folder="Work/Deep") + self.assertEqual([h.note.title for h in hits], ["Beta"]) + + def test_snippets_carry_context_and_respect_cap(self): + hit = search_mod.search(self.conn, "telemetry", max_snippets=1)[0] + self.assertEqual(len(hit.snippets), 1) + self.assertIn("telemetry", hit.snippets[0]) + + # --- folders and show ------------------------------------------------ + + def test_folders_reports_counts_without_trash(self): + self.assertEqual( + search_mod.folders(self.conn), [("Work", 3), ("Work/Deep", 1)] + ) + + def test_show_returns_body_and_metadata(self): + note, body = search_mod.show(self.conn, 1001) + self.assertEqual(note.title, "Alpha") + self.assertEqual(note.folder, "Work") + self.assertIn("telemetry", body) + + def test_show_raises_for_unknown_id(self): + with self.assertRaises(KeyError): + search_mod.show(self.conn, 999999) + + +class ParseSinceTests(unittest.TestCase): + def test_relative_offsets(self): + now = dt.datetime.now(dt.timezone.utc) + self.assertAlmostEqual( + (now - search_mod.parse_since("2d")).total_seconds(), 172800, delta=5 + ) + self.assertAlmostEqual( + (now - search_mod.parse_since("30m")).total_seconds(), 1800, delta=5 + ) + self.assertAlmostEqual( + (now - search_mod.parse_since("1w")).total_seconds(), 604800, delta=5 + ) + + def test_iso_date_is_localized(self): + parsed = search_mod.parse_since("2026-09-01") + self.assertEqual((parsed.year, parsed.month, parsed.day), (2026, 9, 1)) + self.assertIsNotNone(parsed.tzinfo) + + def test_invalid_value_raises_value_error(self): + with self.assertRaises(ValueError): + search_mod.parse_since("last tuesday") + + +if __name__ == "__main__": + unittest.main()