"""
飞书 API 实现函数
所有纯 API 调用函数，不包含 MCP 框架相关的逻辑
可以被测试直接调用
"""
import lark_oapi as lark
from lark_oapi.api.bitable.v1 import (
App,
AppTable,
AppTableRecord,
CreateAppRequest,
CopyAppRequest,
CopyAppRequestBody,
CreateAppTableRequest,
CreateAppTableRequestBody,
CreateAppTableRecordRequest,
UpdateAppTableRecordRequest,
SearchAppTableRecordRequest,
SearchAppTableRecordRequestBody,
BatchCreateAppTableRecordRequest,
BatchCreateAppTableRecordRequestBody,
BatchUpdateAppTableRecordRequest,
BatchUpdateAppTableRecordRequestBody,
BatchGetAppTableRecordRequest,
BatchGetAppTableRecordRequestBody,
BatchDeleteAppTableRecordRequest,
BatchDeleteAppTableRecordRequestBody,
)
from yuppie_mcp_feishu.client import get_client
# ===== 应用级实现函数 =====
def create_bitable_app_impl(name: str, folder_token: str = ""):
"""
创建多维表格应用（实现）
参数:
name: 多维表格名称
folder_token: (可选) 文件夹token，指定创建位置
返回:
应用响应对象
"""
client = get_client()
request = (
CreateAppRequest.builder()
.request_body(App.builder().name(name).folder_token(folder_token).build())
.build()
)
return client.bitable.v1.app.create(request)
def copy_bitable_app_impl(app_token: str, name: str, folder_token: str = ""):
"""
复制多维表格应用（实现）
参数:
app_token: 要复制的多维表格token
name: 新多维表格的名称
folder_token: (可选) 文件夹token，指定创建位置
返回:
应用响应对象
"""
client = get_client()
request = (
CopyAppRequest.builder()
.app_token(app_token)
.request_body(
CopyAppRequestBody.builder().name(name).folder_token(folder_token).build()
)
.build()
)
return client.bitable.v1.app.copy(request)
def create_bitable_table_impl(app_token: str, table_name: str, fields: list = None):
"""
创建数据表（实现）
参数:
app_token: 多维表格的app_token
table_name: 数据表名称
fields: (可选) 字段定义列表，每个字段为dict格式
返回:
数据表响应对象
"""
client = get_client()
# 构建AppTable对象
table_builder = AppTable.builder().name(table_name)
table = table_builder.build()
# 设置字段定义(如果提供)
if fields:
table.fields = fields
# 构建请求
request = (
CreateAppTableRequest.builder()
.app_token(app_token)
.request_body(CreateAppTableRequestBody.builder().table(table).build())
.build()
)
return client.bitable.v1.app_table.create(request)
# ===== 记录级实现函数 =====
def create_bitable_record_impl(app_token: str, table_id: str, fields: dict):
"""
创建记录（实现）
参数:
app_token: 多维表格的token
table_id: 数据表ID
fields: 记录字段字典
返回:
记录响应对象
"""
client = get_client()
request = (
CreateAppTableRecordRequest.builder()
.app_token(app_token)
.table_id(table_id)
.request_body(AppTableRecord.builder().fields(fields).build())
.build()
)
return client.bitable.v1.app_table_record.create(request)
def update_bitable_record_impl(app_token: str, table_id: str, record_id: str, fields: dict):
"""
更新记录（实现）
参数:
app_token: 多维表格的token
table_id: 数据表ID
record_id: 记录ID
fields: 要更新的字段字典
返回:
记录响应对象
"""
client = get_client()
request = (
UpdateAppTableRecordRequest.builder()
.app_token(app_token)
.table_id(table_id)
.record_id(record_id)
.request_body(AppTableRecord.builder().fields(fields).build())
.build()
)
return client.bitable.v1.app_table_record.update(request)
def search_bitable_records_impl(
app_token: str,
table_id: str,
page_size: int = 20,
page_token: str = "",
filter: dict = None,
):
"""
查询记录（实现）
参数:
app_token: 多维表格的token
table_id: 数据表ID
page_size: 每页记录数
page_token: 分页token
filter: 过滤条件
返回:
记录响应对象
"""
client = get_client()
request_builder = (
SearchAppTableRecordRequest.builder()
.app_token(app_token)
.table_id(table_id)
.page_size(page_size)
)
if page_token:
request_builder.page_token(page_token)
body_builder = SearchAppTableRecordRequestBody.builder()
if filter:
# TODO: 需要根据SDK文档正确构建filter
pass
request_builder.request_body(body_builder.build())
request = request_builder.build()
return client.bitable.v1.app_table_record.search(request)
def batch_create_bitable_records_impl(app_token: str, table_id: str, records: list[dict]):
"""
批量创建记录（实现）
参数:
app_token: 多维表格的token
table_id: 数据表ID
records: 记录列表
返回:
记录响应对象
"""
client = get_client()
request = (
BatchCreateAppTableRecordRequest.builder()
.app_token(app_token)
.table_id(table_id)
.request_body(
BatchCreateAppTableRecordRequestBody.builder().records(records).build()
)
.build()
)
return client.bitable.v1.app_table_record.batch_create(request)
def batch_update_bitable_records_impl(app_token: str, table_id: str, records: list[dict]):
"""
批量更新记录（实现）
参数:
app_token: 多维表格的token
table_id: 数据表ID
records: 要更新的记录列表
返回:
记录响应对象
"""
client = get_client()
request = (
BatchUpdateAppTableRecordRequest.builder()
.app_token(app_token)
.table_id(table_id)
.request_body(
BatchUpdateAppTableRecordRequestBody.builder().records(records).build()
)
.build()
)
return client.bitable.v1.app_table_record.batch_update(request)
def batch_get_bitable_records_impl(app_token: str, table_id: str, record_ids: list[str]):
"""
批量获取记录（实现）
参数:
app_token: 多维表格的token
table_id: 数据表ID
record_ids: 记录ID列表
返回:
记录响应对象
"""
client = get_client()
request = (
BatchGetAppTableRecordRequest.builder()
.app_token(app_token)
.table_id(table_id)
.request_body(
BatchGetAppTableRecordRequestBody.builder().record_ids(record_ids).build()
)
.build()
)
return client.bitable.v1.app_table_record.batch_get(request)
def batch_delete_bitable_records_impl(app_token: str, table_id: str, record_ids: list[str]):
"""
批量删除记录（实现）
参数:
app_token: 多维表格的token
table_id: 数据表ID
record_ids: 要删除的记录ID列表
返回:
记录响应对象
"""
client = get_client()
request = (
BatchDeleteAppTableRecordRequest.builder()
.app_token(app_token)
.table_id(table_id)
.request_body(
BatchDeleteAppTableRecordRequestBody.builder().records(record_ids).build()
)
.build()
)
return client.bitable.v1.app_table_record.batch_delete(request)
__all__ = [
# 应用级
"create_bitable_app_impl",
"copy_bitable_app_impl",
"create_bitable_table_impl",
# 记录级
"create_bitable_record_impl",
"update_bitable_record_impl",
"search_bitable_records_impl",
"batch_create_bitable_records_impl",
"batch_update_bitable_records_impl",
"batch_get_bitable_records_impl",
"batch_delete_bitable_records_impl",
# Export 函数（供测试使用）
"create_bitable_app",
"copy_bitable_app",
"create_bitable_table",
"create_bitable_record",
"update_bitable_record",
"search_bitable_records",
"batch_create_bitable_records",
"batch_update_bitable_records",
"batch_get_bitable_records",
"batch_delete_bitable_records",
]
# ===== Export 函数（供测试使用） =====
def create_bitable_app(name: str, folder_token: str = "") -> str:
"""创建多维表格应用"""
response = create_bitable_app_impl(name, folder_token)
return lark.JSON.marshal(response.data, indent=4)
def copy_bitable_app(app_token: str, name: str, folder_token: str = "") -> str:
"""复制多维表格应用"""
response = copy_bitable_app_impl(app_token, name, folder_token)
return lark.JSON.marshal(response.data, indent=4)
def create_bitable_table(app_token: str, table_name: str, fields: list = None) -> str:
"""
创建数据表
参数:
app_token: 多维表格的app_token
table_name: 数据表名称
fields: (可选) 字段定义列表
返回:
JSON格式的数据表信息
"""
response = create_bitable_table_impl(app_token, table_name, fields)
if not response.success():
error_data = {"success": False, "code": response.code, "msg": response.msg}
return lark.JSON.marshal(error_data, indent=4)
return lark.JSON.marshal(response.data, indent=4)
def create_bitable_record(app_token: str, table_id: str, fields: dict) -> str:
"""创建记录"""
response = create_bitable_record_impl(app_token, table_id, fields)
if not response.success():
error_data = {
"success": False,
"code": response.code,
"msg": response.msg,
"error": response.error,
}
return lark.JSON.marshal(error_data, indent=4)
return lark.JSON.marshal(response.data, indent=4)
def update_bitable_record(app_token: str, table_id: str, record_id: str, fields: dict) -> str:
"""更新记录"""
response = update_bitable_record_impl(app_token, table_id, record_id, fields)
if not response.success():
error_data = {
"success": False,
"code": response.code,
"msg": response.msg,
"error": response.error,
}
return lark.JSON.marshal(error_data, indent=4)
return lark.JSON.marshal(response.data, indent=4)
def search_bitable_records(
app_token: str,
table_id: str,
page_size: int = 20,
page_token: str = "",
filter: dict = None,
) -> str:
"""查询记录"""
response = search_bitable_records_impl(
app_token, table_id, page_size, page_token, filter
)
if not response.success():
error_data = {
"success": False,
"code": response.code,
"msg": response.msg,
"error": response.error,
}
return lark.JSON.marshal(error_data, indent=4)
return lark.JSON.marshal(response.data, indent=4)
def batch_create_bitable_records(app_token: str, table_id: str, records: list[dict]) -> str:
"""批量创建记录"""
response = batch_create_bitable_records_impl(app_token, table_id, records)
if not response.success():
error_data = {
"success": False,
"code": response.code,
"msg": response.msg,
"error": response.error,
}
return lark.JSON.marshal(error_data, indent=4)
return lark.JSON.marshal(response.data, indent=4)
def batch_update_bitable_records(app_token: str, table_id: str, records: list[dict]) -> str:
"""批量更新记录"""
response = batch_update_bitable_records_impl(app_token, table_id, records)
if not response.success():
error_data = {
"success": False,
"code": response.code,
"msg": response.msg,
"error": response.error,
}
return lark.JSON.marshal(error_data, indent=4)
return lark.JSON.marshal(response.data, indent=4)
def batch_get_bitable_records(app_token: str, table_id: str, record_ids: list[str]) -> str:
"""批量获取记录"""
response = batch_get_bitable_records_impl(app_token, table_id, record_ids)
if not response.success():
error_data = {
"success": False,
"code": response.code,
"msg": response.msg,
"error": response.error,
}
return lark.JSON.marshal(error_data, indent=4)
return lark.JSON.marshal(response.data, indent=4)
def batch_delete_bitable_records(app_token: str, table_id: str, record_ids: list[str]) -> str:
"""批量删除记录"""
response = batch_delete_bitable_records_impl(app_token, table_id, record_ids)
if not response.success():
error_data = {
"success": False,
"code": response.code,
"msg": response.msg,
"error": response.error,
}
return lark.JSON.marshal(error_data, indent=4)
return lark.JSON.marshal(response.data, indent=4)
