【Python】天气预报及雨量预警到企业微信群的代码实现每天早上7点的关怀提醒雨量预警功能温度数据采集功能

116 阅读 0 评论 77 点赞

我是靠谱客的博主想人陪墨镜，最近开发中收集的这篇文章主要介绍【Python】天气预报及雨量预警到企业微信群的代码实现每天早上7点的关怀提醒雨量预警功能温度数据采集功能，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

最近打算做个关于天气主题的小程序，互联网上有免费的API可以查询天气预报，有降雨量和降雨概率，风向等指标可查，我选择了彩云天气和和风天气两个接口，和风用来做每天早上的关怀提醒，彩云用来每5分钟检测一次雨量，雨量如果在中雨以上就发送消息到企业微信群，相关同事可以做好工作预案。

每天早上7点的关怀提醒

定义查询方法 greeting.py

import datetime
from requests import get
import wxHook
from Constant import QWEATHER_URL, FORMAL_ROBOT
print(datetime.datetime.now().strftime('%m月%d日%H时%M分') + " 晨间播报程序开始")
# 星期函数
def get_week_day(date):
week_day_dict = {
0: '星期一', 1: '星期二', 2: '星期三', 3: '星期四', 4: '星期五', 5: '星期六', 6: '星期天',
}
day = date.weekday()
return week_day_dict[day]
# 晨间播报
def greeting_msg():
weekDay = get_week_day(datetime.datetime.now())
response = get(QWEATHER_URL)
raw_json = response.json()
temperature = raw_json["hourly"][0]["temp"]
weather = raw_json["hourly"][0]["text"]
arrayTemp = [raw_json["hourly"][0]["temp"], raw_json["hourly"][1]["temp"], raw_json["hourly"][2]["temp"],
raw_json["hourly"][3]["temp"], raw_json["hourly"][4]["temp"], raw_json["hourly"][5]["temp"],
raw_json["hourly"][6]["temp"], raw_json["hourly"][7]["temp"], raw_json["hourly"][8]["temp"],
raw_json["hourly"][9]["temp"], raw_json["hourly"][10]["temp"], raw_json["hourly"][11]["temp"],
raw_json["hourly"][12]["temp"], raw_json["hourly"][13]["temp"], raw_json["hourly"][14]["temp"],
raw_json["hourly"][15]["temp"], raw_json["hourly"][16]["temp"], raw_json["hourly"][17]["temp"],
raw_json["hourly"][18]["temp"], raw_json["hourly"][19]["temp"], raw_json["hourly"][20]["temp"],
raw_json["hourly"][21]["temp"], raw_json["hourly"][22]["temp"], raw_json["hourly"][23]["temp"]]
maxTemp = max(arrayTemp)
minTemp = min(arrayTemp)
windDir = raw_json["hourly"][0]["windDir"]
windScale = raw_json["hourly"][0]["windScale"]
windSpeed = raw_json["hourly"][0]["windSpeed"]
pressure = raw_json["hourly"][0]["pressure"]
humidity = raw_json["hourly"][0]["humidity"]
arrayPop = [raw_json["hourly"][0]["pop"], raw_json["hourly"][1]["pop"]]
greeting_info = "n早上好，亲爱的小伙伴n当前时间：" + datetime.datetime.now().strftime('%Y年%m月%d日%H时%M分') + " " + weekDay + 
"n实时天气：" + weather + "n温度：" + temperature + "°C" + "n最高温度：" + maxTemp + "°C" + "n最低温度：" 
+ minTemp + "°C" + "n降雨概率：" + max(arrayPop) + "%" + "n风向：" + windDir + "n风力等级：" + windScale + "级" +
"n风速：" + windSpeed + "公里/小时" + "n气压：" + pressure + "百帕" + "n湿度：" + humidity + "%" + 
"n请注意通勤安全，祝您工作顺利，一整天都心情愉快"
kind = "晨间播报"
wxHook.warning_send(FORMAL_ROBOT, kind, greeting_info)
print(datetime.datetime.now().strftime('%m月%d日%H时%M分') + " " + kind + ' 企业微信群成功发送')

这里将发送企业微信消息封装成方法，传入webhook地址、内容即可触发发送 wxHook.py

def warning_send(RbUrl, kind, info):
headers = {"Content-Type": "text/plain"}
send_data = {
"msgtype": "markdown",
"markdown": {
"content": "**<font color="info">" + kind + "</font>**：" + info
# 文本内容，最长不超过2048个字节，必须是utf8编码
}
}
requests.post(url=RbUrl, headers=headers, json=send_data)

定时执行 scheduleModel.py

from apscheduler.schedulers.blocking import BlockingScheduler
from greeting import greeting_msg
scheduler = BlockingScheduler(timezone='Asia/Chongqing')
scheduler.add_job(greeting_msg, 'cron', day_of_week='mon-sun', hour='7', minute='00', max_instances=500, jitter=5)
scheduler.start()

雨量预警功能

定义雨量预警检测方法，这里接口可以获取类似台风之类的预警，也能获取到未来12小时雨量和降雨概率。为了不重复发送预警，用了一个数组变量装载预警id，当数组长度达到20以后再清空重新装载。用时间戳变量来实现发送一次过后，一段时间内不再发送，这样比较合理。为了不打扰日常的工作，还定义了一个变量，每次发送加1，用于限定一天内最多发送次数，晚上11点后这个变量置为0。

写一个testOneTime.py

import datetime
import time
from requests import get
import wxHook
from Constant import TEST_GAP, CAIYUN_URL, MAX_PER_DAY, FORMAL_ROBOT, divide_img, divide_gmc_img
alert_id_list = []
sendTime = 0
timesInDay = 0
print(datetime.datetime.now().strftime('%m月%d日%H时%M分') + " 预警id和时间戳初始化完毕，检测程序开始")
def test_one():
response = get(CAIYUN_URL)
raw_json = response.json()
alert = raw_json["result"]["alert"]["content"]
precipitation = raw_json["result"]["minutely"]["precipitation"]
probability = raw_json["result"]["minutely"]["probability"]
# 判断预警
if alert:
alert_json = alert[0]
global alert_id_list
if alert_json['alertId'] not in alert_id_list:
# alert_id != alert_json['alertId']:
title = alert_json['title']
description = alert_json['description']
source = alert_json['source']
alert_info = "n" + title + "n内容：" + description + "n来源：" + source
kind = "预警信息"
wxHook.alert_send(FORMAL_ROBOT, kind, alert_info)
print(datetime.datetime.now().strftime('%m月%d日%H时%M分') + " " + kind + ' 企业微信群成功发送')
# 判断并赋值
if len(alert_id_list) < 20:
alert_id_list.append(alert_json['alertId'])
# alert_id = alert_json['alertId']
else:
alert_id_list = [alert_json['alertId']]
# 长度置为1
print("当前预警id：" + str(alert_id_list))
else:
# print(datetime.datetime.now().strftime('%m月%d日%H时%M分') + " 当前id的预警已经发送过了")
pass
else:
# print(datetime.datetime.now().strftime('%m月%d日%H时%M分') + " 当前没有预警")
pass
# 判断雨量
now_time = time.time()
global sendTime
global timesInDay
if now_time - sendTime >= TEST_GAP:
# 大于1小时即可，数字在3600
if int(max(precipitation)) >= 51.30:
# 51.30
red_info = "n当前时间：" + datetime.datetime.now().strftime('%H时%M分') 
+ "n2小时内降雨概率：" + '{:.2f}'.format(max(probability) * 100) + "%" 
+ "n预计降水强度：" + '{:.2f}'.format(max(precipitation)) + "毫米/小时"
kind = "暴雨预警"
if timesInDay <= MAX_PER_DAY:
wxHook.red_send(FORMAL_ROBOT, kind, red_info)
wxHook.wx_image(FORMAL_ROBOT, divide_img)
print(datetime.datetime.now().strftime('%m月%d日%H时%M分') + " " + kind + ' 企业微信群成功发送')
sendTime = time.time()
timesInDay += 1
print("当前时间戳：" + str(sendTime))
print(datetime.datetime.now().strftime('%m月%d日%H时%M分') + ":timesInDay = " + str(timesInDay))
else:
# print("今日发送次数已达5次")
pass
elif int(max(precipitation)) > 11.33:
# 11.33
yellow_info = "n当前时间：" + datetime.datetime.now().strftime('%H时%M分') 
+ "n2小时内降雨概率：" + '{:.2f}'.format(max(probability) * 100) + "%" 
+ "n预计降水强度：" + '{:.2f}'.format(max(precipitation)) + "毫米/小时"
kind = "大雨预警"
if timesInDay <= MAX_PER_DAY:
wxHook.yellow_send(FORMAL_ROBOT, kind, yellow_info)
wxHook.wx_image(FORMAL_ROBOT, divide_img)
print(datetime.datetime.now().strftime('%m月%d日%H时%M分') + " " + kind + ' 企业微信群成功发送')
sendTime = time.time()
timesInDay += 1
print("当前时间戳：" + str(sendTime))
print(datetime.datetime.now().strftime('%m月%d日%H时%M分') + ":timesInDay = " + str(timesInDay))
else:
# print("今日发送次数已达5次")
pass
elif int(max(precipitation)) > 10:
# 10
green_info = "n当前时间：" + datetime.datetime.now().strftime('%H时%M分') 
+ "n2小时内降雨概率：" + '{:.2f}'.format(max(probability) * 100) + "%" 
+ "n预计降水强度：" + '{:.2f}'.format(max(precipitation)) + "毫米/小时"
kind = "中雨预警"
if timesInDay <= MAX_PER_DAY:
wxHook.green_send(FORMAL_ROBOT, kind, green_info)
wxHook.wx_image(FORMAL_ROBOT, divide_img)
print(datetime.datetime.now().strftime('%m月%d日%H时%M分') + " " + kind + ' 企业微信群成功发送')
sendTime = time.time()
timesInDay += 1
print("当前时间戳：" + str(sendTime))
print(datetime.datetime.now().strftime('%m月%d日%H时%M分') + ":timesInDay = " + str(timesInDay))
else:
# print("今日发送次数已达5次")
pass
else:
# print(datetime.datetime.now().strftime('%m月%d日%H时%M分') + " 雨量太小，无需预警")
pass
else:
# print(datetime.datetime.now().strftime('%m月%d日%H时%M分') + " 据上次预警还没到1小时")
pass
# 重置发送每日次数
def reset_timesInDay():
global timesInDay
timesInDay = 0
print(datetime.datetime.now().strftime('%m月%d日%H时%M分') + ":timesInDay重置成功")

定时执行在scheduleModel.py中添加任务

from testOneTime import test_one, reset_timesInDay
scheduler = BlockingScheduler(timezone='Asia/Chongqing')
scheduler.add_job(test_one, 'cron', day_of_week='mon-sun', hour='*', minute='*/5', max_instances=500, jitter=5)
scheduler.add_job(reset_timesInDay, 'cron', day_of_week='mon-sun', hour='23', minute='30', max_instances=500, jitter=5)
scheduler.add_job(reset_timesInDay, 'cron', day_of_week='mon-sun', hour='23', minute='50', max_instances=500, jitter=5)
scheduler.start()

温度数据采集功能

每小时将温度传入大数据数据库 tempRecord.py

import datetime
from requests import get
import MySQLdb
# pip install mysqlclient
from Constant import db_host, db_port, db_user, db_passwd, db_name, CAIYUN_SAVE_URL
def get_status(date):
# 范围时间
on_time = datetime.datetime.strptime(str(datetime.datetime.now().date()) + '10:00', '%Y-%m-%d%H:%M')
off_time_weekend = datetime.datetime.strptime(str(datetime.datetime.now().date()) + '23:00', '%Y-%m-%d%H:%M')
off_time = datetime.datetime.strptime(str(datetime.datetime.now().date()) + '22:30', '%Y-%m-%d%H:%M')
# 判断当前时间是否在范围时间内
day = date.weekday()
now_time = date
if day in [4, 5]:
# 如果是周末
if on_time < now_time < off_time_weekend:
return 1
else:
return 0
else:
# 如果不是周末
if on_time < now_time < off_time:
return 1
else:
return 0
def save_temp():
response = get(CAIYUN_SAVE_URL)
raw_json = response.json()
status = get_status(datetime.datetime.now())
temperature = raw_json["result"]["realtime"]["temperature"]
db = MySQLdb.connect(host=db_host, port=db_port, user=db_user, passwd=db_passwd, db=db_name)
cursor = db.cursor()
sql_string = "INSERT INTO weather (datetime, temperature, status) VALUES (%s, %s, %s);"
cursor.execute(sql_string, (datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'), temperature, status))
# 关闭数据库连接
db.commit()
cursor.close()
db.close()