简介面向农业、交通、旅游、健康等领域研究者以及希望掌握数据采集与处理的Python开发者这套全国天气数据采集方案覆盖2011—2024年全国34个省级行政区、2290个地区的历史天气信息包含气温、风向、风力、生活指数、健康指数等字段。压缩包共300个文件以261个CSV城市天气数据为主另有29个Python采集/处理脚本、6个XML配置、2个Jupyter Notebook和少量工程配置文件整体体积仅1.23MB便于快速下载与二次分析已有163人学习。从中既可获得可直接用于统计分析、可视化与行业应用的多城市多年天气数据也能参考爬虫实现策略、接口获取思路及数据清洗方法适合作为数据科学、爬虫开发课程设计或行业分析的前期素材。1. 全国天气数据采集一份能直接跑的爬虫与清洗落地方案如果你正在做数据分析、算法模型或者可视化大屏八成会卡在“数据从哪来”这一关。天气数据看起来到处都有但真正要批量拿到全国范围的历史与实时数据时手动复制显然不现实而市面上的开放接口不是限频就是需要申请密钥。这份“全国天气数据采集源代码”解决的就是这个痛点它面向公开网页与接口设计了一套采集逻辑配合数据处理脚本能把零散的 JSON 响应整理成干净的表格数据。适合正在做课程设计、毕业项目或想建立个人天气数据集的同学也适合需要定期更新气象数据的从业者。整份资源从请求构造到异常重试再到数据落盘都覆盖到了拿到手不是看文档而是可以直接改参数跑起来。2. 数据采集前的选型与准备接口识别、请求头设计与存储格式选择2.1 先判断数据源类型JSON 接口优先于 HTML 解析很多人在爬虫项目里第一步就栽在解析方式上。拿到一个天气网站新手第一反应是去解析 HTML 页面里的表格或 div 节点但这个方案既脆弱又低效。页面结构一调整正则或 XPath 就全废了。更合理的做法是先打开浏览器的开发者工具切到 Network 面板刷新页面后筛选 XHR 请求看有没有返回 JSON 数据的接口。大部分天气平台的实时数据、逐小时预报和空气质量数据都是通过 JSON 接口下发的页面上的内容只是前端拿到 JSON 后渲染出来的结果。接口地址往往类似于/api/weather/current或/v1/city/101010100这样的路径格式返回体里直接包含城市编码、温度、湿度、气压、风向风速等字段。解析 JSON 比解析 HTML 稳定得多因为接口字段命名通常长期不变而页面模板三天两头调整。这份资源里提供的代码默认就采用了接口优先策略。它内置了一个城市编码表代码先根据城市名找到对应的区域编码再拼出完整的接口 URL。这样做的好处是请求路径完全可控不会出现页面 URL 里参数名猜不准的问题。如果你要接入自己的数据源核心也是先抓接口而不是抓页面。2.2 请求头与 Session 复用降低被拦截的概率拿到了接口地址直接 requests.get 未必能拿到数据。很多服务端会校验请求头重点看 User-Agent 和 Referer。没有这两项服务器很容易识别出这是脚本请求返回 403 或者重定向到验证页面。这份资源的代码里对请求头做了完整定义核心部分是模拟真实浏览器环境import requests HEADERS { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36, Accept: application/json, text/plain, */*, Accept-Language: zh-CN,zh;q0.9, Referer: https://weather.example.com/, Connection: keep-alive, } def create_session(): session requests.Session() session.headers.update(HEADERS) return session这里有几个关键点。Session 对象会自动管理 Cookie如果接口在首次访问时下发 Cookie 做状态校验Session 能替你维护住这个状态。User-Agent 里带上了完整的浏览器版本信息而不是简写成python-requests这能绕过相当一部分基础校验。Accept字段明确声明只接受 JSON告诉服务端你期望的数据格式这也是模拟前端行为的一部分。我自己在实际使用中还会给 Session 挂一个适配器用来调整重试逻辑from requests.adapters import HTTPAdapter session create_session() adapter HTTPAdapter(max_retries3, pool_connections10, pool_maxsize10) session.mount(http://, adapter) session.mount(https://, adapter)这样遇到连接池耗尽或临时网络抖动时requests 会自动重试不必在循环里手动 catch 异常。采集全国城市数据时挨个城市串行请求会非常慢而并发量拉太高又会触发限流。HTTPAdapter 里的连接池参数配合后面会讲到的并发采集能在速度和稳定性之间找到一个相对舒服的平衡点。2.3 城市编码表采集范围的地基全国天气采集的第一步不是写循环而是准备一份完整的城市编码表。接口通常不认城市中文名认的是区域编码。这份资源的代码仓库里提供了一个字典结构键是城市名值是编码覆盖了主要地级市和部分县级市。CITY_CODE_MAP { 北京: 101010100, 上海: 101020100, 广州: 101280101, 深圳: 101280601, # ... 其余城市编码省略实际资源中为完整映射表 } def get_city_code(city_name: str) - str: code CITY_CODE_MAP.get(city_name) if not code: raise ValueError(f未找到城市编码: {city_name}) return code这段逻辑本身不难但有一个实操细节值得提编码并不是简单递增的。有些城市有多个区级站点编码比如某个直辖市可能有“城区”和“郊区”两个编码对应不同的观测站。你要采集的数据粒度如果细化到区县需要单独维护一张区县表如果只到地级市层面上面的字典就够用了。我一般建议把城市编码表单独存成 CSV 文件而不是硬编码在 .py 里。原因很简单一是 CSV 方便随时增删城市不用改代码二是后续做数据校验时可以直接用 pandas 读取这张表跟采集结果做 left join快速找出哪些城市没有采到数据。资源里给的默认做法是硬编码字典但注释里也标注了可以切到 CSV 模式。2.4 存储格式决策CSV 还是 SQLite采集到的数据要落盘存储格式取决于你拿数据做什么。如果只是做一次性的数据分析、画图CSV 足够而且能直接用 Excel 打开检查。但如果要做增量采集、按时间查询历史记录SQLite 更合适。这份资源默认输出 CSV同时提供了一段可选的 SQLite 写入逻辑。我的建议是如果采集任务只是跑一次用 CSV如果你打算每天定时跑把数据累积成时间序列那必须上数据库。原因很简单——CSV 文件的去重和增量合并非常痛苦每次全量重写不仅慢还容易在文件写入过程中损坏数据。SQLite 加一个 UNIQUE 约束在表上插入时执行INSERT OR REPLACE就能优雅地解决重复数据问题。资源里自带的 CSV 输出逻辑大致是这样的import csv from datetime import datetime def save_to_csv(records, filename): fieldnames [city, temp, humidity, pressure, wind_dir, wind_power, weather, update_time] with open(filename, a, newline, encodingutf-8-sig) as f: writer csv.DictWriter(f, fieldnamesfieldnames) if f.tell() 0: writer.writeheader() for row in records: row[update_time] datetime.now().strftime(%Y-%m-%d %H:%M:%S) writer.writerow(row)encodingutf-8-sig这个参数容易忽略但很重要。直接写utf-8生成的 CSV 用 Excel 打开时中文会乱码加上了 BOM字节序标记后 Excel 才能正确识别编码。这是一个典型的血泪经验很多人在数据采完、打开文件的一瞬间心态崩掉原因就是少了这三个字符。3. 代码结构与运行流程从请求构造到并发采集再到落盘3.1 模块化拆分采集器、解析器、存储层各司其职这份资源不是一个大而全的脚本而是按职责拆分了四个模块。采集器负责发起网络请求解析器负责从 JSON 里提取业务字段存储层负责写 CSV 或数据库调度器负责遍历城市列表并控制并发。模块化带来的直接好处是改起来容易——比如某个数据源改了字段名只需要动解析器不用碰采集和存储代码。拆解后的核心调用关系如下# main.py 中调度逻辑 from collector import fetch_weather from parser import parse_weather from storage import save_to_csv, save_to_sqlite def run(cities, concurrency5): results [] for city in cities: code get_city_code(city) raw_json fetch_weather(code) parsed parse_weather(city, raw_json) if parsed: results.append(parsed) save_to_csv(results, weather_data.csv) # 如需入库: save_to_sqlite(results, weather.db)上面的代码里有一个if parsed判断这是解析器返回空字典时的保护逻辑。遇到接口返回异常但不报错的情况——比如业务上返回了空 JSON 体——解析器会返回None调度器要跳过这条记录而不是把空数据写进文件。实际采集时我不建议串行跑完整轮因为城市数量一多串行等待时间呈线性增长。资源里给的默认实现是串行但在注释和文档里指出了并发改造的位置。下面是我在实际项目中验证过的并发版本from concurrent.futures import ThreadPoolExecutor, as_completed def run_concurrent(cities, max_workers8): all_records [] with ThreadPoolExecutor(max_workersmax_workers) as executor: future_map { executor.submit(fetch_one_city, city): city for city in cities } for future in as_completed(future_map): city future_map[future] try: record future.result() if record: all_records.append(record) print(f[OK] {city}) except Exception as e: print(f[FAIL] {city}: {e}) return all_recordsmax_workers8是基于大多数天气接口限流阈值的经验值。并发开到 20 以上很容易触发服务端的频率限制返回 429 状态码。如果你要更快方案不是无限调大线程数而是考虑分布式采集——多个出口 IP 并行请求。但对于课程设计和一般的数据集构建8 个线程足够在几分钟内跑完全国主要城市。3.2 解析器设计字段提取与容错处理解析器是整个采集流程里最容易出错也最容易翻车的一环。接口返回的 JSON 结构通常是嵌套的需要逐层取数据而且部分字段在特殊天气条件下可能缺失。比如台风天气时某些观测站的“气压”字段可能为空雷雨天气时“能见度”字段可能不返回。资源里的解析器采用了显式的字段映射方式每个业务字段对应 JSON 路径中的一层def parse_weather(city, raw_json): try: data raw_json[data] return { city: city, temp: data[temp], humidity: data[humidity], pressure: data[pressure], wind_dir: data[windDir], wind_power: data[windPower], weather: data[weather], } except (KeyError, TypeError) as e: print(f解析失败: {city}, 原因: {e}) return None这里用try...except包裹并返回None而不是让异常向上抛出是经过考虑的。对采集任务来说单个城市解析失败不应该中断整个流程记录下失败信息后继续跑才是正确姿势。TypeError的捕获尤其重要因为接口偶尔会返回异常结构比如data字段不是字典而是字符串此时直接访问data[temp]会抛出TypeError如果没有捕获整个线程就崩了。如果要做更高级的容错可以在except块里把原始 JSON 保存下来import json with open(fdebug_{city}.json, w, encodingutf-8) as f: json.dump(raw_json, f, ensure_asciiFalse, indent2)这样后续排查时能看到接口到底返回了什么而不是盲猜字段名。3.3 数据校验与空值处理解析完成之后直接写文件还是太草率。采集回来的数据一般要做一层校验至少检查温度是否在合理范围内、湿度是否在 0 到 100 之间。如果接口某个时刻返回了异常值比如湿度 158%直接写进数据集会在后续分析阶段产生严重的脏数据。我一般会在存储前加一个简单的校验函数def validate_record(record): errors [] if not -50 record[temp] 60: errors.append(f温度异常: {record[temp]}) if not 0 record[humidity] 100: errors.append(f湿度异常: {record[humidity]}) if not 850 record[pressure] 1100: errors.append(f气压异常: {record[pressure]}) return errors校验规则里的数值边界参考了我国气候的极端情况。温度下限定到 -50 是因为极北地区冬季可能出现接近这个数值的低温上限 60 覆盖夏季极端高温并留有冗余。如果校验不通过可以选择丢弃记录也可以保留并打上脏数据标记。资源里的默认做法是丢弃同时把错误信息打到日志里方便你确认是不是采集环节出了问题。4. 数据处理与清洗从原始 JSON 到可分析的数据集4.1 时间字段标准化与类型转换采集到的原始数据里时间字段往往格式不统一。有的接口返回时间戳有的返回2024-01-15 14:30:00这样的字符串还有的返回2024/1/15 下午2:30这种带中文的格式。直接存进数据集后面没法做时间序列分析。标准化的第一步是统一转成datetime对象或固定格式字符串。import pandas as pd def normalize_time(series): # 处理时间戳单位可能是秒或毫秒 numeric_mask pd.to_numeric(series, errorscoerce).notna() if numeric_mask.any(): # 判断是秒还是毫秒 sample float(series[numeric_mask].iloc[0]) unit ms if sample 10**12 else s ts pd.to_numeric(series, errorscoerce) if unit ms: ts ts / 1000 series.loc[numeric_mask] pd.to_datetime(ts, units) # 剩余字符串格式尝试标准解析 series pd.to_datetime(series, errorscoerce) return series.dt.strftime(%Y-%m-%d %H:%M:%S)这个函数做了两层处理。先判断字段内容是数字还是字符串数字就按时间戳解析同时自动识别精度是秒还是毫秒字符串则交给pd.to_datetime做通用解析。转换失败的记录会被置为NaT后续清洗时统一处理。我在实际项目里遇到过接口返回时间戳精度不固定的情况——大部分时间是秒级偶尔出现毫秒级。如果不做单位判断直接按秒解析会把毫秒级时间戳放大 1000 倍日期直接跳到几年后。上面代码里先采样判断再统一转换就是为了规避这个坑。4.2 缺失值与重复记录处理采集过程难免产生缺失值。网络抖动导致某个城市的数据没采到或者接口本身就没有返回某个字段都会在数据集里留下缺口。处理缺失值没有万能方案取决于你的分析目标。如果做的是某一天的全国天气快照某个城市缺失温度数据可以选择直接删掉这条记录。但如果做的是时间序列分析删除会导致时间轴断裂更适合的做法是保留记录把缺失字段填成NaN后续用插值方法补全。def clean_dataset(df): # 删除完全重复的记录 df df.drop_duplicates(subset[city, update_time]) # 数值列转为 float无法转换的置为 NaN numeric_cols [temp, humidity, pressure] for col in numeric_cols: df[col] pd.to_numeric(df[col], errorscoerce) return df代码里先按“城市时间”两个字段去重这是防止重复采集的关键步骤。如果你用 CSV 累积存储每次跑脚本都会追加新数据同一时刻同一城市可能被写入两次。如果不做去重后面画趋势图时会出现同一条数据被画两遍的玄学问题。pd.to_numeric转换时把无法解析的值变成NaN而不是直接报错中断这样清洗流程能完整跑完问题数据留到下一阶段单独审查。对于某个城市整体缺失的情况我习惯在清洗后打印一份缺失报告def miss_report(df): full_cities set(df[city].unique()) expected_cities set(CITY_CODE_MAP.keys()) missing expected_cities - full_cities print(f缺失城市数量: {len(missing)}) for city in sorted(missing): print(f - {city}) return missing这份报告的价值在于定位采集失败的城市你可以针对性重试而不是整轮重跑。很多时候某些城市连续失败原因不是接口挂了而是这个城市编码不存在或已被下线。此时需要去数据源确认编码是否仍然有效。4.3 特征工程从原始字段生成衍生指标拿到基本的温度、湿度、气压后可以做一层轻量特征工程生成对后续分析更有用的衍生字段。比如体感温度、温度与湿度的组合指标、昼夜温差等。def add_features(df): # 简单体感温度近似计算 df[feels_like] df[temp] - 0.4 * (100 - df[humidity]) / 100 * df[temp] # 露点温度近似计算Magnus 公式 df[dew_point] ( 243.04 * (np.log(df[humidity] / 100) 17.625 * df[temp] / (243.04 df[temp])) / (17.625 - np.log(df[humidity] / 100) - 17.625 * df[temp] / (243.04 df[temp])) ) return df新增字段的计算公式不必过于复杂以稳定可解释为准。体感温度模型有很多种上面用的是兼顾温度和湿度的简化公式适合日常场景。露点温度直接反映空气中实际水汽含量比相对湿度更适合做极端天气分析。这些衍生字段的价值在后续数据分析阶段才会体现出来但提前在清洗环节生成可以避免重复计算。5. 日常运行与常见问题排查从能跑到稳定跑5.1 定时采集与增量更新天气数据是强时效性的数据一次性采集只能得到某一时刻的快照。要想构建长期数据集需要把采集脚本变成定时任务。Windows 上用计划任务Linux 上用 crontab都能实现每天定时执行。一个典型的 crontab 配置# 每天 8 点、14 点、20 点执行采集 0 8,14,20 * * * cd /path/to/project python main.py logs/collect.log 21日志输出必须重定向到文件这是血泪经验。没有日志脚本半夜跑挂了第二天你根本不知道哪里出错。每次采集后检查collect.log末尾的[FAIL]计数能快速判断这轮采集是否健康。增量更新的核心是时间字段。每次写入数据库时带上采集时间查询历史数据时按时间和城市组合过滤就能得到完整的时间序列。CSV 模式下增量更新比较别扭建议定时任务长期跑的场景直接改用 SQLite 落库。5.2 采集速度与限流之间的平衡并发采集时最常见的错误是盲目增加线程数。线程数一旦超过接口的限流阈值返回的就不是数据而是 429 状态码。429 本身不可怕可怕的是频繁触发 429 后可能被封 IP封禁时间从几分钟到几小时不等。资源里默认给的并发数是 8这个数值相对保守但安全。如果你需要判断自己的场景下能不能开更高并发建议先做一次试探import time def probe_limit(session, code, threshold3): status_counts {success: 0, limited: 0} for _ in range(10): resp session.get(f{BASE_URL}/{code}) if resp.status_code 429: status_counts[limited] 1 else: status_counts[success] 1 time.sleep(1) return status_counts连续请求 10 次如果 429 出现次数超过阈值说明当前并发策略已经逼近限流边界需要降低并发数或增加请求间隔。把健康检查前置到正式采集前能替你省去一整个下午的排查时间。5.3 接口字段变更的应对预案数据源接口升级时字段名可能悄然变化。temp改成temperaturewindDir改成wind_dir这类改动不会报错解析器取不到值返回None你只会看到失败记录变多。应对方案是在解析器里做字段兼容处理def get_field(data, *candidates): for key in candidates: if key in data: return data[key] return None temp get_field(data, temp, temperature, t)当主字段名取不到时自动尝试备用字段名。这种做法牺牲了一点代码整洁度但能提高采集程序的鲁棒性。资源里的解析器默认只取单字段但注释里标明了兼容改造的接入点。5.4 三个常见坑的记录第一个坑城市编码表里包含已撤销的编码。某些县级市升格为地级市后原编码废弃接口对无效编码返回一个空数据体而不是错误状态码。现象是采集报告里没有任何异常但清洗后数据集里缺了该城市。解决办法是在校验阶段将空数据体视为解析失败输出到失败日志中。第二个坑并发写入 CSV 时文件损坏。多个线程同时打开同一个文件句柄写入会导致行交错、内容被覆盖。现象是最终文件行数少于预期且部分行只写了半个字段。解决办法是改成单线程写文件或者使用队列把所有采集结果汇总后统一写入。你可以在实际项目里直接参考下方统一写入的写法from queue import Queue result_queue Queue() def collect_worker(city): record fetch_one_city(city) if record: result_queue.put(record) # 主线程在全部 worker 结束后统一写文件 # 避免多线程直接操作同一个文件句柄第三个坑采集过程中内存占用飙升。全国城市数据单次采集量并不大但如果你把每个城市的完整原始 JSON 都保存在内存里而不及时释放几千个城市的累积量足够让内存吃紧。现象是脚本跑到后半程越来越慢甚至被系统杀掉。解决办法是每处理完一个城市就把解析后的精简记录追加到列表原始 JSON 用完后立刻丢弃不要保留。6. 进阶用法把数据集变成可视化看板与异常检测基线拿到一份干净的历史天气数据集后可以做的事远超查天气本身。我常用它来做两件事一是搭建一个轻量级可视化看板二是用它做气象异常检测的基线数据。看板技术栈不需要太复杂前端用一个图表库后端直接读 SQLite。图表库配置好 x 轴为时间、y 轴为温度或湿度值一眼就能看出全国的冷暖分布趋势。这里的关键在于数据加载要高效不能在每次页面刷新时全量查询应用层需要按城市和时间范围做过滤。// 前端查询示例按城市和时间范围过滤 async function loadWeatherData(city, startDate, endDate) { const response await fetch( /api/weather?city${city}start${startDate}end${endDate} ); const rows await response.json(); const dates rows.map((r) r.update_time); const temps rows.map((r) r.temp); // 绘制温度折线图 renderLineChart(dates, temps, ${city} 温度变化); }如果你不熟悉 Web 开发另一个更直接的数据应用是构建简单的异常检测规则。当实时采集到的温度偏离历史同时间段均值超过两个标准差时打上“气象异常”标记。def detect_anomaly(history_df, new_record): city_history history_df[history_df[city] new_record[city]] mean_temp city_history[temp].mean() std_temp city_history[temp].std(ddof0) if abs(new_record[temp] - mean_temp) 2 * std_temp: return {city: new_record[city], alert: True} return {city: new_record[city], alert: False}这个方法不复杂但足以筛出大多数明显异常情形。例如某北方城市 3 月份气温突然跳到 32℃就会被自动标记。对于日均数据两天一跑就能顺带检查清洗结果是否异常。如果你做的是更正式的研究项目还可以把这套逻辑接到定时任务里用邮件或企业微信机器人推送到自己手机上。如果没有告警需求直接在 Jupyter Notebook 里查一次结果也是一样的效果。资源本身的采集与处理能力到这里基本够用了。回看整条链路从城市编码出发经过接口请求、JSON 解析、数据校验、清洗入库最终落到分析与可视化每一步都有至少一个容易翻车但可预防的细节。从那以后我每次跑采集任务都强制走一遍三件事先确认接口返回的字段结构没变再检查缺失城市报告最后用validate_record过滤异常值。三件套做下来数据集的质量基本稳了。希望这份拆解能帮你把天气数据这条路走通需要完整代码直接取用即可。本文还有配套的精品资源点击获取