feat(notion-writer): materialize local images via upload walk

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

custom-skills/32-notion-writer/code/scripts/ntn_files.py

@@ -83,3 +83,54 @@ def upload(path: Path) -> str:
     first_line = result.stdout.strip().splitlines()[0]
     upload_id = first_line.split("\t")[0].strip()
     return upload_id
+
+
+_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

custom-skills/32-notion-writer/code/scripts/test_ntn_files.py

@@ -71,12 +71,61 @@ def test_upload_failure_raises():
             assert "boom" in e.stderr
 
 
+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
+
+
 def run_all():
     tests = [
         test_preflight_missing_ntn,
         test_preflight_returns_workspace,
         test_upload_returns_id,
         test_upload_failure_raises,
+        test_materialize_remote_untouched,
+        test_materialize_local_uploaded,
+        test_materialize_nested_children,
+        test_materialize_missing_file_raises,
     ]
     for t in tests:
         print(f"\n{t.__name__}")