Python自动化批量下载ECWMF和GFS最新预报数据脚本

一、白嫖EC和GFS预报数据

EC的openData部分公开了一部分预报数据，作为普通用户只能访问这些免费预报数据，具体位置在这
在这里插入图片描述

可以发现，由于是Open Data，我们只能获得临近四天的预报结果，虽然时间较短，但是我们可以通过每天都执行脚本储存在本地，日积月累就可以存储完整的EC预报数据

同样的，NCEP的GFS数据也是只提供近几天的数据，因此也是需要一直爬取才能获取连续的预报数据

在这里插入图片描述
注意：该网站的gfs数据是包含41层数据，如果想要选择变量和高度层的话请访问这里

1.1 脚本核心思路

本质上是使用Python的request库通过拼接好的url发送get请求，然后每天执行一次，就可以爬取最新的数据保存下来
我们点进去下载目录中点击一个文件下载，按F12调试中可以从“网络”中抓包下载时候请求的url
详细看一下这个url，可以发现是由几个变量拼接起来的，包括：预报时长（forecast_hour）, 起报时间（start_hour），基本url（base_url），分辨率(resolution)当然这几个名字是我自己取的，跟后面代码中的相对应
这样只需要我们在脚本中指定好这几个参数，之后按一定规律拼接url，并发送get请求，就可以相应下载

1.2 脚本编写中的面向对象思路（模板方法设计模式）

在编写过程中发现两种数据下载的逻辑完全相同，只有使用的base_url和文件名拼接规则不同
将核心功能和公共功能封装为基类
两个继承了基类的具体的实现类只需要根据不同的数据下载需求将filename和url的拼接规则函数重写即可
启示：在python中用子类不实现该方法就会NotImplementedError的方式来实现接口，将不同的逻辑用接口的形式让子类实现，实现多态
后来想起来好像无意中运用了一个之前见过的设计模式，模板方法设计模式，基类规定好算法骨架，部分会随着实现类变化的部分的实现方式延迟到子类中实现

1.3 基类ForecastDownloader类

ForecastDownloader作为基类，其核心方法是start和download，是控制整个下载流程的函数，同时有两个抽象方法需要子类实现，这样做的目的是面对不同的下载需求，其实只有一小部分在变化，将拼接方法延迟到子类中实现就好了

import requests
import os
from datetime import datetime, timedelta
from urllib.parse import urljoin
import sys
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '../utils')))
from utils.logger import Loggerclass ForecastDownloader:"""下载天气预报数据的基类。这个类定义了天气预报数据下载器的基础功能，包括文件的下载和保存。子类应该实现 `construct_filename` 和 `construct_download_url` 方法来提供具体的文件名构造和下载 URL 构造。属性:forecast_hour (list): 预报小时列表，例如 [0, 6, 12, 18]。start_hour (str): 预报开始小时，例如 '00'。save_dir (str): 文件保存目录。base_url (str): 基础下载 URL。previous_days (int): 需要下载的前几天数据。log_savepath (str): 日志记录器路径。方法:start(): 开始下载数据，遍历指定的日期范围。download(time): 根据时间下载数据。construct_filename(time, hour): 构造文件名。子类需实现。construct_download_url(time, filename): 构造下载 URL。子类需实现。"""def __init__(self, forecast_hour, start_hour, save_dir, base_url, previous_days, log_savepath):self.save_dir = save_dirself.forecast_hour = forecast_hourself.start_hour = start_hourself.base_url = base_urlself.previous_days = previous_daysself.logger = Logger.setup_logger(log_savepath)if not os.path.exists(save_dir):os.makedirs(save_dir)def start(self):"""开始下载数据，遍历指定的日期范围。这个方法会从当前日期向过去的日期循环，并调用 `download` 方法来下载数据。"""today = datetime.now().date()for i in range(self.previous_days):date = today - timedelta(days=i)self.logger.info(str(date))time_str = f'{date.year}{date.month:02}{date.day:02}'self.download(time_str)def download(self, time):"""根据时间下载数据。这个方法会下载指定时间的数据，并保存到指定目录。参数:time (str): 数据的时间字符串，例如 '20240721'。"""for hour in self.forecast_hour:filename = self.construct_filename(time, hour)if not os.path.exists(os.path.join(self.save_dir, time, filename)):url = self.construct_download_url(time, filename)try:response = requests.get(url, verify=False)if response.status_code == 200:self.logger.info(f"Downloading {url}")day_dir = os.path.join(self.save_dir, time)if not os.path.exists(day_dir):os.makedirs(day_dir)save_path = os.path.join(self.save_dir, time, filename)with open(save_path, 'wb') as f:f.write(response.content)self.logger.info(f"Finished downloading {url}")else:self.logger.error(f"Failed to download {url}: Status code {response.status_code}")except Exception as e:self.logger.error(f"Error downloading {url}: {e}")else:self.logger.info('data exists')def construct_filename(self, time, hour):"""构造文件名。这个方法应该在子类中实现，用于构造特定格式的文件名。参数:time (str): 数据的时间字符串。hour (int): 预报小时。返回:str: 构造出的文件名。抛出:NotImplementedError: 子类必须实现此方法。"""raise NotImplementedError("This method should be overridden by subclasses")def construct_download_url(self, time, filename):"""构造下载 URL。这个方法应该在子类中实现，用于构造特定格式的下载 URL。参数:time (str): 数据的时间字符串。filename (str): 文件名。返回:str: 构造出的下载 URL。抛出:NotImplementedError: 子类必须实现此方法。"""raise NotImplementedError("This method should be overridden by subclasses")

1.4 子类EcForecastDownloader类

继承自ForecastDownloader类，只需要把baserl改一下和重写两个拼接文件名和url的方法，其他逻辑与基类完全一致，这样就轻松实现了代码复用

class EcForecastDownloader(ForecastDownloader):def __init__(self, forecast_hour, start_hour, save_dir, previous_days, log_savepath='../logs/ecLog.log'):base_url = "https://data.ecmwf.int/forecasts/"super().__init__(forecast_hour, start_hour, save_dir, base_url, previous_days, log_savepath)def construct_filename(self, time, forecast_hour):return f'{time}{self.start_hour}0000-{forecast_hour}h-oper-fc.grib2'def construct_download_url(self, time, filename):return urljoin(self.base_url, f"{time}/{self.start_hour}z/ifs/0p25/oper/{filename}")

1.5 子类NecpForecastDownloader类

class NecpForecastDownloader(ForecastDownloader):def __init__(self, forecast_hour, start_hour, save_dir, previous_days, log_savepath='../logs/ncepLog.log'):base_url = "https://nomads.ncep.noaa.gov/pub/data/nccf/com/gfs/prod/"super().__init__(forecast_hour, start_hour, save_dir, base_url, previous_days, log_savepath)def construct_filename(self, time, forecast_hour):return f'gfs.t{self.start_hour}z.pgrb2.0p25.f0{forecast_hour}'def construct_download_url(self, time, filename):return f'{self.base_url}/gfs.{time}/{self.start_hour}/atmos/{filename}'

1.6 日志类

import logging
import osclass Logger:@staticmethoddef setup_logger(log_file):if not os.path.exists('../logs'):os.makedirs('../logs')logger = logging.getLogger()logger.setLevel(logging.INFO)# 文件处理器file_handler = logging.FileHandler(log_file)file_handler.setLevel(logging.INFO)file_formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')file_handler.setFormatter(file_formatter)logger.addHandler(file_handler)# 控制台处理器console_handler = logging.StreamHandler()console_handler.setLevel(logging.INFO)console_formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')console_handler.setFormatter(console_formatter)logger.addHandler(console_handler)return logger

1.7 控制部分

将下载类和具体的使用逻辑分离，只要指定好构造函数的几个参量，之后间隔一定的小时数之后一直执行下载代码，这样就可以实现不间断更新数据

from forecast_downloader import EcForecastDownloader, NecpForecastDownloader
import time
def download_ec(interval_hour):save_dir = '../download_res/ec'D = EcForecastDownloader([24, 48, 72], '00', save_dir, 4)while True:D.start()time.sleep(interval_hour * 60 * 60)def download_ncep(interval_hour):save_dir = '../download_res/ncep'D = NecpForecastDownloader([24, 48, 72], '00', save_dir, 4)while True:D.start()time.sleep(interval_hour * 60 * 60)if __name__ == '__main__':download_ec(24)