谢田波
谢田波
Published on 2024-12-19 / 5 Visits
0
0

Python对接飞书多维表格

Python 对接飞书多维表格实现增删改查。

相关资料

飞书文档

https://open.feishu.cn/document/server-docs/docs/bitable-v1/bitable-overview

前期准备

创建应用
https://open.feishu.cn/app
image.png

查看应用信息,后续代码需要用到:
image.png

发布版本
image.png

创建表格

先准备好需要操作的多维表格
image.png

开通权限

文档权限配置
image.png

应用中配置表格权限
image.png

参数解析

image.png

运行结果

查询表格

image.png

增加记录

image.png

更新记录

image.png

完整代码

from datetime import datetime  
import requests  
  
class FeishuBitableClient:  
    BASE_URL = "https://open.feishu.cn/open-apis"  
  
    def __init__(self, app_id, app_secret, table_token,table_id):  
        self.app_id = app_id  
        self.app_secret = app_secret  
        self.table_token = table_token  
        self.table_id = table_id  
        self.access_token = None  
  
    def _get_access_token(self):  
        url = f"{self.BASE_URL}/auth/v3/tenant_access_token/internal/"  
        payload = {  
            'app_id': self.app_id,  
            'app_secret': self.app_secret  
        }  
        response = requests.post(url, json=payload)  
        result = response.json()  
        if result.get('code') == 0:  
            return result['tenant_access_token']  
        else:  
            raise Exception(f"Failed to get access token: {result}")  
  
    def _make_request(self, method, endpoint, params=None, data=None):  
        if not self.access_token:  
            self.access_token = self._get_access_token()  
  
        headers = {  
            'Authorization': f'Bearer {self.access_token}',  
            'Content-Type': 'application/json; charset=utf-8'  
        }  
        url = f"{self.BASE_URL}{endpoint}"  
        response = requests.request(method, url, headers=headers, params=params, json=data)  
        result = response.json()  
        if result.get('code') != 0:  
            raise Exception(f"Request failed: {result}")  
        return result  
  
    def get_tables(self):  
        endpoint = f"/bitable/v1/apps/{self.table_token}/tables"  
        return self._make_request('GET', endpoint)  
  
    def get_table_fields(self, table_id):  
        endpoint = f"/bitable/v1/apps/{self.table_token}/tables/{table_id}/fields"  
        return self._make_request('GET', endpoint)  
  
    def get_table_records(self, table_id, page_token=None):  
        endpoint = f"/bitable/v1/apps/{self.table_token}/tables/{table_id}/records"  
        params = {}  
        if page_token:  
            params['page_token'] = page_token  
        return self._make_request('GET', endpoint, params=params)  
  
    def create_record(self, table_id, record_data):  
        endpoint = f"/bitable/v1/apps/{self.table_token}/tables/{table_id}/records"  
        return self._make_request('POST', endpoint, data=record_data)  
  
    def update_record(self, table_id, record_id, record_data):  
        endpoint = f"/bitable/v1/apps/{self.table_token}/tables/{table_id}/records/{record_id}"  
        return self._make_request('PUT', endpoint, data=record_data)  
  
    def delete_record(self, table_id, record_id):  
        endpoint = f"/bitable/v1/apps/{self.table_token}/tables/{table_id}/records/{record_id}"  
        return self._make_request('DELETE', endpoint)  
  
def convert_date_to_timestamp( date_str):  
    """  
    将日期字符串转换为Unix时间戳  
    :param date_str: 日期字符串,格式为 "%Y-%m-%d %H:%M:%S"  
    :return: Unix时间戳  
    """  
    date_obj = datetime.strptime(date_str, "%Y-%m-%d %H:%M:%S")  
    timestamp = int(date_obj.timestamp() * 1000)  # 转换为毫秒  
    return timestamp  
  
def update_record_with_timestamp(record_data):  
    """  
    更新记录数据中的账单日字段为Unix时间戳  
    """  
    date_str = record_data["fields"]["账单日"]  
    timestamp = convert_date_to_timestamp(date_str)  
    record_data["fields"]["账单日"] = timestamp  
    return record_data  
  
  
# 示例用法  
if __name__ == "__main__":  
    app_id = 'xxxxx'  
    app_secret = 'yyyyyy'  
    table_token = 'zzzzz'  
    table_id = 'nnnn'  
  
    client = FeishuBitableClient(app_id, app_secret, table_token, table_id)  
  
    # 获取所有表格  
    tables = client.get_tables()  
    print("Tables:", tables)  
  
    # 获取特定表格的所有行记录  
    if 'data' in tables and 'items' in tables['data']:  
        for table in tables['data']['items']:  
            table_id = table['table_id']  
            all_records = []  
            page_token = None  
  
            while True:  
                response = client.get_table_records(table_id, page_token)  
                records = response.get('data', {}).get('items', [])  
                all_records.extend(records)  
  
                # 检查是否有更多页面  
                page_token = response.get('data', {}).get('page_token')  
                if not page_token:  
                    break  
  
            for record in all_records:  
                print(f"记录: ", record)  
    else:  
        print("No tables found or invalid response.")  
  
    # 创建记录  
    record_data = {  
        "fields": {  
            "账单日": "2024-10-10 00:00:00",  
            "收益/支出": "收益",  
            "种类": "固定工资",  
            "金额": 7777.0  
        }  
    }  
  
    record_data = update_record_with_timestamp(record_data)  
    print("创建记录:",client.create_record(table_id, record_data))  
  
    # 更新记录  
    record_id = 'recutHwoPbzjoI'  
    updated_record_data =  {  
        "fields": {  
            "账单日": "2024-11-11 00:00:00",  
            "收益/支出": "收益",  
            "种类": "固定工资",  
            "金额": 9999.0  
        }  
    }  
    updated_record_data = update_record_with_timestamp(updated_record_data)  
    print("更新记录:",client.update_record(table_id, record_id, updated_record_data))  
  
    # 删除记录  
    delete_record_id = 'recutHyhR2FnbL'  
    print("删除记录:",client.delete_record(table_id, delete_record_id))

结论

通过本文的介绍,你学会了如何使用 Python 快速对接飞书多维表格,实现表格增删改查。


Comment