335 lines
10 KiB
Python
Executable File
335 lines
10 KiB
Python
Executable File
#!/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() |