# WorkBuddy 一个数据人的省钱日记：用 AI 搞定全国343个城市天气数据，省下几百块
·
一个数据人的省钱日记：让 AI 帮我写代码，零成本搞定全国343城天气数据
背景：工作需要全国地级市天气数据，国内付费API动辄几百元。我让 WorkBuddy（AI智能助手）帮我写了一套Python脚本，全程对话式搞定，零成本拿到343城数据，省下几百块。
一、需求背景：天气数据真不便宜
最近工作需要用到全国地级市的天气数据，主要关注温度、是否下雨这些维度，时间范围 2024 年至今，按日统计就行。
听起来不复杂？上手才发现，天气数据在国内是真·付费玩家专属：
| 方案 | 价格 | 备注 |
|---|---|---|
| 和风天气 API | 商用需付费，按量计费 | 343城 × 500+天，估算几百元起 |
| 心知天气 API | 商用套餐起步价不低 | 历史数据更贵 |
| 聚合数据等 | 按次收费，历史数据贵 | 批量拉不划算 |
| 某宝代下 | 50-200元不等 | 质量参差不齐，不一定靠谱 |
对于个人或小团队来说，花几百块买一次性数据，确实肉疼。
二、转机：让 AI 帮我写代码
作为一个数据人，我日常用 WorkBuddy（一款AI智能助手，既能做分析也能写代码）处理各种工作。
我把需求直接丢给了它：
“我需要最近两年全国地级市（333个）的天气，使用 open-meteo，主要关注温度、是否下雨等维度，帮忙写Python脚本。”
接下来就是纯对话式开发，我只需要做决策，代码全交给 WorkBuddy 生成：
| 我的需求 | WorkBuddy 的响应 |
|---|---|
| 每个月每个城市一个CSV | 调整输出格式 |
| 从2024年1月1日开始 | 改起始日期 |
| 每个城市一个CSV文件 | 重构输出逻辑 |
| 帮忙控制并发，貌似被封IP了 | 加429限流保护 |
| 文件已生成的自动跳过 | 加断点续传 |
整个过程中，我只需要在几个关键节点做选择，其余的脚本编写、调试、限流处理全部由 WorkBuddy 完成。
三、WorkBuddy 给出的方案：Open-Meteo
WorkBuddy 帮我调研后，推荐了 Open-Meteo —— 一个完全免费、无需 API Key 的天气数据服务，基于 ECMWF ERA5 再分析数据。
核心特点：
- 完全免费，无需注册，无需 API Key
- 支持历史数据（ERA5 再分析数据）
- 覆盖全球任意经纬度
- 日级数据：最高/最低温、降水、风速、湿度等十几个维度
API 地址：
https://archive-api.open-meteo.com/v1/era5
四、WorkBuddy 写的核心代码
下面是 WorkBuddy 生成的完整脚本（已精简核心逻辑），关键设计点都在注释里：
import requests
import pandas as pd
import time
import json
import os
import random
from datetime import datetime
# ========== 配置区 ==========
START_DATE = "2024-01-01"
END_DATE = "2026-06-09"
OUTPUT_DIR = "weather_data"
CITIES_FILE = "cities.csv"
REQUEST_DELAY = 3.0 # 请求间隔（秒）
MAX_RETRIES = 5 # 最大重试次数
RETRY_BASE_DELAY = 10.0 # 429退避基础延迟（秒）
# =============================
class WeatherDataDownloader:
"""天气数据下载器（WorkBuddy 生成）"""
def __init__(self, output_dir=OUTPUT_DIR):
self.output_dir = output_dir
self.base_url = "https://archive-api.open-meteo.com/v1/era5"
self.progress_file = os.path.join(output_dir, 'progress.json')
self.progress = self._load_progress()
os.makedirs(output_dir, exist_ok=True)
# 复用连接会话，减少连接开销
self.session = requests.Session()
adapter = requests.adapters.HTTPAdapter(
pool_connections=1,
pool_maxsize=1,
max_retries=0
)
self.session.mount('http://', adapter)
self.session.mount('https://', adapter)
def fetch_weather_data(self, lat, lon, start_date, end_date):
"""
获取单个城市天气数据（带指数退避重试）
"""
params = {
'latitude': lat,
'longitude': lon,
'start_date': start_date,
'end_date': end_date,
'daily': [
'temperature_2m_max', 'temperature_2m_min',
'temperature_2m_mean', 'precipitation_sum',
'rain_sum', 'snowfall_sum', 'weathercode',
'windspeed_10m_max', 'winddirection_10m_dominant',
'relative_humidity_2m_mean', 'pressure_msl_mean',
'et0_fao_evapotranspiration'
],
'timezone': 'Asia/Shanghai'
}
last_error = None
for attempt in range(MAX_RETRIES + 1):
try:
response = self.session.get(
self.base_url, params=params, timeout=60
)
# 429 限流处理：指数退避
if response.status_code == 429:
wait = min(RETRY_BASE_DELAY * (2 ** attempt), 300)
print(f"429限流，等待 {wait:.1f}s 后重试")
time.sleep(wait)
last_error = "429 Too Many Requests"
continue
response.raise_for_status()
data = response.json()
daily = data['daily']
df = pd.DataFrame({
'date': daily['time'],
'temp_max': daily['temperature_2m_max'],
'temp_min': daily['temperature_2m_min'],
'temp_mean': daily['temperature_2m_mean'],
'precipitation': daily['precipitation_sum'],
'rain': daily['rain_sum'],
'snowfall': daily['snowfall_sum'],
'weathercode': daily['weathercode'],
'windspeed_max': daily['windspeed_10m_max'],
'wind_direction': daily['winddirection_10m_dominant'],
'humidity_mean': daily['relative_humidity_2m_mean'],
'pressure_mean': daily['pressure_msl_mean'],
'evapotranspiration': daily['et0_fao_evapotranspiration']
})
# 衍生字段：是否下雨、天气描述
df['is_rainy'] = df['rain'].apply(
lambda x: x > 0 if x is not None else False
)
df['weather_desc'] = df['weathercode'].apply(
self._weathercode_to_desc
)
return df
except requests.exceptions.RequestException as e:
wait = 5 * (attempt + 1)
print(f"网络错误，{wait}s后重试: {e}")
time.sleep(wait)
last_error = str(e)
if attempt == MAX_RETRIES:
raise
raise Exception(f"重试{MAX_RETRIES}次后仍失败: {last_error}")
def download_city_data(self, city_info, start_date, end_date):
"""下载单个城市数据，已存在则自动跳过"""
city_name = city_info['city']
province = city_info['province']
lat = float(city_info['lat'])
lon = float(city_info['lon'])
safe_city = city_name.replace('/', '_').replace('\\', '_')
safe_province = province.replace('/', '_').replace('\\', '_')
output_file = os.path.join(
self.output_dir, f"{safe_city}_{safe_province}.csv"
)
# 断点续传：磁盘文件已存在就跳过
if os.path.exists(output_file):
print(f"[SKIP] 文件已存在: {os.path.basename(output_file)}")
return True
try:
df = self.fetch_weather_data(lat, lon, start_date, end_date)
df['city'] = city_name
df['province'] = province
df['lat'] = lat
df['lon'] = lon
df.to_csv(output_file, index=False, encoding='utf-8-sig')
city_key = f"{city_name}_{province}"
self.progress['completed'].append(city_key)
self._save_progress()
return True
except Exception as e:
self.progress['failed'].append({
'city': city_name, 'error': str(e)
})
self._save_progress()
return False
finally:
# 3秒基础 + 0~2秒随机抖动
delay = REQUEST_DELAY + random.uniform(0, 2.0)
time.sleep(delay)
WorkBuddy 设计的三大关键点
1. 断点续传（磁盘文件优先）
if os.path.exists(output_file):
return True # 文件在就不重复下载
即使 progress.json 被删，只要 CSV 在就不会重复请求。
2. 429 限流保护（指数退避）
第1次失败 → 等 10s
第2次失败 → 等 20s
第3次失败 → 等 40s
...最多300s
3. 请求节奏控制
- 每个城市间隔 3s + 0~2s 随机抖动
- 每 10 个城市额外休息 15s
- 连续失败 3 次自动暂停 60s
五、最终成果
数据规模
| 指标 | 数值 |
|---|---|
| 城市数量 | 343 个 |
| 时间范围 | 2024-01-01 ~ 2026-06-09 |
| 数据粒度 | 日级 |
| 字段数量 | 20 个（含衍生字段） |
| 总记录数 | 约 30 万行 |
| 合并文件大小 | 36 MB |
字段说明
| 字段 | 说明 | 单位 |
|---|---|---|
| date | 日期 | YYYY-MM-DD |
| temp_max | 日最高温度 | °C |
| temp_min | 日最低温度 | °C |
| temp_mean | 日平均温度 | °C |
| precipitation | 总降水量 | mm |
| rain | 降雨量 | mm |
| snowfall | 降雪量 | cm |
| weathercode | WMO天气代码 | - |
| windspeed_max | 最大风速 | m/s |
| wind_direction | 主导风向 | ° |
| humidity_mean | 平均相对湿度 | % |
| pressure_mean | 平均海平面气压 | hPa |
| evapotranspiration | 参考蒸散发 | mm |
| is_rainy | 是否下雨 | True/False |
| weather_desc | 天气描述 | 中文 |
| wind_direction_desc | 风向描述 | 中文 |
| city | 城市名 | - |
| province | 省份 | - |
| lat | 纬度 | - |
| lon | 经度 | - |
输出文件结构
weather_data/
├── 七台河_黑龙江省.csv # 每个城市一个文件
├── 三亚_海南省.csv
├── 上海_上海市.csv
├── 北京_北京市.csv
├── ...
├── all_cities.csv # 所有城市合并（30万行，36MB）
├── progress.json # 断点续传进度
└── weather_download.log # 运行日志
数据示例
date,temp_max,temp_min,temp_mean,precipitation,rain,is_rainy,weather_desc,city,province
2024-01-01,-8.6,-24.5,-15.4,0.0,0.0,False,阴天,七台河,黑龙江省
2024-01-02,-9.1,-19.4,-13.2,0.0,0.0,False,阴天,七台河,黑龙江省
2024-01-04,-4.7,-13.7,-9.0,0.3,0.0,False,小雪,七台河,黑龙江省
六、成本对比
| 方案 | 成本 | 数据维度 | 历史数据 |
|---|---|---|---|
| 和风天气商业API | 几百元起 | 丰富 | 付费 |
| 心知天气商业版 | 几百元起 | 丰富 | 付费 |
| 某宝代下 | 50-200元 | 看卖家 | 不稳定 |
| WorkBuddy + Open-Meteo | 0 元 | 13+维度 | 支持 |
实际节省：几百元。
而且数据完全自主可控，需要更新时重跑脚本即可，没有任何边际成本。
七、踩坑记录（都是 WorkBuddy 帮我填的）
坑1：429 Too Many Requests
一开始设的 1 秒间隔，跑了约 50 个城市就被 Open-Meteo 限流了。
WorkBuddy 的修复： 间隔从 1s 提到 3s，加随机抖动，429 时指数退避重试。
坑2：断点续传不可靠
脚本异常退出时 progress.json 可能没及时写入。
WorkBuddy 的修复： 优先检查磁盘文件是否存在，不依赖 progress.json。
坑3：Windows CMD 编码问题
bat 文件里写了中文提示，CMD 报"不是内部或外部命令"。
WorkBuddy 的修复： bat 文件全用英文，中文交给 Python 的 logging。
八、总结
| 维度 | 评价 |
|---|---|
| WorkBuddy | 对话式开发，我只做决策，代码全包 |
| Open-Meteo API | 免费、稳定、数据质量好（ERA5再分析数据） |
| 数据覆盖 | 全国343城，日级，20个字段，够用 |
| 开发成本 | 0 行手写代码，全程对话 |
| 经济成本 | 0 元 |
| 适用场景 | 个人分析、小团队、非商用 |
一句话总结：能用 AI 搞定的事，绝不花冤枉钱。
数据来源：Open-Meteo（https://open-meteo.com/），基于 ECMWF ERA5 再分析数据，仅供个人学习和非商业用途。
代码由 WorkBuddy AI 智能助手生成，经实际运行验证。
Agent 垂直技术社区，欢迎活跃、内容共建。
更多推荐
6
0- 0
已为社区贡献1条内容
所有评论(0)