ourdigital/our-claude-skills
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(notion-writer): --engine routing + two-phase image append

Browse Source 
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Andrew Yim
2026-06-27 09:59:18 +09:00
parent aeba10cea4
commit b14701d92c
2 changed files with 140 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

140
custom-skills/32-notion-writer/code/scripts/notion_writer.py
View File

@@ -17,6 +17,10 @@ from notion_client import Client
from notion_client.errors import APIResponseError
import _notion_compat as compat
import ntn_files
import md_translate
MARKDOWN_NOTION_VERSION = "2026-03-11"
# Load environment variables
load_dotenv(Path(__file__).parent / '.env')
@@ -528,6 +532,27 @@ def create_image_block(alt: str, target: str) -> Dict[str, Any]:
return block
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
def create_table_block(header_cells: List[str], body_rows: List[List[str]]) -> Dict[str, Any]:
"""Build a Notion `table` block with header + body rows.
@@ -817,6 +842,14 @@ Examples:
parser.add_argument('--list', '-l', nargs='?', const='all', choices=['all', 'pages', 'databases'],
help='List accessible pages and/or databases (default: all)')
parser.add_argument('--info', action='store_true', help='Show page/database info')
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')
args = parser.parse_args()
@@ -906,23 +939,61 @@ Examples:
sys.exit(1)
content = file_path.read_text(encoding='utf-8')
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)
# Write to page
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)
mode = 'replace' if args.replace else 'append'
print(f"{'Replacing' if mode == 'replace' else 'Appending'} content to page...")
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
if write_to_page(notion, page_id, content, mode):
print(f"✅ Successfully wrote content to page")
formatted_id = format_id_with_dashes(page_id)
# 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")
@@ -981,6 +1052,8 @@ Examples:
content_blocks = markdown_to_notion_blocks(content) if content else None
existing = None
# Upsert path: look for existing row by the named property
if args.upsert_by:
if args.upsert_by not in schema_props:
@@ -1001,19 +1074,50 @@ Examples:
print(f"Error during upsert lookup: {compat.explain_api_error(exc)}")
sys.exit(1)
if existing:
page_id = existing['id']
print(f"Updating existing row (matched on {args.upsert_by}={lookup_value!r})...")
if not update_page_properties(notion, page_id, properties):
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:
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
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)
if existing:
page_id = existing['id']
print(f"Updating existing row (matched on {args.upsert_by}={lookup_value!r})...")
if not update_page_properties(notion, page_id, properties):
sys.exit(1)
if content_blocks:
if not clear_page_content(notion, page_id):
sys.exit(1)
if content_blocks:
if not clear_page_content(notion, page_id):
sys.exit(1)
if not append_to_page(notion, page_id, content_blocks):
sys.exit(1)
print(f"✅ Successfully updated database row")
print(f" https://notion.so/{page_id.replace('-', '')}")
return
if not append_to_page(notion, page_id, content_blocks):
sys.exit(1)
print(f"✅ Successfully updated database row")
print(f" https://notion.so/{page_id.replace('-', '')}")
return
print(f"Creating database row...")
row_id = create_database_row(notion, data_source_id, properties, content_blocks)

18
custom-skills/32-notion-writer/code/scripts/test_engine_routing.py
View File

@@ -8,6 +8,7 @@ from unittest import mock
sys.path.insert(0, str(Path(__file__).parent))
import _notion_compat as compat
import notion_writer
def _fake_client():
@@ -48,11 +49,28 @@ def test_replace_markdown():
assert kwargs["body"]["replace_content"]["allow_deleting_content"] is True
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
def run_all():
tests = [
test_create_page_markdown,
test_append_markdown,
test_replace_markdown,
test_extract_local_images_splits,
test_content_has_local_images,
]
for t in tests:
print(f"\n{t.__name__}")