背景：

??接上一篇 flask_apscheduler實現(xiàn)定時推送飛書消息，當檢查出的異常結(jié)果比較多的時候，群里會有很多推送消息，一條條檢查工作量會比較大，且容易出現(xiàn)遺漏。
??現(xiàn)在需要將定時任務執(zhí)行的結(jié)果記錄到文件，最好是飛書的云文件中，通過分享云文檔的方式分析給響應的人員。

功能：

??飛書群機器人沒有文件上傳的的功能，滿足這個功能需要使用飛書應用機器人。創(chuàng)建飛書應用后，需要完成機器人配置，以及上傳文件的權限申請。

待使用的接口功能：

1. 實現(xiàn)文件上傳，參考文檔。通過該接口實現(xiàn)將定時任務執(zhí)行結(jié)果保存上傳至飛書云文檔。
   2. 更新云文檔權限設置，參考文檔。修改上傳至云文檔的文件權限，使組織內(nèi)成員可閱讀。
   

實現(xiàn)：

• 實現(xiàn)效果：
  

• 功能代碼：

  # -*- coding:UTF-8 -*-"""@ProjectName  : HotelGo2DelonixPmx@FileName     : webhook@Description  : 飛書消息推送@Time         : 2023/9/17 13:36@Author       : Qredsun"""import os
  import requestsclass FeishuApplication():TENANT_ACCESS_TOKEN_URL = 'https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal'GET_USER_ID_URL = 'https://open.feishu.cn/open-apis/contact/v3/users/batch_get_id'IM_MESSAGES_URL = 'https://open.feishu.cn/open-apis/im/v1/messages'FILES_UPLOAD_URL = 'https://open.feishu.cn/open-apis/drive/v1/files/upload_all'DRIVE_FILES_URL = 'https://open.feishu.cn/open-apis/drive/v1/files'FILE_PERMISSION = 'https://open.feishu.cn/open-apis/drive/v2/permissions/token/public'CREATE_FOLDER = 'https://open.feishu.cn/open-apis/drive/v1/files/create_folder'def __init__(self, app_id, app_secret):self.app_id = app_idself.app_secret = app_secretself.get_tenant_access_token()self._url_prefix = Noneself._file_url_prefix = Nonedef get_tenant_access_token(self):url = self.TENANT_ACCESS_TOKEN_URLdata = {"app_id"    : self.app_id,"app_secret": self.app_secret}response = requests.post(url, json=data)response.raise_for_status()res_data = response.json()if res_data:self._tenant_access_token = res_data['tenant_access_token']self.headers = {'Content-Type' : 'application/json','Authorization': f'Bearer {self._tenant_access_token}'}logger.debug(f'自建應用更新token成功')return self._tenant_access_tokenelse:logger.error(f'自建應用獲取token失敗：{response.text}')return Falsedef get_user_open_id(self, user_info):# 單用戶id查詢url = self.GET_USER_ID_URLparams = {"user_id_type": "open_id"}payload = {"emails" : [],"mobiles": []}if '@' in user_info:payload["emails"].append(user_info)response = requests.post(url, headers=self.headers, params=params, json=payload)elif user_info.isalnum():payload["mobiles"].append(user_info)response = requests.post(url, headers=self.headers, params=params, json=payload)response.raise_for_status()res_data = response.json()if res_data:self.open_id = res_data['data']["user_list"][0]["user_id"]return self.open_idelse:logger.error(f'獲取用戶{user_info} open_id 失敗：{response.text}')return Nonedef send_single_message(self, msg = "single chat msg", open_id = ''):if not open_id:logger.error('缺少對話用戶 open_id ')returnurl = self.IM_MESSAGES_URLparams = { "receive_id_type": "open_id" }msgContent = {"text": msg}req = {"receive_id": open_id,  # chat id"msg_type"  : "text","content"   : json.dumps(msgContent)}payload = json.dumps(req)response = requests.request("POST", url, params=params, headers=self.headers, data=payload)response.raise_for_status()res_data = response.json()if res_data:self.open_id = res_data['data']["chat_id"]return Trueelse:logger.error(f'給用戶 {self.open_id} 發(fā)送消息失敗：{response.text}')return Falsedef remove_file_or_folder(self, file_token, file_type='file'):url = self.DRIVE_FILES_URLurl += f'/{file_token}'payload = ''params = {'type':file_type}response = requests.request("DELETE", url, headers=self.headers, params=params, data=payload)response.raise_for_status()result = response.json()if result.get("code") and result.get("code") != 0:logger.error(f'移除文件失敗：{response.text}')return Falseelse:logger.debug(f'移除文件成功：{response.text}')return Truedef update_permissions(self, folder_token = '', file_type='file'):url = self.FILE_PERMISSIONurl  = url.replace('token', folder_token)params = {'type': file_type}payload = json.dumps({"comment_entity"            : "anyone_can_view","copy_entity"               : "anyone_can_view","external_access_entity"    : "open","link_share_entity"         : "tenant_editable","manage_collaborator_entity": "collaborator_can_view","security_entity"           : "anyone_can_view","share_entity"              : "anyone"})response = requests.request("PATCH", url, headers=self.headers, data=payload, params=params)response.raise_for_status()result = response.json()if result.get("code") and result.get("code") != 0:logger.error(f'更新文件權限失敗：{response.text}')return Falseelse:logger.debug(f'更新文件權限成功：{response.text}')return True"""上傳文件"""def upload_file(self, file_path = "../data/result/23_09_25_訂房檢查任務.xlsx",parent_node = 'ErVlfbxP8lqZ1sdMIWkc11TQn8g'):if not os.path.isfile(file_path):logger.error(f'{file_path} 文件路徑?jīng)]有指定特定文件')returnurl = self.FILES_UPLOAD_URLfile_size = os.path.getsize(file_path)file_name = os.path.basename(file_path)payload = {'file_name'  : file_name,'parent_type': 'explorer','parent_node': parent_node,'size'       : f'{file_size}'}files = [('file', (file_name, open(os.path.abspath(file_path), 'rb'),'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet'))]headers = {'Authorization': self.headers['Authorization']}resp = requests.request("POST", url, headers=headers, data=payload, files=files)resp.raise_for_status()result = resp.json()if result.get("code") and result.get("code") != 0:logger.error(f'文件上傳失敗：{resp.text}')return Falseelse:file_token = result['data']['file_token']logger.debug(f'文件上傳成功：{resp.text}')return file_token"""獲取文件夾下的清單"""def expoler(self, direction = 'DESC', order_by = 'EditedTime'):url = self.DRIVE_FILES_URLparams = {'direction': direction,'order_by' : order_by}resp = requests.request("GET", url, headers=self.headers, params=params)resp.raise_for_status()result = resp.json()if result.get("code") and result.get("code") != 0:logger.error(f'獲取云空間列表失敗：{resp.text}')return Noneelse:self.files = result['data']['files']self.update_url_prefix()logger.debug(f'獲取云空間列表成功: {self.files}')return self.files"""新建文件夾"""def create_folder(self, folder_name = "", folder_token = ""):url = self.CREATE_FOLDERpayload = {"folder_token": folder_token,"name"        : folder_name}resp = requests.request("POST", url, headers=self.headers, json=payload)resp.raise_for_status()result = resp.json()if result.get("code") and result.get("code") != 0:logger.error(f'新建文件夾失敗：{resp.text}')return Noneelse:self.folder_token = result['data']['token']logger.debug(f'新建文件夾成功: {self.folder_token}')folder_url = result['data']['url']start_index = folder_url.find('//') + 2r_index = folder_url.find('/', start_index) + 1self._url_prefix = folder_url[:r_index]logger.debug(f'更新應用地址前綴:{self._url_prefix}')self._file_url_prefix = self._url_prefix + 'file/'logger.debug(f'更新云文件前綴:{self._file_url_prefix}')return self.folder_tokendef update_url_prefix(self):for obj in self.files:if obj['type'] == 'folder':obj_url = obj['url']start_index = obj_url.find('//') + 2r_index = obj_url.find('/', start_index) + 1self._url_prefix = obj_url[:r_index]logger.debug(f'更新應用地址前綴:{self._url_prefix}')self._file_url_prefix = self._url_prefix + 'file/'logger.debug(f'更新云文件前綴:{self._file_url_prefix}')breakreturn self._url_prefixdef upload_schedule_result(upload_file, app_id, app_secret):robot = FeishuApplication(app_id, app_secret)default_folder = 'schedule_demo'file_path = upload_fileparent_node = ''robot.expoler()if not robot.files.__len__():# 創(chuàng)建文件夾result = robot.create_folder(default_folder)if result:parent_node = resultelse:for file in robot.files:if default_folder == file['name']:parent_node = file['parent_token']parent_node = file['token']break# 移除文件robot.remove_file_or_folder('O3MgbgYKgo7NgtxUNc4cqkQZnWe')upload_file_token = robot.upload_file(file_path=file_path, parent_node=parent_node)if upload_file_token:result = robot.update_permissions(upload_file_token)if result:file_url = f'{robot._file_url_prefix}{upload_file_token}'logger.debug(f'待分享的文件url: {file_url}')else:file_url = ''logger.debug('上傳結(jié)果至飛書失敗')return file_urlif __name__ == '__main__':upload_file = "../data/result/23_09_24_訂房檢查任務.xlsx"app_id = 'XXXX'app_secret = 'XXX'upload_schedule_result(upload_file, app_id, app_secret)