数据下载  |  如何使用Python下载GFS数据 2026 ver

师弟问，之前那个从 UCAR 拖 GFS 的脚本，今年怎么一个文件都下不下来了，是不是网络的问题。

是 NCAR 把数据从 RDA 搬到了 GDEX，顺手把老地址的证书也弄坏了，你改哪个参数都没用，博主上周为这事重写了一遍，两百多行，把体积校验和断点续传一起补上了，扔到超算上挂着就能跑，完整代码放在文末，不想听博主啰嗦的直接往下翻。

01

base_url = "https://data.rda.ucar.edu/d084001/{year}/{date}/gfs.0p25.{date}00.f"

现在的实测结果是，data.rda.ucar.edu 直接报 TLS 证书链不受信任，Windows 底下给的是 schannel: SEC_E_UNTRUSTED_ROOT，服务器回你一个 403。

再老一点的 rda.ucar.edu/data/ds084.1/... 会 301 跳到 gdex.ucar.edu/data/ds084.1/...，跳过去是 404，因为 GDEX 的文件路径跟 RDA 根本不是一套东西，d084001、ds084.1 博主都拼过，全 404，它只认网页界面。

02

好消息在 GDEX 自己的页面上，d084001 现在明写着这份数据在 AWS 上有持续更新的开放副本，并且把它列为首选访问方式。地址长这样

https://noaa-gfs-bdp-pds.s3.amazonaws.com/gfs.YYYYMMDD/HH/atmos/gfs.tHHz.pgrb2.0p25.fFFF

免认证，不用注册，支持 Range。

坑在 atmos 这一层，2021 年 4 月之后 AWS 的目录里必须有它，少一层就是 404，照老文章手工拼地址的人八成会踩这个。所以模板这么写

SOURCES = {# AWS 开放数据镜像，2021-04 至今，免认证"aws": "https://noaa-gfs-bdp-pds.s3.amazonaws.com/""gfs.{ymd}/{hh}/atmos/gfs.t{hh}z.pgrb2.0p25.f{fff}",# NOMADS 实时源，只留最近约 10 天"nomads": "https://nomads.ncep.noaa.gov/pub/data/nccf/com/gfs/prod/""gfs.{ymd}/{hh}/atmos/gfs.t{hh}z.pgrb2.0p25.f{fff}",}defbuild_url(source, ymd, hh, fff):return SOURCES[source].format(ymd=ymd, hh=hh, fff="%03d" % fff)

NOMADS 那条是给最近十天用的，AWS 探不到就自动退过去，两个都探不到，脚本会直接告诉你不是网的问题，是这一天真的没数据。

03

下载之前先问服务器要个头，这一步能省很多事

defhead(url):"""HEAD 探测，返回 (状态码, 文件字节数, 是否支持 Range)。"""    req = Request(url, method="HEAD", headers={"User-Agent": USER_AGENT})with urlopen(req, timeout=CONNECT_TIMEOUT) as resp:        length = resp.headers.get("Content-Length")return (resp.status,                int(length) if length elseNone,                (resp.headers.get("Accept-Ranges") or"").lower() == "bytes")

拿到的 Content-Length 有两个用处，一个是判断这个文件到底存不存在，另一个是后面校验的标准值，比你原来那句“大于 400 MB 就算成功”靠谱得多。

顺手留了个 --dry-run，只探不下，把一天 43 个时次的可用性和体积一次列出来，决定要不要下之前先跑它。

04

大文件最怕下到 90% 断掉，老脚本一断就得从零开始。这条用 .part 临时文件加 Range 来解决

part_path = out_path + ".part"have = os.path.getsize(part_path) if os.path.exists(part_path) else0headers = {"User-Agent": USER_AGENT}if have:    headers["Range"] = "bytes=%d-" % have   # 从断点位置接着要mode = "ab"if have else"wb"with urlopen(Request(url, headers=headers), timeout=CONNECT_TIMEOUT) as resp:if have and resp.status == 200:   # 服务端不认 Range，只能从头写，不能追加        have, mode = 0, "wb"

必须是追加模式，因为 Range 请求拿回来的是片段，用 wb 会把前面下过的全抹掉。

那个 resp.status == 200 的判断别省，有些镜像不认 Range，你带了它照样回 200 加整个文件，这时候还按追加写，文件就废了，而且体积还偏偏对得上，你查一晚上都查不出来。

下完先躺在 .part 里，等校验过了才改名

os.replace(part_path, out_path)   # 校验通过才落成正式文件名

这样你中途 Ctrl-C，目录里不会混着半截的 grib2，下次重跑自动接着下，博主实测断在 17.74%，重来就是从 17.74% 开始走的。

体积对了也可能是坏文件。国内网络有时候给你一个 200 的 HTML 报错页，或者 302 到某个认证页，长度对不上还好说，有些代理会补齐长度，这时候只能看文件头

deflooks_like_grib2(path, nbytes=16):"""前 4 字节是 'GRIB'，第 8 字节是版本号 2。"""with open(path, "rb") as fh:        head_bytes = fh.read(nbytes)return len(head_bytes) >= 8and head_bytes[:4] == b"GRIB"and head_bytes[7] == 2

整个 fetch 套了三次重试，老脚本最大的毛病就是没超时，网一挂就永久卡在那儿不动，你在超算上等一晚上，第二天起来发现进程还活着，字节数还是那么多。

05

用法就一行

python download_gfs.py 2026072400                    # f000~f126 每 3 小时，43 个文件python download_gfs.py 2026072400 --dry-run          # 只探不下载，先看有多少可用python download_gfs.py 2026072400 --hours 0-72:3     # 只要前 72 小时python download_gfs.py 20260724 00 --outdir ./gfs724 # 日期和时次分开写也行

博主拿 2026-07-24 00Z 实测了一遍，43 个时次全都在，总共 21.5 GB，f000 是 490.4 MB，f126 是 508.9 MB。

文件名还是老规矩 gfs.0p25.2026072400.f000.grib2，你 namelist.wps 里的 prefix 不用动一行。

博主一般这么干，先 --dry-run 看一眼，确认 43 个全在，再 nohup 挂上去睡觉，第二天早上看最后那行汇总，新增 43、跳过 0、失败 0 就收工，要是失败几个，直接重跑，.part 在那儿等着。

太早的日期别硬试，AWS 从 2021 年 4 月起才有，NOMADS 只留最近十天，两头都够不着的时候脚本会明确告诉你，别在一个根本不存在的时间点上耗。

完整代码在这儿，存成 download_gfs.py 直接就能跑，参数说明都写在开头的注释里。

#!/usr/bin/env python3# -*- coding: utf-8 -*-"""下载 NCEP GFS 0.25° GRIB2 预报场（供 WRF/WPS 的 ungrib 使用）。用法::    python download_gfs.py 2026072400                    # 起报时刻 YYYYMMDDHH，默认 f000~f126 每 3h    python download_gfs.py 20260724 00 --hours 0-72:3    # 日期与时次也可分开写    python download_gfs.py 2026072400 --dry-run          # 只探测可用性与文件大小，不下载    python download_gfs.py 2026072400 --outdir ./gfs724  # 指定输出目录为什么原脚本会失败------------------1. ``https://data.rda.ucar.edu/d084001/...`` 已经不能用了：TLS 证书链不受信任   （``schannel: SEC_E_UNTRUSTED_ROOT``），跳过证书校验后返回 **403**。2. RDA 的旧地址 ``https://rda.ucar.edu/data/ds084.1/...`` 现在 **301 跳转**到   ``https://gdex.ucar.edu/data/ds084.1/...``，而该路径在 GDEX 上返回 **404**   —— 文件路径结构已经变了，脚本里拼出来的地址不再是有效地址。3. NCAR/GDEX 的 d084001 页面现在明确推荐改用 AWS 上的开放数据副本   （``noaa-gfs-bdp-pds``，2021-04 至今，免认证、支持 HTTP Range 断点续传）。   本脚本即以它为首选数据源。与旧脚本相比的改进------------------* 数据源改为 AWS Open Data，并保留 NOMADS 实时源做自动回退；* 用服务端 ``Content-Length`` + GRIB2 魔数双重校验，替代"大于 400 MB 就算成功"的粗判；* 支持断点续传（``.part`` 临时文件 + Range 请求），大文件断了不用从头下；* 每个文件最多重试 3 次（指数退避），带连接/读取超时，不会再静默卡死；* 输出文件名仍沿用 ``gfs.0p25.<YYYYMMDDHH>.f<FFF>.grib2``，下游 ungrib 无需改动。"""import argparseimport osimport sysimport timefrom urllib.error import HTTPError, URLErrorfrom urllib.request import Request, urlopenCHUNK = 1024 * 1024# 1 MiBUSER_AGENT = "Mozilla/5.0 (compatible; gfs-fetch/2.0)"CONNECT_TIMEOUT = 60# 秒RETRIES = 3MIN_FILE_SIZE_MB = 400# 兜底性的最小体积检查（f000 约 490 MB，f126 约 510 MB）# 数据源模板：{ymd}=YYYYMMDD, {hh}=UTC 时次, {fff}=预报时效三位数SOURCES = {"aws": "https://noaa-gfs-bdp-pds.s3.amazonaws.com/""gfs.{ymd}/{hh}/atmos/gfs.t{hh}z.pgrb2.0p25.f{fff}","nomads": "https://nomads.ncep.noaa.gov/pub/data/nccf/com/gfs/prod/""gfs.{ymd}/{hh}/atmos/gfs.t{hh}z.pgrb2.0p25.f{fff}",}defbuild_url(source, ymd, hh, fff):return SOURCES[source].format(ymd=ymd, hh=hh, fff="%03d" % fff)defhead(url):"""HEAD 探测，返回 (状态码, 文件字节数或 None, 是否支持 Range)。"""    req = Request(url, method="HEAD", headers={"User-Agent": USER_AGENT})with urlopen(req, timeout=CONNECT_TIMEOUT) as resp:        length = resp.headers.get("Content-Length")return (resp.status,                int(length) if length elseNone,                (resp.headers.get("Accept-Ranges") or"").lower() == "bytes")defprobe(url):"""探测某个地址是否可用，返回 (可用, 字节数, 是否支持Range)。"""try:        status, length, ranges = head(url)return status == 200, length, rangesexcept HTTPError as exc:return exc.code in (200, 206), None, Falseexcept (URLError, OSError):returnFalse, None, Falsedeflooks_like_grib2(path, nbytes=16):"""校验文件头是不是 GRIB2（前 4 字节 'GRIB'，第 8 字节版本号 = 2）。"""try:with open(path, "rb") as fh:            head_bytes = fh.read(nbytes)except OSError:returnFalsereturn len(head_bytes) >= 8and head_bytes[:4] == b"GRIB"and head_bytes[7] == 2defis_file_valid(path, expected_size=None):"""文件完整性检查：存在 + 体积达标 + 头部是 GRIB2 + （可选）与 Content-Length 完全一致。"""ifnot os.path.exists(path):returnFalse    size = os.path.getsize(path)if size < MIN_FILE_SIZE_MB * 1024 * 1024:returnFalseif expected_size isnotNoneand size != expected_size:returnFalsereturn looks_like_grib2(path)defhuman(nbytes):for unit in ("B", "KB", "MB", "GB"):if nbytes < 1024:return"%.1f %s" % (nbytes, unit)        nbytes /= 1024.0return"%.1f TB" % nbytesdeffetch(url, out_path, expected_size):"""带断点续传与重试的下载，返回 True/False。"""    part_path = out_path + ".part"for attempt in range(1, RETRIES + 1):        have = os.path.getsize(part_path) if os.path.exists(part_path) else0if expected_size isnotNoneand have >= expected_size:            have = 0# 本地残留比服务端还大，直接重下        headers = {"User-Agent": USER_AGENT}if have:            headers["Range"] = "bytes=%d-" % have        mode = "ab"if have else"wb"try:            req = Request(url, headers=headers)with urlopen(req, timeout=CONNECT_TIMEOUT) as resp:# 服务端不支持 Range 时会返回 200，此时必须从头写，不能追加if have and resp.status == 200:                    have, mode = 0, "wb"                total = expected_size or (have + (int(resp.headers.get("Content-Length") or0)) orNone)                done = have                resumed_from = have          # 本次尝试的起点，用于单独计算速率                t0 = time.time()                last = t0with open(part_path, mode) as out:whileTrue:                        chunk = resp.read(CHUNK)ifnot chunk:break                        out.write(chunk)                        done += len(chunk)                        now = time.time()if now - last > 0.5:                            last = now                            speed = (done - resumed_from) / max(now - t0, 1e-6) / 1024 / 1024if total:                                pct = 100.0 * done / total                                sys.stdout.write("\r    %6.2f%%  %s / %s  %5.1f MB/s"                                                 % (pct, human(done), human(total), speed))else:                                sys.stdout.write("\r    %s  %5.1f MB/s" % (human(done), speed))                            sys.stdout.flush()            sys.stdout.write("\n")except (HTTPError, URLError, OSError) as exc:            sys.stdout.write("\n    尝试 %d/%d 失败: %s\n" % (attempt, RETRIES, exc))if attempt < RETRIES:                time.sleep(2 ** attempt)continue        actual = os.path.getsize(part_path)if expected_size isnotNoneand actual != expected_size:            sys.stdout.write("    体积不符（收到 %s，应为 %s），重试\n"                             % (human(actual), human(expected_size)))if attempt < RETRIES:continuereturnFalseifnot looks_like_grib2(part_path):            sys.stdout.write("    文件头不是 GRIB2，重试\n")            os.remove(part_path)if attempt < RETRIES:continuereturnFalse        os.replace(part_path, out_path)   # 校验通过才落成正式文件名returnTrue# 全部重试都失败时保留 .part，下次运行可续传returnFalsedefparse_hours(spec):"""'0-126:3' -> range(0, 127, 3)；'0,6,12' -> [0, 6, 12]"""if":"in spec or"-"in spec:        rng, _, step = spec.partition(":")        start, _, end = rng.partition("-")return list(range(int(start), int(end) + 1, int(step or1)))return [int(x) for x in spec.split(",")]defresolve_source(ymd, hh, fff, requested):"""确定用哪个数据源；requested='auto' 时按 aws -> nomads 依次探测。"""if requested != "auto":return requestedfor name in ("aws", "nomads"):        ok, length, _ = probe(build_url(name, ymd, hh, fff))if ok:return namereturnNonedefmain():    ap = argparse.ArgumentParser(description="下载 NCEP GFS 0.25° GRIB2 预报场")    ap.add_argument("date", help="起报时刻，YYYYMMDD 或 YYYYMMDDHH（UTC）")    ap.add_argument("hour", nargs="?", default=None, help="UTC 时次 00/06/12/18（date 只给了 8 位时使用）")    ap.add_argument("--hours", default="0-126:3", help="预报时效范围，默认 0-126:3")    ap.add_argument("--outdir", default=None, help="输出目录，默认为 起报时刻 字符串")    ap.add_argument("--source", default="auto", choices=["auto", "aws", "nomads"], help="数据源")    ap.add_argument("--dry-run", action="store_true", help="只探测，不下载")    args = ap.parse_args()    date = args.date.strip()if len(date) == 10:        ymd, hh = date[:8], date[8:10]elif len(date) == 8:        ymd, hh = date, (args.hour or"00")else:        ap.error("日期格式应为 YYYYMMDD 或 YYYYMMDDHH")    hours = parse_hours(args.hours)    outdir = args.outdir or (ymd + hh)    stamp = ymd + hh    print("起报时刻 : %s-%s-%s %sZ" % (ymd[:4], ymd[4:6], ymd[6:8], hh))    print("预报时效 : f%03d ~ f%03d，共 %d 个文件" % (hours[0], hours[-1], len(hours)))    print("输出目录 : %s" % os.path.abspath(outdir))    print()    source = resolve_source(ymd, hh, hours[0], args.source)if source isNone:# 两种源都探测不到，通常是：日期超出 NOMADS 保留期(约10天) 或 早于 2021-04（AWS 起始）        print("错误：AWS 与 NOMADS 都没有 %s 这个起报时刻的数据。" % stamp)        print("      注意 AWS 镜像自 2021-04 起，NOMADS 只保留最近约 10 天；")        print("      更早的历史数据需到 https://gdex.ucar.edu/datasets/d084001/ 用网页界面取（需注册）。")return1    print("数据源   : %s" % source)    print()ifnot args.dry_run:        os.makedirs(outdir, exist_ok=True)    ok = skip = fail = 0    total_bytes = 0for fff in hours:        url = build_url(source, ymd, hh, fff)        out_path = os.path.join(outdir, "gfs.0p25.%s.f%03d.grib2" % (stamp, fff))        available, expected, _ = probe(url)ifnot available:            print("f%03d  服务端无此文件，跳过" % fff)            fail += 1continueif expected:            total_bytes += expectedif is_file_valid(out_path, expected):            print("f%03d  已存在且校验通过，跳过（%s）" % (fff, human(os.path.getsize(out_path))))            skip += 1continueif args.dry_run:            print("f%03d  可用，%s" % (fff, human(expected) if expected else"大小未知"))continue        print("f%03d  下载中 … %s" % (fff, human(expected) if expected else""))if fetch(url, out_path, expected):            print("f%03d  完成（%s）" % (fff, human(os.path.getsize(out_path))))            ok += 1else:            print("f%03d  失败，已保留 .part 供下次续传" % fff)            fail += 1    print()    print("汇总：新增 %d，跳过 %d，失败 %d" % (ok, skip, fail))if args.dry_run:        print("预计总下载量：%s" % human(total_bytes))return0if fail == 0else2if __name__ == "__main__":    sys.exit(main())