接口测试 接口自动化之每日天气预报通过钉钉机器人发送消息提醒

xiaoxiao · 2019年12月30日 · 最后由 点点 回复于 2019年12月31日 · 259 次阅读

前提

1、下载安装 iOS【实况天气】(Android 请在豌豆荚搜索 “天气预报” 下载安装【实况天气预报】,注意看图标 icon 跟下面截图对不对),如图:

2、通过 Charles 抓包,获取对应接口的 url 和 request 的 data 信息:

3、在钉钉对应群里设置钉钉机器人,并获取对应的 Webhook 链接,如下图:

Python 代码实现

注:代码涉及到阳历阴历、风速风向、aqi、中英文翻译等的转换,具体查看代码注释

# -*- coding: utf-8 -*-
# @file: get_weather.py
# @author: xulinzhou
# @date  : 2019/12/26

import requests
import json
import sxtwl
import time
from datetime import date
from translate import Translator
import re

# 时间戳转阳历日期
def time_stamp_to_date(time_stamp):
    return time.strftime("%Y--%m--%d", time.localtime(time_stamp))


# 日历中文索引
ymc = [u"十一", u"十二", u"正", u"二", u"三", u"四", u"五", u"六", u"七", u"八", u"九", u"十"]
rmc = [u"初一", u"初二", u"初三", u"初四", u"初五", u"初六", u"初七", u"初八", u"初九", u"初十", \
       u"十一", u"十二", u"十三", u"十四", u"十五", u"十六", u"十七", u"十八", u"十九", \
       u"二十", u"廿一", u"廿二", u"廿三", u"廿四", u"廿五", u"廿六", u"廿七", u"廿八", u"廿九", u"三十", u"卅一"]

# 阳历转阴历,入参格式为:xxxx-xx-xx
def solar_to_lunar(solar_date):
    # 日历库实例化
    lunar = sxtwl.Lunar()

    today = str(solar_date)  # 如 2019-08-08
    today_list = today.split('-')  # ['2019', '08', '08']
    lunar_day = lunar.getDayBySolar((int)(today_list[0]), (int)(today_list[1]), (int)(today_list[2]))  # 输入年月日
    # 判断是否为润年
    if (lunar_day.Lleap):
        # print("润{0}月{1}".format(ymc[lunar_day.Lmc], rmc[lunar_day.Ldi]))
        return "润{0}月{1}".format(ymc[lunar_day.Lmc], rmc[lunar_day.Ldi])
    else:
        # print("{0}月{1}".format(ymc[lunar_day.Lmc], rmc[lunar_day.Ldi]))
        return "{0}月{1}".format(ymc[lunar_day.Lmc], rmc[lunar_day.Ldi])

# 获取星期几
def get_week_day(date):
  week_day_dict = {
    0 : '星期一',
    1 : '星期二',
    2 : '星期三',
    3 : '星期四',
    4 : '星期五',
    5 : '星期六',
    6 : '星期日',
  }
  day = date.weekday()
  return week_day_dict[day]


# 当前天气转换,入参为接口返回的string
def sky_exchange_chinese(sky):
    # 天气:晴、多云、阴、小雨、中雨、大雨、暴雨、小雪、中雪、大雪、暴雪
    if sky.startswith("CLEAR"):
        sky = "晴"
    elif sky.startswith("CLOUDY"):
        sky = "多云"
    elif sky.startswith('PARTLY_CLOUDY'):
        sky = "阴"
    elif sky.startswith("HAZE"):
        sky = "霾"
    elif sky.startswith("RAIN_L"):
        sky = "小雨"
    elif sky.startswith("RAIN_M"):
        sky = "中雨"
    elif sky.startswith("SNOW_L"):
        sky = "小雪"
    elif sky.startswith("SNOW_M"):
        sky = "中雪"
    elif sky == "WIND":
        sky = "风"
    else:
        translator = Translator(from_lang="English", to_lang="chinese")
        sky = translator.translate(sky)
    return sky


# 整天天气转换,参数为早上天气和晚上天气
def daily_sky_exchange(down_sky, up_sky):
    down_sky = sky_exchange_chinese(down_sky)
    up_sky = sky_exchange_chinese(up_sky)
    if down_sky == up_sky:
        tomorrow_sky = down_sky
    else:
        tomorrow_sky = down_sky + "转" + up_sky
    return tomorrow_sky


# aqi转换
def aqi_exchange(aqi):
    if aqi <= 0:
        aqi = ""
    elif aqi <= 50:
        aqi = str(aqi) + "优"
    elif aqi <= 100:
        aqi = str(aqi) + "良"
    elif aqi <= 150:
        aqi = str(aqi) + "轻度污染"
    elif aqi <= 200:
        aqi = str(aqi) + "中度污染"
    elif aqi <= 300:
        aqi = str(aqi) + "重度污染"
    elif aqi > 300:
        aqi = str(aqi) + "严重污染"
    else:
        pass
    return aqi


# 风力转换,单位:km/h
def wind_speed_exchange(windSpeed):
    if windSpeed <= 0.2 * 3.6:
        windSpeed = "无风"
    elif windSpeed <= 1.5 * 3.6:
        windSpeed = "1级软风"
    elif windSpeed <= 3.3 * 3.6:
        windSpeed = "2级轻风"
    elif windSpeed <= 5.4 * 3.6:
        windSpeed = "3级微风"
    elif windSpeed <= 7.9 * 3.6:
        windSpeed = "4级和风"
    elif windSpeed <= 10.7 * 3.6:
        windSpeed = "5级清风"
    elif windSpeed <= 13.8 * 3.6:
        windSpeed = "6级强风"
    elif windSpeed <= 17.1 * 3.6:
        windSpeed = "7级疾风"
    elif windSpeed <= 20.7 * 3.6:
        windSpeed = "8级大风"
    elif windSpeed <= 24.4 * 3.6:
        windSpeed = "9级烈风"
    elif windSpeed <= 28.4 * 3.6:
        windSpeed = "10级狂风"
    elif windSpeed <= 32.6 * 3.6:
        windSpeed = "11级暴风"
    elif windSpeed <= 36.9 * 3.6:
        windSpeed = "12级台风"
    elif windSpeed <= 41.4 * 3.6:
        windSpeed = "13级台风(一级飓风)"
    elif windSpeed <= 46.1 * 3.6:
        windSpeed = "14级强台风(一级飓风)"
    elif windSpeed <= 50.9 * 3.6:
        windSpeed = "15级强台风(二级飓风)"
    elif windSpeed <= 56.0 * 3.6:
        windSpeed = "16级超强台风(三级飓风)"
    elif windSpeed <= 61.2 * 3.6:
        windSpeed = "17级超强台风(四级飓风)"
    elif windSpeed > 61.2 * 3.6:
        windSpeed = "18级超强台风(五级飓风)"
    else:
        pass
    return windSpeed


# 风向转换
def wind_direction_exchange(windDirection):
    if windDirection <= 11.25 or windDirection > 348.76:
        windDirection = "北风"
    elif windDirection < 33.75:
        windDirection = "北东北风"
    elif windDirection < 56.25:
        windDirection = "东北风"
    elif windDirection < 78.75:
        windDirection = "东东北风"
    elif windDirection < 101.25:
        windDirection = "东风"
    elif windDirection < 123.75:
        windDirection = "东东南风"
    elif windDirection < 146.25:
        windDirection = "东南风"
    elif windDirection < 168.75:
        windDirection = "南东南风"
    elif windDirection < 191.25:
        windDirection = "南风"
    elif windDirection < 213.75:
        windDirection = "南西南风"
    elif windDirection < 236.25:
        windDirection = "西南风"
    elif windDirection < 258.75:
        windDirection = "西西南风"
    elif windDirection < 281.25:
        windDirection = "西风"
    elif windDirection < 303.75:
        windDirection = "西西北风"
    elif windDirection < 326.25:
        windDirection = "西北风"
    elif windDirection < 348.75:
        windDirection = "北西北风"
    else:
        pass
    return windDirection


# 当前提示语转换
def des_exchange(des):
    # 用Translator翻译成中文也不是很友好,所以优先人工翻译
    if "clear weather over the next 24 hours" == des:
        des = "未来24小时晴"
    elif "overcast, drizzle after" in des and "o'clock tomorrow midnight, followed by cloudy" in des:
        des = "小雨,明天午夜" + str(re.findall(r"\d+", des)[0]) + "点钟后雨停,转阴,其后小雨"
    elif "drizzle turn into overcast, drizzle expected after" in des and "o'clock tomorrow midnight" in des:
        des = "阴,明天午夜" + str(re.findall(r"\d+", des)[0]) + "点钟后转小雨,其后阴"
    elif "clear weather, cloudy after" in des and "tomorrow afternoon, increasing cloudiness" in des:
        des = "晴,明天傍晚" + str(re.findall(r"\d+", des)[0]) + "点钟后转多云,其后云渐多"
    elif "overcast, drizzle after" in des and "o'clock tomorrow morning, followed by overcast" in des:
        des = "多云,明天早上" + str(re.findall(r"\d+", des)[0]) + "点后有毛毛雨,随后多云"
    else:
        m = re.compile(r'^[a-zA-Z0-9,\'\s.]*$') # 判断是否为英文字符串
        if m.match(des): # 如果是英文字符串,则翻译成中文
            translator = Translator(from_lang="English", to_lang="chinese")
            des = translator.translate(des)
        else:
            pass
    return des


# 未来2小时提示语转换
def desc_exchange(desc):
    # 用Translator翻译成中文也不是很友好,所以优先人工翻译
    if "No snow in the next hour, go out and play" == desc:
        desc = "未来两小时不会下雪,放心出门吧"
    elif "No rain in the next hour, go out and play" == desc:
        desc = "未来两小时不会下雨,放心出门吧"
    elif "Very windy outside! Don't get blown away" == desc:
        desc = "大风起兮...注意不要被风刮跑"
    elif "Cloudy here but it's raining east" in desc and "km away" in desc:
        desc = "您东边" + str(re.findall(r"\d+", desc)[0]) + "公里外正在下雨哦"
    elif "Cloudy here but it's raining south" in desc and "km away" in desc:
        desc = "这里多云,但南边" + str(re.findall(r"\d+", desc)[0]) + "公里处正在下雨"
    elif desc == "Get some rest indoors; the air isn't clean":
        desc = "空气不太好,在室内休息一下吧"
    else:
        m = re.compile(r'^[a-zA-Z0-9,\'\s.]*$') # 判断是否为英文字符串
        if m.match(desc): # 如果是英文字符串,则翻译成中文
            translator = Translator(from_lang="English", to_lang="chinese")
            desc = translator.translate(desc)
        else:
            pass
    return desc


headers = {
    "Host": "wr.bubr.net:8080",
    "Content-Type": "application/json",
    "Accept-Encoding": "gzip, deflate",
    "Accept": "*/*",
    "User-Agent": "LiveWeather/1.8.3 (iPhone; iOS 11.4.1; Scale/2.00)",
    "AccessToken": "b7ee103a47d274940b1a0cce54abac17",
    "Accept-Language": "zh-Hans-CN;q=1, en-CN;q=0.9, zh-Hant-CN;q=0.8, zh-Hant-HK;q=0.7",
    "Content-Length": "464",
    "Connection": "keep-alive"
}


data = {
    "address": {
        "state": "北京市",
        "temp": -6,
        "zt": "Asia\/Shanghai",
        "longitude": "116.3979471",
        "subLocality": "",
        "latitude": "39.9081726",
        "city": "北京市",
        "maxTemp": 0,
        "skycon": "CLEAR_DAY",
        "country": "中国",
        "isScenic": "false",
        "minTemp": 0
    }
}

# 【实况天气】app的首页请求
url = "http://wr.bubr.net:8080/weather/data2"


# @allure.feature("首页")
def get_weather():
    s = requests.post(url, headers=headers, json=data)
    s = json.loads(s.text)
    # print(str(s) + '\n')

    # 获取接口response返回的相关字段
    waring_info = "无"
    try:
        waring_info = s["data"]["alert"]["Description"]
    except:
        pass
    cur_sky = s["data"]["cur"]["skycon"]
    try:
        cur_aqi = s["data"]["cur"]["aqi"]  # 当前空气质量
    except:
        cur_aqi = -1
    cur_pm25 = s["data"]["cur"]["pm25"]  # PM2.5
    cur_pm10 = s["data"]["cur"]["pm10"]  # PM10
    cur_temperature = s["data"]["cur"]["temperature"] # 当前温度
    cur_humidity = s["data"]["cur"]["humidity"] # 空气湿度
    cur_des = s["data"]["cur"]["des"] # 未来24小时晴
    cur_desc = s["data"]["cur"]["desc"] # 未来两小时不会下雪,放心出门吧
    cur_windSpeed = s["data"]["cur"]["windSpeed"]  # 风速8.9
    cur_windDirection = s["data"]["cur"]["windDirection"] # 风向321.09

    today_infos = s["data"]["daily"]["list"][0]
    tomorrow_infos = s["data"]["daily"]["list"][1]
    today_htemp = today_infos[1] # 今日最高温度
    today_ltemp = today_infos[2] # 今日最低温度
    today_sunset = today_infos[18]  # 今日日出
    today_sunrise = today_infos[19] # 今日日落
    tomorrow_htemp = tomorrow_infos[1]  # 明天最高温度
    tomorrow_ltemp = tomorrow_infos[2]  # 明天最低温度
    tomorrow_aqi = tomorrow_infos[5] # 明天空气
    tomorrow_sunrise = tomorrow_infos[19]  # 明天日出
    tomorrow_sunset = tomorrow_infos[18]  # 明天日落
    tomorrow_down_sky = tomorrow_infos[7] # 明天早上天气
    tomorrow_up_sky = tomorrow_infos[8] # 明天晚上天气

    # 将接口拿到的数据进行进一步格式化处理,以下为最后得到的数据
    cur_sky = sky_exchange_chinese(cur_sky)
    today_ltemp = str(round(today_ltemp))
    today_htemp = str(round(today_htemp))
    cur_windDirection = wind_direction_exchange(cur_windDirection)
    cur_windSpeed = wind_speed_exchange(cur_windSpeed)
    cur_wind_info = cur_windDirection + " " + cur_windSpeed
    cur_aqi = aqi_exchange(int(cur_aqi))
    if cur_aqi == "":
        cur_aqi_info = ""
    else:
        cur_aqi_info = "{0}(PM2.5={1},PM10={2}),".format(cur_aqi, cur_pm25, cur_pm10)
    cur_temperature = round(cur_temperature)
    cur_humidity = str(int(float(cur_humidity) * 100)) + "%"
    cur_des = des_exchange(cur_des)
    cur_desc = desc_exchange(cur_desc)
    tomorrow_sky = daily_sky_exchange(tomorrow_down_sky, tomorrow_up_sky)
    tomorrow_htemp = round(tomorrow_htemp)
    tomorrow_ltemp = round(tomorrow_ltemp)
    tomorrow_aqi = aqi_exchange(int(tomorrow_aqi))
    tomorrow_sunrise = tomorrow_sunrise
    tomorrow_sunset = tomorrow_sunset
    # 公历日期(今天)
    solar_date = date.today()
    # 阴历日期
    lunar_date = solar_to_lunar(solar_date)
    # 今天星期几
    today_weekday = get_week_day(solar_date)

    # 最后要发送的信息内容
    notice_info = "{0}({1} {2}):\n北京今日{3},{4} ~ {5}℃,{6},空气质量{7}当前室外温度{8}℃,湿度{9}。{10};{11}\n预警信号:{12}\n\n预计明天:{13},{14} ~ {15}℃,空气质量{16},日出{17} 日落{18}。".format(
        solar_date, lunar_date, today_weekday, cur_sky, today_ltemp, today_htemp, cur_wind_info, cur_aqi_info, cur_temperature, cur_humidity, cur_des, cur_desc,
        waring_info, tomorrow_sky, tomorrow_ltemp, tomorrow_htemp, tomorrow_aqi, tomorrow_sunrise, tomorrow_sunset)

    # 如果是晚上八点到八点半之间执行的脚步,则写入history.txt
    print("time hours:" + str(int(time.strftime("%H", time.localtime()))))
    print("time minutes:" + str(int(time.strftime("%M", time.localtime()))))
    write_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
    if (int(time.strftime("%H", time.localtime())) == 17) and (int(time.strftime("%M", time.localtime())) < 30):
        print("写入history.txt:" + str(write_time))
        # 如果脚本执行时间在设定的时间段内,则将每日天气的信息保存到history.txt中
        with open("history.txt", "a+") as f:
            row = str(write_time) + "    " + str(solar_date) + "(" + str(lunar_date) + ")" + str(today_weekday) + "    " + str(cur_sky) + "    " + str(today_ltemp) + "    " + str(today_htemp) + "    " + str(cur_windDirection) + "    " + str(cur_windSpeed) + "    " + str(cur_aqi_info[:-1]) + "    " + str(cur_humidity) + "    " + str(today_sunrise) + "    " + str(today_sunset) + "    " + str(waring_info) + "\n"
            f.write(row)
    # 否则,测试数据写入test_zzz.txt,用于调试
    else:
        print("写入test_zzz.txt:" + write_time)
        with open("test_zzz.txt", "a+") as f:
            row = str(write_time) + "    " + str(solar_date) + "(" + str(lunar_date) + ")" + str(today_weekday) + "    " + str(cur_sky) + "    " + str(today_ltemp) + "    " + str(today_htemp) + "    " + str(cur_windDirection) + "    " + str(cur_windSpeed) + "    " + str(cur_aqi_info[:-1]) + "    " + str(cur_humidity) + "    " + str(today_sunrise) + "    " + str(today_sunset) + "    " + str(waring_info) + "\n"
            f.write(row)

    return notice_info

if __name__ == "__main__":
    weather_message = get_weather()
    print(weather_message)
    headers = {"Content-Type": "application/json"}
    data = {
        "msgtype": "text",
        "text": {
            "content": weather_message
        }
    }
    robot_url = "https://oapi.dingtalk.com/robot/send?access_token=xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx # 这里替换为你的钉钉机器人的Webhook
    r = requests.post(url=robot_url, headers=headers, json=data)
    print(r.text) # 查看信息的发送状态

调试运行结果:

设置定时器

1、通过 Jenkins 设置
(1)在 Jenkins 新建一个自由风格的项目

(2)进行项目配置,如下图:

注意:日程表的第一个 H 表示随机,如果要设定为 17 点整,可以把日程表改成:H(0-0) 17 * * *

(3)执行项目构建

(4)点击 #1 进入 “控制台输出” 查看构建的日志 log

(5)在设定的 17 点多打开钉钉中查看发送的消息

2、通过 Mac 的 launchctl 设置任务定时器(这里不阐述)

共收到 1 条回复 时间 点赞

实不相瞒,钉钉有自带机器人报告天气预报的

需要 登录 后方可回复, 如果你还没有账号请点击这里 注册