From 5ff9b3d9f5e2c9512ac931004aecdc0b62e60bd5 Mon Sep 17 00:00:00 2001 From: Andrew Yim Date: Sat, 27 Jun 2026 09:59:18 +0900 Subject: [PATCH] feat(notion-writer): --engine routing + two-phase image append Co-Authored-By: Claude Opus 4.8 (1M context) --- .../code/scripts/notion_writer.py | 140 +++++++++++++++--- .../code/scripts/test_engine_routing.py | 18 +++ 2 files changed, 140 insertions(+), 18 deletions(-) diff --git a/custom-skills/32-notion-writer/code/scripts/notion_writer.py b/custom-skills/32-notion-writer/code/scripts/notion_writer.py index e72b2ce..267490d 100644 --- a/custom-skills/32-notion-writer/code/scripts/notion_writer.py +++ b/custom-skills/32-notion-writer/code/scripts/notion_writer.py @@ -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) diff --git a/custom-skills/32-notion-writer/code/scripts/test_engine_routing.py b/custom-skills/32-notion-writer/code/scripts/test_engine_routing.py index 544c7f2..11c6ac8 100644 --- a/custom-skills/32-notion-writer/code/scripts/test_engine_routing.py +++ b/custom-skills/32-notion-writer/code/scripts/test_engine_routing.py @@ -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__}")