Python使用免费天气API,获取全球任意地区的天气情况

大家好，又见面了，我是你们的朋友全栈君。如果您正在找激活码,请点击查看最新教程,关注关注公众号 “全栈程序员社区” 获取激活教程,可能之前旧版本教程已经失效.最新Idea2022.1教程亲测有效,一键激活。

Jetbrains全系列IDE稳定放心使用

需求背景:

公司是做外贸服装的,在亚马逊平台上有多个地区店铺运营,运营人员需要参考地区的天气情况,上新的服装.所以需要能够获取全球任意地区的天气情况.还需要预测未来10-15天的天气情况.

选型API:

天气API中有大把免费的api,如:国内的心知天气,国际的雅虎,还有今天的主角:wunderground

最终选择了wunderground,原因:1,需求是全球任意地区的(国内API请求国外地区需要收费才能访问), 2.wunderground提供是信息最全,最丰富的天气api.雅虎提供的天气API信息非常之简略.

直入主题:

官方API文档

这里的免费api只是说测试账号每天有500次的免费请求,要是公司需求大的话,那么就需要付费了.官网价格

准备工作,你需要在官网注册一个账号,然后随意打开一个API的文档, 你会见到

http://api.wunderground.com/api/Your_Key/conditions/q/CA/San_Francisco.json

当中Your_Key的位置有一串key.请保管记住.

python代码:

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 18年3月6日 下午1:53
# @Author  : dongyouyuan
# @email   : 15099977712@163.com
# @File    : weatherApi.py
# @Software: PyCharm
#
# 通过全球天气预报API:http://api.wunderground.com(信息最多最全) 来获取信息
import requests
import json
import datetime
import time
import operator
def timeStamp_to_dataTime(timestamp, format='%Y-%m-%d'):
"""
时间戳转换成格式化的时间
:param timestamp: 时间戳
:param format: 格式
:return:
"""
if timestamp is None or not timestamp:
return ''
return datetime.datetime.fromtimestamp(int(timestamp)).strftime(format)
def dataTime_to_timeStamp(data_tiem, format='%Y-%m-%d %H:%M'):
"""
格式化时间格式转换成时间戳
:param data_tiem: 格式化时间
:param format: 格式
:return:
"""
if not data_tiem:
return 0
else:
try:
time_array = time.strptime(data_tiem, format)
time_stamp = int(time.mktime(time_array))
return time_stamp
except Exception as error_msg:
print(error_msg)
return 0
def get_recent_10_days(date):
"""
获取前10天的时间格式 20180306
:return: 返回列表
"""
date_list = list()
for i in range(1, 11):
tmp_date = (datetime.datetime.strptime(date, "%Y%m%d") - datetime.timedelta(days=i)).strftime('%Y%m%d')
date_list.append(tmp_date)
return date_list
def sorted_for_list_dict(list_dict, key, reverse=False):
"""
为列表排序，按照元素中dict的key
:param list_dict:
:param key:
:return:
"""
return sorted(list_dict, key=operator.itemgetter(key), reverse=reverse)
class Weather(object):
def __init__(self):
self.search_url = 'http://autocomplete.wunderground.com/aq'
self._url = 'http://api.wunderground.com/api/'
self._key = ''  # Your_ key
self.url = self._url + self._key + '/'
self.timeout = 10
def search(self, query):
"""
搜索地名,返回结果列表
:param query:
:return:
"""
params = {"query": query}
headers = {"Content-Type": "application/json"}
r = requests.get(url=self.search_url, params=params, headers=headers, timeout=self.timeout)
if r.status_code == 200:
return r.json().get('RESULTS', None)
else:
print('请求失败,失败代码:{}'.format(r.status_code))
return None
def get_today(self, zmw):
"""
根据zmw码获取今天的天气信息
:param zmw:
:return:
"""
# temp 温度
# humidity 湿度
# weather 天气情况
# wind 风
# dewpoint 露点
# UV 紫外线强度
# pressure 压力
url = self.url + 'conditions' + '/q/zmw:{}.json'.format(zmw)
headers = {"Content-Type": "application/json"}
r = requests.get(url=url, headers=headers, timeout=self.timeout)
if r.status_code == 200:
# print(r.json())
return r.json().get('current_observation')
else:
print('请求失败,失败代码:{}'.format(r.status_code))
return None
def get_hourly_today(self, zmw):
"""
根据zmw码获取今天每小时的天气信息
:param zmw:
:return:
"""
url = self.url + 'hourly' + '/q/zmw:{}.json'.format(zmw)
headers = {"Content-Type": "application/json"}
r = requests.get(url=url, headers=headers, timeout=self.timeout)
if r.status_code == 200:
# print(r.json())
return r.json().get('hourly_forecast')
else:
print('请求失败,失败代码:{}'.format(r.status_code))
return None
def _get_forecast_10day(self, zmw):
"""
根据zmw码获取未来10天的天气预报
:param zmw:
:return:
"""
# snow 降雪 (白天,晚上,全天)
# wind 风 (最大,最小)
# humidity 湿度(最大,最小,平均)
# conditions 天气情况 icon icon_url
# qpf 降水量(白天,晚上,全天)
# low, high 温度(最高,最低)
# pop 降水概率
url = self.url + 'forecast10day' + '/q/zmw:{}.json'.format(zmw)
headers = {"Content-Type": "application/json"}
r = requests.get(url=url, headers=headers, timeout=self.timeout)
if r.status_code == 200:
return sorted_for_list_dict(r.json().get('forecast').get('simpleforecast').get('forecastday'), 'period')
else:
print('请求失败,失败代码:{}'.format(r.status_code))
return None
def get_forecast_10day(self, zmw):
"""
根据zmw码获取未来10天的天气预报(并且整理数据输出)
:param zmw:
:return:
"""
f_day_list = self._get_forecast_10day(zmw)
if f_day_list:
result_list = list()
for f_day in f_day_list:
forecast_day = dict()
date_time = "{}{}{}".format(f_day['date']['year'], f_day['date']['month'], f_day['date']['day'])
time_stamp = dataTime_to_timeStamp(date_time, format='%Y%m%d')
date_format = timeStamp_to_dataTime(time_stamp, format='%Y%m%d')
# 当天日期
forecast_day['date_format'] = date_format
# 风
wind = {
'ave': f_day.get('avewind', {}),
'max': f_day.get('maxwind', {})
}
forecast_day['wind'] = wind
# 雪
snow = {
'day': f_day.get('snow_day', {}),
'night': f_day.get('snow_night', {}),
'allday': f_day.get('snow_allday', {})
}
forecast_day['snow'] = snow
# 湿度
humidity = {
'min': f_day.get('minhumidity', 0),
'max': f_day.get('maxhumidity', 0),
'ave': f_day.get('avehumidity', 0)
}
forecast_day['humidity'] = humidity
# 天气情况
conditions = {
'icon_url': f_day.get('icon_url', ''),
'icon': f_day.get('icon', ''),
'text': f_day.get('conditions', ''),
'skyicon': f_day.get('skyicon', '')
}
forecast_day['conditions'] = conditions
# 气压
qpf = {
'day': f_day['qpf_day'],
'night': f_day['qpf_night'],
'allday': f_day['qpf_allday']
}
forecast_day['qpf'] = qpf
# 气温
temp = {
'low': f_day.get('low', {'celsius': '0', 'fahrenheit': '0'}),
'high': f_day.get('high', {'celsius': '0', 'fahrenheit': '0'})
}
forecast_day['temp'] = temp
forecast_day['pop'] = f_day.get('pop', 0)
forecast_day['date'] = f_day.get('date', {})
result_list.append(forecast_day)
return result_list
else:
return None
def get_hourly_10day(self, zmw):
"""
根据zmw码获取未来10天,每小时的天气预报
:param zmw:
:return:
"""
url = self.url + 'hourly10day' + '/q/zmw:{}.json'.format(zmw)
headers = {"Content-Type": "application/json"}
r = requests.get(url=url, headers=headers, timeout=self.timeout)
if r.status_code == 200:
# print(r.json())
return r.json()
else:
print('请求失败,失败代码:{}'.format(r.status_code))
return None
def _get_history(self, zmw, date):
"""
根据zmw码获取历史天气
:param zmw:
:param dates: 查询时间 20170608
:return:
"""
url = self.url + 'history_{}'.format(date) + '/q/zmw:{}.json'.format(zmw)
headers = {"Content-Type": "application/json"}
r = requests.get(url=url, headers=headers, timeout=self.timeout)
if r.status_code == 200:
# print(r.json())
return r.json()
else:
print('请求失败,失败代码:{}'.format(r.status_code))
return None
def get_history(self, zmw, date):
"""
根据zmw码获取历史天气(并且整理数据输出)
:param zmw:
:param dates: 查询时间 20170608
:return:
"""
h_day = self._get_history(zmw, date)
if h_day['history']['dailysummary']:
h_day = h_day['history']['dailysummary'][0]
history_day = dict()
# 当天日期
history_day['date_format'] = date
# 风
wind = {
'ave': {'dir': h_day.get('meanwdire', ''), 'degrees': h_day.get('meanwdird', ''),
'kph': h_day.get('meanwindspdi', ''), 'mph': h_day.get('meanwindspdm', '')},
'max': {'dir': '', 'degrees': '0', 'kph': '0', 'mph': '0'}
}
history_day['wind'] = wind
# 雪
snow = {
'day': {'cm': h_day.get('snowdepthm', ''), 'in': h_day.get('snowdepthi', '')},
'night': {'cm': '', 'in': ''},
'allday': {'cm': '', 'in': ''}
}
history_day['snow'] = snow
# 湿度
humidity = {
'min': h_day.get('minhumidity', ''),
'max': h_day.get('maxhumidity', ''),
'ave': h_day.get('humidity', '')
}
history_day['humidity'] = humidity
# 天气情况
conditions = {
'icon_url': '',
'icon': '',
'text': '',
'skyicon': '',
'tornado': h_day.get('tornado', '0'),
'fog': h_day.get('fog', '0'),
'rain': h_day.get('rain', '0'),
'snow': h_day.get('snow', '0'),
'thunder': h_day.get('thunder', '0')
}
history_day['conditions'] = conditions
# 气压
qpf = {
'day': {'mm': h_day.get('maxpressurem', '0.0'), 'in': h_day.get('maxpressurei', '0.0')},
'night': {'mm': h_day.get('minpressurem', '0.0'), 'in': h_day.get('minpressurei', '0.0')},
'allday': {'mm': h_day.get('meanpressurem', '0.0'), 'in': h_day.get('meanpressurei', '0.0')}
}
history_day['qpf'] = qpf
# 温度
temp = {
'low': {'celsius': h_day.get('mintempm', '0'), 'fahrenheit': h_day.get('mintempi', '0')},
'high': {'celsius': h_day.get('maxtempm', '0'), 'fahrenheit': h_day.get('maxtempi', '0')}
}
history_day['temp'] = temp
# 可见度
visi = {
'min': {'km': h_day.get('minvism', '0.0'), 'mi': h_day.get('minvisi', '0.0')},
'mean': {'km': h_day.get('meanvism', '0.0'), 'mi': h_day.get('meanvisi', '0.0')},
'max': {'km': h_day.get('maxvism', '0.0'), 'mi': h_day.get('maxvisi', '0.0')}
}
history_day['visi'] = visi
# 时间信息
history_day['date'] = h_day.get('date', {})
return history_day
else:
return None
def _get_history_10day(self, zmw, date):
"""
根据zmw码获取10天的历史天气(不包括传入天)
:param zmw:
:param dates: 查询时间 20170608
:return:
"""
date_list = get_recent_10_days(date)
result_list = list()
for history_date in date_list:
result_list.append(self._get_history(zmw=zmw, date=history_date))
return result_list
def get_history_10day(self, zmw, date):
"""
根据zmw码获取10天的历史天气(不包括传入天)(并且整理输出)
:param zmw:
:param dates: 查询时间 20170608
:return:
"""
date_list = get_recent_10_days(date)
result_list = list()
for history_date in date_list:
result_list.append(self.get_history(zmw=zmw, date=history_date))
return result_list
if __name__ == '__main__':
weather = Weather()
search_result = weather.search('Paris')
if search_result:
zmw = search_result[0].get('zmw')
if zmw:
print('*'*100)
# 获取未来10天的天气预告
f_10daty = weather.get_forecast_10day(zmw=zmw)
print(json.dumps(f_10daty, indent=4))
print(len(f_10daty))
# 获取某一天的历史天气
# h_10daty = weather.get_history(zmw=zmw, date='20180306')
# print(json.dumps(h_10daty, indent=4))
# print(json.dumps(h_10daty, indent=4))
# print(len(h_10daty))
# 获取某天前10天的历史天气(不包括传入天)
h_10daty = weather.get_history_10day(zmw=zmw, date='20180306')
print(json.dumps(h_10daty, indent=4))
print(len(h_10daty))

代码当中封装了Weather类,你需要把刚才注册得到的key放到类的key中去.

代码解析:

类中并没有按照官网API(/q/地区/地名.json)的方法去做,因为发现这样得出的结果往往不准确的.而是通过查询你需要的字符,你自己获得到zmw(这是个唯一的编码)

然后再通过此码去查询.然后在方法 get_forecast_10day(),get_history(), get_history_10day()封装了一下,原因是官网的不同api获取的天气信息所叫的名称不一样(变量名)

这样用起来有点麻烦,所以统一整理,封装了一下.