|
** python把数据转为图片 / python推送图片到钉钉群 **
需求：通过python访问mysql数据库,统计业务相关数据。把统计的结果数据生成表格形式的图片并发送到钉钉群里。
一：CentOS安装wkhtmltoimage 、wkhtmltopdf
wkhtmltopdf 和 wkhtmltoimage 是一个开源的命令行工具，可以将 HTML 转换为 pdf 文档和图片。
1.1：安装字体等依赖
yum install -y xorg-x11-fonts-75dpi
yum install -y xorg-x11-fonts-Type1
yum install -y fontconfig
yum install -y libX11
yum install -y libXext
yum install -y libXrender
1.2：下载安装文件
官网下载地址：https:///downloads.html
查看自己CentOS版本信息：
[root@centos src]# cat /etc/redhat-release
CentOS Linux release 7.7.1908 (Core)
选择自己对应的系统下载文件即可：
wget https://github.com/wkhtmltopdf/packaging/releases/download/0.12.6-1/wkhtmltox-0.12.6-1.centos7.x86_64.rpm
下载完成的文件：wkhtmltox-0.12.6-1.centos7.x86_64.rpm
1.2：安装
[root@centos src]# rpm -Uvh wkhtmltox-0.12.6-1.centos7.x86_64.rpm
error: Failed dependencies:
fontconfig is needed by wkhtmltox-1:0.12.6-1.centos7.x86_64
libX11 is needed by wkhtmltox-1:0.12.6-1.centos7.x86_64
libXext is needed by wkhtmltox-1:0.12.6-1.centos7.x86_64
libXrender is needed by wkhtmltox-1:0.12.6-1.centos7.x86_64
xorg-x11-fonts-75dpi is needed by wkhtmltox-1:0.12.6-1.centos7.x86_64
xorg-x11-fonts-Type1 is needed by wkhtmltox-1:0.12.6-1.centos7.x86_64
# 安装对应的依赖包后重新安装
重新安装
[root@centos src]# rpm -Uvh wkhtmltox-0.12.6-1.centos7.x86_64.rpm
Preparing... ################################# [100%]
Updating / installing...
1:wkhtmltox-1:0.12.6-1.centos7 ################################# [100%]
[root@centos src]# whereis wkhtmltopdf
wkhtmltopdf: /usr/local/bin/wkhtmltopdf
[root@centos src]# whereis wkhtmltoimage
wkhtmltoimage: /usr/local/bin/wkhtmltoimage
[root@centos src]# /usr/local/bin/wkhtmltoimage --version
wkhtmltoimage 0.12.6 (with patched qt)
[root@centos src]# /usr/local/bin/wkhtmltopdf --version
wkhtmltopdf 0.12.6 (with patched qt)
1.3：问题汇总
1：转换过程中文不显示
查看 Linux 是否有中文字体，无任何输出说明没有中文字体。
fc-list :lang=zh
安装中文字体 simsun.ttc
字体下载 http://www./fontlist/fontlist_1_1.html
将下载后的字体放到文件夹 /usr/share/fonts 下
二：使用python统计mysql数据并返回结果
关键py代码：
class DatabaseMonitor(object):
"""数据库操作对象"""
def __init__(self, server, port, username, password, databaseName):
self.server = server
self.port = port
self.username = username
self.password = password
self.databaseName = databaseName
def mysql_server_connect(self):
try:
self.connect = pymysql.connect(host=self.server, port=self.port, user=self.username, password=self.password,database=self.databaseName, connect_timeout=5) # 服务器名,账户,密码,数据库名
cur = self.connect.cursor()
if cur:
print(f"mysql 连接成功! ")
else:
print(f"mysql连接失败2!")
return cur
except Exception as e:
print(f"mysql连接失败3! 报错信息：{e}")
raise (NameError, "数据库连接异常")
def mysql_server_monitor(self):
try:
cur = self.mysql_server_connect()
# 具体的查询统计SQL
sql1 = (f""" SELECT * FROM table_name WHERE xxx ; """)
### 执行sql获取返回结果集
cur.execute(sql1)
result1Query = cur.fetchall()
return result1Query
# # 关闭游标
cur.close()
# 关闭连接
self.connect.commit()
self.connect.close()
except Exception as e:
print(f"mysql执行存储或语句失败!,\n报错信息：{e}")
三：使用python把获得的数据生成图片
安装依赖
pip3 install html-table
pip3 install imgkit
关键py代码：
## imgTitle标题内容
## jpgPath生成图片保存的路径+文件名称，比如/usr/local/aaa.jpg
## headTuple数据行标题 以元组形式传入， 比如(('列名1','列名2','列名3',...),)
## dataTuple数据行 以元组形式传入，比如(('第一行列1数据','第一行列2数据','第一行列3数据',...),('第二行列1数据','第二行列2数据','第二行列3数据',...),...)
def createImage(imgTitle,jpgPath,headTuple,dataTuple):
""" 根据查询的结果,生成图片"""
# 标题
table = HTMLTable(caption=imgTitle)
# 表头行
table.append_header_rows(headTuple)
# 数据行
table.append_data_rows(dataTuple)
# 标题样式
table.caption.set_style({
'font-size': '24px',
})
# 表格样式，即<table>标签样式
table.set_style({
'border-collapse': 'collapse',
'word-break': 'keep-all',
'white-space': 'nowrap',
'font-size': '14px',
})
# 统一设置所有单元格样式，<td>或<th>
table.set_cell_style({
'width': "250px",
'border-color': '#000',
'border-width': '1px',
'border-style': 'solid',
'padding': '5px',
})
# 表头样式
table.set_header_row_style({
'color': '#fff',
'background-color': '#48a6fb',
'font-size': '18px',
})
# 覆盖表头单元格字体样式
table.set_header_cell_style({
'padding': '15px',
})
# # 调小次表头字体大小
# table[1].set_cell_style({
# 'padding': '8px',
# 'font-size': '15px',
# })
# # 遍历数据行，如果增长量为负，标红背景颜色
# for row in table.iter_data_rows():
# if row[0].value == '所有渠道':
# row.set_style({
# 'background-color': '#ffdddd',
# })
body = table.to_html()
# html的charset='UTF-8'必须加上，否则中文会乱码
html = "<!DOCTYPE html><html><head><meta charset='UTF-8'></head><body>{0}</body></html>".format(body)
# 生成图片
imgkit.from_string(html,jpgPath)
四：把生成本地图片上传到阿里云OSS并获取图片url
4.1：在OSS上创建保存图片的目录
4.2：创建阿里云RAM账号
阿里云主账号AccessKey拥有所有API的访问权限，风险很高。强烈建议您创建并使用RAM账号进行API访问或日常运维。
请登录 https://ram.console.aliyun.com 创建RAM账号。
创建成功后把对应的AccessKey ID和AccessKey Secret记录下来，后面要用到。
然后在用户列表的【添加权限】中添加刚刚创建账号对oss的访问操作权限。
4.3：用python把本地图片上传到阿里云并获取访问链接
官方帮助文档：https://help.aliyun.com/document_detail/32026.html
yum install python-devel
pip3 install oss2
如果在安装oss2的时候出现如下错误：ModuleNotFoundError: No module named 'setuptools_rust'
需要升级一下setuptools ： pip3 install -U pip setuptools
验证是否安装成功：出现ImportError: No module named _crcfunext说明crcmod没有安装成功。
具体原因参考官方描述：https://help.aliyun.com/document_detail/85288.html
如果crcmod未安装成功，执行以下操作。
# 先卸载
pip3 uninstall crcmod
# 重新安装crcmod
pip3 install crcmod
# 我重装了，依旧报错ModuleNotFoundError: No module named 'crcmod._crcfunext'
# 先不管了，安装失败还是照样能用的，按官方说法就是上传性能不是很好。这个以后有空再看。
测试操作：把百度首页内容转换成pdf或jpg图片。
wkhtmltoimage https://www.baidu.com baidu.jpg
wkhtmltopdf https://www.baidu.com/ baidu.pdf
生成图片的py代码如下：
#!/usr/bin/python3
# -*- coding: utf-8 -*-
# @CreateDate : 2021/12/14 13:59
# @Author : dbartist
# @Email : dbartist@163.com
# @ScriptFile : picDemo.py
# @Project : PyScripts
# @Describe :
import imgkit
from HTMLTable import HTMLTable
## 下面的标题，表头行和数据行可以根据自己的需求自行调整。
## 最后的标红背景颜色可以根据需求调整下，其它可以不用动。
# 标题
table = HTMLTable(caption='果园收成表')
# 表头行
table.append_header_rows((
('名称', '产量 (吨)','增长量 (吨)','增长率 (%)'),
))
# 数据行
table.append_data_rows((
('荔枝', 11, 1, 10),
('芒果', 9, -1, -10),
('香蕉', 6, 1, 20),
))
# 标题样式
table.caption.set_style({
'font-size': '15px',
})
# 表格样式，即<table>标签样式
table.set_style({
'border-collapse': 'collapse',
'word-break': 'keep-all',
'white-space': 'nowrap',
'font-size': '14px',
})
# 统一设置所有单元格样式，<td>或<th>
table.set_cell_style({
'width': "250px",
'border-color': '#000',
'border-width': '1px',
'border-style': 'solid',
'padding': '5px',
})
# 表头样式
table.set_header_row_style({
'color': '#fff',
'background-color': '#48a6fb',
'font-size': '18px',
})
# 覆盖表头单元格字体样式
table.set_header_cell_style({
'padding': '15px',
})
# 调小次表头字体大小
table[1].set_cell_style({
'padding': '8px',
'font-size': '15px',
})
# 遍历数据行，如果增长量为负，标红背景颜色
for row in table.iter_data_rows():
if row[2].value < 0:
row.set_style({
'background-color': '#ffdddd',
})
body = table.to_html()
# html的charset='UTF-8'必须加上，否则中文会乱码
html = "<!DOCTYPE html><html><head><meta charset='UTF-8'></head><body>{0}</body></html>".format(body)
# 生成图片
imgkit.from_string(html, 'out.jpg')
上传阿里云oss的关键代码：
# 连接oss
def oss_parser(oss_dirpath,local_dirpath):
endpoint = 'http://oss-cn-hangzhou.' # 在哪个城市就选那个城市的oss-cn
access_key_id = 'LTA****vBS'
access_key_secret = 'yPR****muI'
bucket_name = '******'
# 指定Bucket实例，所有文件相关的方法都需要通过Bucket实例来调用。
bucket = oss2.Bucket(oss2.Auth(access_key_id, access_key_secret), endpoint, bucket_name)
# result = bucket.put_object(f'{dirpath}/{imageName}', img.getvalue())
result = bucket.put_object_from_file(f'{oss_dirpath}', f'{local_dirpath}')
print('jpg upload oss success!')
return result.status
def test():
## 生成图片 ######################################################
imgTitle = f'【统计内容】XXX . . . 统计时间:{datetime.now().strftime("%Y-%m-%d %H:%M:%S")}'
jpgName = f'jrdx{datetime.now().strftime("%Y%m%d%H%M%S")}.jpg'
jpgPath = f'/herajob/scripts_jpg/{jpgName}'
createImage(imgTitle, resultDic, jpgPath)
## 把生成的图片上传到阿里云oss
oss_dirpath = f'smddz/{jpgName}'
local_dirpath =f'/herajob/scripts_jpg/{jpgName}'
oss_parser(oss_dirpath,local_dirpath)
if __name__ == '__main__':
test()
4.4：异常问题
手动执行py脚本生成jpg文件是正常的，但是我放在调度系统(我用的是hera)上执行py脚本就会报错，报错如下：
which: no wkhtmltoimage in (/sbin:/bin:/usr/sbin:/usr/bin) No wkhtmltoimage executable found: "command not found"
which: no wkhtmltoimage in (/sbin:/bin:/usr/sbin:/usr/bin)
Traceback (most recent call last):
File "hera_smddz_rihuo_daily_image.py", line 430, in <module>
test()
File "hera_smddz_rihuo_daily_image.py", line 415, in test
createImage(imgTitle, resultDic, jpgPath)
File "hera_smddz_rihuo_daily_image.py", line 291, in createImage
imgkit.from_string(html,jpgPath)
File "/usr/local/lib/python3.6/site-packages/imgkit/api.py", line 106, in from_string
cover_first=cover_first,
File "/usr/local/lib/python3.6/site-packages/imgkit/imgkit.py", line 36, in __init__
self.wkhtmltoimage = self.config.get_wkhtmltoimage()
File "/usr/local/lib/python3.6/site-packages/imgkit/config.py", line 57, in get_wkhtmltoimage
raise OSError(wkhtmltoimage_error)
OSError:
No wkhtmltoimage executable found: "command not found"
If this file exists please check that this process can read it.
Otherwise please install wkhtmltopdf - http://
报错原因：明明已经成功安装了，还是提示找不到wkhtmltoimage。是因为他从/sbin:/bin:/usr/sbin:/usr/bin目录去which查找。
但是安装wkhtmltoimage默认是在/usr/local/bin目录。所以找不到wkhtmltoimage。
解决方法：可以/usr/bin目录添加软链接指向到/usr/local/bin目录的wkhtmltoimage。
[hdfs@centos ]$ which wkhtmltoimage
/usr/local/bin/wkhtmltoimage
sudo ln -s /usr/local/bin/wkhtmltopdf /usr/bin/wkhtmltopdf
sudo ln -s /usr/local/bin/wkhtmltoimage /usr/bin/wkhtmltoimage
参考文章：https://blog.csdn.net/weixin_44818729/article/details/110181938
五：发送图片到钉钉群
采用markdown格式通过钉钉机器人推送到钉钉群。
关键py代码：
def send_dingding_image(msg):
msgbody = {"msgtype": "markdown", "markdown": {"title":"推送标题", "text": msg}}
try:
requests.post(
"https://oapi.dingtalk.com/robot/send?access_token=xxxxxx",
json=msgbody)
print(f"send ok")
except Exception as e:
print(f'报错信息：{e}')
参考官方API文档：https://open.dingtalk.com/document/group/custom-robot-access
六：完整python脚本代码
# !/usr/bin/python3
# -*- coding: utf-8 -*-
# @CreateDate : 2021/12/14 11:04
# @Author : dbartist
# @Email : dbartist@163.com
# @ScriptFile : pydata2image.py
# @Project : PyScripts
# @Describe : 每日定时统计XXX数据转成图片并推送到钉钉群
import warnings
warnings.simplefilter('ignore', DeprecationWarning)
import sys
import importlib
importlib.reload(sys)
# sys.setdefaultencoding('utf8')
import pymysql
import smtplib
from email.mime.text import MIMEText
from datetime import datetime, date, timedelta
import time
import base64
import requests
from decimal import *
import imgkit
from HTMLTable import HTMLTable
import oss2
import uuid
import io
# mysql 连接信息
mysql_dbhost = "192.168.0.101"
mysql_dbport = 3306
mysql_dbname = "testdb"
mysql_dbuser = "db_user"
mysql_dbpasswd = "db_passwd"
class DatabaseOpt(object):
"""数据库操作对象"""
def __init__(self, server, port, username, password, databaseName):
self.server = server
self.port = port
self.username = username
self.password = password
self.databaseName = databaseName
def mysql_server_connect(self):
try:
self.connect = pymysql.connect(host=self.server, port=self.port, user=self.username, password=self.password,database=self.databaseName, connect_timeout=5) # 服务器名,账户,密码,数据库名
cur = self.connect.cursor()
if cur:
print(f"mysql连接成功! ")
else:
print(f"mysql连接失败2!")
return cur
except Exception as e:
print(f"mysql连接失败3! 报错信息：{e}")
raise (NameError, "数据库连接失败")
def mysql_server_monitor(self):
try:
cur = self.mysql_server_connect()
# 具体查询统计的SQL
sql1 = (f""" SELECT * from table_name where xxxx ;""")
# 执行SQL并返回查询结果集
cur.execute(sql1)
result1Query = cur.fetchall()
return result1Query
# # 关闭游标
cur.close()
# 关闭连接
self.connect.commit()
self.connect.close()
except Exception as e:
print(f"mysql执行存储或语句失败!,\n报错信息：{e}")
def send_dingding_image(msg):
"""给钉钉群发送消息"""
msgbody = {"msgtype": "markdown", "markdown": {"title":"DBA推送标题", "text": msg}}
try:
# 钉钉群机器人webhook地址，换成自己的。
requests.post(
"https://oapi.dingtalk.com/robot/send?*****",
json=msgbody)
print(f"send ok")
except Exception as e:
print(f'报错信息：{e}')
def monitorLogAdd(str):
"""添加日志信息"""
str = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())) + ':' + str
with open('/tmp/pyscript.log', 'a+') as f:
f.write(str)
f.write('\n')
def createImage(imgTitle,jpgPath,headTuple,dataTuple):
""" 根据查询的结果,生成图片"""
# 标题
table = HTMLTable(caption=imgTitle)
# 表头行
table.append_header_rows(headTuple)
# 数据行
table.append_data_rows(dataTuple)
# 标题样式
table.caption.set_style({
'font-size': '24px',
})
# 表格样式，即<table>标签样式
table.set_style({
'border-collapse': 'collapse',
'word-break': 'keep-all',
'white-space': 'nowrap',
'font-size': '14px',
})
# 统一设置所有单元格样式，<td>或<th>
table.set_cell_style({
'width': "250px",
'border-color': '#000',
'border-width': '1px',
'border-style': 'solid',
'padding': '5px',
'text-align':'center',
})
# 表头样式
table.set_header_row_style({
'color': '#fff',
'background-color': '#48a6fb',
'font-size': '18px',
})
# 覆盖表头单元格字体样式
table.set_header_cell_style({
'padding': '15px',
})
# # 调小次表头字体大小
# table[1].set_cell_style({
# 'padding': '8px',
# 'font-size': '15px',
# })
# # 遍历数据行，如果增长量为负，标红背景颜色
# for row in table.iter_data_rows():
# if row[0].value == '所有渠道':
# row.set_style({
# 'background-color': '#ffdddd',
# })
body = table.to_html()
# html的charset='UTF-8'必须加上，否则中文会乱码
html = "<!DOCTYPE html><html><head><meta charset='UTF-8'></head><body>{0}</body></html>".format(body)
# 生成图片
imgkit.from_string(html,jpgPath)
# 连接阿里云oss
def oss_parser(oss_dirpath,local_dirpath):
endpoint = 'http://oss-cn-hangzhou.' # 一般是在哪个城市就选那个城市的oss-cn,换成自己的
access_key_id = 'LTAI5***HoxvBS' # 换成自己的
access_key_secret = 'yPRMJ***YzmuI' # 换成自己的
bucket_name = '***' # 换成自己的
# 指定Bucket实例，所有文件相关的方法都需要通过Bucket实例来调用。
bucket = oss2.Bucket(oss2.Auth(access_key_id, access_key_secret), endpoint, bucket_name)
result = bucket.put_object_from_file(f'{oss_dirpath}', f'{local_dirpath}')
print('jpg upload oss success!')
return result.status
def test():
# 连接mysql,获取统计结果数据
test = DatabaseOpt(mysql_dbhost, mysql_dbport, mysql_dbuser, mysql_dbpasswd, mysql_dbname)
returnQueryResult = test.mysql_server_monitor()
# 定义表格标题元祖和数据元组
headTuple = (('统计内容', '0H', '1H', '2H', '3H', '4H', '5H', '6H', '7H', '8H', '9H', '10H', '11H', '12H',
'13H', '14H', '15H', '16H', '17H', '18H', '19H', '20H', '21H', '22H', '23H', '合计'),)
dataTuple = returnQueryResult
## 生成图片 图片格式: today20211215235959.jpg
imgTitle = f'【今日标题】今日XXX各小时段新增次留统计 . . . 统计时间:{datetime.now().strftime("%Y-%m-%d %H:%M:%S")}'
jpgName = f'today{datetime.now().strftime("%Y%m%d%H%M%S")}.jpg'
jpgPath = f'/herajob/scripts_jpg/{jpgName}'
createImage(imgTitle, jpgPath, headTuple, dataTuple)
## 把生成的图片上传到阿里云oss
oss_dirpath = f'smddz/{jpgName}'
local_dirpath =f'/herajob/scripts_jpg/{jpgName}'
oss_parser(oss_dirpath,local_dirpath)
## 把oss的图片发送到钉钉群
image_url = f'https://******.oss-cn-hangzhou./smddz/{jpgName}'
msg = f'### 今日标题 \n >**今日XXX各小时段新增次留统计!** \n > \n ###### 统计时间 : {datetime.now().strftime("%Y-%m-%d %H:%M:%S")} \n'
send_dingding_image(msg)
# print(f'告警信息： \n{msg}')
if __name__ == '__main__':
test()
推送到钉钉群消息如下：
点击图片详情如下：
至此大功告成！！！后续还有再优化的空间，比如生成趋势图/柱状图/饼图/曲线图等形式的图片推送到钉钉群。
|
