From 80c9efa2826d297257c8c32981bfd6af72dba8da Mon Sep 17 00:00:00 2001 From: Andrew Yim Date: Sat, 27 Jun 2026 08:10:12 +0900 Subject: [PATCH] feat(notion-writer): ntn_files upload + preflight Co-Authored-By: Claude Opus 4.8 (1M context) --- .../code/scripts/ntn_files.py | 85 ++++++++++++++++++ .../code/scripts/test_ntn_files.py | 90 +++++++++++++++++++ 2 files changed, 175 insertions(+) create mode 100644 custom-skills/32-notion-writer/code/scripts/ntn_files.py create mode 100644 custom-skills/32-notion-writer/code/scripts/test_ntn_files.py diff --git a/custom-skills/32-notion-writer/code/scripts/ntn_files.py b/custom-skills/32-notion-writer/code/scripts/ntn_files.py new file mode 100644 index 0000000..bb44401 --- /dev/null +++ b/custom-skills/32-notion-writer/code/scripts/ntn_files.py @@ -0,0 +1,85 @@ +#!/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 diff --git a/custom-skills/32-notion-writer/code/scripts/test_ntn_files.py b/custom-skills/32-notion-writer/code/scripts/test_ntn_files.py new file mode 100644 index 0000000..a42f3c3 --- /dev/null +++ b/custom-skills/32-notion-writer/code/scripts/test_ntn_files.py @@ -0,0 +1,90 @@ +#!/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="", + ) + # upload() opens the file before subprocess.run; mock open so the file + # need not exist (subprocess.run is mocked and never reads the handle). + with mock.patch("builtins.open", mock.mock_open(read_data=b"x")), \ + 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("builtins.open", mock.mock_open(read_data=b"x")), \ + 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()