飞书应用机器人文件上传

背景：

  接上一篇 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.mzph.cn/news/105810.shtml

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

相关文章

C# GFPGAN 图像修复

C# GFPGAN 图像修复

效果 项目 代码 using Microsoft.ML.OnnxRuntime; using Microsoft.ML.OnnxRuntime.Tensors; using OpenCvSharp; using System; using System.Collections.Generic; using System.Drawing; using System.Drawing.Imaging; using System.Windows.Forms;namespace 图像修复 {pu…
阅读更多...
二维数组的行指针与列指针

二维数组的行指针与列指针

二维数组的行指针与列指针 笔记来源：懒猫老师-C语言-用指针访问二维数组（指针与二维数组） 声明：本文笔记来自bili懒猫老师，仅供学习参考 回顾一维数组 int a[3];//其中a或a[0]是数组首地址 //a1指向第二个元素的地址…
阅读更多...
PyTorch 深度学习之卷积神经网络(高级篇)Advanced CNN(十)

PyTorch 深度学习之卷积神经网络(高级篇)Advanced CNN(十)

0. Revision 前面讲的比较简单的是 串行网络结构 1. GoogLeNet 1.1 Inception module w h 要一致 what is 11 convolution? 信息融合-eg.高中各门学科成绩比较(总分) 最主要工作:改变通道数量 why is 11 convolution? 减少10倍 1.2 implementation of inception module 拼…
阅读更多...
AOP面向切面编程

AOP面向切面编程

AOP简介 AOP(Aspect oriented Programming)面向切面编程，就是面向特定的方法编程，将方法比作一个一个的切面，可以向指定的方法执行前/后执行自己的逻辑。如统一获取方法的时间。 应用场景： 记录操作日志、权限控制、事务管理 优…
阅读更多...
Node.js初体验

Node.js初体验

Node.js简介 node.js的运行环境 1.V8引擎对js代码进行解析与执行 2.内置API：fs、path、http...等，提供了一些能力，能够使得js调用这些API去做一些后端的事情 流程：我们在node.js的运行环境中编写待执行的JavaScript代码&#…
阅读更多...
部署k8s dashboard（这里使用Kubepi）

部署k8s dashboard（这里使用Kubepi）

9. 部署k8s dashboard（这里使用Kubepi） Kubepi是一个简单高效的k8s集群图形化管理工具，方便日常管理K8S集群，高效快速的查询日志定位问题的工具 部署KubePI（随便在哪个节点部署，我这里在主节点部署&#…
阅读更多...
黑马JVM总结（三十）

黑马JVM总结（三十）

（1）类加载-连接-解析 new对象，会触发对象的解析和初始化 通过类加载器加载类C，这里只会进行类C的加载并不会导致类C的解析以及初始化，这样加载类C的时候类D也不会解析跟初始化 通过jps找到进程id： 打开…
阅读更多...
Kafka SASL认证授权（四）认证源码解析

Kafka SASL认证授权（四）认证源码解析

Kafka SASL认证授权（四）认证源码解析。 官网地址：https://kafka.apache.org/ 一、认证流程 在了解kafka网络模型的基础上，了解它的认证流程： ApiVersionsRequest->SaslHandshakeRequest->a series of SASL client and server tokens corresponding to the mechani…
阅读更多...
IDEA的使用（二）快捷键 （IntelliJ IDEA 2022.1.3版本）

IDEA的使用（二）快捷键 （IntelliJ IDEA 2022.1.3版本）

1. IDEA中的常用快捷键 1.1 通用型快捷键 1.2 提高编写速度 ctrl shift ↑或↓ 只能在方法里面移动代码。 alt shift ↑或↓ 可以向方法外移动代码。 设置过自动导包，所以不用批量导包啦。 1.3 类结构、查找和查看源码 1.4 查找、替换和关闭 1.5 调整格式 1.6 De…
阅读更多...
Elasticsearch 分片内部原理—使文本可被搜索、动态更新索引

Elasticsearch 分片内部原理—使文本可被搜索、动态更新索引

目录 一、使文本可被搜索 不变性 二、动态更新索引 删除和更新 一、使文本可被搜索 必须解决的第一个挑战是如何使文本可被搜索。 传统的数据库每个字段存储单个值，但这对全文检索并不够。文本字段中的每个单词需要被搜索，对数据库意味着需要单个字…
阅读更多...
【CNN-GRU预测】基于卷积神经网络-门控循环单元的单维时间序列预测研究（Matlab代码实现）

【CNN-GRU预测】基于卷积神经网络-门控循环单元的单维时间序列预测研究（Matlab代码实现）

💥💥💞💞欢迎来到本博客❤️❤️💥💥 🏆博主优势：🌞🌞🌞博客内容尽量做到思维缜密，逻辑清晰，为了方便读者。 ⛳️座右铭&a…
阅读更多...
__main__文件学习测试如下

__main__文件学习测试如下

参考博客 https://www.jb51.net/article/251567.htm test01.py # https://www.jb51.net/article/251567.htmdef step():print(__name__)print("step1 买菜""step2 洗菜""step3 切菜""step4 炒菜")if __name__ __main__:print(准备制…
阅读更多...
【Android知识笔记】图片专题（BitmapDrawable）

【Android知识笔记】图片专题（BitmapDrawable）

如何计算一张图片的占用内存大小？ 注意是占用内存，不是文件大小可以运行时获取重要的是能直接掌握计算方法基础知识 Android 屏幕像素密度分类： （其实还有一种 ldpi = 120，不过这个已经绝种了，所以最低的只需关心mdpi即可） 上表中的比例为：m : h : xh : xxh: xxxh = …
阅读更多...
机器学习-无监督算法之降维

机器学习-无监督算法之降维

降维：将训练数据中的样本从高维空间转换到低维空间，降维是对原始数据线性变换实现的。为什么要降维？高维计算难，泛化能力差，防止维数灾难优点：减少冗余特征，方便数据可视化，减少内存…
阅读更多...
使用VirtualBox安装Ubuntu系统【保姆级】

使用VirtualBox安装Ubuntu系统【保姆级】

本文详细介绍使用VirtualBox安装Ubuntu系统，并实现虚拟机Ubuntu和Windows系统之间共享粘贴板和共享文件夹，主要包含如下5部分内容： 下载VirtualBox软件安装包下载Ubuntu系统镜像安装VirtualBoxVirtualBox新建Ubuntu虚拟机并安装Ubuntu系统Vi…
阅读更多...
巴以冲突中暴露的摄像头正对安全构成威胁

巴以冲突中暴露的摄像头正对安全构成威胁

巴以冲突爆发后，许多配置不当的安全摄像头正暴露给黑客活动分子，使其周遭人员面临巨大安全风险。 Cyber​​news 研究人员发现，在以色列至少有165 个暴露的联网 RTSP 摄像头，在巴勒斯坦有 29 个暴露的 RTSP 摄像头。在巴勒斯坦&am…
阅读更多...
Python 机器学习入门之逻辑回归

Python 机器学习入门之逻辑回归

系列文章目录 第一章 Python 机器学习入门之线性回归 第一章 Python 机器学习入门之梯度下降法 第一章 Python 机器学习入门之牛顿法 第二章 Python 机器学习入门之逻辑回归 逻辑回归 系列文章目录前言一、逻辑回归简介二、逻辑回归推导1、问题2、Sigmoid函数3、目标函数3.1 让…
阅读更多...
Prometheus：优秀和强大的监控报警工具

Prometheus：优秀和强大的监控报警工具

文章目录 概述Prometheus的底层技术和原理数据模型数据采集数据存储查询语言数据可视化 Prometheus的部署Prometheus的使用配置数据采集目标查询监控数据设置警报规则 查看数据可视化总结 概述 Prometheus是一款开源的监控和警报工具，用于收集和存储系统和应用程序…
阅读更多...
mysql主从工作原理、搭建

mysql主从工作原理、搭建

目录 一、mysql为什么要用主从架构？ 二、mysql数据库主从复制原理是什么？ 详细的主从复制过程如下图： 主从复制过程概述： 三、mysql主从如何搭建？ 本次安装的数据库版本为mysql5.7 1、准备两台服务器&#xff08…
阅读更多...
tomcat多实例部署jenkins

tomcat多实例部署jenkins

tomcat多实例部署jenkins 文章目录 tomcat多实例部署jenkins1.简介：2.优缺点：3.工作原理：4.工作流程：5.tomcat多实例部署jenkins流程5.1.环境说明5.2.部署前准备工作5.3.多实例部署tomcat5.4.部署jenkins5.5.创建一个jenkins项目5…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023