飞书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.rhkb.cn/news/317189.html

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

相关文章

完美解决AttributeError: module ‘backend_interagg‘ has no attribute ‘FigureCanvas‘

遇到这种错误通常是因为matplotlib的后端配置问题。在某些环境中,尤其是在某些特定的IDE或Jupyter Notebook环境中,可能会因为后端配置不正确而导致错误。错误信息提示 module backend_interagg has no attribute FigureCanvas 意味着当前matplotlib的后…

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

在互联网时代,IP地址的重要性不言而喻。然而,IP关联问题却成为一项令人担忧的隐私和安全挑战。针对这个问题,多IP浏览器是一种解决方案,可以帮助用户单独配置多个独立且稳定的IP地址,有效地防止IP关联。 一、IP关联是…

【Python小练】求斐波那契数列第n个数

题目 输出斐波那契数列第n个数。 分析 首先我们要知道,斐波那契数列,这个数列从第三位开始等于前两个数的和,要知道数列第n个数(n>2),就要知道其前两相的值,着就需要用到递归了。来看一下吧…

开源、轻量、易用的服务器实时监控工具:哪吒探针

本文首发于只抄博客,欢迎点击原文链接了解更多内容。 前言 哪吒探针是一个开源、轻量、易用的服务器监控、运维工具,它有以下几个特点: 一键安装:可以一键安装面板与 Agent,并且支持 Linux、Windows、MacOS、OpenWRT…

纯血鸿蒙APP实战开发——发布图片评论

介绍 本示例将通过发布图片评论场景,介绍如何使用startAbilityForResult接口拉起相机拍照,并获取相机返回的数据。 效果图预览 使用说明 通过startAbilityForResult接口拉起相机,拍照后获取图片地址。 实现思路 创建CommentData类&#…

VSCode 配置 CMake

VSCode 配置 C/C 环境的详细过程可参考:VSCode 配置 C/C 环境 1 配置C/C编译环境 方案一 如果是在Windows,需要安装 MingW,可以去官网(https://sourceforge.net/projects/mingw-w64/)下载安装包。 注意安装路径不要出现中文。 打开 windows…

虚析构与纯虚析构

这里的new Cat("Tom"&#xff09;是由于基类函数中的构造函数里面带有string变量 1. 法一:利用虚函数&#xff0c;虚化基类中的析构函数 virtual ~Animal() { cout << "动物的析构函数调用" << endl; } 2. 法二:利用纯…

VitePress 构建的博客如何部署到 Netlify 平台?

VitePress 构建的博客如何部署到 Netlify 平台&#xff1f; 前言 之前写了篇文章【使用 Vitepress 构建博客并部署到 github 平台】&#xff0c;有个老哥说 github page 访问太慢了&#xff0c;希望放到 Netlify 平台上面。 咱也没部署过&#xff0c;就试了一下&#xff0c;发…

低功耗数字IC后端设计实现典型案例| UPF Flow如何避免工具乱用Always On Buffer?

下图所示为咱们社区低功耗四核A7 Top Hierarchical Flow后端训练营中的一个案例&#xff0c;设计中存在若干个Power Domain&#xff0c;其中Power Domain2(简称PD2)为default Top Domain&#xff0c;Power Domain1&#xff08;简称PD1&#xff09;为一个需要power off的domain&…

【再探】设计模式—抽象工厂及建造者模式

抽象工厂模式和建造者模式都属于创建型模式。两者都能创建对应的对象&#xff0c;而创建者模式更侧重于创建复杂对象&#xff0c;将对象的创建过程封装起来&#xff0c;让客户端不需要知道对象的内部细节。 1 抽象工厂模式 需求&#xff1a; 在使用工厂方法模式时&#xff0…

Postgresql 从小白到高手 十一 :数据迁移ETL方案

文章目录 Postgresql 数据迁移ETL方案1、Pg 同类型数据库2 、Pg 和 不同数据库 Postgresql 数据迁移ETL方案 1、Pg 同类型数据库 备份 : pg_dump -U username -d dbname -f backup.sql插入数据&#xff1a; psql -U username -d dbname -f backup.sqlpg_restore -U username…

【Godot4.2】自定义Todo清单类 - myTodoList

概述 在写myList类的时候&#xff0c;就想到可以写一个类似的Todo清单类。 基础思路 本质还是在内部维护一个数组&#xff0c;在其基础上进行增删改查操作的封装为了方便存储数据&#xff0c;编写一个自定义内置类TodoItem&#xff0c;内部数组就变成了Array[TodoItem]类型的…

JENKINS 安装,学习运维从这里开始

Download and deployJenkins – an open source automation server which enables developers around the world to reliably build, test, and deploy their softwarehttps://www.jenkins.io/download/首先点击上面。下载Jenkins 为了学习&#xff0c;从windows开始&#x…

mysql-sql练习-5-行列互转

目录 成绩单 简单互转 需求 多行转多列 分组 判断 聚合 理解 分组 合并 逆向需求 多列转多行 输出 合并 abc 去重 合并 拆分 需求 建表 多行转多列 逆向需求 多列转多行 拆分 按长度 拆分 按个数 成绩单 简单互转 需求 多行转多列 分组 判断 聚合 with tmp as(--…

【快速入门 LVGL】-- 5、Gui Guider界面移植到STM32工程

上篇&#xff0c;我们已学习&#xff1a;【快速入门 LVGL】-- 4、显示中文 工程中添加了两个按钮作示范。运行效果如图&#xff1a; 本篇&#xff1a;把Gui Guider设计好的界面&#xff0c;移植到STM32工程。 特别地&#xff1a; 在使用Gui Guider进行界面设计时&#xff0c;应…

如何在Linux上安装Python?2024Python安装教程

在Linux上安装Python并不难&#xff0c;对于Ubuntu或Debian系统&#xff0c;使用命令sudo apt install python3&#xff1b;对于CentOS、Red Hat或Fedora系统&#xff0c;使用命令sudo yum install python3。 如何在Linux上安装Python&#xff1f; 确切的安装步骤有所不同&am…

LabVIEW 2024安装教程(附免费安装包资源)

鼠标右击软件压缩包&#xff0c;选择“解压到LabVIEW.2024”。 返回解压后的文件夹&#xff0c;鼠标右击“ni_labview-2024”选择“装载”。 鼠标右击“Install”选择“以管理员身份运行”。 点击“我接受上述2条许可协议”&#xff0c;然后点击“下一步”。 点击“下一步”。 …

用html画一个四叶草

<!DOCTYPE html> <html lang"en" > <head> <meta charset"UTF-8"> <title>四叶草</title> <link href"" rel"stylesheet"> <link rel"stylesheet" href"css/style.css&q…

xLua详解

目录 环境准备xLua导入 C#调用LuaLua解析器Lua文件加载重定向Lua解析管理器全局变量的获取全局函数的获取List和Dictionary映射table类映射table接口映射tableLuaTable映射table Lua调用C#准备工作Lua使用C#类Lua调用C#枚举Lua使用C# 数组 List 字典数组List字典 Lua使用C#扩展…

Ubuntu如何更换 PyTorch 版本

环境&#xff1a; Ubuntu22.04 WLS2 问题描述&#xff1a; Ubuntu如何更换 PyTorch 版本考虑安装一个为 CUDA 11.5 编译的 PyTorch 版本。如何安装旧版本 解决方案&#xff1a; 决定不升级CUDA版本&#xff0c;而是使用一个与CUDA 11.5兼容的PyTorch版本&#xff0c;您可…