our-claude-skills/custom-skills/32-notion-writer/code/scripts/_notion_compat.py

"""
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