如何从Visual Crossing Weather API获取逐小时天气数据并存入MySQL

问题分析 #

原代码存在以下关键问题，导致无法正常获取逐小时数据：

• API响应解析错误：调用response.json()后又使用json.loads(response.content)，会引发类型错误（response.json()已将响应转为字典，无content属性）
• 请求逻辑不完整：仅在跨日期请求时发送API请求，单日请求分支未处理数据
• 数据遍历逻辑错误：未正确访问API返回结构中的days和hours层级
• 数据库操作缺失：cursor初始化错误，无完整SQL执行语句

修正后的完整代码 #

import mysql.connector
import requests
from datetime import datetime

# 封装数据库连接，避免重复代码
def get_db_conn():
    return mysql.connector.connect(
        host="localhost",
        user="root",
        passwd="qwww",
        database="weather_information"
    )

BaseURL = 'https://weather.visualcrossing.com/VisualCrossingWebServices/rest/services/timeline/'
GeoUser = "xxx"
API_KEY = "wwww"  # 统一管理API密钥

# 验证城市有效性
while True:
    try:
        Location = input("输入城市名称：").strip()
        Geo_params = {"q": Location, "maxRows": 5, "username": GeoUser}
        response = requests.get('http://api.geonames.org/searchJSON', params=Geo_params, verify=True)
        Locateinfo = response.json()
        if Locateinfo["totalResultsCount"] == 0:
            print("无效城市，请重新输入")
        else:
            print("城市有效！")
            break
    except Exception as e:
        print(f"连接错误：{str(e)}")

# 获取起始日期
while True:
    try:
        SDate_String = input("输入起始日期（格式：yyyy/mm/dd）：").strip()
        StartDate = datetime.strptime(SDate_String, '%Y/%m/%d')
        break
    except ValueError:
        print("日期格式错误，请重新输入")

# 获取结束日期
while True:
    try:
        EDate_String = input("输入结束日期（格式：yyyy/mm/dd）：").strip()
        EndDate = datetime.strptime(EDate_String, '%Y/%m/%d')
        print('加载中...')
        break
    except ValueError:
        print("日期格式错误，请重新输入")

# 格式化日期为API要求的YYYY-MM-DD格式
start_date = StartDate.strftime('%Y-%m-%d')
end_date = EndDate.strftime('%Y-%m-%d')

# 构建API请求
params = {
    "unitGroup": "metric",
    "key": API_KEY,
    "contentType": "json"
}
request_url = f"{BaseURL}{Location}/{start_date}/{end_date}"
print(f"请求URL：{request_url}")

# 发送请求并解析数据
try:
    response = requests.get(request_url, params=params)
    response.raise_for_status()  # 捕获HTTP请求错误
    weather_data = response.json()
except requests.exceptions.RequestException as e:
    print(f"API请求失败：{str(e)}")
    exit()

# 处理数据并写入数据库
db = get_db_conn()
cursor = db.cursor()

# 确保目标表存在（可根据实际需求调整字段）
create_table_sql = """
CREATE TABLE IF NOT EXISTS hourly_weather (
    id INT AUTO_INCREMENT PRIMARY KEY,
    location VARCHAR(255) NOT NULL,
    record_date DATE NOT NULL,
    record_hour TIME NOT NULL,
    weather_condition TEXT,
    uv_index INT,
    temperature DECIMAL(5,2),
    sunrise TIME,
    sunset TIME
)
"""
cursor.execute(create_table_sql)

# 遍历每日数据，再处理逐小时信息
for day in weather_data['days']:
    # 提取当日日出日落
    sunrise_time = datetime.strptime(day['sunrise'], '%H:%M:%S').time()
    sunset_time = datetime.strptime(day['sunset'], '%H:%M:%S').time()
    current_date = datetime.strptime(day['datetime'], '%Y-%m-%d').date()
    
    # 遍历逐小时数据
    for hour_item in day['hours']:
        hour_time = datetime.strptime(hour_item['datetime'], '%H:%M:%S').time()
        condition = hour_item['conditions']
        uv_index = hour_item['uvindex']
        temp = hour_item['temp']
        
        # 插入数据到数据库
        insert_sql = """
        INSERT INTO hourly_weather (location, record_date, record_hour, weather_condition, uv_index, temperature, sunrise, sunset)
        VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
        """
        data_values = (
            weather_data['resolvedAddress'],
            current_date,
            hour_time,
            condition,
            uv_index,
            temp,
            sunrise_time,
            sunset_time
        )
        cursor.execute(insert_sql, data_values)

# 提交事务并关闭连接
db.commit()
cursor.close()
db.close()

print("逐小时天气数据已成功保存到数据库！")

关键修正说明 #

• 修复数据解析：移除错误的json.loads(response.content)，直接使用response.json()解析API响应
• 统一日期处理：用strftime直接生成API要求的日期格式，简化代码逻辑
• 完善请求逻辑：无论单日还是跨日期请求，都执行完整的API调用和数据处理流程
• 正确遍历层级：先遍历weather_data['days']数组，再遍历每个day下的hours数组，获取逐小时数据
• 补全数据库操作：添加表创建语句、完整插入SQL，修复cursor初始化错误，确保数据正常写入
• 增强错误处理：添加HTTP请求异常捕获，避免程序意外崩溃

内容的提问来源于stack exchange，提问作者Biani