How to sync a Notion page from Markdown and upload local images (agent-safe Python script)

notion_md_sync.py

#!/usr/bin/env python3
"""How to sync a Notion page from local Markdown and attach images.

Replace a Notion page with a local Markdown file and upload relative images
as native Notion files. Built for agents that write the Markdown, then push
it to Notion without wiping comments or child pages.

Python 3.10+, stdlib only.

Usage:
  python3 notion_md_sync.py push PAGE_URL_OR_ID /path/to/post.md
  python3 notion_md_sync.py push PAGE_URL_OR_ID /path/to/post.md --apply
  python3 notion_md_sync.py upload-images PAGE_URL_OR_ID image.png --apply

Safety:
- Dry-run by default. Nothing is written unless you pass --apply.
- --apply refuses when the page still has comments (a full replace drops them).
- Child pages and databases are never deleted.
- Set NOTION_PAT or NOTION_TOKEN in the environment. Never paste tokens into chat.
"""
import argparse
import json
import mimetypes
import os
import re
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
import uuid
from pathlib import Path
from typing import Any, NamedTuple, Sequence


# Authentication and API configuration

API_BASE = "https://api.notion.com"
NOTION_VERSION = "2026-03-11"
HTTP_TIMEOUT_SECONDS = 30
UPLOAD_TIMEOUT_SECONDS = 120
ASYNC_DEADLINE_SECONDS = 180
MAX_POLL_SECONDS = 10.0
MAX_SINGLE_UPLOAD_BYTES = 20 * 1024 * 1024
IMAGE_PLACEHOLDER_BASE = f"{API_BASE}/notion-md-sync-placeholder/"

ACTIVE_ASYNC_STATUSES = {"queued", "running", "retrying"}
TERMINAL_ASYNC_STATUSES = {"succeeded", "failed"}
# Blocks whose children live on another page (or are shared elsewhere) and must
# never be walked when pairing local images to blocks.
OPAQUE_BLOCK_TYPES = {"child_page", "child_database", "synced_block"}
DISCUSSION_MARKERS = ("discussion-urls=", "discussion://")
IMAGE_MIME_TYPES = {
    ".apng": "image/apng",
    ".avif": "image/avif",
    ".bmp": "image/bmp",
    ".gif": "image/gif",
    ".heic": "image/heic",
    ".ico": "image/vnd.microsoft.icon",
    ".jpeg": "image/jpeg",
    ".jpg": "image/jpeg",
    ".png": "image/png",
    ".svg": "image/svg+xml",
    ".tif": "image/tiff",
    ".tiff": "image/tiff",
    ".webp": "image/webp",
}


def get_token() -> str:
    """Read a Notion integration token from the environment."""
    token = (
        os.environ.get("NOTION_PAT", "").strip()
        or os.environ.get("NOTION_TOKEN", "").strip()
    )
    if not token:
        raise RuntimeError(
            "Set NOTION_PAT or NOTION_TOKEN in the environment before using --apply."
        )
    return token


# HTTP

def bounded_request_timeout(
    remaining_seconds: float,
    *,
    max_timeout: int = HTTP_TIMEOUT_SECONDS,
) -> int:
    """Cap an HTTP timeout so a poll cannot outlive a wall-clock deadline."""
    if remaining_seconds <= 0:
        raise RuntimeError(
            f"Notion async task did not finish within "
            f"{ASYNC_DEADLINE_SECONDS} seconds."
        )
    return max(1, min(max_timeout, int(remaining_seconds)))


def _short_message(value: object, token: str = "") -> str:
    message = re.sub(r"\s+", " ", str(value)).strip()
    if token:
        message = message.replace(token, "[redacted]")
    return message[:240] or "request failed"


def _api_error_message(status: int, body: bytes, token: str) -> str:
    code = "api_error"
    message = "request failed"
    try:
        payload = json.loads(body.decode("utf-8"))
    except (UnicodeDecodeError, json.JSONDecodeError):
        payload = None
    if isinstance(payload, dict):
        if isinstance(payload.get("code"), str):
            code = payload["code"]
        if isinstance(payload.get("message"), str):
            message = payload["message"]
    return f"Notion API HTTP {status} {code}: {_short_message(message, token)}"


class NotionClient:
    """Small stdlib-only Notion API client with bounded requests."""

    def __init__(self, token: str) -> None:
        self._token = token

    def request_json(
        self,
        method: str,
        path_or_url: str,
        payload: dict[str, Any] | None = None,
        *,
        timeout: int = HTTP_TIMEOUT_SECONDS,
        allow_http_errors: set[int] | None = None,
    ) -> tuple[int, dict[str, Any]]:
        url = (
            path_or_url
            if path_or_url.startswith("https://")
            else f"{API_BASE}{path_or_url}"
        )
        data = None if payload is None else json.dumps(payload).encode("utf-8")
        headers = {
            "Authorization": f"Bearer {self._token}",
            "Accept": "application/json",
            "Notion-Version": NOTION_VERSION,
        }
        if data is not None:
            headers["Content-Type"] = "application/json"
        request = urllib.request.Request(
            url,
            data=data,
            method=method,
            headers=headers,
        )
        status, body = self._open(
            request,
            timeout,
            allow_http_errors=allow_http_errors,
        )
        try:
            decoded = json.loads(body.decode("utf-8")) if body else {}
        except (UnicodeDecodeError, json.JSONDecodeError) as exc:
            raise RuntimeError(
                f"Notion API HTTP {status} returned invalid JSON."
            ) from exc
        if not isinstance(decoded, dict):
            raise RuntimeError(
                f"Notion API HTTP {status} returned an unexpected response."
            )
        return status, decoded

    def request_multipart(
        self,
        upload_url: str,
        path: Path,
        mime_type: str,
    ) -> dict[str, Any]:
        parsed = urllib.parse.urlsplit(upload_url)
        if parsed.scheme != "https" or parsed.hostname != "api.notion.com":
            raise RuntimeError("Notion returned an unsafe file upload URL.")

        boundary = f"----notion-md-sync-{uuid.uuid4().hex}"
        filename = path.name.replace("\\", "_").replace('"', "_")
        body = b"".join(
            [
                f"--{boundary}\r\n".encode(),
                (
                    'Content-Disposition: form-data; name="file"; '
                    f'filename="{filename}"\r\n'
                ).encode("utf-8"),
                f"Content-Type: {mime_type}\r\n\r\n".encode(),
                path.read_bytes(),
                b"\r\n",
                f"--{boundary}--\r\n".encode(),
            ]
        )
        request = urllib.request.Request(
            upload_url,
            data=body,
            method="POST",
            headers={
                "Authorization": f"Bearer {self._token}",
                "Accept": "application/json",
                "Content-Type": f"multipart/form-data; boundary={boundary}",
                "Notion-Version": NOTION_VERSION,
            },
        )
        status, response_body = self._open(request, UPLOAD_TIMEOUT_SECONDS)
        try:
            decoded = json.loads(response_body.decode("utf-8"))
        except (UnicodeDecodeError, json.JSONDecodeError) as exc:
            raise RuntimeError(
                f"Notion API HTTP {status} returned invalid upload JSON."
            ) from exc
        if not isinstance(decoded, dict):
            raise RuntimeError("Notion returned an unexpected upload response.")
        return decoded

    def _open(
        self,
        request: urllib.request.Request,
        timeout: int,
        *,
        allow_http_errors: set[int] | None = None,
    ) -> tuple[int, bytes]:
        try:
            with urllib.request.urlopen(request, timeout=timeout) as response:
                return response.status, response.read()
        except urllib.error.HTTPError as exc:
            body = exc.read(65536)
            if allow_http_errors and exc.code in allow_http_errors:
                return exc.code, body
            raise RuntimeError(
                _api_error_message(exc.code, body, self._token)
            ) from exc
        except urllib.error.URLError as exc:
            raise RuntimeError(
                "Notion API connection failed: "
                f"{_short_message(exc.reason, self._token)}"
            ) from exc
        except (TimeoutError, OSError) as exc:
            raise RuntimeError(
                f"Notion API request failed: {_short_message(exc, self._token)}"
            ) from exc


# Page IDs

PAGE_ID_RE = re.compile(
    r"(?<![0-9a-fA-F])"
    r"([0-9a-fA-F]{8}(?:-?[0-9a-fA-F]{4}){3}-?[0-9a-fA-F]{12})"
    r"(?![0-9a-fA-F])"
)


def parse_page_id(value: str) -> str:
    """Return a canonical UUID from a Notion page URL or ID."""
    match = PAGE_ID_RE.search(value.strip())
    if not match:
        raise ValueError("Expected a Notion page URL or 32-character page ID.")
    try:
        return str(uuid.UUID(match.group(1)))
    except ValueError as exc:
        raise ValueError("Expected a valid Notion page ID.") from exc


# Markdown preparation

FENCE_RE = re.compile(r"^\s*(`{3,}|~{3,})")
H1_RE = re.compile(r"^\s*#(?:\s+|$)")
IMAGE_RE = re.compile(
    r"!\[[^\]]*\]\(\s*"
    r"(?:<([^>]+)>|([^\s)]+))"
    r"(?:\s+(?:\"[^\"]*\"|'[^']*'|\([^)]*\)))?"
    r"\s*\)"
)


def _fence_marker(line: str) -> tuple[str, int] | None:
    match = FENCE_RE.match(line)
    if not match:
        return None
    marker = match.group(1)
    return marker[0], len(marker)


def strip_html_comments(markdown: str) -> str:
    """Remove HTML comments outside fenced code blocks."""
    output: list[str] = []
    in_comment = False
    fence: tuple[str, int] | None = None

    for line in markdown.splitlines(keepends=True):
        started_in_comment = in_comment
        marker = _fence_marker(line)
        if not in_comment and marker:
            if fence is None:
                fence = marker
            elif marker[0] == fence[0] and marker[1] >= fence[1]:
                fence = None
            output.append(line)
            continue
        if fence is not None:
            output.append(line)
            continue

        remaining = line
        rendered = ""
        while remaining:
            if in_comment:
                end = remaining.find("-->")
                if end < 0:
                    remaining = ""
                    break
                in_comment = False
                remaining = remaining[end + 3 :]
                continue

            start = remaining.find("<!--")
            if start < 0:
                rendered += remaining
                remaining = ""
                break
            rendered += remaining[:start]
            remaining = remaining[start + 4 :]
            in_comment = True

        if started_in_comment and not in_comment and not rendered.strip():
            continue
        if rendered:
            output.append(rendered)
        elif line.endswith(("\n", "\r")) and not in_comment:
            output.append(line[-2:] if line.endswith("\r\n") else line[-1:])
        elif line.endswith(("\n", "\r")) and "<!--" in line:
            output.append(line[-2:] if line.endswith("\r\n") else line[-1:])

    return "".join(output)


def strip_first_leading_h1(markdown: str) -> str:
    """Remove only the first nonblank line when it is an H1."""
    lines = markdown.splitlines(keepends=True)
    for index, line in enumerate(lines):
        if not line.strip():
            continue
        if H1_RE.match(line):
            del lines[index]
        break
    return "".join(lines)


def prepare_markdown(markdown: str) -> str:
    """Prepare Markdown for replacement while preserving fenced examples."""
    return strip_first_leading_h1(strip_html_comments(markdown))


def _markdown_without_fenced_code(markdown: str) -> str:
    lines: list[str] = []
    fence: tuple[str, int] | None = None
    for line in strip_html_comments(markdown).splitlines(keepends=True):
        marker = _fence_marker(line)
        if marker:
            if fence is None:
                fence = marker
            elif marker[0] == fence[0] and marker[1] >= fence[1]:
                fence = None
            lines.append("\n")
            continue
        lines.append(line if fence is None else "\n")
    return "".join(lines)


def extract_image_refs(markdown: str) -> list[str]:
    """Extract Markdown image destinations outside fenced code blocks."""
    searchable = _markdown_without_fenced_code(markdown)
    refs: list[str] = []
    for match in IMAGE_RE.finditer(searchable):
        ref = match.group(1) or match.group(2)
        refs.append(urllib.parse.unquote(ref.strip()))
    return refs


def is_local_image_ref(ref: str) -> bool:
    parsed = urllib.parse.urlsplit(ref)
    return (
        not parsed.scheme
        and not parsed.netloc
        and not ref.startswith(("/", "\\", "#"))
    )


def is_remote_image_ref(ref: str) -> bool:
    parsed = urllib.parse.urlsplit(ref)
    return parsed.scheme in {"http", "https"} and bool(parsed.netloc)


def assert_supported_image_refs(refs: Sequence[str]) -> None:
    """Reject absolute paths and non-http(s) schemes; keep relative or remote URLs."""
    for ref in refs:
        if is_local_image_ref(ref) or is_remote_image_ref(ref):
            continue
        raise ValueError(
            "Unsupported image reference (use a relative path or http(s) URL): "
            f"{ref}"
        )


def replace_local_images_with_placeholders(markdown: str) -> str:
    """Replace local image destinations with temporary HTTPS placeholders."""
    output: list[str] = []
    fence: tuple[str, int] | None = None
    image_number = 0

    def replace_match(match: re.Match[str]) -> str:
        nonlocal image_number
        group_number = 1 if match.group(1) is not None else 2
        ref = match.group(group_number)
        if not is_local_image_ref(ref):
            return match.group(0)
        decoded_ref = urllib.parse.unquote(ref)
        filename = Path(urllib.parse.urlsplit(decoded_ref).path).name or "image"
        placeholder = (
            IMAGE_PLACEHOLDER_BASE
            + urllib.parse.quote(filename, safe="")
            + f"?ref={image_number}"
        )
        image_number += 1
        start, end = match.span(group_number)
        relative_start = start - match.start()
        relative_end = end - match.start()
        return (
            match.group(0)[:relative_start]
            + placeholder
            + match.group(0)[relative_end:]
        )

    for line in markdown.splitlines(keepends=True):
        marker = _fence_marker(line)
        if marker:
            if fence is None:
                fence = marker
            elif marker[0] == fence[0] and marker[1] >= fence[1]:
                fence = None
            output.append(line)
            continue
        output.append(IMAGE_RE.sub(replace_match, line) if fence is None else line)
    return "".join(output)


# Images

class ImageBlock(NamedTuple):
    block_id: str
    url: str
    file_kind: str


def resolve_local_image(ref: str, root: Path) -> Path:
    """Resolve a relative image path and enforce root containment."""
    if not is_local_image_ref(ref):
        raise ValueError(f"Image path must be relative: {ref}")
    parsed = urllib.parse.urlsplit(ref)
    relative = Path(urllib.parse.unquote(parsed.path))
    resolved_root = root.expanduser().resolve()
    resolved = (resolved_root / relative).resolve()
    if not resolved.is_relative_to(resolved_root):
        raise ValueError(f"Image path resolves outside the allowed directory: {ref}")
    if not resolved.is_file():
        raise ValueError(f"Image file not found: {ref}")
    return resolved


def validate_image(path: Path) -> str:
    """Validate an image for Notion single-part upload and return its MIME type."""
    suffix = path.suffix.lower()
    mime_type = IMAGE_MIME_TYPES.get(suffix)
    guessed = mimetypes.guess_type(path.name)[0]
    if not mime_type or (guessed and not guessed.startswith("image/")):
        raise ValueError(f"Unsupported image extension: {suffix or '(none)'}")
    size = path.stat().st_size
    if size >= MAX_SINGLE_UPLOAD_BYTES:
        raise ValueError(
            f"Image must be smaller than 20 MB for single-part upload: {path.name}"
        )
    return mime_type


def _block_url(block: dict[str, Any]) -> tuple[str, str]:
    image = block.get("image")
    if not isinstance(image, dict):
        return "", ""
    kind = image.get("type")
    if not isinstance(kind, str):
        kind = ""
    file_data = image.get(kind)
    if not isinstance(file_data, dict):
        return "", kind
    url = file_data.get("url")
    return (url if isinstance(url, str) else ""), kind


def collect_image_blocks(
    client: NotionClient,
    parent_id: str,
) -> list[ImageBlock]:
    """Recursively collect image blocks on one page, in document order.

    Child pages and databases are not descended into: their images belong to a
    different page and must never be swapped by a push to this one.
    """
    images: list[ImageBlock] = []
    cursor: str | None = None
    while True:
        query = "?page_size=100"
        if cursor:
            query += "&start_cursor=" + urllib.parse.quote(cursor, safe="")
        _, payload = client.request_json(
            "GET",
            f"/v1/blocks/{urllib.parse.quote(parent_id, safe='')}/children{query}",
        )
        results = payload.get("results")
        if not isinstance(results, list):
            raise RuntimeError("Notion block response did not contain results.")
        for block in results:
            if not isinstance(block, dict):
                continue
            block_id = block.get("id")
            if not isinstance(block_id, str):
                continue
            block_type = block.get("type")
            if block_type == "image":
                url, kind = _block_url(block)
                images.append(ImageBlock(block_id, url, kind))
            descend = (
                block.get("has_children") is True
                and block_type not in OPAQUE_BLOCK_TYPES
            )
            if descend:
                images.extend(collect_image_blocks(client, block_id))

        if payload.get("has_more") is not True:
            break
        cursor_value = payload.get("next_cursor")
        if not isinstance(cursor_value, str) or not cursor_value:
            raise RuntimeError("Notion pagination response omitted next_cursor.")
        cursor = cursor_value
    return images


def _filename_stem(value: str) -> str:
    path = urllib.parse.urlsplit(value).path
    return Path(urllib.parse.unquote(path)).stem.casefold()


def _is_replaceable_placeholder(block: ImageBlock) -> bool:
    if not block.url:
        return True
    if block.url.startswith(IMAGE_PLACEHOLDER_BASE):
        return True
    if block.file_kind == "external":
        return True
    host = (urllib.parse.urlsplit(block.url).hostname or "").casefold()
    return not (
        block.file_kind == "file"
        or host.startswith("prod-files-secure.")
        or "notion-static.com" in host
    )


def pair_images_to_blocks(
    paths: Sequence[Path],
    blocks: Sequence[ImageBlock],
    *,
    remote_ref_count: int = 0,
) -> list[tuple[Path, ImageBlock]]:
    """Safely pair local files to image blocks using stems, then strict order."""
    if not paths:
        return []
    if not blocks:
        raise RuntimeError("No image blocks were found on the Notion page.")

    paired: dict[int, int] = {}
    remaining_paths = set(range(len(paths)))
    remaining_blocks = set(range(len(blocks)))

    path_stems: dict[str, list[int]] = {}
    block_stems: dict[str, list[int]] = {}
    for index, path in enumerate(paths):
        path_stems.setdefault(path.stem.casefold(), []).append(index)
    for index, block in enumerate(blocks):
        stem = _filename_stem(block.url)
        if stem:
            block_stems.setdefault(stem, []).append(index)

    for stem, path_indexes in path_stems.items():
        block_indexes = block_stems.get(stem, [])
        if len(path_indexes) == 1 and len(block_indexes) == 1:
            path_index = path_indexes[0]
            block_index = block_indexes[0]
            paired[path_index] = block_index
            remaining_paths.discard(path_index)
            remaining_blocks.discard(block_index)

    if remaining_paths:
        if remote_ref_count:
            raise RuntimeError(
                "Image mapping is ambiguous because local and remote image "
                "references are mixed and filename matching was incomplete."
            )
        ordered_paths = sorted(remaining_paths)
        replaceable = [
            index
            for index in sorted(remaining_blocks)
            if _is_replaceable_placeholder(blocks[index])
        ]
        if len(ordered_paths) != len(replaceable):
            raise RuntimeError(
                "Image mapping is ambiguous. Use unique matching filename stems "
                "or make the local file count equal the replaceable image count."
            )
        paired.update(zip(ordered_paths, replaceable))

    return [(paths[index], blocks[paired[index]]) for index in range(len(paths))]


def upload_file(
    client: NotionClient,
    path: Path,
    mime_type: str,
) -> str:
    """Upload one image to Notion and return its File Upload ID."""
    _, created = client.request_json(
        "POST",
        "/v1/file_uploads",
        {
            "mode": "single_part",
            "filename": path.name,
            "content_type": mime_type,
        },
    )
    upload_id = created.get("id")
    upload_url = created.get("upload_url")
    if not isinstance(upload_id, str) or not isinstance(upload_url, str):
        raise RuntimeError("Notion did not return a usable file upload object.")

    uploaded = client.request_multipart(upload_url, path, mime_type)
    if uploaded.get("status") != "uploaded":
        raise RuntimeError("Notion did not confirm that the file was uploaded.")
    return upload_id


def attach_upload_to_image_block(
    client: NotionClient,
    block_id: str,
    upload_id: str,
) -> None:
    """Swap an existing image block to a Notion-hosted upload."""
    client.request_json(
        "PATCH",
        f"/v1/blocks/{urllib.parse.quote(block_id, safe='')}",
        {"image": {"file_upload": {"id": upload_id}}},
    )


# Push

def count_page_level_comments(client: NotionClient, page_id: str) -> int | None:
    """Count unresolved page-level comments, or None if comment read is denied."""
    status, payload = client.request_json(
        "GET",
        (
            "/v1/comments?block_id="
            + urllib.parse.quote(page_id, safe="")
            + "&page_size=100"
        ),
        allow_http_errors={403},
    )
    if status != 200:
        return None
    results = payload.get("results")
    return len(results) if isinstance(results, list) else 0


def find_inline_comment_anchors(
    client: NotionClient,
    page_id: str,
) -> tuple[bool, str | None]:
    """Look for inline comment anchors in the page markdown export.

    Returns (export_readable, reason). The reason is set only when anchors are
    actually present.
    """
    status, payload = client.request_json(
        "GET",
        f"/v1/pages/{urllib.parse.quote(page_id, safe='')}/markdown",
        allow_http_errors={403},
    )
    if status != 200:
        return False, None
    markdown = payload.get("markdown")
    if isinstance(markdown, str) and any(
        marker in markdown for marker in DISCUSSION_MARKERS
    ):
        return True, "page has inline Notion comment anchors in its markdown export"
    if payload.get("truncated") is True:
        print(
            "warning: the page markdown export was truncated, so the inline "
            "comment check covered only part of the page.",
            file=sys.stderr,
        )
    return True, None


def page_has_blocking_comments(client: NotionClient, page_id: str) -> str | None:
    """Return a refusal reason if the page has comments that replace would drop.

    Two independent checks, because neither alone is sufficient: the Comments
    API with a page ID returns only page-level comments, while inline comments
    anchored to individual blocks show up as discussion markers in the markdown
    export. Both kinds are destroyed by a full replace.
    """
    reasons: list[str] = []

    comment_count = count_page_level_comments(client, page_id)
    if comment_count:
        reasons.append(
            f"page has {comment_count} unresolved page-level Notion comment(s)"
        )
    if comment_count is None:
        print(
            "warning: could not list Notion comments (grant the integration "
            "read comment capability for a stronger check).",
            file=sys.stderr,
        )

    export_readable, anchor_reason = find_inline_comment_anchors(client, page_id)
    if anchor_reason:
        reasons.append(anchor_reason)

    if comment_count is None and not export_readable:
        return (
            "neither the Comments API nor the page markdown export is readable "
            "by this integration, so unresolved comments cannot be ruled out; "
            "refusing replace. Grant the integration read content and read "
            "comment capability, then retry."
        )

    if not reasons:
        return None
    return (
        "; ".join(reasons)
        + ". Refusing replace so comment anchors are not dropped. "
        "Resolve or remove the comments, then retry."
    )


def replace_page_markdown(
    client: NotionClient,
    page_id: str,
    markdown: str,
) -> None:
    """Replace page content and wait for any asynchronous task."""
    status, response = client.request_json(
        "PATCH",
        f"/v1/pages/{urllib.parse.quote(page_id, safe='')}/markdown",
        {
            "type": "replace_content",
            "replace_content": {
                "new_str": markdown,
                "allow_deleting_content": False,
            },
            "allow_async": True,
        },
    )
    if status == 202 or response.get("object") == "async_task":
        poll_async_task(client, response)


def poll_async_task(
    client: NotionClient,
    task: dict[str, Any],
) -> dict[str, Any]:
    """Poll a Notion async task to a known terminal status."""
    task_id = task.get("id")
    if not isinstance(task_id, str) or not task_id:
        raise RuntimeError("Notion async response did not include a task ID.")
    deadline = time.monotonic() + ASYNC_DEADLINE_SECONDS
    current = task

    while True:
        status = current.get("status")
        if status == "succeeded":
            return current
        if status == "failed":
            error = current.get("error")
            if isinstance(error, dict):
                code = error.get("code", "async_failed")
                message = error.get("message", "Markdown replacement failed")
                raise RuntimeError(
                    f"Notion async task failed {code}: {_short_message(message)}"
                )
            raise RuntimeError("Notion async task failed.")
        if status not in ACTIVE_ASYNC_STATUSES:
            known = ", ".join(sorted(ACTIVE_ASYNC_STATUSES | TERMINAL_ASYNC_STATUSES))
            raise RuntimeError(
                f"Notion async task returned unknown status {status!r}. "
                f"Known statuses: {known}."
            )

        poll_after = current.get("poll_after_seconds", 1)
        if not isinstance(poll_after, (int, float)) or poll_after < 0:
            poll_after = 1
        delay = min(max(float(poll_after), 0.1), MAX_POLL_SECONDS)
        remaining = deadline - time.monotonic()
        if remaining <= 0:
            raise RuntimeError(
                f"Notion async task did not finish within "
                f"{ASYNC_DEADLINE_SECONDS} seconds."
            )
        time.sleep(min(delay, remaining))
        remaining_after_sleep = deadline - time.monotonic()
        if remaining_after_sleep <= 0:
            raise RuntimeError(
                f"Notion async task did not finish within "
                f"{ASYNC_DEADLINE_SECONDS} seconds."
            )
        _, current = client.request_json(
            "GET",
            f"/v1/async_tasks/{urllib.parse.quote(task_id, safe='')}",
            timeout=bounded_request_timeout(remaining_after_sleep),
        )


def _upload_and_attach(
    client: NotionClient,
    pairs: Sequence[tuple[Path, ImageBlock]],
) -> None:
    upload_cache: dict[Path, str] = {}
    for path, block in pairs:
        upload_id = upload_cache.get(path)
        if upload_id is None:
            upload_id = upload_file(client, path, validate_image(path))
            upload_cache[path] = upload_id
        attach_upload_to_image_block(client, block.block_id, upload_id)
        print(f"Attached {path.name} to image block {block.block_id}.")


def run_push(args: argparse.Namespace) -> int:
    page_id = parse_page_id(args.page)
    post = args.post.expanduser().resolve()
    if not post.is_file() or post.suffix.lower() not in {".md", ".markdown"}:
        raise RuntimeError(f"Markdown post not found: {post}")
    try:
        source = post.read_text(encoding="utf-8")
    except UnicodeDecodeError as exc:
        raise RuntimeError(f"Markdown post is not valid UTF-8: {post}") from exc

    prepared = prepare_markdown(source)
    all_refs = extract_image_refs(prepared)
    assert_supported_image_refs(all_refs)
    local_refs = [ref for ref in all_refs if is_local_image_ref(ref)]
    local_paths = [resolve_local_image(ref, post.parent) for ref in local_refs]
    for path in local_paths:
        validate_image(path)
    remote_refs = [ref for ref in all_refs if is_remote_image_ref(ref)]
    remote_ref_count = len(remote_refs)
    if remote_refs:
        local_stems = [path.stem.casefold() for path in local_paths]
        remote_stems = {_filename_stem(ref) for ref in remote_refs}
        if len(local_stems) != len(set(local_stems)) or any(
            stem in remote_stems for stem in local_stems
        ):
            raise RuntimeError(
                "Mixed local and remote images require unique, non-overlapping "
                "local filename stems."
            )
    markdown = replace_local_images_with_placeholders(prepared)

    if not args.apply:
        print("Dry run. No Notion data was changed.")
        print(f"Page: {page_id}")
        print(f"Post: {post}")
        print(f"Prepared Markdown characters: {len(markdown)}")
        print(f"Local image references: {len(local_paths)}")
        print(f"Remote image references left unchanged: {remote_ref_count}")
        print("Run again with --apply to replace the page.")
        print(
            "Note: --apply refuses to replace if the page has unresolved "
            "Notion comments."
        )
        return 0

    client = NotionClient(get_token())
    comment_block = page_has_blocking_comments(client, page_id)
    if comment_block:
        raise RuntimeError(comment_block)

    replace_page_markdown(client, page_id, markdown)
    print("Replaced page Markdown.")

    if local_paths:
        blocks = collect_image_blocks(client, page_id)
        pairs = pair_images_to_blocks(
            local_paths,
            blocks,
            remote_ref_count=remote_ref_count,
        )
        _upload_and_attach(client, pairs)
    return 0


# upload-images

def run_upload_images(args: argparse.Namespace) -> int:
    page_id = parse_page_id(args.page)
    root = Path.cwd().resolve()
    paths = [resolve_local_image(str(path), root) for path in args.images]
    for path in paths:
        validate_image(path)

    if not args.apply:
        print("Dry run. No Notion data was changed.")
        print(f"Page: {page_id}")
        print(f"Image root: {root}")
        for path in paths:
            print(f"Would upload: {path.relative_to(root)}")
        print("Run again with --apply to inspect blocks, map images, and upload.")
        return 0

    client = NotionClient(get_token())
    blocks = collect_image_blocks(client, page_id)
    pairs = pair_images_to_blocks(paths, blocks)
    _upload_and_attach(client, pairs)
    return 0


# Entrypoint

def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description=(
            "Replace a Notion page from a local Markdown file and attach "
            "local images. Agents: run this script with python3 and args."
        )
    )
    subparsers = parser.add_subparsers(dest="command", required=True)

    push = subparsers.add_parser(
        "push",
        help="Replace a page from a local Markdown post",
    )
    push.add_argument("page", help="Notion page URL or page ID")
    push.add_argument("post", type=Path, help="Local .md or .markdown post")
    push.add_argument(
        "--apply",
        action="store_true",
        help="Perform the replacement and uploads",
    )
    push.set_defaults(handler=run_push)

    upload = subparsers.add_parser(
        "upload-images",
        help="Upload local images onto existing image blocks",
    )
    upload.add_argument("page", help="Notion page URL or page ID")
    upload.add_argument(
        "images",
        nargs="+",
        type=Path,
        help="Images relative to the current directory",
    )
    upload.add_argument(
        "--apply",
        action="store_true",
        help="Perform the uploads and block updates",
    )
    upload.set_defaults(handler=run_upload_images)
    return parser


def parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace:
    return build_parser().parse_args(argv)


def main(argv: Sequence[str] | None = None) -> int:
    args = parse_args(argv)
    try:
        return args.handler(args)
    except (RuntimeError, ValueError) as exc:
        print(f"error: {exc}", file=sys.stderr)
        return 1


if __name__ == "__main__":
    raise SystemExit(main())