Files
our-claude-skills/docs/superpowers/plans/2026-06-27-notion-writer-cli-enhancements.md
Andrew Yim bfade2b722 docs(notion-writer): fix plan — circular import, columns loop, error hint
Pre-flight review fixes: lazy import in md_translate to break the
notion_writer<->md_translate cycle; clean columns state machine; markdown
deletion hint in explain_api_error.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-27 11:12:48 +09:00

1332 lines
50 KiB
Markdown
Raw Blame History

This file contains invisible Unicode characters
This file contains invisible Unicode characters that are indistinguishable to humans but may be processed differently by a computer. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# Notion Writer CLI Enhancements Implementation Plan
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
**Goal:** Add local file/image uploads (via the `ntn` CLI) and an opt-in native-markdown write engine to the `notion-writer` skill, without regressing the existing block-converter path.
**Architecture:** Keep the token-based `notion-client` core. Add two focused, pure-where-possible modules — `ntn_files.py` (subprocess upload) and `md_translate.py` (dialect→Notion-markdown). The block parser stays pure; uploads happen in an isolated post-parse walk. A `--engine` flag selects between the existing block converter (default) and the new markdown endpoints.
**Tech Stack:** Python 3.12, `notion-client>=2.0.0`, `python-dotenv`, stdlib `subprocess`/`shutil`/`unittest.mock`, the `ntn` CLI (v0.18.0, already installed + logged in), Notion API `2025-09-03` (blocks) / `2026-03-11` (markdown).
## Global Constraints
- All work happens on branch `notion-writer-cli-enhancements` in repo `~/Project/our-claude-skills`.
- Skill code dir (all relative paths below are under it): `custom-skills/32-notion-writer/code/scripts/`.
- Tests use the existing custom harness (plain `assert` functions + a `run_all()` at bottom), run with `venv/bin/python test_<name>.py`. **No pytest.**
- The existing `test_parser.py` has **32 tests** and must stay green after every task. (The spec's "28" is a stale count; the real number is 32.)
- Default engine is `blocks` — existing CLI calls must behave identically. Markdown is opt-in via `--engine markdown`.
- Markdown *write* calls use a client built with `notion_version="2026-03-11"`. Schema/property/upsert-lookup calls stay on the default-version client.
- Parser purity is preserved: `markdown_to_notion_blocks` / `_parse_lines` make no I/O calls. Uploads live only in `ntn_files.py` and the `materialize_local_media` walk.
- `client.request(path=..., method=..., body=...)` paths omit the `v1/` prefix (the SDK adds it): use `"pages"` and `f"pages/{page_id}/markdown"`.
- Commit after every task. Use `git add <explicit files>` (never `git add -A`). End commit messages with the trailer:
`Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>`
---
## File Structure
| File | Status | Responsibility |
|---|---|---|
| `ntn_files.py` | create | `preflight()`, `upload(path)`, `materialize_local_media(blocks, base_dir, upload_fn)`, `NtnUploadError` |
| `md_translate.py` | create | `translate(content)` + per-construct translators (callout/columns/toggle/mention) |
| `notion_writer.py` | modify | `IMAGE_RE`, `create_image_block`, `extract_local_images`, `_content_has_local_images`, CLI flags, engine routing, two-phase image append |
| `_notion_compat.py` | modify | `make_client(api_key, notion_version=None)`, `create_page_markdown`, `append_markdown`, `replace_markdown`, markdown cases in `explain_api_error` |
| `test_ntn_files.py` | create | unit tests for `ntn_files` (subprocess + walk mocked) |
| `test_md_translate.py` | create | unit tests for the translator |
| `test_engine_routing.py` | create | unit tests for compat markdown helpers + CLI routing (client mocked) |
| `test_parser.py` | modify | image-block tests |
| `requirements.txt` | modify | (no dep change expected; touched only if needed) |
| repo `.gitignore` | modify | ignore the skill's `venv/` |
| `code/CLAUDE.md`, `SKILL.md` | modify | document engines, uploads, flags; v1.3.0 |
---
### Task 1: Environment baseline (venv + green existing tests)
**Files:**
- Create: `custom-skills/32-notion-writer/code/scripts/venv/` (not committed)
- Modify: repo `.gitignore`
**Interfaces:**
- Consumes: nothing.
- Produces: a working `venv/bin/python` with deps installed; baseline of 32 green tests.
- [ ] **Step 1: Create the venv**
Run:
```bash
cd ~/Project/our-claude-skills/custom-skills/32-notion-writer/code/scripts
python3 -m venv venv
```
- [ ] **Step 2: Install dependencies**
Run:
```bash
venv/bin/python -m pip install -r requirements.txt
```
Expected: installs `python-dotenv` and `notion-client` (and transitive deps) with no error.
- [ ] **Step 3: Run the existing test suite (baseline)**
Run:
```bash
venv/bin/python test_parser.py
```
Expected: ends with `✅ All 32 tests passed`.
- [ ] **Step 4: Ignore the venv in git**
Append to the repo-root `.gitignore` (`~/Project/our-claude-skills/.gitignore`); create the file if missing. Add this line:
```
custom-skills/32-notion-writer/code/scripts/venv/
```
Verify it is ignored:
```bash
cd ~/Project/our-claude-skills && git status --short custom-skills/32-notion-writer/code/scripts/
```
Expected: no `venv/` entries appear.
- [ ] **Step 5: Commit**
```bash
cd ~/Project/our-claude-skills
git add .gitignore
git commit -m "$(printf 'chore(notion-writer): ignore scripts venv\n\nCo-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>')"
```
---
### Task 2: `ntn_files.py` — preflight + upload
**Files:**
- Create: `custom-skills/32-notion-writer/code/scripts/ntn_files.py`
- Test: `custom-skills/32-notion-writer/code/scripts/test_ntn_files.py`
**Interfaces:**
- Consumes: nothing (stdlib only).
- Produces:
- `class NtnUploadError(Exception)` — attributes `.path` (str), `.stderr` (str).
- `preflight() -> dict` — returns `{"workspace_name": str, "workspace_id": str}`; raises `NtnUploadError` if `ntn` is missing or not logged in. Cached after first success.
- `upload(path) -> str` — uploads a local file via `ntn files create --plain`, returns the `file_upload_id`. Raises `NtnUploadError` on failure.
- [ ] **Step 1: Write the failing tests**
Create `test_ntn_files.py`:
```python
#!/usr/bin/env python3
"""Tests for ntn_files.py — run with `python test_ntn_files.py`."""
import sys
import subprocess
from pathlib import Path
from unittest import mock
sys.path.insert(0, str(Path(__file__).parent))
import ntn_files
from ntn_files import NtnUploadError
def _reset_cache():
ntn_files._WORKSPACE = None
def test_preflight_missing_ntn():
_reset_cache()
with mock.patch("shutil.which", return_value=None):
try:
ntn_files.preflight()
assert False, "expected NtnUploadError"
except NtnUploadError as e:
assert "ntn" in str(e).lower()
def test_preflight_returns_workspace():
_reset_cache()
fake = subprocess.CompletedProcess(
args=[], returncode=0,
stdout='{"bot":{"workspace_name":"D.Intelligence",'
'"workspace_id":"ws-123"}}',
stderr="",
)
with mock.patch("shutil.which", return_value="/usr/bin/ntn"), \
mock.patch("subprocess.run", return_value=fake):
info = ntn_files.preflight()
assert info["workspace_name"] == "D.Intelligence"
assert info["workspace_id"] == "ws-123"
def test_upload_returns_id():
_reset_cache()
fake = subprocess.CompletedProcess(
args=[], returncode=0,
stdout="43833259-72ae-404e-8441-b6577f3159b4\tphoto.png\tuploaded\n",
stderr="",
)
with mock.patch("subprocess.run", return_value=fake):
upload_id = ntn_files.upload(Path("/tmp/photo.png"))
assert upload_id == "43833259-72ae-404e-8441-b6577f3159b4"
def test_upload_failure_raises():
_reset_cache()
fake = subprocess.CompletedProcess(
args=[], returncode=1, stdout="", stderr="boom: invalid file",
)
with mock.patch("subprocess.run", return_value=fake):
try:
ntn_files.upload(Path("/tmp/bad.png"))
assert False, "expected NtnUploadError"
except NtnUploadError as e:
assert e.path == "/tmp/bad.png"
assert "boom" in e.stderr
def run_all():
tests = [
test_preflight_missing_ntn,
test_preflight_returns_workspace,
test_upload_returns_id,
test_upload_failure_raises,
]
for t in tests:
print(f"\n{t.__name__}")
t()
print("\n" + "=" * 50)
print(f"✅ All {len(tests)} tests passed")
print("=" * 50)
if __name__ == "__main__":
run_all()
```
- [ ] **Step 2: Run tests to verify they fail**
Run: `venv/bin/python test_ntn_files.py`
Expected: FAIL with `ModuleNotFoundError: No module named 'ntn_files'`.
- [ ] **Step 3: Write `ntn_files.py`**
Create `ntn_files.py`:
```python
#!/usr/bin/env python3
"""Local file uploads to Notion via the `ntn` CLI.
Owns all subprocess I/O for the skill. The CLI handles the full File Upload
lifecycle (create -> send bytes -> complete) and prints the upload ID.
"""
from __future__ import annotations
import json
import shutil
import subprocess
import sys
from pathlib import Path
from typing import Dict, List, Optional, Callable, Any
_WORKSPACE: Optional[Dict[str, str]] = None
class NtnUploadError(Exception):
"""Raised when `ntn` is unavailable or a file upload fails."""
def __init__(self, message: str, path: str = "", stderr: str = ""):
super().__init__(message)
self.path = path
self.stderr = stderr
def preflight() -> Dict[str, str]:
"""Verify `ntn` is installed and logged in; return its target workspace.
Cached for the process. Raises NtnUploadError with an actionable hint
on failure. Prints one informational line naming the workspace `ntn`
targets (file uploads are workspace-scoped).
"""
global _WORKSPACE
if _WORKSPACE is not None:
return _WORKSPACE
if shutil.which("ntn") is None:
raise NtnUploadError(
"ntn CLI not found. Install it with: "
"curl -fsSL https://ntn.dev | bash"
)
result = subprocess.run(
["ntn", "api", "v1/users/me"],
capture_output=True, text=True,
)
if result.returncode != 0:
raise NtnUploadError(
"ntn is not logged in. Run: ntn login\n" + result.stderr.strip()
)
try:
me = json.loads(result.stdout)
bot = me.get("bot", {})
info = {
"workspace_name": bot.get("workspace_name", "unknown"),
"workspace_id": bot.get("workspace_id", "unknown"),
}
except (ValueError, AttributeError):
info = {"workspace_name": "unknown", "workspace_id": "unknown"}
print(f'ntn -> workspace "{info["workspace_name"]}"', file=sys.stderr)
_WORKSPACE = info
return info
def upload(path: Path) -> str:
"""Upload a local file via `ntn files create --plain`; return its id."""
path = Path(path)
with open(path, "rb") as fh:
result = subprocess.run(
["ntn", "files", "create", "--plain"],
stdin=fh, capture_output=True, text=True,
)
if result.returncode != 0:
raise NtnUploadError(
f"ntn upload failed for {path}", path=str(path),
stderr=result.stderr.strip(),
)
first_line = result.stdout.strip().splitlines()[0]
upload_id = first_line.split("\t")[0].strip()
return upload_id
```
- [ ] **Step 4: Run tests to verify they pass**
Run: `venv/bin/python test_ntn_files.py`
Expected: `✅ All 4 tests passed`.
- [ ] **Step 5: Commit**
```bash
cd ~/Project/our-claude-skills
git add custom-skills/32-notion-writer/code/scripts/ntn_files.py \
custom-skills/32-notion-writer/code/scripts/test_ntn_files.py
git commit -m "$(printf 'feat(notion-writer): ntn_files upload + preflight\n\nCo-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>')"
```
---
### Task 3: Image blocks in the parser (pure)
**Files:**
- Modify: `custom-skills/32-notion-writer/code/scripts/notion_writer.py`
- Test: `custom-skills/32-notion-writer/code/scripts/test_parser.py`
**Interfaces:**
- Consumes: `parse_rich_text` (existing).
- Produces:
- `IMAGE_RE` — compiled regex matching a standalone image line.
- `create_image_block(alt: str, target: str) -> dict` — always external shape, `caption` = `parse_rich_text(alt)`.
- `_parse_lines` emits an image block for standalone `![alt](target)` lines.
- [ ] **Step 1: Write the failing tests**
Add these test functions to `test_parser.py` (above `run_all`), and add their names to the `tests` list inside `run_all`:
```python
def test_image_remote_external():
blocks = markdown_to_notion_blocks("![a chart](https://ex.com/c.png)")
assert len(blocks) == 1
b = blocks[0]
assert b["type"] == "image"
assert b["image"]["type"] == "external"
assert b["image"]["external"]["url"] == "https://ex.com/c.png"
assert b["image"]["caption"][0]["text"]["content"] == "a chart"
def test_image_local_external_shape_preupload():
blocks = markdown_to_notion_blocks("![local](./pics/x.png)")
assert len(blocks) == 1
assert blocks[0]["image"]["external"]["url"] == "./pics/x.png"
def test_image_only_when_standalone():
# An inline bang-bracket inside prose is NOT an image block.
blocks = markdown_to_notion_blocks("see ![x](y.png) inline")
assert blocks[0]["type"] == "paragraph"
```
Add to the `tests` list in `run_all`:
```python
test_image_remote_external,
test_image_local_external_shape_preupload,
test_image_only_when_standalone,
```
- [ ] **Step 2: Run tests to verify they fail**
Run: `venv/bin/python test_parser.py`
Expected: FAIL — `test_image_remote_external` errors because the first block is a `paragraph` (or the `!` leaks), not an `image`.
- [ ] **Step 3: Add `IMAGE_RE` and `create_image_block`**
In `notion_writer.py`, add the constant next to `TABLE_SEPARATOR_RE` (near line 58):
```python
IMAGE_RE = re.compile(r'^\s*!\[([^\]]*)\]\(([^)\s]+)\)\s*$')
```
Add the factory alongside the other `create_*_block` helpers (e.g., after `create_divider_block`):
```python
def create_image_block(alt: str, target: str) -> Dict[str, Any]:
"""Image block in external shape. Local targets are converted to
file_upload shape later by ntn_files.materialize_local_media."""
block: Dict[str, Any] = {
"object": "block",
"type": "image",
"image": {"type": "external", "external": {"url": target}},
}
if alt:
block["image"]["caption"] = parse_rich_text(alt)
return block
```
- [ ] **Step 4: Add the image detector to `_parse_lines`**
In `_parse_lines`, add this branch **before** the table detector (before the `if _is_table_row(line)...` line near line 213), so a standalone image line is caught before other block detectors:
```python
image_match = IMAGE_RE.match(line)
if image_match:
blocks.append(create_image_block(
image_match.group(1), image_match.group(2)))
i += 1
continue
```
- [ ] **Step 5: Run tests to verify they pass**
Run: `venv/bin/python test_parser.py`
Expected: `✅ All 35 tests passed` (32 existing + 3 new).
- [ ] **Step 6: Commit**
```bash
cd ~/Project/our-claude-skills
git add custom-skills/32-notion-writer/code/scripts/notion_writer.py \
custom-skills/32-notion-writer/code/scripts/test_parser.py
git commit -m "$(printf 'feat(notion-writer): parse standalone images to image blocks\n\nCo-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>')"
```
---
### Task 4: `materialize_local_media` walk
**Files:**
- Modify: `custom-skills/32-notion-writer/code/scripts/ntn_files.py`
- Test: `custom-skills/32-notion-writer/code/scripts/test_ntn_files.py`
**Interfaces:**
- Consumes: `create_image_block` output shape (`block["image"]["external"]["url"]`).
- Produces:
- `materialize_local_media(blocks, base_dir, upload_fn=upload) -> list` — recursively walks blocks + nested `children`. For each `image` block whose `external.url` is not `http(s):` and resolves (relative to `base_dir`, or absolute) to an existing file, calls `upload_fn(resolved_path)` and rewrites the block to file_upload shape. Missing local file raises `NtnUploadError`. Remote/external untouched. `upload_fn` is injectable for testing.
- [ ] **Step 1: Write the failing tests**
Add to `test_ntn_files.py` (and register in `run_all`):
```python
def _img(url):
return {"object": "block", "type": "image",
"image": {"type": "external", "external": {"url": url}}}
def test_materialize_remote_untouched():
blocks = [_img("https://ex.com/a.png")]
out = ntn_files.materialize_local_media(
blocks, Path("/tmp"), upload_fn=lambda p: "SHOULD_NOT_RUN")
assert out[0]["image"]["type"] == "external"
def test_materialize_local_uploaded(tmp_path=None):
import tempfile, os
d = Path(tempfile.mkdtemp())
(d / "x.png").write_bytes(b"fake")
blocks = [_img("x.png")]
out = ntn_files.materialize_local_media(
blocks, d, upload_fn=lambda p: "UP123")
assert out[0]["image"]["type"] == "file_upload"
assert out[0]["image"]["file_upload"]["id"] == "UP123"
def test_materialize_nested_children():
import tempfile
d = Path(tempfile.mkdtemp())
(d / "y.png").write_bytes(b"fake")
toggle = {"object": "block", "type": "toggle",
"toggle": {"rich_text": [], "children": [_img("y.png")]}}
out = ntn_files.materialize_local_media(
[toggle], d, upload_fn=lambda p: "UPNESTED")
child = out[0]["toggle"]["children"][0]
assert child["image"]["file_upload"]["id"] == "UPNESTED"
def test_materialize_missing_file_raises():
blocks = [_img("nope.png")]
try:
ntn_files.materialize_local_media(
blocks, Path("/tmp"), upload_fn=lambda p: "x")
assert False, "expected NtnUploadError"
except NtnUploadError as e:
assert "nope.png" in str(e) or "nope.png" in e.path
```
Register in `run_all`'s `tests` list:
```python
test_materialize_remote_untouched,
test_materialize_local_uploaded,
test_materialize_nested_children,
test_materialize_missing_file_raises,
```
- [ ] **Step 2: Run tests to verify they fail**
Run: `venv/bin/python test_ntn_files.py`
Expected: FAIL — `AttributeError: module 'ntn_files' has no attribute 'materialize_local_media'`.
- [ ] **Step 3: Implement the walk**
Append to `ntn_files.py`:
```python
_REMOTE_PREFIXES = ("http://", "https://")
def _resolve_local(url: str, base_dir: Path) -> Optional[Path]:
"""Return the resolved path for a local media url, or None if remote."""
if url.startswith(_REMOTE_PREFIXES):
return None
p = Path(url)
return p if p.is_absolute() else Path(base_dir) / p
def materialize_local_media(
blocks: List[Dict[str, Any]],
base_dir: Path,
upload_fn: Callable[[Path], str] = upload,
) -> List[Dict[str, Any]]:
"""Walk blocks (and nested children); upload local image files and
rewrite them to file_upload shape. Remote images are left as-is."""
for block in blocks:
if block.get("type") == "image":
img = block["image"]
if img.get("type") == "external":
url = img.get("external", {}).get("url", "")
resolved = _resolve_local(url, base_dir)
if resolved is not None:
if not resolved.is_file():
raise NtnUploadError(
f"image references missing file: {url}",
path=str(resolved),
)
upload_id = upload_fn(resolved)
caption = img.get("caption")
new_img: Dict[str, Any] = {
"type": "file_upload",
"file_upload": {"id": upload_id},
}
if caption:
new_img["caption"] = caption
block["image"] = new_img
# Recurse into any nested children list.
btype = block.get("type")
nested = block.get(btype, {}) if isinstance(block.get(btype), dict) else {}
if isinstance(nested.get("children"), list):
materialize_local_media(nested["children"], base_dir, upload_fn)
if btype == "column_list":
for col in nested.get("children", []):
col_children = col.get("column", {}).get("children", [])
materialize_local_media(col_children, base_dir, upload_fn)
return blocks
```
- [ ] **Step 4: Run tests to verify they pass**
Run: `venv/bin/python test_ntn_files.py`
Expected: `✅ All 8 tests passed`.
- [ ] **Step 5: Commit**
```bash
cd ~/Project/our-claude-skills
git add custom-skills/32-notion-writer/code/scripts/ntn_files.py \
custom-skills/32-notion-writer/code/scripts/test_ntn_files.py
git commit -m "$(printf 'feat(notion-writer): materialize local images via upload walk\n\nCo-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>')"
```
---
### Task 5: `md_translate.py` — dialect translator
**Files:**
- Create: `custom-skills/32-notion-writer/code/scripts/md_translate.py`
- Test: `custom-skills/32-notion-writer/code/scripts/test_md_translate.py`
**Interfaces:**
- Consumes: `notion_writer.extract_notion_id` (existing) for mention resolution.
- Produces: `translate(content: str) -> str` — converts skill dialect to Notion enhanced markdown. Pure.
Translation rules (everything not listed passes through unchanged): GitHub alert `> [!TYPE]` block → `<callout icon=... color="..._bg">` with tab-indented body; Pandoc `::: columns/column/:::``<columns>/<column>` with tab-indented children; `<details>`/`<summary>` toggle → children re-indented one tab; inline `@[Title](id|url)``<mention-page url="...">Title</mention-page>` (unresolvable → plain `@Title`).
- [ ] **Step 1: Write the failing tests**
Create `test_md_translate.py`:
```python
#!/usr/bin/env python3
"""Tests for md_translate.py — run with `python test_md_translate.py`."""
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent))
import md_translate
def test_passthrough_basics():
src = "# Heading\n\n- item\n\n```python\nx=1\n```"
assert md_translate.translate(src) == src
def test_callout_note():
out = md_translate.translate("> [!NOTE]\n> Be careful")
assert '<callout icon="" color="blue_bg">' in out
assert "\tBe careful" in out
assert "</callout>" in out
def test_callout_warning_color():
out = md_translate.translate("> [!WARNING]\n> Danger")
assert 'color="yellow_bg"' in out
assert 'icon="⚠️"' in out
def test_columns():
src = "::: columns\n::: column\nLeft\n:::\n::: column\nRight\n:::\n:::"
out = md_translate.translate(src)
assert "<columns>" in out
assert out.count("<column>") == 2
assert "</columns>" in out
assert "\tLeft" in out or "\t\tLeft" in out
def test_toggle_children_indented():
src = "<details>\n<summary>More</summary>\nBody line\n</details>"
out = md_translate.translate(src)
assert "<summary>More</summary>" in out
assert "\tBody line" in out
def test_mention_url():
out = md_translate.translate(
"See @[ADR](https://notion.so/X-abcdef0123456789abcdef0123456789).")
assert "<mention-page url=" in out
assert ">ADR</mention-page>" in out
def test_mention_invalid_plain():
out = md_translate.translate("ping @[Bob](not-an-id)")
assert "@Bob" in out
assert "mention-page" not in out
def run_all():
tests = [
test_passthrough_basics,
test_callout_note,
test_callout_warning_color,
test_columns,
test_toggle_children_indented,
test_mention_url,
test_mention_invalid_plain,
]
for t in tests:
print(f"\n{t.__name__}")
t()
print("\n" + "=" * 50)
print(f"✅ All {len(tests)} tests passed")
print("=" * 50)
if __name__ == "__main__":
run_all()
```
- [ ] **Step 2: Run tests to verify they fail**
Run: `venv/bin/python test_md_translate.py`
Expected: FAIL with `ModuleNotFoundError: No module named 'md_translate'`.
- [ ] **Step 3: Implement `md_translate.py`**
Create `md_translate.py`:
```python
#!/usr/bin/env python3
"""Translate the notion-writer markdown dialect into Notion enhanced
markdown for the markdown write engine. Pure functions, no I/O."""
from __future__ import annotations
import re
from typing import List
# NOTE: `notion_writer` is imported lazily inside _translate_mentions to avoid
# a circular import (notion_writer imports this module at its top level).
# Skill alert type -> (emoji, Notion enhanced-markdown background color)
CALLOUT_MAP = {
"NOTE": ("", "blue_bg"),
"TIP": ("💡", "green_bg"),
"IMPORTANT": ("☝️", "purple_bg"),
"WARNING": ("⚠️", "yellow_bg"),
"CAUTION": ("🚨", "red_bg"),
}
_ALERT_RE = re.compile(r'^\s*\[!(NOTE|TIP|IMPORTANT|WARNING|CAUTION)\]\s*$')
_MENTION_RE = re.compile(r'@\[([^\]]+)\]\(([^)\s]+)\)')
def _translate_mentions(text: str) -> str:
from notion_writer import extract_notion_id, format_id_with_dashes # lazy: breaks import cycle
def repl(m: "re.Match") -> str:
title, target = m.group(1), m.group(2)
page_id = extract_notion_id(target)
if page_id:
return (f'<mention-page url="{format_id_with_dashes(page_id)}">'
f'{title}</mention-page>')
return f"@{title}"
return _MENTION_RE.sub(repl, text)
def _indent(lines: List[str]) -> List[str]:
return ["\t" + ln if ln.strip() else ln for ln in lines]
def translate(content: str) -> str:
lines = content.split("\n")
out: List[str] = []
i = 0
while i < len(lines):
line = lines[i]
# Callout: > [!TYPE] then contiguous > body lines
if line.lstrip().startswith(">"):
body_first = line.lstrip()[1:].strip()
alert = _ALERT_RE.match(body_first)
if alert:
emoji, color = CALLOUT_MAP[alert.group(1)]
i += 1
body: List[str] = []
while i < len(lines) and lines[i].lstrip().startswith(">"):
body.append(lines[i].lstrip()[1:].lstrip())
i += 1
out.append(f'<callout icon="{emoji}" color="{color}">')
out.extend(_indent([_translate_mentions(b) for b in body]))
out.append("</callout>")
continue
# Columns: ::: columns / ::: column / :::
# State machine: "::: column" opens a column; ":::" closes a column
# when one is open, otherwise closes the wrapper. The Pandoc layout
# emits N "::: column" opens, N ":::" column-closes, then one final
# ":::" wrapper-close.
if line.strip() == "::: columns":
i += 1
cols: List[List[str]] = []
cur: List[str] = None
while i < len(lines):
s = lines[i].strip()
if s == "::: column":
if cur is not None:
cols.append(cur)
cur = []
i += 1
elif s == ":::":
if cur is not None: # close the open column
cols.append(cur)
cur = None
i += 1
else: # close the wrapper
i += 1
break
else:
if cur is not None:
cur.append(lines[i])
i += 1
out.append("<columns>")
for col in cols:
out.append("\t<column>")
out.extend(_indent(_indent(
[_translate_mentions(c) for c in col])))
out.append("\t</column>")
out.append("</columns>")
continue
# Toggle: <details><summary>..</summary> body </details>
if line.strip() == "<details>":
out.append("<details>")
i += 1
if i < len(lines) and lines[i].lstrip().startswith("<summary>"):
out.append(lines[i].strip())
i += 1
body = []
while i < len(lines) and lines[i].strip() != "</details>":
body.append(_translate_mentions(lines[i]))
i += 1
out.extend(_indent(body))
out.append("</details>")
if i < len(lines): # skip closing </details>
i += 1
continue
out.append(_translate_mentions(line))
i += 1
return "\n".join(out)
```
- [ ] **Step 4: Run tests to verify they pass**
Run: `venv/bin/python test_md_translate.py`
Expected: `✅ All 7 tests passed`.
- [ ] **Step 5: Confirm the parser suite still passes (no import cycle)**
Run: `venv/bin/python test_parser.py`
Expected: `✅ All 35 tests passed` (importing `md_translate` must not break `notion_writer`).
- [ ] **Step 6: Commit**
```bash
cd ~/Project/our-claude-skills
git add custom-skills/32-notion-writer/code/scripts/md_translate.py \
custom-skills/32-notion-writer/code/scripts/test_md_translate.py
git commit -m "$(printf 'feat(notion-writer): dialect->Notion enhanced markdown translator\n\nCo-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>')"
```
---
### Task 6: `_notion_compat.py` markdown helpers + version-aware client
**Files:**
- Modify: `custom-skills/32-notion-writer/code/scripts/_notion_compat.py`
- Test: `custom-skills/32-notion-writer/code/scripts/test_engine_routing.py`
**Interfaces:**
- Consumes: `notion_client.Client.request(path, method, body)`.
- Produces:
- `make_client(api_key, notion_version=None) -> Client` (adds optional version).
- `create_page_markdown(client, parent, properties, markdown) -> dict``request("pages", "POST", body={parent, properties, markdown})`.
- `append_markdown(client, page_id, markdown) -> dict``request(f"pages/{page_id}/markdown", "PATCH", body={"type":"insert_content","insert_content":{"content":markdown,"position":{"type":"end"}}})`.
- `replace_markdown(client, page_id, markdown, allow_deleting=False) -> dict``request(f"pages/{page_id}/markdown", "PATCH", body={"type":"replace_content","replace_content":{"new_str":markdown,"allow_deleting_content":allow_deleting}})`.
- [ ] **Step 1: Write the failing tests**
Create `test_engine_routing.py`:
```python
#!/usr/bin/env python3
"""Tests for markdown transport helpers — run with `python test_engine_routing.py`."""
import sys
from pathlib import Path
from unittest import mock
sys.path.insert(0, str(Path(__file__).parent))
import _notion_compat as compat
def _fake_client():
c = mock.MagicMock()
c.request.return_value = {"object": "page", "id": "p1"}
return c
def test_create_page_markdown():
c = _fake_client()
parent = {"type": "data_source_id", "data_source_id": "ds1"}
compat.create_page_markdown(c, parent, {"Name": {"title": []}}, "# Hi")
_, kwargs = c.request.call_args
assert kwargs["path"] == "pages"
assert kwargs["method"] == "POST"
assert kwargs["body"]["markdown"] == "# Hi"
assert kwargs["body"]["parent"] == parent
def test_append_markdown():
c = _fake_client()
compat.append_markdown(c, "page-123", "## More")
_, kwargs = c.request.call_args
assert kwargs["path"] == "pages/page-123/markdown"
assert kwargs["method"] == "PATCH"
assert kwargs["body"]["type"] == "insert_content"
assert kwargs["body"]["insert_content"]["content"] == "## More"
assert kwargs["body"]["insert_content"]["position"] == {"type": "end"}
def test_replace_markdown():
c = _fake_client()
compat.replace_markdown(c, "page-123", "# Fresh", allow_deleting=True)
_, kwargs = c.request.call_args
assert kwargs["path"] == "pages/page-123/markdown"
assert kwargs["body"]["type"] == "replace_content"
assert kwargs["body"]["replace_content"]["new_str"] == "# Fresh"
assert kwargs["body"]["replace_content"]["allow_deleting_content"] is True
def run_all():
tests = [
test_create_page_markdown,
test_append_markdown,
test_replace_markdown,
]
for t in tests:
print(f"\n{t.__name__}")
t()
print("\n" + "=" * 50)
print(f"✅ All {len(tests)} tests passed")
print("=" * 50)
if __name__ == "__main__":
run_all()
```
- [ ] **Step 2: Run tests to verify they fail**
Run: `venv/bin/python test_engine_routing.py`
Expected: FAIL — `AttributeError: module '_notion_compat' has no attribute 'create_page_markdown'`.
- [ ] **Step 3: Implement the helpers**
In `_notion_compat.py`, replace `make_client` (lines 19-21) with:
```python
def make_client(api_key: str, notion_version: str = None) -> Client:
"""Build a sync Notion client. Pass notion_version to override the SDK
default (needed: 2026-03-11 for the markdown content endpoints)."""
if notion_version:
return Client(auth=api_key, notion_version=notion_version)
return Client(auth=api_key)
```
Add at the end of `_notion_compat.py`:
```python
def create_page_markdown(client, parent, properties, markdown):
"""POST /v1/pages with the enhanced-markdown body param."""
return client.request(
path="pages", method="POST",
body={"parent": parent, "properties": properties, "markdown": markdown},
)
def append_markdown(client, page_id, markdown):
"""PATCH /v1/pages/:id/markdown — append at end (insert_content)."""
return client.request(
path=f"pages/{page_id}/markdown", method="PATCH",
body={"type": "insert_content",
"insert_content": {"content": markdown,
"position": {"type": "end"}}},
)
def replace_markdown(client, page_id, markdown, allow_deleting=False):
"""PATCH /v1/pages/:id/markdown — replace all content."""
return client.request(
path=f"pages/{page_id}/markdown", method="PATCH",
body={"type": "replace_content",
"replace_content": {"new_str": markdown,
"allow_deleting_content": allow_deleting}},
)
```
Also extend the `ValidationError` branch of `explain_api_error` (currently lines ~212-213) so markdown-replace deletion errors are actionable. Replace:
```python
if code == APIErrorCode.ValidationError:
return f"Validation error{suffix}: {exc.body.get('message', str(exc))}"
```
with:
```python
if code == APIErrorCode.ValidationError:
msg = exc.body.get('message', str(exc))
if 'delet' in msg.lower():
msg += " — re-run with --allow-deleting-content to permit this."
return f"Validation error{suffix}: {msg}"
```
- [ ] **Step 4: Run tests to verify they pass**
Run: `venv/bin/python test_engine_routing.py`
Expected: `✅ All 3 tests passed`.
- [ ] **Step 5: Confirm no regression**
Run: `venv/bin/python test_parser.py`
Expected: `✅ All 35 tests passed`.
- [ ] **Step 6: Commit**
```bash
cd ~/Project/our-claude-skills
git add custom-skills/32-notion-writer/code/scripts/_notion_compat.py \
custom-skills/32-notion-writer/code/scripts/test_engine_routing.py
git commit -m "$(printf 'feat(notion-writer): markdown endpoint helpers + version-aware client\n\nCo-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>')"
```
---
### Task 7: Wire the CLI — flags, routing, two-phase image append
**Files:**
- Modify: `custom-skills/32-notion-writer/code/scripts/notion_writer.py`
- Test: `custom-skills/32-notion-writer/code/scripts/test_engine_routing.py`
**Interfaces:**
- Consumes: `ntn_files.preflight/upload/materialize_local_media`, `md_translate.translate`, `compat.create_page_markdown/append_markdown/replace_markdown`, `compat.make_client(..., notion_version=...)`.
- Produces:
- `extract_local_images(content: str) -> tuple[str, list[tuple[str, str]]]` — removes standalone local (`!http`) image lines, returns `(content_without_local_images, [(alt, target), ...])`.
- `_content_has_local_images(content: str) -> bool`.
- `--engine`, `--notion-version`, `--allow-deleting-content` argparse flags.
- `MARKDOWN_NOTION_VERSION = "2026-03-11"` constant.
- [ ] **Step 1: Write the failing tests**
Add to `test_engine_routing.py` (register names in `run_all`). These import from `notion_writer` and exercise the pure helpers:
```python
import notion_writer
def test_extract_local_images_splits():
content = ("# Title\n\n![remote](https://ex.com/a.png)\n\n"
"![local](./b.png)\n\nbody")
without, imgs = notion_writer.extract_local_images(content)
assert imgs == [("local", "./b.png")]
assert "![local](./b.png)" not in without
assert "![remote](https://ex.com/a.png)" in without # remote stays
def test_content_has_local_images():
assert notion_writer._content_has_local_images("![x](./y.png)") is True
assert notion_writer._content_has_local_images("![x](https://e/y.png)") is False
assert notion_writer._content_has_local_images("no images") is False
```
Add to `run_all`'s `tests` list:
```python
test_extract_local_images_splits,
test_content_has_local_images,
```
- [ ] **Step 2: Run tests to verify they fail**
Run: `venv/bin/python test_engine_routing.py`
Expected: FAIL — `AttributeError: module 'notion_writer' has no attribute 'extract_local_images'`.
- [ ] **Step 3: Add imports + helpers + constant**
Near the top of `notion_writer.py`, after `import _notion_compat as compat` (line 19), add:
```python
import ntn_files
import md_translate
MARKDOWN_NOTION_VERSION = "2026-03-11"
```
Add these helpers near `create_image_block`:
```python
def _content_has_local_images(content: str) -> bool:
for line in content.split('\n'):
m = IMAGE_RE.match(line)
if m and not m.group(2).startswith(('http://', 'https://')):
return True
return False
def extract_local_images(content: str):
"""Remove standalone LOCAL image lines; keep remote ones inline.
Returns (content_without_local_images, [(alt, target), ...])."""
kept, imgs = [], []
for line in content.split('\n'):
m = IMAGE_RE.match(line)
if m and not m.group(2).startswith(('http://', 'https://')):
imgs.append((m.group(1), m.group(2)))
continue
kept.append(line)
return "\n".join(kept), imgs
```
- [ ] **Step 4: Run the new pure-helper tests**
Run: `venv/bin/python test_engine_routing.py`
Expected: `✅ All 5 tests passed`.
- [ ] **Step 5: Add the argparse flags**
In `main()`, after the `--info` argument (around line 798), add:
```python
parser.add_argument('--engine', choices=['blocks', 'markdown'],
default='blocks',
help='Write engine: blocks (default) or markdown')
parser.add_argument('--notion-version', dest='notion_version',
help='Override Notion API version')
parser.add_argument('--allow-deleting-content', dest='allow_deleting',
action='store_true',
help='Markdown replace may delete child pages/dbs')
```
- [ ] **Step 6: Add a base-dir helper + media-client wiring in `main()`**
In `main()`, right after `content` is resolved (after the `elif args.file:` block near line 886), add:
```python
base_dir = Path(args.file).parent if args.file else Path.cwd()
def _md_client():
version = args.notion_version or MARKDOWN_NOTION_VERSION
return compat.make_client(NOTION_TOKEN, notion_version=version)
def _upload_blocks_for(images):
"""Build + materialize image blocks for two-phase markdown writes."""
blocks = [create_image_block(alt, target) for alt, target in images]
return ntn_files.materialize_local_media(blocks, base_dir, ntn_files.upload)
```
- [ ] **Step 7: Route the page-write path through the engine**
Replace the page-write block in `main()` (the `if args.page:` block, lines ~889-909) with:
```python
if args.page:
if not content:
print("Error: No content provided. Use --file or --stdin")
sys.exit(1)
page_id = extract_notion_id(args.page)
if not page_id:
print(f"Error: Invalid Notion page URL/ID: {args.page}")
sys.exit(1)
formatted_id = format_id_with_dashes(page_id)
if args.engine == 'markdown':
if _content_has_local_images(content):
ntn_files.preflight()
md_body, local_imgs = extract_local_images(content)
md_body = md_translate.translate(md_body)
mc = _md_client()
try:
if args.replace:
compat.replace_markdown(mc, formatted_id, md_body,
allow_deleting=args.allow_deleting)
else:
compat.append_markdown(mc, formatted_id, md_body)
except APIResponseError as exc:
print(f"Error: {compat.explain_api_error(exc, formatted_id)}")
sys.exit(1)
if local_imgs:
append_to_page(notion, page_id, _upload_blocks_for(local_imgs))
print("✅ Successfully wrote content to page (markdown engine)")
print(f" https://notion.so/{formatted_id.replace('-', '')}")
return
# blocks engine (default)
blocks = markdown_to_notion_blocks(content)
if _content_has_local_images(content):
ntn_files.preflight()
blocks = ntn_files.materialize_local_media(blocks, base_dir, ntn_files.upload)
if not blocks:
print("No content to write")
sys.exit(1)
if args.replace and not clear_page_content(notion, page_id):
sys.exit(1)
if append_to_page(notion, page_id, blocks):
print("✅ Successfully wrote content to page")
print(f" https://notion.so/{formatted_id.replace('-', '')}")
else:
print("❌ Failed to write content")
sys.exit(1)
return
```
- [ ] **Step 8: Route the database-row path through the engine**
In the `if args.database:` block, after `content_blocks = markdown_to_notion_blocks(content) if content else None` (line ~961), insert the markdown-engine branch and guard the blocks-engine upload. Replace that line and the create/upsert tail with:
```python
if args.engine == 'markdown':
md_content = content or ""
if _content_has_local_images(md_content):
ntn_files.preflight()
md_body, local_imgs = extract_local_images(md_content)
md_body = md_translate.translate(md_body)
parent = compat.build_data_source_parent(data_source_id)
mc = _md_client()
try:
if args.upsert_by and 'existing' in dir() and existing:
compat.replace_markdown(mc, existing['id'], md_body,
allow_deleting=args.allow_deleting)
update_page_properties(notion, existing['id'], properties)
new_id = existing['id']
else:
result = compat.create_page_markdown(mc, parent, properties, md_body)
new_id = result['id']
except APIResponseError as exc:
print(f"Error: {compat.explain_api_error(exc)}")
sys.exit(1)
if local_imgs:
append_to_page(notion, new_id, _upload_blocks_for(local_imgs))
print("✅ Successfully wrote database row (markdown engine)")
print(f" https://notion.so/{new_id.replace('-', '')}")
return
content_blocks = markdown_to_notion_blocks(content) if content else None
if content_blocks and _content_has_local_images(content or ""):
ntn_files.preflight()
content_blocks = ntn_files.materialize_local_media(
content_blocks, base_dir, ntn_files.upload)
```
> Note: the existing upsert lookup that defines `existing` runs earlier in this block (the `if args.upsert_by:` section). Keep that section above this code so `existing` is in scope. If the markdown branch runs before that section in the current file order, move the markdown branch to just after the upsert lookup. The blocks-engine create/upsert tail below stays unchanged.
- [ ] **Step 9: Run all suites**
Run:
```bash
venv/bin/python test_parser.py && \
venv/bin/python test_ntn_files.py && \
venv/bin/python test_md_translate.py && \
venv/bin/python test_engine_routing.py
```
Expected: every suite prints its `✅ All N tests passed` line (35, 8, 7, 5).
- [ ] **Step 10: Smoke-check the CLI help (no API calls)**
Run: `venv/bin/python notion_writer.py --help`
Expected: help text lists `--engine`, `--notion-version`, `--allow-deleting-content`.
- [ ] **Step 11: Commit**
```bash
cd ~/Project/our-claude-skills
git add custom-skills/32-notion-writer/code/scripts/notion_writer.py \
custom-skills/32-notion-writer/code/scripts/test_engine_routing.py
git commit -m "$(printf 'feat(notion-writer): --engine routing + two-phase image append\n\nCo-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>')"
```
---
### Task 8: Documentation (CLAUDE.md + SKILL.md, v1.3.0)
**Files:**
- Modify: `custom-skills/32-notion-writer/code/CLAUDE.md`
- Modify: `custom-skills/32-notion-writer/SKILL.md`
**Interfaces:** none (docs only).
- [ ] **Step 1: Document the new capabilities in `code/CLAUDE.md`**
Add a "File uploads" subsection under Markdown Support describing `![alt](local.png)` auto-upload via `ntn` (requires `ntn` installed + `ntn login`; workspace-scoped). Add an "Engines" subsection: `--engine blocks` (default) vs `--engine markdown` (native endpoints, `2026-03-11`, dialect auto-translated, local images appended at page end). Document `--notion-version` and `--allow-deleting-content`. Add example:
```bash
# Markdown engine, create a row from a doc with callouts/columns
python notion_writer.py -d DB_URL -t "Notes" --engine markdown -f notes.md
# Blocks engine with a local image (auto-uploaded via ntn)
python notion_writer.py -p PAGE_URL -f post.md # post.md contains ![chart](./chart.png)
```
- [ ] **Step 2: Bump the version footer in `code/CLAUDE.md`**
Change the footer line to `*Version 1.3.0 | Claude Code | 2026-06-27*` and add a changelog entry:
```markdown
- 1.3.0 — Local file/image uploads via the `ntn` CLI (`![](local)` → file_upload image blocks). New `--engine markdown` path writing through Notion's native enhanced-markdown endpoints with a dialect translator. Added `--notion-version` and `--allow-deleting-content`.
```
- [ ] **Step 3: Mirror the summary into `SKILL.md`**
Under the "Supported Markdown" / capabilities area of `SKILL.md`, add a one-paragraph note on engines and image upload, with the `--engine markdown` example above. Keep it short — `SKILL.md` is the trigger surface.
- [ ] **Step 4: Commit**
```bash
cd ~/Project/our-claude-skills
git add custom-skills/32-notion-writer/code/CLAUDE.md \
custom-skills/32-notion-writer/SKILL.md
git commit -m "$(printf 'docs(notion-writer): document engines + file uploads (v1.3.0)\n\nCo-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>')"
```
---
### Task 9: Live smoke test (both engines)
**Files:** none (manual verification against the live API).
**Interfaces:** uses the real `NOTION_API_KEY` (1Password) and the "Working with AI" data source `f8f19ede-32bd-43ac-9f60-0651f6f40afe`.
- [ ] **Step 1: Connection + token**
Run:
```bash
cd ~/Project/our-claude-skills/custom-skills/32-notion-writer/code/scripts
NOTION_API_KEY="$(op read 'op://Development/Notion - Claude Agent/api-key')" \
venv/bin/python notion_writer.py --test
```
Expected: `✅ Connected successfully!`. If the 1Password item path differs, use the working credential path from SKILL.md's "Credential handling".
- [ ] **Step 2: Blocks engine — local image upload**
Create `/tmp/nw_img.md` with a heading and `![smoke](./logo.png)` (place any small PNG at `./logo.png` in the scripts dir). Run an upsert into the AI DB:
```bash
NOTION_API_KEY="$(op read 'op://Development/Notion - Claude Agent/api-key')" \
venv/bin/python notion_writer.py -d f8f19ede-32bd-43ac-9f60-0651f6f40afe \
-t "[smoke] blocks image" --upsert-by "Name" -f /tmp/nw_img.md
```
Expected: prints the `ntn -> workspace` line and a success URL; the row shows an embedded image (not a broken link). **Verify the `ntn` workspace line matches the token's workspace** — if not, the image won't attach (documented constraint).
- [ ] **Step 3: Markdown engine — callout + columns + appended image**
Create `/tmp/nw_md.md`:
```markdown
# Smoke: markdown engine
> [!NOTE]
> Rendered as a Notion callout.
::: columns
::: column
Left column
:::
::: column
Right column
:::
:::
![smoke](./logo.png)
```
Run:
```bash
NOTION_API_KEY="$(op read 'op://Development/Notion - Claude Agent/api-key')" \
venv/bin/python notion_writer.py -d f8f19ede-32bd-43ac-9f60-0651f6f40afe \
-t "[smoke] markdown engine" --upsert-by "Name" --engine markdown -f /tmp/nw_md.md
```
Expected: success URL; the page shows a callout, two columns, and the image at the end. **This confirms the open pipe-table/version items** — if the callout or columns render as raw text, check the translator's tab indentation; if the API rejects the body, check the `2026-03-11` version and the data-source parent.
- [ ] **Step 4: Markdown engine — replace**
Run the same command as Step 3 but with `-p <the page URL from Step 3>`, `--engine markdown`, `--replace`, and a trivial `/tmp/nw_md2.md` (`# Replaced\nNew body.`):
```bash
NOTION_API_KEY="$(op read 'op://Development/Notion - Claude Agent/api-key')" \
venv/bin/python notion_writer.py -p PAGE_URL_FROM_STEP_3 \
--engine markdown --replace -f /tmp/nw_md2.md
```
Expected: page content replaced via `replace_content`. If it errors about deleting child content, re-run with `--allow-deleting-content`.
- [ ] **Step 5: Record results**
If all four steps pass, the feature is verified end-to-end. If the pipe-table passthrough failed in Step 3, open a follow-up per the spec's parking lot (`<table>` rewrite). No commit (manual test task).
---
## Notes for the implementer
- **Run from the scripts dir.** All `venv/bin/python ...` commands assume cwd = `custom-skills/32-notion-writer/code/scripts/`.
- **Line numbers** in "Modify" hints are from the current `notion_writer.py` (1013 lines) and `_notion_compat.py` (256 lines); if they've drifted, locate by the quoted anchor code instead.
- **Don't reformat** untouched code. Keep diffs minimal and additive.
- After Task 7, the four suites total 55 tests (35 + 8 + 7 + 5). Keep them all green through Tasks 8-9.