截止到上一篇,终于通过飞书 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 pd
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 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
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'])
filter_feishu_datas = filter_platform(feishu_datas, '淘宝')
return filter_feishu_datas
if __name__ == '__main__':
feishu_datas = main()
print(feishu_datas)
思路二完整代码:
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, 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 = 5
has_more = True
feishu_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_datas
if __name__ == '__main__':
feishu_datas = main()
print(feishu_datas)
思路三完整代码:
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, 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 = 5
has_more = True
feishu_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_datas
if __name__ == '__main__':
feishu_datas = main()
print(feishu_datas)