Notion Integrations 开发:实现一个自动格式化工具
背景与目标
Notion 没有内建“中文与英文/数字之间自动加空格”的格式化能力(飞书文档有类似的小组件)。为了把排版这件事自动化,我用 Notion 的 Integrations + API 写了一个小工具:扫描页面的文本块,把需要加空格的位置统一修正,并回写到 Notion。
工具的目标很明确:
- 在 中文 ↔ 英文/数字 之间自动补空格
- 处理标题与正文(标题是 page property,不在 blocks 里)
- 不处理 code block(避免破坏代码)
- 能处理嵌套 block(toggle/quote/列表等)
- 兼顾工程细节:限频、冲突、失败重试
Notion Integration 是什么
Notion 把“外部程序访问 workspace 数据”的能力封装为 Integration(连接)。它需要用户显式授权才能访问具体页面/数据库。

对这个小工具来说,我们只需要三件事:鉴权、读取页面内容、更新页面内容。
1)创建集成并授权页面
在 我的集成 创建一个新集成,勾选:
- 读取内容(Read content)
- 更新内容(Update content)
- 插入内容(Insert content,视需求而定)
保存后会生成一个 内部集成密钥(integration token),它是 API 鉴权凭证。

然后把你要处理的页面“分享”给这个集成:
- 页面右上角菜单 → 连接(Connections)
- 搜索你的集成名称并添加
注意:Notion 的授权是“页面级”的。想处理一个父页面下的子页面,可以只给父页面授权,子页面会继承权限。
2)准备环境与依赖
Notion 提供了官方 API 与多语言 SDK。这里用 Python:
pip install notion-client
3)核心:文本格式化函数
格式化规则可以先收敛到最小可用:在中文与英文/数字之间补空格。
import re
def format_text(text: str) -> str:
# 中文后跟英文/数字:补空格
text = re.sub(r"([\u4e00-\u9fa5])([A-Za-z0-9])", r"\1 \2", text)
# 英文/数字后跟中文:补空格
text = re.sub(r"([A-Za-z0-9])([\u4e00-\u9fa5])", r"\1 \2", text)
return text
这一段很短,但决定了效果边界:如果你后续还想处理标点、括号、全角半角等,也应该以“规则可解释、可回滚”为前提逐步扩展。
4)读取页面:标题与正文是两套入口
Notion 文档是 block 化的,但页面标题不在 blocks 里,而在 page.properties 里(property 名字可能是 title、Name 等)。
4.1 连接鉴权
建议把 token 放到环境变量里:
import os
from notion_client import AsyncClient
NOTION_TOKEN = os.environ["NOTION_TOKEN"]
notion = AsyncClient(auth=NOTION_TOKEN)
4.2 读取并更新标题(page properties)
page = await notion.pages.retrieve(page_id=page_id)
title_prop = None
for prop_name, prop_value in page["properties"].items():
if prop_value["type"] == "title":
title_prop = prop_name
break
找到 title_prop 后,你就能拿到标题的 rich_text 并做同样的格式化处理,再用 pages.update 回写(完整实现见仓库)。
4.3 读取正文(blocks + 递归 children)
正文需要通过 blocks API 拉取,且要处理嵌套:
block = await notion.blocks.retrieve(block_id=block_id)
children = await notion.blocks.children.list(block_id=block_id)
# 递归处理 children(toggle、列表、引用等都可能带 children)
你不需要把所有 block 类型背下来。实践里更可行的做法是:
- 只处理“确实包含
rich_text的文本块” - 明确跳过
code等不应修改的块
5)修改正文:更新 rich_text
Notion 的更新基本就是把目标 block 的 rich_text 改写回去:
await notion.blocks.update(
block_id=block_id,
**{block_type: {"rich_text": new_rich_text}},
)
这里关键不是 update 调用本身,而是你如何在遍历时构造 new_rich_text:建议只修改 plain_text 的表现,不要动链接、标注(bold/italic)、mention 等结构信息(完整实现见仓库)。
6)工程化:限频、并发与冲突重试
Notion 有明确的 API 限频策略(以文档中常见的口径为例):
- 每个集成:3 requests/second
- 每个工作区:30 requests/second
另外,页面内容可能被用户或其他自动化同时修改。实践中常见现象是:你读到的 block 状态过期,更新时出现冲突或失败。
因此实现里建议至少包含三件事:
- 批量并发上限:并发数控制在 3 左右
- 更新前拉最新状态:用
last_edited_time判断是否发生变化 - 失败重试:指数退避(或递增退避)+ 上限次数,最终进入失败队列
示例(核心逻辑):
async def update_single_block(self, block: dict, retry_count: int = 3) -> bool:
for attempt in range(retry_count):
try:
current_block = await notion.blocks.retrieve(block_id=block["block_id"])
if current_block.get("last_edited_time", "") != block["last_edited_time"]:
content = current_block[block["block_type"]]["rich_text"]
updated_content, modified = self.process_rich_text(content)
if not modified:
return True
block["content"] = updated_content
await notion.blocks.update(
block_id=block["block_id"],
**{block["block_type"]: {"rich_text": block["content"]}},
)
return True
except Exception as e:
if attempt < retry_count - 1:
wait_time = (attempt + 1) * 0.5
await asyncio.sleep(wait_time)
else:
return False
批量更新(并发 3 + 节流):
async def update_blocks(self, batch_size: int = 3):
total = len(self.blocks_to_update)
for i in range(0, total, batch_size):
batch = self.blocks_to_update[i : i + batch_size]
results = await asyncio.gather(*[self.update_single_block(b) for b in batch])
self.failed_blocks.extend([b for b, ok in zip(batch, results) if not ok])
await asyncio.sleep(0.5)
使用效果
初始文档:

执行格式化:

最终效果:

标题与文本块都被更新;code block 不做处理,符合预期。
完整代码
完整实现见 GitHub 仓库。