216 lines
6.8 KiB
Python
216 lines
6.8 KiB
Python
import json
|
||
import re
|
||
import requests
|
||
from urllib.parse import urljoin
|
||
|
||
# 基础URL地址
|
||
BASE_API_URL = "https://open.feishu.cn/open-apis/"
|
||
|
||
|
||
class BitableService:
|
||
"""
|
||
飞书多维表格服务类
|
||
"""
|
||
|
||
@staticmethod
|
||
def make_request(method, url, headers=None, params=None, json_data=None):
|
||
"""
|
||
发送请求到飞书API
|
||
|
||
Args:
|
||
method: 请求方法，GET/POST等
|
||
url: API路径，不含基础URL
|
||
headers: 请求头
|
||
params: URL参数
|
||
json_data: JSON数据体
|
||
|
||
Returns:
|
||
dict: 响应数据
|
||
"""
|
||
full_url = urljoin(BASE_API_URL, url)
|
||
|
||
if headers is None:
|
||
headers = {}
|
||
|
||
response = requests.request(
|
||
method=method,
|
||
url=full_url,
|
||
headers=headers,
|
||
params=params,
|
||
json=json_data
|
||
)
|
||
|
||
# 检查响应
|
||
if not response.ok:
|
||
error_msg = f"API 请求失败: {response.status_code}, 响应: {response.text}"
|
||
print(error_msg)
|
||
raise Exception(error_msg)
|
||
|
||
return response.json()
|
||
|
||
@staticmethod
|
||
def extract_params_from_url(table_url):
|
||
"""
|
||
从URL中提取app_token和table_id
|
||
|
||
Args:
|
||
table_url: 飞书多维表格URL
|
||
|
||
Returns:
|
||
tuple: (app_token, table_id) 元组
|
||
|
||
Raises:
|
||
ValueError: 如果无法从URL中提取必要参数
|
||
"""
|
||
app_token_match = re.search(r'base/([^?]+)', table_url)
|
||
table_id_match = re.search(r'table=([^&]+)', table_url)
|
||
|
||
if not app_token_match or not table_id_match:
|
||
raise ValueError("无法从URL中提取必要参数，请确认URL格式正确")
|
||
|
||
return app_token_match.group(1), table_id_match.group(1)
|
||
|
||
@staticmethod
|
||
def get_metadata(app_token, table_id, access_token):
|
||
"""
|
||
获取多维表格元数据
|
||
|
||
Args:
|
||
app_token: 应用令牌
|
||
table_id: 表格ID
|
||
access_token: 访问令牌
|
||
|
||
Returns:
|
||
dict: 表格元数据
|
||
"""
|
||
try:
|
||
# 构造请求
|
||
url = f"bitable/v1/apps/{app_token}/tables/{table_id}"
|
||
headers = {
|
||
"Authorization": f"Bearer {access_token}",
|
||
"Content-Type": "application/json"
|
||
}
|
||
|
||
# 发送请求
|
||
response = BitableService.make_request("GET", url, headers=headers)
|
||
|
||
# 检查响应
|
||
if response and "code" in response and response["code"] == 0:
|
||
return response.get("data", {}).get("table", {})
|
||
|
||
# 发生错误
|
||
error_msg = f"获取多维表格元数据失败: {json.dumps(response)}"
|
||
print(error_msg)
|
||
raise Exception(error_msg)
|
||
|
||
except Exception as e:
|
||
# 如果正常API调用失败，使用替代方法
|
||
print(f"获取多维表格元数据失败: {str(e)}")
|
||
# 简单返回一个基本名称
|
||
return {
|
||
"name": f"table_{table_id}",
|
||
"description": "自动创建的表格"
|
||
}
|
||
|
||
@staticmethod
|
||
def search_records(app_token, table_id, access_token, filter_exp=None, sort=None, page_size=20, page_token=None):
|
||
"""
|
||
查询多维表格记录
|
||
|
||
Args:
|
||
app_token: 应用令牌
|
||
table_id: 表格ID
|
||
access_token: 访问令牌
|
||
filter_exp: 过滤条件
|
||
sort: 排序条件
|
||
page_size: 每页大小
|
||
page_token: 分页标记
|
||
|
||
Returns:
|
||
dict: 查询结果
|
||
|
||
Raises:
|
||
Exception: 查询失败时抛出异常
|
||
"""
|
||
try:
|
||
# 构造请求URL
|
||
url = f"bitable/v1/apps/{app_token}/tables/{table_id}/records/search"
|
||
|
||
# 请求头
|
||
headers = {
|
||
"Authorization": f"Bearer {access_token}",
|
||
"Content-Type": "application/json"
|
||
}
|
||
|
||
# 查询参数
|
||
params = {"page_size": page_size}
|
||
if page_token:
|
||
params["page_token"] = page_token
|
||
|
||
# 请求体
|
||
json_data = {}
|
||
if filter_exp:
|
||
json_data["filter"] = filter_exp
|
||
if sort:
|
||
json_data["sort"] = sort
|
||
|
||
# 发送请求
|
||
response = BitableService.make_request("POST", url, headers=headers, params=params, json_data=json_data)
|
||
|
||
# 检查响应
|
||
if response and "code" in response and response["code"] == 0:
|
||
return response.get("data", {})
|
||
|
||
# 发生错误
|
||
error_msg = f"查询飞书多维表格失败: {json.dumps(response)}"
|
||
print(error_msg)
|
||
raise Exception(error_msg)
|
||
|
||
except Exception as e:
|
||
# 记录详细错误
|
||
print(f"查询飞书多维表格发生错误: {str(e)}")
|
||
raise
|
||
|
||
@staticmethod
|
||
def get_table_fields(app_token, table_id, access_token):
|
||
"""
|
||
获取多维表格的字段信息
|
||
|
||
Args:
|
||
app_token: 应用令牌
|
||
table_id: 表格ID
|
||
access_token: 访问令牌
|
||
|
||
Returns:
|
||
list: 字段信息列表
|
||
"""
|
||
try:
|
||
# 构造请求
|
||
url = f"bitable/v1/apps/{app_token}/tables/{table_id}/fields"
|
||
headers = {
|
||
"Authorization": f"Bearer {access_token}",
|
||
"Content-Type": "application/json"
|
||
}
|
||
params = {"page_size": 100}
|
||
|
||
# 发送请求
|
||
response = BitableService.make_request("GET", url, headers=headers, params=params)
|
||
|
||
# 检查响应
|
||
if response and "code" in response and response["code"] == 0:
|
||
return response.get("data", {}).get("items", [])
|
||
|
||
# 发生错误
|
||
error_msg = f"获取多维表格字段失败: {json.dumps(response)}"
|
||
print(error_msg)
|
||
raise Exception(error_msg)
|
||
|
||
except Exception as e:
|
||
# 记录详细错误
|
||
print(f"获取字段信息失败: {str(e)}")
|
||
# 如果获取失败，返回一个基本字段集
|
||
return [
|
||
{"field_name": "title", "type": "text", "property": {}},
|
||
{"field_name": "description", "type": "text", "property": {}},
|
||
{"field_name": "created_time", "type": "datetime", "property": {}}
|
||
] |