自己建个网站,企业官网与公开财报,wordpress菜单对齐修改,软件工程师岗位职责背景#xff1a; 接上一篇 flask_apscheduler实现定时推送飞书消息#xff0c;当检查出的异常结果比较多的时候#xff0c;群里会有很多推送消息#xff0c;一条条检查工作量会比较大#xff0c;且容易出现遗漏。 现在需要将定时任务执行的结果记录到文件#xff0c;…背景 接上一篇 flask_apscheduler实现定时推送飞书消息当检查出的异常结果比较多的时候群里会有很多推送消息一条条检查工作量会比较大且容易出现遗漏。 现在需要将定时任务执行的结果记录到文件最好是飞书的云文件中通过分享云文档的方式分析给响应的人员。
功能 飞书群机器人没有文件上传的的功能满足这个功能需要使用飞书应用机器人。创建飞书应用后需要完成机器人配置以及上传文件的权限申请。 待使用的接口功能
实现文件上传参考文档。通过该接口实现将定时任务执行结果保存上传至飞书云文档。 2. 更新云文档权限设置参考文档。修改上传至云文档的文件权限使组织内成员可阅读。
实现 实现效果 功能代码 # -*- coding:UTF-8 -*-ProjectName : HotelGo2DelonixPmxFileName : webhookDescription : 飞书消息推送Time : 2023/9/17 13:36Author : Qredsunimport os
import requestsclass FeishuApplication():TENANT_ACCESS_TOKEN_URL https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internalGET_USER_ID_URL https://open.feishu.cn/open-apis/contact/v3/users/batch_get_idIM_MESSAGES_URL https://open.feishu.cn/open-apis/im/v1/messagesFILES_UPLOAD_URL https://open.feishu.cn/open-apis/drive/v1/files/upload_allDRIVE_FILES_URL https://open.feishu.cn/open-apis/drive/v1/filesFILE_PERMISSION https://open.feishu.cn/open-apis/drive/v2/permissions/token/publicCREATE_FOLDER https://open.feishu.cn/open-apis/drive/v1/files/create_folderdef __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, jsondata)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: fBearer {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, headersself.headers, paramsparams, jsonpayload)elif user_info.isalnum():payload[mobiles].append(user_info)response requests.post(url, headersself.headers, paramsparams, jsonpayload)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 idmsg_type : text,content : json.dumps(msgContent)}payload json.dumps(req)response requests.request(POST, url, paramsparams, headersself.headers, datapayload)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} 发送消息失败{response.text})return Falsedef remove_file_or_folder(self, file_token, file_typefile):url self.DRIVE_FILES_URLurl f/{file_token}payload params {type:file_type}response requests.request(DELETE, url, headersself.headers, paramsparams, datapayload)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_typefile):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, headersself.headers, datapayload, paramsparams)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} 文件路径没有指定特定文件)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, headersheaders, datapayload, filesfiles)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, headersself.headers, paramsparams)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, headersself.headers, jsonpayload)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_demofile_path upload_fileparent_node robot.expoler()if not robot.files.__len__():# 创建文件夹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_pathfile_path, parent_nodeparent_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(上传结果至飞书失败)return file_urlif __name__ __main__:upload_file ../data/result/23_09_24_订房检查任务.xlsxapp_id XXXXapp_secret XXXupload_schedule_result(upload_file, app_id, app_secret)
