feat(Tools): Refactor the base table plugin (#9182)

Co-authored-by: 黎斌 <libin.23@bytedance.com>

api/core/tools/utils/feishu_api_utils.py

@@ -1,3 +1,4 @@
+import json
 from typing import Optional
 
 import httpx
@@ -17,6 +18,41 @@ def auth(credentials):
         raise ToolProviderCredentialValidationError(str(e))
 
 
+def convert_add_records(json_str):
+    try:
+        data = json.loads(json_str)
+        if not isinstance(data, list):
+            raise ValueError("Parsed data must be a list")
+        converted_data = [{"fields": json.dumps(item, ensure_ascii=False)} for item in data]
+        return converted_data
+    except json.JSONDecodeError:
+        raise ValueError("The input string is not valid JSON")
+    except Exception as e:
+        raise ValueError(f"An error occurred while processing the data: {e}")
+
+
+def convert_update_records(json_str):
+    try:
+        data = json.loads(json_str)
+        if not isinstance(data, list):
+            raise ValueError("Parsed data must be a list")
+
+        converted_data = [
+            {"fields": json.dumps(record["fields"], ensure_ascii=False), "record_id": record["record_id"]}
+            for record in data
+            if "fields" in record and "record_id" in record
+        ]
+
+        if len(converted_data) != len(data):
+            raise ValueError("Each record must contain 'fields' and 'record_id'")
+
+        return converted_data
+    except json.JSONDecodeError:
+        raise ValueError("The input string is not valid JSON")
+    except Exception as e:
+        raise ValueError(f"An error occurred while processing the data: {e}")
+
+
 class FeishuRequest:
     API_BASE_URL = "https://lark-plugin-api.solutionsuite.cn/lark-plugin"
 
@@ -517,3 +553,270 @@ class FeishuRequest:
         }
         res = self._send_request(url, method="GET", params=params)
         return res.get("data")
+
+    def create_base(
+        self,
+        name: str,
+        folder_token: str,
+    ) -> dict:
+        # 创建多维表格
+        url = f"{self.API_BASE_URL}/base/create_base"
+        payload = {
+            "name": name,
+            "folder_token": folder_token,
+        }
+        res = self._send_request(url, payload=payload)
+        return res.get("data")
+
+    def add_records(
+        self,
+        app_token: str,
+        table_id: str,
+        table_name: str,
+        records: str,
+        user_id_type: str = "open_id",
+    ) -> dict:
+        # 新增多条记录
+        url = f"{self.API_BASE_URL}/base/add_records"
+        params = {
+            "app_token": app_token,
+            "table_id": table_id,
+            "table_name": table_name,
+            "user_id_type": user_id_type,
+        }
+        payload = {
+            "records": convert_add_records(records),
+        }
+        res = self._send_request(url, params=params, payload=payload)
+        return res.get("data")
+
+    def update_records(
+        self,
+        app_token: str,
+        table_id: str,
+        table_name: str,
+        records: str,
+        user_id_type: str,
+    ) -> dict:
+        # 更新多条记录
+        url = f"{self.API_BASE_URL}/base/update_records"
+        params = {
+            "app_token": app_token,
+            "table_id": table_id,
+            "table_name": table_name,
+            "user_id_type": user_id_type,
+        }
+        payload = {
+            "records": convert_update_records(records),
+        }
+        res = self._send_request(url, params=params, payload=payload)
+        return res.get("data")
+
+    def delete_records(
+        self,
+        app_token: str,
+        table_id: str,
+        table_name: str,
+        record_ids: str,
+    ) -> dict:
+        # 删除多条记录
+        url = f"{self.API_BASE_URL}/base/delete_records"
+        params = {
+            "app_token": app_token,
+            "table_id": table_id,
+            "table_name": table_name,
+        }
+        if not record_ids:
+            record_id_list = []
+        else:
+            try:
+                record_id_list = json.loads(record_ids)
+            except json.JSONDecodeError:
+                raise ValueError("The input string is not valid JSON")
+        payload = {
+            "records": record_id_list,
+        }
+        res = self._send_request(url, params=params, payload=payload)
+        return res.get("data")
+
+    def search_record(
+        self,
+        app_token: str,
+        table_id: str,
+        table_name: str,
+        view_id: str,
+        field_names: str,
+        sort: str,
+        filters: str,
+        page_token: str,
+        automatic_fields: bool = False,
+        user_id_type: str = "open_id",
+        page_size: int = 20,
+    ) -> dict:
+        # 查询记录，单次最多查询 500 行记录。
+        url = f"{self.API_BASE_URL}/base/search_record"
+        params = {
+            "app_token": app_token,
+            "table_id": table_id,
+            "table_name": table_name,
+            "user_id_type": user_id_type,
+            "page_token": page_token,
+            "page_size": page_size,
+        }
+
+        if not field_names:
+            field_name_list = []
+        else:
+            try:
+                field_name_list = json.loads(field_names)
+            except json.JSONDecodeError:
+                raise ValueError("The input string is not valid JSON")
+
+        if not sort:
+            sort_list = []
+        else:
+            try:
+                sort_list = json.loads(sort)
+            except json.JSONDecodeError:
+                raise ValueError("The input string is not valid JSON")
+
+        if not filters:
+            filter_dict = {}
+        else:
+            try:
+                filter_dict = json.loads(filters)
+            except json.JSONDecodeError:
+                raise ValueError("The input string is not valid JSON")
+
+        payload = {}
+
+        if view_id:
+            payload["view_id"] = view_id
+        if field_names:
+            payload["field_names"] = field_name_list
+        if sort:
+            payload["sort"] = sort_list
+        if filters:
+            payload["filter"] = filter_dict
+        if automatic_fields:
+            payload["automatic_fields"] = automatic_fields
+        res = self._send_request(url, params=params, payload=payload)
+        return res.get("data")
+
+    def get_base_info(
+        self,
+        app_token: str,
+    ) -> dict:
+        # 获取多维表格元数据
+        url = f"{self.API_BASE_URL}/base/get_base_info"
+        params = {
+            "app_token": app_token,
+        }
+        res = self._send_request(url, method="GET", params=params)
+        return res.get("data")
+
+    def create_table(
+        self,
+        app_token: str,
+        table_name: str,
+        default_view_name: str,
+        fields: str,
+    ) -> dict:
+        # 新增一个数据表
+        url = f"{self.API_BASE_URL}/base/create_table"
+        params = {
+            "app_token": app_token,
+        }
+        if not fields:
+            fields_list = []
+        else:
+            try:
+                fields_list = json.loads(fields)
+            except json.JSONDecodeError:
+                raise ValueError("The input string is not valid JSON")
+        payload = {
+            "name": table_name,
+            "fields": fields_list,
+        }
+        if default_view_name:
+            payload["default_view_name"] = default_view_name
+        res = self._send_request(url, params=params, payload=payload)
+        return res.get("data")
+
+    def delete_tables(
+        self,
+        app_token: str,
+        table_ids: str,
+        table_names: str,
+    ) -> dict:
+        # 删除多个数据表
+        url = f"{self.API_BASE_URL}/base/delete_tables"
+        params = {
+            "app_token": app_token,
+        }
+        if not table_ids:
+            table_id_list = []
+        else:
+            try:
+                table_id_list = json.loads(table_ids)
+            except json.JSONDecodeError:
+                raise ValueError("The input string is not valid JSON")
+
+        if not table_names:
+            table_name_list = []
+        else:
+            try:
+                table_name_list = json.loads(table_names)
+            except json.JSONDecodeError:
+                raise ValueError("The input string is not valid JSON")
+
+        payload = {
+            "table_ids": table_id_list,
+            "table_names": table_name_list,
+        }
+        res = self._send_request(url, params=params, payload=payload)
+        return res.get("data")
+
+    def list_tables(
+        self,
+        app_token: str,
+        page_token: str,
+        page_size: int = 20,
+    ) -> dict:
+        # 列出多维表格下的全部数据表
+        url = f"{self.API_BASE_URL}/base/list_tables"
+        params = {
+            "app_token": app_token,
+            "page_token": page_token,
+            "page_size": page_size,
+        }
+        res = self._send_request(url, method="GET", params=params)
+        return res.get("data")
+
+    def read_records(
+        self,
+        app_token: str,
+        table_id: str,
+        table_name: str,
+        record_ids: str,
+        user_id_type: str = "open_id",
+    ) -> dict:
+        url = f"{self.API_BASE_URL}/base/read_records"
+        params = {
+            "app_token": app_token,
+            "table_id": table_id,
+            "table_name": table_name,
+        }
+        if not record_ids:
+            record_id_list = []
+        else:
+            try:
+                record_id_list = json.loads(record_ids)
+            except json.JSONDecodeError:
+                raise ValueError("The input string is not valid JSON")
+        payload = {
+            "record_ids": record_id_list,
+            "user_id_type": user_id_type,
+        }
+        res = self._send_request(url, method="GET", params=params, payload=payload)
+        return res.get("data")