ai_member_xiaobian/scripts/feishu_sheet_writer.py

#!/usr/bin/env python3
"""
飞书电子表格剧本写入工具
将儿童互动英语剧本以4列格式写入飞书电子表格。
自动去除台词中的 ** 知识点标记符，输出干净纯文本。

用法:
  python scripts/feishu_sheet_writer.py create --title "U22 剧本" --credential xiaobian
  python scripts/feishu_sheet_writer.py write --token <spreadsheet_token> --sheet <sheet_id> \
      --data <json_file> --credential xiaobian
  python scripts/feishu_sheet_writer.py append --token <spreadsheet_token> --sheet <sheet_id> \
      --data <json_file> --credential xiaobian
  python scripts/feishu_sheet_writer.py create-sheet --token <spreadsheet_token> \
      --title "U22_L1_起" --credential xiaobian
"""

import json
import sys
import os
import re
import argparse
import requests

BASE_URL = "https://open.feishu.cn/open-apis"
CRED_DIR = "/root/.openclaw/credentials"


def get_token(credential_name):
    """获取 Bot tenant_access_token"""
    config_path = os.path.join(CRED_DIR, credential_name, "config.json")
    if not os.path.exists(config_path):
        print(f"❌ 凭证文件不存在: {config_path}", file=sys.stderr)
        sys.exit(1)

    with open(config_path) as f:
        config = json.load(f)

    app_id = config.get("apps", [{}])[0].get("appId", "")
    app_secret = config.get("apps", [{}])[0].get("appSecret", "")

    resp = requests.post(
        f"{BASE_URL}/auth/v3/tenant_access_token/internal",
        json={"app_id": app_id, "app_secret": app_secret},
        timeout=10,
    )
    data = resp.json()
    if data.get("code") != 0:
        print(f"❌ 获取token失败: {data}", file=sys.stderr)
        sys.exit(1)
    return data["tenant_access_token"]


def set_tenant_editable(spreadsheet_token, token):
    """
    设置表格权限：组织内任何人可通过链接编辑。
    非致命操作，失败不阻塞主流程。
    """
    resp = requests.patch(
        f"{BASE_URL}/drive/v1/permissions/{spreadsheet_token}/public?type=sheet",
        headers={
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json",
        },
        json={"link_share_entity": "tenant_editable"},
        timeout=10,
    )
    data = resp.json()
    if data.get("code") != 0:
        print(f"⚠️ 设置权限失败（非致命）: {data}", file=sys.stderr)
    else:
        print(json.dumps({"permission": "tenant_editable"}))


def strip_kp_markers(text):
    """
    去除文本中的 ** 知识点标记符，返回干净纯文本。
    Feishu Sheets V2 不支持单元格内富文本，所有台词以纯文本存储。
    """
    if not isinstance(text, str):
        return text
    return re.sub(r'\*\*([^*]+)\*\*', r'\1', text)


def create_spreadsheet(title, token):
    """创建新的电子表格"""
    resp = requests.post(
        f"{BASE_URL}/sheets/v3/spreadsheets",
        headers={
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json",
        },
        json={"title": title},
        timeout=10,
    )
    data = resp.json()
    if data.get("code") != 0:
        print(f"❌ 创建表格失败: {data}", file=sys.stderr)
        sys.exit(1)
    
    spreadsheet = data["data"]["spreadsheet"]
    spreadsheet_token = spreadsheet["spreadsheet_token"]
    
    # 自动设置组织内可编辑权限
    set_tenant_editable(spreadsheet_token, token)
    
    print(json.dumps({
        "token": spreadsheet_token,
        "url": spreadsheet["url"],
        "title": spreadsheet["title"],
    }))


def create_sheet(spreadsheet_token, title, token):
    """在已有表格中创建新sheet"""
    resp = requests.post(
        f"{BASE_URL}/sheets/v2/spreadsheets/{spreadsheet_token}/sheets_batch_update",
        headers={
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json",
        },
        json={
            "requests": [{
                "addSheet": {
                    "properties": {"title": title},
                }
            }]
        },
        timeout=10,
    )
    data = resp.json()
    if data.get("code") != 0:
        print(f"❌ 创建sheet失败: {data}", file=sys.stderr)
        sys.exit(1)
    
    reply = data["data"]["replies"][0]
    props = reply["addSheet"]["properties"]
    print(json.dumps({
        "sheet_id": props["sheetId"],
        "title": props["title"],
        "index": props.get("index", -1),
    }))


def write_data(spreadsheet_token, sheet_id, data_file, token, append=False, start_row=1):
    """写入或追加数据到表格。所有单元格均为纯文本，** 标记符自动去除。"""
    with open(data_file) as f:
        data = json.load(f)
    
    rows = data.get("rows", [])
    
    # Convert rows to plain-text API format, stripping ** markers
    values = []
    for row in rows:
        formatted_row = []
        for cell in row:
            if isinstance(cell, dict):
                # 兼容旧格式：dict → 提取 text 字段
                formatted = strip_kp_markers(cell.get("text", ""))
            elif isinstance(cell, str):
                formatted = strip_kp_markers(cell)
            else:
                formatted = str(cell) if cell else ""
            formatted_row.append(formatted)
        values.append(formatted_row)
    
    if not values:
        print("⚠️ 没有数据可写入", file=sys.stderr)
        return
    
    num_rows = len(values)
    num_cols = max(len(row) for row in values) if values else 1
    col_letter = chr(64 + num_cols) if num_cols <= 26 else "Z"
    
    if append:
        resp = requests.post(
            f"{BASE_URL}/sheets/v2/spreadsheets/{spreadsheet_token}/values_append",
            headers={
                "Authorization": f"Bearer {token}",
                "Content-Type": "application/json",
            },
            json={
                "valueRange": {
                    "range": f"{sheet_id}!A{start_row}:{col_letter}{start_row + num_rows - 1}",
                    "values": values,
                },
                "insertDataOption": "INSERT_ROWS",
            },
            timeout=30,
        )
    else:
        resp = requests.put(
            f"{BASE_URL}/sheets/v2/spreadsheets/{spreadsheet_token}/values",
            headers={
                "Authorization": f"Bearer {token}",
                "Content-Type": "application/json",
            },
            json={
                "valueRange": {
                    "range": f"{sheet_id}!A1:{col_letter}{num_rows}",
                    "values": values,
                },
            },
            timeout=30,
        )
    
    data_resp = resp.json()
    if data_resp.get("code") != 0:
        print(f"❌ 写入数据失败: {data_resp}", file=sys.stderr)
        sys.exit(1)
    
    print(json.dumps({
        "status": "ok",
        "rows": num_rows,
        "columns": num_cols,
        "mode": "append" if append else "overwrite",
    }))


def set_styles(spreadsheet_token, sheet_id, ranges_styles, token):
    """设置表头样式（粗体 + 灰底）"""
    resp = requests.put(
        f"{BASE_URL}/sheets/v2/spreadsheets/{spreadsheet_token}/style",
        headers={
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json",
        },
        json={
            "appendStyle": {
                "range": f"{sheet_id}!A1:D1",
                "style": {
                    "font": {"bold": True},
                    "backColor": "#e8e8e8",
                },
            },
        },
        timeout=10,
    )
    data = resp.json()
    if data.get("code") != 0:
        print(f"⚠️ 设置样式失败（非致命）: {data}", file=sys.stderr)
    else:
        print(json.dumps({"style_status": "ok"}))


def main():
    parser = argparse.ArgumentParser(description="飞书电子表格剧本写入工具")
    subparsers = parser.add_subparsers(dest="command", required=True)
    
    create_parser = subparsers.add_parser("create", help="创建新的电子表格")
    create_parser.add_argument("--title", required=True, help="表格标题")
    create_parser.add_argument("--credential", default="xiaobian", help="凭证名称")
    
    sheet_parser = subparsers.add_parser("create-sheet", help="在已有表格中创建新sheet")
    sheet_parser.add_argument("--token", required=True, help="spreadsheet token")
    sheet_parser.add_argument("--title", required=True, help="sheet标题")
    sheet_parser.add_argument("--credential", default="xiaobian", help="凭证名称")
    
    write_parser = subparsers.add_parser("write", help="写入数据（覆盖）")
    write_parser.add_argument("--token", required=True, help="spreadsheet token")
    write_parser.add_argument("--sheet", required=True, help="sheet ID")
    write_parser.add_argument("--data", required=True, help="数据JSON文件路径")
    write_parser.add_argument("--credential", default="xiaobian", help="凭证名称")
    
    append_parser = subparsers.add_parser("append", help="追加数据")
    append_parser.add_argument("--token", required=True, help="spreadsheet token")
    append_parser.add_argument("--sheet", required=True, help="sheet ID")
    append_parser.add_argument("--data", required=True, help="数据JSON文件路径")
    append_parser.add_argument("--start-row", type=int, default=1, help="起始行号")
    append_parser.add_argument("--credential", default="xiaobian", help="凭证名称")
    
    style_parser = subparsers.add_parser("style", help="设置表头样式")
    style_parser.add_argument("--token", required=True, help="spreadsheet token")
    style_parser.add_argument("--sheet", required=True, help="sheet ID")
    style_parser.add_argument("--credential", default="xiaobian", help="凭证名称")
    
    args = parser.parse_args()
    token = get_token(args.credential)
    
    if args.command == "create":
        create_spreadsheet(args.title, token)
    elif args.command == "create-sheet":
        create_sheet(args.token, args.title, token)
    elif args.command == "write":
        write_data(args.token, args.sheet, args.data, token, append=False)
        set_styles(args.token, args.sheet, None, token)
    elif args.command == "append":
        write_data(args.token, args.sheet, args.data, token, append=True, start_row=args.start_row)
    elif args.command == "style":
        set_styles(args.token, args.sheet, None, token)


if __name__ == "__main__":
    main()