Source snapshot: 97d7f0fae96dc04b7ddad56fc1db6a108ed662cc [SEC-CLEAN] · pre-push-clean v1.0 · 109处敏感信息已自动转乱码
3.1 KiB
3.1 KiB
②notion_writer.py · Notion写入工具 · 2026-05-21
HLDP://juzi/tools/notion_writer/2026-05-21
├── 作用: 自动把拆书结果写入 Notion 页面
├── 操作: 整段复制粘贴到终端,回车
└── 看到「✅ notion_writer.py 创建完成」就成功了
复制这整段到终端,回车
cat > ~/chenxing-aircraft/tools/notion_writer.py << 'ENDOFFILE'
#!/usr/bin/env python3
"""
Notion 写入工具 · notion_writer.py
"""
import os
import requests
NOTION_API_KEY = os.environ.get("NOTION_API_KEY", "")
NOTION_VERSION = "2022-06-28"
BASE_URL = "https://api.notion.com/v1"
def _headers():
return {
"Authorization": f"Bearer {NOTION_API_KEY}",
"Content-Type": "application/json",
"Notion-Version": NOTION_VERSION,
}
def _md_to_blocks(text: str) -> list:
"""Markdown 转 Notion blocks(简化版)"""
blocks = []
for line in text.split("\n"):
if not line.strip():
continue
if line.startswith("# "):
blocks.append({"type":"heading_1","heading_1":{"rich_text":[{"type":"text","text":{"content":line[2:100]}}]}})
elif line.startswith("## "):
blocks.append({"type":"heading_2","heading_2":{"rich_text":[{"type":"text","text":{"content":line[3:100]}}]}})
elif line.startswith("### "):
blocks.append({"type":"heading_3","heading_3":{"rich_text":[{"type":"text","text":{"content":line[4:100]}}]}})
elif line.startswith("- "):
blocks.append({"type":"bulleted_list_item","bulleted_list_item":{"rich_text":[{"type":"text","text":{"content":line[2:2000]}}]}})
else:
blocks.append({"type":"paragraph","paragraph":{"rich_text":[{"type":"text","text":{"content":line[:2000]}}]}})
return blocks
def create_notion_page(title: str, content: str, parent_page_id: str) -> bool:
"""在指定父页面下创建新页面"""
if not NOTION_API_KEY:
print("⚠️ 未配置 NOTION_API_KEY,跳过写入Notion")
return False
blocks = _md_to_blocks(content)
data = {
"parent": {"page_id": parent_page_id},
"properties": {"title": {"title": [{"type": "text", "text": {"content": title[:255]}}]}},
"children": blocks[:100]
}
try:
r = requests.post(f"{BASE_URL}/pages", headers=_headers(), json=data, timeout=30)
if r.status_code in [200, 201]:
page_id = r.json()["id"]
if len(blocks) > 100:
for i in range(100, len(blocks), 100):
requests.patch(
f"{BASE_URL}/blocks/{page_id}/children",
headers=_headers(),
json={"children": blocks[i:i+100]},
timeout=30
)
return True
else:
print(f"Notion API 错误 {r.status_code}: {r.text[:300]}")
return False
except Exception as e:
print(f"请求失败: {e}")
return False
ENDOFFILE
echo "✅ notion_writer.py 创建完成"
看到
✅ notion_writer.py 创建完成→ 告诉宝宝,进行下一步 💜