别再到处找天气预报接口了！这个免费API（JSON格式）我用Python爬虫实测可用

用Python玩转免费天气API从接口调用到数据可视化的完整指南最近在开发个人天气小程序时我几乎翻遍了全网所有的免费天气接口要么限制调用次数要么返回数据格式混乱直到发现这个稳定可靠的JSON格式API。它不仅完全免费而且响应速度快数据结构清晰特别适合个人开发者和小型项目使用。本文将带你从零开始用Python实现完整的天气数据获取、解析和应用流程。1. 环境准备与接口初探在开始之前确保你的开发环境已经安装了Python 3.6版本和requests库。如果没有安装requests可以通过以下命令快速安装pip install requests这个天气API的基本调用格式非常简单只需要在URL末尾加上城市代码即可。例如获取北京天气数据的接口地址是http://t.weather.itboy.net/api/weather/city/101010100其中101010100就是北京的城市代码。我们先来写一个最简单的请求函数测试一下接口import requests def get_weather(city_code): url fhttp://t.weather.itboy.net/api/weather/city/{city_code} response requests.get(url) return response.json() # 测试获取北京天气 beijing_weather get_weather(101010100) print(beijing_weather)运行这段代码你应该能看到返回的JSON数据包含了丰富的天气信息从实时温度到未来几天预报一应俱全。2. 深入解析API返回数据结构理解返回数据的结构对于后续处理至关重要。让我们仔细看看这个API返回的主要字段cityInfo: 包含城市基本信息如城市名称、ID等data: 核心天气数据shidu: 湿度百分比pm25: PM2.5数值wendu: 当前温度(摄氏度)forecast: 未来15天预报列表date: 日期high/low: 最高/最低温度type: 天气类型(晴、雨等)fx/fengli: 风向/风力下面是一个典型返回数据的简化示例{ cityInfo: { city: 北京市, cityId: 101010100 }, data: { shidu: 32%, pm25: 35, wendu: 26, forecast: [ { date: 2023-07-20, high: 高温 30℃, low: 低温 22℃, type: 晴, fengxiang: 南风, fengli: 3级 } ] } }3. 构建健壮的天气查询工具基础的请求功能实现后我们需要考虑实际应用中的各种异常情况。以下是几个常见的处理点网络请求超时添加合理的超时设置城市代码不存在处理404或其他错误响应API限流添加适当的重试机制数据解析错误验证JSON格式和关键字段改进后的完整代码如下import requests import time from typing import Dict, Optional class WeatherAPI: def __init__(self, max_retries3, timeout5): self.base_url http://t.weather.itboy.net/api/weather/city/ self.max_retries max_retries self.timeout timeout def get_weather(self, city_code: str) - Optional[Dict]: url f{self.base_url}{city_code} for attempt in range(self.max_retries): try: response requests.get(url, timeoutself.timeout) response.raise_for_status() data response.json() # 验证必要字段是否存在 if not all(key in data for key in [cityInfo, data]): raise ValueError(Invalid API response structure) return data except requests.exceptions.RequestException as e: print(fAttempt {attempt 1} failed: {str(e)}) if attempt self.max_retries - 1: return None time.sleep(1) # 简单的退避策略 return None # 使用示例 weather_api WeatherAPI() result weather_api.get_weather(101010100) if result: print(f当前温度: {result[data][wendu]}℃) else: print(获取天气信息失败)4. 城市代码管理与自动补全手动查找和输入城市代码很不方便我们可以构建一个本地城市代码数据库。原始数据中的城市代码是JSON格式我们可以将其保存为本地文件import json # 保存城市代码到本地 city_codes { 城市代码: [ { 省: 北京, 市: [ {市名: 北京, 编码: 101010100}, # 其他城市... ] } # 其他省份... ] } with open(city_codes.json, w, encodingutf-8) as f: json.dump(city_codes, f, ensure_asciiFalse, indent2)然后创建一个城市代码查询工具class CityCodeFinder: def __init__(self, data_filecity_codes.json): with open(data_file, encodingutf-8) as f: self.city_data json.load(f)[城市代码] def find_code(self, city_name: str) - Optional[str]: for province in self.city_data: for city in province[市]: if city[市名] city_name: return city[编码] return None def search(self, keyword: str) - List[Dict]: results [] for province in self.city_data: for city in province[市]: if keyword in city[市名]: results.append({ province: province[省], city: city[市名], code: city[编码] }) return results # 使用示例 finder CityCodeFinder() print(finder.find_code(北京)) # 输出: 101010100 print(finder.search(海)) # 搜索包含海字的城市5. 数据可视化与实用功能扩展获取到天气数据后我们可以进行各种有趣的可视化和实用功能开发。以下是几个可能的扩展方向5.1 温度趋势图表使用matplotlib绘制未来几天温度变化曲线import matplotlib.pyplot as plt from datetime import datetime def plot_temperature_forecast(weather_data): dates [] highs [] lows [] for day in weather_data[data][forecast][:7]: # 取未来7天数据 dates.append(day[date]) highs.append(int(day[high].split( )[1].replace(℃, ))) lows.append(int(day[low].split( )[1].replace(℃, ))) plt.figure(figsize(10, 6)) plt.plot(dates, highs, label最高温度, markero) plt.plot(dates, lows, label最低温度, markero) plt.fill_between(dates, highs, lows, alpha0.1) plt.title(f{weather_data[cityInfo][city]}未来7天温度预报) plt.xlabel(日期) plt.ylabel(温度(℃)) plt.legend() plt.grid(True) plt.xticks(rotation45) plt.tight_layout() plt.show() # 使用示例 weather_data weather_api.get_weather(101010100) if weather_data: plot_temperature_forecast(weather_data)5.2 天气预警通知我们可以编写一个简单的天气预警系统当出现极端天气时发送通知def check_weather_alert(weather_data): alerts [] today weather_data[data][forecast][0] # 检查高温预警 high_temp int(today[high].split( )[1].replace(℃, )) if high_temp 35: alerts.append(f高温预警: 今日最高温度{high_temp}℃) # 检查降雨 if 雨 in today[type]: alerts.append(f降雨预警: 今日天气{today[type]}) # 检查大风 if 风 in today[type] or any(x in today[fengli] for x in [4级, 5级]): alerts.append(f大风预警: {today[fengxiang]}{today[fengli]}) return alerts # 使用示例 alerts check_weather_alert(weather_data) if alerts: print(天气预警:) for alert in alerts: print(f- {alert})5.3 将天气数据存入数据库对于需要历史天气数据的应用我们可以将获取的数据保存到SQLite数据库import sqlite3 from contextlib import contextmanager contextmanager def weather_db_connection(db_fileweather.db): conn sqlite3.connect(db_file) try: yield conn finally: conn.close() def init_weather_db(): with weather_db_connection() as conn: conn.execute( CREATE TABLE IF NOT EXISTS weather_records ( id INTEGER PRIMARY KEY AUTOINCREMENT, city_code TEXT NOT NULL, city_name TEXT NOT NULL, temperature INTEGER, humidity TEXT, pm25 INTEGER, weather_type TEXT, record_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ) def save_weather_record(weather_data): if not weather_data: return record { city_code: weather_data[cityInfo][cityId], city_name: weather_data[cityInfo][city], temperature: weather_data[data][wendu], humidity: weather_data[data][shidu], pm25: weather_data[data][pm25], weather_type: weather_data[data][forecast][0][type] } with weather_db_connection() as conn: conn.execute( INSERT INTO weather_records (city_code, city_name, temperature, humidity, pm25, weather_type) VALUES (:city_code, :city_name, :temperature, :humidity, :pm25, :weather_type) , record) # 使用示例 init_weather_db() save_weather_record(weather_data)6. 构建完整的天气查询命令行工具将上述功能整合我们可以创建一个功能完善的命令行天气查询工具import argparse def main(): parser argparse.ArgumentParser(description命令行天气查询工具) subparsers parser.add_subparsers(destcommand, requiredTrue) # 查询天气命令 query_parser subparsers.add_parser(query, help查询城市天气) query_parser.add_argument(city, help城市名称) # 搜索城市命令 search_parser subparsers.add_parser(search, help搜索城市代码) search_parser.add_argument(keyword, help搜索关键词) args parser.parse_args() finder CityCodeFinder() api WeatherAPI() if args.command query: city_code finder.find_code(args.city) if not city_code: print(f找不到城市: {args.city}) return weather api.get_weather(city_code) if weather: print(f\n{weather[cityInfo][city]}天气情况:) print(f当前温度: {weather[data][wendu]}℃) print(f湿度: {weather[data][shidu]}) print(fPM2.5: {weather[data][pm25]}) print(\n今日预报:) today weather[data][forecast][0] print(f{today[date]} {today[type]}) print(f温度: {today[low]} ~ {today[high]}) print(f风向: {today[fengxiang]} {today[fengli]}) elif args.command search: results finder.search(args.keyword) if results: print(\n搜索结果:) for city in results: print(f{city[province]} {city[city]}: {city[code]}) else: print(没有找到匹配的城市) if __name__ __main__: main()使用示例python weather_tool.py query 北京 python weather_tool.py search 海7. 性能优化与最佳实践在实际项目中我们还需要考虑一些性能优化和最佳实践缓存机制天气数据不需要实时更新可以添加缓存减少API调用异步请求使用aiohttp实现异步请求提高效率配置管理将API地址、超时设置等提取到配置文件中日志记录添加详细的日志记录方便调试单元测试编写测试用例确保核心功能稳定缓存实现示例from functools import lru_cache import time class CachedWeatherAPI(WeatherAPI): lru_cache(maxsize100) def get_weather(self, city_code: str, expiry3600) - Optional[Dict]: # 简单实现缓存1小时 return super().get_weather(city_code) # 使用示例 cached_api CachedWeatherAPI() # 第一次调用会请求API weather1 cached_api.get_weather(101010100) # 第二次调用会直接返回缓存结果 weather2 cached_api.get_weather(101010100)异步请求示例import aiohttp import asyncio class AsyncWeatherAPI: def __init__(self, max_retries3, timeout5): self.base_url http://t.weather.itboy.net/api/weather/city/ self.max_retries max_retries self.timeout aiohttp.ClientTimeout(totaltimeout) async def get_weather(self, city_code: str) - Optional[Dict]: url f{self.base_url}{city_code} async with aiohttp.ClientSession(timeoutself.timeout) as session: for attempt in range(self.max_retries): try: async with session.get(url) as response: response.raise_for_status() data await response.json() if not all(key in data for key in [cityInfo, data]): raise ValueError(Invalid API response structure) return data except Exception as e: print(fAttempt {attempt 1} failed: {str(e)}) if attempt self.max_retries - 1: return None await asyncio.sleep(1) return None # 使用示例 async def main(): api AsyncWeatherAPI() weather await api.get_weather(101010100) print(weather) asyncio.run(main())通过本文介绍的方法你可以轻松地将这个免费天气API集成到你的各种项目中无论是开发微信小程序、个人网站还是制作自动化天气通知脚本。这个API的稳定性和数据完整性在实际使用中表现相当出色完全能够满足个人开发者和小型项目的需求。