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": {}} ]