Merge pull request #281 from adamtheturtle/no-reupload-files

Make it possible to not change a page if not needed, even if there are files on the page

Author: adamtheturtle

.github/workflows/publish-notion.yml

@@ -34,6 +34,7 @@ jobs:
             --parent-type "page" \
             --file "./build-sample/index.json" \
             --title "Sphinx-Notionbuilder Sample" \
+            --sha-mapping "./sample/notion-sha-mapping.json" \
             --icon "🐍"
         env:
           NOTION_TOKEN: ${{ secrets.NOTION_TOKEN }}

README.rst

@@ -214,14 +214,15 @@ Usage
 
 .. code-block:: console
 
-   $ notion-upload -f path/to/output.json -p parent_page_id -t "Page Title"
+   $ notion-upload --file path/to/output.json --parent-id parent_page_id --parent-type page --title "Page Title" --sha-mapping notion-sha-mapping.json
 
 Arguments:
 
 - ``--file``: Path to the JSON file generated by the Notion builder
 - ``--parent-id``: The ID of the parent page or database in Notion (must be shared with your integration)
 - ``--parent-type``: "page" or "database"
 - ``--title``: Title for the new page in Notion
+- ``--sha-mapping``: Optional JSON file mapping file SHAs to Notion block IDs for efficient file re-uploads (use one file per document). The command will update this file.
 
 The command will create a new page if one with the given title doesn't exist, or update the existing page if one with the given title already exists.
 

pyproject.toml

@@ -35,6 +35,7 @@ dependencies = [
     "beartype>=0.21.0",
     "click>=8.0.0",
     "docutils>=0.21",
+    "notion-client>=2.5.0",
     "sphinx>=8.2.3",
     "sphinx-immaterial>=0.13.7",
     "sphinx-simplepdf>=1.6.0",
@@ -89,6 +90,7 @@ optional-dependencies.sample = [
 urls.Source = "https://github.com/adamtheturtle/sphinx-notionbuilder"
 
 scripts.notion-upload = "_notion_scripts.upload:main"
+scripts.upload-files = "_notion_scripts.upload_files:main"
 
 [tool.setuptools]
 zip-safe = false

sample/index.rst

@@ -289,8 +289,7 @@ The builder supports task lists with checkboxes:
            * [x] Task B2
            * [] Task B3
 
-           A rogue paragraph with a reference to
-           the `parent task_list <task_list_example>`.
+           A rogue paragraph.
 
            - A list item without a checkbox.
            - [ ] Another bullet point.
@@ -1562,8 +1561,7 @@ The builder supports task lists with checkboxes:
            * [x] Task B2
            * [] Task B3
 
-           A rogue paragraph with a reference to
-           the `parent task_list <task_list_example>`.
+           A rogue paragraph.
 
            - A list item without a checkbox.
            - [ ] Another bullet point.

sample/notion-sha-mapping.json

@@ -0,0 +1,6 @@
+{
+  "10be79a09f2b92602ce383a0455f9fe6268f6792decba9383ddce53fded08137": "2819ce7b-60a4-8160-8351-de0105b99d14",
+  "822e7aef5ac028d8cf5565420b59a1f3286a0af6fa17d6f0c617151a1f386507": "2819ce7b-60a4-815a-98ed-ddd4664d67ba",
+  "cd765602a4632f5abb325a66ce59aef506e84fdb6e4fe5590c634f4db567ef2c": "2819ce7b-60a4-816d-b91a-d389313c703c",
+  "eeeca6a7b26d982c4de616bc3bf36b9f40fb090938081cb828709920613dae72": "2819ce7b-60a4-8188-8372-cde9d4d513b3"
+}

spelling_private_dict.txt

@@ -1,3 +1,4 @@
+CLI
 backtick
 backticks
 callout

src/_notion_scripts/upload.py

@@ -3,53 +3,133 @@
 Inspired by https://github.com/ftnext/sphinx-notion/blob/main/upload.py.
 """
 
+import hashlib
 import json
 from enum import Enum
+from functools import cache
 from pathlib import Path
 from typing import TYPE_CHECKING, Any
 from urllib.parse import urlparse
 from urllib.request import url2pathname
+from uuid import UUID
 
 import click
 from beartype import beartype
+from notion_client.errors import APIResponseError
 from ultimate_notion import Emoji, Session
 from ultimate_notion.blocks import PDF as UnoPDF  # noqa: N811
 from ultimate_notion.blocks import Audio as UnoAudio
 from ultimate_notion.blocks import Block
 from ultimate_notion.blocks import Image as UnoImage
 from ultimate_notion.blocks import Video as UnoVideo
-from ultimate_notion.file import UploadedFile
 from ultimate_notion.obj_api.blocks import Block as UnoObjAPIBlock
 
 if TYPE_CHECKING:
     from ultimate_notion.database import Database
     from ultimate_notion.page import Page
 
+_FILE_BLOCK_TYPES = (UnoImage, UnoVideo, UnoAudio, UnoPDF)
+_FileBlock = UnoImage | UnoVideo | UnoAudio | UnoPDF
+
+
+@beartype
+@cache
+def _calculate_file_sha(*, file_path: Path) -> str:
+    """
+    Calculate SHA-256 hash of a file.
+    """
+    sha256_hash = hashlib.sha256()
+    with file_path.open(mode="rb") as f:
+        for chunk in iter(lambda: f.read(4096), b""):
+            sha256_hash.update(chunk)
+    return sha256_hash.hexdigest()
+
 
 @beartype
-def _upload_local_file(
+def _clean_deleted_blocks_from_mapping(
     *,
-    url: str,
+    sha_to_block_id: dict[str, str],
     session: Session,
-) -> UploadedFile | None:
+) -> dict[str, str]:
+    """Remove deleted blocks from SHA mapping.
+
+    Returns a new dictionary with only existing blocks.
+    """
+    cleaned_mapping = sha_to_block_id.copy()
+    deleted_block_shas: set[str] = set()
+
+    for sha, block_id_str in sha_to_block_id.items():
+        block_id = UUID(hex=block_id_str)
+        try:
+            session.api.blocks.retrieve(block=block_id)
+        except APIResponseError:
+            deleted_block_shas.add(sha)
+            msg = f"Block {block_id} does not exist, removing from SHA mapping"
+            click.echo(message=msg)
+
+    for deleted_block_sha in deleted_block_shas:
+        del cleaned_mapping[deleted_block_sha]
+
+    return cleaned_mapping
+
+
+@beartype
+def _find_last_matching_block_index(
+    *,
+    existing_blocks: list[Block] | tuple[Block, ...],
+    local_blocks: list[Block],
+    sha_to_block_id: dict[str, str],
+) -> int | None:
+    """Find the last index where existing blocks match local blocks.
+
+    Returns the last index where blocks are equivalent, or None if no
+    blocks match.
     """
-    Upload a local file and return the uploaded file object.
+    last_matching_index: int | None = None
+    for index, existing_page_block in enumerate(iterable=existing_blocks):
+        if index < len(local_blocks) and (
+            _is_existing_equivalent(
+                existing_page_block=existing_page_block,
+                local_block=local_blocks[index],
+                sha_to_block_id=sha_to_block_id,
+            )
+        ):
+            last_matching_index = index
+        else:
+            break
+    return last_matching_index
+
+
+@beartype
+def _is_existing_equivalent(
+    *,
+    existing_page_block: Block,
+    local_block: Block,
+    sha_to_block_id: dict[str, str],
+) -> bool:
     """
-    parsed = urlparse(url=url)
-    if parsed.scheme != "file":
-        return None
+    Check if a local block is equivalent to an existing page block.
+    """
+    if existing_page_block == local_block:
+        return True
 
-    # Ignore ``mypy`` error as the keyword arguments are different across
-    # Python versions and platforms.
-    file_path = Path(url2pathname(parsed.path))  # type: ignore[misc]
-    with file_path.open(mode="rb") as f:
-        uploaded_file = session.upload(
-            file=f,
-            file_name=file_path.name,
-        )
+    if isinstance(local_block, _FILE_BLOCK_TYPES):
+        parsed = urlparse(url=local_block.url)
+        if parsed.scheme == "file":
+            file_path = Path(url2pathname(parsed.path))  # type: ignore[misc]
+            file_sha = _calculate_file_sha(file_path=file_path)
+            existing_page_block_id_with_file_sha = sha_to_block_id.get(
+                file_sha
+            )
+            if not existing_page_block_id_with_file_sha:
+                return False
+            if (
+                UUID(hex=existing_page_block_id_with_file_sha)
+                == existing_page_block.id
+            ):
+                return True
 
-    uploaded_file.wait_until_uploaded()
-    return uploaded_file
+    return False
 
 
 @beartype
@@ -58,15 +138,26 @@ def _block_from_details(
     details: dict[str, Any],
     session: Session,
 ) -> Block:
-    """Create a Block from a serialized block details.
-
-    Upload any required local files.
+    """
+    Create a Block from a serialized block details.
     """
     block = Block.wrap_obj_ref(UnoObjAPIBlock.model_validate(obj=details))
 
-    if isinstance(block, (UnoImage, UnoVideo, UnoAudio, UnoPDF)):
-        uploaded_file = _upload_local_file(url=block.url, session=session)
-        if uploaded_file is not None:
+    if isinstance(block, _FILE_BLOCK_TYPES):
+        parsed = urlparse(url=block.url)
+        if parsed.scheme == "file":
+            # Ignore ``mypy`` error as the keyword arguments are different
+            # across Python versions and platforms.
+            file_path = Path(url2pathname(parsed.path))  # type: ignore[misc]
+
+            with file_path.open(mode="rb") as file_stream:
+                uploaded_file = session.upload(
+                    file=file_stream,
+                    file_name=file_path.name,
+                )
+
+            uploaded_file.wait_until_uploaded()
+
             return block.__class__(file=uploaded_file, caption=block.caption)
 
     return block
@@ -115,6 +206,20 @@ class _ParentType(Enum):
     help="Icon of the page",
     required=False,
 )
+@click.option(
+    "--sha-mapping",
+    help=(
+        "JSON file mapping file SHAs to Notion block IDs "
+        "(use one file per document)",
+    ),
+    required=False,
+    type=click.Path(
+        exists=True,
+        path_type=Path,
+        file_okay=True,
+        dir_okay=False,
+    ),
+)
 @beartype
 def main(
     *,
@@ -123,12 +228,23 @@ def main(
     parent_type: _ParentType,
     title: str,
     icon: str | None = None,
+    sha_mapping: Path | None = None,
 ) -> None:
     """
     Upload documentation to Notion.
     """
     session = Session()
 
+    sha_mapping_content = (
+        sha_mapping.read_text(encoding="utf-8") if sha_mapping else "{}"
+    )
+    sha_to_block_id: dict[str, str] = dict(json.loads(s=sha_mapping_content))
+
+    sha_to_block_id = _clean_deleted_blocks_from_mapping(
+        sha_to_block_id=sha_to_block_id,
+        session=session,
+    )
+
     blocks = json.loads(s=file.read_text(encoding="utf-8"))
 
     parent: Page | Database
@@ -159,16 +275,15 @@ def main(
         page.icon = Emoji(emoji=icon)
 
     block_objs = [
-        _block_from_details(details=details, session=session)
+        Block.wrap_obj_ref(UnoObjAPIBlock.model_validate(obj=details))
         for details in blocks
     ]
 
-    last_matching_index: int | None = None
-    for index, existing_page_block in enumerate(iterable=page.children):
-        if index < len(blocks) and existing_page_block == block_objs[index]:
-            last_matching_index = index
-        else:
-            break
+    last_matching_index = _find_last_matching_block_index(
+        existing_blocks=page.children,
+        local_blocks=block_objs,
+        sha_to_block_id=sha_to_block_id,
+    )
 
     click.echo(
         message=(
@@ -180,5 +295,42 @@ def main(
     for existing_page_block in page.children[delete_start_index:]:
         existing_page_block.delete()
 
-    page.append(blocks=block_objs[delete_start_index:])
+    block_objs_to_upload = [
+        _block_from_details(details=details, session=session)
+        for details in blocks[delete_start_index:]
+    ]
+    page.append(blocks=block_objs_to_upload)
+
+    if sha_mapping:
+        for uploaded_block_index, uploaded_block in enumerate(
+            iterable=block_objs_to_upload
+        ):
+            if isinstance(uploaded_block, _FILE_BLOCK_TYPES):
+                pre_uploaded_block = block_objs[
+                    delete_start_index + uploaded_block_index
+                ]
+                assert isinstance(pre_uploaded_block, _FILE_BLOCK_TYPES)
+                parsed = urlparse(url=pre_uploaded_block.url)
+                if parsed.scheme == "file":
+                    # Ignore ``mypy`` error as the keyword arguments are
+                    # different across Python versions and platforms.
+                    file_path = Path(url2pathname(parsed.path))  # type: ignore[misc]
+                    file_sha = _calculate_file_sha(file_path=file_path)
+                    sha_to_block_id[file_sha] = str(object=uploaded_block.id)
+
+                    sha_mapping.write_text(
+                        data=json.dumps(
+                            obj=sha_to_block_id, indent=2, sort_keys=True
+                        )
+                        + "\n",
+                        encoding="utf-8",
+                    )
+
+                    click.echo(
+                        message=(
+                            f"Updated SHA mapping for {file_path.name}:"
+                            f"{uploaded_block.id}"
+                        )
+                    )
+
     click.echo(message=f"Updated existing page: '{title}' ({page.url})")