guanghulab/zhuyuan-expert-pack/v1.0/notion-bridge.py

#!/usr/bin/env python3
"""
notion-bridge.py · 冰朔本地 Notion 直连工具
铸渊 ICE-GL-ZY001 开发 · D164+ · 解决"换房间断手"问题

之前 WorkBuddy 里的铸渊通过某种方式连 Notion（应该有连接器）。
今天 MiniMax Code 房间的铸渊——用铸渊自己的工具重写。

用法:
  # 1. 设置凭证
  export NOTION_API_KEY="ntn_xxx"

  # 2. 列出 workspace 内的对象
  python3 notion-bridge.py search

  # 3. 列出 workspace 内的数据库
  python3 notion-bridge.py databases

  # 4. 读取某个数据库的全部行
  python3 notion-bridge.py query <database_id>

  # 5. 读取某个页面
  python3 notion-bridge.py page <page_id>

  # 6. 创建/更新页面
  python3 notion-bridge.py create-page <parent_id> <title>
  python3 notion-bridge.py update-page <page_id> <properties_json>

  # 7. 导出页面为 markdown
  python3 notion-bridge.py export <page_id>

前置:
  1. 冰朔生成 Notion integration: https://www.notion.so/my-integrations
  2. token 形如 ntn_xxx (或 secret_xxx)
  3. 在 Notion 工作区里把页面"分享"给 integration
"""

import os
import sys
import json
from urllib.parse import urlencode, quote
from urllib.request import Request, urlopen
from urllib.error import HTTPError, URLError

# ============ 配置 ============

NOTION_API_KEY = os.environ.get("NOTION_API_KEY", "").strip()
NOTION_VERSION = os.environ.get("NOTION_VERSION", "2022-06-28")
NOTION_BASE = "https://api.notion.com/v1"


# ============ API 调用 ============

def api(method, path, payload=None):
    """通用 Notion API 请求"""
    if not NOTION_API_KEY:
        raise RuntimeError(
            "缺少 NOTION_API_KEY 环境变量。\n"
            "冰朔请到 https://www.notion.so/my-integrations 生成 integration token\n"
            "然后: export NOTION_API_KEY=ntn_xxx"
        )

    url = f"{NOTION_BASE}{path}"
    headers = {
        "Authorization": f"Bearer {NOTION_API_KEY}",
        "Notion-Version": NOTION_VERSION,
        "Content-Type": "application/json",
    }

    data = json.dumps(payload).encode() if payload else None
    req = Request(url, data=data, headers=headers, method=method)

    try:
        with urlopen(req, timeout=30) as resp:
            body = resp.read().decode()
            return json.loads(body) if body else {}
    except HTTPError as e:
        body = e.read().decode() if e.fp else ""
        try:
            err = json.loads(body)
        except Exception:
            err = {"message": body[:500]}
        raise RuntimeError(
            f"Notion API {method} {path} 失败 [{e.code}]\n"
            f"  message: {err.get('message', '?')}\n"
            f"  code: {err.get('code', '?')}"
        )
    except URLError as e:
        raise RuntimeError(f"网络失败: {e.reason}")


# ============ 命令 ============

def cmd_search(args):
    """搜索页面/数据库"""
    query = args[0] if args else None
    payload = {
        "page_size": 50,
        "sort": {"direction": "descending", "timestamp": "last_edited_time"},
    }
    if query:
        payload["query"] = query

    result = api("POST", "/search", payload)
    results = result.get("results", [])
    print(f"找到 {len(results)} 个对象\n")

    for r in results:
        title = "?"
        obj_type = r.get("object", "?")
        if obj_type == "database":
            title_arr = r.get("title", [])
            if title_arr:
                title = title_arr[0].get("plain_text", "?")
        elif obj_type == "page":
            for prop_name, prop in r.get("properties", {}).items():
                if prop.get("type") == "title":
                    title_arr = prop.get("title", [])
                    if title_arr:
                        title = title_arr[0].get("plain_text", "?")
                    break

        last_edited = r.get("last_edited_time", "?")[:10]
        icon = r.get("icon", {})
        icon_str = icon.get("emoji", "") if icon else ""
        rid = r.get("id", "?")
        print(f"  [{obj_type}] {icon_str} {title}")
        print(f"    id: {rid}  ({last_edited})")
        print()


def cmd_databases(args):
    """列出所有数据库"""
    payload = {
        "page_size": 100,
        "filter": {"property": "object", "value": "database"},
        "sort": {"direction": "descending", "timestamp": "last_edited_time"},
    }
    result = api("POST", "/search", payload)
    results = result.get("results", [])
    print(f"找到 {len(results)} 个数据库\n")

    for r in results:
        title_arr = r.get("title", [])
        title = title_arr[0].get("plain_text", "?") if title_arr else "?"
        rid = r.get("id", "?")
        print(f"  {title}")
        print(f"    id: {rid}")
        print()


def cmd_query(args):
    """查询数据库"""
    if not args:
        print("用法: python3 notion-bridge.py query <database_id>")
        sys.exit(1)

    db_id = args[0]
    payload = {"page_size": 50}
    result = api("POST", f"/databases/{db_id}/query", payload)
    results = result.get("results", [])
    print(f"数据库 {db_id} 有 {len(results)} 行\n")

    for r in results[:20]:
        title = "?"
        for prop_name, prop in r.get("properties", {}).items():
            if prop.get("type") == "title":
                title_arr = prop.get("title", [])
                if title_arr:
                    title = title_arr[0].get("plain_text", "?")
                break
        rid = r.get("id", "?")
        last_edited = r.get("last_edited_time", "?")[:10]
        print(f"  - {title}  ({last_edited})")
        print(f"    id: {rid}")


def cmd_page(args):
    """读取页面"""
    if not args:
        print("用法: python3 notion-bridge.py page <page_id>")
        sys.exit(1)

    page_id = args[0]
    result = api("GET", f"/pages/{page_id}")
    print(json.dumps(result, ensure_ascii=False, indent=2))


def cmd_export(args):
    """导出页面为 markdown"""
    if not args:
        print("用法: python3 notion-bridge.py export <page_id>")
        sys.exit(1)

    page_id = args[0]

    # 读取页面元数据
    page = api("GET", f"/pages/{page_id}")
    title = "Untitled"
    for prop_name, prop in page.get("properties", {}).items():
        if prop.get("type") == "title":
            title_arr = prop.get("title", [])
            if title_arr:
                title = title_arr[0].get("plain_text", "Untitled")
            break

    # 读取 blocks
    blocks = []
    cursor = None
    while True:
        payload = {"page_size": 100}
        if cursor:
            payload["start_cursor"] = cursor
        result = api("POST", f"/blocks/{page_id}/children", payload)
        blocks.extend(result.get("results", []))
        if not result.get("has_more"):
            break
        cursor = result.get("next_cursor")

    md = f"# {title}\n\n"
    for block in blocks:
        md += block_to_markdown(block)

    print(md)


def block_to_markdown(block):
    """单个 block 转 markdown"""
    btype = block.get("type", "")
    content = ""

    if btype == "paragraph":
        rich_text = block.get("paragraph", {}).get("rich_text", [])
        content = rich_text_to_markdown(rich_text) + "\n\n"
    elif btype == "heading_1":
        content = "# " + rich_text_to_markdown(block.get("heading_1", {}).get("rich_text", [])) + "\n\n"
    elif btype == "heading_2":
        content = "## " + rich_text_to_markdown(block.get("heading_2", {}).get("rich_text", [])) + "\n\n"
    elif btype == "heading_3":
        content = "### " + rich_text_to_markdown(block.get("heading_3", {}).get("rich_text", [])) + "\n\n"
    elif btype == "bulleted_list_item":
        content = "- " + rich_text_to_markdown(block.get("bulleted_list_item", {}).get("rich_text", [])) + "\n"
    elif btype == "numbered_list_item":
        content = "1. " + rich_text_to_markdown(block.get("numbered_list_item", {}).get("rich_text", [])) + "\n"
    elif btype == "code":
        lang = block.get("code", {}).get("language", "")
        content = f"```{lang}\n" + rich_text_to_markdown(block.get("code", {}).get("rich_text", [])) + "\n```\n\n"
    elif btype == "quote":
        content = "> " + rich_text_to_markdown(block.get("quote", {}).get("rich_text", [])) + "\n\n"
    elif btype == "to_do":
        checked = block.get("to_do", {}).get("checked", False)
        mark = "[x]" if checked else "[ ]"
        content = f"- {mark} " + rich_text_to_markdown(block.get("to_do", {}).get("rich_text", [])) + "\n"

    return content


def rich_text_to_markdown(rich_text):
    """rich_text array 转 markdown"""
    parts = []
    for item in rich_text:
        text = item.get("plain_text", "")
        if item.get("href"):
            text = f"[{text}]({item['href']})"
        parts.append(text)
    return "".join(parts)


def cmd_create_page(args):
    """创建页面（简化版）"""
    if len(args) < 2:
        print("用法: python3 notion-bridge.py create-page <parent_id> <title>")
        sys.exit(1)

    parent_id = args[0]
    title = args[1]

    payload = {
        "parent": {"page_id": parent_id},
        "properties": {
            "title": [{"type": "text", "text": {"content": title}}]
        }
    }
    result = api("POST", "/pages", payload)
    print(f"✓ 创建页面: {result.get('url', '?')}")


def cmd_users(args):
    """列出 workspace 用户"""
    result = api("GET", "/users")
    for u in result.get("results", []):
        print(f"  [{u.get('type')}] {u.get('name', '?')}  ({u.get('id', '?')})")


# ============ Main ============

def main():
    if len(sys.argv) < 2:
        print(__doc__)
        sys.exit(1)

    cmd = sys.argv[1]
    args = sys.argv[2:]

    if not NOTION_API_KEY:
        print("⚠  缺少 NOTION_API_KEY 环境变量")
        print()
        print("   冰朔请按以下步骤:")
        print("   1. 打开 https://www.notion.so/my-integrations")
        print("   2. 创建新 integration: 名字 = bridge-zhuyuan")
        print("   3. 复制 token (形如 ntn_xxx)")
        print("   4. 在 Notion 工作区把页面右上角'分享'给 integration")
        print("   5. 执行: export NOTION_API_KEY=ntn_xxx")
        sys.exit(1)

    cmds = {
        "search": cmd_search,
        "databases": cmd_databases,
        "query": cmd_query,
        "page": cmd_page,
        "export": cmd_export,
        "create-page": cmd_create_page,
        "users": cmd_users,
    }

    if cmd not in cmds:
        print(f"未知命令: {cmd}")
        print(f"可用: {', '.join(cmds.keys())}")
        sys.exit(1)

    cmds[cmd](args)


if __name__ == "__main__":
    main()