332 lines
11 KiB
Python
332 lines
11 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
飞书电子表格剧本写入工具
|
||
将儿童互动英语剧本以8列格式写入飞书电子表格，支持知识点富文本标注。
|
||
|
||
用法:
|
||
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 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 format_cell_value(text, knowledge_points=None, is_user_line=False):
|
||
"""
|
||
将文本格式化为飞书富文本单元格值。
|
||
|
||
Args:
|
||
text: 单元格文本
|
||
knowledge_points: 知识点列表，如 ["afternoon", "adventure"]
|
||
is_user_line: 是否为User台词（User台词知识点=红色+粗体，NPC=粗体）
|
||
|
||
Returns:
|
||
飞书单元格值对象（带textFormatRuns）
|
||
"""
|
||
if not knowledge_points or not text:
|
||
return text
|
||
|
||
# 找出所有知识点的位置
|
||
format_runs = []
|
||
text_lower = text.lower()
|
||
|
||
# 去重并按在文本中的位置排序
|
||
found_kps = []
|
||
for kp in knowledge_points:
|
||
kp_lower = kp.lower()
|
||
idx = 0
|
||
while True:
|
||
idx = text_lower.find(kp_lower, idx)
|
||
if idx == -1:
|
||
break
|
||
found_kps.append({
|
||
"start": idx,
|
||
"end": idx + len(kp_lower),
|
||
"original": text[idx:idx + len(kp_lower)],
|
||
"kp": kp,
|
||
})
|
||
idx += len(kp_lower)
|
||
|
||
# 按位置排序，合并重叠区间
|
||
found_kps.sort(key=lambda x: x["start"])
|
||
|
||
if not found_kps:
|
||
return text
|
||
|
||
# 构建 textFormatRuns
|
||
for kp_info in found_kps:
|
||
fmt = {"bold": True}
|
||
if is_user_line:
|
||
fmt["foregroundColor"] = {
|
||
"red": 245, "green": 74, "blue": 69, "alpha": 255
|
||
} # #f54a45
|
||
format_runs.append({
|
||
"startIndex": kp_info["start"],
|
||
"format": fmt,
|
||
})
|
||
|
||
return {
|
||
"type": "text",
|
||
"text": text,
|
||
"textFormatRuns": format_runs,
|
||
}
|
||
|
||
|
||
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"]
|
||
print(json.dumps({
|
||
"token": spreadsheet["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)
|
||
|
||
# data format: list of rows, each row is a list of column values
|
||
# Column values can be strings or objects with {text, knowledge_points, is_user}
|
||
rows = data.get("rows", [])
|
||
knowledge_points = data.get("knowledge_points", [])
|
||
|
||
# Convert rows to API format
|
||
values = []
|
||
header_written = False
|
||
|
||
for row_idx, row in enumerate(rows):
|
||
formatted_row = []
|
||
for col_idx, cell in enumerate(row):
|
||
if isinstance(cell, dict):
|
||
is_user = cell.get("is_user", False)
|
||
formatted = format_cell_value(
|
||
cell.get("text", ""),
|
||
knowledge_points,
|
||
is_user_line=is_user,
|
||
)
|
||
else:
|
||
formatted = str(cell) if cell else ""
|
||
formatted_row.append(formatted)
|
||
values.append(formatted_row)
|
||
|
||
if not values:
|
||
print("⚠️ 没有数据可写入", file=sys.stderr)
|
||
return
|
||
|
||
# Calculate range
|
||
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:
|
||
# Append mode - use append API
|
||
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:
|
||
# Overwrite mode
|
||
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:H1",
|
||
"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 command
|
||
create_parser = subparsers.add_parser("create", help="创建新的电子表格")
|
||
create_parser.add_argument("--title", required=True, help="表格标题")
|
||
create_parser.add_argument("--credential", default="xiaobian", help="凭证名称")
|
||
|
||
# create-sheet command
|
||
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 command
|
||
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 command
|
||
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 command
|
||
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()
|