人工智能-项目实践-灾害预警-基于和风天气API的Workflow，支持灾害预警、AQI、今日天气、三日内天气查询 需自行配

#!/usr/bin/env python # -*- encoding: utf-8 -*- # @Time : 2020/08/10 15:49:04 # @Update : 2020/08/12 01:10:54 # @Author : Leon/Arthals # @File : HeWeather.py # @Contact : [email protected] # @Software: Visual Studio Code import sys import os from datetime import datetime, timedelta from workflow import Workflow3 reload(sys) sys.setdefaultencoding('utf8') # ☉☀☂𐑺☾ # 定义查询天气函数 def query_weather(wf, query_location='', query_adm=''): import requests # 获取APIKey api_key = os.getenv('api_key') # 判断是否含有一级、二级参数，如果有，获取输入参数；如果没有，获取环境参数 if query_location != '': location = query_location if query_adm == '': adm = '' elif query_adm != '': adm = query_adm else: location = os.getenv('location') adm = os.getenv('adm') # 状态码字典，用于提示错误信息 error_dic = { '200': '请求成功', '204': '请求成功，但你查询的地区暂时没有你需要的数据。', '400': '请求错误，可能包含错误的请求参数或缺少必选的请求参数。', '401': '认证失败，可能使用了错误的KEY、数字签名错误、KEY的类型错误（如使用SDK的KEY去访问Web API）。', '402': '超过访问次数或余额不足以支持继续访问服务，你可以充值、升级访问量或等待访问量重置。', '403': '无访问权限，可能是绑定的PackageName、BundleID、域名IP地址不一致，或者是需要额外付费的数据。', '404': '查询的数据或地区不存在。', '429': '超过限定的QPM（每分钟访问次数），请参考QPM说明', '500': '无响应或超时，接口服务异常请联系我们' } # 判断APIKey是否未设置 if api_key == '': wf.add_item(title=u'APIKey未设置', subtitle=u'请设置api_key的环境变量', quicklookurl=u'https://dev.heweather.com/') return # 获取城市ID geo_url = 'https://geoapi.heweather.net/v2/city/lookup?' # 提前获取一下预设区域的时区，保证后期时间校正正确 r = requests.get(geo_url, params={ 'location': os.getenv('location'), 'adm': os.getenv('adm'), 'key': api_key }) geo_info = r.json() local_utc = geo_info['location'][0]['utcOffset'] r = requests.get(geo_url, params={ 'location': location, 'adm': adm, 'key': api_key }) geo_info = r.json() # 判断是否正常返回数据 if geo_info['status'] != '200': failure_title = '地理位置请求异常,错误状态码：' + geo_info['status'] failure_sub_title = error_dic[geo_info['status']] wf.add_item(failure_title, failure_sub_title) return # 保存城市信息变量 geo_num = geo_info['location'][0]['id'] # 城市ID，数字识别码 geo_name = geo_info['location'][0]['name'] # 城市名称 geo_adm1 = geo_info['location'][0]['adm1'] # 城市一级区划（省级） geo_adm2 = geo_info['location'][0]['adm2'] # 城市二级区划（市级） geo_country = geo_info['location'][0]['country'] # 城市所在国家 # 判断是否为中国所属，如果不处于中国境内，输出信息中加入国家信息 if geo_country == '中国': geo_c1 = '' else: geo_c1 = '· {} '.format(geo_country) # 判断是否是直辖市 if geo_adm1 == geo_adm2: geo_a1 = '' else: geo_a1 = '· {} '.format(geo_adm1) # 判断是否具体到了具体区县 if geo_name == geo_adm2: geo_a2 = '' else: geo_a2 = '· {} '.format(geo_adm2) API_title = u'【和风天气 · {}{}{}{}】'.format(geo_name, geo_c1, geo_a1, geo_a2) API_sub_title = u' Code by Leon/Arthals，API by HeWeather，各项具体内容可以选中条目同时按住⇧\ 预览' wf.add_item(API_title, API_sub_title) # 获取数据 base_url = 'https://devapi.heweather.net/v7/{want_type}/{want_time}?' # 判断是否为中国境内，如果不是不请求灾害预警和AQI数据 if geo_country == '中国': # 获取灾害预警 warn_url = base_url.format(want_type='warning', want_time='now') r = requests.get(warn_url, params={ 'location': geo_num, 'key': api_key }) warn_info = r.json() # 判断是否正常返回数据 if warn_info['code'] != '200': failure_title = '灾害预警请求异常,错误状态码：' + warn_info['code'] failure_sub_title = error_dic[warn_info['code']] wf.add_item(failure_title, failure_sub_title) return # 判断预警信息是否为空，如果不是则执行格式化输出 if warn_info['warning']: warning_list = [] warn_link = warn_info['fxLink'] # 灾害预警自适应网页 # 获取全部预警信息，加入warning_list for i in range(len(warn_info['warning'])): if warn_info['warning'][i]['status'] == 'active': warn_info_title = warn_info['warning'][i][ 'typeName'] + warn_info['warning'][i]['level'] + '预警' warning_list.append(warn_info_title) # 列表去重 warning_list = list(set(warning_list)) # 如果只有一条预警，直接输出 if len(warning_list) == 1: warn_title = u'【{}】{}'.format(geo_name, warning_list[0]) # 如果有两条及以上的预警，以'&'分隔后输出 else: warning_all = '&'.join(warning_list) warn_title = u'【{}】{}'.format(geo_name, warning_all) # 对时间进行切片操作，格式化输出 warn_pubTime_list = [] warn_pubTime_list.append(warn_info['updateTime'][:-6]) warn_pubTime_list.append(warn_info['updateTime'][-6:]) pubTime = str( datetime.strptime( warn_pubTime_list[0], '%Y-%m-%dT%H:%M')) + ' UTC' + warn_pubTime_list[1] warn_sub_title = u' ↻ {} 蓝色<黄色<橙色<红色 具体内容请按住⇧预览'.format( pubTime) warn_icon_path = './res/icon-warn/warn.png' wf.add_item(warn_title, warn_sub_title, icon=warn_icon_path, quicklookurl=warn_link) # 获取 AQI aqi_url = base_url.format(want_type='air', want_time='now') r = requests.get(aqi_url, params={'location': geo_num, 'key': api_key}) aqi_info = r.json() # 判断是否正常返回数据 if aqi_info['code'] != '200': failure_title = 'AQI请求异常,错误状态码：' + aqi_info['code'] failure_sub_title = error_dic[aqi_info['code']] wf.add_item(failure_title, failure_sub_title) return # 保存AQI信息 aqi_link = aqi_info['fxLink'] # AQI自适应网页 aqi_num = int(aqi_info['now']['aqi']) # AQI程度描述 aqi_state = aqi_info['now']['category'] # AQI程度描述 # 获取AQI程度对应等级，以调整icon路径，副标题提示 if 0 <= int(aqi_num) <= 50: aqi_state_en = '1' aqi_sub_sug = '空气质量令人满意，基本无空气污染' elif 50 < int(aqi_num) <= 100: aqi_state_en = '2' aqi_sub_sug = '空气质量可接