飞书API（6）：使用 pandas 处理数据并写入 MySQL 数据库

news2024/12/24 0:57:22

一、引入

上一篇了解了飞书 28 种数据类型通过接口读取到的数据结构，本文开始探讨如何将这些数据写入 MySQL 数据库。
这个工作流的起点是从 API 获取到的一个完整的数据，终点是写入 MySQL 数据表，表结构和维格表结构类似。在过程中可以有不同的工作流程，可以是将接口返回的所有数据作为一个值，直接写入 MySQL 表中，再使用 MySQL 对该值进行解析，处理成不同的列，然后再新建一张表单存储，这种方法入库比较简单粗暴，但是 MySQL 的处理会比较复杂，更侧重 MySQL 的对 json 结构的解析处理能力；也可以使用 Python 对接口数据进行进行处理，提取出各个数据列以及对应的值，再入库。
本文主要探讨后者。

虽然飞书的多维表提供了 28 中数据类型，但是本质上，很多数据类型记录的内容从 MySQL 的数据类型的角度上看是相似的，可能有点绕，举个例子：如下图，文本列和单选列虽然在飞书多维表是分为两个数据类型，但是它们的列值：单选1、单选2、这是文本111、这是文本222，本质上都是一个字符串，所以在入库处理时可以都设置为 MySQL 的 varchar 数据类型。
image.png

从 MySQL 的数据类型的角度上看，我们可以将飞书的这 28 中数据类型划分为五类，分别是字符串、数据、时间、列表和布尔值，参考如下：

类型描述MySQL 常用数据类型飞书数据类型编码飞书数据类型中文描述
字符串text、varchar、char1、3、11、13、15、22、23、1003、1004、1005多行文本、条码、Email邮箱、单选、人员、电话号码、超链接、附件、地理位置、群组、创建人、修改人、自动编号
数字double、float、bigint、int2数字、进度、货币、评分
布尔值bool7复选框
时间datetime、date、timestamp5、1001、1002日期、创建时间、最后更新时间
列表格式字符串json4、17、18、21、19、20多选、附件、单向关联、双向关联、查找引用、公式

下面开始探讨相关的数据处理。
本文结构：先对每个数据类型进行处理，然后在 MySQL 创建数据表，最后将数据写入数据表。

二、使用 pandas 处理每个数据类型的数据

2.1 环境说明

Python 3.9.12，相关第三方库如下：

requests == 2.31.0
pandas == 1.3.5
SQLAlchemy == 1.4.32
jupyter == 1.0.0

MySQL 8.0

2.2 准备工作

由于飞书应用的限制，飞书多维表无法设置公开给任一应用读取使用，所以需要自行创建一个包含 28 种数据类型的多维表，然后给应用授权（参考《飞书API（3）：Python 自动读取多维表所有分页数据的三种方法》的3、创建多维表，并设置应用操作多维表的权限）。

我用于测试的数据结构如下，可能你的和我的命名不同，所以以下代码的列名称根据你的列名进行修改即可。
我的测试数据为：
image.png

image.png

为了方便调试，本次使用 jupyter notebook 来做数据处理。
如果你未安装过 jupyter，可以考虑以下方案：

  • 如果安装 Python 是使用 Anaconda 3 包，一般会自动安装 jupyter；
  • 如果安装 Python 是使用官方的 Python 包，可以通过pip install jupyter安装；
  • 不想安装，也可以直接跑 .py 文件；
  • 当然也有替代方案，直接在命令行或终端输入python回车调用 Python 的测试环境，或者输入ipython回车调用 ipython 来测试。

我们取《飞书API（3）：Python 自动读取多维表所有分页数据的三种方法》的【2.1 while 循环读取分页数据】来读取所有的数据，代码如下。

import requests
import json

def get_tenant_access_token(app_id, app_secret):
    url = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal"
    payload = json.dumps({
        "app_id": app_id,
        "app_secret": app_secret
    })
    headers = {'Content-Type': 'application/json'}
    response = requests.request("POST", url, headers=headers, data=payload)
    # print(response.text)
    return response.json()['tenant_access_token']

def get_bitable_datas(tenant_access_token, app_token, table_id, page_token='', page_size=20):

    url = f"https://open.feishu.cn/open-apis/bitable/v1/apps/{app_token}/tables/{table_id}/records/search?page_size={page_size}&page_token={page_token}&user_id_type=user_id"
    payload = json.dumps({})
    headers = {
        'Content-Type': 'application/json',
        'Authorization': f'Bearer {tenant_access_token}'
    }
    response = requests.request("POST", url, headers=headers, data=payload)
    # print(response.text)
    return response.json()

def main():
    app_id = 'your_app_id'
    app_secret = 'your_app_secret'
    tenant_access_token = get_tenant_access_token(app_id, app_secret)
    app_token = 'your_app_token'
    table_id = 'your_table_id'

    page_token = ''
    page_size = 5
    has_more = True
    feishu_datas = []
    while has_more:
        response = get_bitable_datas(tenant_access_token, app_token, table_id, page_token, page_size)
        if response['code'] == 0:
            page_token = response['data'].get('page_token')
            has_more = response['data'].get('has_more')
            # print(response['data'].get('items'))
            # print('\n--------------------------------------------------------------------\n')
            feishu_datas.extend(response['data'].get('items'))
        else:
            raise Exception(response['msg'])

    return feishu_datas

if __name__ == '__main__':
    feishu_datas = main()
    print(feishu_datas)

将代码放到 jupyter notebook 中运行，得到结果如下
image.png

接下来将获取到的数据feishu_datas通过 Pandas 的 DataFrame 来处理。

import pandas as pd
feishu_df = pd.DataFrame(feishu_datas)
feishu_df

打印结果如下：
image.png

上篇，我们已经了解到每个飞书数据类型返回的数据结构，接下来就结合它来取值。

2.3 提取字符串数据列

2.3.1 多行文本、条码、Email邮箱

多行文本、条码、Email邮箱这三者的数据结构类型类似，可以统一处理，即取“text”的值。

数据类型编码数据类型中文描述数据类型对应英文描述数据示例
1多行文本、条码Text,Barcode“多行文本”: [{“text”: “我是文本1”,“type”: “text”}]
1Email邮箱Email“Email”: [{“link”: “mailto:ceshi@ceshi.com”,“text”: “ceshi@ceshi.com”,“type”: “url”}]

但是，实际生产中的数据可能没有那么完美，会存在很多空值，直接取“text”即get("文本")[0].get("text")会报错：TypeError: ‘NoneType’ object is not subscriptable，大致意思就是空值类型不能索引。如果值不为空，正常返回列表，才可以通过[0]进行索引。
image.png

解决该问题，其实很简单，填充一个默认值即可，这个填充是在get()传递第二个参数，而不是对 pandas 列进行填充。该参数需要根据后面取值的结构进行适配，这里后面通过索引取一次，再根据键取一次，索引需要给一个嵌套字典的列表结构，即[{}]。

顺带说一下，为什么要使用x.get("文本")，而不使用x["文本"]？
第一，前者兼容空值，即使是空值也不会报错，而是返回 None，后者则直接报错找不到对应的键；
第二，前者可以传递第二个参数，当对象是空值时，返回该参数，从而支持后续再次取值。

当然，对 pandas 列进行空值填充也是一种方案。代码示例如下：

feishu_df.fields.apply(lambda x:x.get("文本", [{}])[0].get("text"))

结果参考如下：
image.png

同理，可以对 Email邮箱 类型的列做同样的处理，将列名进行修改即可，示例如下：
image.png

2.3.2 单选、电话号码、自动编号

单选、电话号码、自动编号这三者的数据结构类型类似，直接取列值即可，直接使用get()方法，可以不用考虑空值的问题。

数据类型编码数据类型中文描述数据类型对应英文描述数据示例
3单选SingleSelect“单选”: “单选11”
13电话号码Phone“电话号码”: “13549857286”
1005自动编号AutoNumber“自动编号”: “1”

参考如下

feishu_df.fields.apply(lambda x:x.get("单选"))
feishu_df.fields.apply(lambda x:x.get("电话号码"))
feishu_df.fields.apply(lambda x:x.get("自动编号"))

结果如下：
image.png

2.3.3 人员、群组、创建人、修改人

人员、群组、创建人、修改人这四者的数据结构类型类似，取值方法也打通小异。我们这里保留“name”的值，你可以根据实际的应用场景判断，是否改为保留“id”的值，或者二者都需要等。

数据类型编码数据类型中文描述数据类型对应英文描述数据示例
11人员User“人员”: [{“email”: “”,“en_name”: “user1”,“id”: “ou_4007a8a82cc6e0874524edda12ce94b1”,“name”: “user1”}]
23群组GroupChat“群组”: [{“avatar_url”: “https://s1-imfile.feishucdn.com/static-resource/avatar/default-avatar_9fb72564-d52a-49b0-9de8-f79071a02286_96.webp”,“id”: “oc_8b6ac124bd908dce5c5facfb41c4dd4e”,“name”: “(无主题)”}]
1003创建人CreatedUser同人员
1004修改人ModifiedUser同人员

处理方式和文本类似，参考代码如下：

feishu_df.fields.apply(lambda x:x.get("人员1", [{}])[0].get("name"))
feishu_df.fields.apply(lambda x:x.get("群组1", [{}])[0].get("name"))
feishu_df.fields.apply(lambda x:x.get("创建人", [{}])[0].get("name"))
feishu_df.fields.apply(lambda x:x.get("修改人", [{}])[0].get("name"))

结果如下：
image.png

上面是单值的情况，如果是多值，还需要进行遍历取值。对于多值的情况，此处我的处理方法是把多个值通过逗号链接起来。比如选择了“张三”和“李四”，处理的结构为：张三,李四。
注意：这里的默认值需要做一层处理，因为 Nonetype 不能使用join()连接，空字符串才可以。
参考处理逻辑如下：

feishu_df.fields.apply(\
    lambda x: ','.join([val.get("name") for val in x.get("人员2", [{"name":""}])]))
feishu_df.fields.apply(\
    lambda x: ','.join([val.get("name") for val in x.get("群组2", [{"name":""}])]))

结果如下：
image.png

2.3.4 超链接

超链接和文本的取值差不多，只不过少了一层列表的取值，默认填充值也相应去掉列表层。

数据类型编码数据类型中文描述数据类型对应英文描述数据示例
15超链接Url“超链接”: {
“link”: “https://xxx.feishu.cn/base/PtRdbPjCFa5Og5sry0lcD1yPnKg?table=tbl3cvd797CmyEnN&view=vewdFnsmWn”,
“text”: “测试数据类型”
}

这里保留原始的链接，即“link”的值，参考代码如下：

feishu_df.fields.apply(lambda x:x.get("超链接",{}).get("link"))

结果如下：
image.png

2.3.5 地理位置

地理位置的数据结构和超链接类似，取值方式改下关键字即可。

数据类型编码数据类型中文描述数据类型对应英文描述数据示例
22地理位置Location“地理位置”: {
“address”: “东长安街”,
“adname”: “东城区”,
“cityname”: “北京市”,
“full_address”: “天安门广场，北京市北京市东城区东长安街”,
“location”: “116.397755,39.903179”,
“name”: “天安门广场”,
“pname”: “北京市”
}

这里我保留“full_address”的值，参考代码如下：

feishu_df.fields.apply(lambda x:x.get("地理位置",{}).get("full_address"))

结果如下：
image.png

2.4 提取数字与布尔值数据列

2.4.1 数字、进度、货币、评分

数字、进度、货币、评分这四者的数据结构和单选类似，取值是修改一下列名即可。

数据类型编码数据类型中文描述数据类型对应英文描述数据示例
2数字、进度、货币、评分Number,Progress,Currency,Rating“数字”: 1.33

参考代码：

feishu_df.fields.apply(lambda x:x.get("数字"))
feishu_df.fields.apply(lambda x:x.get("进度"))
feishu_df.fields.apply(lambda x:x.get("货币"))
feishu_df.fields.apply(lambda x:x.get("评分"))

结果如下：
image.png

2.4.2 复选框

复选框取值方式和数字的取值方式一样。

数据类型编码数据类型中文描述数据类型对应英文描述数据示例
7复选框Checkbox“复选框”: true

参考代码：

feishu_df.fields.apply(lambda x:x.get("复选框"))

结果如下：
image.png

在实际的生产过程中，见到比较多表示布尔值的字段，可能是使用数字 0 和 1 来表示，如果要改为数字，可加一个三元表达式进行判断，参考代码如下：

feishu_df.fields.apply(\
    lambda x:1 if x.get("复选框") else(None if x.get("复选框") is None else 0))

结果如下：
image.png

2.5 提取时间数据列

时间列的格式和数字一样，如果直接存时间戳，可以参考数字的取值逻辑，直接取即可，后续在读表的时候再进行格式转换。但是这种方式不够直观，可读性较差，这里我把它转为时间格式：年-月-日 时:分:秒。

数据类型编码数据类型中文描述数据类型对应英文描述数据示例
5日期DateTime“日期”: 1711900800000
1001创建时间CreatedTime同日期
1002最后更新时间ModifiedTime同日期

由于三者一模一样，这里只取数据类型编码 5 来处理。
处理时间，特别是时间戳的转换，需要特别注意时区的问题。Pandas 默认是 0 时区，所以需要加上 8 小时（28800 秒），由于pd.to_datetime()方法不能处理 NoneType 对象，所以需要给默认值，我这里给 1000（飞书日期列的单位是毫秒，其他值也可以），最终反应为时间格式是“1970-01-01 08:00:01”。

feishu_df.fields.apply(lambda x:28800 + int(x.get('日期',1000)/1000))

pd.to_datetime(
    feishu_df.fields.apply(lambda x:28800 + int(x.get('日期',1000)/1000)),
    unit='s')

结果如下：
image.png

2.6 提取列表格式字符串数据列

注意：这是 json 并不是列表！需要使用 json 库将列表转为 json 格式。

2.6.1 多选

多选列存为列表，直接取列值即可。

数据类型编码数据类型中文描述数据类型对应英文描述数据示例
4多选MultiSelect“多选”: [“多选11”,“多选22”]

参考代码：

feishu_df.fields.apply(lambda x:json.dumps(x.get("多选")))

结果如下：
image.png

2.6.2 查找引用、公式

查找引用、公式这两个类型由于是可变的，这里暂时保留原数据，处理为列表格式字符串，实际生产过程可以根据列的特性进行定制修改。

数据类型编码数据类型中文描述数据类型对应英文描述数据示例
19查找引用Lookup“查找引用”: {“type”: 1,“value”: [{“text”: “我是文本1”,“type”: “text”}]}
20公式Formula“公式-数字”: {“type”: 2,“value”: [10]}
“公式-文本”: {“type”: 1,“value”: [{“text”: “公式1”,“type”: “text”}]}

直接取列值，转为字符串即可，参考代码如下：

feishu_df.fields.apply(lambda x:json.dumps(x.get("查找引用数值")))
feishu_df.fields.apply(lambda x:json.dumps(x.get("数字公式")))

结果如下：
image.png

2.6.3 单向关联、双向关联

单向关联、双向关联的列表数据都在“link_record_ids”键中，处理逻辑一样。

数据类型编码数据类型中文描述数据类型对应英文描述数据示例
18单向关联SingleLink“单向关联”: {“link_record_ids”: [“recuax3DpzWCW4”]}
21双向关联DuplexLink“双向关联”: {“link_record_ids”: [“recuax3DpzWCW4”]}

单向关联和双向关联会返回默认值{}，所以不存在 None 值，get()方法不需要传递第二个参数。
参考代码：

feishu_df.fields.apply(lambda x:json.dumps(x.get("单向关联").get("link_record_ids")))
feishu_df.fields.apply(lambda x:json.dumps(x.get("双向关联").get("link_record_ids")))

结果如下：
image.png

2.6.4 附件

附件可能有一个也可能有多个，其实和人员的取值逻辑差不多，不过这里把它处理为列表格式，存放所有图片的“url”。
注意：由于鉴权的限制，浏览器无法直接展示，需要使用飞书应用调用接口下载完图片才可以查看，在生产应用端的数据分析领域可能都不会使用该字段。

数据类型编码数据类型中文描述数据类型对应英文描述数据示例
17附件Attachment“附件”: [{
“file_token”: “Cm3Vb8fe4oLPw4xgChZcOa2Mnhe”,
“name”: “image.png”,
“size”: 956,
“tmp_url”: “https://open.feishu.cn/open-apis/drive/v1/medias/batch_get_tmp_download_url?file_tokens=Cm3Vb8fe4oLPw4xgChZcOa2Mnhe”,
“type”: “image/png”,
“url”: “https://open.feishu.cn/open-apis/drive/v1/medias/Cm3Vb8fe4oLPw4xgChZcOa2Mnhe/download”}]

参考处理代码如下：

feishu_df.fields.apply(\
    lambda x: json.dumps([val.get("url") for val in x.get("附件", [{"url":""}])]))

结果如下：
image.png

2.7 其他数据列

流程和按钮列没有返回值，所以不需要处理。

数据类型编码数据类型中文描述数据类型对应英文描述数据示例
24流程Stage无返回值
3001按钮Button无返回值

2.8 数据类型预处理小结

目前上面处理好的数据，还没有保存起来，需要在feishu_df创建一个新列，将处理好的数据作为列值插入，后续直接把feishu_df数据入库即可。
创建新列，涉及到一个命名问题，需要给每个列起一个英文名，后续 MySQL 建表参考该英文名进行建表。

结合上面的处理逻辑，在feishu_df新建需要入库的字段，参考代码如下：

feishu_df['field_text'] = feishu_df.fields.apply(lambda x:x.get("文本", [{}])[0].get("text"))
feishu_df['field_email'] = feishu_df.fields.apply(lambda x:x.get("email", [{}])[0].get("text"))
feishu_df['field_select'] = feishu_df.fields.apply(lambda x:x.get("单选"))
feishu_df['field_mobile'] = feishu_df.fields.apply(lambda x:x.get("电话号码"))
feishu_df['field_no'] = feishu_df.fields.apply(lambda x:x.get("自动编号"))
feishu_df['field_member1'] = feishu_df.fields.apply(lambda x:x.get("人员1", [{}])[0].get("name"))
feishu_df['field_group1'] = feishu_df.fields.apply(lambda x:x.get("群组1", [{}])[0].get("name"))
feishu_df['field_creator'] = feishu_df.fields.apply(lambda x:x.get("创建人", [{}])[0].get("name"))
feishu_df['field_modifier'] = feishu_df.fields.apply(lambda x:x.get("修改人", [{}])[0].get("name"))
feishu_df['field_member2'] = feishu_df.fields.apply(\
    lambda x: ','.join([val.get("name") for val in x.get("人员2", [{"name":""}])]))
feishu_df['field_group2'] = feishu_df.fields.apply(\
    lambda x: ','.join([val.get("name") for val in x.get("群组2", [{"name":""}])]))
feishu_df['field_url'] = feishu_df.fields.apply(lambda x:x.get("超链接",{}).get("link"))
feishu_df['field_location'] = feishu_df.fields.apply(lambda x:x.get("地理位置",{}).get("full_address"))
feishu_df['field_number'] = feishu_df.fields.apply(lambda x:x.get("数字"))
feishu_df['field_progress'] = feishu_df.fields.apply(lambda x:x.get("进度"))
feishu_df['field_money'] = feishu_df.fields.apply(lambda x:x.get("货币"))
feishu_df['field_rating'] = feishu_df.fields.apply(lambda x:x.get("评分"))
feishu_df['field_bool'] = feishu_df.fields.apply(lambda x:x.get("复选框"))
feishu_df['field_date'] = pd.to_datetime(
    feishu_df.fields.apply(lambda x:28800 + int(x.get('日期',1000)/1000)),
    unit='s')
feishu_df['field_createdtime'] = pd.to_datetime(
    feishu_df.fields.apply(lambda x:28800 + int(x.get('创建时间',1000)/1000)),
    unit='s')
feishu_df['field_updatedtime'] = pd.to_datetime(
    feishu_df.fields.apply(lambda x:28800 + int(x.get('更新时间',1000)/1000)),
    unit='s')
feishu_df['field_mulselect'] = feishu_df.fields.apply(lambda x:json.dumps(x.get("多选")))
feishu_df['field_findnum'] = feishu_df.fields.apply(lambda x:json.dumps(x.get("查找引用数值")))
feishu_df['field_numformula'] = feishu_df.fields.apply(lambda x:json.dumps(x.get("数字公式")))
feishu_df['field_singleunion'] = feishu_df.fields.apply(lambda x:json.dumps(x.get("单向关联").get("link_record_ids")))
feishu_df['field_doubleunion'] = feishu_df.fields.apply(lambda x:json.dumps(x.get("双向关联").get("link_record_ids")))
feishu_df['field_file'] = feishu_df.fields.apply(\
    lambda x: json.dumps([val.get("url") for val in x.get("附件", [{"url":""}])]))

# 查看前3行
feishu_df.head(3)

执行结果如下：
image.png

原本feishu_df带有2个列：“fields”和“record_id”，数据列已经解析出来了，可以把“fields”删除，保留“record_id”即可。

feishu_df.drop(['fields'],axis=1,inplace=True)

三、使用 sqlalchemy 写入 MySQL 数据库

为什么使用 sqlalchemy？

  • 主要是因为 pandas 可以利用它直接将 DataFrame 数据写入 MySQL 数据库

3.1 MySQL 建表

建表可以在 MySQL 直接创建，也可以通过 Python 调用数据库创建，为了统一处理，此处使用后者。
Python 使用 sqlalchemy 库创建 MySQL 数据表的通用代码结构参考如下，修改 MySQL 的配置，并且传递 SQL 语句即可建表。

from sqlalchemy import create_engine, text

# 创建 SQLAlchemy engine 对象，这里以 MySQL 为例
# engine = create_engine('mysql://username:password@host:port/dbname')
engine = create_engine('mysql://root:password@127.0.0.1:3306/test')

# 定义一个建表的 SQL 语句
create_table_sql = ''''''

# 使用 execute() 方法执行 SQL 语句
with engine.connect() as connection:
    connection.execute(text(create_table_sql))
    print('创建成功！')

建表语句：将建表语句传递给create_table_sql变量，执行之后便可以在数据库中建表。

create table if not exists test.feishu_data_type_test(
    record_id               varchar(256)     comment '飞书记录id'
   ,field_text              varchar(256)     comment '文本'
   ,field_email             varchar(256)     comment 'email'
   ,field_select            varchar(256)     comment '单选'
   ,field_mobile            varchar(256)     comment '电话号码'
   ,field_no                varchar(256)     comment '自动编号'
   ,field_member1           varchar(256)     comment '人员1'
   ,field_group1            varchar(256)     comment '群组1'
   ,field_creator           varchar(256)     comment '创建人'
   ,field_modifier          varchar(256)     comment '修改人'
   ,field_member2           varchar(256)     comment '人员2'
   ,field_group2            varchar(256)     comment '群组2'
   ,field_url               varchar(256)     comment '超链接'
   ,field_location          varchar(256)     comment '地理位置'
   ,field_number            float            comment '数字'
   ,field_progress          float            comment '进度'
   ,field_money             float            comment '货币'
   ,field_rating            float            comment '评分'
   ,field_bool              bool             comment '复选框'
   ,field_date              datetime         comment '日期'
   ,field_createdtime       datetime         comment '创建时间'
   ,field_updatedtime       datetime         comment '更新时间'
   ,field_mulselect         json             comment '多选'
   ,field_findnum           json             comment '查找引用数值'
   ,field_numformula        json             comment '数字公式'
   ,field_singleunion       json             comment '单向关联'
   ,field_doubleunion       json             comment '双向关联'
   ,field_file              json             comment '附件'
)

测试结果如下：
image.png

执行之后，可以登录数据库，查看相关的表单信息。截图如下：
image.png

3.2 写入 MySQL 表

pandas.DataFrame.to_sql() 方法可以调用 sqlalchemy 库的 create_engine 模块实现和 MySQL 数据联通，直接将 DataFrame 数据写入 MySQL 中。参考代码如下：

from sqlalchemy import create_engine

# 创建 SQLAlchemy engine 对象，这里以 MySQL 为例
# engine = create_engine('mysql://username:password@host:port/dbname')
engine = create_engine('mysql://root:password@127.0.0.1:3306/test')

# 将 DataFrame 直接写入 MySQL 数据库
feishu_df.to_sql(name='feishu_data_type_test', con=engine, if_exists='append', index=False)
print('写入成功！')

说明：

  • df.to_sql 方法将 DataFrame 写入到数据库中，name 参数指定表名，con 参数指定数据库引擎，if_exists 参数指定了如果表已经存在应该如何处理（例如，‘fail’、‘replace’ 或 ‘append’），index 参数表示是否将 DataFrame 的索引写入数据库，默认为 True，这里设置为 False 表示不保存索引。
  • 默认情况下，to_sql 方法基于列名匹配，和顺序无关，即匹配 DataFrame 的列名与数据库表中的列名，所以只要保证 DataFrame 的列名与数据库表中的列名匹配即可。当然了，数据类型也必须兼容如果尝试将一个含有字符串的 DataFrame 列插入到数据库的整数字段中将会报错。
  • 如果 if_exists 参数传递“replace”，则会根据 DataFrame 的列重新建表。

执行代码测试结果：
image.png

连接数据，查看表数据，截图如下：
image.png

image.png

image.png

可以看到，所有的数据写入均符合预期！完美！！

四、小结

本文介绍了如何处理飞书的 28 中数据类型，以及通过 sqlalchemy 库将处理好的数据入库。
在探索一个未知的领域时，一般会从个别典型的案例先入手，随着认知的不断深入，逐渐有全局观，便会考虑通用案例，然后在通用案例下，再考虑局部的需求定制。

本文是一个比较全面的案例，更多的是在提供一个解决思路，相关代码的可拓展性较差，下一遍介绍另外一个飞书的 API 读取多维表的元数据来优化可拓展性问题，使得代码变得更加通用。下下篇则在通用的基础上再做定制化需求。

五、附：最终代码

  • 在循环取数的代码的基础上新增三个函数
    • extract_key_fields(feishu_datas)：提取飞书接口数据的关键列
    • cre_mysql_table(create_table_sql)：在 MySQL 中建表
    • insert_mysql_table(feishu_df, table_name)：将提取的关键列数据写入 MySQL 数据表
  • 注意点：
    • 需要修改配置信息，包含 MySQL 的配置信息、飞书的 APP 配置信息、飞书多维表的配置信息。
    • 需要修改飞书多维表的列名，对应的英文命名，还有 MySQL 的建表语句
    • 目前该代码的可拓展性较差，下一遍介绍另外一个飞书的 API 读取多维表的元数据来优化可拓展性问题，使得代码变得更加通用。
import requests
import json
import pandas as pd
from sqlalchemy import create_engine, text

def get_tenant_access_token(app_id, app_secret):
    url = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal"
    payload = json.dumps({
        "app_id": app_id,
        "app_secret": app_secret
    })
    headers = {'Content-Type': 'application/json'}
    response = requests.request("POST", url, headers=headers, data=payload)
    # print(response.text)
    return response.json()['tenant_access_token']

def get_bitable_datas(tenant_access_token, app_token, table_id, page_token='', page_size=20):

    url = f"https://open.feishu.cn/open-apis/bitable/v1/apps/{app_token}/tables/{table_id}/records/search?page_size={page_size}&page_token={page_token}&user_id_type=user_id"
    payload = json.dumps({})
    headers = {
        'Content-Type': 'application/json',
        'Authorization': f'Bearer {tenant_access_token}'
    }
    response = requests.request("POST", url, headers=headers, data=payload)
    # print(response.text)
    return response.json()

def extract_key_fields(feishu_datas):
    feishu_df = pd.DataFrame(feishu_datas)
    feishu_df['field_text'] = feishu_df.fields.apply(lambda x:x.get("文本", [{}])[0].get("text"))
    feishu_df['field_email'] = feishu_df.fields.apply(lambda x:x.get("email", [{}])[0].get("text"))
    feishu_df['field_select'] = feishu_df.fields.apply(lambda x:x.get("单选"))
    feishu_df['field_mobile'] = feishu_df.fields.apply(lambda x:x.get("电话号码"))
    feishu_df['field_no'] = feishu_df.fields.apply(lambda x:x.get("自动编号"))
    feishu_df['field_member1'] = feishu_df.fields.apply(lambda x:x.get("人员1", [{}])[0].get("name"))
    feishu_df['field_group1'] = feishu_df.fields.apply(lambda x:x.get("群组1", [{}])[0].get("name"))
    feishu_df['field_creator'] = feishu_df.fields.apply(lambda x:x.get("创建人", [{}])[0].get("name"))
    feishu_df['field_modifier'] = feishu_df.fields.apply(lambda x:x.get("修改人", [{}])[0].get("name"))
    feishu_df['field_member2'] = feishu_df.fields.apply(\
        lambda x: ','.join([val.get("name") for val in x.get("人员2", [{"name":""}])]))
    feishu_df['field_group2'] = feishu_df.fields.apply(\
        lambda x: ','.join([val.get("name") for val in x.get("群组2", [{"name":""}])]))
    feishu_df['field_url'] = feishu_df.fields.apply(lambda x:x.get("超链接",{}).get("link"))
    feishu_df['field_location'] = feishu_df.fields.apply(lambda x:x.get("地理位置",{}).get("full_address"))
    feishu_df['field_number'] = feishu_df.fields.apply(lambda x:x.get("数字"))
    feishu_df['field_progress'] = feishu_df.fields.apply(lambda x:x.get("进度"))
    feishu_df['field_money'] = feishu_df.fields.apply(lambda x:x.get("货币"))
    feishu_df['field_rating'] = feishu_df.fields.apply(lambda x:x.get("评分"))
    feishu_df['field_bool'] = feishu_df.fields.apply(lambda x:x.get("复选框"))
    feishu_df['field_date'] = pd.to_datetime(
        feishu_df.fields.apply(lambda x:28800 + int(x.get('日期',1000)/1000)),
        unit='s')
    feishu_df['field_createdtime'] = pd.to_datetime(
        feishu_df.fields.apply(lambda x:28800 + int(x.get('创建时间',1000)/1000)),
        unit='s')
    feishu_df['field_updatedtime'] = pd.to_datetime(
        feishu_df.fields.apply(lambda x:28800 + int(x.get('更新时间',1000)/1000)),
        unit='s')
    feishu_df['field_mulselect'] = feishu_df.fields.apply(lambda x:json.dumps(x.get("多选")))
    feishu_df['field_findnum'] = feishu_df.fields.apply(lambda x:json.dumps(x.get("查找引用数值")))
    feishu_df['field_numformula'] = feishu_df.fields.apply(lambda x:json.dumps(x.get("数字公式")))
    feishu_df['field_singleunion'] = feishu_df.fields.apply(lambda x:json.dumps(x.get("单向关联").get("link_record_ids")))
    feishu_df['field_doubleunion'] = feishu_df.fields.apply(lambda x:json.dumps(x.get("双向关联").get("link_record_ids")))
    feishu_df['field_file'] = feishu_df.fields.apply(\
        lambda x: json.dumps([val.get("url") for val in x.get("附件", [{"url":""}])]))
    
    feishu_df.drop(['fields'],axis=1,inplace=True)
    return feishu_df

def cre_mysql_table(create_table_sql):
    # from sqlalchemy import create_engine, text
    # 创建 SQLAlchemy engine 对象，这里以 MySQL 为例
    # engine = create_engine('mysql://username:password@host:port/dbname')
    engine = create_engine('mysql://root:password@127.0.0.1:3306/test')
    
    # 定义一个建表的 SQL 语句
    # create_table_sql = ''''''
    
    # 使用 execute() 方法执行 SQL 语句
    with engine.connect() as connection:
        connection.execute(text(create_table_sql))
        print('创建成功！')

def insert_mysql_table(feishu_df, table_name):
    # from sqlalchemy import create_engine
    
    # 创建 SQLAlchemy engine 对象，这里以 MySQL 为例
    # engine = create_engine('mysql://username:password@host:port/dbname')
    engine = create_engine('mysql://root:password@127.0.0.1:3306/test')
    
    # 将 DataFrame 直接写入 MySQL 数据库
    feishu_df.to_sql(name=table_name, con=engine, if_exists='append', index=False)
    print('写入成功！')

def main():
    app_id = 'your_app_id'
    app_secret = 'your_app_secret'
    tenant_access_token = get_tenant_access_token(app_id, app_secret)
    app_token = 'your_app_token'
    table_id = 'your_table_id'

    page_token = ''
    page_size = 5
    has_more = True
    feishu_datas = []
    while has_more:
        response = get_bitable_datas(tenant_access_token, app_token, table_id, page_token, page_size)
        if response['code'] == 0:
            page_token = response['data'].get('page_token')
            has_more = response['data'].get('has_more')
            # print(response['data'].get('items'))
            # print('\n--------------------------------------------------------------------\n')
            feishu_datas.extend(response['data'].get('items'))
        else:
            raise Exception(response['msg'])

    # 提取关键字段
    feishu_df = extract_key_fields(feishu_datas)
    # MySQL 建表
    create_table_sql = '''
        create table if not exists test.feishu_data_type_test(
        record_id               varchar(256)     comment '飞书记录id'
       ,field_text              varchar(256)     comment '文本'
       ,field_email             varchar(256)     comment 'email'
       ,field_select            varchar(256)     comment '单选'
       ,field_mobile            varchar(256)     comment '电话号码'
       ,field_no                varchar(256)     comment '自动编号'
       ,field_member1           varchar(256)     comment '人员1'
       ,field_group1            varchar(256)     comment '群组1'
       ,field_creator           varchar(256)     comment '创建人'
       ,field_modifier          varchar(256)     comment '修改人'
       ,field_member2           varchar(256)     comment '人员2'
       ,field_group2            varchar(256)     comment '群组2'
       ,field_url               varchar(256)     comment '超链接'
       ,field_location          varchar(256)     comment '地理位置'
       ,field_number            float            comment '数字'
       ,field_progress          float            comment '进度'
       ,field_money             float            comment '货币'
       ,field_rating            float            comment '评分'
       ,field_bool              bool             comment '复选框'
       ,field_date              datetime         comment '日期'
       ,field_createdtime       datetime         comment '创建时间'
       ,field_updatedtime       datetime         comment '更新时间'
       ,field_mulselect         json             comment '多选'
       ,field_findnum           json             comment '查找引用数值'
       ,field_numformula        json             comment '数字公式'
       ,field_singleunion       json             comment '单向关联'
       ,field_doubleunion       json             comment '双向关联'
       ,field_file              json             comment '附件'
    )
    '''
    cre_mysql_table(create_table_sql)
    # MySQL 表插入数据
    table_name = 'feishu_data_type_test'
    insert_mysql_table(feishu_df, table_name)

if __name__ == '__main__':
    main()

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.coloradmin.cn/o/1636675.html

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈，一经查实，立即删除！

相关文章

大型企业总分支多区域数据传输，效率为先还是安全为先？

大型企业总分支多区域数据传输，效率为先还是安全为先？

大型企业为了业务拓展需要，会在全国乃至全球各地设立分公司和办事机构，以便更好地处理当地事务，并进行市场的开拓和客户维护，此时，企业内部就衍生出了新的业务需求，即多区域数据传输。 多区域很难准确定义&…
阅读更多...
C++相关概念和易错语法（10）（定位new、模板）

C++相关概念和易错语法（10）（定位new、模板）

1.定位new 我们使用类来实例化对象，开辟空间的时候会自动去调用它的构造函数。但在那篇博客我就特意强调过，使用a.A()的方式是错误的，A()根本不会被识别为一个构造函数，而会被识别为A类型。因此我们要注意最好在实例化对象&#…
阅读更多...
test4282

test4282

欢迎关注博主 Mindtechnist 或加入【智能科技社区】一起学习和分享Linux、C、C、Python、Matlab，机器人运动控制、多机器人协作，智能优化算法，滤波估计、多传感器信息融合，机器学习，人工智能等相关领域的知识和技术。关…
阅读更多...
ChatGPT 网络安全秘籍（一）

ChatGPT 网络安全秘籍（一）

原文：zh.annas-archive.org/md5/6b2705e0d6d24d8c113752f67b42d7d8 译者：飞龙 协议：CC BY-NC-SA 4.0 前言 在不断发展的网络安全领域中，由 OpenAI 推出的 ChatGPT 所代表的生成式人工智能和大型语言模型（LLMs&#xf…
阅读更多...
首页最新 多IP浏览器防关联：如何配置多个独立且稳定的IP地址？

首页最新 多IP浏览器防关联：如何配置多个独立且稳定的IP地址？

在互联网时代，IP地址的重要性不言而喻。然而，IP关联问题却成为一项令人担忧的隐私和安全挑战。针对这个问题，多IP浏览器是一种解决方案，可以帮助用户单独配置多个独立且稳定的IP地址，有效地防止IP关联。 一、IP关联是…
阅读更多...
【HarmonyOS4学习笔记】《HarmonyOS4+NEXT星河版入门到企业级实战教程》课程学习笔记（七）

【HarmonyOS4学习笔记】《HarmonyOS4+NEXT星河版入门到企业级实战教程》课程学习笔记（七）

课程地址： 黑马程序员HarmonyOS4NEXT星河版入门到企业级实战教程，一套精通鸿蒙应用开发 （本篇笔记对应课程第 14 节） P14《13.ArkUI组件-自定义组件》 将可变部分封装成组件的成员变量： 1、首先给标题添加两个图标&am…
阅读更多...
百川crm系统 教育crm系统 一款高效的培训机构管理系统

百川crm系统 教育crm系统 一款高效的培训机构管理系统

在教育培训行业日益竞争激烈的今天，如何精准把握客户需求、提升服务质量、实现客户价值最大化，成为了每一家教育培训机构都必须面对的问题。为此，一款高效、智能的CRM客户管理系统成为了教育培训机构不可或缺的得力助手。本文将为您详细介绍这…
阅读更多...
使用xshell工具连接ubuntu的root账户被拒绝的解决方法

使用xshell工具连接ubuntu的root账户被拒绝的解决方法

问题描述： 我在使用xshell工具远程连接Ubuntu虚拟机的过程中，如果连接的是的普通用户则xshell工具可以正常连接，但是当我向连接ubuntu系统的root用户，即便是密码输入正确但还是不能连接成功。不能连接成功的截图如下： …
阅读更多...
C#---使用Coravel实现定时任务

C#---使用Coravel实现定时任务

Coravel是一款框架轻，使用简单，支持秒级定时任务。 1.添加NuGet引用 2.定义自己的工作任务 using Coravel.Invocable; using Microsoft.Extensions.Logging; using System; using System.Collections.Generic; using System.Linq; using System.Thread…
阅读更多...
【navicat】oracle library is not loaded 问题复现和解决方案

【navicat】oracle library is not loaded 问题复现和解决方案

问题原因：客户端oci版本安装错误，navicat需要64位的oci,但是使用32位的oci。 解决方案：官网下载64位oci进行配置。本次演示的解决多了splplus，其实不必要安装也能运行。 首先判断是否数据库已经打开 尝试使用splplus连接数据库 1…
阅读更多...
MYSQL自联结

MYSQL自联结

文章目录 查找：生产ID为DTNTR物品的供应商，生产的其他物品。子查询自联结比较 查找：生产ID为DTNTR物品的供应商，生产的其他物品。 子查询 SELECTprod_id,prod_name FROMproducts WHEREvend_id ( SELECT vend_id FROM product…
阅读更多...
静电纺聚丙烯腈（PAN）纳米纤维膜

静电纺聚丙烯腈（PAN）纳米纤维膜

静电纺聚丙烯腈（PAN）纳米纤维膜是通过静电纺丝技术制备的一种纳米级纤维膜材料。静电纺丝技术利用高压电场使带电的聚合物溶液或熔体在喷丝口形成细流，经过拉伸、固化后形成纳米纤维，最终收集形成纳米纤维膜。 PAN纳米纤维膜具有以…
阅读更多...
Unity涂鸦纹理实现

Unity涂鸦纹理实现

文章目录 前言实现过程UV坐标和UI坐标对齐修改像素代码 前言 心血来潮实现下场景中提供一张纹理进行涂鸦的功能。 最终实现效果: 实现过程 UV坐标和UI坐标对齐 这里的纹理使用了UGUI的Canvas进行显示，所以这里使用一张RawImage。 因为Unity的视口坐标是以左下角…
阅读更多...
微服务之SpringCloud AlibabaSeata处理分布式事务

微服务之SpringCloud AlibabaSeata处理分布式事务

一、概述 1.1背景 一次业务操作需要跨多个数据源或需要跨多个系统进行远程调用，就会产生分布式事务问题 but 关系型数据库提供的能力是基于单机事务的，一旦遇到分布式事务场景，就需要通过更多其他技术手段来解决问题。 全局事务：…
阅读更多...
SOLIDWORKS2021：革新设计界的里程碑，引领您迈向数字化制造未来

SOLIDWORKS2021：革新设计界的里程碑，引领您迈向数字化制造未来

在2021年，SOLIDWORKS 2021的发布犹如一股清新的创新之风，吹遍了整个设计与工程领域。作为业界领先的3D CAD解决方案，SOLIDWORKS 2021不仅巩固了其在设计软件领域的领导地位，更以前所未有的方式激发了设计者的创造力与生产力。亿达…
阅读更多...
Flask教程3：jinja2模板引擎

Flask教程3：jinja2模板引擎

文章目录 模板的导入与使用 模板的导入与使用 Flask通过render_template来实现模板的渲染，要使用这个方法，我们需要导入from flask import rander_template，模板中注释需放在{# #}中 模板的第一个参数为指定的模板文件名称，如自定…
阅读更多...
有关CSS中排版常见问题（清除默认样式问题 + 元素居中问题 + 元素之间的空白问题 + 行内块的幽灵空白问题）

有关CSS中排版常见问题（清除默认样式问题 + 元素居中问题 + 元素之间的空白问题 + 行内块的幽灵空白问题）

前言：在练习CSS排版的时候，我们经常会遇到一些排版上的问题，那么我们如何去解决这些问题呢？本篇文章给出了一些新手在练习排版时候可能会遇到的问题的解决方案。 ✨✨✨这里是秋刀鱼不做梦的BLOG ✨✨✨想要了解更多内容可以访问我…
阅读更多...
1天搞定uniApp+Vue3+vite+Element UI或者Element Plus开发学习，使用vite构建管理项目,HBuilderX做为开发者工具

1天搞定uniApp+Vue3+vite+Element UI或者Element Plus开发学习，使用vite构建管理项目,HBuilderX做为开发者工具

我们通常给小程序或者app开发后台时，不可避免的要用到可视化的数据管理后台，而vue和Element是我们目前比较主流的开发管理后台的主流搭配。所以今天石头哥就带大家来一起学习下vue3和Element plus的开发。 准备工作 1，下载HBuilderX 开发者…
阅读更多...
Python 可以对数据进行哪些可视化？

Python 可以对数据进行哪些可视化？

Python 可视化 一、条形图（或柱状图） 1.代码如下： import matplotlib.pyplot as plt import pandas as pddf pd.DataFrame({County:[America,Canada,Australia,Germany,French,China],GDP:[80,30,70,80,60,75] })plt.bar(df[County],df[G…
阅读更多...
Linux 虚拟主机切换php版本及参数

Linux 虚拟主机切换php版本及参数

我使用的Hostease的Linux虚拟主机产品,由于网站程序需要支持高版本的PHP,程序已经上传到主机，但是没有找到切换PHP以及查看PHP有哪些版本的位置，因此咨询了Hostease的技术支持，寻求帮助了解到可以实现在cPanel面板上找到此切换PHP版本的按钮&…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023