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

一、引入

上一篇了解了飞书 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 jsondef 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 = 5has_more = Truefeishu_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_datasif __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, textdef 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_dfdef 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 = 5has_more = Truefeishu_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.mzph.cn/news/830846.shtml

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

相关文章

【Leetcode每日一题】 动态规划 - 简单多状态 dp 问题 - 按摩师（难度⭐）（64）

【Leetcode每日一题】 动态规划 - 简单多状态 dp 问题 - 按摩师（难度⭐）（64）

1. 题目解析 题目链接：面试题 17.16. 按摩师 这个问题的理解其实相当简单，只需看一下示例，基本就能明白其含义了。 2.算法原理 一、状态定义 在解决这类动态规划问题时，首先我们需要明确状态的定义。对于本题，我们…
阅读更多...
在mac上安装node.js及使用npm,yarn相关命令教程

在mac上安装node.js及使用npm,yarn相关命令教程

1、安装node.js 官网：Node.js — Download Node.js 选择需要的版本，点击DownLoad 2、点击继续，直到安装成功。 2.1打开终端输入命令node -v 显示版本号则说明已安装成功 3、全局安装yarn命令 1、sudo npm install --global yarn &#xf…
阅读更多...
Git学习笔记（五）IDEA使用Git

Git学习笔记（五）IDEA使用Git

在前面几篇文章中，我们已经介绍了git的基础知识，知道了其主要作用是用来进行代码的版本管理；并且已经介绍了Git操作的常用命令。在日常的开发环境下，除了通过Bash命令行来操作Git之外，我们另外一种常用的操作方式则是直…
阅读更多...
基于STC12C5A60S2系列1T 8051单片机的Proteus中的单片机发送一帧或一串数据给串口调试助手软件接收区显示出来的串口通信应用

基于STC12C5A60S2系列1T 8051单片机的Proteus中的单片机发送一帧或一串数据给串口调试助手软件接收区显示出来的串口通信应用

基于STC12C5A60S2系列1T 8051单片机的Proteus中的单片机发送一帧或一串数据给串口调试助手软件接收区显示出来的串口通信应用 STC12C5A60S2系列1T 8051单片机管脚图STC12C5A60S2系列1T 8051单片机串口通信介绍STC12C5A60S2系列1T 8051单片机串口通信的结构基于STC12C5A60S2系列…
阅读更多...
Python_GUI框架 PyQt 与 Pyside6的介绍

Python_GUI框架 PyQt 与 Pyside6的介绍

Python_GUI框架 PyQt 与 Pyside6的介绍 一、简介 在Python的GUI（图形用户界面）开发领域，PyQt和PySide6是两个非常重要的工具包。它们都基于Qt库，为Python开发者提供了丰富的GUI组件和强大的功能。当然Python也有一些其他的GUI工…
阅读更多...
手把手教数据结构与算法：优先级队列（银行排队问题）

手把手教数据结构与算法：优先级队列（银行排队问题）

队列 基本概念 队列的定义 队列（Queue）：队列是一种常见的数据结构，遵循先进先出（First-In-First-Out, FIFO）的原则。在队列中，元素按照进入队列的顺序排列。队列是一个线性的数据结构&#x…
阅读更多...
【团体程序设计天梯赛】往年关键真题 L2-036 网红点打卡攻略 模拟 L2-037 包装机 栈和队列 详细分析完整AC代码

【团体程序设计天梯赛】往年关键真题 L2-036 网红点打卡攻略 模拟 L2-037 包装机 栈和队列 详细分析完整AC代码

【团体程序设计天梯赛 往年关键真题 详细分析&完整AC代码】搞懂了赛场上拿下就稳 【团体程序设计天梯赛 往年关键真题 25分题合集 详细分析&完整AC代码】（L2-001 - L2-024）搞懂了赛场上拿下就稳了 【团体程序设计天梯赛 往年关键真题 25分题合…
阅读更多...
《Redis使用手册之列表》

《Redis使用手册之列表》

《Redis使用手册之列表》 目录 **《Redis使用手册之列表》****LPUSH：将元素推入列表左端****LPUSHX、RPUSHX：只对已存在的列表执行推入操作****LPOP：弹出列表最左端的元素****RPOP：弹出列表最右端的元素****RPOPLPUSH：…
阅读更多...
ElasticSearch教程入门到精通——第二部分（基于ELK技术栈elasticsearch 7.x新特性）

ElasticSearch教程入门到精通——第二部分（基于ELK技术栈elasticsearch 7.x新特性）

ElasticSearch教程入门到精通——第二部分（基于ELK技术栈elasticsearch 7.x新特性） 1. JavaAPI-环境准备1.1 新建Maven工程——添加依赖1.2 HelloElasticsearch 2. 索引2.1 索引——创建2.2 索引——查询2.3 索引——删除 3. 文档3.1 文档——重构3.2 文…
阅读更多...
SQL 基础 | BETWEEN 的常见用法

SQL 基础 | BETWEEN 的常见用法

在SQL中，BETWEEN是一个操作符，用于选取介于两个值之间的数据。 它包含这两个边界值。BETWEEN操作符常用于WHERE子句中，以便选取某个范围内的值。 以下是BETWEEN的一些常见用法： 选取介于两个值之间的值： 使用 BETWEEN来…
阅读更多...
基于昇腾AI | 英码科技EA500I使用AscendCL实现垃圾分类和视频物体分类应用

基于昇腾AI | 英码科技EA500I使用AscendCL实现垃圾分类和视频物体分类应用

现如今，人工智能迅猛发展，AI赋能产业发展的速度正在加快，“AI”的需求蜂拥而来，但AI应用快速落地的过程中仍存在很大的挑战：向下需要适配的硬件，向上需要完善的技术支持，两者缺一不可。 基于此&…
阅读更多...
【配置】Docker搭建JSON在线解析网站

【配置】Docker搭建JSON在线解析网站

云服务器打开端口8787 连接上docker运行 docker run -id --name jsonhero -p 8787:8787 -e SESSION_SECRETabc123 henryclw/jsonhero-webhttp://ip:8787访问 Github：地址
阅读更多...
STM32 看门狗WDG

STM32 看门狗WDG

一、看门狗（Watchdog） 看门狗可以监控程序的运行状态，当程序因为设计漏洞、硬件故障、电磁干扰等原因，出现卡死或跑飞现象时，看门狗能及时复位程序，避免程序陷入长时间的罢工状态，保证系统的可靠…
阅读更多...
Django后台项目开发实战一

Django后台项目开发实战一

开发环境使用 Anaconda, IDE 使用 pycharm 第一阶段 创建 Django 项目 在 Anaconda Prompt 中逐步输入下面的命令（之后的所有命令都在这个） 首先创建一个虚拟环境，名称自拟，python 版本我这里使用 3.9.18 关于 python 版本和…
阅读更多...
PotatoPie 4.0 实验教程(28) —— FPGA实现sobel算子对摄像头图像进行边缘提取

PotatoPie 4.0 实验教程(28) —— FPGA实现sobel算子对摄像头图像进行边缘提取

什么是sobel算子？ Sobel 算子是一种常用的边缘检测算子，用于在图像中检测边缘。它基于对图像进行梯度运算，可以帮助识别图像中灰度值变化较大的区域，从而找到图像中的边缘。 Sobel 算子通过计算图像的水平和垂直方向的一阶导数来…
阅读更多...
Linux 虚拟主机切换php版本及参数

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

我使用的Hostease的Linux虚拟主机产品,由于网站程序需要支持高版本的PHP,程序已经上传到主机，但是没有找到切换PHP以及查看PHP有哪些版本的位置，因此咨询了Hostease的技术支持，寻求帮助了解到可以实现在cPanel面板上找到此切换PHP版本的按钮&…
阅读更多...
2024 五一杯高校数学建模邀请赛（A题）|钢板最优切割路径问题|建模秘籍文章代码思路大全

2024 五一杯高校数学建模邀请赛（A题）|钢板最优切割路径问题|建模秘籍文章代码思路大全

铛铛！小秘籍来咯！ 小秘籍团队独辟蹊径，运用Dijkstra算法，最优路径切割等强大工具，构建了这一题的详细解答哦！ 为大家量身打造创新解决方案。小秘籍团队，始终引领着建模问题求解的风潮。 抓紧小秘…
阅读更多...
Stability AI 推出稳定音频 2.0：为创作者提供先进的 AI 生成音频

Stability AI 推出稳定音频 2.0：为创作者提供先进的 AI 生成音频

概述 Stability AI 的发布再次突破了创新的界限。这一尖端模型以其前身的成功为基础，引入了一系列突破性的功能，有望彻底改变艺术家和音乐家创建和操作音频内容的方式。 Stable Audio 2.0 代表了人工智能生成音频发展的一个重要里程碑，为质量…
阅读更多...
Flask模版详解

Flask模版详解

Flask模版详解 概述Jinja2模板引擎渲染模版的步骤变量控制结构自定义错误页面链接静态文件 概述 模板是一个包含响应文本的文件，其中包含用占位变量表示的动态部分，其具体值只在请求的上下文中才能知道。使用真实值替换变量，再返回最终得到的…
阅读更多...
Android4.4真机移植过程笔记（一）

Android4.4真机移植过程笔记（一）

1、RK源码编译 获取内核源码： git clone git172.28.1.172:rk3188_kernel -b xtc_ok1000 内核编译环境： 从172.28.1.132编译服务器的/data1/ZouZhiPing目录下拷贝toolchain.tar.gz（交叉编译工具链）并解压到与rk3188_kernel同级目…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023