330 lines
12 KiB
Python
330 lines
12 KiB
Python
|
import json
|
|||
|
|
import re
|
|||
|
|
import logging
|
|||
|
|
import requests
|
|||
|
|
from urllib.parse import urljoin, urlparse, parse_qs
|
|||
|
|
|
|||
|
|
# 基础URL地址
|
|||
|
|
BASE_API_URL = "https://open.feishu.cn/open-apis/"
|
|||
|
|
|
|||
|
|
# 获取日志记录器
|
|||
|
|
logger = logging.getLogger(__name__)
|
|||
|
|
|
|||
|
|
|
|||
|
|
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 = {}
|
|||
|
|
|
|||
|
|
# 记录请求信息，避免记录敏感信息
|
|||
|
|
logger.info(f"请求飞书API: {method} {full_url}")
|
|||
|
|
if params:
|
|||
|
|
logger.info(f"请求参数: {params}")
|
|||
|
|
|
|||
|
|
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}"
|
|||
|
|
logger.error(error_msg)
|
|||
|
|
|
|||
|
|
# 解析错误响应
|
|||
|
|
try:
|
|||
|
|
error_json = response.json()
|
|||
|
|
error_code = error_json.get("code")
|
|||
|
|
error_msg = error_json.get("msg", "")
|
|||
|
|
|
|||
|
|
# 根据错误代码提供更具体的错误信息
|
|||
|
|
if error_code == 91402: # NOTEXIST
|
|||
|
|
error_detail = "请求的资源不存在，请检查app_token和table_id是否正确，以及应用是否有权限访问该资源"
|
|||
|
|
logger.error(f"资源不存在错误: {error_detail}")
|
|||
|
|
raise Exception(f"资源不存在: {error_detail}")
|
|||
|
|
elif error_code == 99991663: # TOKEN_INVALID
|
|||
|
|
error_detail = "访问令牌无效或已过期"
|
|||
|
|
logger.error(f"令牌错误: {error_detail}")
|
|||
|
|
raise Exception(f"访问令牌错误: {error_detail}")
|
|||
|
|
else:
|
|||
|
|
logger.error(f"飞书API错误: 代码={error_code}, 消息={error_msg}")
|
|||
|
|
except ValueError:
|
|||
|
|
# 响应不是有效的JSON
|
|||
|
|
pass
|
|||
|
|
|
|||
|
|
raise Exception(error_msg)
|
|||
|
|
|
|||
|
|
return response.json()
|
|||
|
|
|
|||
|
|
@staticmethod
|
|||
|
|
def extract_params_from_url(table_url):
|
|||
|
|
"""
|
|||
|
|
从URL中提取app_token和table_id
|
|||
|
|
支持多种飞书多维表格URL格式：
|
|||
|
|
1. https://xxx.feishu.cn/base/{app_token}?table={table_id}
|
|||
|
|
2. https://xxx.feishu.cn/wiki/{app_token}?sheet={table_id}
|
|||
|
|
3. https://xxx.feishu.cn/wiki/wikcnXXX?sheet=XXX
|
|||
|
|
4. https://xxx.feishu.cn/base/bascnXXX?table=tblXXX
|
|||
|
|
|
|||
|
|
Args:
|
|||
|
|
table_url: 飞书多维表格URL
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
tuple: (app_token, table_id) 元组
|
|||
|
|
|
|||
|
|
Raises:
|
|||
|
|
ValueError: 如果无法从URL中提取必要参数
|
|||
|
|
"""
|
|||
|
|
# 记录原始URL
|
|||
|
|
logger.info(f"解析多维表格URL: {table_url}")
|
|||
|
|
|
|||
|
|
# 处理URL中的多余空格
|
|||
|
|
table_url = table_url.strip()
|
|||
|
|
|
|||
|
|
# 解析URL
|
|||
|
|
parsed_url = urlparse(table_url)
|
|||
|
|
query_params = parse_qs(parsed_url.query)
|
|||
|
|
path = parsed_url.path
|
|||
|
|
|
|||
|
|
logger.debug(f"URL路径: {path}, 查询参数: {query_params}")
|
|||
|
|
|
|||
|
|
# 尝试从查询参数中获取table_id
|
|||
|
|
table_id = None
|
|||
|
|
if 'table' in query_params:
|
|||
|
|
table_id = query_params['table'][0].strip()
|
|||
|
|
elif 'sheet' in query_params:
|
|||
|
|
table_id = query_params['sheet'][0].strip()
|
|||
|
|
|
|||
|
|
# 尝试从路径中获取app_token
|
|||
|
|
app_token = None
|
|||
|
|
# 处理标准格式 /base/{app_token} 或 /wiki/{app_token}
|
|||
|
|
standard_match = re.search(r'/(base|wiki)/([^/?]+)', path)
|
|||
|
|
if standard_match:
|
|||
|
|
app_token = standard_match.group(2).strip()
|
|||
|
|
else:
|
|||
|
|
# 处理简短ID格式，如 /wiki/wikcnXXX
|
|||
|
|
short_id_match = re.search(r'/(base|wiki)/([a-zA-Z0-9]+)', path)
|
|||
|
|
if short_id_match:
|
|||
|
|
app_token = short_id_match.group(2).strip()
|
|||
|
|
|
|||
|
|
# 检查是否成功提取
|
|||
|
|
if not app_token or not table_id:
|
|||
|
|
error_msg = "无法从URL中提取必要参数，请确认URL格式正确"
|
|||
|
|
logger.error(f"{error_msg}. URL: {table_url}")
|
|||
|
|
raise ValueError(error_msg)
|
|||
|
|
|
|||
|
|
logger.info(f"成功从URL提取参数: app_token={app_token}, table_id={table_id}")
|
|||
|
|
return app_token, table_id
|
|||
|
|
|
|||
|
|
@staticmethod
|
|||
|
|
def validate_access(app_token, access_token):
|
|||
|
|
"""
|
|||
|
|
验证应用是否有权限访问多维表格
|
|||
|
|
|
|||
|
|
Args:
|
|||
|
|
app_token: 应用令牌
|
|||
|
|
access_token: 访问令牌
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
bool: 是否有权限
|
|||
|
|
"""
|
|||
|
|
try:
|
|||
|
|
# 构造请求
|
|||
|
|
url = f"bitable/v1/apps/{app_token}"
|
|||
|
|
headers = {
|
|||
|
|
"Authorization": f"Bearer {access_token}",
|
|||
|
|
"Content-Type": "application/json"
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
# 发送请求
|
|||
|
|
response = BitableService.make_request("GET", url, headers=headers)
|
|||
|
|
|
|||
|
|
# 检查响应
|
|||
|
|
return response and "code" in response and response["code"] == 0
|
|||
|
|
|
|||
|
|
except Exception as e:
|
|||
|
|
logger.error(f"验证访问权限失败: {str(e)}")
|
|||
|
|
return False
|
|||
|
|
|
|||
|
|
@staticmethod
|
|||
|
|
def get_metadata(app_token, table_id, access_token):
|
|||
|
|
"""
|
|||
|
|
获取多维表格元数据
|
|||
|
|
|
|||
|
|
Args:
|
|||
|
|
app_token: 应用令牌
|
|||
|
|
table_id: 表格ID
|
|||
|
|
access_token: 访问令牌
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
dict: 表格元数据
|
|||
|
|
"""
|
|||
|
|
try:
|
|||
|
|
# 先验证应用是否有权限访问
|
|||
|
|
if not BitableService.validate_access(app_token, access_token):
|
|||
|
|
logger.warning(f"应用无权限访问多维表格: app_token={app_token}")
|
|||
|
|
|
|||
|
|
# 构造请求
|
|||
|
|
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:
|
|||
|
|
metadata = response.get("data", {}).get("table", {})
|
|||
|
|
logger.info(f"成功获取多维表格元数据: {metadata.get('name', table_id)}")
|
|||
|
|
return metadata
|
|||
|
|
|
|||
|
|
# 发生错误
|
|||
|
|
error_msg = f"获取多维表格元数据失败: {json.dumps(response)}"
|
|||
|
|
logger.error(error_msg)
|
|||
|
|
raise Exception(error_msg)
|
|||
|
|
|
|||
|
|
except Exception as e:
|
|||
|
|
# 如果正常API调用失败，使用替代方法
|
|||
|
|
logger.error(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:
|
|||
|
|
logger.info(f"查询多维表格记录: app_token={app_token}, table_id={table_id}")
|
|||
|
|
|
|||
|
|
# 构造请求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:
|
|||
|
|
result = response.get("data", {})
|
|||
|
|
logger.info(f"查询成功，获取到 {len(result.get('items', []))} 条记录")
|
|||
|
|
return result
|
|||
|
|
|
|||
|
|
# 发生错误
|
|||
|
|
error_msg = f"查询飞书多维表格失败: {json.dumps(response)}"
|
|||
|
|
logger.error(error_msg)
|
|||
|
|
raise Exception(error_msg)
|
|||
|
|
|
|||
|
|
except Exception as e:
|
|||
|
|
# 记录详细错误
|
|||
|
|
logger.error(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:
|
|||
|
|
logger.info(f"获取多维表格字段: app_token={app_token}, table_id={table_id}")
|
|||
|
|
|
|||
|
|
# 构造请求
|
|||
|
|
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:
|
|||
|
|
fields = response.get("data", {}).get("items", [])
|
|||
|
|
logger.info(f"成功获取到 {len(fields)} 个字段")
|
|||
|
|
return fields
|
|||
|
|
|
|||
|
|
# 发生错误
|
|||
|
|
error_msg = f"获取多维表格字段失败: {json.dumps(response)}"
|
|||
|
|
logger.error(error_msg)
|
|||
|
|
raise Exception(error_msg)
|
|||
|
|
|
|||
|
|
except Exception as e:
|
|||
|
|
# 记录详细错误
|
|||
|
|
logger.error(f"获取字段信息失败: {str(e)}")
|
|||
|
|
# 如果获取失败，返回一个基本字段集
|
|||
|
|
logger.warning("返回默认字段集")
|
|||
|
|
return [
|
|||
|
|
{"field_name": "title", "type": "text", "property": {}},
|
|||
|
|
{"field_name": "description", "type": "text", "property": {}},
|
|||
|
|
{"field_name": "created_time", "type": "datetime", "property": {}}
|
|||
|
|
]
|