← 返回列表

实时天气查询 API 接口

编写一个简单的 API 客户端，向天气服务提供商发送 HTTP 请求，解析返回的 JSON 数据，并在终端实时显示指定城市的温度与天气状况。

快速预览：

import requests
city = "Shanghai"
api_key = "YOUR_API_KEY"
url = f"http://api.weather.com/v1?q={city}&key={api_key}"
data = requests.get(url).json()
print(f"当前温度: {data['main']['temp']}°C")

应用场景与价值

天气数据在众多应用场景中不可或缺，通过 API 获取实时天气信息可以实现：

  • 智能提醒: 出门前自动提醒是否需要带伞或添衣
  • 农业应用: 根据天气预报优化灌溉和施肥计划
  • 物流优化: 根据天气情况调整运输路线和时间
  • 旅游规划: 为旅行者提供目的地天气信息
  • 智能家居: 根据天气自动调节室内温度和湿度
  • 数据分析: 收集历史天气数据进行气候研究

本案例将教你如何对接第三方天气 API，掌握 RESTful API 调用的核心技能，这些技能可迁移到其他任何 API 集成场景。

基础知识

RESTful API 基础

REST（Representational State Transfer）是一种 Web 服务架构风格：

  • HTTP 方法: GET（获取数据）、POST（提交数据）、PUT（更新）、DELETE（删除）
  • URL 结构: https://api.example.com/v1/resource?param=value
  • 状态码: 200（成功）、400（请求错误）、401（未授权）、404（未找到）、500（服务器错误）
  • 数据格式: 通常使用 JSON 格式传输数据

常用天气 API 服务

服务 特点 免费额度
OpenWeatherMap 全球覆盖，数据详细 60次/分钟
WeatherAPI 简单易用，响应快速 1,000,000次/月
高德天气 国内数据准确 根据账号等级
和风天气 中文友好 1,000次/天

JSON 数据格式

JSON（JavaScript Object Notation）是轻量级的数据交换格式：

{
  "location": {
    "name": "Shanghai",
    "country": "China"
  },
  "current": {
    "temp_c": 18.5,
    "condition": {
      "text": "Partly cloudy"
    }
  }
}

在 Python 中使用 json 模块或 requests 的 .json() 方法解析。

代码详解

让我们深入分析天气 API 调用的核心代码：

import requests

导入 requests 库，用于发送 HTTP 请求。这是 Python 最流行的 HTTP 客户端库。

city = "Shanghai"
api_key = "YOUR_API_KEY"

设置查询参数。api_key 是 API 提供商颁发的认证密钥，需要在官网注册获取。

url = f"http://api.weather.com/v1?q={city}&key={api_key}"

构建 API 请求 URL。使用 f-string 格式化，将城市名和 API 密钥作为查询参数。

data = requests.get(url).json()

发送 GET 请求并解析 JSON 响应。requests.get() 返回响应对象，.json() 方法将 JSON 文本解析为 Python 字典。

print(f"当前温度: {data['main']['temp']}°C")

从解析的字典中提取温度数据并显示。需要根据实际 API 的数据结构调整键名。

完整代码

基础版本：使用 OpenWeatherMap API

import requests

def get_weather(city, api_key):
    """
    获取指定城市的当前天气

    参数:
        city: 城市名称（英文）
        api_key: OpenWeatherMap API 密钥

    返回:
        天气数据字典，如果失败返回 None
    """
    base_url = "https://api.openweathermap.org/data/2.5/weather"

    # 构建请求参数
    params = {
        'q': city,
        'appid': api_key,
        'units': 'metric',  # 使用摄氏度
        'lang': 'zh_cn'     # 中文描述
    }

    try:
        response = requests.get(base_url, params=params, timeout=10)
        response.raise_for_status()  # 检查 HTTP 状态码

        data = response.json()

        # 提取关键信息
        weather_info = {
            '城市': data['name'],
            '国家': data['sys']['country'],
            '温度': f"{data['main']['temp']:.1f}°C",
            '体感温度': f"{data['main']['feels_like']:.1f}°C",
            '天气': data['weather'][0]['description'],
            '湿度': f"{data['main']['humidity']}%",
            '气压': f"{data['main']['pressure']} hPa",
            '风速': f"{data['wind']['speed']} m/s",
            '能见度': f"{data.get('visibility', 0) / 1000:.1f} km"
        }

        return weather_info

    except requests.exceptions.RequestException as e:
        print(f"请求失败: {e}")
        return None
    except KeyError as e:
        print(f"数据解析错误: {e}")
        return None


def display_weather(weather_info):
    """格式化显示天气信息"""
    if not weather_info:
        print("无法获取天气数据")
        return

    print("\n" + "=" * 40)
    print(f"  {weather_info['城市']}, {weather_info['国家']} - 天气预报")
    print("=" * 40)

    for key, value in weather_info.items():
        if key not in ['城市', '国家']:
            print(f"{key:8s}: {value}")

    print("=" * 40 + "\n")


# 使用示例
if __name__ == "__main__":
    # 注意：需要在 https://openweathermap.org/api 注册获取 API 密钥
    API_KEY = "your_api_key_here"

    city_name = input("请输入城市名称（英文）: ") or "Shanghai"
    weather = get_weather(city_name, API_KEY)
    display_weather(weather)

进阶版本：多城市对比查询

import requests
from concurrent.futures import ThreadPoolExecutor

def get_weather_concurrent(cities, api_key):
    """
    并发查询多个城市的天气

    参数:
        cities: 城市列表
        api_key: API 密钥

    返回:
        天气数据列表
    """
    def fetch_single_city(city):
        return city, get_weather(city, api_key)

    # 使用线程池并发请求
    with ThreadPoolExecutor(max_workers=5) as executor:
        results = list(executor.map(lambda c: fetch_single_city(c), cities))

    return results


def compare_cities(cities, api_key):
    """对比多个城市的天气"""
    print("正在查询天气数据...\n")

    results = get_weather_concurrent(cities, api_key)

    print(f"{'城市':<15} {'温度':<10} {'天气':<15} {'湿度':<8}")
    print("-" * 60)

    for city, weather in results:
        if weather:
            print(f"{weather['城市']:<15} {weather['温度']:<10} {weather['天气']:<15} {weather['湿度']:<8}")
        else:
            print(f"{city:<15} {'数据获取失败'}")


# 使用示例
cities = ['Shanghai', 'Beijing', 'Guangzhou', 'Shenzhen', 'Hangzhou']
compare_cities(cities, API_KEY)

高级版本：天气预警和持久化存储

import requests
import json
from datetime import datetime
import sqlite3

class WeatherMonitor:
    """天气监控类"""

    def __init__(self, api_key, db_path='weather.db'):
        self.api_key = api_key
        self.db_path = db_path
        self.init_database()

    def init_database(self):
        """初始化数据库"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS weather_history (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                city TEXT,
                temperature REAL,
                description TEXT,
                humidity INTEGER,
                timestamp DATETIME DEFAULT CURRENT_TIMESTAMP
            )
        ''')
        conn.commit()
        conn.close()

    def fetch_weather(self, city):
        """获取天气数据"""
        return get_weather(city, self.api_key)

    def save_to_database(self, city, weather_data):
        """保存到数据库"""
        if not weather_data:
            return

        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()

        cursor.execute('''
            INSERT INTO weather_history (city, temperature, description, humidity)
            VALUES (?, ?, ?, ?)
        ''', (
            city,
            float(weather_data['温度'].replace('°C', '')),
            weather_data['天气'],
            int(weather_data['湿度'].replace('%', ''))
        ))

        conn.commit()
        conn.close()

    def check_alert(self, weather_data):
        """检查天气预警"""
        alerts = []

        temp = float(weather_data['温度'].replace('°C', ''))
        humidity = int(weather_data['湿度'].replace('%', ''))

        if temp > 35:
            alerts.append("⚠️ 高温预警：气温超过35°C，注意防暑")
        elif temp < 0:
            alerts.append("❄️ 低温预警：气温低于0°C，注意保暖")

        if humidity > 80:
            alerts.append("💧 高湿度预警：湿度超过80%，体感闷热")

        if '雨' in weather_data['天气']:
            alerts.append("☔ 降雨提醒：建议携带雨具")

        return alerts

    def monitor(self, city):
        """监控天气并生成报告"""
        print(f"\n正在监控 {city} 的天气...")

        weather = self.fetch_weather(city)

        if weather:
            display_weather(weather)

            # 保存历史数据
            self.save_to_database(city, weather)

            # 检查预警
            alerts = self.check_alert(weather)
            if alerts:
                print("📢 天气预警:")
                for alert in alerts:
                    print(f"   {alert}")
            else:
                print("✅ 天气状况良好，无需特别注意")
        else:
            print("❌ 获取天气数据失败")

    def get_history(self, city, days=7):
        """获取历史天气数据"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()

        cursor.execute('''
            SELECT temperature, description, humidity, timestamp
            FROM weather_history
            WHERE city = ?
            ORDER BY timestamp DESC
            LIMIT ?
        ''', (city, days * 24))  # 假设每小时记录一次

        rows = cursor.fetchall()
        conn.close()

        return rows


# 使用示例
monitor = WeatherMonitor(API_KEY)

# 监控单个城市
monitor.monitor('Shanghai')

# 查看历史数据
history = monitor.get_history('Shanghai', days=7)
print(f"\n过去7天的天气记录: {len(history)} 条")

命令行交互版本

import requests
import sys

def interactive_weather_app(api_key):
    """交互式天气查询应用"""
    print("=" * 50)
    print("     欢迎使用天气查询系统")
    print("=" * 50)

    while True:
        print("\n请选择操作:")
        print("1. 查询单个城市天气")
        print("2. 对比多个城市天气")
        print("3. 查看天气历史")
        print("4. 退出")

        choice = input("\n请输入选项 (1-4): ").strip()

        if choice == '1':
            city = input("请输入城市名称（英文）: ").strip()
            if city:
                weather = get_weather(city, api_key)
                display_weather(weather)

        elif choice == '2':
            cities_input = input("请输入城市列表（用逗号分隔）: ").strip()
            cities = [c.strip() for c in cities_input.split(',')]
            compare_cities(cities, api_key)

        elif choice == '3':
            city = input("请输入城市名称: ").strip()
            monitor = WeatherMonitor(api_key)
            history = monitor.get_history(city, days=3)

            if history:
                print(f"\n{city} 最近的天气记录:")
                print(f"{'温度':<10} {'天气':<15} {'湿度':<8} {'时间'}")
                print("-" * 60)
                for temp, desc, hum, time in history[:10]:
                    print(f"{temp:.1f}°C    {desc:<15} {hum}%     {time}")
            else:
                print("暂无历史记录")

        elif choice == '4':
            print("\n感谢使用！再见！")
            sys.exit(0)

        else:
            print("无效选项，请重新输入")


# 运行交互式应用
if __name__ == "__main__":
    API_KEY = "your_api_key_here"
    interactive_weather_app(API_KEY)

运行示例

获取 API 密钥

  1. 访问 OpenWeatherMap
  2. 注册账号
  3. 在 API Keys 页面生成免费 API 密钥
  4. 将密钥复制到代码中

安装依赖

uv add requests

执行查询

uv run main.py

预期输出

========================================
  Shanghai, CN - 天气预报
========================================
温度      : 18.5°C
体感温度  : 17.2°C
天气      : 多云
湿度      : 65%
气压      : 1013 hPa
风速      : 3.5 m/s
能见度    : 10.0 km
========================================

📢 天气预警:
   💧 高湿度预警：湿度超过80%，体感闷热

扩展思路

1. 天气预报（未来几天）

def get_forecast(city, api_key, days=5):
    """获取未来几天的天气预报"""
    url = f"https://api.openweathermap.org/data/2.5/forecast"
    params = {
        'q': city,
        'appid': api_key,
        'units': 'metric',
        'cnt': days * 8  # 每3小时一条数据
    }

    response = requests.get(url, params=params)
    data = response.json()

    for item in data['list']:
        date = item['dt_txt']
        temp = item['main']['temp']
        desc = item['weather'][0]['description']
        print(f"{date}: {temp}°C, {desc}")

2. 天气图表可视化

import matplotlib.pyplot as plt

def plot_temperature_trend(history):
    """绘制温度趋势图"""
    temps = [row[0] for row in history]
    times = [row[3] for row in history]

    plt.figure(figsize=(12, 6))
    plt.plot(times, temps, marker='o')
    plt.title('温度趋势')
    plt.xlabel('时间')
    plt.ylabel('温度 (°C)')
    plt.xticks(rotation=45)
    plt.grid(True)
    plt.tight_layout()
    plt.savefig('temperature_trend.png')
    print("图表已保存: temperature_trend.png")

3. 邮件通知异常天气

import smtplib
from email.mime.text import MIMEText

def send_weather_alert(weather_data, alerts, to_email):
    """发送天气预警邮件"""
    if not alerts:
        return

    subject = f"天气预警 - {weather_data['城市']}"
    body = f"当前天气:\n温度: {weather_data['温度']}\n\n预警:\n" + "\n".join(alerts)

    # 调用之前的邮件发送函数
    send_email(subject, body, to_email, ...)

4. 多语言支持

def get_weather_multilang(city, api_key, lang='zh_cn'):
    """支持多语言的天气查询"""
    params = {
        'q': city,
        'appid': api_key,
        'units': 'metric',
        'lang': lang  # en, zh_cn, ja, ko, etc.
    }
    # ... 其余代码

5. 缓存机制

from functools import lru_cache
from datetime import datetime, timedelta

@lru_cache(maxsize=100)
def cached_get_weather(city, api_key, cache_time):
    """带缓存的天气查询（避免频繁请求）"""
    return get_weather(city, api_key)

# 使用时传入当前时间的整点作为 cache_time
current_hour = datetime.now().replace(minute=0, second=0, microsecond=0)
weather = cached_get_weather('Shanghai', API_KEY, current_hour)

6. Webhook 集成

def send_to_webhook(weather_data, webhook_url):
    """发送天气数据到 Webhook（如 Slack、Discord）"""
    payload = {
        "text": f"天气更新: {weather_data['城市']}, {weather_data['温度']}, {weather_data['天气']}"
    }
    requests.post(webhook_url, json=payload)

7. 定位服务

def get_weather_by_coordinates(lat, lon, api_key):
    """根据经纬度查询天气"""
    url = "https://api.openweathermap.org/data/2.5/weather"
    params = {
        'lat': lat,
        'lon': lon,
        'appid': api_key,
        'units': 'metric'
    }
    response = requests.get(url, params=params)
    return response.json()

# 使用示例：查询上海的天气（纬度31.23, 经度121.47）
weather = get_weather_by_coordinates(31.23, 121.47, API_KEY)

注意事项

API 使用规范

  1. 速率限制: 遵守 API 提供商的调用频率限制，避免账号被封禁
  2. 密钥安全: 不要将 API 密钥提交到公开代码仓库，使用环境变量存储
  3. 错误处理: 妥善处理网络错误、超时、无效响应等异常情况
  4. 缓存策略: 同一城市短时间内不要重复请求，使用缓存减少 API 调用

数据准确性

  1. 数据来源: 不同 API 提供商的数据可能有差异
  2. 更新频率: 天气数据通常15-30分钟更新一次
  3. 预报精度: 短期预报（1-3天）较准确，长期预报误差较大

开发建议

  1. 环境变量: 使用 .env 文件存储 API 密钥
  2. 日志记录: 记录 API 调用日志，便于调试和监控
  3. 单元测试: 编写测试用例，模拟 API 响应
  4. 文档阅读: 仔细阅读 API 官方文档，了解所有可用功能

通过掌握天气 API 集成技术，你不仅可以构建实用的天气应用，还能将这些技能应用到对接任何第三方 API 的场景中，为你的项目增添丰富的外部数据支持。