feat: 飞书知识库读写工具：增删查改节点、写入页面、创建早报、记录踩坑，支持 Markdown 表格原生渲染。
This commit is contained in:
NightStar
@@ -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,262 @@
|
||||
#!/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
|
||||
|
||||
Supports: headings, lists, dividers, inline formatting, and markdown tables.
|
||||
Table blocks (31) are output as nested structures with cell blocks (32).
|
||||
"""
|
||||
|
||||
import re, json, sys
|
||||
|
||||
TABLE_BLOCK = 31
|
||||
CELL_BLOCK = 32
|
||||
TEXT_BLOCK = 2
|
||||
PREFIX = 'ecc_'
|
||||
|
||||
def parse_inline(text):
|
||||
"""Parse a line of inline markdown into text_run elements."""
|
||||
elements = []
|
||||
pos = 0
|
||||
while pos < len(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
|
||||
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
|
||||
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
|
||||
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
|
||||
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:
|
||||
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 is_sep_line(line):
|
||||
"""Check if a line is a markdown table separator, e.g. |---|:---:|---|"""
|
||||
s = line.strip()
|
||||
if not s.startswith('|') or not s.endswith('|'):
|
||||
return False
|
||||
parts = s[1:-1].split('|')
|
||||
return all(re.match(r'^[\s\-:]+$', p) for p in parts)
|
||||
|
||||
def is_table_row(line):
|
||||
"""Check if a line looks like a markdown table row."""
|
||||
s = line.strip()
|
||||
return s.startswith('|') and s.endswith('|') and '|' in s[1:-1]
|
||||
|
||||
def split_row(line):
|
||||
return [c.strip() for c in line.strip()[1:-1].split('|')]
|
||||
|
||||
def parse_table(lines, start_idx):
|
||||
"""Parse a markdown table. Returns (blocks, end_idx) or (None, start_idx)."""
|
||||
i = start_idx
|
||||
headers = []
|
||||
has_header = False
|
||||
|
||||
# Look ahead: need at least 2 lines (header + separator) or 1 plain row
|
||||
if i >= len(lines) or not is_table_row(lines[i]):
|
||||
return None, start_idx
|
||||
|
||||
# First line: potential header
|
||||
first_cells = split_row(lines[i])
|
||||
if not first_cells:
|
||||
return None, start_idx
|
||||
|
||||
i += 1
|
||||
# Check for separator
|
||||
if i < len(lines) and is_sep_line(lines[i]):
|
||||
headers = first_cells
|
||||
has_header = True
|
||||
i += 1
|
||||
|
||||
# Collect data rows
|
||||
data_rows = []
|
||||
while i < len(lines) and is_table_row(lines[i]):
|
||||
row = split_row(lines[i])
|
||||
if row:
|
||||
data_rows.append(row)
|
||||
i += 1
|
||||
|
||||
# No header case: first line was a data row
|
||||
if not has_header:
|
||||
data_rows.insert(0, first_cells)
|
||||
|
||||
if not data_rows:
|
||||
return None, start_idx
|
||||
|
||||
num_rows = len(data_rows)
|
||||
num_cols = max(len(r) for r in data_rows)
|
||||
|
||||
# Build nested blocks
|
||||
blocks = []
|
||||
table_id = f'{PREFIX}t0'
|
||||
cell_ids = []
|
||||
|
||||
# Table container
|
||||
table_block = {
|
||||
'block_id': table_id,
|
||||
'block_type': TABLE_BLOCK,
|
||||
'table': {
|
||||
'property': {
|
||||
'row_size': num_rows,
|
||||
'column_size': num_cols,
|
||||
'header_row': has_header
|
||||
}
|
||||
},
|
||||
'children': [],
|
||||
'__table': True,
|
||||
'__rows': num_rows,
|
||||
'__cols': num_cols
|
||||
}
|
||||
|
||||
for row_idx, row in enumerate(data_rows):
|
||||
for col_idx in range(num_cols):
|
||||
cell_text = row[col_idx] if col_idx < len(row) else ''
|
||||
cell_id = f'{PREFIX}t0r{row_idx}c{col_idx}'
|
||||
text_id = f'{PREFIX}t0r{row_idx}c{col_idx}t'
|
||||
cell_ids.append(cell_id)
|
||||
|
||||
is_header_cell = has_header and row_idx == 0
|
||||
elements = parse_inline(cell_text)
|
||||
if not elements:
|
||||
elements = [{'text_run': {'content': cell_text, 'text_element_style': {}}}]
|
||||
if is_header_cell:
|
||||
for el in elements:
|
||||
if 'text_run' in el:
|
||||
el['text_run']['text_element_style']['bold'] = True
|
||||
|
||||
text_block = {
|
||||
'block_id': text_id,
|
||||
'block_type': TEXT_BLOCK,
|
||||
'text': {'elements': elements, 'style': {}},
|
||||
'children': []
|
||||
}
|
||||
cell_block = {
|
||||
'block_id': cell_id,
|
||||
'block_type': CELL_BLOCK,
|
||||
'table_cell': {},
|
||||
'children': [text_id]
|
||||
}
|
||||
|
||||
table_block['children'].append(cell_id)
|
||||
blocks.append(cell_block)
|
||||
blocks.append(text_block)
|
||||
|
||||
blocks.insert(0, table_block)
|
||||
return blocks, i
|
||||
|
||||
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():
|
||||
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() in ('---', '***'):
|
||||
blocks.append({'block_type': 22, 'divider': {}})
|
||||
i += 1
|
||||
continue
|
||||
|
||||
# Table detection
|
||||
if '|' in line:
|
||||
tbl, ni = parse_table(lines, i)
|
||||
if tbl is not None:
|
||||
blocks.extend(tbl)
|
||||
i = ni
|
||||
continue
|
||||
|
||||
# Headings
|
||||
h = re.match(r'^(#{1,3})\s+(.+)$', line)
|
||||
if h:
|
||||
level = len(h.group(1))
|
||||
bt = 2 + level
|
||||
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,155 @@
|
||||
#!/usr/bin/env python3
|
||||
"""写入 Wiki 页面内容。支持 Markdown 文件（含表格）和 Block JSON。
|
||||
表格块（block_type=31）自动走 descendant API 创建嵌套结构。
|
||||
"""
|
||||
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 strip_table_meta(block):
|
||||
"""Remove __table/__rows/__cols internal markers."""
|
||||
out = {}
|
||||
for k, v in block.items():
|
||||
if not k.startswith('__'):
|
||||
out[k] = v
|
||||
return out
|
||||
|
||||
def write_blocks(node_token, blocks):
|
||||
import requests
|
||||
token = get_token()
|
||||
h = {'Authorization': f'Bearer {token}', 'Content-Type': 'application/json'}
|
||||
|
||||
# Get doc_id
|
||||
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 existing content
|
||||
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 if isinstance(c, str) else 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)}
|
||||
)
|
||||
|
||||
# Separate table groups from normal blocks
|
||||
normal_blocks = []
|
||||
table_groups = []
|
||||
i = 0
|
||||
while i < len(blocks):
|
||||
b = blocks[i]
|
||||
if b.get('__table'):
|
||||
group = []
|
||||
table_prefix = b.get('block_id', '')
|
||||
j = i
|
||||
# Collect table + cells + cell content (all tagged with __table or have matching ecc_ block_id)
|
||||
while j < len(blocks):
|
||||
bj = blocks[j]
|
||||
bid = bj.get('block_id', '')
|
||||
# Same table group: __table marker, cell/child with matching prefix, or continuation
|
||||
if bj.get('__table') and j != i:
|
||||
break
|
||||
if bj.get('block_type') == 31 and not bj.get('__table'):
|
||||
break
|
||||
# Stop at unrelated blocks (non-table, non-cell blocks without matching prefix)
|
||||
if not bj.get('__table') and not bid.startswith('ecc_t'):
|
||||
# Only include empty text blocks with block_id (spacers next to table)
|
||||
if bj.get('block_type') == 2 and bid:
|
||||
elements = bj.get('text', {}).get('elements', [])
|
||||
if elements and all(e.get('text_run', {}).get('content', '') == '' for e in elements):
|
||||
pass # empty spacer included
|
||||
else:
|
||||
break
|
||||
else:
|
||||
break
|
||||
group.append(bj)
|
||||
j += 1
|
||||
table_groups.append(group)
|
||||
i = j
|
||||
else:
|
||||
normal_blocks.append(b)
|
||||
i += 1
|
||||
|
||||
# Write normal blocks in batches
|
||||
batch_size = 25
|
||||
for idx in range(0, len(normal_blocks), batch_size):
|
||||
batch = normal_blocks[idx:idx + 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 {idx // batch_size} failed: {r.json().get('msg')}")
|
||||
|
||||
# Write table groups via descendant API
|
||||
for group_idx, group in enumerate(table_groups):
|
||||
table_block = group[0]
|
||||
descendants = [strip_table_meta(b) for b in group]
|
||||
table_bid = table_block.get('block_id', table_block.get('children', [''])[0])
|
||||
payload = {
|
||||
'index': -1,
|
||||
'children_id': [table_bid],
|
||||
'descendants': descendants
|
||||
}
|
||||
r = requests.post(
|
||||
f'https://open.feishu.cn/open-apis/docx/v1/documents/{doc_id}/blocks/{doc_id}/descendant?document_revision_id=-1',
|
||||
headers=h, json=payload
|
||||
)
|
||||
rd = r.json()
|
||||
if rd.get('code') != 0:
|
||||
raise Exception(f"Table {group_idx} descendant API failed: {rd.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'):
|
||||
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 = 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