Andrew Yim
4e692d0e9a
Code review caught that all 9 other create_*_block factories include "object": "block" as the first key, but the new callout factory was missing it. The Notion API accepts both forms, but the inconsistency would compound; align with the existing convention. 22/22 tests still pass. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
838 lines
29 KiB
Python
838 lines
29 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Notion Writer - Push markdown content to Notion pages or databases.
|
||
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, Union
|
||
|
||
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') or os.getenv('NOTION_TOKEN')
|
||
|
||
|
||
def extract_notion_id(url_or_id: str) -> Optional[str]:
|
||
"""Extract Notion page/database ID from URL or raw ID."""
|
||
# Already a raw ID (32 chars hex, with or without dashes)
|
||
clean_id = url_or_id.replace('-', '')
|
||
if re.match(r'^[a-f0-9]{32}$', clean_id):
|
||
return clean_id
|
||
|
||
# Notion URL patterns
|
||
patterns = [
|
||
r'notion\.so/(?:[^/]+/)?([a-f0-9]{32})', # notion.so/workspace/page-id
|
||
r'notion\.so/(?:[^/]+/)?[^-]+-([a-f0-9]{32})', # notion.so/workspace/Page-Title-id
|
||
r'notion\.site/(?:[^/]+/)?([a-f0-9]{32})', # public notion.site
|
||
r'notion\.site/(?:[^/]+/)?[^-]+-([a-f0-9]{32})',
|
||
r'([a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12})', # UUID format
|
||
]
|
||
|
||
for pattern in patterns:
|
||
match = re.search(pattern, url_or_id)
|
||
if match:
|
||
return match.group(1).replace('-', '')
|
||
|
||
return None
|
||
|
||
|
||
def format_id_with_dashes(raw_id: str) -> str:
|
||
"""Format 32-char ID to UUID format with dashes."""
|
||
if len(raw_id) == 32:
|
||
return f"{raw_id[:8]}-{raw_id[8:12]}-{raw_id[12:16]}-{raw_id[16:20]}-{raw_id[20:]}"
|
||
return raw_id
|
||
|
||
|
||
TABLE_SEPARATOR_RE = re.compile(r'^\s*\|?(\s*:?-{3,}:?\s*\|)+\s*:?-{3,}:?\s*\|?\s*$')
|
||
|
||
|
||
def _is_table_row(line: str) -> bool:
|
||
stripped = line.strip()
|
||
return stripped.startswith('|') and stripped.endswith('|') and stripped.count('|') >= 2
|
||
|
||
|
||
def _split_table_row(line: str) -> List[str]:
|
||
stripped = line.strip()
|
||
if stripped.startswith('|'):
|
||
stripped = stripped[1:]
|
||
if stripped.endswith('|'):
|
||
stripped = stripped[:-1]
|
||
return [cell.strip() for cell in stripped.split('|')]
|
||
|
||
|
||
def markdown_to_notion_blocks(content: Union[str, List[str]]) -> List[Dict[str, Any]]:
|
||
"""Convert markdown text (or pre-split lines) to Notion block objects.
|
||
|
||
Reentrant: container detectors recursively call _parse_lines on inner content.
|
||
"""
|
||
if isinstance(content, str):
|
||
lines = content.split('\n')
|
||
else:
|
||
lines = list(content)
|
||
return _parse_lines(lines)
|
||
|
||
|
||
def _parse_lines(lines: List[str]) -> List[Dict[str, Any]]:
|
||
"""Walk lines and emit Notion blocks. Called recursively by container detectors."""
|
||
blocks = []
|
||
i = 0
|
||
while i < len(lines):
|
||
line = lines[i]
|
||
if not line.strip():
|
||
i += 1
|
||
continue
|
||
if _is_table_row(line) and i + 1 < len(lines) and TABLE_SEPARATOR_RE.match(lines[i + 1]):
|
||
header_cells = _split_table_row(line)
|
||
i += 2
|
||
body_rows: List[List[str]] = []
|
||
while i < len(lines) and _is_table_row(lines[i]):
|
||
body_rows.append(_split_table_row(lines[i]))
|
||
i += 1
|
||
blocks.append(create_table_block(header_cells, body_rows))
|
||
continue
|
||
if line.startswith('######'):
|
||
blocks.append(create_heading_block(line[6:].strip(), 3))
|
||
elif line.startswith('#####'):
|
||
blocks.append(create_heading_block(line[5:].strip(), 3))
|
||
elif line.startswith('####'):
|
||
blocks.append(create_heading_block(line[4:].strip(), 3))
|
||
elif line.startswith('###'):
|
||
blocks.append(create_heading_block(line[3:].strip(), 3))
|
||
elif line.startswith('##'):
|
||
blocks.append(create_heading_block(line[2:].strip(), 2))
|
||
elif line.startswith('#'):
|
||
blocks.append(create_heading_block(line[1:].strip(), 1))
|
||
elif line.startswith('```'):
|
||
language = line[3:].strip() or 'plain text'
|
||
code_lines = []
|
||
i += 1
|
||
while i < len(lines) and not lines[i].startswith('```'):
|
||
code_lines.append(lines[i])
|
||
i += 1
|
||
blocks.append(create_code_block('\n'.join(code_lines), language))
|
||
# Checkbox / Todo (must come before generic bullet match)
|
||
elif line.strip().startswith('- [ ]'):
|
||
text = line.strip()[5:].strip()
|
||
blocks.append(create_todo_block(text, False))
|
||
elif line.strip().startswith('- [x]') or line.strip().startswith('- [X]'):
|
||
text = line.strip()[5:].strip()
|
||
blocks.append(create_todo_block(text, True))
|
||
elif line.strip().startswith('- ') or line.strip().startswith('* '):
|
||
text = line.strip()[2:]
|
||
blocks.append(create_bulleted_list_block(text))
|
||
elif re.match(r'^\d+\.\s', line.strip()):
|
||
text = re.sub(r'^\d+\.\s', '', line.strip())
|
||
blocks.append(create_numbered_list_block(text))
|
||
# Blockquote — GitHub-alert callout takes priority over generic quote
|
||
elif line.startswith('>'):
|
||
first_body = line[1:].strip()
|
||
alert_match = ALERT_RE.match(first_body)
|
||
if alert_match:
|
||
alert_type = alert_match.group(1)
|
||
body_lines: List[str] = []
|
||
i += 1
|
||
while i < len(lines) and lines[i].lstrip().startswith('>'):
|
||
body_lines.append(lines[i].lstrip()[1:].lstrip())
|
||
i += 1
|
||
body_text = '\n'.join(body_lines)
|
||
blocks.append(create_callout_block(body_text, alert_type))
|
||
continue
|
||
blocks.append(create_quote_block(first_body))
|
||
elif line.strip() in ['---', '***', '___']:
|
||
blocks.append(create_divider_block())
|
||
else:
|
||
blocks.append(create_paragraph_block(line))
|
||
i += 1
|
||
return blocks
|
||
|
||
|
||
INLINE_PATTERNS = [
|
||
('code', re.compile(r'`([^`\n]+)`')),
|
||
('link', re.compile(r'\[([^\]]+)\]\(([^)\s]+)\)')),
|
||
('bold_star', re.compile(r'\*\*([^*\n]+)\*\*')),
|
||
('bold_under', re.compile(r'__([^_\n]+)__')),
|
||
('strike', re.compile(r'~~([^~\n]+)~~')),
|
||
('italic_star', re.compile(r'\*([^*\n]+)\*')),
|
||
('italic_under', re.compile(r'(?<![a-zA-Z0-9_])_([^_\n]+)_(?![a-zA-Z0-9_])')),
|
||
]
|
||
|
||
ALERT_TYPES = {
|
||
'NOTE': ('ℹ️', 'blue_background'),
|
||
'TIP': ('💡', 'green_background'),
|
||
'IMPORTANT': ('☝️', 'purple_background'),
|
||
'WARNING': ('⚠️', 'yellow_background'),
|
||
'CAUTION': ('🚨', 'red_background'),
|
||
}
|
||
ALERT_RE = re.compile(r'^\s*\[!(NOTE|TIP|IMPORTANT|WARNING|CAUTION)\]\s*$')
|
||
|
||
|
||
_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] = {
|
||
"type": "text",
|
||
"text": {"content": content[:2000]},
|
||
}
|
||
link = annotations.pop('link', None)
|
||
if link:
|
||
span["text"]["link"] = {"url": link}
|
||
if annotations:
|
||
span["annotations"] = {k: v for k, v in annotations.items() if v}
|
||
return span
|
||
|
||
|
||
def parse_rich_text(text: str) -> List[Dict[str, Any]]:
|
||
"""Parse markdown inline formatting into Notion rich-text spans.
|
||
|
||
Supports: **bold**, __bold__, *italic*, _italic_, `code`, ~~strike~~, [text](url).
|
||
No nesting (Notion rich-text flattens annotations onto each span anyway).
|
||
"""
|
||
if not text:
|
||
return []
|
||
|
||
spans: List[Dict[str, Any]] = []
|
||
remaining = text
|
||
|
||
while remaining:
|
||
# Find the earliest inline token in what's left.
|
||
best: Optional[tuple] = None # (kind, match_obj)
|
||
for kind, pattern in INLINE_PATTERNS:
|
||
m = pattern.search(remaining)
|
||
if m and (best is None or m.start() < best[1].start()):
|
||
best = (kind, m)
|
||
|
||
if not best:
|
||
spans.append(_rich_span(remaining))
|
||
break
|
||
|
||
kind, m = best
|
||
if m.start() > 0:
|
||
spans.append(_rich_span(remaining[:m.start()]))
|
||
|
||
if kind == 'code':
|
||
spans.append(_rich_span(m.group(1), code=True))
|
||
elif kind == 'link':
|
||
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':
|
||
spans.append(_rich_span(m.group(1), strikethrough=True))
|
||
elif kind in ('italic_star', 'italic_under'):
|
||
spans.append(_rich_span(m.group(1), italic=True))
|
||
|
||
remaining = remaining[m.end():]
|
||
|
||
return spans
|
||
|
||
|
||
def create_paragraph_block(text: str) -> Dict[str, Any]:
|
||
return {
|
||
"object": "block",
|
||
"type": "paragraph",
|
||
"paragraph": {
|
||
"rich_text": parse_rich_text(text)
|
||
}
|
||
}
|
||
|
||
|
||
def create_heading_block(text: str, level: int) -> Dict[str, Any]:
|
||
heading_type = f"heading_{level}"
|
||
return {
|
||
"object": "block",
|
||
"type": heading_type,
|
||
heading_type: {
|
||
"rich_text": parse_rich_text(text)
|
||
}
|
||
}
|
||
|
||
|
||
def create_bulleted_list_block(text: str) -> Dict[str, Any]:
|
||
return {
|
||
"object": "block",
|
||
"type": "bulleted_list_item",
|
||
"bulleted_list_item": {
|
||
"rich_text": parse_rich_text(text)
|
||
}
|
||
}
|
||
|
||
|
||
def create_numbered_list_block(text: str) -> Dict[str, Any]:
|
||
return {
|
||
"object": "block",
|
||
"type": "numbered_list_item",
|
||
"numbered_list_item": {
|
||
"rich_text": parse_rich_text(text)
|
||
}
|
||
}
|
||
|
||
|
||
def create_todo_block(text: str, checked: bool) -> Dict[str, Any]:
|
||
return {
|
||
"object": "block",
|
||
"type": "to_do",
|
||
"to_do": {
|
||
"rich_text": parse_rich_text(text),
|
||
"checked": checked
|
||
}
|
||
}
|
||
|
||
|
||
def create_quote_block(text: str) -> Dict[str, Any]:
|
||
return {
|
||
"object": "block",
|
||
"type": "quote",
|
||
"quote": {
|
||
"rich_text": parse_rich_text(text)
|
||
}
|
||
}
|
||
|
||
|
||
def create_callout_block(text: str, alert_type: str) -> Dict[str, Any]:
|
||
"""Create a callout block with icon + color from a GitHub alert type."""
|
||
icon, color = ALERT_TYPES[alert_type]
|
||
return {
|
||
"object": "block",
|
||
"type": "callout",
|
||
"callout": {
|
||
"rich_text": parse_rich_text(text),
|
||
"icon": {"type": "emoji", "emoji": icon},
|
||
"color": color,
|
||
},
|
||
}
|
||
|
||
|
||
def create_code_block(code: str, language: str) -> Dict[str, Any]:
|
||
return {
|
||
"object": "block",
|
||
"type": "code",
|
||
"code": {
|
||
"rich_text": parse_rich_text(code),
|
||
"language": language.lower()
|
||
}
|
||
}
|
||
|
||
|
||
def create_divider_block() -> Dict[str, Any]:
|
||
return {
|
||
"object": "block",
|
||
"type": "divider",
|
||
"divider": {}
|
||
}
|
||
|
||
|
||
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.
|
||
|
||
All rows are padded to the header width so Notion accepts the payload.
|
||
Each cell renders through `parse_rich_text` so inline formatting works
|
||
inside tables too.
|
||
"""
|
||
width = max(len(header_cells), max((len(r) for r in body_rows), default=0))
|
||
|
||
def row_block(cells: List[str]) -> Dict[str, Any]:
|
||
padded = cells + [''] * (width - len(cells))
|
||
return {
|
||
"object": "block",
|
||
"type": "table_row",
|
||
"table_row": {
|
||
"cells": [parse_rich_text(c) for c in padded]
|
||
},
|
||
}
|
||
|
||
children = [row_block(header_cells)] + [row_block(r) for r in body_rows]
|
||
|
||
return {
|
||
"object": "block",
|
||
"type": "table",
|
||
"table": {
|
||
"table_width": width,
|
||
"has_column_header": True,
|
||
"has_row_header": False,
|
||
"children": children,
|
||
},
|
||
}
|
||
|
||
|
||
def list_accessible_content(notion: Client, filter_type: str = 'all') -> None:
|
||
"""List all accessible pages and databases."""
|
||
print("=" * 70)
|
||
print("Accessible Notion Content (Claude-D.intelligence)")
|
||
print("=" * 70)
|
||
|
||
databases = []
|
||
pages = []
|
||
|
||
try:
|
||
# Search all content and filter locally
|
||
response = notion.search()
|
||
all_results = response.get('results', [])
|
||
|
||
# Handle pagination
|
||
while response.get('has_more'):
|
||
response = notion.search(start_cursor=response.get('next_cursor'))
|
||
all_results.extend(response.get('results', []))
|
||
|
||
# Separate databases and pages
|
||
for item in all_results:
|
||
if item.get('object') == 'database':
|
||
databases.append(item)
|
||
elif item.get('object') == 'page':
|
||
pages.append(item)
|
||
|
||
# Show databases
|
||
if filter_type in ['all', 'database']:
|
||
print("\n📊 DATABASES")
|
||
print("-" * 70)
|
||
|
||
if databases:
|
||
for i, db in enumerate(databases, 1):
|
||
title = ""
|
||
if db.get('title'):
|
||
title = db['title'][0].get('plain_text', 'Untitled') if db['title'] else 'Untitled'
|
||
db_id = db['id']
|
||
url = db.get('url', '')
|
||
props = list(db.get('properties', {}).keys())[:5]
|
||
print(f"\n{i}. {title}")
|
||
print(f" ID: {db_id}")
|
||
print(f" URL: {url}")
|
||
print(f" Properties: {', '.join(props)}")
|
||
else:
|
||
print(" No databases accessible")
|
||
|
||
# Show pages
|
||
if filter_type in ['all', 'page']:
|
||
print("\n\n📄 PAGES")
|
||
print("-" * 70)
|
||
|
||
if pages:
|
||
for i, page in enumerate(pages, 1):
|
||
title = "Untitled"
|
||
if 'properties' in page:
|
||
for prop in page['properties'].values():
|
||
if prop.get('type') == 'title':
|
||
title_arr = prop.get('title', [])
|
||
if title_arr:
|
||
title = title_arr[0].get('plain_text', 'Untitled')
|
||
break
|
||
page_id = page['id']
|
||
url = page.get('url', '')
|
||
parent_type = page.get('parent', {}).get('type', 'unknown')
|
||
print(f"\n{i}. {title}")
|
||
print(f" ID: {page_id}")
|
||
print(f" URL: {url}")
|
||
print(f" Parent: {parent_type}")
|
||
else:
|
||
print(" No pages accessible")
|
||
|
||
print("\n" + "=" * 70)
|
||
db_count = len(databases) if filter_type != 'page' else 0
|
||
page_count = len(pages) if filter_type != 'database' else 0
|
||
print(f"Total: {db_count} databases, {page_count} pages")
|
||
print("=" * 70)
|
||
|
||
except Exception as e:
|
||
print(f"Error listing content: {e}")
|
||
|
||
|
||
def get_page_info(notion: Client, page_id: str) -> Optional[Dict]:
|
||
"""Get page information."""
|
||
try:
|
||
formatted_id = format_id_with_dashes(page_id)
|
||
return notion.pages.retrieve(page_id=formatted_id)
|
||
except Exception as e:
|
||
print(f"Error retrieving page: {e}")
|
||
return None
|
||
|
||
|
||
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(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
|
||
|
||
|
||
def append_to_page(notion: Client, page_id: str, blocks: List[Dict]) -> bool:
|
||
"""Append blocks to an existing Notion page."""
|
||
try:
|
||
formatted_id = format_id_with_dashes(page_id)
|
||
|
||
# Notion API limits to 100 blocks per request
|
||
for i in range(0, len(blocks), 100):
|
||
batch = blocks[i:i+100]
|
||
notion.blocks.children.append(
|
||
block_id=formatted_id,
|
||
children=batch
|
||
)
|
||
|
||
return True
|
||
except Exception as e:
|
||
print(f"Error appending to page: {e}")
|
||
return False
|
||
|
||
|
||
def clear_page_content(notion: Client, page_id: str) -> bool:
|
||
"""Clear all content from a page."""
|
||
try:
|
||
formatted_id = format_id_with_dashes(page_id)
|
||
|
||
# Get all child blocks
|
||
children = notion.blocks.children.list(block_id=formatted_id)
|
||
|
||
# Delete each block (skip already archived blocks)
|
||
for block in children.get('results', []):
|
||
if block.get('archived'):
|
||
continue
|
||
notion.blocks.delete(block_id=block['id'])
|
||
|
||
return True
|
||
except Exception as e:
|
||
print(f"Error clearing page: {e}")
|
||
return False
|
||
|
||
|
||
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": compat.build_data_source_parent(data_source_id),
|
||
"properties": properties,
|
||
}
|
||
|
||
if content_blocks:
|
||
page_data["children"] = content_blocks[:100] # API limit on creation
|
||
|
||
result = notion.pages.create(**page_data)
|
||
|
||
# If more than 100 blocks, append the rest
|
||
if content_blocks and len(content_blocks) > 100:
|
||
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)
|
||
|
||
if not blocks:
|
||
print("No content to write")
|
||
return False
|
||
|
||
if mode == 'replace':
|
||
if not clear_page_content(notion, page_id):
|
||
return False
|
||
|
||
return append_to_page(notion, page_id, blocks)
|
||
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(
|
||
description='Push markdown content to Notion pages or databases',
|
||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||
epilog='''
|
||
Examples:
|
||
# Append markdown file to a page
|
||
python notion_writer.py --page https://notion.so/My-Page-abc123 --file content.md
|
||
|
||
# Replace page content
|
||
python notion_writer.py --page PAGE_ID --file content.md --replace
|
||
|
||
# Pipe content from stdin
|
||
cat content.md | python notion_writer.py --page PAGE_ID --stdin
|
||
|
||
# Create database row with title
|
||
python notion_writer.py --database DB_URL --title "New Entry" --file content.md
|
||
|
||
# Test connection
|
||
python notion_writer.py --test
|
||
'''
|
||
)
|
||
|
||
parser.add_argument('--page', '-p', help='Notion page 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 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)')
|
||
parser.add_argument('--info', action='store_true', help='Show page/database info')
|
||
|
||
args = parser.parse_args()
|
||
|
||
if not NOTION_TOKEN:
|
||
print("Error: NOTION_API_KEY not set in .env file")
|
||
print("Get your integration token from: https://www.notion.so/my-integrations")
|
||
sys.exit(1)
|
||
|
||
notion = compat.make_client(NOTION_TOKEN)
|
||
|
||
# Test connection
|
||
if args.test:
|
||
try:
|
||
me = notion.users.me()
|
||
print("=" * 50)
|
||
print("Notion API Connection Test")
|
||
print("=" * 50)
|
||
print(f"\n✅ Connected successfully!")
|
||
print(f" Bot: {me.get('name', 'Unknown')}")
|
||
print(f" Type: {me.get('type', 'Unknown')}")
|
||
print("\n" + "=" * 50)
|
||
return
|
||
except Exception as e:
|
||
print(f"❌ Connection failed: {e}")
|
||
sys.exit(1)
|
||
|
||
# List accessible content
|
||
if args.list:
|
||
filter_map = {'all': 'all', 'pages': 'page', 'databases': 'database'}
|
||
list_accessible_content(notion, filter_map.get(args.list, 'all'))
|
||
return
|
||
|
||
# Show info
|
||
if args.info:
|
||
if args.page:
|
||
page_id = extract_notion_id(args.page)
|
||
if page_id:
|
||
info = get_page_info(notion, page_id)
|
||
if info:
|
||
print("=" * 50)
|
||
print("Page Information")
|
||
print("=" * 50)
|
||
title = ""
|
||
if 'properties' in info:
|
||
for prop in info['properties'].values():
|
||
if prop.get('type') == 'title':
|
||
title_arr = prop.get('title', [])
|
||
if title_arr:
|
||
title = title_arr[0].get('plain_text', '')
|
||
print(f"Title: {title}")
|
||
print(f"ID: {info['id']}")
|
||
print(f"Created: {info.get('created_time', 'N/A')}")
|
||
print(f"URL: {info.get('url', 'N/A')}")
|
||
return
|
||
|
||
if args.database:
|
||
db_id = extract_notion_id(args.database)
|
||
if db_id:
|
||
info = get_database_info(notion, db_id)
|
||
if info:
|
||
print("=" * 50)
|
||
print("Database Information")
|
||
print("=" * 50)
|
||
title = ""
|
||
if info.get('title'):
|
||
title = info['title'][0].get('plain_text', '') if info['title'] else ''
|
||
print(f"Title: {title}")
|
||
print(f"ID: {info['id']}")
|
||
print(f"Properties: {', '.join(info.get('properties', {}).keys())}")
|
||
return
|
||
|
||
print("Please specify --page or --database with --info")
|
||
return
|
||
|
||
# Get content
|
||
content = None
|
||
if args.stdin:
|
||
content = sys.stdin.read()
|
||
elif args.file:
|
||
file_path = Path(args.file)
|
||
if not file_path.exists():
|
||
print(f"Error: File not found: {args.file}")
|
||
sys.exit(1)
|
||
content = file_path.read_text(encoding='utf-8')
|
||
|
||
# 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)
|
||
|
||
mode = 'replace' if args.replace else 'append'
|
||
print(f"{'Replacing' if mode == 'replace' else 'Appending'} content to page...")
|
||
|
||
if write_to_page(notion, page_id, content, mode):
|
||
print(f"✅ Successfully wrote content to page")
|
||
formatted_id = format_id_with_dashes(page_id)
|
||
print(f" https://notion.so/{formatted_id.replace('-', '')}")
|
||
else:
|
||
print("❌ Failed to write content")
|
||
sys.exit(1)
|
||
return
|
||
|
||
# Create or upsert database row
|
||
if args.database:
|
||
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)
|
||
|
||
# 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)
|
||
|
||
# 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: Data source has no title property; cannot create row.")
|
||
sys.exit(1)
|
||
|
||
# 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, data_source_id, properties, content_blocks)
|
||
|
||
if row_id:
|
||
print(f"✅ Successfully created database row")
|
||
print(f" https://notion.so/{row_id.replace('-', '')}")
|
||
else:
|
||
print("❌ Failed to create database row")
|
||
sys.exit(1)
|
||
return
|
||
|
||
# No action specified
|
||
parser.print_help()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|