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

每天早上7点的关怀提醒

定义查询方法 greeting.py

import datetime
from requests import get
import wxHook
from Constant import QWEATHER_URL, FORMAL_ROBOTprint(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_msgscheduler = 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_imgalert_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_listif 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来源：" + sourcekind = "预警信息"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']]  # 长度置为1print("当前预警id：" + str(alert_id_list))else:# print(datetime.datetime.now().strftime('%m月%d日%H时%M分') + " 当前id的预警已经发送过了")passelse:# print(datetime.datetime.now().strftime('%m月%d日%H时%M分') + " 当前没有预警")pass# 判断雨量now_time = time.time()global sendTimeglobal timesInDayif now_time - sendTime >= TEST_GAP:  # 大于1小时即可，数字在3600if int(max(precipitation)) >= 51.30:    # 51.30red_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 += 1print("当前时间戳：" + str(sendTime))print(datetime.datetime.now().strftime('%m月%d日%H时%M分') + ":timesInDay = " + str(timesInDay))else:# print("今日发送次数已达5次")passelif int(max(precipitation)) > 11.33:   # 11.33yellow_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 += 1print("当前时间戳：" + str(sendTime))print(datetime.datetime.now().strftime('%m月%d日%H时%M分') + ":timesInDay = " + str(timesInDay))else:# print("今日发送次数已达5次")passelif int(max(precipitation)) > 10:  # 10green_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 += 1print("当前时间戳：" + str(sendTime))print(datetime.datetime.now().strftime('%m月%d日%H时%M分') + ":timesInDay = " + str(timesInDay))else:# print("今日发送次数已达5次")passelse:# print(datetime.datetime.now().strftime('%m月%d日%H时%M分') + " 雨量太小，无需预警")passelse:# print(datetime.datetime.now().strftime('%m月%d日%H时%M分') + " 据上次预警还没到1小时")pass# 重置发送每日次数
def reset_timesInDay():global timesInDaytimesInDay = 0print(datetime.datetime.now().strftime('%m月%d日%H时%M分') + ":timesInDay重置成功")

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

from testOneTime import test_one, reset_timesInDayscheduler = 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 mysqlclientfrom Constant import db_host, db_port, db_user, db_passwd, db_name, CAIYUN_SAVE_URLdef 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 = dateif day in [4, 5]:  # 如果是周末if on_time < now_time < off_time_weekend:return 1else:return 0else:  # 如果不是周末if on_time < now_time < off_time:return 1else:return 0def 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()

用这个思路，其实可以判断交通是否拥堵、工作任务提醒、行程提醒等场景，还可以做一些非工作的个人功能，比如定时喝水、疫情通报、每日鸡汤等。

【Python】天气预报及雨量预警到企业微信群的代码实现相关推荐

python 企业微信群机器人_企业微信群机器人应用：使用python从网站抓取行业资讯并定时推送...
在企业经营过程中,及时了解行业相关信息(市场动态.竞品策略.行业数据等等)是非常必要的.通常情况下,商品部门.营销部门.市场部门可能都会安排专门的人员定期进行这些信息的搜集.整理,再进行内部的分享. ...
div区域内容抓取_企业微信群机器人应用：使用python从网站抓取行业资讯并定时推送...
在企业经营过程中,及时了解行业相关信息(市场动态.竞品策略.行业数据等等)是非常必要的.通常情况下,商品部门.营销部门.市场部门可能都会安排专门的人员定期进行这些信息的搜集.整理,再进行内部的分享. ...
python实现由通知接口发送企业微信通知
python实现由通知接口发送企业微信通知问题背景要点说明 1. 请求接口鉴权并返回token值 2.请求通知接口发送企业微信通知 3.通知内容中传入参数问题背景项目中需要通过 python ...
Python实现发送警告通知到企业微信方法详解
常见的报警方式有:邮件,电话,短信,微信.本文将介绍如何利用Python发送警告通知到企业微信,文中的示例代码有一定的参考价值,感兴趣的可以了解一下.编程资料点击免费领取目录 1. 新建应用 2. ...
Python实现企业微信群机器人自动化推送
人工智能(Artificial Intelligence),英文缩写为AI.它是研究.开发用于模拟.延伸和扩展人的智能的理论.方法.技术及应用系统的一门新的技术科学. --<百度百科> 文 ...
运用python实现企业微信群机器人消息推送
使用场景:将BI报表精准推送入(群),精准触达用户目的:提高管理层对数据的感知度工具:python+企业微信步骤: 1.创建企业微信群机器人,提取Webhook地址(群机器人地址) 2.编写代码 ...
企业微信群机器人是什么？企微机器人如何自动发消息？
经常有很多用户会问我们,企业微信群机器人是什么,有什么用,有什么限制?企业微信群机器人提供的是一个webhook消息,如何可以通过这个来自动发消息呢?之前写过一些关于企业微信群机器人如何发消息的教程, ...
【SCF CLI实践】腾讯云serverless + 企业微信群机器人，轻松解决告警通知问题
市面上有什么好用的从服务器推报警和日志的工具?之前私下用的是[Server酱]的服务,非常方便. 但是考虑到安全原因,这个服务如果用在生产环境心里还是有点慌(虽然我相信Server酱是很有节操的). ...
企业微信机器人脚本python_Python 操控企业微信群机器人
目标企业微信群机器人常用来作为通知工具,群发消息给群内成员,充当小助手的角色.但若按照官方 API 文档来构建请求,也确实不太方便.本文通过 Python 第三方库来控制企业微信群机器人发送消息. ...

【Python】天气预报及雨量预警到企业微信群的代码实现

每天早上7点的关怀提醒

雨量预警功能

温度数据采集功能

【Python】天气预报及雨量预警到企业微信群的代码实现相关推荐

最新文章

热门文章