企业官网建设流程全解析

简介：由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标准库+requests+sqlite3的组合，城市清单内置在脚本里，采集逻辑和数据落库解耦，这样既能快速跑通，后续加功能也不至于推倒重来。

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, retries=3): url = f"https://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, headers=headers, timeout=10) 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秒）。timeout=10是必须的，没有超时的采集脚本一旦遇到连接挂起，整个任务会卡死在那个城市上，后面的城市全部排队等。

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="", encoding="utf-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里没有声明charset=utf-8，而requests默认按ISO-8859-1解码，中文自然乱码。

解决方法是拿到响应后先检查resp.encoding，再强制指定编码：

resp = requests.get(url, headers=headers, timeout=10) 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 2>&1

Windows环境下可以用任务计划程序，触发条件设为“每天”，操作指向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文件，前端用纯HTML+JavaScript读取，展示主要城市当天的天气卡片和温度对比。

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", encoding="utf-8") as f: json.dump(rows, f, ensure_ascii=False, indent=2) conn.close()

前端只需要一个fetch("latest_weather.json")就能渲染，不需要引入任何框架。对于个人项目来说，这套方案的性价比远高于搭一套完整后端。

5.4 后续还能扩展的方向

数据源本身还可以继续加厚：接入实时分钟级数据、AQI空气质量、风向风力，把原来“每天一次快照”升级成“每小时一次快照”。分析侧可以做天气异常告警，比如某城市温度突破历史极值时推送通知；也可以把历史气温数据接进时序数据库，配合Grafana画出整年的气温变化曲线。

我个人的建议是：先把“稳定采集、不重不漏”这件事做到位，再考虑可视化。数据采集是最底层的地基，地基稳了，上面盖什么建筑都容易。

最后分享一个实际操作中的体会：这套方案的代码量不大，但真正让它变得可靠的是那些“看起来不起眼”的细节——超时重试、编码判断、唯一键约束、失败容错、日志输出。每一行都对应一次真实的踩坑。你如果准备自己搭一套，不用一步到位，先把主流程跑起来，然后跑个三五天，把日志翻一翻，你会发现自己项目的稳定性和当初的Demo版本完全是两个东西。

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