飞书应用机器人文件上传

背景:

  接上一篇 flask_apscheduler实现定时推送飞书消息,当检查出的异常结果比较多的时候,群里会有很多推送消息,一条条检查工作量会比较大,且容易出现遗漏。
  现在需要将定时任务执行的结果记录到文件,最好是飞书的云文件中,通过分享云文档的方式分析给响应的人员。

功能:

  飞书群机器人没有文件上传的的功能,满足这个功能需要使用飞书应用机器人。创建飞书应用后,需要完成机器人配置,以及上传文件的权限申请。
在这里插入图片描述在这里插入图片描述
待使用的接口功能:

  1. 实现文件上传,参考文档。通过该接口实现将定时任务执行结果保存上传至飞书云文档。
    在这里插入图片描述2. 更新云文档权限设置,参考文档。修改上传至云文档的文件权限,使组织内成员可阅读。
    在这里插入图片描述

实现:

  • 实现效果:
    在这里插入图片描述

  • 功能代码:

    # -*- 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} 发送消息失败:{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} 文件路径没有指定特定文件')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__():# 创建文件夹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('上传结果至飞书失败')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)
    

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.rhkb.cn/news/159089.html

如若内容造成侵权/违法违规/事实不符,请联系长河编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

使用EasyDarwin+ffmpeg+EasyPlayerPro完成rtsp的推流操作和拉流操作

本文分享在做视频类测试过程中所用到的工具EasyDarwinffmpegEasyPlayerPro 首先说一下EasyDarwin,简单来讲,它就是个推流和拉流及系统消耗的监测软件,具体使用方法我会写在下方。 EasyDarwin 1、解压下载好的EasyDarwin压缩包,并找到EasyD…

后端:推荐 2 个 .NET 操作的 Redis 客户端类库

目录 Redis特点 Redis场景 1. StackExchange.Redis 2. FreeRedis 🚀 快速入门 🎣 Master-Slave (读写分离) 💻 Pipeline (管道)示例 🌌 Redis Cluster (集群) Redis ,是一个高性能(NOSQL)的key-value数据库,Re…

【TensorFlow2 之014】在 TF 2.0 中实现 LeNet-5

一、说明 在这篇文章中,我们将展示如何在 TensorFlow 中实现像 \(LeNet-5\) 这样的基础卷积神经网络。LeNet-5 架构由 Yann LeCun 于 1998 年发明,是第一个卷积神经网络。 数据黑客变种rs 深度学习 机器学习 TensorFlow 2020 年 2 月 29 日 | 0 …

GB28181平台简介

产品简介 LiveMedia视频中间件是支持部署到本地服务器或者云服务器的纯软件服务,也提供服务器、GPU一体机全包服务,提供视频设备管理、无插件、跨平台的实时视频、历史回放、语音对讲、设备控制等基础功能,支持视频协议有海康、大华私有协议…

竞赛 深度学习LSTM新冠数据预测

文章目录 0 前言1 课题简介2 预测算法2.1 Logistic回归模型2.2 基于动力学SEIR模型改进的SEITR模型2.3 LSTM神经网络模型 3 预测效果3.1 Logistic回归模型3.2 SEITR模型3.3 LSTM神经网络模型 4 结论5 最后 0 前言 🔥 优质竞赛项目系列,今天要分享的是 …

Idea创建springboot工程的时候,发现pom文件没有带<parent>标签

今天创建springboot工程,加载maven的时候报错: 这个问题以前遇到过,这是因为 mysql-connector-j 没有带版本号的原因,但是springboot的依赖的版本号不是都统一交给spring-boot-starter-parent管理了吗,为什么还会报错&…

华为云云耀云服务器L实例评测|华为云耀云服务器L实例评测用例(五)

六、华为云耀云服务器L实例评测用例: “兵马未动,粮草先行”,随着企业业务的快速发展,服务器在数字化建设体系至关重要,为了保证服务器的稳定性、可靠性,需要对服务器进行评测,以确保服务器能够…

kafka详解(三)

2.2 Kafka命令行操作 2.2.1 主题命令行操作 1)查看操作主题命令参数 [aahadoop102 kafka]$ bin/kafka-topics.sh2)查看当前服务器中的所有topic (配置了环境变量不需要写bin/) [aahadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop10…

Linux gcc和make学习

文章目录 GCCgcc的安装gcc的工作流程 makefilemakefile的规则工作原理自动生成makefile的变量自定义变量预定义变量自动变量 模式匹配函数wildcard函数patsubst函数 伪声明 GCC gcc全程是(GNU compiler collection CNU编译器套件),是由GNU开发…

想要精通算法和SQL的成长之路 - 分割数组的最大值

想要精通算法和SQL的成长之路 - 分割数组的最大值 前言一. 分割数组的最大值1.1 二分法 前言 想要精通算法和SQL的成长之路 - 系列导航 一. 分割数组的最大值 原题链接 首先面对这个题目,我们可以捕获几个关键词: 非负整数。非空连续子数组。 那么我…

线性排序:如何根据年龄给100万用户数据排序?

文章来源于极客时间前google工程师−王争专栏。 桶排序、计数排序、基数排序时间复杂度是O(n),所以这类排序算法叫作线性排序。 线性的原因:三个算法是非基于比较的排序算法,都不涉及元素之间的比较操作。 三种排序对排序的数据要求苛刻&am…

CCF CSP认证 历年题目自练Day30

题目一 试题编号: 202203-1 试题名称: 未初始化警告 时间限制: 1.0s 内存限制: 512.0MB 问题描述: 题目背景 一个未经初始化的变量,里面存储的值可能是任意的。因此直接使用未初始化的变量,比…

太强了,真的太强了!

国庆之后gpt4上线了很多强大的功能,有超级强大的数据分析和挖掘的功能,有可以比肩AI绘图神器Midjourney的绘图功能(前面写了一篇泰酷辣!目前最强的AI绘画神器!文生图模型 DALLE 3来啦!)&#xf…

Python正则表达式

正则表达式 当处理文本数据时,正则表达式是一种强大的工具,它允许我们根据特定的模式来匹配、搜索和处理字符串。 正则表达式由一系列字符和特殊字符组成,用于描述文本模式。这些模式可以包含普通字符(如字母、数字和标点符号&a…

【TensorFlow2 之012】TF2.0 中的 TF 迁移学习

#012 TensorFlow 2.0 中的 TF 迁移学习 一、说明 在这篇文章中,我们将展示如何在不从头开始构建计算机视觉模型的情况下构建它。迁移学习背后的想法是,在大型数据集上训练的神经网络可以将其知识应用于以前从未见过的数据集。也就是说,为什么…

蓝桥杯 第 1 场算法双周赛 第1题 三带一 c++ map 巧解 加注释

题目 三带一【算法赛】https://www.lanqiao.cn/problems/5127/learning/?contest_id144 问题描述 小蓝和小桥玩斗地主,小蓝只剩四张牌了,他想知道是否是“三带一”牌型。 所谓“三带一”牌型,即四张手牌中,有三张牌一样&#…

CSS学习基础知识

CSS学习笔记 <!DOCTYPE html> <html lang"en"> <head><meta charset"UTF-8"><meta http-equiv"X-UA-Compatible" content"IEedge"><meta name"viewport" content"widthdevice-width,…

独立式三相无源逆变电源设计

摘要 面对全球日趋严重的能源危机问题&#xff0c;可再生能源的开发和利用得到了人们的高度重视。其中辐射到地球太阳能资源是十分富饶的&#xff0c;绿色清洁的太阳能不会危害我们的生存环境&#xff0c;因而受到了人们的广泛利用。光伏发电作为可再生能源被广泛的应用&#x…

RabbitMq启用TLS

Windows环境 查看配置文件的位置 选择使用的节点 查看当前节点配置文件的配置 配置TLS 将证书放到同配置相同目录中 编辑配置文件添加TLS相关配置 [{ssl, [{versions, [tlsv1.2]}]},{rabbit, [{ssl_listeners, [5671]},{ssl_options, [{cacertfile,"C:/Users/17126…

如何定制化跑腿小程序源码

跑腿小程序源码为您提供了一个强大的起点&#xff0c;但要创建一个成功的本地服务平台&#xff0c;您通常需要对源码进行定制化。这篇文章将介绍如何定制化跑腿小程序源码&#xff0c;包括添加新功能、修改界面和优化用户体验。 选择合适的跑腿小程序源码 首先&#xff0c;您…