标准下载脚本示例文档

本文档旨在为用户提供标准下载气象数据的脚本示例，帮助您高效、顺利获取所需数据。

1. 文档说明

1.1 步骤概览

获取您的认证令牌：从数据下载页面复制下载链接，并从中提取您的个人认证令牌
选择并配置脚本：选择脚本类型，修改“用户配置区”的参数。
运行脚本：执行脚本，它将自动创建目录并下载文件。
查看结果：在您指定的输出目录中检查下载的数据文件。

1.2 术语解释

中文名称	参数	说明
认证令牌	AUTH_TOKEN/key	用于验证用户下载权限的唯一标识，需从数据下载链接获取
起始时间	BASE_TIME	数据起始时间，格式为 “yyyymmddHH”（如 2025101000 ）
时效	FORECAST_HOURS	如 f001 代表 1 小时时效，不同数据集、不同起始时间的时效范围不同
要素名称	ELEMENT_NAME	气象数据的具体指标名称，需与官方技术文档一致（如 base_reflectivity）
数据类型	DATA_TYPE	可选 TJ1HCN、TJ1GB、TJCN 或 TJGB

2. 下载准备

在开始下载前，请先完成准备工作，包括环境、所需数据信息与个人 key 的确认等，以确保脚本能顺利运行。

2.1 环境准备

网络环境：确保设备网络连接稳定，避免下载过程中断。
存储空间：根据预估文件大小预留足够存储空间（可通过第 4 部分的检测脚本预估文件大小）。
软件安装：
- 若使用 Python 脚本：安装 Python 3.x 及依赖库。
- 若使用 Bash 脚本：确保运行环境为 Linux/macOS 系统，并已安装 curl。

2.2 信息准备

您需提前确认并记录以下关键信息：

数据信息：明确所需数据的数据类型、起始时间（格式需符合 yyyymmddHH）与所需时效范围。
要素名称：从下载页面的【要素清单】或官方【技术文档】→【数据要素说明】中获取准确的要素名称，不同数据集提供的要素数量不同，需仔细确认。
认证令牌（key）：从数据下载链接中提取您的个人{key}，可在 [2.3 下载链接确认] 中查看详细方法。

2.3 下载链接解析（可选）

2.3.1 链接获取途径

（1）访问【数据下载】页面（https://www.tjweather.com/Download），选取您所需的数据包。

（2）在【下载】模块的【标准下载】栏目，依次选择起始时间、要素名，点击图标[...]即可复制认证下载链接。

2.3.2 链接格式解析

复制后的链接格式示例：

https://www.tjweather.com/gateway/admin/download/fixed/{key}/TJ1HCN/2025101000_base_reflectivity_f001.nc

各部分含义解析如下：

“https://www.tjweather.com/gateway/admin/download/fixed”：固定路径。
{key}：需替换为用户个人认证信息。
TJ1HCN：代表数据集类型。
2025101018base_reflectivity_f001.nc：文件名，遵循 “起始日期要素名称_时效.nc” 的格式。

3. 数据下载脚本

我们提供 Python 和 Bash 两种脚本，您可根据操作系统与使用习惯选择其一。两种脚本均已自动处理网站要求的重定向功能，您无需额外配置。

[说明]文件下载使用串行方式下载，目前暂不支持并发下载

3.1 Python 下载脚本

3.1.1 代码说明

使用方法：

配置参数：在“用户配置区”修改起始日期、要素名称、输出目录、时效、认证令牌、数据类型等参数。
检查环境：确认已安装 Python 3.x。
运行脚本：在终端执行 python xxx.py，脚本会自动创建输出目录（若不存在）并开始下载。
查看结果：下载完成后，在 OUTPUT_DIR 指定的目录中查看文件；若下载失败，终端会提示错误原因
关于 FORECAST_HOURS 的示例：
- "all"：下载所有可用时效。
- 12：仅下载第 12 小时时效。
- (1, 24)：下载从第 1 小时到第 24 小时的所有时效。
- (1, 3, 24)：从第 1 小时至第 24 小时开始，每隔 3 小时下载一次，并自动匹配从 0 时开始计算，即下载 3, 6, 9, ... , 24 时效数据。

3.1.2 完整代码

Python

import os
import sys
import time
import requests
from tqdm import tqdm

# =====================
# --- 请在此处配置您的下载信息 ---
# 下载起始日期
BASE_TIME = "2025102912"  # <-- 请修改为您需要的起始日期，数据格式: yyyymmddHH
# 要素名称
ELEMENT_NAME = "base_reflectivity"  # <-- 请修改为您需要的要素名称，可参考网站技术文档-数据要素说明
# 输出目录
OUTPUT_DIR = "./downloaded_files"   # <-- 请修改为您希望保存文件的目录，可以是相对路径如 './data' 或绝对路径如 '/tmp/netcdf_files'
# 时效选择
FORECAST_HOURS = "all"  # <-- 可选: "all", 整数（如：12）, 或 (起始, 结束)（如：(1, 24)），或 (起始, 间隔, 结束)（如：(1, 3, 24)）
# 认证令牌
AUTH_TOKEN = "xxxxxxxxxxxxxx"  # <-- 请修改为您自己的key
# 数据类型
DATA_TYPE = "TJ1HCN"  # <-- 可选：TJ1HCN、TJ1GB、TJCN 或 TJGB
# =====================

BASE_URL = f"https://www.tjweather.com/gateway/admin/download/fixed/{AUTH_TOKEN}/{DATA_TYPE}"

def get_max_forecast_hours(base_time, data_type):
    """根据起始日期和数据类型确定最大时效"""
    try:
        hour = int(base_time[-2:])
    except Exception:
        raise ValueError(f"无法解析起始日期 {base_time}，应为 yyyymmddHH 格式")

    data_config = {
        "TJ1HCN": {0: 360, 6: 72, 12: 360, 18: 72},
        "TJ1GB": {0: 360, 6: 72, 12: 360, 18: 72},
        "TJCN": {0: 240, 12: 240},
        "TJGB": {0: 1104, 12: 1104},
    }

    if data_type not in data_config:
        raise ValueError(f"不支持的数据类型: {data_type}")

    if hour not in data_config[data_type]:
        valid_hours = list(data_config[data_type].keys())
        raise ValueError(f"起始日期 {base_time} 对于数据类型{data_type}不是标准时间{valid_hours}")

    return data_config[data_type][hour]

def get_forecast_hours_list(base_time, forecast_cfg, data_type):
    """根据配置生成要下载的时效列表"""
    max_hours = get_max_forecast_hours(base_time, data_type)

    if forecast_cfg == "all":
        if data_type == "TJGB":
            print("[!] 说明：TJGB数据集360小时之后仅提供6小时分辨率的数据。")
            return list(range(1, 361)) + list(range(366, max_hours + 1, 6))
        return list(range(1, max_hours + 1))
    elif isinstance(forecast_cfg, int):
        if not 1 <= forecast_cfg <= max_hours:
            raise ValueError(f"指定的时效 {forecast_cfg} 超出范围(1-{max_hours})")
        return [forecast_cfg]
    elif isinstance(forecast_cfg, (tuple, list)):
        if len(forecast_cfg) == 2:
            start, end = forecast_cfg
            step = 1
        elif len(forecast_cfg) == 3:
            start, step, end = forecast_cfg
        else:
            raise ValueError("时效设置格式错误，应为2或3个参数 (起始,结束) 或 (起始,间隔,结束)")

        if start < 1 or end > max_hours or start > end:
            raise ValueError(f"时效范围无效: 起始={start}, 结束={end}, 最大={max_hours}")

        hours_list = []

        if data_type == "TJGB" and end > 360:
            print("[!] 说明：TJGB数据集360小时之后仅提供6小时分辨率的数据。")
            h = step
            while h <= end:
                if h < start:
                    h += step
                    continue
                if h <= 360:
                    hours_list.append(h)
                    h += step
                else:
                    if h % 6 != 0:
                        h = ((h // 6) + 1) * 6
                    hours_list.append(h)
                    h += 6
            hours_list = [h for h in hours_list if start <= h <= end]
        else:
            h = step
            while h <= end:
                if h >= start:
                    hours_list.append(h)
                h += step
            hours_list = [h for h in hours_list if h <= end]
        return hours_list

    else:
        raise ValueError("无效的预报时效配置")

def download_file(url, save_path, filename):
    """下载单个文件并显示进度条"""
    max_retries = 3
    for attempt in range(max_retries):
        try:
            with requests.get(url, stream=True, timeout=60) as r:
                if r.status_code == 404:
                    print(f"[!] 文件不存在，跳过: {filename}")
                    return False

                r.raise_for_status()
                total = int(r.headers.get("content-length", 0))

                with open(save_path, "wb") as f, tqdm(
                    total=total, unit="B", unit_scale=True, desc=filename
                ) as bar:
                    for chunk in r.iter_content(chunk_size=8192):
                        if chunk:
                            f.write(chunk)
                            bar.update(len(chunk))

            print(f"[+] 下载成功: {filename}")
            return True

        except requests.exceptions.RequestException as e:
            if attempt < max_retries - 1:
                print(f"[!] 下载失败 ({attempt+1}/{max_retries})，5秒后重试: {e}")
                time.sleep(5)
            else:
                print(f"[x] 下载失败，已达最大重试次数: {e}")
                return False
        except Exception as e:
            print(f"[x] 未知错误: {e}")
            return False

    return False

def main():
    """主函数"""
    if not all([BASE_TIME, ELEMENT_NAME, AUTH_TOKEN, DATA_TYPE]):
        sys.exit("[x] 请先配置 BASE_TIME、ELEMENT_NAME、AUTH_TOKEN、DATA_TYPE。")

    if len(BASE_TIME) != 10 or not BASE_TIME.isdigit():
        sys.exit(f"[x] 起始时间格式错误: {BASE_TIME}，应为10位数字 (yyyymmddHH)")

    try:
        forecast_hours = get_forecast_hours_list(BASE_TIME, FORECAST_HOURS, DATA_TYPE)
        max_forecast = get_max_forecast_hours(BASE_TIME, DATA_TYPE)
    except Exception as e:
        sys.exit(f"[x] 配置错误: {e}")

    os.makedirs(OUTPUT_DIR, exist_ok=True)

    print(f"[*] 起始时间: {BASE_TIME}")
    print(f"[*] 最大时效: {max_forecast} 小时")
    print(f"[*] 数据类型: {DATA_TYPE}")
    print(f"[*] 计划下载: {len(forecast_hours)} 个文件")
    print(f"[*] 时效列表: {forecast_hours}")
    print(f"[*] 输出目录: {OUTPUT_DIR}")
    print("[*] 开始下载...\n")
    success = 0
    for i, fh in enumerate(forecast_hours, 1):
        forecast_str = f"f{fh:04d}" if fh >= 1000 else f"f{fh:03d}"
        filename = f"{BASE_TIME}_{ELEMENT_NAME}_{forecast_str}.nc"
        url = f"{BASE_URL}/{filename}"
        path = os.path.join(OUTPUT_DIR, filename)
        print(f"[*] 下载文件 ({i}/{len(forecast_hours)}): {filename}")
        if download_file(url, path, filename):
            success += 1

        if i < len(forecast_hours):
            time.sleep(2)

    print(f"\n[+] 下载完成!")
    print(f"    成功: {success} 个文件，失败: {len(forecast_hours) - success} 个文件")
    print(f"    保存位置: {OUTPUT_DIR}")

if __name__ == "__main__":
    main()

3.2 Bash 下载脚本

3.2.1 代码说明

使用方法：

配置参数：在“用户配置区”修改起始日期、要素名称、输出目录、时效、认证令牌、数据类型等参数。
赋予权限：在终端执行 chmod +x xxx.sh ，赋予脚本可执行权限
运行脚本：执行./xxx.sh，脚本会自动处理目录创建与文件下载
结果查看：下载完成后，在 OUTPUT_DIR 目录中确认文件；若下载失败，终端会提示错误原因

时效选项类型与 Python 脚本一致，具体格式可查看脚本中的注释。

3.2.2 完整代码

Bash

#!/bin/bash

# 注：本脚本使用curl命令，请确保环境安装该命令

# --- 请在此处配置您的下载信息 ---
# 下载起始日期
BASE_TIME="2025102912"      # <-- 请修改为您需要的起始日期，数据格式:yyyymmddHH 例如"2025101000"
# 要素名称
ELEMENT_NAME="base_reflectivity"   # <-- 请修改为您需要的要素名称，可参考网站技术文档-数据要素说明
# 输出目录
OUTPUT_DIR="./downloaded_files"   # <-- 请修改为您希望保存文件的目录，可以是相对路径如 './data' 或绝对路径如 '/tmp/netcdf_files'
# 时效选择
FORECAST_HOURS="all"  # <-- 可选: "all", 整数（如：12）, 或 "起始时效, 结束时效"（如： "1-24" ），或"起始时效，间隔，结束时效"（如："1, 3, 24" ）
# 认证令牌
AUTH_TOKEN="xxxxxxxxxxxxxx" # <-- 请修改为您自己的key
# 数据类型
DATA_TYPE="TJ1HCN"  # <-- 可选：TJ1HCN、TJ1GB、TJCN 或 TJGB
# ------------------------------------

BASE_URL="https://www.tjweather.com/gateway/admin/download/fixed/${AUTH_TOKEN}/${DATA_TYPE}"

RED='\033[0;31m'; GREEN='\033[0;32m'; YELLOW='\033[1;33m'; NC='\033[0m'
error() { echo -e "${RED}[x] $1${NC}" >&2; exit 1; }
info()  { echo -e "[*] $1"; }
success() { echo -e "${GREEN}[+] $1${NC}"; }
warn()  { echo -e "${YELLOW}[!] $1${NC}" >&2; }

get_max_hours() {
    local hour=${1: -2}
    [[ ! "$hour" =~ ^[0-9]{2}$ ]] && error "无法解析起始日期 $1，应为 yyyymmddHH 格式"
    hour=$((10#$hour))

    declare -A config=([TJ1HCN]="0:360 6:72 12:360 18:72" [TJ1GB]="0:360 6:72 12:360 18:72" [TJCN]="0:240 12:240" [TJGB]="0:1104 12:1104")
    [[ -z "${config[$2]}" ]] && error "不支持的数据类型: $2"

    local max_hour
    for item in ${config[$2]}; do
        if [[ "${item%:*}" == "$hour" ]]; then
            max_hour="${item#*:}"
            break
        fi
    done

    [[ -z "$max_hour" ]] && error "$2 数据不支持 ${hour} 时起报"
    echo "$max_hour"
}

generate_list() {
    local max_hours=$(get_max_hours "$1" "$3")

    if [[ "$2" == "all" ]]; then
        if [[ "$3" == "TJGB" ]]; then
            warn "TJGB数据集360小时之后仅提供每6小时的文件。"
            for ((h=1; h<=max_hours; h++)); do
                ((h <= 360 || h % 6 == 0)) && echo "$h"
            done
        else
            seq 1 "$max_hours"
        fi
        return
    fi

    if [[ "$2" =~ ^[0-9]+$ ]]; then
        [[ $2 -lt 1 || $2 -gt $max_hours ]] && error "时效 $2 超出范围(1-$max_hours)"
        echo "$2"
        return
    fi

    if [[ "$2" =~ ^[0-9]+-[0-9]+$ ]]; then
        local start=${2%-*} end=${2#*-}
        [[ $start -lt 1 || $end -gt $max_hours || $start -gt $end ]] && error "时效范围无效"
        if [[ "$3" == "TJGB" && $end -gt 360 ]]; then
            warn "TJGB数据集360小时之后仅提供每6小时的文件。"
            for ((h=$start; h<=$end; h++)); do
                ((h <= 360 || h % 6 == 0)) && echo "$h"
            done
        else
            seq "$start" "$end"
        fi
        return
    fi

    if [[ "$2" =~ , ]]; then
        IFS=, read -r start step end <<< "$2"
        [[ -z "$step" ]] && step=1
        [[ $start -lt 1 || $end -gt $max_hours || $start -gt $end ]] && error "时效范围无效"
        if [[ "$3" == "TJGB" && $end -gt 360 ]]; then
            warn "TJGB数据集360小时之后仅提供每6小时的文件。"
            for ((h=$step; h<=$end; )); do
                if ((h >= start && (h <= 360 || (h > 360 && h % 6 == 0)))); then
                    echo "$h"
                fi
                if ((h < 360)); then
                    h=$((h + step))
                else
                    ((h % 6 == 0)) || h=$(( (h/6 + 1) * 6 ))
                    h=$((h + 6))
                fi
            done
        else
            for ((h=$step; h<=$end; h+=step)); do
                ((h >= start)) && echo "$h"
            done
        fi
        return
    fi
    error "无效的预报时效配置: $2"
}

download_file() {
    local max_retries=3
    for attempt in $(seq 1 $max_retries); do
        if curl -L --fail --connect-timeout 60 --progress-bar -o "$2" "$1"; then
            return 0
        else
            rm -f "$2"
            [[ $attempt -lt $max_retries ]] && { warn "下载失败 ($attempt/$max_retries)，5秒后重试: $3"; sleep 5; } || return 1
        fi
    done
}

command -v curl >/dev/null 2>&1 || error "未检测到 curl，请先安装"
[[ -z "$BASE_TIME$ELEMENT_NAME$AUTH_TOKEN$DATA_TYPE" ]] && error "请设置所有必需参数"
[[ ${#BASE_TIME} -ne 10 || ! "$BASE_TIME" =~ ^[0-9]{10}$ ]] && error "起始时间格式错误: $BASE_TIME"

max_hours=$(get_max_hours "$BASE_TIME" "$DATA_TYPE") || exit 1
forecast_list=$(generate_list "$BASE_TIME" "$FORECAST_HOURS" "$DATA_TYPE")
total_files=$(echo "$forecast_list" | wc -l)

info "起始时间 $BASE_TIME，最大时效 $max_hours 小时，计划下载 $total_files 个文件"
info "时效列表: $(echo $forecast_list | tr '\n' ' ')"
info "数据类型: $DATA_TYPE，输出目录: $OUTPUT_DIR"
mkdir -p "$OUTPUT_DIR" || error "无法创建输出目录"

success=0; i=0
while read -r fh; do
    [[ -z "$fh" ]] && continue
    i=$((i+1))
    tag=$(printf "f%03d" "$fh"); [[ $fh -ge 1000 ]] && tag=$(printf "f%04d" "$fh")
    fname="${BASE_TIME}_${ELEMENT_NAME}_${tag}.nc"

    info "($i/$total_files) 下载: $fname"
    if download_file "${BASE_URL}/${fname}" "${OUTPUT_DIR}/${fname}" "$fname"; then
        success=$((success+1)); success "下载成功: $fname"
    else
        warn "下载失败: $fname"
    fi
    [[ $i -lt $total_files ]] && sleep 2
done <<< "$forecast_list"

info "下载完成! 成功: $success, 失败: $((total_files-success))"
[[ $success -eq 0 ]] && error "所有文件下载失败" || success "文件保存路径: $OUTPUT_DIR"

4. 其他参考代码补充

下载链接可用性检测脚本

我们提供了一个 Python 检测脚本，用于在正式下载前检查链接有效性和预估文件大小。

使用方法：

安装依赖：若未安装 pycurl，执行命令 pip install pycurl。
配置参数：修改认证令牌（key）、数据类型、起始日期等信息。
运行脚本：执行 python xxx.py，查看检测结果。
结果判断：
- 若显示 “HTTP 状态码：200 → OK” 且文件大小正常：可继续执行第 3 部分中的下载脚本。
- 若显示其他状态码（如 401、404）：根据状态码说明修正配置（如重新获取 key、核对要素名称），并重新检测。

完整代码：

Python

import pycurl
from io import BytesIO

def explain_http_status(code: int) -> str:
    """根据 HTTP 状态码返回简要说明"""
    # 这只是常见状态码的简单映射，可根据需要扩展
    mapping = {
        200: "OK — 请求成功",
        201: "Created — 资源已创建",
        202: "Accepted — 请求已接受但尚未处理",
        204: "No Content — 请求成功但没有返回内容",
        301: "Moved Permanently — 永久重定向",
        302: "Found / Moved Temporarily — 临时重定向",
        304: "Not Modified — 资源未修改",
        400: "Bad Request — 请求语法错误",
        401: "Unauthorized — 未经授权（需要身份验证）",
        403: "Forbidden — 禁止访问资源",  # 403 表示服务器理解请求但拒绝执行  [oai_citation:0‡维基百科](https://en.wikipedia.org/wiki/HTTP_403?utm_source=chatgpt.com)
        404: "Not Found — 资源未找到",
        405: "Method Not Allowed — 请求方法不被允许",
        408: "Request Timeout — 请求超时",
        413: "Payload Too Large — 请求实体过大",
        500: "Internal Server Error — 服务器内部错误",
        502: "Bad Gateway — 错误网关",
        503: "Service Unavailable — 服务不可用（服务器过载或维护）",
        504: "Gateway Timeout — 网关超时",
    }
    return mapping.get(code, f"未知状态码 {code}")

def get_file_size_with_pycurl(url):
    buf = BytesIO()
    c = pycurl.Curl()
    c.setopt(c.URL, url)
    c.setopt(c.WRITEDATA, buf)
    c.setopt(c.NOBODY, True)   # 只请求头部，不下载主体
    c.setopt(c.FOLLOWLOCATION, True)

    c.perform()

    # 获取 HTTP 状态码和 Content-Length（字节）
    http_code = c.getinfo(pycurl.HTTP_CODE)
    size_bytes = c.getinfo(pycurl.CONTENT_LENGTH_DOWNLOAD)
    c.close()

    # 处理 size_bytes，转换为 KB
    size_info = "无法获取大小"
    if size_bytes is not None and size_bytes >= 0:
        size_kb = size_bytes / 1024.0
        size_info = f"{size_kb:.2f} KB"

    # 状态说明
    status_desc = explain_http_status(http_code)

    # 输出结果（您也可以改成 return 字典等形式）
    print(f"HTTP 状态码: {http_code} → {status_desc}")
    print(f"文件大小（估算）: {size_info}")

    return http_code, size_bytes

# 例子
if __name__ == "__main__":
    AUTH_TOKEN = "xxxxxxxxxxxxxx" # <-- 请修改为您自己的key
    DATA_TYPE = "TJ1HCN"  # <-- 可选：TJ1HCN、TJ1GB、TJCN 或 TJGB
    BASE_TIME = "2025101000"  # <-- 请修改为您需要的起始日期
    url = f"https://www.tjweather.com/gateway/admin/download/fixed/{AUTH_TOKEN}/{DATA_TYPE}/{BASE_TIME}_bdsf_ave_f001.nc" # <-- 可修改为您需要的要素
    get_file_size_with_pycurl(url)

标准下载脚本示例文档 ​

​

1. 文档说明 ​

1.1 步骤概览 ​

1.2 术语解释 ​

2. 下载准备 ​

2.1 环境准备 ​

2.2 信息准备 ​

2.3 下载链接解析（可选） ​

2.3.1 链接获取途径 ​

2.3.2 链接格式解析 ​

3. 数据下载脚本 ​

3.1 Python 下载脚本 ​

3.1.1 代码说明 ​

3.1.2 完整代码 ​

3.2 Bash 下载脚本 ​

3.2.1 代码说明 ​

3.2.2 完整代码 ​

4. 其他参考代码补充 ​

标准下载脚本示例文档

1. 文档说明

1.1 步骤概览

1.2 术语解释

2. 下载准备

2.1 环境准备

2.2 信息准备

2.3 下载链接解析（可选）

2.3.1 链接获取途径

2.3.2 链接格式解析

3. 数据下载脚本

3.1 Python 下载脚本

3.1.1 代码说明

3.1.2 完整代码

3.2 Bash 下载脚本

3.2.1 代码说明

3.2.2 完整代码

4. 其他参考代码补充