python实现:从superset接口中读取数据,把数据以excel、pdf、图片、csv格式发送到企业微信群

本文主要是介绍python实现:从superset接口中读取数据,把数据以excel、pdf、图片、csv格式发送到企业微信群,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

python实现:从接口中读取数据,把数据以excel、pdf、图片、csv格式发送到企业微信群

接口文檔地址:https://developer.work.weixin.qq.com/document/path/99110

1.發送圖文

1.對接接口

import json
import time
import requests
import time_utilrbt_key = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=key"
headers = {'Content-Type': 'application/json; charset=UTF-8'}
text = '未拉OA提醒'##toke
def get_superset_token():url = "http://hadoop102:8787/api/v1/security/login"request_param = {"password": "passwd","provider": "db","refresh": True,"username": "user"}headers = {"Content-Type": "application/json"}response = requests.post(url, data=json.dumps(request_param), headers=headers)return response.json()["access_token"]def get_superset_data():access_token = get_superset_token()# target_date = time_util.convert_to_cst(time.time(), 'Asia/Shanghai').strftime('%Y-%m-%d')target_date = '2024-03-21'target_user = 'ALLEN.X'url = "http://hadoop102:8787/api/v1/chart/data"headers = {'Authorization': f'Bearer {access_token}',  # 添加正确格式的 Bearer Token'Content-Type': 'application/json'}request_param = {"datasource": {"id": 94,"type": "table"},"queries": [{"columns": ["user_name","日程類型","标题", "日程开始时间","日程结束时间","是否属于全天日程","OA状态"],"orderby": [["user_name",True]],"row_limit": 100,"order_desc": True,"result_format": "string","form_data": "string","where": "user_name = '" + target_user + "' and start_date = '" + target_date + "'"}]}response = requests.post(url, data=json.dumps(request_param), headers=headers)if response.status_code == 200:return response.json()['result'][0]['data']else:return 'error'def send_text(text):  # 添加 result 參數用於傳遞消息內容data = {"msgtype": "markdown","markdown": {"content": text + "\n<@allen>"}}requests.post(rbt_key, headers=headers, json=data)def send_image():  # 添加 result 參數用於傳遞消息內容# send_text(text)data = {"msgtype": "news","news": {"articles": [{"title": "==未拉OA提醒==","description": "提醒各位拉OA","url": "\\\\ip地址\\is-department\\小組\\myoa\\image\\schedule_table.png","picurl": "https://pic.616pic.com/ys_img/00/99/99/AdcfGUOyJR.jpg"}]}}requests.post(rbt_key, headers=headers, json=data)

數據時間處理


from datetime import datetime, timedelta
import pytz# 時間戳轉美國和中國時區
def convert_to_cst(timestamp, time_zone):dt = datetime.fromtimestamp(timestamp, pytz.utc)  # 将时间戳转换为datetime对象(UTC时间)pst = dt.astimezone(pytz.timezone(time_zone))  # 转换为美国PST时区时间return pst# return pst.strftime('%Y-%m-%d %H:%M:%S')  # 格式化为字符串def convert_to_pst(timestamp):dt = datetime.fromtimestamp(timestamp/1000)  # 将时间戳转换为datetime对象(UTC时间)pst = dt + timedelta(hours=7)  # 增加8小时,即美国PST时间return pst.strftime('%Y-%m-%d %H:%M:%S')  # 格式化为字符串

主程序

import data_util, time_util
import pandas as pd
import matplotlib.pyplot as pltif __name__ == '__main__':data = data_util.get_superset_data()if(len(data)>0):# 应用函数转换时间戳for entry in data:entry['日程开始时间'] = time_util.convert_to_pst(entry['日程开始时间'])entry['日程结束时间'] = time_util.convert_to_pst(entry['日程结束时间'])# 创建DataFramedf = pd.DataFrame(data)# 如果不为空,则绘制表格# 指定中文字体plt.rcParams['font.sans-serif'] = ['SimHei']plt.rcParams['axes.unicode_minus'] = False# 保存为图片,并指定dpi参数fig, ax = plt.subplots(figsize=(10, 6), dpi=500)ax.axis('tight')ax.axis('off')ax.table(cellText=df.values, colLabels=df.columns, loc='center', cellLoc='center')# 自定义保存路径save_path = 'image/schedule_table.png'print(save_path)plt.savefig(save_path)# 可选:显示保存成功的提示print(f"图片已保存至:{save_path}")else:print("error")data_util.send_image()

2.發送純圖片

主程序

import data_util, time_util
import pandas as pd
import matplotlib.pyplot as plt
import base64
import hashlib
if __name__ == '__main__':data = data_util.get_superset_data()if (len(data) > 0):# 应用函数转换时间戳for entry in data:entry['日程开始时间'] = time_util.convert_to_pst(entry['日程开始时间'])entry['日程结束时间'] = time_util.convert_to_pst(entry['日程结束时间'])# 创建DataFramedf = pd.DataFrame(data)# 如果不为空,则绘制表格# 指定中文字体plt.rcParams['font.sans-serif'] = ['SimHei']plt.rcParams['axes.unicode_minus'] = False# 保存为图片,并指定dpi参数fig, ax = plt.subplots(figsize=(10, 6), dpi=500)ax.axis('tight')ax.axis('off')ax.table(cellText=df.values, colLabels=df.columns, loc='center', cellLoc='center')# 自定义保存路径save_path = 'image/schedule_table.png'# print(save_path)plt.savefig(save_path)# 可选:显示保存成功的提示print(f"图片已保存至:{save_path}")# 1. 读取图片文件内容with open("image/schedule_table.png", "rb") as f:image_data = f.read()# 2. 对图片内容进行Base64编码base64_image = base64.b64encode(image_data).decode('utf-8')# 3. 计算图片内容的MD5值md5_hash = hashlib.md5(image_data).hexdigest()data_util.send_image(base64_image, md5_hash)else:print("error")

數據接口


def send_image(data, md5):  # 添加 result 參數用於傳遞消息內容# send_text(text)data = {"msgtype": "image","image": {"base64": data,"md5": md5}}requests.post(rbt_key, headers=headers, json=data)

3.發送excel,csv,pdf

主程序

import data_util, time_util
import pandas as pd
import matplotlib.pyplot as plt
from matplotlib.backends.backend_pdf import PdfPagesif __name__ == '__main__':data = data_util.get_superset_data()df = ''plt.rcParams['font.sans-serif'] = ['SimHei']plt.rcParams['axes.unicode_minus'] = Falseif (len(data) > 0):# 应用函数转换时间戳for entry in data:entry['日程开始时间'] = time_util.convert_to_pst(entry['日程开始时间'])entry['日程结束时间'] = time_util.convert_to_pst(entry['日程结束时间'])if isinstance(data, dict):if data:df = pd.DataFrame.from_records([data])else:print("Data is an empty dictionary")df = pd.DataFrame()elif isinstance(data, list):dfs = [pd.DataFrame.from_records([item]) for item in data if isinstance(item, dict)]df = pd.concat(dfs, ignore_index=True)else:print("Data is not a list or a dictionary")df = pd.DataFrame()# print(df)df.to_excel('media.xlsx', index=False)df.to_csv('media.csv', index=False)# 创建一个PDF文件with PdfPages('media.pdf') as pdf:# 将DataFrame绘制成表格fig, ax = plt.subplots(figsize=(8.27, 11.69), dpi=400)  # 设置dpi为300,提高图像质量ax.axis('tight')ax.axis('off')ax.table(cellText=df.values, colLabels=df.columns, loc='top', cellLoc='center')pdf.savefig(dpi=400)  # 将当前图形保存到PDF,并设置dpiplt.close()print("PDF文件已生成")data = data_util.send_file('media.pdf')if data != 'error':data_util.get_file(data)

調用接口

def send_file(file):# 设置请求参数url = "https://qyapi.weixin.qq.com/cgi-bin/webhook/upload_media?key=key&type=file"data = {'file': open(file, 'rb')}  # post jasonresponse = requests.post(url=url, files=data)  # post 请求上传文件if response.status_code == 200:return response.json()['media_id']else:return 'error'def get_file(media_id):  # 添加 result 參數用於傳遞消息內容data = {"msgtype": "file","file": {"media_id": media_id}}requests.post(rbt_key, headers=headers, json=data)

这篇关于python实现:从superset接口中读取数据,把数据以excel、pdf、图片、csv格式发送到企业微信群的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/838170

相关文章

pandas中位数填充空值的实现示例

《pandas中位数填充空值的实现示例》中位数填充是一种简单而有效的方法,用于填充数据集中缺失的值,本文就来介绍一下pandas中位数填充空值的实现,具有一定的参考价值,感兴趣的可以了解一下... 目录什么是中位数填充?为什么选择中位数填充?示例数据结果分析完整代码总结在数据分析和机器学习过程中,处理缺失数

Golang HashMap实现原理解析

《GolangHashMap实现原理解析》HashMap是一种基于哈希表实现的键值对存储结构,它通过哈希函数将键映射到数组的索引位置,支持高效的插入、查找和删除操作,:本文主要介绍GolangH... 目录HashMap是一种基于哈希表实现的键值对存储结构,它通过哈希函数将键映射到数组的索引位置,支持

Pandas使用AdaBoost进行分类的实现

《Pandas使用AdaBoost进行分类的实现》Pandas和AdaBoost分类算法,可以高效地进行数据预处理和分类任务,本文主要介绍了Pandas使用AdaBoost进行分类的实现,具有一定的参... 目录什么是 AdaBoost?使用 AdaBoost 的步骤安装必要的库步骤一:数据准备步骤二:模型

Pandas统计每行数据中的空值的方法示例

《Pandas统计每行数据中的空值的方法示例》处理缺失数据(NaN值)是一个非常常见的问题,本文主要介绍了Pandas统计每行数据中的空值的方法示例,具有一定的参考价值,感兴趣的可以了解一下... 目录什么是空值?为什么要统计空值?准备工作创建示例数据统计每行空值数量进一步分析www.chinasem.cn处

使用Pandas进行均值填充的实现

《使用Pandas进行均值填充的实现》缺失数据(NaN值)是一个常见的问题,我们可以通过多种方法来处理缺失数据,其中一种常用的方法是均值填充,本文主要介绍了使用Pandas进行均值填充的实现,感兴趣的... 目录什么是均值填充?为什么选择均值填充?均值填充的步骤实际代码示例总结在数据分析和处理过程中,缺失数

Java对象转换的实现方式汇总

《Java对象转换的实现方式汇总》:本文主要介绍Java对象转换的多种实现方式,本文通过实例代码给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友参考下吧... 目录Java对象转换的多种实现方式1. 手动映射(Manual Mapping)2. Builder模式3. 工具类辅助映

如何使用 Python 读取 Excel 数据

《如何使用Python读取Excel数据》:本文主要介绍使用Python读取Excel数据的详细教程,通过pandas和openpyxl,你可以轻松读取Excel文件,并进行各种数据处理操... 目录使用 python 读取 Excel 数据的详细教程1. 安装必要的依赖2. 读取 Excel 文件3. 读

Go语言开发实现查询IP信息的MCP服务器

《Go语言开发实现查询IP信息的MCP服务器》随着MCP的快速普及和广泛应用,MCP服务器也层出不穷,本文将详细介绍如何在Go语言中使用go-mcp库来开发一个查询IP信息的MCP... 目录前言mcp-ip-geo 服务器目录结构说明查询 IP 信息功能实现工具实现工具管理查询单个 IP 信息工具的实现服

Python的time模块一些常用功能(各种与时间相关的函数)

《Python的time模块一些常用功能(各种与时间相关的函数)》Python的time模块提供了各种与时间相关的函数,包括获取当前时间、处理时间间隔、执行时间测量等,:本文主要介绍Python的... 目录1. 获取当前时间2. 时间格式化3. 延时执行4. 时间戳运算5. 计算代码执行时间6. 转换为指

利用Python调试串口的示例代码

《利用Python调试串口的示例代码》在嵌入式开发、物联网设备调试过程中,串口通信是最基础的调试手段本文将带你用Python+ttkbootstrap打造一款高颜值、多功能的串口调试助手,需要的可以了... 目录概述:为什么需要专业的串口调试工具项目架构设计1.1 技术栈选型1.2 关键类说明1.3 线程模