【Python自动化】定时自动采集,并发送微信告警通知,全流程案例讲解!

【Python自动化】定时自动采集,并发送微信告警通知,全流程案例讲解!

一、概要您好!我是@马哥python说,一名10年程序猿。

我原创开发了一套定时自动化爬取方案,完整开发流程如下:

采集数据 -> 筛选数据 -> 存MySQL数据库 -> 发送邮件 -> 微信提醒 -> 定时执行

如果您现在苦于每天繁琐、重复的数据采集工作,可尝试套用该自动化方案,节省人力,降本增效!

二、效果演示基于数据隐私保护,部分数据已脱敏。

数据库结果(部分数据已脱敏):

▲ 图2.1 MySQL里的数据

Excel结果(部分数据已脱敏):

图片▲ 图2.2 csv数据

微信消息提醒:

图片▲ 图2.3 微信提醒

收件箱告警邮件:

图片▲ 图2.4 告警邮件

邮件正文内容(部分数据已脱敏):

图片▲ 图2.5 邮件正文

以上。

三、代码讲解3.1 爬虫采集行政处罚数据爬取目标是某公示平台的行政处罚数据。

通过浏览器的开发者模式分析接口,页面显示数据与接口返回数据一致,故可通过爬取接口的方式采集。

图片▲ 图3.1 开发者模式

首先,导入需要用到的库:

代码语言:javascript代码运行次数:0运行复制import requests # 发送请求

import pandas as pd # 存取csv

import os # 判断本地文件

import random # 随机

from time import sleep # 等待时长,防止反爬

from sqlalchemy import create_engine # 连接数据库

import json # 转换数据格式

然后,定义请求头(真实地址已脱敏):

代码语言:javascript代码运行次数:0运行复制# 请求头

headers = {

'Accept': 'application/json, text/javascript, */*; q=0.01',

'Accept-Encoding': 'gzip, deflate, br',

'Accept-Language': 'zh-CN,zh;q=0.9,en-US;q=0.8,en;q=0.7',

'Connection': 'keep-alive',

'Content-Type': 'application/json;charset=UTF-8',

'Cookie': '换成自己的cookie',

'Host': 'xxx',

'Origin': 'xxx',

'Referer': 'xxx',

'sec-ch-ua': '"Chromium";v="112", "Google Chrome";v="112", "Not:A-Brand";v="99"',

'sec-ch-ua-mobile': '?0',

'sec-ch-ua-platform': '"macOS"',

'Sec-Fetch-Dest': 'empty',

'Sec-Fetch-Mode': 'cors',

'Sec-Fetch-Site': 'same-origin',

'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/112.0.0.0 Safari/537.36',

'X-Requested-With': 'XMLHttpRequest'

}

定义请求参数,即开发者中的payload参数,不再展示。

发送post请求并接收返回数据:

代码语言:javascript代码运行次数:0运行复制# 发送post请求

r = requests.post(url, headers=headers, data=json.dumps(data))

# 查看响应码

print(r.status_code)

# json解析数据

resp_json = r.json()

data_list = resp_json['data']['list']

逐个解析返回数据,以"处罚金额"为例:

代码语言:javascript代码运行次数:0运行复制for data in data_list:

# 处罚金额

punish_amount = data['f_2022118615143']

try:

if float(punish_amount) >= threshold: # 如果超过告警阈值,就存入数据

punish_amount_list.append(punish_amount)

else: # 否则进入下一轮循环

continue

except:

continue

这里解释一下:如果处罚金额超过事先设定好的阈值,则存入数据,否则不存,continue进入下一轮循环,由此达成告警的目的。

其他字段(案件名称、处罚人姓名、处罚日期、处罚机关等)同理,不再赘述代码。

最后是保存数据到csv文件:

代码语言:javascript代码运行次数:0运行复制# 保存数据到Dataframe

df = pd.DataFrame(

{

'页码': page,

'案件名称': case_name_list,

'处罚人姓名': punish_name_list,

'处罚金额': punish_amount_list,

'处罚日期': punish_time_list,

'处罚机关': punish_org_list,

}

)

# 保存到csv文件

df.to_csv(result_file, mode='a+', header=header, index=False, encoding='utf_8_sig')

数据存入csv效果,查看图2.2。

3.2 存MySQL数据库如上所述,数据保存到csv文件作为临时存储,下面保存到MySQL数据库作为持久性存储。

我采用sqlalchemy和pandas的to_sql结合的方式,把csv数据快速导入MySQL数据库。

关键代码(真实信息已脱敏):

代码语言:javascript代码运行次数:0运行复制# 创建MySQL数据库连接

engine = create_engine('mysql+pymysql://用户名:密码@IP地址/数据库名')

# 读取csv数据

df_punish = pd.read_csv(result_file)

# 把csv数据导入MySQL数据库

df_punish.to_sql(name='table_name', con=engine, chunksize=, if_exists='replace', index=False)

print('导入数据库完成!')

这样的简单3行代码,即实现了csv数据导入MySQL数据库的目的。

注意,to_sql中的if_exists代表如果表中存在数据,那么replace覆盖原始数据,这样不会产生重复数据。当然,也可以加个ist_time插入时间的区分字段,这样可以把if_exists设置为append,保证每次爬取都留痕,你可以试试,我认为这样代码更完善、效果更好。

数据入库效果,查看图2.1。

3.3 发送告警邮件&微信通知自动发送邮件,我采用zmail库实现。

关键代码(真实信息已脱敏):

代码语言:javascript代码运行次数:0运行复制def send_email(v_location, v_content):

# 当前时间戳

now = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')

# 设置邮件信息

mail_content = {

'subject': '税务行政处罚预警邮件 | {} | {}'.format(v_location, now), # 邮件标题

'content_text': '尊敬的管理员,您好!\n\n以下是【{}】税务行政处罚预警名单,请注意查收。\n\n'.format(

v_location) + v_content # 邮件内容

}

# 发件人的用户名和密码

server = zmail.server('xxx@qq.com', 'xxx') # 发件人

# 发送邮件

server.send_mail('xxx@qq.com', mail_content) # 收件人

print('邮件已发送完毕:', v_location)

注意,发件人的smtp服务必须开启,否则会邮件发送失败,如下:

图片▲ 图3.2 邮箱开启smtp服务

邮件收到了,怎么实现微信通知呢?秘密就是把微信的邮件功能打开,如下:

图片▲ 图3.3 微信设置

微信:我 -> 设置 -> 通用 -> 辅助功能 -> QQ邮箱提醒 -> 开启功能

这样就能让微信收到消息提醒了,效果如图2.3所示。

多说一句,请勿尝试用itchat、wxpy等第三方库操作微信,微信官方已停用这类工具,且存在封号风险!不信你就逝世![手动狗头]

3.4 定时机制定时执行也是采用简单粗暴的方式,直接os.system调用分别的py文件,死循环加sleep的方式:

代码语言:javascript代码运行次数:0运行复制while True:

# 执行爬虫

print(get_now(), '开始爬虫!')

os.system('python 爬虫.py')

print(get_now(), '爬取结束!')

sleep()

# 发邮件

print(get_now(), '开始发预警邮件!')

os.system('python 发预警邮件.py')

print(get_now(), '结束发预警邮件!')

sleep()

# 等待时长60分钟

print(get_now(), '开始等待60分钟..')

sleep()

这样,程序只要在后台一直运行就好了,每隔3600秒(即1小时)自动执行一次,可自定义设置间隔时长。

最终得到的效果就是每隔1小时微信收到一次消息通知,效果如图2.3所示。

四、总结这套定时采集通知方案,全流程均由本人原创开发,可套用到类似业务需求上,其中部分环节可根据需要,扩展相关功能。请小伙伴们打开脑洞,码上开发!

如文中所说,部分信息涉及隐私保护,所以不提供完整代码,有类似需求的小伙伴可私信讨论。

📚 相关推荐

惠州市行政区划
以前游戏分“电信”和“网通区”,为什么现在不分了?看完心里清楚了
江湖风云录如何快速提升等级
合肥博大教育
从零到月入过万:写小说赚钱的6大实战指南(附真实案例)
电动车电瓶分几种类型