add sync_xbot_activity_info

app/utils/utils.py

@@ -11,6 +11,7 @@ from lark_oapi.api.im.v1 import *
 from lark_oapi.api.bitable.v1 import *
 import lark_oapi as lark
 from lark_oapi.api.drive.v1 import *
+from lark_oapi.api.bitable.v1 import *
 
 
 def update_record(client, bitable_token, table_id, record_id, data):
@@ -197,3 +198,86 @@ def send_message(client, receive_id, content):
 
     # 处理业务结果
     lark.logger.info(lark.JSON.marshal(response.data, indent=4))
+
+
+def get_filtered_record_by_uuid(client, bitable_token, table_id, view_id, field_name, uuid):
+    request: SearchAppTableRecordRequest = (
+        SearchAppTableRecordRequest.builder()
+        .app_token(bitable_token)
+        .table_id(table_id)
+        .user_id_type("user_id")
+        .page_size(20)
+        .request_body(
+            SearchAppTableRecordRequestBody.builder()
+            .view_id(view_id)
+            .field_names([field_name])
+            .sort([Sort.builder().field_name(field_name).desc(True).build()])
+            .filter(
+                FilterInfo.builder()
+                .conjunction("and")
+                .conditions(
+                    [
+                        Condition.builder()
+                        .field_name(field_name)
+                        .operator("is")
+                        .value([uuid])
+                        .build()
+                    ]
+                )
+                .build()
+            )
+            .automatic_fields(False)
+            .build()
+        )
+        .build()
+    )
+
+    # 发起请求
+    option = (
+        lark.RequestOption.builder()
+        .user_access_token("u-caH0NMIK50kUMfBMUqgMtr1l6blBghh1Mi0015.82GbP")
+        .build()
+    )
+    response: SearchAppTableRecordResponse = client.bitable.v1.app_table_record.search(
+        request, option
+    )
+
+    # 处理失败返回
+    if not response.success():
+        # lark.logger.error(
+        #     f"client.bitable.v1.app_table_record.search failed, code: {response.code}, msg: {response.msg}, log_id: {response.get_log_id()}"
+        # )
+        return
+
+    # 处理业务结果
+    lark.logger.info(lark.JSON.marshal(response.data))
+
+    if response.data.total == 1:
+        return response.data.items[0].record_id
+
+    if response.data.total == 2:
+        print("找到了多条记录")
+
+if __name__ == "__main__":
+
+    app_id = os.environ["APP_ID"]
+    app_secret = os.environ["APP_SECRET"]
+
+    bitable_token = os.environ["BITABLE_TOKEN"]
+    table_id = os.environ["TABLE_ID"]
+
+    bitable_token = "XDmHbhbZcaY4gZsOvBJcLcHonKh"
+    table_id = "tbl2GcBvkvMtOBnl"
+
+    uuid = "4e478138-c8d6-426c-9761-3d1f4fd3cb05"
+    view_id = "vewIAewVD2"
+
+    # 创建client
+    client = (
+        lark.Client.builder()
+        .app_id(app_id)
+        .app_secret(app_secret)
+        .log_level(lark.LogLevel.DEBUG)
+        .build()
+    )
+    get_filtered_record_by_uuid(client, bitable_token, table_id, view_id, uuid)