magnus919_agent-skills/notion/scripts/notion-cli

#!/usr/bin/env python3
"""notion-cli - bounded, agent-first command line for the Notion API.

Reads Notion pages and databases and, with explicit confirmation, creates or
updates pages over HTTPS using only the Python standard library. Covers pages,
databases (query), search, and page updates.

Design rules:

- Read-only by default. Every state-changing command (creating or updating a
  page) is a guarded mutation: it requires --dry-run to preview, then --yes to
  confirm. Mutation requires explicit confirmation.
- Bounded reads: search and database queries cap results with --limit and
  never page past the requested cap.
- --json emits machine-readable JSON; the default is human-readable text.
- --help works with no NOTION_TOKEN set and makes no network calls.

Environment:
  NOTION_TOKEN         Notion integration token (secret_...)
  NOTION_VERSION       API version header (default: 2022-06-28)

Exit codes: 0 success, 1 Notion API error or failed check, 2 usage error.
"""
import argparse
import json
import os
import sys
import urllib.error
import urllib.request
from typing import Any, Dict, List, Optional

API_BASE = os.environ.get("NOTION_API_BASE", "https://api.notion.com/v1")
DEFAULT_VERSION = "2022-06-28"
DEFAULT_LIMIT = 20
MAX_LIMIT = 100
REQUEST_TIMEOUT = 15
TEXT_TRUNCATE = 500


class NotionError(Exception):
    """Raised when the Notion API returns an error or transport fails."""


def get_token() -> str:
    token = os.environ.get("NOTION_TOKEN", "")
    if not token:
        raise NotionError("NOTION_TOKEN environment variable is not set")
    return token


def headers(token: str) -> Dict[str, str]:
    return {
        "Authorization": f"Bearer {token}",
        "Notion-Version": os.environ.get("NOTION_VERSION", DEFAULT_VERSION),
        "Content-Type": "application/json",
    }


def api_request(method: str, path: str, token: str, body: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
    """Send a JSON request to a Notion API path and return the payload."""
    data = json.dumps(body).encode("utf-8") if body is not None else None
    request = urllib.request.Request(
        f"{API_BASE}/{path.lstrip('/')}",
        data=data,
        headers=headers(token),
        method=method,
    )
    try:
        with urllib.request.urlopen(request, timeout=REQUEST_TIMEOUT) as response:
            return json.loads(response.read().decode("utf-8"))
    except urllib.error.HTTPError as error:
        detail = ""
        try:
            detail = json.loads(error.read().decode("utf-8")).get("message", "")
        except (json.JSONDecodeError, UnicodeDecodeError):
            pass
        raise NotionError(f"Notion API HTTP {error.code}: {detail or error.reason}") from error
    except urllib.error.URLError as error:
        raise NotionError(f"Notion API unreachable: {error.reason}") from error
    except json.JSONDecodeError as error:
        raise NotionError(f"Notion API returned non-JSON: {error}") from error


def truncate(text: str, limit: int = TEXT_TRUNCATE) -> str:
    if len(text) <= limit:
        return text
    return text[: limit - 1] + "…"


def page_title(page: Dict[str, Any]) -> str:
    """Best-effort title extraction from a Notion page object."""
    properties = page.get("properties", {})
    for key in ("title", "Name"):
        prop = properties.get(key)
        if not prop:
            continue
        if prop.get("type") == "title":
            parts = prop.get("title", [])
            if parts:
                return truncate(parts[0].get("plain_text", ""))
    return page.get("id", "")


def summarize_page(page: Dict[str, Any]) -> Dict[str, Any]:
    return {
        "id": page.get("id", ""),
        "title": page_title(page),
        "url": page.get("url", ""),
        "object": page.get("object", "page"),
        "last_edited_time": page.get("last_edited_time", ""),
        "created_time": page.get("created_time", ""),
    }


def summarize_database(database: Dict[str, Any]) -> Dict[str, Any]:
    title_parts = database.get("title", [])
    title = truncate("".join(p.get("plain_text", "") for p in title_parts)) if title_parts else ""
    return {
        "id": database.get("id", ""),
        "title": title,
        "url": database.get("url", ""),
        "last_edited_time": database.get("last_edited_time", ""),
    }


# --------------------------------------------------------------------------
# Command implementations
# --------------------------------------------------------------------------


def cmd_pages_get(args: argparse.Namespace, token: str) -> Dict[str, Any]:
    page = api_request("GET", f"pages/{args.page_id}", token)
    return {"ok": True, "page": summarize_page(page)}


def cmd_pages_create(args: argparse.Namespace, token: str) -> Dict[str, Any]:
    if not args.dry_run and not args.yes:
        raise NotionError(
            "refusing to create a page without confirmation: pass --dry-run to "
            "preview or --yes to confirm the mutation"
        )
    parent = {"type": "page_id", "page_id": args.parent_page} if args.parent_page \
        else {"type": "database_id", "database_id": args.parent_database}
    body: Dict[str, Any] = {"parent": parent, "properties": {}}
    if args.title:
        body["properties"]["title"] = {"title": [{"text": {"content": args.title}}]}
    if args.dry_run:
        return {"ok": True, "dry_run": True, "would_create": body}
    page = api_request("POST", "pages", token, body)
    return {"ok": True, "page": summarize_page(page)}


def cmd_pages_update(args: argparse.Namespace, token: str) -> Dict[str, Any]:
    if not args.dry_run and not args.yes:
        raise NotionError(
            "refusing to update a page without confirmation: pass --dry-run to "
            "preview or --yes to confirm the mutation"
        )
    try:
        with open(args.properties, "r", encoding="utf-8") as handle:
            properties = json.load(handle)
    except OSError as error:
        raise NotionError(f"cannot read properties file {args.properties}: {error}") from error
    except json.JSONDecodeError as error:
        raise NotionError(f"properties file is not valid JSON: {error}") from error
    if not isinstance(properties, dict):
        raise NotionError("properties file must contain a JSON object of property values")
    body = {"properties": properties}
    if args.dry_run:
        return {"ok": True, "dry_run": True, "would_update": {"page_id": args.page_id, "properties": properties}}
    page = api_request("PATCH", f"pages/{args.page_id}", token, body)
    return {"ok": True, "page": summarize_page(page)}


def cmd_databases_query(args: argparse.Namespace, token: str) -> Dict[str, Any]:
    body: Dict[str, Any] = {"page_size": args.limit}
    if args.filter:
        try:
            with open(args.filter, "r", encoding="utf-8") as handle:
                body["filter"] = json.load(handle)
        except OSError as error:
            raise NotionError(f"cannot read filter file {args.filter}: {error}") from error
        except json.JSONDecodeError as error:
            raise NotionError(f"filter file is not valid JSON: {error}") from error
    payload = api_request("POST", f"databases/{args.database_id}/query", token, body)
    pages = [summarize_page(p) for p in payload.get("results", [])]
    return {"ok": True, "database_id": args.database_id, "pages": pages,
            "has_more": payload.get("has_more", False)}


def cmd_search(args: argparse.Namespace, token: str) -> Dict[str, Any]:
    payload = api_request("POST", "search", token, {"query": args.query, "page_size": args.limit})
    results = []
    for item in payload.get("results", []):
        if item.get("object") == "database":
            results.append({"object": "database", **summarize_database(item)})
        else:
            results.append({"object": "page", **summarize_page(item)})
    return {"ok": True, "query": args.query, "results": results,
            "has_more": payload.get("has_more", False)}


# --------------------------------------------------------------------------
# Output helpers
# --------------------------------------------------------------------------


def emit(data: Dict[str, Any], json_mode: bool) -> int:
    if json_mode:
        print(json.dumps(data, indent=2, sort_keys=True))
    else:
        _emit_human(data)
    return 0


def _emit_human(data: Dict[str, Any]) -> None:
    if "page" in data and "dry_run" not in data:
        page = data["page"]
        print(f"{page['title']}  <{page['id']}>  {page['url']}")
    elif "pages" in data:
        print(f"database <{data['database_id']}> (has_more={data['has_more']}):")
        for page in data["pages"]:
            print(f"  {page['title']}  <{page['id']}>")
    elif "results" in data:
        print(f"search '{data['query']}' (has_more={data['has_more']}):")
        for result in data["results"]:
            print(f"  [{result['object']}] {result['title']}  <{result['id']}>")
    elif data.get("dry_run"):
        if "would_create" in data:
            print("DRY RUN (nothing created):")
            print(json.dumps(data["would_create"], indent=2))
        else:
            print("DRY RUN (nothing updated):")
            print(json.dumps(data["would_update"], indent=2))
    else:
        print(json.dumps(data, indent=2, sort_keys=True))


# --------------------------------------------------------------------------
# CLI
# --------------------------------------------------------------------------


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="notion-cli",
        description=(
            "Bounded, agent-first CLI for the Notion API: pages, database "
            "queries, search, and guarded page updates. Read-only by default; "
            "create/update require --dry-run then --yes."
        ),
    )
    parser.add_argument("--json", action="store_true", help="emit machine-readable JSON output")
    parser.add_argument("--limit", type=int, default=DEFAULT_LIMIT, metavar="N",
                        help=f"cap reads at N results (default {DEFAULT_LIMIT}, max {MAX_LIMIT})")
    sub = parser.add_subparsers(dest="command", required=True)

    pages = sub.add_parser("pages", help="get, create, or update pages")
    page_sub = pages.add_subparsers(dest="action", required=True)

    page_get = page_sub.add_parser("get", help="retrieve a page (read-only)")
    page_get.add_argument("--page-id", required=True, help="page ID (32-hex or UUID form)")

    page_create = page_sub.add_parser("create", help="create a page (guarded mutation)")
    page_create.add_argument("--parent-page", help="parent page ID for the new page")
    page_create.add_argument("--parent-database", help="parent database ID for the new page")
    page_create.add_argument("--title", help="page title text")
    page_create.add_argument("--dry-run", action="store_true", help="preview the payload without creating")
    page_create.add_argument("--yes", action="store_true", help="confirm the mutation and create")

    page_update = page_sub.add_parser("update", help="update page properties (guarded mutation)")
    page_update.add_argument("--page-id", required=True, help="page ID to update")
    page_update.add_argument("--properties", required=True,
                             help="path to a JSON file of property values")
    page_update.add_argument("--dry-run", action="store_true", help="preview the payload without updating")
    page_update.add_argument("--yes", action="store_true", help="confirm the mutation and update")

    databases = sub.add_parser("databases", help="query a database (read-only)")
    db_sub = databases.add_subparsers(dest="action", required=True)
    db_query = db_sub.add_parser("query", help="query database pages")
    db_query.add_argument("--database-id", required=True, help="database ID to query")
    db_query.add_argument("--filter", help="path to a JSON filter object")

    search = sub.add_parser("search", help="search pages and databases (read-only)")
    search.add_argument("action", nargs="?", default="query", choices=["query"])
    search.add_argument("--query", required=True, help="search query text")

    return parser


def main(argv: Optional[List[str]] = None) -> int:
    parser = build_parser()
    args = parser.parse_args(argv)
    if args.limit < 1 or args.limit > MAX_LIMIT:
        parser.error(f"--limit must be between 1 and {MAX_LIMIT}")
    try:
        token = get_token()
        if args.command == "pages":
            if args.action == "get":
                result = cmd_pages_get(args, token)
            elif args.action == "create":
                result = cmd_pages_create(args, token)
            else:
                result = cmd_pages_update(args, token)
        elif args.command == "databases":
            result = cmd_databases_query(args, token)
        elif args.command == "search":
            result = cmd_search(args, token)
        else:  # pragma: no cover - argparse prevents this
            parser.error(f"unknown command: {args.command}")
        return emit(result, args.json)
    except NotionError as error:
        if args.json:
            print(json.dumps({"ok": False, "error": str(error)}, indent=2, sort_keys=True))
        else:
            print(f"notion-cli: {error}", file=sys.stderr)
        return 1


if __name__ == "__main__":
    sys.exit(main())