一、引入
上一篇了解了飞书 28 种数据类型通过接口读取到的数据结构，本文开始探讨如何将这些数据写入 MySQL 数据库。
这个工作流的起点是从 API 获取到的一个完整的数据，终点是写入 MySQL 数据表，表结构和维格表结构类似。在过程中可以有不同的工作流程，可以是将接口返回的所有数据作为一个值，直接写入 MySQL 表中，再使用 MySQL 对该值进行解析，处理成不同的列，然后再新建一张表单存储，这种方法入库比较简单粗暴，但是 MySQL 的处理会比较复杂，更侧重 MySQL 的对 json 结构的解析处理能力；也可以使用 Python 对接口数据进行进行处理，提取出各个数据列以及对应的值，再入库。
本文主要探讨后者。
虽然飞书的多维表提供了 28 中数据类型，但是本质上，很多数据类型记录的内容从 MySQL 的数据类型的角度上看是相似的，可能有点绕，举个例子：如下图，文本列和单选列虽然在飞书多维表是分为两个数据类型，但是它们的列值：单选1、单选2、这是文本111、这是文本222，本质上都是一个字符串，所以在入库处理时可以都设置为 MySQL 的 varchar 数据类型。
从 MySQL 的数据类型的角度上看，我们可以将飞书的这 28 中数据类型划分为五类，分别是字符串、数据、时间、列表和布尔值，参考如下：
| 类型描述 | MySQL 常用数据类型 | 飞书数据类型编码 | 飞书数据类型中文描述 |
|---|---|---|---|
| 字符串 | text、varchar、char | 1、3、11、13、15、22、23、1003、1004、1005 | 多行文本、条码、Email邮箱、单选、人员、电话号码、超链接、附件、地理位置、群组、创建人、修改人、自动编号 |
| 数字 | double、float、bigint、int | 2 | 数字、进度、货币、评分 |
| 布尔值 | bool | 7 | 复选框 |
| 时间 | datetime、date、timestamp | 5、1001、1002 | 日期、创建时间、最后更新时间 |
| 列表格式字符串 | json | 4、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、创建多维表，并设置应用操作多维表的权限）。
我用于测试的数据结构如下，可能你的和我的命名不同，所以以下代码的列名称根据你的列名进行修改即可。
我的测试数据为：
为了方便调试，本次使用 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 中运行，得到结果如下
接下来将获取到的数据feishu_datas通过 Pandas 的 DataFrame 来处理。
import pandas as pd
feishu_df = pd.DataFrame(feishu_datas)
feishu_df
打印结果如下：
上篇，我们已经了解到每个飞书数据类型返回的数据结构，接下来就结合它来取值。
2.3 提取字符串数据列
2.3.1 多行文本、条码、Email邮箱
多行文本、条码、Email邮箱这三者的数据结构类型类似，可以统一处理，即取“text”的值。
| 数据类型编码 | 数据类型中文描述 | 数据类型对应英文描述 | 数据示例 |
|---|---|---|---|
| 1 | 多行文本、条码 | Text,Barcode | “多行文本”: [{“text”: “我是文本1”,“type”: “text”}] |
| 1 | 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]进行索引。
解决该问题，其实很简单，填充一个默认值即可，这个填充是在get()传递第二个参数，而不是对 pandas 列进行填充。该参数需要根据后面取值的结构进行适配，这里后面通过索引取一次，再根据键取一次，索引需要给一个嵌套字典的列表结构，即[{}]。
顺带说一下，为什么要使用
x.get("文本")，而不使用x["文本"]？
第一，前者兼容空值，即使是空值也不会报错，而是返回 None，后者则直接报错找不到对应的键；
第二，前者可以传递第二个参数，当对象是空值时，返回该参数，从而支持后续再次取值。
当然，对 pandas 列进行空值填充也是一种方案。代码示例如下：
feishu_df.fields.apply(lambda x:x.get("文本", [{}])[0].get("text"))
结果参考如下：
同理，可以对 Email邮箱 类型的列做同样的处理，将列名进行修改即可，示例如下：
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("自动编号"))
结果如下：
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"))
结果如下：
上面是单值的情况，如果是多值，还需要进行遍历取值。对于多值的情况，此处我的处理方法是把多个值通过逗号链接起来。比如选择了“张三”和“李四”，处理的结构为：张三,李四。
注意：这里的默认值需要做一层处理，因为 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":""}])]))
结果如下：
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"))
结果如下：
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"))
结果如下：
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("评分"))
结果如下：
2.4.2 复选框
复选框取值方式和数字的取值方式一样。
| 数据类型编码 | 数据类型中文描述 | 数据类型对应英文描述 | 数据示例 |
|---|---|---|---|
| 7 | 复选框 | Checkbox | “复选框”: true |
参考代码：
feishu_df.fields.apply(lambda x:x.get("复选框"))
结果如下：
在实际的生产过程中，见到比较多表示布尔值的字段，可能是使用数字 0 和 1 来表示，如果要改为数字，可加一个三元表达式进行判断，参考代码如下：
feishu_df.fields.apply(\
lambda x:1 if x.get("复选框") else(None if x.get("复选框") is None else 0))
结果如下：
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')
结果如下：
2.6 提取列表格式字符串数据列
注意：这是 json 并不是列表！需要使用 json 库将列表转为 json 格式。
2.6.1 多选
多选列存为列表，直接取列值即可。
| 数据类型编码 | 数据类型中文描述 | 数据类型对应英文描述 | 数据示例 |
|---|---|---|---|
| 4 | 多选 | MultiSelect | “多选”: [“多选11”,“多选22”] |
参考代码：
feishu_df.fields.apply(lambda x:json.dumps(x.get("多选")))
结果如下：
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("数字公式")))
结果如下：
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")))
结果如下：
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":""}])]))
结果如下：
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'