John Qiu
712063071c
skills/ → skills-dev(9), skills-req(10), skills-ops(4), skills-integration(8), skills-biz(4), skills-workflow(7) generate-marketplace.py 改为自动扫描所有 skills-* 目录。 Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
410 lines
13 KiB
Python
410 lines
13 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
飞书多维表格 Demo
|
||
使用 zhiyun.ai 凭证演示多维表格的完整操作流程
|
||
"""
|
||
|
||
import requests
|
||
from datetime import datetime, timedelta
|
||
from typing import Optional, List, Dict
|
||
|
||
# ========== 配置 ==========
|
||
ZHIYUN_APP_ID = "cli_a9f29dca82b9dbef"
|
||
ZHIYUN_APP_SECRET = "sDfhjG7QT1S4gfHiMVYSygmPQPN1R2Ho"
|
||
BASE_URL = "https://open.feishu.cn/open-apis"
|
||
|
||
|
||
# ========== 工具类 ==========
|
||
class FeishuBitable:
|
||
"""飞书多维表格操作工具类"""
|
||
|
||
def __init__(self, app_id: str = ZHIYUN_APP_ID, app_secret: str = ZHIYUN_APP_SECRET):
|
||
self.app_id = app_id
|
||
self.app_secret = app_secret
|
||
self._token = None
|
||
self._token_expires = None
|
||
|
||
@property
|
||
def token(self) -> str:
|
||
"""获取或刷新 access token"""
|
||
if self._token and self._token_expires and datetime.now() < self._token_expires:
|
||
return self._token
|
||
|
||
url = f"{BASE_URL}/auth/v3/tenant_access_token/internal"
|
||
response = requests.post(url, json={
|
||
"app_id": self.app_id,
|
||
"app_secret": self.app_secret
|
||
})
|
||
data = response.json()
|
||
|
||
if data.get("code") != 0:
|
||
raise Exception(f"获取 token 失败: {data}")
|
||
|
||
self._token = data["tenant_access_token"]
|
||
self._token_expires = datetime.now() + timedelta(seconds=data.get("expire", 7200) - 60)
|
||
print(f"[OK] Token 获取成功，有效期至 {self._token_expires}")
|
||
return self._token
|
||
|
||
@property
|
||
def headers(self) -> Dict[str, str]:
|
||
return {
|
||
"Authorization": f"Bearer {self.token}",
|
||
"Content-Type": "application/json"
|
||
}
|
||
|
||
# ========== 多维表格管理 ==========
|
||
def create_bitable(self, name: str, folder_token: str = None) -> Dict:
|
||
"""
|
||
创建新的多维表格
|
||
|
||
Args:
|
||
name: 多维表格名称
|
||
folder_token: 文件夹 token（可选，不指定则创建在根目录）
|
||
"""
|
||
url = f"{BASE_URL}/bitable/v1/apps"
|
||
payload = {"name": name}
|
||
if folder_token:
|
||
payload["folder_token"] = folder_token
|
||
|
||
response = requests.post(url, headers=self.headers, json=payload)
|
||
data = response.json()
|
||
|
||
if data.get("code") != 0:
|
||
raise Exception(f"创建多维表格失败: {data}")
|
||
|
||
app_info = data["data"]["app"]
|
||
print(f"[OK] 多维表格创建成功")
|
||
print(f" 名称: {app_info['name']}")
|
||
print(f" app_token: {app_info['app_token']}")
|
||
print(f" URL: {app_info.get('url', 'N/A')}")
|
||
return app_info
|
||
|
||
def list_tables(self, app_token: str) -> List[Dict]:
|
||
"""列出多维表格中的所有数据表"""
|
||
url = f"{BASE_URL}/bitable/v1/apps/{app_token}/tables"
|
||
response = requests.get(url, headers=self.headers)
|
||
data = response.json()
|
||
|
||
if data.get("code") != 0:
|
||
raise Exception(f"获取数据表列表失败: {data}")
|
||
|
||
tables = data["data"].get("items", [])
|
||
print(f"[OK] 找到 {len(tables)} 个数据表")
|
||
for t in tables:
|
||
print(f" - {t['name']} (table_id: {t['table_id']})")
|
||
return tables
|
||
|
||
def create_table(self, app_token: str, name: str, fields: List[Dict]) -> Dict:
|
||
"""
|
||
在多维表格中创建数据表
|
||
|
||
Args:
|
||
app_token: 多维表格 app_token
|
||
name: 数据表名称
|
||
fields: 字段定义列表
|
||
"""
|
||
url = f"{BASE_URL}/bitable/v1/apps/{app_token}/tables"
|
||
payload = {
|
||
"table": {
|
||
"name": name,
|
||
"default_view_name": "默认视图",
|
||
"fields": fields
|
||
}
|
||
}
|
||
|
||
response = requests.post(url, headers=self.headers, json=payload)
|
||
data = response.json()
|
||
|
||
if data.get("code") != 0:
|
||
raise Exception(f"创建数据表失败: {data}")
|
||
|
||
table_info = data["data"]
|
||
print(f"[OK] 数据表创建成功")
|
||
print(f" 名称: {name}")
|
||
print(f" table_id: {table_info['table_id']}")
|
||
return table_info
|
||
|
||
# ========== 记录操作 ==========
|
||
def list_records(self, app_token: str, table_id: str,
|
||
filter_str: str = None, page_size: int = 100) -> List[Dict]:
|
||
"""列出所有记录（自动分页）"""
|
||
url = f"{BASE_URL}/bitable/v1/apps/{app_token}/tables/{table_id}/records"
|
||
all_records = []
|
||
page_token = None
|
||
|
||
while True:
|
||
params = {"page_size": min(page_size, 500)}
|
||
if page_token:
|
||
params["page_token"] = page_token
|
||
if filter_str:
|
||
params["filter"] = filter_str
|
||
|
||
response = requests.get(url, headers=self.headers, params=params)
|
||
data = response.json()
|
||
|
||
if data.get("code") != 0:
|
||
raise Exception(f"查询失败: {data}")
|
||
|
||
items = data["data"].get("items", [])
|
||
all_records.extend(items)
|
||
|
||
if not data["data"].get("has_more"):
|
||
break
|
||
page_token = data["data"]["page_token"]
|
||
|
||
return all_records
|
||
|
||
def create_record(self, app_token: str, table_id: str, fields: Dict) -> Dict:
|
||
"""创建单条记录"""
|
||
url = f"{BASE_URL}/bitable/v1/apps/{app_token}/tables/{table_id}/records"
|
||
response = requests.post(url, headers=self.headers, json={"fields": fields})
|
||
data = response.json()
|
||
|
||
if data.get("code") != 0:
|
||
raise Exception(f"创建失败: {data}")
|
||
return data["data"]["record"]
|
||
|
||
def batch_create(self, app_token: str, table_id: str,
|
||
records: List[Dict], batch_size: int = 500) -> List[Dict]:
|
||
"""批量创建记录"""
|
||
url = f"{BASE_URL}/bitable/v1/apps/{app_token}/tables/{table_id}/records/batch_create"
|
||
created = []
|
||
|
||
for i in range(0, len(records), batch_size):
|
||
batch = [{"fields": r} if "fields" not in r else r for r in records[i:i+batch_size]]
|
||
response = requests.post(url, headers=self.headers, json={"records": batch})
|
||
data = response.json()
|
||
|
||
if data.get("code") != 0:
|
||
raise Exception(f"批量创建失败: {data}")
|
||
created.extend(data["data"]["records"])
|
||
|
||
return created
|
||
|
||
def get_fields(self, app_token: str, table_id: str) -> List[Dict]:
|
||
"""获取字段定义"""
|
||
url = f"{BASE_URL}/bitable/v1/apps/{app_token}/tables/{table_id}/fields"
|
||
response = requests.get(url, headers=self.headers)
|
||
data = response.json()
|
||
|
||
if data.get("code") != 0:
|
||
raise Exception(f"获取字段失败: {data}")
|
||
return data["data"]["items"]
|
||
|
||
|
||
# ========== Demo 函数 ==========
|
||
def demo_verify_credentials():
|
||
"""验证凭证是否有效"""
|
||
print("\n" + "="*50)
|
||
print("Step 1: 验证飞书应用凭证")
|
||
print("="*50)
|
||
|
||
bitable = FeishuBitable()
|
||
token = bitable.token # 触发 token 获取
|
||
print(f" Token 前缀: {token[:20]}...")
|
||
return bitable
|
||
|
||
|
||
def demo_create_bitable(bitable: FeishuBitable) -> str:
|
||
"""创建新的多维表格"""
|
||
print("\n" + "="*50)
|
||
print("Step 2: 创建多维表格")
|
||
print("="*50)
|
||
|
||
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||
name = f"Claude Code Demo - {timestamp}"
|
||
|
||
app_info = bitable.create_bitable(name)
|
||
return app_info["app_token"]
|
||
|
||
|
||
def demo_create_task_table(bitable: FeishuBitable, app_token: str) -> str:
|
||
"""创建任务管理数据表"""
|
||
print("\n" + "="*50)
|
||
print("Step 3: 创建任务管理数据表")
|
||
print("="*50)
|
||
|
||
# 定义字段
|
||
fields = [
|
||
{
|
||
"field_name": "任务名称",
|
||
"type": 1 # 文本
|
||
},
|
||
{
|
||
"field_name": "状态",
|
||
"type": 3, # 单选
|
||
"property": {
|
||
"options": [
|
||
{"name": "待处理", "color": 0},
|
||
{"name": "进行中", "color": 1},
|
||
{"name": "已完成", "color": 2},
|
||
{"name": "已取消", "color": 3}
|
||
]
|
||
}
|
||
},
|
||
{
|
||
"field_name": "优先级",
|
||
"type": 3, # 单选
|
||
"property": {
|
||
"options": [
|
||
{"name": "高", "color": 4},
|
||
{"name": "中", "color": 5},
|
||
{"name": "低", "color": 6}
|
||
]
|
||
}
|
||
},
|
||
{
|
||
"field_name": "负责人",
|
||
"type": 1 # 文本
|
||
},
|
||
{
|
||
"field_name": "截止日期",
|
||
"type": 5, # 日期
|
||
"property": {
|
||
"date_formatter": "yyyy/MM/dd"
|
||
}
|
||
},
|
||
{
|
||
"field_name": "工时(小时)",
|
||
"type": 2 # 数字
|
||
},
|
||
{
|
||
"field_name": "备注",
|
||
"type": 1 # 文本
|
||
}
|
||
]
|
||
|
||
table_info = bitable.create_table(app_token, "任务看板", fields)
|
||
return table_info["table_id"]
|
||
|
||
|
||
def demo_add_sample_data(bitable: FeishuBitable, app_token: str, table_id: str):
|
||
"""添加示例数据"""
|
||
print("\n" + "="*50)
|
||
print("Step 4: 添加示例数据")
|
||
print("="*50)
|
||
|
||
# 准备示例数据
|
||
now = datetime.now()
|
||
sample_tasks = [
|
||
{
|
||
"任务名称": "完成产品需求文档",
|
||
"状态": "已完成",
|
||
"优先级": "高",
|
||
"负责人": "张三",
|
||
"截止日期": int((now - timedelta(days=2)).timestamp() * 1000),
|
||
"工时(小时)": 8,
|
||
"备注": "PRD 已评审通过"
|
||
},
|
||
{
|
||
"任务名称": "设计系统架构方案",
|
||
"状态": "进行中",
|
||
"优先级": "高",
|
||
"负责人": "李四",
|
||
"截止日期": int((now + timedelta(days=3)).timestamp() * 1000),
|
||
"工时(小时)": 16,
|
||
"备注": "正在编写技术方案"
|
||
},
|
||
{
|
||
"任务名称": "开发用户登录模块",
|
||
"状态": "待处理",
|
||
"优先级": "中",
|
||
"负责人": "王五",
|
||
"截止日期": int((now + timedelta(days=7)).timestamp() * 1000),
|
||
"工时(小时)": 24,
|
||
"备注": "等待架构方案确定"
|
||
},
|
||
{
|
||
"任务名称": "编写单元测试",
|
||
"状态": "待处理",
|
||
"优先级": "中",
|
||
"负责人": "赵六",
|
||
"截止日期": int((now + timedelta(days=10)).timestamp() * 1000),
|
||
"工时(小时)": 12,
|
||
"备注": ""
|
||
},
|
||
{
|
||
"任务名称": "部署测试环境",
|
||
"状态": "待处理",
|
||
"优先级": "低",
|
||
"负责人": "钱七",
|
||
"截止日期": int((now + timedelta(days=14)).timestamp() * 1000),
|
||
"工时(小时)": 4,
|
||
"备注": "需要申请服务器资源"
|
||
}
|
||
]
|
||
|
||
# 批量创建记录
|
||
created = bitable.batch_create(app_token, table_id, sample_tasks)
|
||
print(f"[OK] 成功创建 {len(created)} 条示例记录")
|
||
|
||
# 显示创建的记录
|
||
for i, record in enumerate(created, 1):
|
||
fields = record["fields"]
|
||
print(f" {i}. {fields.get('任务名称')} - {fields.get('状态')} ({fields.get('优先级')}优先级)")
|
||
|
||
return created
|
||
|
||
|
||
def demo_query_data(bitable: FeishuBitable, app_token: str, table_id: str):
|
||
"""查询数据演示"""
|
||
print("\n" + "="*50)
|
||
print("Step 5: 查询数据演示")
|
||
print("="*50)
|
||
|
||
# 查询所有记录
|
||
all_records = bitable.list_records(app_token, table_id)
|
||
print(f"[OK] 共 {len(all_records)} 条记录")
|
||
|
||
# 获取字段定义
|
||
fields = bitable.get_fields(app_token, table_id)
|
||
print(f"[OK] 共 {len(fields)} 个字段:")
|
||
for f in fields:
|
||
print(f" - {f['field_name']} (类型: {f['type']})")
|
||
|
||
return all_records
|
||
|
||
|
||
def main():
|
||
"""运行完整 Demo"""
|
||
print("\n" + "#"*60)
|
||
print("# 飞书多维表格 Demo - zhiyun.ai")
|
||
print("#"*60)
|
||
|
||
try:
|
||
# Step 1: 验证凭证
|
||
bitable = demo_verify_credentials()
|
||
|
||
# Step 2: 创建多维表格
|
||
app_token = demo_create_bitable(bitable)
|
||
|
||
# Step 3: 创建数据表
|
||
table_id = demo_create_task_table(bitable, app_token)
|
||
|
||
# Step 4: 添加示例数据
|
||
demo_add_sample_data(bitable, app_token, table_id)
|
||
|
||
# Step 5: 查询数据
|
||
demo_query_data(bitable, app_token, table_id)
|
||
|
||
# 完成
|
||
print("\n" + "="*60)
|
||
print("Demo 完成!")
|
||
print("="*60)
|
||
print(f"\n多维表格信息:")
|
||
print(f" app_token: {app_token}")
|
||
print(f" table_id: {table_id}")
|
||
print(f"\n访问地址:")
|
||
print(f" https://zhiyun-ai.feishu.cn/base/{app_token}?table={table_id}")
|
||
print()
|
||
|
||
return app_token, table_id
|
||
|
||
except Exception as e:
|
||
print(f"\n[ERROR] {e}")
|
||
raise
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|