伏羲天气预报模型Python爬虫实战：气象数据自动化采集与处理

最近在做一个农业气象相关的项目，需要自动化获取和处理全国多个站点的气象数据，然后喂给伏羲天气预报模型做预测。整个过程从数据抓取到清洗，再到格式转换，踩了不少坑，也总结了一套还算好用的方法。今天就把这个实战流程整理出来，如果你也在做类似的气象数据分析，或者想学习如何用Python爬虫配合AI模型处理时序数据，这篇应该能帮到你。

我们的目标是：写一个脚本，能自动从公开的气象网站抓取数据，把乱七八糟的原始数据整理成伏羲模型认识的格式，最后还能把预测结果存下来。整个过程全自动，每天跑一次就行。

1. 环境准备与数据源确认

开始写代码之前，有两件事得先搞定：一是把Python环境搭好，二是确定去哪里抓数据。

1.1 Python环境与必要库

你电脑上得有Python，建议用3.8或以上的版本。然后，我们需要安装几个关键的Python库。打开你的命令行终端，执行下面的命令：

pip install requests beautifulsoup4 pandas numpy

简单解释一下这几个库是干什么的：

• requests：用来向气象网站发送请求，获取网页数据，这是爬虫的基础。
• beautifulsoup4：江湖人称“美味汤”，专门用来解析HTML网页，从一堆标签里把我们需要的气温、湿度这些数据“捞”出来。
• pandas 和 numpy：数据处理的黄金搭档。pandas的DataFrame就像一张超级智能的Excel表格，整理数据特别方便；numpy则提供高效的数值计算。

如果你的项目后期需要更复杂的分析，可能还会用到scikit-learn或matplotlib，这里我们先以完成核心流程为主。

1.2 寻找可靠的气象数据源

公开的气象数据源其实不少，但稳定性和数据格式千差万别。这里我列举几个常用的，你可以根据需求选择：

• 政府气象部门网站：比如中国气象局下属的数据服务网站。这类数据权威性高，但可能需要注册，且API可能有访问频率限制。
• 开源气象数据平台：像OpenWeatherMap、Weatherbit这些平台提供免费的API（通常有每日调用次数限制），返回结构化的JSON数据，用起来最省心。
• 大学或研究机构：国内外一些大学的气象系或研究机构会公开部分观测站数据。

重要提示：在编写爬虫时，务必遵守网站的robots.txt协议，并合理设置请求间隔（例如每次请求后time.sleep(2)），避免对目标服务器造成压力。对于API接口，请严格按照其文档说明使用。

为了教程的通用性，我们假设从一个模拟的、结构简单的公开气象数据页面进行抓取。真实场景中，你需要将代码中的URL替换成实际的数据源地址。

2. 第一步：编写爬虫抓取原始数据

数据源确定了，我们就开始动手写爬虫。这一步的目标是拿到最原始的网页数据。

2.1 分析网页结构与定位数据

爬虫的第一步永远是“人肉观察”。用浏览器打开目标数据页面，按下F12打开开发者工具，查看网页的HTML结构。我们需要找到包含气象数据（如日期、温度、风速）的HTML标签和它们的属性（比如class或id）。

假设我们发现需要的数据在一个<table>表格里，每一行<tr>代表一天，每个单元格<td>里是具体的数据。

2.2 爬虫代码实现

下面是一个基础的爬虫脚本，它完成了请求网页、解析HTML、提取表格数据的过程。

import requests
from bs4 import BeautifulSoup
import pandas as pd
import time

def fetch_weather_data(url):
    """
    从指定URL抓取气象数据
    """
    headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'  # 模拟浏览器访问
    }
    
    try:
        print(f"正在抓取数据: {url}")
        response = requests.get(url, headers=headers, timeout=10)
        response.raise_for_status()  # 检查请求是否成功
        response.encoding = 'utf-8'  # 设置编码，防止乱码
    except requests.RequestException as e:
        print(f"网络请求失败: {e}")
        return None
    
    # 使用BeautifulSoup解析HTML
    soup = BeautifulSoup(response.text, 'html.parser')
    
    # 假设数据在第一个表格（table）里，这里需要根据实际网页结构调整
    table = soup.find('table')
    if not table:
        print("未在页面中找到表格数据。")
        return None
    
    # 提取表头和数据行
    data_rows = []
    # 假设第一行是表头
    headers = [th.get_text(strip=True) for th in table.find('tr').find_all(['th', 'td'])]
    
    for row in table.find_all('tr')[1:]:  # 跳过表头行
        cells = row.find_all(['td', 'th'])
        row_data = [cell.get_text(strip=True) for cell in cells]
        if row_data:  # 避免空行
            data_rows.append(row_data)
    
    # 将数据转换为pandas DataFrame
    df_raw = pd.DataFrame(data_rows, columns=headers)
    print(f"成功抓取 {len(df_raw)} 条数据。")
    return df_raw

# 示例：使用一个模拟的测试URL（实际使用时请替换）
test_url = "https://example-weather-site.com/data"  # 请替换为真实URL
raw_data = fetch_weather_data(test_url)

if raw_data is not None:
    print("原始数据前5行预览：")
    print(raw_data.head())
    # 可以先保存一份原始数据，以备查验
    raw_data.to_csv('raw_weather_data.csv', index=False, encoding='utf-8-sig')

运行这段代码，如果一切顺利，你就能在控制台看到抓取到的数据预览，并且当前目录下会生成一个raw_weather_data.csv文件。不过，这时候的数据往往很“毛糙”，不能直接使用。

3. 第二步：数据清洗与格式化

从网上抓下来的数据，经常会有缺失值、格式不一致、单位不统一等问题。这一步就是当“数据保洁”，把乱七八糟的数据收拾干净。

3.1 常见的脏数据问题及处理

我们的清洗脚本主要处理以下几个典型问题：

def clean_weather_data(df):
    """
    清洗和预处理气象数据DataFrame
    """
    if df is None or df.empty:
        return None
    
    df_clean = df.copy()
    
    # 1. 处理列名：去除空格，统一格式
    df_clean.columns = [col.strip().replace(' ', '_').lower() for col in df_clean.columns]
    print("标准化后的列名：", df_clean.columns.tolist())
    
    # 2. 处理缺失值：这里以数值列为例，用前后值的平均值填充（可根据实际情况选择策略）
    numeric_cols = df_clean.select_dtypes(include=['number']).columns
    for col in numeric_cols:
        if df_clean[col].isnull().any():
            # 使用该列前后非空值的平均值进行填充
            df_clean[col] = df_clean[col].fillna(df_clean[col].interpolate())
            print(f"已填充列 '{col}' 的缺失值。")
    
    # 3. 转换数据类型：确保日期、数值等类型正确
    # 假设有'date'列，将其转换为datetime类型
    if 'date' in df_clean.columns:
        df_clean['date'] = pd.to_datetime(df_clean['date'], errors='coerce')
        # 删除转换失败的日期行（可选）
        df_clean = df_clean.dropna(subset=['date'])
    
    # 4. 处理异常值：简单的基于标准差（Z-score）的方法
    from scipy import stats
    for col in numeric_cols:
        # 计算Z-score，绝对值大于3的通常视为极端异常值
        z_scores = stats.zscore(df_clean[col].dropna())
        abs_z_scores = abs(z_scores)
        if (abs_z_scores > 3).any():
            # 这里选择用中位数替换异常值，而非直接删除，以保持数据连续性
            median_val = df_clean[col].median()
            df_clean.loc[abs_z_scores > 3, col] = median_val
            print(f"已处理列 '{col}' 中的异常值。")
    
    # 5. 去除完全重复的行
    initial_len = len(df_clean)
    df_clean = df_clean.drop_duplicates()
    if len(df_clean) < initial_len:
        print(f"删除了 {initial_len - len(df_clean)} 条重复记录。")
    
    # 重置索引
    df_clean.reset_index(drop=True, inplace=True)
    
    print(f"数据清洗完成。清洗后数据形状: {df_clean.shape}")
    return df_clean

# 应用清洗函数
cleaned_data = clean_weather_data(raw_data)
if cleaned_data is not None:
    cleaned_data.to_csv('cleaned_weather_data.csv', index=False, encoding='utf-8-sig')
    print("\n清洗后数据预览：")
    print(cleaned_data.head())

3.2 转换为模型输入格式

伏羲这类天气预报模型，通常期望输入的是规整的时序数据。常见的要求包括：

• 时间列作为索引或单独的一列，且必须是连续的。
• 特征列（如温度、气压、湿度）需要是数值型。
• 不能有缺失值。

下面的代码将清洗后的数据转换为一个更规整的、适合模型读取的格式：

def format_for_model(df, date_col='date', feature_cols=None):
    """
    将数据格式化为适合时序预测模型的格式。
    """
    if df is None or df.empty:
        return None
    
    df_model = df.copy()
    
    # 确保日期列是索引（很多时序模型要求）
    if date_col in df_model.columns:
        df_model.set_index(date_col, inplace=True)
    
    # 按时间排序
    df_model.sort_index(inplace=True)
    
    # 如果未指定特征列，默认使用所有数值列
    if feature_cols is None:
        feature_cols = df_model.select_dtypes(include=['number']).columns.tolist()
    
    # 确保只保留需要的特征列
    df_model = df_model[feature_cols]
    
    # 再次检查并确保没有缺失值（对于模型输入至关重要）
    if df_model.isnull().any().any():
        print("警告：格式化后的数据仍存在缺失值，将进行最终填充。")
        df_model = df_model.fillna(method='ffill').fillna(method='bfill')  # 前向后向填充
    
    print(f"模型输入数据格式：{df_model.shape}， 时间范围：{df_model.index.min()} 至 {df_model.index.max()}")
    return df_model

# 假设我们关心的特征列是这些（根据你的数据调整）
selected_features = ['temperature', 'humidity', 'wind_speed', 'pressure']
model_ready_data = format_for_model(cleaned_data, feature_cols=selected_features)

if model_ready_data is not None:
    model_ready_data.to_csv('model_ready_weather_data.csv')
    print("\n模型就绪数据预览：")
    print(model_ready_data.head())

现在，我们得到了一个干净的、格式规整的DataFrame，它的索引是时间，列是我们选定的气象特征。这份数据已经可以准备喂给伏羲模型进行预测了。

4. 第三步：集成伏羲模型进行预测

数据准备好了，就到了最核心的一步——使用伏羲天气预报模型进行预测。这里我以调用一个假设的模型API为例，展示如何将数据处理流程与模型调用串联起来。

4.1 准备模型输入与调用

假设伏羲模型提供了一个API接口，它接收一个JSON格式的数据，包含历史时序特征，并返回未来一段时间的预测结果。

import json

def call_fuxi_prediction_api(model_input_df, api_endpoint, forecast_hours=24):
    """
    将处理好的数据发送给伏羲模型API，获取预测结果。
    注意：这是一个示例函数，实际API参数和调用方式需根据伏羲模型的官方文档调整。
    """
    # 1. 将DataFrame转换为模型API要求的格式（例如JSON列表）
    # 这里假设API需要每个时间点的特征值作为一个字典
    historical_data = model_input_df.reset_index().to_dict('records')
    
    request_payload = {
        "historical_data": historical_data,
        "forecast_hours": forecast_hours,
        # 可能还有其他参数，如地理位置ID、模型版本等
    }
    
    headers = {
        'Content-Type': 'application/json',
        # 如果API需要认证，在这里添加Token
        # 'Authorization': 'Bearer YOUR_API_TOKEN'
    }
    
    try:
        print(f"正在向模型API发送请求，预测未来{forecast_hours}小时...")
        response = requests.post(api_endpoint, data=json.dumps(request_payload), headers=headers, timeout=30)
        response.raise_for_status()
        
        prediction_result = response.json()
        print("模型预测调用成功！")
        return prediction_result
        
    except requests.RequestException as e:
        print(f"模型API调用失败: {e}")
        # 在实际应用中，这里可以加入重试逻辑或降级方案
        return None

# 示例：假设的API端点（请替换为真实地址）
# API_ENDPOINT = "https://api.fuxi-weather.com/v1/predict"
# 由于是示例，我们这里模拟一个成功的返回
print("【模拟调用】模型接收到数据，正在计算...")
# 模拟生成一些预测数据
last_time = model_ready_data.index[-1]
import numpy as np
future_times = pd.date_range(start=last_time + pd.Timedelta(hours=1), periods=24, freq='H')
simulated_predictions = {
    "forecast_timestamps": [t.isoformat() for t in future_times],
    "forecast_temperature": list(np.random.uniform(15, 25, 24)),
    "forecast_humidity": list(np.random.uniform(40, 80, 24)),
    "status": "success"
}

4.2 解析与保存预测结果

拿到模型的返回结果后，我们需要将其解析成容易查看和使用的格式，比如再次转成pandas的DataFrame。

def parse_and_save_prediction(prediction_json, output_path='weather_forecast.csv'):
    """
    解析模型预测结果并保存为CSV文件。
    """
    if prediction_json is None or prediction_json.get('status') != 'success':
        print("无效的预测结果。")
        return None
    
    try:
        # 根据API返回的实际结构解析数据
        # 这里根据上面的模拟结构编写
        forecast_df = pd.DataFrame({
            'forecast_time': pd.to_datetime(prediction_json['forecast_timestamps']),
            'predicted_temperature': prediction_json['forecast_temperature'],
            'predicted_humidity': prediction_json['forecast_humidity']
        })
        
        forecast_df.set_index('forecast_time', inplace=True)
        forecast_df.to_csv(output_path, encoding='utf-8-sig')
        print(f"预测结果已保存至: {output_path}")
        print(forecast_df.head())
        return forecast_df
    except KeyError as e:
        print(f"解析预测结果时出错，键错误: {e}")
        return None

# 解析并保存模拟的预测结果
forecast_data = parse_and_save_prediction(simulated_predictions)

5. 第四步：构建自动化流程

单个步骤跑通后，我们就可以把它们串起来，写一个主函数，实现“一键”从抓取到预测的全流程。再加上定时任务，就能实现真正的自动化。

5.1 整合完整脚本

def run_full_weather_pipeline(data_url, feature_cols, model_api_endpoint=None, forecast_hours=24):
    """
    运行完整的气象数据爬取、清洗、预测流水线。
    """
    print("="*50)
    print("开始执行气象数据自动化管道")
    print("="*50)
    
    # 步骤1: 抓取数据
    raw_df = fetch_weather_data(data_url)
    if raw_df is None:
        print("管道终止：数据抓取失败。")
        return
    
    # 步骤2: 清洗数据
    cleaned_df = clean_weather_data(raw_df)
    
    # 步骤3: 格式化数据
    model_df = format_for_model(cleaned_df, feature_cols=feature_cols)
    
    if model_df is None or model_df.empty:
        print("管道终止：数据格式化后为空。")
        return
    
    # 步骤4: 调用模型预测 (如果提供了API端点)
    if model_api_endpoint:
        prediction = call_fuxi_prediction_api(model_df, model_api_endpoint, forecast_hours)
        if prediction:
            parse_and_save_prediction(prediction, 'latest_forecast.csv')
        else:
            print("模型预测步骤未执行或失败。")
    else:
        print("未提供模型API，管道执行至数据准备阶段。")
        # 即使不预测，也保存好处理完的数据
        model_df.to_csv('latest_processed_weather_data.csv')
    
    print("="*50)
    print("气象数据自动化管道执行完毕！")
    print("="*50)

# 配置参数并运行
if __name__ == "__main__":
    # 你的实际数据源URL
    DATA_SOURCE_URL = "https://your-real-weather-data-source.com"
    # 你选择的气象特征列
    SELECTED_FEATURES = ['temperature', 'humidity', 'wind_speed', 'pressure']
    # 伏羲模型API地址（如果暂时没有，设为None）
    FUXI_API_ENDPOINT = None  # 替换为真实API地址，例如 "https://api.example.com/predict"
    
    run_full_weather_pipeline(DATA_SOURCE_URL, SELECTED_FEATURES, FUXI_API_ENDPOINT)

5.2 设置定时任务

想让这个脚本每天自动运行？在Linux服务器上，可以使用cron；在Windows上，可以使用“任务计划程序”。

Linux/Mac 示例 (Cron)：

1. 打开终端，输入 crontab -e 编辑定时任务。
2. 添加一行，例如下面这行表示每天凌晨2点执行一次我们的脚本（需要替换为你的实际Python和脚本路径）：

   0 2 * * * /usr/bin/python3 /path/to/your/weather_pipeline.py >> /path/to/logfile.log 2>&1
   

Windows 示例：

1. 搜索并打开“任务计划程序”。
2. 创建基本任务，设置每日触发，时间设为凌晨2点。
3. 操作为“启动程序”，程序或脚本填写你的python.exe完整路径，参数填写你的脚本weather_pipeline.py的完整路径。

6. 总结

走完这一整套流程，你会发现，把Python爬虫和伏羲这样的天气预报模型结合起来，并没有想象中那么复杂。核心就是三个环节：稳定地拿到数据、耐心地洗干净数据、正确地喂给模型。本教程提供的代码是一个坚实的起点，你可以根据自己实际的数据源和模型接口进行调整。

在实际项目中，你可能还会遇到更多细节问题，比如应对网站反爬机制、处理更复杂的数据缺失模式、优化模型API的调用效率等等。但只要你把握住“数据获取-清洗-转换-应用”这个核心链路，大部分问题都能找到解决思路。建议你先用一个站点的数据把整个流程跑通，然后再考虑扩展多站点、增加错误重试机制、加入结果可视化等功能，一步步搭建起属于你自己的、可靠的气象数据智能分析系统。

---

获取更多AI镜像

想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。