飞书API（4）：筛选数据的三种思路

截止到上一篇，终于通过飞书 API 完整获取到飞书多维表的数据。
但是，有些场景，比如数据源会出现脏数据，毕竟如果是运营过程多人协作维护的数据，要想保持数据完美简直是天方夜谭！再比如我们不需要完整的数据，只需要某个类别的数据即可，这个时候，就需要我们对数据进行筛选处理。

本文探讨三种处理思路：

思路一：先读取所有数据，再处理；
思路二：调接口读取时直接处理；
思路三：新建视图进行筛选。

思路一：先读取所有数据，再处理

这种思路比较简单，就是在上一篇的基础上，使用 Python 进行数据处理。
上一篇，获取到的数据都以列表feishu_datas的形式返回了，采用 pandas 库来进行处理，把列表 feishu_datas 转为 DataFrame 之后。对对应的列进行筛选处理即可。
比如说只筛选出支付平台是“淘宝”的数据。
为了便于维护和对比查看，这里直接新增一个函数，该函数传入两个参数，一个是飞书的列表数据feishu_datas，一个是需要筛选的平台名称，然后在main()函数中调用它即可。

def filter_platform(feishu_datas, platform):df = pd.DataFrame(feishu_datas)df_platform = df[df.fields.apply(lambda x: x.get('支付平台')) == platform]filter_feishu_datas = df_platform.to_dict(orient='records')return filter_feishu_datas

在这里插入图片描述

如果是其他的筛选条件可以参考类似的方法执行筛选。

思路二：调接口读取时直接处理

前面在调用飞书接口的时候都是使用默认方式读取，即直接读取所有的数据，未使用非必须参数。
在非必须参数中，有一个 filter 参数，它就可以实现筛选，本思路就是使用它在调接口的时候对数据进行筛选。
官方文档如下：

从文档中可以看到，filter 参数需要给：conjunction（条件之间的关系）和 conditions（条件）。
- conjunction：当有多个条件的时候，是否同时满足所有条件（值：and）才返回数据还是满足任一个条件（值：or）即可。注意：不支持二者嵌套使用，比如条件A and (条件B or 条件C)。
- conditions：支持多个条件，每个条件使用大括号{}包裹，每个条件给定三个键值对，键分别是：field_name、operator 和 value，值根据需求给即可。

还是拿上面的例子，筛选支付平台是“淘宝”的数据。
该思路只需要在get_bitable_datas()函数中，将payload赋值内容改为以下代码即可。

payload = json.dumps({"filter":{"conjunction":"and","conditions":[{"field_name":"支付平台","operator":"is","value":["淘宝"]}]}})

改动图示如下：

该示例的代码是硬代码，为了更好维护，可以将变量payload的值改为参数进行传递，从函数外部传递payload_data值。

补充：观察每一页打印的数据，可以发现，接口数据也是按照所有数据进行分页，然后按页筛选数据。而不是将所有符合的数据筛选完再分页返回。

思路三：新建多维表视图进行筛选

该方法的原理和前面两种有点不同，前面两种都是通过 API 对所有数据进行筛选。该方法是先在飞书上新建视图进行筛选数据，然后通过 API 访问对应视图数据，读取时仅读取筛选后的结果。
还是以筛选支付平台是“淘宝”的数据。
先在多维表上新建一个视图，命名为“淘宝”，筛选：“支付平台=淘宝”。

然后在非必选参数中，需要指定对应的视图id（view_id），参数说明如下图：

在这里插入图片描述

view_id 的值在链接上直接获取即可：选择对应的视图，然后复制“view=”后面的字符串。

取思路二的代码，将payload_data修改为下面的值即可。

payload_data = {"view_id":"vewUjtcfIX"}

改完执行结果如下：

小结

本文探讨了三种处理多维表筛选数据的思路：

思路一：获取所有数据，使用 pandas 进行数据筛选；
思路二：调查询记录接口时，传递 filter 参数进行数据筛选；
思路三：在多维表新建视图进行筛选，然后指定视图，读取所有数据。

附录

注：使用上一篇的 while 循环代码进行修改。
思路一完整代码：

import requests
import json
import pandas as pddef 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 filter_platform(feishu_datas, platform):df = pd.DataFrame(feishu_datas)df_platform = df[df.fields.apply(lambda x: x.get('支付平台')) == platform]filter_feishu_datas = df_platform.to_dict(orient='records')return filter_feishu_datasdef 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'])filter_feishu_datas = filter_platform(feishu_datas, '淘宝')return filter_feishu_datasif __name__ == '__main__':feishu_datas = main()print(feishu_datas)

思路二完整代码：

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, payload_data={}):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(payload_data)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 = []payload_data = {"filter":{"conjunction":"and","conditions":[{"field_name":"支付平台","operator":"is","value":["淘宝"]}]}}while has_more:response = get_bitable_datas(tenant_access_token, app_token, table_id, page_token, page_size, payload_data)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)

思路三完整代码：

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, payload_data={}):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(payload_data)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 = []payload_data = {"view_id":"vewUjtcfIX"}while has_more:response = get_bitable_datas(tenant_access_token, app_token, table_id, page_token, page_size, payload_data)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)