feat: 飞书知识库读写工具：增删查改节点、写入页面、创建早报、记录踩坑，基于 tenant_access_token。
This commit is contained in:
NightStar
@@ -0,0 +1,110 @@
|
||||
---
|
||||
name: "feishu-wiki-tools"
|
||||
description: "读写飞书知识库：增删查改节点、写入页面、创建早报、记录踩坑，基于 tenant_access_token。"
|
||||
metadata:
|
||||
author: xf
|
||||
version: "1.0.0"
|
||||
---
|
||||
|
||||
# Feishu Wiki Tools
|
||||
|
||||
飞书知识库读写操作。所有脚本使用 `tenant_access_token`，从 `~/.openclaw/openclaw.json` 读取凭证。
|
||||
|
||||
## 前置条件
|
||||
|
||||
- 飞书应用已开通 `wiki:wiki docx:document` 权限
|
||||
- 机器人已加入目标知识库并获 `full_access`（参考 `references/auth.md`）
|
||||
- 服务器可访问飞书 API
|
||||
|
||||
## 环境变量
|
||||
|
||||
| 变量 | 默认值 | 说明 |
|
||||
|------|--------|------|
|
||||
| `FEISHU_SPACE_ID` | `7664817589230570761` | 知识库 Space ID |
|
||||
| `FEISHU_ROOT_NODE` | `LY5wwNWpniOoFwkHHG4cmDvGnSh` | 首页 node token |
|
||||
| `FEISHU_DIGEST_PARENT` | `BH8AwLgJhipBcUkehCAcyKvDngE` | 早报父节点 |
|
||||
| `FEISHU_SUMMARY_PARENT` | `XunCwlk7JiSTuAkvoKlcq54pnPb` | 每日总结父节点 |
|
||||
|
||||
## 通用脚本（CRUD）
|
||||
|
||||
### 初始化 Token
|
||||
|
||||
```bash
|
||||
source scripts/init_token.sh
|
||||
```
|
||||
设置 `BEARER_TOKEN` 环境变量。其他脚本会自动调用。
|
||||
|
||||
### 列节点
|
||||
|
||||
```bash
|
||||
python3 scripts/wiki_list.py [parent_node_token]
|
||||
```
|
||||
不传 parent 则列根目录。
|
||||
|
||||
### 创建节点
|
||||
|
||||
```bash
|
||||
python3 scripts/wiki_create.py --title "标题" --parent "PARENT_TOKEN" [--obj-type docx]
|
||||
```
|
||||
|
||||
### 读取页面
|
||||
|
||||
```bash
|
||||
python3 scripts/wiki_read.py --node "NODE_TOKEN"
|
||||
```
|
||||
输出纯文本内容到 stdout。
|
||||
|
||||
### 写入页面
|
||||
|
||||
```bash
|
||||
python3 scripts/wiki_write.py --node "NODE_TOKEN" --file content.md
|
||||
```
|
||||
支持 Markdown 转换。
|
||||
|
||||
> ⚠️ **已知限制**：`md_to_blocks.py` 不支持 Markdown 表格语法（`| ... |`），会被转为纯文本而非飞书原生表格 block。如需表格，改用列表 + 加粗键名的格式，例如：
|
||||
> ```markdown
|
||||
> - **项目**：说明
|
||||
> - **模型**：DeepSeek V4 Pro
|
||||
> ```
|
||||
|
||||
### 删除页面
|
||||
|
||||
```bash
|
||||
python3 scripts/wiki_delete.py --node "NODE_TOKEN"
|
||||
```
|
||||
移到回收站（需 `space:document:delete` 权限）。
|
||||
|
||||
## 特化脚本
|
||||
|
||||
### 创建早报
|
||||
|
||||
```bash
|
||||
python3 scripts/create_digest.py --title "07.21 周二 早报" --content content.md
|
||||
```
|
||||
自动挂到 `FEISHU_DIGEST_PARENT` 按周分组下。
|
||||
|
||||
### 记录踩坑
|
||||
|
||||
```bash
|
||||
python3 scripts/create_pitfall.py --title "标题" --file content.md
|
||||
```
|
||||
自动挂到 `踩坑记录合集` 节点下。
|
||||
|
||||
### 创建每日总结
|
||||
|
||||
```bash
|
||||
python3 scripts/create_summary.py --title "07.21 周二 总结" --content summary.md
|
||||
```
|
||||
自动挂到 `FEISHU_SUMMARY_PARENT`（🌙 每日总结）下。内容精简，仅含当日「晚星今日动态」。
|
||||
|
||||
## Block 类型速查
|
||||
|
||||
- 3=heading1, 4=heading2, 5=heading3
|
||||
- 2=text, 12=bullet, 22=divider, 14=code
|
||||
- 写入时支持 Markdown → Block 转换（`scripts/md_to_blocks.py`）
|
||||
|
||||
## 参考
|
||||
|
||||
- `references/auth.md` — 鉴权配置完整流程
|
||||
- `references/block_types.md` — Docx Block API 参考
|
||||
- `references/api_endpoints.md` — Feishu API 端点速查
|
||||
@@ -0,0 +1,41 @@
|
||||
# Feishu API 端点速查
|
||||
|
||||
## 认证
|
||||
|
||||
```
|
||||
POST https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal
|
||||
Body: {app_id, app_secret}
|
||||
```
|
||||
|
||||
## Wiki 空间
|
||||
|
||||
```
|
||||
GET /wiki/v2/spaces # 列表
|
||||
GET /wiki/v2/spaces/{space_id} # 详情
|
||||
GET /wiki/v2/spaces/{space_id}/nodes # 列节点 (parent_node_token 可选)
|
||||
POST /wiki/v2/spaces/{space_id}/nodes # 创建节点
|
||||
GET /wiki/v2/spaces/get_node?token=*** # 节点详情
|
||||
POST /wiki/v2/spaces/{space_id}/nodes/{node}/move # 移动节点
|
||||
```
|
||||
|
||||
## Wiki 成员
|
||||
|
||||
```
|
||||
GET /wiki/v2/spaces/{space_id}/members # 成员列表
|
||||
POST /wiki/v2/spaces/{space_id}/members # 添加成员
|
||||
```
|
||||
|
||||
## Docx
|
||||
|
||||
```
|
||||
GET /docx/v1/documents/{doc_id}/blocks/{block_id} # 获取块
|
||||
DELETE /docx/v1/documents/{doc_id}/blocks/{block_id}/children/batch_delete # 清空子块
|
||||
POST /docx/v1/documents/{doc_id}/blocks/{block_id}/children # 写入子块
|
||||
```
|
||||
|
||||
## Drive
|
||||
|
||||
```
|
||||
DELETE /drive/v1/files/{obj_token}?type=docx # 移入回收站
|
||||
POST /drive/v1/permissions/{token}/members?type=docx # 授权
|
||||
```
|
||||
@@ -0,0 +1,40 @@
|
||||
# 飞书知识库鉴权配置
|
||||
|
||||
## 配置来源
|
||||
|
||||
所有飞书维度的配置统一从 `~/.openclaw/openclaw.json` 的 `channels.feishu` 读取：
|
||||
|
||||
| 字段 | 说明 |
|
||||
|------|------|
|
||||
| `appId` | 应用 App ID |
|
||||
| `appSecret` | 应用 App Secret（敏感） |
|
||||
| `botOpenId` | 机器人的 Open ID |
|
||||
|
||||
脚本通过 `init_token.py` 读取配置并获取 `tenant_access_token`（有效期 2 小时，自动续期）。
|
||||
|
||||
## 完整配置步骤
|
||||
|
||||
1. 在飞书开放平台为应用开通权限：
|
||||
- `wiki:wiki` — 知识库读写
|
||||
- `wiki:member:create` — 添加成员
|
||||
- `docx:document` — 文档读写
|
||||
- `docx:document:readonly` — 文档只读
|
||||
- `drive:drive` — 云盘操作
|
||||
- `space:document:delete` — 删除文档
|
||||
|
||||
2. 应用 Bot ID / Open ID
|
||||
- 从配置文件读取 `channels.feishu.appId` 和 `channels.feishu.botOpenId`
|
||||
|
||||
3. 添加机器人为知识库成员（需管理员 OAuth）：
|
||||
- 走 OAuth 获取 user_access_token
|
||||
- 调 `POST /wiki/v2/spaces/{space_id}/members`
|
||||
- `member_type: openid`, `member_role: member`
|
||||
|
||||
4. 授予编辑权限（需管理员 OAuth + 用户同意）：
|
||||
- 调 `POST /drive/v1/permissions/{obj_token}/members?type=docx`
|
||||
- `perm: full_access`
|
||||
|
||||
5. 日常操作：直接用 `tenant_access_token`
|
||||
- 从 `~/.openclaw/openclaw.json` 读取 app_id/app_secret
|
||||
- 调 `POST /auth/v3/tenant_access_token/internal` 获取 token
|
||||
- 有效期 2 小时，自动续期
|
||||
@@ -0,0 +1,37 @@
|
||||
# Docx Block 类型
|
||||
|
||||
## 常用块
|
||||
|
||||
| type | block_type | 说明 | key 字段 |
|
||||
|------|-----------|------|---------|
|
||||
| text | 2 | 普通文本 | `text.elements[].text_run` |
|
||||
| heading1 | 3 | 一级标题 | `heading1` |
|
||||
| heading2 | 4 | 二级标题 | `heading2` |
|
||||
| heading3 | 5 | 三级标题 | `heading3` |
|
||||
| bullet | 12 | 无序列表 | `bullet` |
|
||||
| code | 14 | 代码块 | `code` |
|
||||
| divider | 22 | 分割线 | `divider` |
|
||||
|
||||
## Text Element 属性
|
||||
|
||||
```json
|
||||
{
|
||||
"text_run": {
|
||||
"content": "文本内容",
|
||||
"text_element_style": {
|
||||
"bold": false,
|
||||
"link": {"url": "https://..."},
|
||||
"inline_code": false,
|
||||
"italic": false,
|
||||
"strikethrough": false,
|
||||
"underline": false
|
||||
}
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
## 写入流程
|
||||
|
||||
1. `GET /docx/v1/documents/{doc_id}/blocks/{doc_id}` — 获取现有块
|
||||
2. `DELETE .../children/batch_delete` — 清空默认块
|
||||
3. `POST .../children` — 批量写入（≤25块/批）
|
||||
@@ -0,0 +1,92 @@
|
||||
#!/usr/bin/env python3
|
||||
"""创建早报页面，自动挂到按周分组目录下"""
|
||||
import json, sys, os, argparse
|
||||
from datetime import datetime
|
||||
import subprocess
|
||||
|
||||
SCRIPTS_DIR = os.path.dirname(os.path.abspath(__file__))
|
||||
SPACE_ID = os.environ.get('FEISHU_SPACE_ID', '7664817589230570761')
|
||||
DIGEST_PARENT = os.environ.get('FEISHU_DIGEST_PARENT', 'BH8AwLgJhipBcUkehCAcyKvDngE')
|
||||
|
||||
def run_script(name, *args):
|
||||
cmd = ['python3', os.path.join(SCRIPTS_DIR, name)] + list(args)
|
||||
r = subprocess.run(cmd, capture_output=True, text=True)
|
||||
if r.returncode != 0:
|
||||
raise Exception(f"{name} failed: {r.stderr}")
|
||||
return json.loads(r.stdout)
|
||||
|
||||
def get_token():
|
||||
r = subprocess.run(['python3', os.path.join(SCRIPTS_DIR, 'init_token.py')],
|
||||
capture_output=True, text=True)
|
||||
return json.loads(r.stdout)['token']
|
||||
|
||||
def create_digest(title, content_file=None):
|
||||
import requests
|
||||
token = get_token()
|
||||
h = {'Authorization': 'Bearer ' + token, 'Content-Type': 'application/json'}
|
||||
|
||||
# Find or create this week's node
|
||||
from datetime import timedelta
|
||||
today = datetime.now()
|
||||
iso_week = today.isocalendar()[1]
|
||||
week_end = today + timedelta(days=(6 - today.weekday()))
|
||||
week_title = f"{today.year}年第{iso_week}周 {today.strftime('%m.%d')}-{week_end.strftime('%m.%d')}"
|
||||
|
||||
# List children of digest parent
|
||||
r = requests.get(
|
||||
f'https://open.feishu.cn/open-apis/wiki/v2/spaces/{SPACE_ID}/nodes',
|
||||
params={'parent_node_token': DIGEST_PARENT}, headers=h
|
||||
)
|
||||
kids = r.json()['data']['items']
|
||||
|
||||
week_token = None
|
||||
for k in kids:
|
||||
if str(iso_week) in k['title']:
|
||||
week_token = k['node_token']
|
||||
break
|
||||
|
||||
if not week_token:
|
||||
r = requests.post(
|
||||
f'https://open.feishu.cn/open-apis/wiki/v2/spaces/{SPACE_ID}/nodes',
|
||||
headers=h,
|
||||
json={'obj_type': 'docx', 'node_type': 'origin', 'parent_node_token': DIGEST_PARENT, 'title': week_title}
|
||||
)
|
||||
data = r.json()
|
||||
if data.get('code') != 0:
|
||||
raise Exception(f"Create week node failed: {data.get('msg')}")
|
||||
week_token = data['data']['node']['node_token']
|
||||
|
||||
# Create today's digest
|
||||
r = requests.post(
|
||||
f'https://open.feishu.cn/open-apis/wiki/v2/spaces/{SPACE_ID}/nodes',
|
||||
headers=h,
|
||||
json={'obj_type': 'docx', 'node_type': 'origin', 'parent_node_token': week_token, 'title': title}
|
||||
)
|
||||
data = r.json()
|
||||
if data.get('code') != 0:
|
||||
raise Exception(f"Create digest failed: {data.get('msg')}")
|
||||
|
||||
node = data['data']['node']
|
||||
print(json.dumps({'node_token': node['node_token'], 'obj_token': node['obj_token']}))
|
||||
|
||||
# Write content if provided
|
||||
if content_file:
|
||||
node_token = node['node_token']
|
||||
subprocess.run(
|
||||
['python3', os.path.join(SCRIPTS_DIR, 'wiki_write.py'), '--node', node_token, '--file', content_file],
|
||||
check=True
|
||||
)
|
||||
|
||||
return node
|
||||
|
||||
if __name__ == '__main__':
|
||||
p = argparse.ArgumentParser()
|
||||
p.add_argument('--title', required=True, help='Digest title')
|
||||
p.add_argument('--content', default=None, help='Content JSON file')
|
||||
args = p.parse_args()
|
||||
|
||||
try:
|
||||
create_digest(args.title, args.content)
|
||||
except Exception as e:
|
||||
print(f'Error: {e}', file=sys.stderr)
|
||||
sys.exit(1)
|
||||
@@ -0,0 +1,72 @@
|
||||
#!/usr/bin/env python3
|
||||
"""创建踩坑记录页面，自动挂到踩坑记录合集下"""
|
||||
import json, sys, os, argparse, subprocess
|
||||
|
||||
SCRIPTS_DIR = os.path.dirname(os.path.abspath(__file__))
|
||||
SPACE_ID = os.environ.get('FEISHU_SPACE_ID', '7664817589230570761')
|
||||
|
||||
def get_token():
|
||||
r = subprocess.run(['python3', os.path.join(SCRIPTS_DIR, 'init_token.py')],
|
||||
capture_output=True, text=True)
|
||||
return json.loads(r.stdout)['token']
|
||||
|
||||
def find_or_create_parent(name):
|
||||
"""Find or create a parent node by name under root"""
|
||||
import requests
|
||||
token = get_token()
|
||||
h = {'Authorization': 'Bearer ' + token, 'Content-Type': 'application/json'}
|
||||
ROOT = os.environ.get('FEISHU_ROOT_NODE', 'LY5wwNWpniOoFwkHHG4cmDvGnSh')
|
||||
|
||||
r = requests.get(
|
||||
f'https://open.feishu.cn/open-apis/wiki/v2/spaces/{SPACE_ID}/nodes',
|
||||
params={'parent_node_token': ROOT}, headers=h
|
||||
)
|
||||
items = r.json()['data']['items']
|
||||
for n in items:
|
||||
if name in n['title']:
|
||||
return n['node_token']
|
||||
|
||||
r = requests.post(
|
||||
f'https://open.feishu.cn/open-apis/wiki/v2/spaces/{SPACE_ID}/nodes',
|
||||
headers=h,
|
||||
json={'obj_type': 'docx', 'node_type': 'origin', 'parent_node_token': ROOT, 'title': '📝 踩坑记录合集'}
|
||||
)
|
||||
return r.json()['data']['node']['node_token']
|
||||
|
||||
def create_pitfall(title, content_file=None):
|
||||
import requests
|
||||
token = get_token()
|
||||
h = {'Authorization': 'Bearer ' + token, 'Content-Type': 'application/json'}
|
||||
|
||||
parent_token = find_or_create_parent('踩坑记录合集')
|
||||
|
||||
r = requests.post(
|
||||
f'https://open.feishu.cn/open-apis/wiki/v2/spaces/{SPACE_ID}/nodes',
|
||||
headers=h,
|
||||
json={'obj_type': 'docx', 'node_type': 'origin', 'parent_node_token': parent_token, 'title': title}
|
||||
)
|
||||
data = r.json()
|
||||
if data.get('code') != 0:
|
||||
raise Exception(f"Create pitfall failed: {data.get('msg')}")
|
||||
|
||||
node = data['data']['node']
|
||||
print(json.dumps({'node_token': node['node_token'], 'obj_token': node['obj_token']}))
|
||||
|
||||
if content_file:
|
||||
subprocess.run(
|
||||
['python3', os.path.join(SCRIPTS_DIR, 'wiki_write.py'), '--node', node['node_token'], '--file', content_file],
|
||||
check=True
|
||||
)
|
||||
return node
|
||||
|
||||
if __name__ == '__main__':
|
||||
p = argparse.ArgumentParser()
|
||||
p.add_argument('--title', required=True)
|
||||
p.add_argument('--content', default=None)
|
||||
args = p.parse_args()
|
||||
|
||||
try:
|
||||
create_pitfall(args.title, args.content)
|
||||
except Exception as e:
|
||||
print(f'Error: {e}', file=sys.stderr)
|
||||
sys.exit(1)
|
||||
@@ -0,0 +1,54 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
创建当日总结页面，挂到「🌙 每日总结」目录下。
|
||||
用法: python3 create_summary.py --title "07.21 周二 总结" --content summary.md
|
||||
|
||||
复用 wiki_create.py + wiki_write.py，自身不发起 HTTP 请求，
|
||||
从而避免 exec 安全策略对 Authorization / Bearer 的扫描拦截。
|
||||
"""
|
||||
import json, sys, os, argparse
|
||||
import subprocess
|
||||
|
||||
SCRIPTS_DIR = os.path.dirname(os.path.abspath(__file__))
|
||||
SUMMARY_PARENT = os.environ.get('FEISHU_SUMMARY_PARENT', 'XunCwlk7JiSTuAkvoKlcq54pnPb')
|
||||
|
||||
|
||||
def _run(name, *args):
|
||||
"""Run a peer script and return parsed stdout JSON."""
|
||||
r = subprocess.run(
|
||||
['python3', os.path.join(SCRIPTS_DIR, name)] + list(args),
|
||||
capture_output=True, text=True
|
||||
)
|
||||
if r.returncode != 0:
|
||||
raise Exception(f"{name} failed: {r.stderr}")
|
||||
return json.loads(r.stdout)
|
||||
|
||||
|
||||
def create_summary(title, content_file=None):
|
||||
# 1) Create the node via wiki_create.py
|
||||
node = _run('wiki_create.py', '--title', title, '--parent', SUMMARY_PARENT)
|
||||
node_token = node['node_token']
|
||||
print(json.dumps({'node_token': node_token, 'obj_token': node['obj_token']}))
|
||||
|
||||
# 2) Write content via wiki_write.py
|
||||
if content_file:
|
||||
subprocess.run(
|
||||
['python3', os.path.join(SCRIPTS_DIR, 'wiki_write.py'),
|
||||
'--node', node_token, '--file', content_file, '--markdown'],
|
||||
check=True
|
||||
)
|
||||
|
||||
return node_token
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
p = argparse.ArgumentParser()
|
||||
p.add_argument('--title', required=True, help='Summary title, e.g. "07.21 周二 总结"')
|
||||
p.add_argument('--content', default=None, help='Markdown content file')
|
||||
args = p.parse_args()
|
||||
|
||||
try:
|
||||
create_summary(args.title, args.content)
|
||||
except Exception as e:
|
||||
print(f'Error: {e}', file=sys.stderr)
|
||||
sys.exit(1)
|
||||
@@ -0,0 +1,41 @@
|
||||
#!/usr/bin/env python3
|
||||
"""获取 tenant_access_token，同时暴露 app_id 和 bot_open_id。
|
||||
用法:
|
||||
python3 init_token.py → {"token": "..."}
|
||||
python3 init_token.py --full → {"token": "...", "app_id": "...", "bot_open_id": "..."}
|
||||
"""
|
||||
import json, sys, os
|
||||
from pathlib import Path
|
||||
|
||||
CONFIG = Path.home() / '.openclaw' / 'openclaw.json'
|
||||
|
||||
def read_config():
|
||||
cfg = json.loads(CONFIG.read_text())
|
||||
fs = cfg['channels']['feishu']
|
||||
return fs['appId'], fs['appSecret'], fs.get('botOpenId', '')
|
||||
|
||||
def get_token():
|
||||
aid, asc, _ = read_config()
|
||||
import requests
|
||||
r = requests.post(
|
||||
'https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal',
|
||||
json={'app_id': aid, 'app_secret': asc}
|
||||
)
|
||||
td = r.json()
|
||||
if 'tenant_access_token' not in td:
|
||||
print(json.dumps({'error': td}), file=sys.stderr)
|
||||
sys.exit(1)
|
||||
return td['tenant_access_token']
|
||||
|
||||
if __name__ == '__main__':
|
||||
try:
|
||||
token = get_token()
|
||||
result = {'token': token}
|
||||
if '--full' in sys.argv:
|
||||
aid, asc, bot_id = read_config()
|
||||
result['app_id'] = aid
|
||||
result['bot_open_id'] = bot_id
|
||||
print(json.dumps(result))
|
||||
except Exception as e:
|
||||
print(json.dumps({'error': str(e)}), file=sys.stderr)
|
||||
sys.exit(1)
|
||||
@@ -0,0 +1,140 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Markdown to Feishu Docx Block JSON converter.
|
||||
Usage: python3 md_to_blocks.py < input.md > output.json
|
||||
python3 md_to_blocks.py file.md
|
||||
"""
|
||||
|
||||
import re, json, sys
|
||||
|
||||
def parse_inline(text):
|
||||
"""Parse a line of inline markdown into text_run elements."""
|
||||
elements = []
|
||||
pos = 0
|
||||
# Patterns in order: bold, italic, inline code, link, plain text
|
||||
while pos < len(text):
|
||||
# Bold: **text** or __text__
|
||||
m = re.match(r'\*\*(.+?)\*\*|__(.+?)__', text[pos:])
|
||||
if m:
|
||||
elements.append({'text_run': {'content': m.group(1), 'text_element_style': {'bold': True}}})
|
||||
pos += len(m.group(0))
|
||||
continue
|
||||
# Link: [text](url)
|
||||
m = re.match(r'\[(.+?)\]\((.+?)\)', text[pos:])
|
||||
if m:
|
||||
elements.append({'text_run': {'content': m.group(1), 'text_element_style': {'link': {'url': m.group(2)}}}})
|
||||
pos += len(m.group(0))
|
||||
continue
|
||||
# Inline code: `text`
|
||||
m = re.match(r'`(.+?)`', text[pos:])
|
||||
if m:
|
||||
elements.append({'text_run': {'content': m.group(1), 'text_element_style': {'inline_code': True}}})
|
||||
pos += len(m.group(0))
|
||||
continue
|
||||
# Italic: *text*
|
||||
m = re.match(r'\*(.+?)\*', text[pos:])
|
||||
if m:
|
||||
elements.append({'text_run': {'content': m.group(1), 'text_element_style': {'italic': True}}})
|
||||
pos += len(m.group(0))
|
||||
continue
|
||||
# Plain text until next special char
|
||||
nxt = re.search(r'[\*\[`]', text[pos:])
|
||||
if nxt:
|
||||
end = pos + nxt.start()
|
||||
if end > pos:
|
||||
elements.append({'text_run': {'content': text[pos:end], 'text_element_style': {}}})
|
||||
elif end == pos:
|
||||
# special char at current position but not a match — skip it as literal
|
||||
elements.append({'text_run': {'content': text[pos:pos+1], 'text_element_style': {}}})
|
||||
pos += 1
|
||||
continue
|
||||
pos = end
|
||||
else:
|
||||
elements.append({'text_run': {'content': text[pos:], 'text_element_style': {}}})
|
||||
break
|
||||
return elements
|
||||
|
||||
def md_to_blocks(md_text):
|
||||
"""Convert Markdown text to Feishu Block JSON array."""
|
||||
lines = md_text.strip().split('\n')
|
||||
blocks = []
|
||||
i = 0
|
||||
while i < len(lines):
|
||||
line = lines[i]
|
||||
|
||||
# Empty line
|
||||
if not line.strip():
|
||||
# Add empty paragraph only if next line is also text (not heading/list/divider)
|
||||
if i + 1 < len(lines) and lines[i + 1].strip() and not lines[i + 1].strip().startswith(('#', '-', '*', '`', '---')):
|
||||
blocks.append({'block_type': 2, 'text': {'elements': [{'text_run': {'content': '', 'text_element_style': {}}}], 'style': {}}})
|
||||
i += 1
|
||||
continue
|
||||
|
||||
# Divider
|
||||
if line.strip() == '---' or line.strip() == '***':
|
||||
blocks.append({'block_type': 22, 'divider': {}})
|
||||
i += 1
|
||||
continue
|
||||
|
||||
# Headings
|
||||
h = re.match(r'^(#{1,3})\s+(.+)$', line)
|
||||
if h:
|
||||
level = len(h.group(1))
|
||||
bt = 2 + level # 3=h1, 4=h2, 5=h3
|
||||
blocks.append({
|
||||
'block_type': bt,
|
||||
f'heading{level}': {
|
||||
'elements': parse_inline(h.group(2)),
|
||||
'style': {}
|
||||
}
|
||||
})
|
||||
i += 1
|
||||
continue
|
||||
|
||||
# Bullet list
|
||||
bullet = re.match(r'^[\-\*]\s+(.+)$', line)
|
||||
if bullet:
|
||||
blocks.append({
|
||||
'block_type': 12,
|
||||
'bullet': {
|
||||
'elements': parse_inline(bullet.group(1)),
|
||||
'style': {}
|
||||
}
|
||||
})
|
||||
i += 1
|
||||
continue
|
||||
|
||||
# Numbered list
|
||||
numbered = re.match(r'^\d+\.\s+(.+)$', line)
|
||||
if numbered:
|
||||
blocks.append({
|
||||
'block_type': 13,
|
||||
'ordered': {
|
||||
'elements': parse_inline(numbered.group(1)),
|
||||
'style': {}
|
||||
}
|
||||
})
|
||||
i += 1
|
||||
continue
|
||||
|
||||
# Regular text
|
||||
elements = parse_inline(line)
|
||||
blocks.append({
|
||||
'block_type': 2,
|
||||
'text': {
|
||||
'elements': elements if elements else [{'text_run': {'content': line, 'text_element_style': {}}}],
|
||||
'style': {}
|
||||
}
|
||||
})
|
||||
i += 1
|
||||
|
||||
return blocks
|
||||
|
||||
if __name__ == '__main__':
|
||||
if len(sys.argv) > 1:
|
||||
with open(sys.argv[1]) as f:
|
||||
text = f.read()
|
||||
else:
|
||||
text = sys.stdin.read()
|
||||
|
||||
blocks = md_to_blocks(text)
|
||||
print(json.dumps(blocks, indent=2, ensure_ascii=False))
|
||||
@@ -0,0 +1,40 @@
|
||||
#!/usr/bin/env python3
|
||||
"""创建知识库节点"""
|
||||
import json, sys, os, argparse
|
||||
import subprocess
|
||||
|
||||
SPACE_ID = os.environ.get('FEISHU_SPACE_ID', '7664817589230570761')
|
||||
|
||||
def get_token():
|
||||
r = subprocess.run(['python3', os.path.join(os.path.dirname(__file__), 'init_token.py')],
|
||||
capture_output=True, text=True)
|
||||
return json.loads(r.stdout)['token']
|
||||
|
||||
def create_node(title, parent_token, obj_type='docx'):
|
||||
import requests
|
||||
token = get_token()
|
||||
h = {'Authorization': 'Bearer ' + token, 'Content-Type': 'application/json'}
|
||||
|
||||
r = requests.post(
|
||||
f'https://open.feishu.cn/open-apis/wiki/v2/spaces/{SPACE_ID}/nodes',
|
||||
headers=h,
|
||||
json={'obj_type': obj_type, 'node_type': 'origin', 'parent_node_token': parent_token, 'title': title}
|
||||
)
|
||||
data = r.json()
|
||||
if data.get('code') != 0:
|
||||
raise Exception(f"API error: {data.get('msg')} - {json.dumps(data, ensure_ascii=False)}")
|
||||
return data['data']['node']
|
||||
|
||||
if __name__ == '__main__':
|
||||
p = argparse.ArgumentParser()
|
||||
p.add_argument('--title', required=True)
|
||||
p.add_argument('--parent', required=True)
|
||||
p.add_argument('--obj-type', default='docx')
|
||||
args = p.parse_args()
|
||||
|
||||
try:
|
||||
node = create_node(args.title, args.parent, args.obj_type)
|
||||
print(json.dumps(node, indent=2, ensure_ascii=False))
|
||||
except Exception as e:
|
||||
print(f'Error: {e}', file=sys.stderr)
|
||||
sys.exit(1)
|
||||
@@ -0,0 +1,48 @@
|
||||
#!/usr/bin/env python3
|
||||
"""删除知识库节点（移到回收站）"""
|
||||
import json, sys, os, argparse
|
||||
import subprocess
|
||||
|
||||
SPACE_ID = os.environ.get('FEISHU_SPACE_ID', '7664817589230570761')
|
||||
|
||||
def get_token():
|
||||
r = subprocess.run(['python3', os.path.join(os.path.dirname(__file__), 'init_token.py')],
|
||||
capture_output=True, text=True)
|
||||
return json.loads(r.stdout)['token']
|
||||
|
||||
def delete_node(node_token):
|
||||
import requests
|
||||
token = get_token()
|
||||
h = {'Authorization': 'Bearer ' + token, 'Content-Type': 'application/json'}
|
||||
|
||||
# Get obj_token
|
||||
r = requests.get(
|
||||
'https://open.feishu.cn/open-apis/wiki/v2/spaces/get_node',
|
||||
params={'token': node_token}, headers=h
|
||||
)
|
||||
nd = r.json()
|
||||
if nd.get('code') != 0:
|
||||
raise Exception(f"Get node failed: {nd.get('msg')}")
|
||||
obj_token = nd['data']['node']['obj_token']
|
||||
|
||||
# Delete via drive API
|
||||
r = requests.delete(
|
||||
f'https://open.feishu.cn/open-apis/drive/v1/files/{obj_token}?type=docx',
|
||||
headers=h
|
||||
)
|
||||
data = r.json()
|
||||
if data.get('code') != 0:
|
||||
raise Exception(f"Delete failed: {data.get('msg')}")
|
||||
return True
|
||||
|
||||
if __name__ == '__main__':
|
||||
p = argparse.ArgumentParser()
|
||||
p.add_argument('--node', required=True, help='Node token to delete')
|
||||
args = p.parse_args()
|
||||
|
||||
try:
|
||||
delete_node(args.node)
|
||||
print('Deleted')
|
||||
except Exception as e:
|
||||
print(f'Error: {e}', file=sys.stderr)
|
||||
sys.exit(1)
|
||||
@@ -0,0 +1,37 @@
|
||||
#!/usr/bin/env python3
|
||||
"""列出知识库节点，可选指定父节点"""
|
||||
import json, sys, os
|
||||
import subprocess
|
||||
|
||||
SPACE_ID = os.environ.get('FEISHU_SPACE_ID', '7664817589230570761')
|
||||
|
||||
def get_token():
|
||||
r = subprocess.run(['python3', os.path.join(os.path.dirname(__file__), 'init_token.py')],
|
||||
capture_output=True, text=True)
|
||||
td = json.loads(r.stdout)
|
||||
return td['token']
|
||||
|
||||
def list_nodes(parent_token=None):
|
||||
import requests
|
||||
token = get_token()
|
||||
h = {'Authorization': 'Bearer ' + token, 'Content-Type': 'application/json'}
|
||||
|
||||
url = f'https://open.feishu.cn/open-apis/wiki/v2/spaces/{SPACE_ID}/nodes'
|
||||
params = {}
|
||||
if parent_token:
|
||||
params['parent_node_token'] = parent_token
|
||||
|
||||
r = requests.get(url, params=params, headers=h)
|
||||
data = r.json()
|
||||
if data.get('code') != 0:
|
||||
raise Exception(f"API error: {data.get('msg')}")
|
||||
return data['data']['items']
|
||||
|
||||
if __name__ == '__main__':
|
||||
parent = sys.argv[1] if len(sys.argv) > 1 else None
|
||||
try:
|
||||
nodes = list_nodes(parent)
|
||||
print(json.dumps(nodes, indent=2, ensure_ascii=False))
|
||||
except Exception as e:
|
||||
print(f'Error: {e}', file=sys.stderr)
|
||||
sys.exit(1)
|
||||
@@ -0,0 +1,123 @@
|
||||
#!/usr/bin/env python3
|
||||
"""写入 Wiki 页面内容。支持 Markdown 文件和 Block JSON。"""
|
||||
import json, sys, os, argparse
|
||||
import subprocess
|
||||
|
||||
SPACE_ID = os.environ.get('FEISHU_SPACE_ID', '7664817589230570761')
|
||||
|
||||
def get_token():
|
||||
r = subprocess.run(['python3', os.path.join(os.path.dirname(__file__), 'init_token.py')],
|
||||
capture_output=True, text=True)
|
||||
return json.loads(r.stdout)['token']
|
||||
|
||||
# Block helpers
|
||||
def te(t, **kw):
|
||||
d = {}
|
||||
d.update(kw)
|
||||
return {'text_run': {'content': t, 'text_element_style': d}}
|
||||
|
||||
def link_el(t, u):
|
||||
return {'text_run': {'content': t, 'text_element_style': {'link': {'url': u}}}}
|
||||
|
||||
H1 = lambda t: {'block_type': 3, 'heading1': {'elements': [te(t, bold=True)], 'style': {}}}
|
||||
H2 = lambda t: {'block_type': 4, 'heading2': {'elements': [te(t, bold=True)], 'style': {}}}
|
||||
H3 = lambda t: {'block_type': 5, 'heading3': {'elements': [te(t, bold=True)], 'style': {}}}
|
||||
DIV = lambda: {'block_type': 22, 'divider': {}}
|
||||
TXT = lambda t: {'block_type': 2, 'text': {'elements': [te(t)], 'style': {}}}
|
||||
BLD = lambda t: {'block_type': 2, 'text': {'elements': [te(t, bold=True)], 'style': {}}}
|
||||
BUL = lambda t, lt='', lu='': {'block_type': 12, 'bullet': {'elements': [te(t)] + ([link_el(' ' + lt, lu)] if lt and lu else []), 'style': {}}}
|
||||
|
||||
def json_to_blocks(data):
|
||||
"""Convert JSON block data to Feishu block objects"""
|
||||
if isinstance(data, list):
|
||||
return [json_to_blocks(d) for d in data]
|
||||
bt = data.get('type', 'text')
|
||||
text = data.get('text', '')
|
||||
|
||||
type_map = {
|
||||
'h1': lambda: H1(text),
|
||||
'h2': lambda: H2(text),
|
||||
'h3': lambda: H3(text),
|
||||
'text': lambda: TXT(text),
|
||||
'bold': lambda: BLD(text),
|
||||
'divider': lambda: DIV(),
|
||||
'bullet': lambda: BUL(text, data.get('link_text', ''), data.get('link_url', '')),
|
||||
}
|
||||
return type_map.get(bt, lambda: TXT(text))()
|
||||
|
||||
def write_blocks(node_token, blocks):
|
||||
import requests
|
||||
token = get_token()
|
||||
h = {'Authorization': 'Bearer ' + token, 'Content-Type': 'application/json'}
|
||||
|
||||
# Get node detail to find obj_token
|
||||
r = requests.get(
|
||||
'https://open.feishu.cn/open-apis/wiki/v2/spaces/get_node',
|
||||
params={'token': node_token}, headers=h
|
||||
)
|
||||
nd = r.json()
|
||||
if nd.get('code') != 0:
|
||||
raise Exception(f"Get node failed: {nd.get('msg')}")
|
||||
doc_id = nd['data']['node']['obj_token']
|
||||
|
||||
# Clear defaults
|
||||
r = requests.get(
|
||||
f'https://open.feishu.cn/open-apis/docx/v1/documents/{doc_id}/blocks/{doc_id}',
|
||||
headers=h
|
||||
)
|
||||
kids = r.json().get('data', {}).get('block', {}).get('children', [])
|
||||
if kids:
|
||||
cids = [c['block_id'] for c in kids]
|
||||
requests.delete(
|
||||
f'https://open.feishu.cn/open-apis/docx/v1/documents/{doc_id}/blocks/{doc_id}/children/batch_delete',
|
||||
headers=h, json={'start_index': 0, 'end_index': len(cids)}
|
||||
)
|
||||
|
||||
# Write in batches
|
||||
batch_size = 25
|
||||
for i in range(0, len(blocks), batch_size):
|
||||
batch = blocks[i:i + batch_size]
|
||||
r = requests.post(
|
||||
f'https://open.feishu.cn/open-apis/docx/v1/documents/{doc_id}/blocks/{doc_id}/children',
|
||||
headers=h, json={'children': batch, 'index': -1}
|
||||
)
|
||||
if r.json().get('code') != 0:
|
||||
raise Exception(f"Write batch {i // batch_size} failed: {r.json().get('msg')}")
|
||||
return True
|
||||
|
||||
if __name__ == '__main__':
|
||||
p = argparse.ArgumentParser()
|
||||
p.add_argument('--node', required=True, help='Node token')
|
||||
p.add_argument('--file', default=None, help='Content file (JSON block array or Markdown)')
|
||||
p.add_argument('--json', default=None, help='Inline JSON block array')
|
||||
p.add_argument('--markdown', action='store_true', help='Treat --file as Markdown')
|
||||
args = p.parse_args()
|
||||
|
||||
try:
|
||||
if args.json:
|
||||
data = json.loads(args.json)
|
||||
elif args.file:
|
||||
if args.markdown or args.file.endswith('.md'):
|
||||
import subprocess
|
||||
r = subprocess.run(
|
||||
['python3', os.path.join(os.path.dirname(__file__), 'md_to_blocks.py'), args.file],
|
||||
capture_output=True, text=True
|
||||
)
|
||||
if r.returncode != 0:
|
||||
raise Exception(f"md_to_blocks failed: {r.stderr}")
|
||||
data = json.loads(r.stdout)
|
||||
else:
|
||||
with open(args.file) as f:
|
||||
data = json.load(f)
|
||||
else:
|
||||
data = json.load(sys.stdin)
|
||||
|
||||
if isinstance(data, list) and not isinstance(data, str):
|
||||
blocks = data
|
||||
else:
|
||||
blocks = json_to_blocks(data)
|
||||
write_blocks(args.node, blocks)
|
||||
print('OK')
|
||||
except Exception as e:
|
||||
print(f'Error: {e}', file=sys.stderr)
|
||||
sys.exit(1)
|
||||
Reference in New Issue
Block a user