aigc-test/feishu.py

import json
from utils import request_post, request_get
from config import set_config

config_ = set_config()


class FeiShuHelper(object):
    @staticmethod
    def get_tenant_access_token():
        """获取自建应用的tenant_access_token"""
        url = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal"
        headers = {"Content-Type": "application/json; charset=utf-8"}
        request_data = config_.FEISHU_TOKEN
        data = request_post(request_url=url, headers=headers, request_data=request_data)
        if data is not None:
            tenant_access_token = data.get('tenant_access_token')
            return tenant_access_token

    def get_data(self, spreadsheet_token, sheet_id):
        """读取电子表格数据"""
        tenant_access_token = self.get_tenant_access_token()
        url = f"https://open.feishu.cn/open-apis/sheets/v2/spreadsheets/{spreadsheet_token}/values_batch_get"
        headers = {
            "Content-Type": "application/json; charset=utf-8",
            "Authorization": f"Bearer {tenant_access_token}"
        }
        params = {
            'ranges': sheet_id,
        }
        data = request_get(request_url=url, headers=headers, params=params)
        values = []
        if data is not None:
            try:
                values = data['data']['valueRanges'][0].get('values')
            except:
                values = []
        return values

    def data_to_feishu_sheet(self, sheet_token, sheet_id, data, start_row, start_column, end_column):
        tenant_access_token = self.get_tenant_access_token()
        url = f"https://open.feishu.cn/open-apis/sheets/v2/spreadsheets/{sheet_token}/values_prepend"
        headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {tenant_access_token}"
        }
        for i in range(len(data) // 10 + 1):
            values = data[i * 10:(i + 1) * 10]
            start_index = start_row + i * 10
            end_index = start_index + len(values)-1
            print(len(values), start_index, end_index)
            post_data = {
                "valueRange": {
                    "range": f"{sheet_id}!{start_column}{start_index}:{end_column}{end_index}",
                    "values": values
                }
            }
            r2 = request_post(request_url=url, headers=headers, request_data=post_data)
            # print(r2["msg"])
            print(r2)

    def update_values(self, sheet_token, sheet_id, data, start_row, start_column, end_column):
        """写入数据"""
        tenant_access_token = self.get_tenant_access_token()
        url = f"https://open.feishu.cn/open-apis/sheets/v2/spreadsheets/{sheet_token}/values_append"
        headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {tenant_access_token}"
        }
        for i in range(len(data) // 10 + 1):
            values = data[i * 10:(i + 1) * 10]
            start_index = start_row + i * 10
            end_index = start_index + len(values) - 1
            print(len(values), start_index, end_index)
            post_data = {
                "valueRange": {
                    "range": f"{sheet_id}!{start_column}{start_index}:{end_column}{end_index}",
                    "values": values
                }
            }
            r2 = request_post(request_url=url, headers=headers, request_data=post_data)
            # print(r2["msg"])
            print(r2)


if __name__ == '__main__':
    # sheet_info = config_.SHEET_INFO['汉语常用词汇表']
    # FeiShuHelper().get_data(spreadsheet_token=sheet_info.get('spreadsheet_token'), sheet_id=sheet_info.get('sheet_id'))
    FeiShuHelper().data_to_feishu_sheet(sheet_token='DkiUsqwJ6hmBxstBYyEcNE4ante',
                                        sheet_id='08d4cc',
                                        data=[['1', 2, 3, 4]],
                                        start_row=1,
                                        start_column='A',
                                        end_column='D')