office_xbot_feishu/app/utils/utils.py

import os
import json
from itertools import chain

import lark_oapi as lark

from flask import Flask
from dotenv import load_dotenv
from lark_oapi.adapter.flask import *
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):
    # 构造请求对象
    request: UpdateAppTableRecordRequest = (
        UpdateAppTableRecordRequest.builder()
        .app_token(bitable_token)
        .table_id(table_id)
        .record_id(record_id)
        .request_body(AppTableRecord.builder().fields(data).build())
        .build()
    )

    # 发起请求
    option = (
        lark.RequestOption.builder()
        .user_access_token("u-fEsiJAcUZaOFrBI0iYa6N61l4zvBghNxr200l1K82GvD")
        .build()
    )
    response: UpdateAppTableRecordResponse = client.bitable.v1.app_table_record.update(
        request, option
    )

    # 处理失败返回
    if not response.success():
        lark.logger.error(
            f"client.bitable.v1.app_table_record.update failed, code: {response.code}, msg: {response.msg}, log_id: {response.get_log_id()}"
        )
        return

    # 处理业务结果
    lark.logger.info(lark.JSON.marshal(response.data, indent=4))


def add_record(client, bitable_token, table_id, data):
    # 构造请求对象
    request: CreateAppTableRecordRequest = (
        CreateAppTableRecordRequest.builder()
        .app_token(bitable_token)
        .table_id(table_id)
        .request_body(AppTableRecord.builder().fields(data).build())
        .build()
    )

    # 发起请求
    response: CreateAppTableRecordResponse = client.bitable.v1.app_table_record.create(
        request
    )

    # 处理失败返回
    if not response.success():
        lark.logger.error(
            f"client.bitable.v1.app_table_record.create failed, code: {response.code}, msg: {response.msg}, log_id: {response.get_log_id()}"
        )
        return

    # 处理业务结果
    lark.logger.debug(lark.JSON.marshal(response.data, indent=4))

    return response.data.record.record_id


def get_message_resource(client, message_id, file_key, file_type, save_dir):
    request: GetMessageResourceRequest = (
        GetMessageResourceRequest.builder()
        .message_id(message_id)
        .file_key(file_key)
        .type(file_type)
        .build()
    )
    response: GetMessageResourceResponse = client.im.v1.message_resource.get(request)

    if not response.success():
        lark.logger.error(
            f"client.im.v1.message_resource.get failed, code: {response.code}, msg: {response.msg}, log_id: {response.get_log_id()}"
        )
        return

    # 处理业务结果

    file_path = os.path.join(save_dir, file_key + ".png")
    with open(file_path, "wb") as f:
        f.write(response.file.read())

    return file_path


def upload_media(client, file_path, parent_type, parent_node):
    # 构造请求对象
    _, filename = os.path.split(file_path)
    file_size = os.path.getsize(file_path)

    with open(file_path, "rb") as f:
        request: UploadAllMediaRequest = (
            UploadAllMediaRequest.builder()
            .request_body(
                UploadAllMediaRequestBody.builder()
                .file_name(filename)
                .parent_type(parent_type)
                .parent_node(parent_node)
                .size(file_size)
                .file(f)
                .build()
            )
            .build()
        )

        # 发起请求
        option = (
            lark.RequestOption.builder()
            .user_access_token("u-dF0r5rD5BcUFhORI38zmlk1l61TBghPzry00gl.82Haz")
            .build()
        )
        response: UploadAllMediaResponse = client.drive.v1.media.upload_all(
            request, option
        )

        # # 处理失败返回
        # if not response.success():
        #     lark.logger.error(
        #         f"client.drive.v1.media.upload_all failed, code: {response.code}, msg: {response.msg}, log_id: {response.get_log_id()}"
        #     )
        #     return

        # 处理业务结果
        lark.logger.info(lark.JSON.marshal(response.data, indent=4))

        return response.data.file_token


def send_group_message(client, message_id, content):
    # 构造请求对象

    request: ReplyMessageRequest = (
        ReplyMessageRequest.builder()
        .message_id(message_id)
        .request_body(
            ReplyMessageRequestBody.builder()
            .content(content=content)
            .msg_type("text")
            .reply_in_thread(True)
            .build()
        )
        .build()
    )

    # 发起请求
    response: ReplyMessageResponse = client.im.v1.message.reply(request)

    # 处理失败返回
    if not response.success():
        lark.logger.error(
            f"client.im.v1.message.reply failed, code: {response.code}, msg: {response.msg}, log_id: {response.get_log_id()}"
        )

    # 处理业务结果
    lark.logger.info(lark.JSON.marshal(response.data, indent=4))


def send_message(client, receive_id, content):
    # 构造请求对象
    request: CreateMessageRequest = (
        CreateMessageRequest.builder()
        .receive_id_type("open_id")
        .request_body(
            CreateMessageRequestBody.builder()
            .receive_id(receive_id)
            .msg_type("text")
            .content(content)
            .build()
        )
        .build()
    )

    # 发起请求
    response: CreateMessageResponse = client.im.v1.message.create(request)

    # 处理失败返回
    if not response.success():
        lark.logger.error(
            f"client.im.v1.message.create failed, code: {response.code}, msg: {response.msg}, log_id: {response.get_log_id()}"
        )
        return

    # 处理业务结果
    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()
    )