使用飞书开发平台提供的api接口,去爬取多维表格的数据,并保存在本地,也可以爬取电子表格的数据,但是电子表格的相关api在我使用的时候是不提供调试的,可以将电子表格转变成多维表格 第一次看这种api文档来书写代码完成业务逻辑,并且我本身是用的java语言,对于python并不是很熟悉,使用python代码写的并不是很好,只是看见csdn上面关于飞书api的帖子较少,(当时自己也搜过,没有看见自己想要的),所以才写了这个博客(也是自己第一次写),希望给大家做一个参考,如果有什么问题的话,也希望大家指正 大概的流程: 1:第一步在飞书开发者后台创建自己的应用,并赋予相应的权限 2:看开发文档,相应的api,那些参数的含义,调试后,它那里会有写好的代码(响应体),写 代码逻辑获取自己想要的内容 我的代码大概的流程:控制台读取相关参数,加密写入到本地的配置文件,通过读取配置文件的相关参数,获取信息,传参,爬取数据 import http.client import json import hashlib from grammar import TransferParametersAndEncryption from urllib.parse import urlparse, parse_qs import hashlib import xml.etree.ElementTree as ET import base64 #加密:md5加密是不可逆的,这里应该使用其他的加密方式,AES或bcrypt,sm4,但是我导报的时候会出现问题,这里需要改进 def encrypt(parameter): # 使用MD5进行密码加密,你也可以选择其他更安全的加密算法 hash_object = hashlib.md5(parameter.encode()) return hash_object.hexdigest() #传参:向xml写入table的一些信息 def input_parameters(): # 获取app_id,app_secret,url path = input("配置文件保存的地址 ") user_appp_id = input("请输入app_id ") user_app_secret = input("请输入app_secret ") url = input("多维表格的url地址: ") table_name = input("请输入你要爬取的表名 ") # 加密appid,appSecret encrypted_app_id = encrypt(user_appp_id) encrypted_app_secret = encrypt(user_app_secret) # 创建XML文档 root = ET.Element('table_info') user_element = ET.SubElement(root, 'Path') user_element.text = path # 先使用原来的明文 password_element = ET.SubElement(root, 'User_appp_id') # password_element.text = encrypted_app_id password_element.text = user_appp_id password_element = ET.SubElement(root, 'User_app_secret') # password_element.text = encrypted_app_secret password_element.text = user_app_secret password_element = ET.SubElement(root, 'Url') password_element.text = url password_element = ET.SubElement(root, 'Table_name') password_element.text = table_name # 创建XML文档对象 tree = ET.ElementTree(root) # 将XML文档保存到文件 tree.write(path, encoding='utf-8', xml_declaration=True) print("用户名和密码已保存到XML配置文件中。") #获取xml中的信息,以字典的类型返回 def get_parameters(): # 获取app_id,app_secret,url path = input("爬取数据所需要的配置文件的地址 ") tree = ET.parse(path) root = tree.getroot() path = None user_appp_id = None user_app_secret = None url = None table_name = None table_info_dir = {} for child in root: if child.tag == 'Path': table_info_dir["Path"] = child.text elif child.tag == 'User_appp_id': table_info_dir["User_appp_id"] = child.text elif child.tag == 'User_app_secret': table_info_dir["User_app_secret"] = child.text elif child.tag == 'Url': table_info_dir["Url"] = child.text elif child.tag == 'Table_name': table_info_dir["Table_name"] = child.text # 解密密码 # password = decrypt_password(encrypted_password) return table_info_dir #获取access_token def accss_token(app_id,app_secret): conn = http.client.HTTPSConnection("open.feishu.cn") payload = json.dumps({ "app_id": app_id, "app_secret": app_secret }) headers = { 'Content-Type': 'application/json' } conn.request("POST", "/open-apis/auth/v3/app_access_token/internal", payload, headers) res = conn.getresponse() data = res.read() # 格式化成json data_dir = json.loads(data.decode("utf-8")) print(data_dir["app_access_token"]) return data_dir['app_access_token'] #获取app_token #最好是使用api调用,返回的结果与传入的多维表格名字进行判断,再获取对应的token,但是他的api调用的话返回的是null #这里我采用的是截取传入的连接中的url的值 def get_app_token(url): app_token = url.split('/')[-1].split('?')[0] if app_token: print(app_token) return app_token else: return None #获取table_id #这里通过url截取,也可以通过官方api获取 #如果传入的sheetID不是tableid而是子表名,可以通过api获取字典,获取相应的值,放到字典中通过key返回相应的tableID def get_table_id(url): parsed_url = urlparse(url) query_params = parse_qs(parsed_url.query) table_param = query_params.get('table') if table_param: table_id = table_param[0] print(table_id) return table_id else: print("table parameter not found in the URL.") return None def get_table_idByName(url,tableName,app_id,app_secret): conn = http.client.HTTPSConnection("open.feishu.cn") payload = '' headers = { 'Authorization': 'Bearer '+accss_token(app_id,app_secret) } conn.request("GET", "/open-apis/bitable/v1/apps/"+get_app_token(url)+"/tables?page_size=100", payload, headers) res = conn.getresponse() data = res.read() data_dir = json.loads(data) table_info_list = data_dir["data"]["items"] table_name = "内容类型" #print(data_dir["data"]["items"]) table_info_dir = {} for table_info in table_info_list: table_info_dir[table_info["name"]] = table_info["table_id"] print(table_info_dir[tableName]) return table_info_dir[tableName] #写入到本地文件 def write_data(url,tableName,app_id,app_secret): conn = http.client.HTTPSConnection("open.feishu.cn") payload = '' path = input("爬取数据保存的地址 ") headers = { 'Authorization': 'Bearer '+accss_token(app_id,app_secret) } conn.request("GET", "/open-apis/bitable/v1/apps/"+get_app_token(url)+"/tables/"+get_table_idByName(url,tableName,app_id,app_secret)+"/records?page_size=500", payload, headers) res = conn.getresponse() data = res.read() #print(data.decode("utf-8")) #print(data['data']) #格式化成json data_dir=json.loads(data.decode("utf-8")) #数据的list,里面每一条记录是以json存在 value_list=data_dir['data']["items"] #输出查看list #print(value_list) count=0 for kv in value_list: result = "" for key, value in kv['fields'].items(): result += f"{key}: {value}|^|" with open(path , 'a') as file: # 文件路径为完整的绝对路径 "/Users/username/Documents/output.txt" file.write(result) file.write("\n") print(result) count+=1 def run(): input_parameters() table_info_dir = get_parameters() print(table_info_dir) write_data(table_info_dir['Url'],table_info_dir['Table_name'],table_info_dir['User_appp_id'],table_info_dir['User_app_secret']) if __name__ == '__main__': url = 'https://hatuxjd72ne.feishu.cn/base/MvPfbLKcsaHSoUsGJdqcSi0Xnde?table=tbldgaotuABaSQUT&view=vewovUyg71' #path = 'D:/migu/data/data2.txt' tableName='page_id' #write_data(url,path,tableName) run() #write_data(t)
文章来源地址https://uudwc.com/A/nPLbA
文章来源:https://uudwc.com/A/nPLbA