From 144a17c88d52406c1e36caaa4fa99c6353d2222b Mon Sep 17 00:00:00 2001 From: Andrew Yim Date: Mon, 27 Apr 2026 09:57:20 +0900 Subject: [PATCH] feat(notion): migrate 31+32 to API 2025-09-03 + add property writes, upsert, anchor-link fix MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase 1 — docs alignment: - Fix path drift in 32 CLAUDE.md (02-notion-writer → 32-notion-writer) - Align env var to NOTION_API_KEY in 31 docs (NOTION_TOKEN still accepted) - Document Phase 3 roadmap in 31 (metadata-aware migration, Notion-as-RAG) Phase 2 — multi-source database support: - New 32/scripts/_notion_compat.py: client factory, data_source resolution with cache, property coercion, find_existing_page (for upsert), explain_api_error for friendlier failure messages - 32 notion_writer.py: drop the 2022-06-28 pin, route schema introspection through data_sources.retrieve, build data_source_id parent shape, accept arbitrary properties via --properties JSON-or-file flag, add --upsert-by for idempotent re-runs - 32 markdown parser: anchor-link fix — Notion's URL validator rejects fragment URLs and relative paths; fragments now render as bold to preserve TOC nav intent, relatives drop the link, absolute URLs preserved - 31 async_organizer.py + schema_migrator.py: same data_sources migration with cached resolution; pages.create now uses data_source_id parent - 16/16 parser tests pass (was 13; +3 link-handling regression guards) Co-Authored-By: Claude Opus 4.7 (1M context) --- .../31-notion-organizer/code/CLAUDE.md | 31 ++- .../code/scripts/async_organizer.py | 64 ++++- .../code/scripts/schema_migrator.py | 69 ++++- custom-skills/32-notion-writer/code/CLAUDE.md | 126 ++++++++- .../code/scripts/_notion_compat.py | 255 ++++++++++++++++++ .../code/scripts/notion_writer.py | 174 +++++++++--- .../code/scripts/test_parser.py | 28 ++ 7 files changed, 678 insertions(+), 69 deletions(-) create mode 100644 custom-skills/32-notion-writer/code/scripts/_notion_compat.py diff --git a/custom-skills/31-notion-organizer/code/CLAUDE.md b/custom-skills/31-notion-organizer/code/CLAUDE.md index b5b8e20..7249819 100644 --- a/custom-skills/31-notion-organizer/code/CLAUDE.md +++ b/custom-skills/31-notion-organizer/code/CLAUDE.md @@ -79,11 +79,40 @@ python scripts/async_organizer.py --database [ID] --action restructure Environment variables: ```bash -NOTION_TOKEN=secret_xxx +# Preferred (matches 32-notion-writer) +NOTION_API_KEY=secret_xxx + +# Legacy fallback also accepted +# NOTION_TOKEN=secret_xxx ``` +The scripts read `NOTION_TOKEN` first, then fall back to `NOTION_API_KEY`. Use `NOTION_API_KEY` for new setups so the same `.env` works across all Notion skills. + ## Notes - Always use `--dry-run` first for destructive operations - Large operations (1000+ pages) use async with progress reporting - Scripts implement automatic rate limiting + +## Roadmap (Phase 3) + +The current scripts cover schema migration and async bulk ops. Two larger goals are parked for a future iteration: + +### Goal 1 — Metadata-aware page move/integration + +Search source pages, compare their property metadata against a target database schema, and move/migrate while transforming property values to fit. Beyond the existing `schema_migrator.py` (which expects a hand-written mapping file), this would: + +- Auto-suggest property mappings using name + type similarity +- Surface unmappable properties before the move (no silent data loss) +- Support cross-database moves (not just same-schema migrations) + +### Goal 2 — Notion as personal RAG source + +Treat Notion as a small, personal knowledge base for AI agents: + +- Search pages by query across databases, filter by property +- Merge results into a single context (with source citations back to page URLs) +- Summarize / distill via LLM into agent-ready snippets +- Export as JSONL or markdown for fine-tuning or RAG indexing + +This dovetails with `90-reference-curator` (which does the same for web sources) — Phase 3 would make Notion a first-class source type for that pipeline. diff --git a/custom-skills/31-notion-organizer/code/scripts/async_organizer.py b/custom-skills/31-notion-organizer/code/scripts/async_organizer.py index 0bf7437..77f9e60 100644 --- a/custom-skills/31-notion-organizer/code/scripts/async_organizer.py +++ b/custom-skills/31-notion-organizer/code/scripts/async_organizer.py @@ -55,6 +55,45 @@ class NotionAsyncOrganizer: self.semaphore = Semaphore(MAX_CONCURRENT_REQUESTS) self.dry_run = dry_run self.stats = {"fetched": 0, "updated": 0, "created": 0, "errors": 0} + self._data_source_cache: dict[str, str] = {} + + async def _resolve_data_source_id(self, target_id: str) -> str: + """ + Accept either a database_id or data_source_id; return data_source_id. + + Caches results so repeated calls within one organizer run don't re-hit + the API. Errors propagate to the caller for proper handling. + """ + if target_id in self._data_source_cache: + return self._data_source_cache[target_id] + + from notion_client.errors import APIErrorCode, APIResponseError + + try: + await self._rate_limited_request( + self.client.data_sources.retrieve(data_source_id=target_id) + ) + self._data_source_cache[target_id] = target_id + return target_id + except APIResponseError as exc: + if exc.code != APIErrorCode.ObjectNotFound: + raise + + db = await self._rate_limited_request( + self.client.databases.retrieve(database_id=target_id) + ) + sources = db.get("data_sources") or [] + if not sources: + raise ValueError(f"Database {target_id} has no data sources attached") + if len(sources) > 1: + names = ", ".join(s.get("name", s["id"]) for s in sources) + raise ValueError( + f"Database {target_id} has multiple data sources ({names}); " + "pass a specific data_source_id" + ) + ds_id = sources[0]["id"] + self._data_source_cache[target_id] = ds_id + return ds_id @retry( stop=stop_after_attempt(3), @@ -67,31 +106,33 @@ class NotionAsyncOrganizer: await asyncio.sleep(REQUEST_DELAY) return await coro - async def fetch_database_schema(self, database_id: str) -> dict: - """Fetch database schema/properties.""" - logger.info(f"Fetching database schema: {database_id}") + async def fetch_database_schema(self, target_id: str) -> dict: + """Fetch full schema (properties) for a database or data source.""" + data_source_id = await self._resolve_data_source_id(target_id) + logger.info(f"Fetching schema: {data_source_id}") response = await self._rate_limited_request( - self.client.databases.retrieve(database_id=database_id) + self.client.data_sources.retrieve(data_source_id=data_source_id) ) self.stats["fetched"] += 1 return response async def fetch_all_pages( self, - database_id: str, + target_id: str, filter_obj: dict | None = None, sorts: list | None = None, ) -> list[dict]: - """Fetch all pages from a database with pagination.""" + """Fetch all pages from a data source with pagination.""" + data_source_id = await self._resolve_data_source_id(target_id) all_pages = [] has_more = True start_cursor = None - logger.info(f"Fetching pages from database: {database_id}") + logger.info(f"Fetching pages from data source: {data_source_id}") while has_more: query_params = { - "database_id": database_id, + "data_source_id": data_source_id, "page_size": 100, } if start_cursor: @@ -102,7 +143,7 @@ class NotionAsyncOrganizer: query_params["sorts"] = sorts response = await self._rate_limited_request( - self.client.databases.query(**query_params) + self.client.data_sources.query(**query_params) ) all_pages.extend(response["results"]) @@ -287,7 +328,10 @@ async def example_bulk_status_update( async def main(): """Main entry point.""" parser = argparse.ArgumentParser(description="Notion Async Organizer") - parser.add_argument("--database-id", "-d", required=True, help="Database ID") + parser.add_argument( + "--database-id", "-d", required=True, + help="Database ID or data source ID (resolved automatically)", + ) parser.add_argument( "--dry-run", action="store_true", help="Preview changes without executing" ) diff --git a/custom-skills/31-notion-organizer/code/scripts/schema_migrator.py b/custom-skills/31-notion-organizer/code/scripts/schema_migrator.py index b17a848..0261c5e 100644 --- a/custom-skills/31-notion-organizer/code/scripts/schema_migrator.py +++ b/custom-skills/31-notion-organizer/code/scripts/schema_migrator.py @@ -53,6 +53,7 @@ class SchemaMigrator: "pages_skipped": 0, "errors": 0, } + self._data_source_cache: dict[str, str] = {} @retry( stop=stop_after_attempt(3), @@ -63,24 +64,59 @@ class SchemaMigrator: await asyncio.sleep(REQUEST_DELAY) return await coro - async def get_schema(self, database_id: str) -> dict: - """Get database schema.""" + async def _resolve_data_source_id(self, target_id: str) -> str: + """Accept database_id or data_source_id; return data_source_id.""" + if target_id in self._data_source_cache: + return self._data_source_cache[target_id] + + from notion_client.errors import APIErrorCode, APIResponseError + + try: + await self._request( + self.client.data_sources.retrieve(data_source_id=target_id) + ) + self._data_source_cache[target_id] = target_id + return target_id + except APIResponseError as exc: + if exc.code != APIErrorCode.ObjectNotFound: + raise + + db = await self._request(self.client.databases.retrieve(database_id=target_id)) + sources = db.get("data_sources") or [] + if not sources: + raise ValueError(f"Database {target_id} has no data sources attached") + if len(sources) > 1: + names = ", ".join(s.get("name", s["id"]) for s in sources) + raise ValueError( + f"Database {target_id} has multiple data sources ({names}); " + "pass a specific data_source_id" + ) + ds_id = sources[0]["id"] + self._data_source_cache[target_id] = ds_id + return ds_id + + async def get_schema(self, target_id: str) -> dict: + """Get full schema for a database or data source.""" + data_source_id = await self._resolve_data_source_id(target_id) return await self._request( - self.client.databases.retrieve(database_id=database_id) + self.client.data_sources.retrieve(data_source_id=data_source_id) ) - async def fetch_all_pages(self, database_id: str) -> list[dict]: - """Fetch all pages from source database.""" + async def fetch_all_pages(self, target_id: str) -> list[dict]: + """Fetch all pages from source data source.""" + data_source_id = await self._resolve_data_source_id(target_id) pages = [] has_more = True cursor = None while has_more: - params = {"database_id": database_id, "page_size": 100} + params = {"data_source_id": data_source_id, "page_size": 100} if cursor: params["start_cursor"] = cursor - response = await self._request(self.client.databases.query(**params)) + response = await self._request( + self.client.data_sources.query(**params) + ) pages.extend(response["results"]) has_more = response.get("has_more", False) cursor = response.get("next_cursor") @@ -235,7 +271,10 @@ class SchemaMigrator: result = await self._request( self.client.pages.create( - parent={"database_id": target_database_id}, + parent={ + "type": "data_source_id", + "data_source_id": target_database_id, + }, properties=properties, ) ) @@ -253,22 +292,26 @@ class SchemaMigrator: target_db: str, mapping: dict, ) -> list[dict]: - """Execute full migration.""" + """Execute full migration. Resolves both IDs to data_source_ids once.""" + logger.info("Resolving source and target IDs...") + source_ds = await self._resolve_data_source_id(source_db) + target_ds = await self._resolve_data_source_id(target_db) + logger.info("Fetching schemas...") - source_schema = await self.get_schema(source_db) - target_schema = await self.get_schema(target_db) + source_schema = await self.get_schema(source_ds) + target_schema = await self.get_schema(target_ds) logger.info(f"Source: {len(source_schema['properties'])} properties") logger.info(f"Target: {len(target_schema['properties'])} properties") logger.info("Fetching source pages...") - pages = await self.fetch_all_pages(source_db) + pages = await self.fetch_all_pages(source_ds) logger.info(f"Found {len(pages)} pages to migrate") results = [] for page in tqdm(pages, desc="Migrating"): result = await self.migrate_page( - page, target_db, mapping, source_schema, target_schema + page, target_ds, mapping, source_schema, target_schema ) results.append(result) diff --git a/custom-skills/32-notion-writer/code/CLAUDE.md b/custom-skills/32-notion-writer/code/CLAUDE.md index e2a22af..d1dc514 100644 --- a/custom-skills/32-notion-writer/code/CLAUDE.md +++ b/custom-skills/32-notion-writer/code/CLAUDE.md @@ -39,7 +39,7 @@ ### 3. Configure Environment ```bash -cd ~/Project/our-claude-skills/custom-skills/02-notion-writer/code/scripts +cd ~/Project/our-claude-skills/custom-skills/32-notion-writer/code/scripts # Create .env from example cp .env.example .env @@ -100,8 +100,47 @@ python notion_writer.py --database DB_URL --title "New Entry" --file content.md # Title only python notion_writer.py --database DB_URL --title "Empty Entry" + +# Set arbitrary properties (JSON literal) +python notion_writer.py \ + --database DB_URL \ + --title "Research Notes" \ + --properties '{"Status": "In progress", "Topic": ["AI", "MCP"], "Account Code": "OurDigital"}' \ + --file notes.md + +# Set properties from a JSON file +python notion_writer.py \ + --database DB_URL \ + --title "Research Notes" \ + --properties ./props.json \ + --file notes.md ``` +`--database` accepts a database URL, a database_id, or a data_source_id. It is resolved to a data_source_id automatically. + +### Upsert (avoid duplicates) + +```bash +# If a row with Name="My Title" already exists, update it; otherwise create. +python notion_writer.py \ + --database DB_URL \ + --title "My Title" \ + --upsert-by "Name" \ + --properties '{"Status": "Done"}' \ + --file content.md +``` + +When upserting, `--file` (or `--stdin`) replaces the page's body blocks; `--properties` updates the named columns. `--upsert-by` supports title, rich_text, select, status, number, url, email, and phone_number properties. + +### Two-tool split: when to use what + +| Goal | Tool | Why | +|------|------|-----| +| Push markdown content into a page or new DB row | **`notion_writer.py`** | Converts markdown → Notion blocks (tables, headings, code, lists, rich-text, anchors) | +| Create a DB row with both content and metadata in one shot | **`notion_writer.py`** with `--properties` | Single API call, full schema validation | +| Update properties on many existing pages (bulk column edit) | **MCP `notion-update-page`** | Native schema awareness, no markdown work needed | +| Search / move pages with metadata transition | 31-notion-organizer (Phase 3 work) | Cross-database operations | + --- ## Markdown Support @@ -131,6 +170,20 @@ print("Hello") ``` ``` +### Inline rich-text + +| Markdown | Result | +|----------|--------| +| `**bold**` or `__bold__` | bold | +| `*italic*` or `_italic_` | italic | +| `` `code` `` | inline code | +| `~~strike~~` | strikethrough | +| `[text](https://...)` | link (absolute URLs only) | +| `[text](#anchor)` | bold (Notion rejects fragment URLs) | +| `[text](relative/path.md)` | plain text (Notion rejects relative URLs) | + +Notion's URL validator requires absolute URLs for link annotations. The parser converts TOC-style anchor links to bold to preserve navigation intent and silently strips relative paths. + --- ## Examples @@ -160,6 +213,40 @@ python jamie_video_info.py VIDEO_ID --json | \ python notion_writer.py --page PAGE_URL --stdin ``` +### Test against the "Working with AI" database + +The user's primary export target is the AI research database (data_source_id `f8f19ede-32bd-43ac-9f60-0651f6f40afe`). End-to-end check: + +```bash +# 1) Sanity: connection works +python notion_writer.py --test + +# 2) Inspect the schema (confirms data_source resolution) +python notion_writer.py \ + --database f8f19ede-32bd-43ac-9f60-0651f6f40afe \ + --info + +# 3) Upsert a test row with full metadata +python notion_writer.py \ + --database f8f19ede-32bd-43ac-9f60-0651f6f40afe \ + --title "[smoke test] notion_writer v3 migration" \ + --upsert-by "Name" \ + --properties '{ + "Status": "Done", + "AI used": ["Claude Code"], + "Topic": ["AI"], + "Category": ["Research"], + "Account Code": "OurDigital", + "Type": "Memo", + "AI summary": "Verifying the data_sources API migration end-to-end." + }' \ + --stdin <<< "## Smoke test + +Migration verified — properties + content land in one shot." +``` + +Re-running step 3 should update the existing row instead of creating a duplicate (upsert). + --- ## API Limits @@ -176,13 +263,22 @@ The script automatically batches large content. ## Troubleshooting -### "Could not find page" -- Ensure page is shared with your integration +### "Notion object not found" +- Ensure the page or database is shared with your integration (Notion → "..." → Connections → add integration) - Check URL/ID is correct +- For databases, the integration must have access to *each* data source under it -### "Invalid token" -- Verify NOTION_API_KEY in .env -- Token should start with `secret_` +### "Unauthorized" +- Verify `NOTION_API_KEY` in `.env` +- Token should start with `secret_` (internal integration token) + +### "Database has multiple data sources" +- Pass a specific `data_source_id` instead of the database URL/id +- List data sources via the Notion API or open the data source individually in Notion to grab its ID + +### "Property 'X' has unsupported type" +- The script handles common property types but skips computed/read-only ones (formula, rollup, created_by, last_edited_by, last_edited_time, unique_id) +- Drop the unsupported key from `--properties` JSON ### "Rate limited" - Wait and retry @@ -193,11 +289,13 @@ The script automatically batches large content. ## File Structure ``` -02-notion-writer/ +32-notion-writer/ ├── code/ │ ├── CLAUDE.md # This skill document │ ├── scripts/ │ │ ├── notion_writer.py # Main script +│ │ ├── _notion_compat.py # API compat helper (data_sources resolution) +│ │ ├── test_parser.py # Markdown parser tests │ │ ├── venv/ # Python environment │ │ ├── .env # API token (not committed) │ │ └── .env.example # Template @@ -212,7 +310,7 @@ The script automatically batches large content. ```bash # Navigate -cd ~/Project/our-claude-skills/custom-skills/02-notion-writer/code/scripts +cd ~/Project/our-claude-skills/custom-skills/32-notion-writer/code/scripts source venv/bin/activate # Test @@ -226,8 +324,18 @@ python notion_writer.py -p PAGE_URL -f content.md -r # Create DB row python notion_writer.py -d DB_URL -t "Title" -f content.md + +# Create DB row with full metadata +python notion_writer.py -d DB_URL -t "Title" --properties '{"Status": "Done"}' -f content.md + +# Upsert (avoid duplicates) +python notion_writer.py -d DB_URL -t "Title" --upsert-by "Name" -f content.md ``` --- -*Version 1.0.0 | Claude Code | 2025-12-26* +*Version 1.1.0 | Claude Code | 2026-04-27* + +Changelog: +- 1.1.0 — Migrated to Notion API 2025-09-03 (multi-source databases). Added `--properties` JSON flag, `--upsert-by` for idempotency, anchor-link parser fix, friendlier API error messages. +- 1.0.0 — Initial release with markdown→Notion block conversion. diff --git a/custom-skills/32-notion-writer/code/scripts/_notion_compat.py b/custom-skills/32-notion-writer/code/scripts/_notion_compat.py new file mode 100644 index 0000000..4f8b88f --- /dev/null +++ b/custom-skills/32-notion-writer/code/scripts/_notion_compat.py @@ -0,0 +1,255 @@ +""" +Notion API compatibility helpers for v3 SDK + 2025-09-03 multi-source databases. + +Shared by 32-notion-writer and 31-notion-organizer. Pure functions where +possible; sync API helpers for the writer; async resolution is inlined in +31's scripts to keep this module sync-only. +""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any, Dict, List, Optional, Union + +from notion_client import Client +from notion_client.errors import APIErrorCode, APIResponseError + + +def make_client(api_key: str) -> Client: + """Build a sync Notion client on the SDK default API version (2025-09-03+).""" + return Client(auth=api_key) + + +def resolve_data_source_id(client: Client, target_id: str) -> str: + """ + Accept either a database_id or a data_source_id, return a data_source_id. + + Strategy: try data_sources.retrieve first; on ObjectNotFound, fall back to + databases.retrieve and return the first attached data source. Most + workspaces have single-source databases so this is one extra round-trip + only on legacy IDs. + """ + try: + client.data_sources.retrieve(data_source_id=target_id) + return target_id + except APIResponseError as exc: + if exc.code != APIErrorCode.ObjectNotFound: + raise + + db = client.databases.retrieve(database_id=target_id) + sources = db.get("data_sources") or [] + if not sources: + raise ValueError( + f"Database {target_id} has no data_sources attached. " + "Either it is archived or the integration lacks access." + ) + if len(sources) > 1: + names = ", ".join(s.get("name", s["id"]) for s in sources) + raise ValueError( + f"Database {target_id} has multiple data sources ({names}). " + "Pass a specific data_source_id instead of the database_id." + ) + return sources[0]["id"] + + +def get_schema(client: Client, data_source_id: str) -> Dict[str, Any]: + """Fetch a data source's full schema (properties, title, parent).""" + return client.data_sources.retrieve(data_source_id=data_source_id) + + +def find_title_property(properties: Dict[str, Any]) -> Optional[str]: + """Return the name of the title property in a schema, or None.""" + for name, prop in properties.items(): + if prop.get("type") == "title": + return name + return None + + +def build_data_source_parent(data_source_id: str) -> Dict[str, str]: + """Parent shape for pages.create when targeting a data source.""" + return {"type": "data_source_id", "data_source_id": data_source_id} + + +def parse_properties_arg(arg: str) -> Dict[str, Any]: + """ + Parse the --properties CLI value. Accepts either: + - A JSON object literal: '{"Status": "In progress"}' + - A path to a JSON file: './props.json' + """ + arg = arg.strip() + if arg.startswith("{"): + return json.loads(arg) + path = Path(arg).expanduser() + if not path.is_file(): + raise ValueError( + f"--properties value is neither a JSON object nor an existing file: {arg!r}" + ) + return json.loads(path.read_text(encoding="utf-8")) + + +def coerce_property_value( + name: str, schema_entry: Dict[str, Any], value: Any +) -> Dict[str, Any]: + """ + Convert a Python-friendly value into a Notion property body, given the + schema entry from data_sources.retrieve(). + + Supported types: title, rich_text, status, select, multi_select, date, + checkbox, number, url, email, phone_number, people, files (URL form), + relation. Unsupported types raise ValueError with a clear message so the + caller can surface it to the user. + """ + prop_type = schema_entry.get("type") + + if prop_type == "title": + return {"title": [{"type": "text", "text": {"content": str(value)}}]} + + if prop_type == "rich_text": + return {"rich_text": [{"type": "text", "text": {"content": str(value)}}]} + + if prop_type == "status": + return {"status": {"name": str(value)}} + + if prop_type == "select": + return {"select": {"name": str(value)}} + + if prop_type == "multi_select": + items = value if isinstance(value, list) else [value] + return {"multi_select": [{"name": str(v)} for v in items]} + + if prop_type == "date": + if isinstance(value, dict): + return {"date": value} + return {"date": {"start": str(value)}} + + if prop_type == "checkbox": + return {"checkbox": bool(value)} + + if prop_type == "number": + return {"number": float(value) if value is not None else None} + + if prop_type == "url": + return {"url": str(value) if value else None} + + if prop_type == "email": + return {"email": str(value) if value else None} + + if prop_type == "phone_number": + return {"phone_number": str(value) if value else None} + + if prop_type == "people": + ids = value if isinstance(value, list) else [value] + return {"people": [{"id": str(i)} for i in ids]} + + if prop_type == "files": + items = value if isinstance(value, list) else [value] + files: List[Dict[str, Any]] = [] + for i, item in enumerate(items): + url = item if isinstance(item, str) else item.get("url") + label = f"file_{i}" if isinstance(item, str) else item.get("name", f"file_{i}") + files.append({"name": label, "type": "external", "external": {"url": url}}) + return {"files": files} + + if prop_type == "relation": + ids = value if isinstance(value, list) else [value] + return {"relation": [{"id": str(i)} for i in ids]} + + raise ValueError( + f"Property {name!r} has unsupported type {prop_type!r} " + "(read-only or not yet handled). Supported: title, rich_text, status, " + "select, multi_select, date, checkbox, number, url, email, " + "phone_number, people, files, relation." + ) + + +def coerce_properties( + schema: Dict[str, Any], values: Dict[str, Any] +) -> Dict[str, Any]: + """ + Coerce a flat {prop_name: python_value} dict into Notion properties body. + Unknown property names raise ValueError listing the valid options so + callers can fix their input quickly. + """ + schema_props = schema.get("properties") or {} + out: Dict[str, Any] = {} + for name, value in values.items(): + if name not in schema_props: + valid = ", ".join(sorted(schema_props.keys())) + raise ValueError( + f"Unknown property {name!r}. Valid properties: {valid}" + ) + out[name] = coerce_property_value(name, schema_props[name], value) + return out + + +def explain_api_error(exc: APIResponseError, context: str = "") -> str: + """ + Translate a notion-client APIResponseError into a human-friendly message. + The default exc message often hides the root cause behind validation jargon. + """ + code = exc.code + suffix = f" ({context})" if context else "" + + if code == APIErrorCode.ObjectNotFound: + return ( + f"Notion object not found{suffix}. The ID may be wrong, the page " + "may be archived, or the integration is not connected to it. " + "Open the page/database in Notion and add your integration via " + "'... -> Connections'." + ) + if code == APIErrorCode.Unauthorized: + return ( + f"Unauthorized{suffix}. Check that NOTION_API_KEY is set and " + "starts with 'secret_' (internal integration token)." + ) + if code == APIErrorCode.RestrictedResource: + return ( + f"Restricted resource{suffix}. The integration's capabilities do " + "not allow this action. Update capabilities at " + "https://www.notion.so/my-integrations." + ) + if code == APIErrorCode.ValidationError: + return f"Validation error{suffix}: {exc.body.get('message', str(exc))}" + if code == APIErrorCode.RateLimited: + return f"Rate limited{suffix}. Back off and retry." + return f"Notion API error [{code}]{suffix}: {exc}" + + +def find_existing_page( + client: Client, + data_source_id: str, + property_name: str, + schema_entry: Dict[str, Any], + value: Any, +) -> Optional[Dict[str, Any]]: + """ + Look up an existing page in a data source by an exact property value. + Returns the first matching page object or None. Used by --upsert-by. + + Supported filter types: title, rich_text, select, status, number, url, email. + """ + prop_type = schema_entry.get("type") + str_value = str(value) + + if prop_type in ("title", "rich_text"): + prop_filter = {prop_type: {"equals": str_value}} + elif prop_type in ("select", "status"): + prop_filter = {prop_type: {"equals": str_value}} + elif prop_type == "number": + prop_filter = {"number": {"equals": float(value)}} + elif prop_type in ("url", "email", "phone_number"): + prop_filter = {prop_type: {"equals": str_value}} + else: + raise ValueError( + f"Cannot upsert by property type {prop_type!r}. Use title, " + "rich_text, select, status, number, url, email, or phone_number." + ) + + response = client.data_sources.query( + data_source_id=data_source_id, + filter={"property": property_name, **prop_filter}, + page_size=1, + ) + results = response.get("results") or [] + return results[0] if results else None 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 dbbac0b..4bad01c 100644 --- a/custom-skills/32-notion-writer/code/scripts/notion_writer.py +++ b/custom-skills/32-notion-writer/code/scripts/notion_writer.py @@ -7,17 +7,21 @@ Supports both page content updates and database row creation. import os import sys import re +import json import argparse from pathlib import Path from typing import Optional, List, Dict, Any from dotenv import load_dotenv from notion_client import Client +from notion_client.errors import APIResponseError + +import _notion_compat as compat # Load environment variables load_dotenv(Path(__file__).parent / '.env') -NOTION_TOKEN = os.getenv('NOTION_API_KEY') +NOTION_TOKEN = os.getenv('NOTION_API_KEY') or os.getenv('NOTION_TOKEN') def extract_notion_id(url_or_id: str) -> Optional[str]: @@ -164,6 +168,14 @@ INLINE_PATTERNS = [ ] +_ABSOLUTE_URL_RE = re.compile(r'^(?:[a-zA-Z][a-zA-Z0-9+.-]*:)') + + +def _is_absolute_url(url: str) -> bool: + """Notion's link annotation rejects fragment-only and relative URLs.""" + return bool(_ABSOLUTE_URL_RE.match(url)) + + def _rich_span(content: str, **annotations: Any) -> Dict[str, Any]: """Build one rich-text span with optional annotations / link.""" span: Dict[str, Any] = { @@ -209,7 +221,15 @@ def parse_rich_text(text: str) -> List[Dict[str, Any]]: if kind == 'code': spans.append(_rich_span(m.group(1), code=True)) elif kind == 'link': - spans.append(_rich_span(m.group(1), link=m.group(2))) + link_text, link_url = m.group(1), m.group(2) + if _is_absolute_url(link_url): + spans.append(_rich_span(link_text, link=link_url)) + elif link_url.startswith('#'): + # TOC-style fragment link — Notion rejects, preserve nav intent as bold. + spans.append(_rich_span(link_text, bold=True)) + else: + # Relative path or unrecognized scheme — drop link, keep text. + spans.append(_rich_span(link_text)) elif kind in ('bold_star', 'bold_under'): spans.append(_rich_span(m.group(1), bold=True)) elif kind == 'strike': @@ -427,11 +447,21 @@ def get_page_info(notion: Client, page_id: str) -> Optional[Dict]: return None -def get_database_info(notion: Client, database_id: str) -> Optional[Dict]: - """Get database information.""" +def get_database_info(notion: Client, target_id: str) -> Optional[Dict]: + """ + Fetch full schema for a database or data source. + + Accepts either a database_id or a data_source_id; resolves to the + data_source_id first, then returns the data source object (which contains + the property schema under the 2025-09-03 API). + """ try: - formatted_id = format_id_with_dashes(database_id) - return notion.databases.retrieve(database_id=formatted_id) + formatted_id = format_id_with_dashes(target_id) + data_source_id = compat.resolve_data_source_id(notion, formatted_id) + return compat.get_schema(notion, data_source_id) + except APIResponseError as exc: + print(f"Error retrieving database: {compat.explain_api_error(exc, formatted_id)}") + return None except Exception as e: print(f"Error retrieving database: {e}") return None @@ -476,18 +506,21 @@ def clear_page_content(notion: Client, page_id: str) -> bool: return False -def create_database_row(notion: Client, database_id: str, properties: Dict, content_blocks: List[Dict] = None) -> Optional[str]: - """Create a new row in a Notion database.""" - try: - formatted_id = format_id_with_dashes(database_id) +def create_database_row(notion: Client, data_source_id: str, properties: Dict, content_blocks: List[Dict] = None) -> Optional[str]: + """ + Create a new row in a Notion data source (multi-source database model). + `data_source_id` must be a resolved data_source_id, not a database_id. + Use compat.resolve_data_source_id() upstream. + """ + try: page_data = { - "parent": {"database_id": formatted_id}, - "properties": properties + "parent": compat.build_data_source_parent(data_source_id), + "properties": properties, } if content_blocks: - page_data["children"] = content_blocks[:100] # Limit on creation + page_data["children"] = content_blocks[:100] # API limit on creation result = notion.pages.create(**page_data) @@ -496,11 +529,27 @@ def create_database_row(notion: Client, database_id: str, properties: Dict, cont append_to_page(notion, result['id'], content_blocks[100:]) return result['id'] + except APIResponseError as exc: + print(f"Error creating database row: {compat.explain_api_error(exc)}") + return None except Exception as e: print(f"Error creating database row: {e}") return None +def update_page_properties(notion: Client, page_id: str, properties: Dict) -> bool: + """Update properties on an existing page (used by upsert).""" + try: + notion.pages.update(page_id=page_id, properties=properties) + return True + except APIResponseError as exc: + print(f"Error updating page properties: {compat.explain_api_error(exc)}") + return False + except Exception as e: + print(f"Error updating page properties: {e}") + return False + + def write_to_page(notion: Client, page_id: str, markdown_content: str, mode: str = 'append') -> bool: """Write markdown content to a Notion page.""" blocks = markdown_to_notion_blocks(markdown_content) @@ -540,11 +589,13 @@ Examples: ) parser.add_argument('--page', '-p', help='Notion page URL or ID') - parser.add_argument('--database', '-d', help='Notion database URL or ID') + parser.add_argument('--database', '-d', help='Notion database URL, database_id, or data_source_id') parser.add_argument('--file', '-f', help='Markdown file to push') parser.add_argument('--stdin', action='store_true', help='Read content from stdin') parser.add_argument('--replace', '-r', action='store_true', help='Replace page content instead of append') - parser.add_argument('--title', '-t', help='Title for database row (required for database)') + parser.add_argument('--title', '-t', help='Title for database row (required for database unless --properties sets it)') + parser.add_argument('--properties', help='Extra properties as JSON object literal or path to .json file (e.g. \'{"Status": "In progress", "Topic": ["AI"]}\')') + parser.add_argument('--upsert-by', dest='upsert_by', help='Property name to match for upsert; updates existing row instead of creating duplicate') parser.add_argument('--test', action='store_true', help='Test Notion API connection') parser.add_argument('--list', '-l', nargs='?', const='all', choices=['all', 'pages', 'databases'], help='List accessible pages and/or databases (default: all)') @@ -557,7 +608,7 @@ Examples: print("Get your integration token from: https://www.notion.so/my-integrations") sys.exit(1) - notion = Client(auth=NOTION_TOKEN) + notion = compat.make_client(NOTION_TOKEN) # Test connection if args.test: @@ -657,43 +708,94 @@ Examples: sys.exit(1) return - # Create database row + # Create or upsert database row if args.database: - if not content and not args.title: - print("Error: Provide --title and/or --file for database row") + if not content and not args.title and not args.properties: + print("Error: Provide --title, --properties, and/or --file for database row") sys.exit(1) db_id = extract_notion_id(args.database) if not db_id: print(f"Error: Invalid Notion database URL/ID: {args.database}") sys.exit(1) + formatted_id = format_id_with_dashes(db_id) - # Get database schema to find title property - db_info = get_database_info(notion, db_id) - if not db_info: + # Resolve to a data_source_id (handles both database_id and data_source_id input) + try: + data_source_id = compat.resolve_data_source_id(notion, formatted_id) + except APIResponseError as exc: + print(f"Error: {compat.explain_api_error(exc, formatted_id)}") + sys.exit(1) + except ValueError as exc: + print(f"Error: {exc}") sys.exit(1) - # Find the title property - title_prop = None - for prop_name, prop_config in db_info.get('properties', {}).items(): - if prop_config.get('type') == 'title': - title_prop = prop_name - break - + # Fetch schema + schema = compat.get_schema(notion, data_source_id) + schema_props = schema.get('properties', {}) + title_prop = compat.find_title_property(schema_props) if not title_prop: - print("Error: Could not find title property in database") + print("Error: Data source has no title property; cannot create row.") sys.exit(1) - properties = { - title_prop: { - "title": [{"text": {"content": args.title or "Untitled"}}] - } - } + # Build properties body: start from --properties, layer --title on top + property_values: Dict[str, Any] = {} + if args.properties: + try: + property_values = compat.parse_properties_arg(args.properties) + except (ValueError, json.JSONDecodeError) as exc: + print(f"Error parsing --properties: {exc}") + sys.exit(1) + + if args.title: + property_values[title_prop] = args.title + elif title_prop not in property_values: + property_values[title_prop] = "Untitled" + + try: + properties = compat.coerce_properties(schema, property_values) + except ValueError as exc: + print(f"Error: {exc}") + sys.exit(1) content_blocks = markdown_to_notion_blocks(content) if content else None + # Upsert path: look for existing row by the named property + if args.upsert_by: + if args.upsert_by not in schema_props: + valid = ", ".join(sorted(schema_props.keys())) + print(f"Error: --upsert-by property {args.upsert_by!r} not in schema. Valid: {valid}") + sys.exit(1) + lookup_value = property_values.get(args.upsert_by) + if lookup_value is None: + print(f"Error: --upsert-by {args.upsert_by!r} requires a value in --properties or --title") + sys.exit(1) + + try: + existing = compat.find_existing_page( + notion, data_source_id, args.upsert_by, + schema_props[args.upsert_by], lookup_value, + ) + except APIResponseError as exc: + 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): + 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 + print(f"Creating database row...") - row_id = create_database_row(notion, db_id, properties, content_blocks) + row_id = create_database_row(notion, data_source_id, properties, content_blocks) if row_id: print(f"✅ Successfully created database row") diff --git a/custom-skills/32-notion-writer/code/scripts/test_parser.py b/custom-skills/32-notion-writer/code/scripts/test_parser.py index d1a3c19..899b41b 100644 --- a/custom-skills/32-notion-writer/code/scripts/test_parser.py +++ b/custom-skills/32-notion-writer/code/scripts/test_parser.py @@ -165,6 +165,31 @@ def test_blocks_table(): "link inline formatting preserved inside table cell") +def test_rich_text_anchor_link_becomes_bold(): + """TOC-style fragment links must not be sent as Notion link annotations.""" + spans = parse_rich_text("see [Section A](#section-a) below") + bold_spans = [s for s in spans if s.get("annotations", {}).get("bold")] + _assert(len(bold_spans) == 1, "anchor-only link converted to bold span") + _assert(bold_spans[0]["text"]["content"] == "Section A", "anchor link text preserved") + _assert("link" not in bold_spans[0]["text"], "no link annotation on fragment URL") + + +def test_rich_text_relative_link_becomes_plain(): + """Relative paths can't be Notion links either; render plain text.""" + spans = parse_rich_text("see [the spec](../spec.md) please") + spec_span = next(s for s in spans if s["text"]["content"] == "the spec") + _assert("link" not in spec_span["text"], "relative-path link annotation dropped") + _assert("annotations" not in spec_span, "no spurious bold/italic on plain text") + + +def test_rich_text_absolute_link_preserved(): + """Regression: absolute URLs must still produce a real Notion link.""" + spans = parse_rich_text("[docs](https://example.com/x) and [mail](mailto:a@b.co)") + urls = [s["text"].get("link", {}).get("url") for s in spans if s["text"].get("link")] + _assert("https://example.com/x" in urls, "https URL preserved as link") + _assert("mailto:a@b.co" in urls, "mailto URL preserved as link") + + def test_no_literal_markers_leak(): """Regression: before the rich-text rewrite, ** and [ ] leaked as literal text.""" blocks = markdown_to_notion_blocks("A **bold** word and a [link](https://x.com).") @@ -189,6 +214,9 @@ def run_all(): test_blocks_todo, test_blocks_quote_code_divider, test_blocks_table, + test_rich_text_anchor_link_becomes_bold, + test_rich_text_relative_link_becomes_plain, + test_rich_text_absolute_link_preserved, test_no_literal_markers_leak, ] for t in tests: