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

2024-03-23 11:28
文章标签 python 数据 实现 excel 接口 图片 格式 pdf 企业 csv 读取数据 发送到 superset 微信群

本文主要是介绍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。

相关文章

大模型研发全揭秘：客服工单数据标注的完整攻略

大模型研发全揭秘：客服工单数据标注的完整攻略

在人工智能（AI）领域，数据标注是模型训练过程中至关重要的一步。无论你是新手还是有经验的从业者，掌握数据标注的技术细节和常见问题的解决方案都能为你的AI项目增添不少价值。在电信运营商的客服系统中，工单数据是客户问题和解决方案的重要记录。通过对这些工单数据进行有效标注，不仅能够帮助提升客服自动化系统的智能化水平，还能优化客户服务流程，提高客户满意度。本文将详细介绍如何在电信运营商客服工单的背景下进行
阅读更多...
基于MySQL Binlog的Elasticsearch数据同步实践

基于MySQL Binlog的Elasticsearch数据同步实践

一、为什么要做 随着马蜂窝的逐渐发展，我们的业务数据越来越多，单纯使用 MySQL 已经不能满足我们的数据查询需求，例如对于商品、订单等数据的多维度检索。 使用 Elasticsearch 存储业务数据可以很好的解决我们业务中的搜索需求。而数据进行异构存储后，随之而来的就是数据同步的问题。 二、现有方法及问题 对于数据同步，我们目前的解决方案是建立数据中间表。把需要检索的业务数据，统一放到一张M
阅读更多...
关于数据埋点，你需要了解这些基本知识

关于数据埋点，你需要了解这些基本知识

产品汪每天都在和数据打交道，你知道数据来自哪里吗？ 移动app端内的用户行为数据大多来自埋点，了解一些埋点知识，能和数据分析师、技术侃大山，参与到前期的数据采集，更重要是让最终的埋点数据能为我所用，否则可怜巴巴等上几个月是常有的事。   埋点类型 根据埋点方式，可以区分为： 手动埋点半自动埋点全自动埋点 秉承“任何事物都有两面性”的道理：自动程度高的，能解决通用统计，便于统一化管理，但个性化定
阅读更多...
python: 多模块(.py)中全局变量的导入

python: 多模块(.py)中全局变量的导入

文章目录 global关键字可变类型和不可变类型数据的内存地址单模块(单个py文件)的全局变量示例总结 多模块(多个py文件)的全局变量from x import x导入全局变量示例 import x导入全局变量示例 总结 global关键字 global 的作用范围是模块(.py)级别： 当你在一个模块（文件）中使用 global 声明变量时，这个变量只在该模块的全局命名空
阅读更多...
Hadoop企业开发案例调优场景

Hadoop企业开发案例调优场景

需求 （1）需求：从1G数据中，统计每个单词出现次数。服务器3台，每台配置4G内存，4核CPU，4线程。 （2）需求分析： 1G / 128m = 8个MapTask；1个ReduceTask；1个mrAppMaster 平均每个节点运行10个 / 3台 ≈ 3个任务（4    3    3） HDFS参数调优 （1）修改：hadoop-env.sh export HDFS_NAMENOD
阅读更多...
使用SecondaryNameNode恢复NameNode的数据

使用SecondaryNameNode恢复NameNode的数据

1）需求： NameNode进程挂了并且存储的数据也丢失了，如何恢复NameNode 此种方式恢复的数据可能存在小部分数据的丢失。 2）故障模拟 （1）kill -9 NameNode进程 [lytfly@hadoop102 current]$ kill -9 19886 （2）删除NameNode存储的数据（/opt/module/hadoop-3.1.4/data/tmp/dfs/na
阅读更多...
异构存储（冷热数据分离）

异构存储（冷热数据分离）

异构存储主要解决不同的数据，存储在不同类型的硬盘中，达到最佳性能的问题。 异构存储Shell操作 （1）查看当前有哪些存储策略可以用 [lytfly@hadoop102 hadoop-3.1.4]$ hdfs storagepolicies -listPolicies （2）为指定路径（数据存储目录）设置指定的存储策略 hdfs storagepolicies -setStoragePo
阅读更多...
Hadoop集群数据均衡之磁盘间数据均衡

Hadoop集群数据均衡之磁盘间数据均衡

生产环境，由于硬盘空间不足，往往需要增加一块硬盘。刚加载的硬盘没有数据时，可以执行磁盘数据均衡命令。（Hadoop3.x新特性） plan后面带的节点的名字必须是已经存在的，并且是需要均衡的节点。 如果节点不存在，会报如下错误： 如果节点只有一个硬盘的话，不会创建均衡计划： （1）生成均衡计划 hdfs diskbalancer -plan hadoop102 （2）执行均衡计划 hd
阅读更多...
hdu1043（八数码问题，广搜 + hash（实现状态压缩） ）

hdu1043（八数码问题，广搜 + hash（实现状态压缩） ）

利用康拓展开将一个排列映射成一个自然数，然后就变成了普通的广搜题。 #include<iostream>#include<algorithm>#include<string>#include<stack>#include<queue>#include<map>#include<stdio.h>#include<stdlib.h>#include<ctype.h>#inclu
阅读更多...
使用opencv优化图片（画面变清晰）

使用opencv优化图片（画面变清晰）

文章目录 需求影响照片清晰度的因素 实现降噪测试代码 锐化空间锐化Unsharp Masking频率域锐化对比测试 对比度增强常用算法对比测试 需求 对图像进行优化，使其看起来更清晰，同时保持尺寸不变，通常涉及到图像处理技术如锐化、降噪、对比度增强等 影响照片清晰度的因素 影响照片清晰度的因素有很多，主要可以从以下几个方面来分析 1. 拍摄设备 相机传感器：相机传
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2