feishu-wiki-tools/scripts/wiki_write.py

#!/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)