knowledge/utils/fei_shu.py

import os
import uuid
from typing import Any, Dict, Optional

import lark_oapi as lark
from lark_oapi.core.enum import LogLevel


class FeiShu(object):
    _instance = None

    def __new__(cls, *args, **kwargs):
        if cls._instance is None:
            cls._instance = super().__new__(cls, *args, **kwargs)
        return cls._instance

    def __init__(self):
        app_id = os.getenv('FEISHU_APP_ID')
        app_secret = os.getenv('FEISHU_APP_SECRET')
        self.file_token = os.getenv('FEISHU_FILE_TOKEN')
        self.client = (lark.client.Client().builder()
                       .app_id(app_id)
                       .app_secret(app_secret)
                       .log_level(LogLevel.INFO)
                       .build())

    def create_table(self, name: str) -> str:
        request = (lark.bitable.v1.CreateAppTableRequest.builder()
                   .app_token(self.file_token)
                   .request_body(lark.bitable.v1.CreateAppTableRequestBody.builder()
                                 .table(lark.bitable.v1.ReqTable.builder()
                                        .name(name)
                                        .fields([
                                            lark.bitable.v1.AppTableCreateHeader.builder().field_name('原文链接').type(15).build(),
                                            lark.bitable.v1.AppTableCreateHeader.builder().field_name('抓取结果').type(1).build(),
                                            lark.bitable.v1.AppTableCreateHeader.builder().field_name('识别结果').type(1).build(),
                                            lark.bitable.v1.AppTableCreateHeader.builder().field_name('初步理解').type(1).build(),
                                        ])
                                        .build())
                                 .build())
                   .build())
        response = self.client.bitable.v1.app_table.create(request)
        if not response.success():
            raise RuntimeError(f'多维表格创建新数据表失败: {lark.JSON.marshal(response.error)}')
        return response.data.table_id

    def get_all_records(self, table_id: str, page_token: Optional[str] = None):
        request = (lark.bitable.v1.SearchAppTableRecordRequest.builder()
                   .app_token(self.file_token)
                   .table_id(table_id)
                   .user_id_type('open_id')
                   .page_size(500)
                   .request_body(lark.bitable.v1.SearchAppTableRecordRequestBody.builder()
                                 .automatic_fields(True)
                                 .build()))
        if page_token:
            request = request.page_token(page_token)
        request = request.build()
        response = self.client.bitable.v1.app_table_record.search(request)
        if not response.success():
            raise RuntimeError(f'获取多维表格记录失败: {lark.JSON.marshal(response.error)}')
        return response.data

    def get_record(self, table_id: str, *record_ids: str):
        request = (lark.bitable.v1.BatchGetAppTableRecordRequest.builder()
                   .app_token(self.file_token)
                   .table_id(table_id)
                   .request_body(lark.bitable.v1.BatchGetAppTableRecordRequestBody.builder()
                                 .record_ids(list(record_ids))
                                 .user_id_type('open_id')
                                 .with_shared_url(True)
                                 .automatic_fields(True)
                                 .build())
                   .build())
        response = self.client.bitable.v1.app_table_record.batch_get(request)
        if not response.success():
            raise RuntimeError(f'获取多维表格指定记录失败: {lark.JSON.marshal(response.error)}')
        return response.data

    def create_record(self, table_id: str, *fields: Dict[str, Any]):
        request = (lark.bitable.v1.BatchCreateAppTableRecordRequest.builder()
                   .app_token(self.file_token)
                   .table_id(table_id)
                   .user_id_type('open_id')
                   .client_token(str(uuid.uuid4()))
                   .ignore_consistency_check(False)
                   .request_body(lark.bitable.v1.BatchCreateAppTableRecordRequestBody.builder()
                                 .records([lark.bitable.v1.AppTableRecord.builder()
                                          .fields(item)
                                          .build() for item in fields])
                                 .build())
                   .build())
        response = self.client.bitable.v1.app_table_record.batch_create(request)
        if not response.success():
            raise RuntimeError(f'向多维表格添加记录失败: {lark.JSON.marshal(response.error)}')
        return response.data

    def update_record(self, table_id: str, *records: lark.bitable.v1.AppTableRecord):
        request = (lark.bitable.v1.BatchUpdateAppTableRecordRequest.builder()
                   .app_token(self.file_token)
                   .table_id(table_id)
                   .user_id_type('open_id')
                   .ignore_consistency_check(False)
                   .request_body(lark.bitable.v1.BatchUpdateAppTableRecordRequestBody.builder()
                                 .records(list(records))
                                 .build())
                   .build())
        response = self.client.bitable.v1.app_table_record.batch_update(request)
        if not response.success():
            raise RuntimeError(f'更新多维表格指定记录失败: {lark.JSON.marshal(response.error)}')
        return response.data

    def delete_record(self, table_id: str, *record_ids: str):
        request = (lark.bitable.v1.BatchDeleteAppTableRecordRequest.builder()
                   .app_token(self.file_token)
                   .table_id(table_id)
                   .request_body(lark.bitable.v1.BatchDeleteAppTableRecordRequestBody.builder()
                                 .records(list(record_ids))
                                 .build())
                   .build())
        response = self.client.bitable.v1.app_table_record.batch_delete(request)
        if not response.success():
            raise RuntimeError(f'删除多维表格指定记录失败: {lark.JSON.marshal(response.error)}')
        return response.data


if __name__ == '__main__':
    feishu = FeiShu()
    # 创建数据表
    new_table_id = feishu.create_table('测试数据表')
    # 新增记录
    new_fields = [
        {
            '原文链接': {
                'link': 'https://www.baidu.com',
                'text': 'https://www.baidu.com',
            },
            '抓取结果': '这是抓取结果1',
        },
        {
            '原文链接': {
                'link': 'https://www.qq.com',
                'text': 'https://www.qq.com',
            },
            '抓取结果': '这是抓取结果2',
        }
    ]
    feishu.create_record(new_table_id, *new_fields)
    # 获取全部记录
    get_result = feishu.get_all_records(new_table_id)
    has_more = get_result.has_more  # 是否有下一页
    next_page_token = get_result.page_token  # 下一页token
    new_record_ids = []
    for record in get_result.items:
        new_record_ids.append(record.record_id)
        print(record.fields)
    # 更新记录
    new_record = (lark.bitable.v1.AppTableRecord.builder()
                  .record_id(new_record_ids[0])
                  .fields({'识别结果': '这是识别结果'})
                  .build())
    feishu.update_record(new_table_id, new_record)
    # 获取指定ID记录
    get_result = feishu.get_record(new_table_id, *new_record_ids)
    for record in get_result.records:
        new_record_ids.append(record.record_id)
        print(record.fields)
    # 删除指定ID记录
    feishu.delete_record(new_table_id, new_record_ids[1])