简介由CSV数据表与Python源码组成的天气资源包沉淀了全国三十四个省、两千二百九十个地区在二零一一年至二零二四年间超过十年的历史天气记录涵盖气温、风向、风力以及生活指数、健康指数、旅游指数等字段适合农业、交通、旅游等行业的数据分析人员也适合正在学习爬虫与数据清洗的开发者。压缩包内共有三百个文件其中二百六十一个CSV文件直接存放各城市天气明细二十九个Python脚本负责数据采集与预处理六个XML文件用于定义爬虫请求或解析规则两个Jupyter文件演示了从抓取到分析的完整链路整包体积仅一点二三兆字节轻量但结构完整。目前已经有一百六十一人学习或下载属于偏实用型案例资源可用来快速上手天气数据的整理与分析。使用者既能直接读取全国多年天气数据用于统计分析也能参照其爬虫设计与处理逻辑在合规前提下改造为特定地区的实时天气采集工具或进一步拓展可视化展示与预测建模等应用。 做天气类的小项目很多人第一反应是找个现成的SDK或者API包几行代码就能拿到数据。但一旦你面对的是“全国城市”这几个字情况就完全变了SDK往往只开放单点查询免费额度撑不起全量轮询底层字段你也没法自定义。我这次干脆从零搭了一套全国天气数据采集的脚本单文件Python实现支持全国主要城市逐日抓取、CSV/SQLite双写落库还带重试和增量更新。整套方案的源代码就是核心交付物放到服务器上配个定时任务就能长期跑。这篇文章不只是贴代码我会把数据源选择、字段解析、踩过的坑和工程化思路全部拆开讲适合正在做爬虫入门、数据分析预处理、或者想给个人项目加一个稳定天气数据源的开发者参考。1. 为什么自己搭全国天气采集而不是直接调一个现成库天气类API一直不缺免费的有、付费的也有封装好的Python库更是一抓一大把。那为什么还要自己写一套采集我在动手之前其实做过一轮权衡这里直接说结论。1.1 现成SDK在“全国全量”场景下的三个硬伤第一个硬伤是接口形态。大多数天气SDK的设计思路是“按城市查”你给它一个城市ID它回一个城市的实时数据或预报数据。如果你的业务只覆盖三五个城市这完全够用。但要采集全国几百上千个城市你就得自己维护城市清单、自己写循环、自己处理限流这时候SDK的“封装”反而成了累赘你绕不开它还得按它的限制来。第二个硬伤是数据所有权。SDK拿到的数据一般只能在它的终端组件里展示想导出成结构化数据做二次分析很多服务商在协议里是不允许的或者只给很小的导出配额。而天气数据最有价值的用法恰恰是长期积累——历史归档、同期对比、趋势分析这些都需要你自己掌握原始数据。第三个硬伤是版本和稳定性。第三方SDK升级频繁今天这个版本能用明天接口字段改名你可能要跟着改业务代码。自建采集脚本用的是底层HTTP接口只要数据源协议不变代码就能一直跑维护成本反而更低。1.2 这套方案适合谁用我自己这次搭建的目标很明确拿到全国主要城市的每日天气快照字段包含日期、天气现象、最高温、最低温数据能够增量存储、方便导出。适用场景包括个人爬虫练手、课程设计中的数据分析模块、校园天气大屏的数据底座、以及需要离线归档的行业报表。如果你也是类似需求这套源代码的思路完全可以复用到同类公开数据采集上不只是天气。1.3 整体设计目标动手前我给自己列了三条硬性要求单文件可运行、不依赖重型组件、服务器上能常驻。所以最终方案定为Python标准库requestssqlite3的组合城市清单内置在脚本里采集逻辑和数据落库解耦这样既能快速跑通后续加功能也不至于推倒重来。2. 数据源选择与采集策略先摸清接口再动手很多人一上来就写爬虫结果爬到一半发现字段对不上或者被限流回头才研究数据源。我建议顺序反过来先花半天把数据源摸透再写代码整体效率高很多。2.1 可用的公开天气数据源有哪些目前网络上可用的公开天气HTTP接口主要有几类。一类是商业天气服务商提供的免费开发档比如和风天气、高德天气它们的JSON结构规范字段命名接近自然语言适合直接解析。另一类是气象部门公开的基础数据服务这类接口通常没有复杂鉴权但字段风格偏原始需要自己做一层映射。我这次选择了“免费档结构化JSON”的数据源原因有三点返回体干净不需要用正则去匹配杂乱的HTML。有明确的请求频率限制说明方便设计采集频率。字段基本覆盖天气现象、最高温、最低温、日期不需要额外拼接。2.2 城市清单怎么组织采集全国天气最基础的是维护一张城市编码表。市面上的天气接口基本都支持两种入参城市ID或者经纬度坐标。城市ID的好处是稳定、可读、不容易被反爬策略盯上所以我的脚本里直接用字典维护了一张城市ID映射表取前几十个主要城市作为默认集合CITY_MAP { 北京: 101010100, 上海: 101020100, 广州: 101280101, 深圳: 101280601, 杭州: 101210101, 成都: 101270101, 武汉: 101200101, 西安: 101110101, }这个表的好处是扩展容易要加城市直接往字典里塞一条就行。如果城市数量上千可以改成读外部JSON文件不必动主逻辑。提示不同数据源的城市编码体系不一样换数据源时第一件事就是把编码表整体替换掉否则容易出现“查出来的天气是另外一个城市的”这种隐蔽错误。2.3 请求频率怎么设计才不会被限流公开接口一般都有每分钟请求数的隐性限制三五十次可能没事一口气发几百个请求大概率触发封禁。我的设计原则是全国城市哪怕有几百个也循环采集每次请求之间至少间隔0.5秒失败自动退避。城市多的时候整体耗时会长一些但对个人项目来说完全可接受。与其追求几分钟跑完全国不如保证任务能稳定执行一个月不挂。3. 全国天气采集核心代码逐段拆解下面进入正题源代码本身。整个脚本只有一个weather_collector.py文件代码分四块城市映射、请求封装、解析入库、主调度逻辑。我按顺序拆开讲。3.1 请求封装超时和重试是底线采集脚本跑在服务器上网络抖动是常态。所以请求封装我一开始就加了超时和重试import requests import time def fetch_weather(city_code, retries3): url fhttps://weather-data.example.com/api/weather?cityid{city_code} headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 } for attempt in range(retries): try: resp requests.get(url, headersheaders, timeout10) if resp.status_code 200: return resp.json() # 429 或 5xx 状态码等待后重试 time.sleep(2 * (attempt 1)) except requests.RequestException: time.sleep(2 * (attempt 1)) return None这段代码的思路很简单每次请求最多重试3次失败后等待时间按指数递增2秒、4秒、6秒。timeout10是必须的没有超时的采集脚本一旦遇到连接挂起整个任务会卡死在那个城市上后面的城市全部排队等。3.2 解析逻辑把接口字段映射成统一结构不同天气接口的字段命名差异很大有的叫high有的叫tmp_max还有的直接把温度拼在字符串里。我的做法是写一个parse_weather函数把接口原始数据统一映射成内部结构体后面入库和导出都基于这个结构不直接碰原始字段def parse_weather(raw): if not raw: return None data raw.get(data, {}) forecast_list data.get(forecast, []) if not forecast_list: return None today forecast_list[0] return { city_code: raw.get(cityid, ), city_name: raw.get(city, ), forecast_date: today.get(date, ), weather: today.get(type, ), high: today.get(high, ).replace(高温 , ).replace(℃, ), low: today.get(low, ).replace(低温 , ).replace(℃, ), }这里有个细节值得说high字段如果直接存原始值会出现“高温 23℃”这种带前缀的字符串入库后做数值分析很麻烦。所以我在解析层就做了清洗只保留数字部分。类似这种清洗逻辑集中在解析函数里比散落在各处好维护得多。3.3 落库方案CSV方便看SQLite方便查数据解析出来之后要存下来。我选了双写方案CSV和SQLite同时落一份。为什么这么做CSV文件的好处是可以用Excel直接打开核对数据方便SQLite的好处是支持结构化查询后续做增量更新、按城市检索、去重统计都很轻松。SQLite表结构如下CREATE TABLE IF NOT EXISTS daily_weather ( city_code TEXT, city_name TEXT, forecast_date TEXT, weather TEXT, high TEXT, low TEXT, updated_at TEXT, PRIMARY KEY (city_code, forecast_date) );注意PRIMARY KEY这一行它把城市编码预报日期设成了唯一键。这个设计直接解决了重复采集的问题同一天同一个城市的数据再次写入时用INSERT OR REPLACE覆盖旧记录不会产生脏数据。写入函数import sqlite3 import csv import os DB_PATH weather.db CSV_PATH weather_data.csv def save_to_sqlite(items): conn sqlite3.connect(DB_PATH) cursor conn.cursor() cursor.execute(CREATE TABLE IF NOT EXISTS daily_weather ( city_code TEXT, city_name TEXT, forecast_date TEXT, weather TEXT, high TEXT, low TEXT, updated_at TEXT, PRIMARY KEY (city_code, forecast_date) )) for item in items: cursor.execute(INSERT OR REPLACE INTO daily_weather (city_code, city_name, forecast_date, weather, high, low, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?), (item[city_code], item[city_name], item[forecast_date], item[weather], item[high], item[low], time.strftime(%Y-%m-%d %H:%M:%S))) conn.commit() conn.close() def save_to_csv(items): file_exists os.path.exists(CSV_PATH) with open(CSV_PATH, a, newline, encodingutf-8) as f: writer csv.DictWriter(f, fieldnames[city_code, city_name, forecast_date, weather, high, low, updated_at]) if not file_exists: writer.writeheader() for item in items: writer.writerow(item)3.4 主调度逻辑主函数做的事很简单遍历城市表、逐城请求、解析、收集结果、最后统一入库。这里我不建议“每请求一个城市就立刻写库”而是攒一批再写减少数据库连接次数def main(): items [] for city_name, city_code in CITY_MAP.items(): raw fetch_weather(city_code) parsed parse_weather(raw) if parsed: parsed[city_name] city_name items.append(parsed) print(f[OK] {city_name} | {parsed[forecast_date]} | {parsed[weather]} | {parsed[high]}/{parsed[low]}) else: print(f[FAIL] {city_name}) time.sleep(0.5) if items: save_to_sqlite(items) save_to_csv(items) if __name__ __main__: main()实际跑一轮的效果是这样[OK] 北京 | 2025-01-10 | 晴 | 5℃/-6℃ [OK] 上海 | 2025-01-10 | 多云 | 8℃/1℃ [OK] 广州 | 2025-01-10 | 小雨 | 15℃/10℃到这里一套能跑、能存、能查的全国天气采集脚本就完成了。但说实话脚本跑通只是第一步真正让人头疼的是后面那些偶发问题。4. 实测中绕不开的五个坑及排查思路这个脚本我在本地和服务器上都跑过积累了一些真实遇到的坑。每个问题我都按照“现象→排查→解决”的顺序记录方便你以后遇到类似情况直接对照。4.1 返回的JSON里中文全部乱码现象是接口返回的字符串里城市名和天气现象变成了类似渴这样的乱码。排查方向很明确HTTP响应头的Content-Type里没有声明charsetutf-8而requests默认按ISO-8859-1解码中文自然乱码。解决方法是拿到响应后先检查resp.encoding再强制指定编码resp requests.get(url, headersheaders, timeout10) if resp.encoding is None or resp.encoding.lower() ! utf-8: resp.encoding utf-8更好的做法是直接按resp.apparent_encoding来它能根据响应内容自动判断编码但偶尔会判断错所以明确指定更稳。4.2 预报日期和本地日期对不上这是个很容易被忽略的问题。有些接口返回的forecast[0]是“今天”的预报但如果你在凌晨运行脚本接口可能已经把第一天的数据切成了明天导致当天数据缺失。我一开始没注意直接拿本地日期去归档结果数据库里某一天的数据死活对不上当天的天气。解决方法是完全信任接口返回的date字段不要用本地时间拼接日期。归档时以forecast_date为准这样数据链条才是闭合的。4.3 一次性请求太猛触发限流第一次跑全国城市时我把所有城市ID塞进循环中间没加延时跑了不到50个城市后面全部返回403。排查时先看状态码是403 Forbidden而不是普通的404基本可以确定是被限流了。加上每请求间隔0.5秒、失败退避重试之后整套流程再也没触发过限流。4.4 城市编码表有坑城市编码表不是一成不变的。个别城市的ID在某次数据源升级后被废弃或者同一个城市在不同接口里ID完全不一样。我的排查思路是把每次请求失败的city_code单独记到一个日志表里运行一段时间后看哪个城市频繁失败再回头人工核对。为了避免因为一个城市挂了导致整轮中止采集循环里要对单城市失败做容错——记录失败、继续下一个而不是抛出异常终止整个任务。4.5 同一天跑多次数据重复最开始我只存CSV每天定时任务跑一次没注意重复问题。后来手动补数据一天跑了三次CSV里同一个城市同一天出现三行。加SQLite之前我先在代码里做了一次“先查后插”发现并发场景下还是会重复最后直接用INSERT OR REPLACE加唯一键彻底解决。这个重复问题的本质是“没有唯一约束”靠业务代码判断永远有漏洞最省心的还是数据库层面加主键。5. 工程化升级定时任务、增量存储与轻量可视化采集脚本能跑通之后接下来就是让它变成一套“无人值守”的常驻服务。这里分享我实际用的几个工程化方案。5.1 定时任务配置Linux环境下我用crontab每天的早上7点抓一次早间天气0 7 * * * cd /home/weather /usr/bin/python3 weather_collector.py weather.log 21Windows环境下可以用任务计划程序触发条件设为“每天”操作指向python.exe weather_collector.py。注意日志重定向一定要写不然脚本出错时你完全不知道发生了什么。5.2 增量存储和历史归档拆开之前的daily_weather表已经能通过唯一键增量更新。如果还要做历史归档我建议再建一张weather_history表只追加不覆盖每次采集都把当日快照完整插入。这样daily_weather负责“查最近数据”weather_history负责“做长期分析”两个表职责清晰查询效率也高。CREATE TABLE IF NOT EXISTS weather_history ( id INTEGER PRIMARY KEY AUTOINCREMENT, city_code TEXT, city_name TEXT, forecast_date TEXT, weather TEXT, high TEXT, low TEXT, created_at TEXT DEFAULT (datetime(now, localtime)) );5.3 做一个轻量可视化面板数据积累起来之后只躺在数据库里有点可惜。我用一个最简单的方案做展示脚本每次采集完把SQLite里的最新数据导出成JSON文件前端用纯HTMLJavaScript读取展示主要城市当天的天气卡片和温度对比。import json def export_to_json(): conn sqlite3.connect(DB_PATH) cursor conn.cursor() cursor.execute(SELECT city_name, forecast_date, weather, high, low FROM daily_weather WHERE forecast_date (SELECT MAX(forecast_date) FROM daily_weather)) rows cursor.fetchall() with open(latest_weather.json, w, encodingutf-8) as f: json.dump(rows, f, ensure_asciiFalse, indent2) conn.close()前端只需要一个fetch(latest_weather.json)就能渲染不需要引入任何框架。对于个人项目来说这套方案的性价比远高于搭一套完整后端。5.4 后续还能扩展的方向数据源本身还可以继续加厚接入实时分钟级数据、AQI空气质量、风向风力把原来“每天一次快照”升级成“每小时一次快照”。分析侧可以做天气异常告警比如某城市温度突破历史极值时推送通知也可以把历史气温数据接进时序数据库配合Grafana画出整年的气温变化曲线。我个人的建议是先把“稳定采集、不重不漏”这件事做到位再考虑可视化。数据采集是最底层的地基地基稳了上面盖什么建筑都容易。最后分享一个实际操作中的体会这套方案的代码量不大但真正让它变得可靠的是那些“看起来不起眼”的细节——超时重试、编码判断、唯一键约束、失败容错、日志输出。每一行都对应一次真实的踩坑。你如果准备自己搭一套不用一步到位先把主流程跑起来然后跑个三五天把日志翻一翻你会发现自己项目的稳定性和当初的Demo版本完全是两个东西。本文还有配套的精品资源点击获取 从零搭建全国天气数据采集系统：Python脚本实现与工程化实践
简介由CSV数据表与Python源码组成的天气资源包沉淀了全国三十四个省、两千二百九十个地区在二零一一年至二零二四年间超过十年的历史天气记录涵盖气温、风向、风力以及生活指数、健康指数、旅游指数等字段适合农业、交通、旅游等行业的数据分析人员也适合正在学习爬虫与数据清洗的开发者。压缩包内共有三百个文件其中二百六十一个CSV文件直接存放各城市天气明细二十九个Python脚本负责数据采集与预处理六个XML文件用于定义爬虫请求或解析规则两个Jupyter文件演示了从抓取到分析的完整链路整包体积仅一点二三兆字节轻量但结构完整。目前已经有一百六十一人学习或下载属于偏实用型案例资源可用来快速上手天气数据的整理与分析。使用者既能直接读取全国多年天气数据用于统计分析也能参照其爬虫设计与处理逻辑在合规前提下改造为特定地区的实时天气采集工具或进一步拓展可视化展示与预测建模等应用。 做天气类的小项目很多人第一反应是找个现成的SDK或者API包几行代码就能拿到数据。但一旦你面对的是“全国城市”这几个字情况就完全变了SDK往往只开放单点查询免费额度撑不起全量轮询底层字段你也没法自定义。我这次干脆从零搭了一套全国天气数据采集的脚本单文件Python实现支持全国主要城市逐日抓取、CSV/SQLite双写落库还带重试和增量更新。整套方案的源代码就是核心交付物放到服务器上配个定时任务就能长期跑。这篇文章不只是贴代码我会把数据源选择、字段解析、踩过的坑和工程化思路全部拆开讲适合正在做爬虫入门、数据分析预处理、或者想给个人项目加一个稳定天气数据源的开发者参考。1. 为什么自己搭全国天气采集而不是直接调一个现成库天气类API一直不缺免费的有、付费的也有封装好的Python库更是一抓一大把。那为什么还要自己写一套采集我在动手之前其实做过一轮权衡这里直接说结论。1.1 现成SDK在“全国全量”场景下的三个硬伤第一个硬伤是接口形态。大多数天气SDK的设计思路是“按城市查”你给它一个城市ID它回一个城市的实时数据或预报数据。如果你的业务只覆盖三五个城市这完全够用。但要采集全国几百上千个城市你就得自己维护城市清单、自己写循环、自己处理限流这时候SDK的“封装”反而成了累赘你绕不开它还得按它的限制来。第二个硬伤是数据所有权。SDK拿到的数据一般只能在它的终端组件里展示想导出成结构化数据做二次分析很多服务商在协议里是不允许的或者只给很小的导出配额。而天气数据最有价值的用法恰恰是长期积累——历史归档、同期对比、趋势分析这些都需要你自己掌握原始数据。第三个硬伤是版本和稳定性。第三方SDK升级频繁今天这个版本能用明天接口字段改名你可能要跟着改业务代码。自建采集脚本用的是底层HTTP接口只要数据源协议不变代码就能一直跑维护成本反而更低。1.2 这套方案适合谁用我自己这次搭建的目标很明确拿到全国主要城市的每日天气快照字段包含日期、天气现象、最高温、最低温数据能够增量存储、方便导出。适用场景包括个人爬虫练手、课程设计中的数据分析模块、校园天气大屏的数据底座、以及需要离线归档的行业报表。如果你也是类似需求这套源代码的思路完全可以复用到同类公开数据采集上不只是天气。1.3 整体设计目标动手前我给自己列了三条硬性要求单文件可运行、不依赖重型组件、服务器上能常驻。所以最终方案定为Python标准库requestssqlite3的组合城市清单内置在脚本里采集逻辑和数据落库解耦这样既能快速跑通后续加功能也不至于推倒重来。2. 数据源选择与采集策略先摸清接口再动手很多人一上来就写爬虫结果爬到一半发现字段对不上或者被限流回头才研究数据源。我建议顺序反过来先花半天把数据源摸透再写代码整体效率高很多。2.1 可用的公开天气数据源有哪些目前网络上可用的公开天气HTTP接口主要有几类。一类是商业天气服务商提供的免费开发档比如和风天气、高德天气它们的JSON结构规范字段命名接近自然语言适合直接解析。另一类是气象部门公开的基础数据服务这类接口通常没有复杂鉴权但字段风格偏原始需要自己做一层映射。我这次选择了“免费档结构化JSON”的数据源原因有三点返回体干净不需要用正则去匹配杂乱的HTML。有明确的请求频率限制说明方便设计采集频率。字段基本覆盖天气现象、最高温、最低温、日期不需要额外拼接。2.2 城市清单怎么组织采集全国天气最基础的是维护一张城市编码表。市面上的天气接口基本都支持两种入参城市ID或者经纬度坐标。城市ID的好处是稳定、可读、不容易被反爬策略盯上所以我的脚本里直接用字典维护了一张城市ID映射表取前几十个主要城市作为默认集合CITY_MAP { 北京: 101010100, 上海: 101020100, 广州: 101280101, 深圳: 101280601, 杭州: 101210101, 成都: 101270101, 武汉: 101200101, 西安: 101110101, }这个表的好处是扩展容易要加城市直接往字典里塞一条就行。如果城市数量上千可以改成读外部JSON文件不必动主逻辑。提示不同数据源的城市编码体系不一样换数据源时第一件事就是把编码表整体替换掉否则容易出现“查出来的天气是另外一个城市的”这种隐蔽错误。2.3 请求频率怎么设计才不会被限流公开接口一般都有每分钟请求数的隐性限制三五十次可能没事一口气发几百个请求大概率触发封禁。我的设计原则是全国城市哪怕有几百个也循环采集每次请求之间至少间隔0.5秒失败自动退避。城市多的时候整体耗时会长一些但对个人项目来说完全可接受。与其追求几分钟跑完全国不如保证任务能稳定执行一个月不挂。3. 全国天气采集核心代码逐段拆解下面进入正题源代码本身。整个脚本只有一个weather_collector.py文件代码分四块城市映射、请求封装、解析入库、主调度逻辑。我按顺序拆开讲。3.1 请求封装超时和重试是底线采集脚本跑在服务器上网络抖动是常态。所以请求封装我一开始就加了超时和重试import requests import time def fetch_weather(city_code, retries3): url fhttps://weather-data.example.com/api/weather?cityid{city_code} headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 } for attempt in range(retries): try: resp requests.get(url, headersheaders, timeout10) if resp.status_code 200: return resp.json() # 429 或 5xx 状态码等待后重试 time.sleep(2 * (attempt 1)) except requests.RequestException: time.sleep(2 * (attempt 1)) return None这段代码的思路很简单每次请求最多重试3次失败后等待时间按指数递增2秒、4秒、6秒。timeout10是必须的没有超时的采集脚本一旦遇到连接挂起整个任务会卡死在那个城市上后面的城市全部排队等。3.2 解析逻辑把接口字段映射成统一结构不同天气接口的字段命名差异很大有的叫high有的叫tmp_max还有的直接把温度拼在字符串里。我的做法是写一个parse_weather函数把接口原始数据统一映射成内部结构体后面入库和导出都基于这个结构不直接碰原始字段def parse_weather(raw): if not raw: return None data raw.get(data, {}) forecast_list data.get(forecast, []) if not forecast_list: return None today forecast_list[0] return { city_code: raw.get(cityid, ), city_name: raw.get(city, ), forecast_date: today.get(date, ), weather: today.get(type, ), high: today.get(high, ).replace(高温 , ).replace(℃, ), low: today.get(low, ).replace(低温 , ).replace(℃, ), }这里有个细节值得说high字段如果直接存原始值会出现“高温 23℃”这种带前缀的字符串入库后做数值分析很麻烦。所以我在解析层就做了清洗只保留数字部分。类似这种清洗逻辑集中在解析函数里比散落在各处好维护得多。3.3 落库方案CSV方便看SQLite方便查数据解析出来之后要存下来。我选了双写方案CSV和SQLite同时落一份。为什么这么做CSV文件的好处是可以用Excel直接打开核对数据方便SQLite的好处是支持结构化查询后续做增量更新、按城市检索、去重统计都很轻松。SQLite表结构如下CREATE TABLE IF NOT EXISTS daily_weather ( city_code TEXT, city_name TEXT, forecast_date TEXT, weather TEXT, high TEXT, low TEXT, updated_at TEXT, PRIMARY KEY (city_code, forecast_date) );注意PRIMARY KEY这一行它把城市编码预报日期设成了唯一键。这个设计直接解决了重复采集的问题同一天同一个城市的数据再次写入时用INSERT OR REPLACE覆盖旧记录不会产生脏数据。写入函数import sqlite3 import csv import os DB_PATH weather.db CSV_PATH weather_data.csv def save_to_sqlite(items): conn sqlite3.connect(DB_PATH) cursor conn.cursor() cursor.execute(CREATE TABLE IF NOT EXISTS daily_weather ( city_code TEXT, city_name TEXT, forecast_date TEXT, weather TEXT, high TEXT, low TEXT, updated_at TEXT, PRIMARY KEY (city_code, forecast_date) )) for item in items: cursor.execute(INSERT OR REPLACE INTO daily_weather (city_code, city_name, forecast_date, weather, high, low, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?), (item[city_code], item[city_name], item[forecast_date], item[weather], item[high], item[low], time.strftime(%Y-%m-%d %H:%M:%S))) conn.commit() conn.close() def save_to_csv(items): file_exists os.path.exists(CSV_PATH) with open(CSV_PATH, a, newline, encodingutf-8) as f: writer csv.DictWriter(f, fieldnames[city_code, city_name, forecast_date, weather, high, low, updated_at]) if not file_exists: writer.writeheader() for item in items: writer.writerow(item)3.4 主调度逻辑主函数做的事很简单遍历城市表、逐城请求、解析、收集结果、最后统一入库。这里我不建议“每请求一个城市就立刻写库”而是攒一批再写减少数据库连接次数def main(): items [] for city_name, city_code in CITY_MAP.items(): raw fetch_weather(city_code) parsed parse_weather(raw) if parsed: parsed[city_name] city_name items.append(parsed) print(f[OK] {city_name} | {parsed[forecast_date]} | {parsed[weather]} | {parsed[high]}/{parsed[low]}) else: print(f[FAIL] {city_name}) time.sleep(0.5) if items: save_to_sqlite(items) save_to_csv(items) if __name__ __main__: main()实际跑一轮的效果是这样[OK] 北京 | 2025-01-10 | 晴 | 5℃/-6℃ [OK] 上海 | 2025-01-10 | 多云 | 8℃/1℃ [OK] 广州 | 2025-01-10 | 小雨 | 15℃/10℃到这里一套能跑、能存、能查的全国天气采集脚本就完成了。但说实话脚本跑通只是第一步真正让人头疼的是后面那些偶发问题。4. 实测中绕不开的五个坑及排查思路这个脚本我在本地和服务器上都跑过积累了一些真实遇到的坑。每个问题我都按照“现象→排查→解决”的顺序记录方便你以后遇到类似情况直接对照。4.1 返回的JSON里中文全部乱码现象是接口返回的字符串里城市名和天气现象变成了类似渴这样的乱码。排查方向很明确HTTP响应头的Content-Type里没有声明charsetutf-8而requests默认按ISO-8859-1解码中文自然乱码。解决方法是拿到响应后先检查resp.encoding再强制指定编码resp requests.get(url, headersheaders, timeout10) if resp.encoding is None or resp.encoding.lower() ! utf-8: resp.encoding utf-8更好的做法是直接按resp.apparent_encoding来它能根据响应内容自动判断编码但偶尔会判断错所以明确指定更稳。4.2 预报日期和本地日期对不上这是个很容易被忽略的问题。有些接口返回的forecast[0]是“今天”的预报但如果你在凌晨运行脚本接口可能已经把第一天的数据切成了明天导致当天数据缺失。我一开始没注意直接拿本地日期去归档结果数据库里某一天的数据死活对不上当天的天气。解决方法是完全信任接口返回的date字段不要用本地时间拼接日期。归档时以forecast_date为准这样数据链条才是闭合的。4.3 一次性请求太猛触发限流第一次跑全国城市时我把所有城市ID塞进循环中间没加延时跑了不到50个城市后面全部返回403。排查时先看状态码是403 Forbidden而不是普通的404基本可以确定是被限流了。加上每请求间隔0.5秒、失败退避重试之后整套流程再也没触发过限流。4.4 城市编码表有坑城市编码表不是一成不变的。个别城市的ID在某次数据源升级后被废弃或者同一个城市在不同接口里ID完全不一样。我的排查思路是把每次请求失败的city_code单独记到一个日志表里运行一段时间后看哪个城市频繁失败再回头人工核对。为了避免因为一个城市挂了导致整轮中止采集循环里要对单城市失败做容错——记录失败、继续下一个而不是抛出异常终止整个任务。4.5 同一天跑多次数据重复最开始我只存CSV每天定时任务跑一次没注意重复问题。后来手动补数据一天跑了三次CSV里同一个城市同一天出现三行。加SQLite之前我先在代码里做了一次“先查后插”发现并发场景下还是会重复最后直接用INSERT OR REPLACE加唯一键彻底解决。这个重复问题的本质是“没有唯一约束”靠业务代码判断永远有漏洞最省心的还是数据库层面加主键。5. 工程化升级定时任务、增量存储与轻量可视化采集脚本能跑通之后接下来就是让它变成一套“无人值守”的常驻服务。这里分享我实际用的几个工程化方案。5.1 定时任务配置Linux环境下我用crontab每天的早上7点抓一次早间天气0 7 * * * cd /home/weather /usr/bin/python3 weather_collector.py weather.log 21Windows环境下可以用任务计划程序触发条件设为“每天”操作指向python.exe weather_collector.py。注意日志重定向一定要写不然脚本出错时你完全不知道发生了什么。5.2 增量存储和历史归档拆开之前的daily_weather表已经能通过唯一键增量更新。如果还要做历史归档我建议再建一张weather_history表只追加不覆盖每次采集都把当日快照完整插入。这样daily_weather负责“查最近数据”weather_history负责“做长期分析”两个表职责清晰查询效率也高。CREATE TABLE IF NOT EXISTS weather_history ( id INTEGER PRIMARY KEY AUTOINCREMENT, city_code TEXT, city_name TEXT, forecast_date TEXT, weather TEXT, high TEXT, low TEXT, created_at TEXT DEFAULT (datetime(now, localtime)) );5.3 做一个轻量可视化面板数据积累起来之后只躺在数据库里有点可惜。我用一个最简单的方案做展示脚本每次采集完把SQLite里的最新数据导出成JSON文件前端用纯HTMLJavaScript读取展示主要城市当天的天气卡片和温度对比。import json def export_to_json(): conn sqlite3.connect(DB_PATH) cursor conn.cursor() cursor.execute(SELECT city_name, forecast_date, weather, high, low FROM daily_weather WHERE forecast_date (SELECT MAX(forecast_date) FROM daily_weather)) rows cursor.fetchall() with open(latest_weather.json, w, encodingutf-8) as f: json.dump(rows, f, ensure_asciiFalse, indent2) conn.close()前端只需要一个fetch(latest_weather.json)就能渲染不需要引入任何框架。对于个人项目来说这套方案的性价比远高于搭一套完整后端。5.4 后续还能扩展的方向数据源本身还可以继续加厚接入实时分钟级数据、AQI空气质量、风向风力把原来“每天一次快照”升级成“每小时一次快照”。分析侧可以做天气异常告警比如某城市温度突破历史极值时推送通知也可以把历史气温数据接进时序数据库配合Grafana画出整年的气温变化曲线。我个人的建议是先把“稳定采集、不重不漏”这件事做到位再考虑可视化。数据采集是最底层的地基地基稳了上面盖什么建筑都容易。最后分享一个实际操作中的体会这套方案的代码量不大但真正让它变得可靠的是那些“看起来不起眼”的细节——超时重试、编码判断、唯一键约束、失败容错、日志输出。每一行都对应一次真实的踩坑。你如果准备自己搭一套不用一步到位先把主流程跑起来然后跑个三五天把日志翻一翻你会发现自己项目的稳定性和当初的Demo版本完全是两个东西。本文还有配套的精品资源点击获取