最近打算做个关于天气主题的小程序，互联网上有免费的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()

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