Guanghu Domestic Migration a27e87cb99 chore: import sanitized domestic snapshot for REPO-007
Source snapshot: 97d7f0fae96dc04b7ddad56fc1db6a108ed662cc

[SEC-CLEAN] · pre-push-clean v1.0 · 109处敏感信息已自动转乱码
2026-07-17 15:59:55 +08:00

3.1 KiB
Raw Permalink Blame History

②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 创建完成 → 告诉宝宝,进行下一步 💜