openclaw-home-pc/workspace/create_feishu_doc.py

#!/usr/bin/env python3
"""
使用飞书 API 创建云文档
"""
import requests
import json
import sys

# 飞书应用凭证
APP_ID = "cli_a93815b250b9dcb5"
APP_SECRET = "NogaPY8DiMHMyadOKDW26bqkGPnrOkND"

def get_tenant_access_token():
    """获取飞书 tenant_access_token"""
    url = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal"
    payload = {
        "app_id": APP_ID,
        "app_secret": APP_SECRET
    }
    resp = requests.post(url, json=payload, timeout=10)
    data = resp.json()
    if data.get("code") != 0:
        raise Exception(f"获取 access_token 失败：{data}")
    return data["tenant_access_token"]

def create_document(token, title, content):
    """创建飞书云文档"""
    url = "https://open.feishu.cn/open-apis/docx/v1/documents"
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }
    payload = {
        "title": title,
        "folder_token": "",  # 空表示创建在个人空间根目录
    }
    resp = requests.post(url, json=payload, headers=headers, timeout=10)
    data = resp.json()
    if data.get("code") != 0:
        raise Exception(f"创建文档失败：{data}")
    
    doc_token = data["data"]["document"]["token"]
    doc_id = data["data"]["document"]["doc_id"]
    
    # 更新文档内容
    update_document_content(token, doc_token, content)
    
    return doc_token, doc_id

def update_document_content(token, doc_token, content):
    """更新文档内容（使用批量更新 API）"""
    url = f"https://open.feishu.cn/open-apis/docx/v1/documents/{doc_token}/batch_update"
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }
    
    # 将 Markdown 内容转换为飞书文档块
    # 简单处理：将内容作为一个文本块插入
    blocks = []
    
    # 解析 Markdown 内容，转换为飞书文档块
    lines = content.split('\n')
    current_block = []
    block_type = "text"
    
    for line in lines:
        if line.startswith('# '):
            if current_block:
                blocks.append({"text": "\n".join(current_block)})
                current_block = []
            blocks.append({"heading1": {"elements": [{"text_run": {"content": line[2:]}}]}})
        elif line.startswith('## '):
            if current_block:
                blocks.append({"text": "\n".join(current_block)})
                current_block = []
            blocks.append({"heading2": {"elements": [{"text_run": {"content": line[3:]}}]}})
        elif line.startswith('### '):
            if current_block:
                blocks.append({"text": "\n".join(current_block)})
                current_block = []
            blocks.append({"heading3": {"elements": [{"text_run": {"content": line[4:]}}]}})
        elif line.startswith('> '):
            if current_block:
                blocks.append({"text": "\n".join(current_block)})
                current_block = []
            blocks.append({"quote": {"elements": [{"text_run": {"content": line[2:]}}]}})
        elif line.startswith('- ') or line.startswith('* '):
            if current_block:
                blocks.append({"text": "\n".join(current_block)})
                current_block = []
            blocks.append({"bullet": {"elements": [{"text_run": {"content": line[2:]}}]}})
        elif line.startswith('1. ') or line.startswith('2. ') or line.startswith('3. '):
            if current_block:
                blocks.append({"text": "\n".join(current_block)})
                current_block = []
            # 简单处理有序列表
            content_text = line.split('. ', 1)[1] if '. ' in line else line
            blocks.append({"ordered": {"elements": [{"text_run": {"content": content_text}}]}})
        elif line.startswith('```'):
            if current_block:
                blocks.append({"text": "\n".join(current_block)})
                current_block = []
            block_type = 'code' if block_type != 'code' else 'text'
        elif line == '---':
            if current_block:
                blocks.append({"text": "\n".join(current_block)})
                current_block = []
            blocks.append({"divider": {}})
        elif line.strip() == '':
            if current_block:
                blocks.append({"text": "\n".join(current_block)})
                current_block = []
        else:
            current_block.append(line)
    
    if current_block:
        blocks.append({"text": "\n".join(current_block)})
    
    # 构建请求体
    requests_body = {
        "requests": []
    }
    
    for i, block in enumerate(blocks):
        requests_body["requests"].append({
            "action": "append_block",
            "args": {
                "block": block,
                "parent_block_id": "",
                "index": i
            }
        })
    
    resp = requests.post(url, json=requests_body, headers=headers, timeout=30)
    data = resp.json()
    if data.get("code") != 0:
        print(f"警告：更新文档内容部分失败：{data}")
    else:
        print(f"成功更新 {len(blocks)} 个块")

def main():
    if len(sys.argv) < 3:
        print("用法：python3 create_feishu_doc.py <title> <content_file>")
        sys.exit(1)
    
    title = sys.argv[1]
    content_file = sys.argv[2]
    
    # 读取文档内容
    with open(content_file, 'r', encoding='utf-8') as f:
        content = f.read()
    
    print(f"创建文档：{title}")
    print(f"内容来源：{content_file}")
    
    # 获取访问令牌
    print("正在获取访问令牌...")
    token = get_tenant_access_token()
    print("✓ 获取访问令牌成功")
    
    # 创建文档
    print("正在创建文档...")
    doc_token, doc_id = create_document(token, title, content)
    print(f"✓ 文档创建成功")
    print(f"文档 Token: {doc_token}")
    print(f"文档 ID: {doc_id}")
    print(f"文档 URL: https://www.feishu.cn/docx/{doc_token}")
    
    return doc_token, doc_id

if __name__ == "__main__":
    main()