当前位置：首页 > news >正文

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

news 2026/5/5 22:40:25

用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 = f"http://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_retries=3, timeout=5): 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, timeout=self.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(f"Attempt {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", encoding="utf-8") as f: json.dump(city_codes, f, ensure_ascii=False, indent=2)

然后创建一个城市代码查询工具：

class CityCodeFinder: def __init__(self, data_file="city_codes.json"): with open(data_file, encoding="utf-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="最高温度", marker="o") plt.plot(dates, lows, label="最低温度", marker="o") plt.fill_between(dates, highs, lows, alpha=0.1) plt.title(f"{weather_data['cityInfo']['city']}未来7天温度预报") plt.xlabel("日期") plt.ylabel("温度(℃)") plt.legend() plt.grid(True) plt.xticks(rotation=45) 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_file="weather.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(dest="command", required=True) # 查询天气命令 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(f"PM2.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(maxsize=100) def get_weather(self, city_code: str, expiry=3600) -> 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_retries=3, timeout=5): self.base_url = "http://t.weather.itboy.net/api/weather/city/" self.max_retries = max_retries self.timeout = aiohttp.ClientTimeout(total=timeout) async def get_weather(self, city_code: str) -> Optional[Dict]: url = f"{self.base_url}{city_code}" async with aiohttp.ClientSession(timeout=self.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(f"Attempt {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的稳定性和数据完整性在实际使用中表现相当出色，完全能够满足个人开发者和小型项目的需求。

查看全文

http://www.jsqmd.com/news/759842/