全国天气数据采集实战：从爬虫到SQLite定时入库

简介：这是一份覆盖全国2290个地区、时间跨度为2011年至2024年的历史天气数据集，配套完整的Python爬虫与数据处理源代码，适合数据分析初学者、气象研究者以及需要长期天气数据进行农业、交通、旅游等领域分析的用户。压缩包内共300个文件，以261个CSV数据文件为核心，每个文件对应一个城市的历史天气记录，包含气温、风向、风力等逐日数据；另有29个Python脚本用于抓取、清洗与合并数据，6个XML和2个Jupyter Notebook文件辅助说明与演示，整体仅1.23MB，结构清晰、便于快速上手。资源目前已有161人学习下载，作者还提供了数据来源说明与抓取策略概述，帮助读者理解合法获取数据的方式。通过实际动手，学习者既能获得可直接使用的多年份天气数据集，也能掌握网页爬虫编写、异常处理和多表合并等实用技能，为进一步的数据挖掘与可视化分析打下基础。 做全国天气数据采集这个项目，起因特别简单：我手头有一个数据可视化大屏，需要把全国主要城市的天气情况实时展示出来，一开始打算直接用现成的天气API，但要么收费，要么有调用次数限制，要么返回字段跟我们需求对不上。最后决定自己写一套全国天气数据采集源代码，抓取目标城市天气数据后存进自己的数据库，想怎么用就怎么用。

整套代码前后跑了快一周，从城市列表管理、HTTP请求封装、JSON解析、SQLite入库到定时增量更新，各环节都踩过不少坑，最终稳定跑下来的完整流程我整理成这篇文章。如果你正准备做数据采集、入门爬虫、或者刚好需要给可视化项目准备天气数据源，这篇内容可以直接拿去做底稿。

1. 动手前的思路：先想清楚三件事

1.1 数据源怎么选，免费接口背后的取舍

做天气采集，最先卡住人的就是“数据从哪来”。市面上绝大多数商用天气API都有严格QPS限制，日调用量也卡得很死，真要覆盖全国几百个城市，一天几十万次请求根本扛不住。我的做法是选一个免费、无需复杂鉴权、返回结构稳定的接口作为基础数据源。

这里有一个核心思路：采集方案的重心不是“哪个接口数据全”，而是“接口返回结构是否稳定、能否低成本做容错”。就算接口偶尔抽风，只要解析层做了足够的异常兜底，整体采集任务依然能自愈运行。我实际选用的接口只需要传入城市ID，返回JSON包含城市名、天气状况、温度区间、湿度、风力和风向，完全覆盖大屏展示需求。

1.2 整体架构：采集、解析、入库、调度的闭环

整个项目的标准闭环是：定时调度器触发采集任务 → 遍历城市列表 → 发起HTTP请求 → 解析返回数据 → 清洗字段 → 写入数据库 → 记录日志。简单画一下流程就是：

城市列表(JSON) -> 采集模块(requests) -> 解析模块(json) -> 入库模块(sqlite3) -> 日志模块(logging) ^ | |________________ 定时调度(schedule) __________________________|

这样设计的好处是每一层都能独立测试，比如今天接口字段变了，我只需要改解析模块；想换MySQL存储，入库模块单独替换即可。对于这种生命周期可能很长的数据采集任务，模块化可维护性是第一位的，别把代码写成一坨只能在本地跑通的脚本。

2. 核心代码：采集、解析、入库三步走

2.1 城市列表与请求封装

第一步是准备城市列表。全国所有区县的天气城市ID是固定编码，不需要手动收集，直接把我预置好的城市列表放进JSON文件或者Python字典里。我这边用了一个包含城市名和城市ID的列表，覆盖了直辖市、省会城市以及主要地级市。

# city_list.py CITIES = [ {"name": "北京", "id": "101010100"}, {"name": "上海", "id": "101020100"}, {"name": "广州", "id": "101280101"}, {"name": "深圳", "id": "101280601"}, {"name": "杭州", "id": "101210101"}, {"name": "成都", "id": "101270101"}, {"name": "武汉", "id": "101200101"}, {"name": "西安", "id": "101110101"}, {"name": "南京", "id": "101190101"}, {"name": "重庆", "id": "101040100"}, ]

请求层封装时，有几点很容易被忽略：必须设置超时时间、必须设置User-Agent、必须对返回状态码做预检。超时时间我习惯设成5秒，某些接口在晚间高峰期响应会明显变慢，不设超时的话整个任务会被单个卡死的请求拖住。

# weather_client.py import requests import json 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" } def fetch_weather(city_id: str, city_name: str) -> dict: url = f"https://example-weather-api.com/data/{city_id}.html" try: resp = requests.get(url, headers=HEADERS, timeout=5) resp.raise_for_status() data = resp.json() return {"city": city_name, "raw": data, "ok": True} except requests.Timeout: return {"city": city_name, "raw": None, "ok": False, "error": "timeout"} except requests.RequestException as e: return {"city": city_name, "raw": None, "ok": False, "error": str(e)}

注意：生产环境建议把真实接口地址替换成你有权限调用的数据源，并且把API Key一类凭证放到环境变量或配置中心，不要硬编码在源代码里。这一点对于所有数据采集项目都通用，养成习惯能少踩很多泄露风险方面的坑。

2.2 解析逻辑：从JSON到结构化数据

接口返回的JSON结构哪怕再简单，也建议在解析层做一层“字段映射”，而不是在业务代码里到处直接访问字典键。一来是便于应对上游字段改名，二来是可以把异常数据统一拦在入口。

def parse_weather(city_name: str, raw: dict) -> dict: try: info = raw["data"] return { "city": city_name, "weather": info.get("weather", "未知"), "temp_max": info.get("temp_max", ""), "temp_min": info.get("temp_min", ""), "humidity": info.get("humidity", ""), "wind": info.get("wind", ""), "update_time": info.get("date", ""), "fetch_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S") } except (KeyError, TypeError): return None

返回值直接统一成我们自己的字段结构，与上游解耦。温度字段有时候返回的是“22℃/12℃”这种带单位的字符串，如果业务层需要做数值计算，建议在解析层顺手把数字部分提取出来，存成单独的temp_high和temp_low字段。

2.3 入库设计：SQLite轻量够用，注意去重

存储层我用的是SQLite，原因很简单：单机部署、不需要单独安装数据库服务、Python内置支持。表结构设计如下：

CREATE TABLE IF NOT EXISTS weather_daily ( id INTEGER PRIMARY KEY AUTOINCREMENT, city TEXT NOT NULL, weather TEXT, temp_max TEXT, temp_min TEXT, humidity TEXT, wind TEXT, fetch_time TEXT );

写入时需要做去重：同一个城市两小时内可能被采集多次，流量费和时间都浪费在重复写入上。我在表上加了(city, fetch_time)的唯一索引，写入策略改成INSERT OR IGNORE，这样重复调度时自动跳过已有数据。

def save_weather(conn, item: dict) -> None: sql = """ INSERT OR IGNORE INTO weather_daily (city, weather, temp_max, temp_min, humidity, wind, fetch_time) VALUES (?, ?, ?, ?, ?, ?, ?) """ conn.execute(sql, ( item["city"], item["weather"], item["temp_max"], item["temp_min"], item["humidity"], item["wind"], item["fetch_time"] )) conn.commit()

实操心得：SQLite在频繁并发写入时会有锁冲突，我们这里是单线程顺序写入，完全没问题。如果以后要并发采集提速，数据库换成PostgreSQL或MySQL就行，入库函数基本不用改。

3. 让数据自动更新：定时调度与工程化落地

3.1 定时调度：schedule库五分钟搞定

天气数据一般一小时更新一次即可，太频繁不仅浪费请求资源，还容易触发接口限流。我用了Python的schedule库做定时任务，每天每整点执行一次全量采集。

import schedule import time def job(): print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 开始采集全国天气数据") run_collect() schedule.every().hour.at(":05").do(job) while True: schedule.run_pending() time.sleep(1)

定时时间我特意设成整点过5分钟——避开接口整点缓存刷新的高并发时间段，被限流的概率会低一些。这个细节是从实际运行日志里总结出来的，头几天整点整分跑，时不时就撞上429状态码。

3.2 日志与容错：采集任务必须能自愈

数据采集脚本一旦部署到服务器上，大概率会遇到网络抖动、接口超时、上游返回脏数据等问题。所以日志和错误处理绝对不能省。我用了标准库logging，同时输出到控制台和文件，生产环境建议配一套loguru或接入ELK，排查问题时体验完全不一样。

import logging logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", handlers=[ logging.StreamHandler(), logging.FileHandler("weather_collect.log", encoding="utf-8") ] )

单城市采集失败时，我采用“记录错误、继续下一个城市”的策略，绝不让一个故障站点中断整个采集流程。全部城市跑完后，统计成功数和失败数，失败比例超过阈值时通过企业微信机器人或邮件告警，这样不用天天盯着日志，出了问题也能第一时间介入。

3.3 控制请求频率，别把自己搞进黑名单

采集脚本运行一段时间后，我曾因为请求太密集被接口短暂封禁。现在的做法是在每两个请求之间加一个随机延时，时间在0.5到1.5秒之间波动，模拟人工访问的节奏。全国300个城市，一轮采集也就五六分钟，完全够用。

import random import time for city in CITIES: result = fetch_weather(city["id"], city["name"]) if result["ok"]: item = parse_weather(city["name"], result["raw"]) if item: save_weather(conn, item) time.sleep(random.uniform(0.5, 1.5))

4. 踩坑实录与排查清单

4.1 城市ID解析失败，接口返回“无此城市”

排查思路：这类问题大多是城市ID不合法或者接口数据源覆盖范围有限。可以先手工在浏览器打开接口地址，确认城市ID能否正常返回数据；如果返回为空，换一个邻近城市的ID测试，判断是不是该城市本身就没有数据。

4.2 请求偶尔返回429状态码，触发限流

处理办法是退避重试加请求间隔。我在代码里加了一个简单的指数退避：第一次失败等2秒重试，第二次等4秒，最多重试3次。重试仍然失败就跳过当前城市，留到下一轮调度再采。

def fetch_with_retry(city_id: str, city_name: str, retries: int = 3): for attempt in range(retries): result = fetch_weather(city_id, city_name) if result["ok"]: return result wait_time = 2 ** (attempt + 1) time.sleep(wait_time) return result

4.3 写入数据库后中文乱码

解决方案：连接SQLite时显式设置好编码，SQLite本身不处理编码问题，写入前把字符串统一转成UTF-8，读取展示侧也统一UTF-8。Python3 源码文件默认就是UTF-8，但Windows下如果用了记事本编辑源码，务必另存为UTF-8编码，否则字面量里的中文同样会乱。

4.4 常见问题排查速查表

免费接口数据源稳定性有一定波动，这是所有免费数据采集方案都要接受的事实。如果你对数据实时性、准确性要求极高，建议在源码里预留好付费API的替换入口——解析层返回统一结构，切换数据源时只要改客户端函数和解析函数，其余代码零改动。

最后分享一个实战小技巧

跑了一段时间后，我发现很多天气接口在每天凌晨会做数据校准，凌晨采集到的数据偶尔出现短暂异常。所以我把每天的第一轮全量采集从凌晨1点调到了早上7点，并且把历史数据保留下来做了一份按城市、按周的天气趋势统计，用来反推接口数据质量。如果某天某个城市的最高温突然和前后两天差距过大，基本可以断定是当天数据异常，自动打标后就不再把这条数据放进可视化大屏。

另外，在入门阶段认定“源代码”这三个字时，我的建议是不要只关注跑通效果，更要关注代码的工程结构。数据采集这个方向，项目真正值钱的地方通常是异常处理做得稳不稳、日志记录全不全、调度模块能不能独立复用，而不是哪一行请求代码写得有多漂亮。希望这套全国天气数据采集源代码能给你提供一份可行的脚手架，剩下的功能，就看你自己的业务需要往里填了。

本文还有配套的精品资源，点击获取