去年底我接手了一个区域气候分析的小型项目需求听上去特别简单把过去三年每隔一小时的气温、湿度、气压、风速、降水量全部拉下来供后续建模用。结果真正动手才发现光数据这一关就折腾得我够呛。现成的气象公开API要么有请求次数限制要么字段粒度对不上好不容易找到一个能看历史数据的气象信息网页却发现数据分散在几千个详情页里手动复制粘贴到猴年马月也弄不完。后来我索性决定用 Python 网络爬虫做一套气候环境数据采集系统把“人工点点点”变成“定时自己跑”断断续续踩坑两周终于跑通了全流程。这篇文章就把这套系统的完整设计思路、实现细节和真实踩坑记录分享出来希望能帮你少走点弯路。这套系统适合谁参考如果你是做环境数据分析、农业气象研究、学术课题或者手头刚好有类似“定时获取大量网页数据”的需求却不知道从哪下手那这篇文章就是给你准备的。不需要你有多深的爬虫基础但最好有一点 Python 语法常识会装依赖包就够了。1. 数据拿到手的第一步往往不是写代码很多初学者听到“爬虫”两个字第一反应就是翻开教程写 requests、写 BeautifulSoup。但真正到了气候环境数据这种长期采集场景最不该急着动的就是代码。我自己的习惯是先花半天时间把“要什么、从哪拿、怎么拿、拿到怎么存”这四个问题想清楚后面写代码反而顺风顺水。1.1 真实场景为什么现成接口不够用市面上确实有一些天气服务商提供免费接口但实际用起来通常有几个痛点一是免费额度很小我当年调试了两次就触顶了等配额刷新要等一整天二是历史数据往往需要付费解锁免费只给当前天气三是返回字段固定比如有些接口只给温度和天气现象没有气压、没有逐时降水。可做气候分析恰恰最需要的就是长周期、多要素、连续稳定的历史序列。反过来说很多公开气象信息网站把历史观测数据放在网页上数据本身是公开的没有登录墙也没有验证码只是没有提供现成接口。这种场景下用 Python 网络爬虫定时去抓取反而成了最灵活的方案字段可以自己选频率可以自己定数据格式完全可控。1.2 公开API与自建爬虫的取舍我整理了一份当时做的对比直接看表更清楚对比维度现成API自建爬虫数据字段供应商预设调整空间小按需提取完全自定义历史深度通常受套餐限制只要页面能翻到就能抓到请求限制配额用尽即停只要合理控制频率可持续运行稳定性接口变更由对方负责页面改版需要自己跟法律合规按授权使用需遵守目标站点的服务条款与 robots 协议这表格不是劝你凡事都上爬虫。如果需求只是每天看一眼天气调用现成API显然是更明智的选择。但如果你要做的是“长期、高频、多字段”的环境数据积累自建采集系统的价值就体现出来了。我当时还特地确认过目标站点的服务条款和 robots.txt只抓公开数据、不碰登录接口这是底线。1.3 把目标拆成可验收的模块在动手前我把系统拆成了下面这几个模块每个模块都有明确的输入输出请求模块负责下载目标页面的 HTML处理超时、重试、请求头伪装解析模块从 HTML 中抽取温度、湿度、气压、风速、降水等字段清洗模块处理缺测值、异常值、单位换算统一时间格式存储模块把干净数据写入本地数据库调度模块每小时或每天定时触发采集任务日志模块记录每次采集的结果方便排障。不要小看这个拆分动作。爬虫最忌讳把所有代码堆在一个脚本里当时能用一个月后想改一个字段改一处坏一片。模块化之后每个环节出了问题单独看日志就能定位。2. 技术栈与目录设计一套能长期维护的骨架技术选型没有太多花哨我的原则是“用社区最成熟、文档最多、遇到问题能搜到答案的方案”。整套系统用的都是 Python 生态里非常基础但非常稳的库。2.1 技术选型为什么是这套组合核心依赖如下Python 3.10当前主流版本语法特性足够兼容性也好requestsHTTP 请求库比内置 urllib 好用太多处理重定向、超时、Session 都很顺手BeautifulSoup4 lxmlHTML 解析库CSS 选择器提取标签非常方便pandas做数据清洗、格式统一、缺失值处理比手写循环高效得多SQLite零配置文件型数据库单机采集完全够用数据量到千万级也扛得住APSchedulerPython 界的定时任务库支持 cron 表达式和固定间隔还能设置任务实例数防止重入。这些库如果没装直接一行命令搞定pip install requests beautifulsoup4 lxml pandas apscheduler顺便提一句环境配置的坑。很多新手在安装 Python 阶段就卡住了我见过最典型的错误是 Windows 下命令行输入 python 却弹出了 Microsoft Store这是因为系统 PATH 里没有指定 Python 安装路径。解决办法是安装时勾选 Add Python to PATH装完在命令行执行python --version确认版本。另外写完代码跑不起来的时候先检查是不是在用pip给 A 环境装包却用 B 环境的解释器跑代码这类环境错乱问题占了初学阶段报错的半壁江山。2.2 摸底数据源先看清页面再动手写代码前我花了大半天“研究”目标网页的结构。观察什么主要是三件事第一数据在页面里是直接渲染的还是通过异步接口加载的。最简单的判断方法是用浏览器右键查看网页源代码按 CtrlF 搜一个数据字段比如一个具体温度值。如果能在原始 HTML 里搜到说明可以用 requests 直接抓如果搜不到说明数据大概率是页面加载完后由 JavaScript 请求接口填充的那就还要再一步抓包找 XHR 接口。第二数据的分页/翻页规律。观察 URL 的变化模式是参数里带着 offset 还是 date这决定了我们循环请求的 URL 规则。第三每条观测数据对应的信息页链接藏在哪个标签里。大多数详情页链接都是标准的a href...但有些站点会套一层 JS 跳转这种情况解析时要额外处理。2.3 项目目录结构参考我最终的项目目录长这样供你参考weather_crawler/ ├── config.py # 全局配置目标URL、请求头、重试次数 ├── fetcher.py # 请求模块 ├── parser.py # 解析模块 ├── cleaner.py # 清洗模块 ├── storage.py # 存储模块 ├── scheduler_job.py # 定时任务入口 ├── requirements.txt # 依赖清单 └── logs/ └── crawler.logconfig.py 单独放的意义是未来站点 URL 变了、请求频率要调整、某字段要新增都只需要改配置文件不用动业务代码。这是长期可维护性的关键。3. 请求、解析、清洗三个最容易翻车的环节核心功能实现起来没有玄学就是按前面拆分好的模块一个个填代码。但每个模块里都有一些“坑”我一个个说。3.1 请求模块超时、重试与请求头缺一不可请求模块是整个系统的门面它要是罢工后面全是空转。我的 fetcher.py 核心逻辑如下import time import random import requests HEADERS { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/120.0 Safari/537.36, Accept-Language: zh-CN,zh;q0.9,en;q0.8, Accept: text/html,application/xhtmlxml,application/xml;q0.9,*/*;q0.8 } def fetch_page(url, retries3, timeout10): for attempt in range(retries): try: resp requests.get(url, headersHEADERS, timeouttimeout) if resp.status_code 200: return resp raise requests.RequestException(fHTTP {resp.status_code}) except requests.RequestException: wait_time 2 ** attempt random.uniform(0, 1) time.sleep(wait_time) return None这里有几个容易被忽略的点值得展开说。第一是timeout参数。很多初学者调用 requests.get 时不加 timeout结果就是目标站点响应慢的时候线程一直挂着不返回定时任务一个接一个堆积。设成 10 秒超时就重试宁可失败也不要干等。第二是User-Agent。不少网站会对默认的 python-requests 客户端直接返回 403。加一个浏览器的 User-Agent 是最基本的请求头伪装但如果你连续用一个固定的 UA 高频请求照样会被识别。更稳的做法是准备一个 UA 列表每次随机取一个USER_AGENTS [...] HEADERS[User-Agent] random.choice(USER_AGENTS)第三是指数退避重试。第一次失败后等 2 秒第二次等 4 秒左右第三次等 8 秒左右。这种策略比固定等 5 秒更合理对临时性网络抖动等得短一点快速恢复对目标站点可能已经报警的情况等得长一点避免火上浇油。另外两次请求之间我还会加一个 1 到 3 秒的随机 sleep模仿人的浏览节奏也避免把对方服务器打爆。3.2 解析模块从 HTML 中稳定抽取气象字段拿到 HTML 之后解析就轮到 BeautifulSoup 上场了。假设目标页面的表格结构是这样的table classweather-table trth时间/thth气温/thth湿度/thth气压/thth风速/thth降水量/th/tr trtd2025-01-01 00:00/tdtd5.2℃/tdtd78%/tdtd1013.4hPa/tdtd12.6km/h/tdtd0.0mm/td/tr /table解析代码可以这样写from bs4 import BeautifulSoup def parse_weather_table(html): soup BeautifulSoup(html, html.parser) rows soup.select(table.weather-table tr) result [] for row in rows[1:]: # 跳过表头 cells [cell.get_text(stripTrue) for cell in row.find_all(td)] if len(cells) 6: continue result.append({ time: cells[0], temperature: cells[1], humidity: cells[2], pressure: cells[3], wind_speed: cells[4], precipitation: cells[5] }) return result解析时最怕的是网页结构和预期不一致。所以我每次写完 parse 函数都会故意拿一页真实数据测试把所有字段打印出来人工核对。这里有一个实用技巧解析时尽量用稳定的属性定位比如table.weather-table这种 class 选择器而不要用body div table这种绝对路径因为后者只要页面多了一层 div解析就全断。3.3 清洗与校验别让脏数据污染分析结果解析出来的数据是“原始字符串”比如气温可能是5.2℃、湿度是78%这样的数据没法直接进模型。清洗模块用 pandas 会非常省力import pandas as pd def clean_weather_data(raw_list): df pd.DataFrame(raw_list) if df.empty: return df df[temperature] pd.to_numeric( df[temperature].astype(str).str.replace(℃, ), errorscoerce ) df[humidity] pd.to_numeric( df[humidity].astype(str).str.replace(%, ), errorscoerce ) df[pressure] pd.to_numeric( df[pressure].astype(str).str.replace(hPa, ), errorscoerce ) df[wind_speed] pd.to_numeric( df[wind_speed].astype(str).str.replace(km/h, ), errorscoerce ) df[precipitation] pd.to_numeric( df[precipitation].astype(str).str.replace(mm, ), errorscoerce ) df[observed_at] pd.to_datetime(df[time], errorscoerce) # 去掉时间和核心字段缺失的行 df df.dropna(subset[observed_at, temperature]) # 合理性校验湿度不可能小于0或大于100 df df[(df[humidity] 0) (df[humidity] 100)] # 气压合理范围通常在850~1100hPa df df[(df[pressure] 850) (df[pressure] 1100)] return dferrorscoerce的意思是转不了数字的字符串直接变成 NaN后续一起处理。这里关键的思路是爬虫拿到的数据不能“闭眼入库”一定要做一轮范围和合理性的校验否则等到分析阶段再发现异常值溯源就非常痛苦了。4. 入库与定时调度数据落地才叫系统解析完的数据如果只是躺在内存里关了电脑就没了。真正把它变成一个“系统”还需要让数据持久化下去并且定时自动化地跑起来。4.1 SQLite 表结构设计与写入策略我用 SQLite 做存储主要是图它轻量一个 .db 文件就能带走不需要单独装数据库服务。建表语句如下CREATE TABLE IF NOT EXISTS weather_observation ( id INTEGER PRIMARY KEY AUTOINCREMENT, station_id TEXT NOT NULL, observed_at TEXT NOT NULL, temperature_c REAL, humidity_percent REAL, pressure_hpa REAL, wind_speed_kmh REAL, precipitation_mm REAL, created_at TEXT DEFAULT CURRENT_TIMESTAMP, UNIQUE(station_id, observed_at) );这里最核心的是UNIQUE(station_id, observed_at)这个约束。它保证的是同一站点同一观测时刻最多只有一条记录后面重复采集也不会插重。这个约束在后续运行中帮我挡掉了大量重复数据。写入的代码也很简单使用 pandas 的to_sql方法import sqlite3 def save_to_sqlite(df, db_path, station_id): if df is None or df.empty: return 0 df df.copy() df[station_id] station_id # 只保留需要的列 df df[[station_id, observed_at, temperature_c, humidity_percent, pressure_hpa, wind_speed_kmh, precipitation_mm]] conn sqlite3.connect(db_path) try: # if_existsappend 表示追加写入 df.to_sql(weather_observation, conn, if_existsappend, indexFalse) finally: conn.close() return len(df)要注意的是UNIQUE约束在数据重复时会直接抛异常。实际使用中我用的是INSERT OR IGNORE的思路但 pandas 的to_sql没有直接提供这个选项。更稳妥的做法是先查库里已有的时间点把要插入的数据过滤一遍再写。数据量不大的时候这样最干净。4.2 APScheduler 定时调度与防重入采集任务本身不难难的是让它稳定地跑。这里我用到了 APScheduler可以精确控制采集频率。比如每小时采集一次from apscheduler.schedulers.blocking import BlockingScheduler from datetime import datetime def job(): print(f[{datetime.now()}] start collecting...) # 获取目标URL列表 # 逐个请求、解析、清洗、入库 scheduler BlockingScheduler() scheduler.add_job( job, interval, hours1, next_run_timedatetime.now(), max_instances1, coalesceTrue ) scheduler.start()max_instances1非常关键它的作用是如果上一次任务还没跑完下一次触发就直接跳过而不是开一个并发任务。气候数据采集有时候会遇到目标站点响应缓慢一个任务可能跑四五分钟如果没有这个参数任务就会叠起来轻则数据重复重则被目标站点封掉。coalesceTrue则是在错过多次调度后只补跑一次避免积压一堆任务。4.3 日志与监控故障时快速定位程序跑起来容易但你在睡觉的时候它悄悄挂了才是最头疼的。所以我给系统加了两道保险日志和告警。日志用 Python 自带的 logging 就行关键是要把每次请求的 URL、状态码、解析行数、入库条数都打出来import logging logging.basicConfig( filenamelogs/crawler.log, levellogging.INFO, format%(asctime)s %(levelname)s %(message)s ) def record(url, status, parsed, saved): logging.info(furl{url}, status{status}, parsed{parsed}, saved{saved})告警层面我用了一个“心跳文件”的思路每次成功入库后在指定目录写一个带时间戳的心跳文件然后用一个外部监控脚本比如 cron 任务检查心跳文件是否在最近两个小时内更新过。如果超过两小时没更新就发一封邮件通知自己。这套方案虽然没有专业监控平台那么花哨但胜在简单可靠跑了小半年没出过大漏子。5. 那些让采集器“突然罢工”的坑及排查过程系统上线头两周我几乎每天都会收到自己的告警邮件。每次排查都是一次经验积累我把其中最有代表性的三个问题整理出来每个都附上了完整的排查链路。5.1 第一坑页面编码导致的中文乱码某天我发现入库的数据里站点名称字段全是乱码像天气这种。第一反应是解析写错了仔细一查发现源页面返回的编码是gb2312而 requests 却按ISO-8859-1去解码了。排查过程是这样一步步走的先在浏览器开发者工具里看响应头的 Content-Type发现没有明确 charset再用 requests 打印resp.apparent_encoding得到的是GB2312最终确认问题根源是响应头缺失编码声明。解决办法是在请求时手动指定编码或者在解析前强制转码resp.encoding resp.apparent_encoding这里有个细节resp.text的取值时机很关键。如果你调用过resp.textrequests 就已经帮你解码了之后的resp.encoding修改不会再影响resp.text只能通过resp.content.decode(gb2312, errorsignore)重新取。所以我建议拿到响应后第一时间确定编码再决定用哪种方式取内容。这个坑我栽过后来在 fetcher 模块里固定写死先设置 encoding再取 text。5.2 第二坑字段错位解析出来的数据对不上表头第二次告警是因为入库数据出现了“湿度比气温还高”的荒唐值。我最初以为是清洗校验没生效后来打印出原始解析结果才发现某一页的表格结构跟之前的页面不一样——表头变了列的顺序换了有两个字段多了一个嵌套标签。BeautifulSoup 按td顺序取值自然就把温度和湿度放错了位置。这次的排查链路其实很短取一页最新数据把解析结果逐行打印出来和页面肉眼比对瞬间就看到了错位。但根本原因值得反思目标网站改版不是一次性的而是某些页面已更新、某些页面还是旧结构。这提醒我解析时必须加“字段位置校验”而不能默认表头永远不变。我的做法是解析时先读取表头列名把它映射到我们约定的字段名再按列名去取数据。这样即使列顺序变了只要列名还在就不会错位。def parse_weather_table_with_header(html): soup BeautifulSoup(html, html.parser) table soup.find(table, class_weather-table) header_cells table.find(tr).find_all([th, td]) header [c.get_text(stripTrue) for c in header_cells] columns [] for h in header: if 时间 in h: columns.append(time) elif 气温 in h: columns.append(temperature) elif 湿度 in h: columns.append(humidity) # ... 其他字段映射 rows table.find_all(tr)[1:] result [] for row in rows: cells [c.get_text(stripTrue) for c in row.find_all(td)] result.append(dict(zip(columns, cells))) return result5.3 第三坑目标站点返回 403被识别为机器人运行到第三周某天任务突然连续失败日志里清一色的 403。这算是爬虫生命周期里必然会遇到的问题对方服务器检测到高频同类请求决定拒绝服务。我的排查链路是这样的先确认不是自己代码报错——看日志发现请求确实发出去了响应确实返回了 403再人工打开目标网址浏览器能正常访问说明是服务器的反爬机制在起作用最后检查自己最近的请求频率发现因为之前加了指数退避重试某个失败瞬间会短时间内重试三次加上采集周期本身设置得太短累计请求量确实上去了。解决方案分几步拉长请求间隔把每小时一次改成每两小时一次请求头里随机换 User-Agent请求失败后不再连续快速重试而是把失败 URL 记到日志里等下一轮调度再补。这里要强调一点我的原则是不做绕过对方安全措施的事只是把自己这边的请求方式做得更合理、更克制。实际上只要采集频率控制在正常人浏览的范围内绝大多数公开数据站点都不会为难你。6. 后续扩展从一个站点到一片数据网络这套采集系统跑稳定后我的注意力就从“怎么不挂”转向了“怎么用得更好”。这里分享几个我觉得值得继续投入的方向这些也是我目前在逐步落地的功能。6.1 多站点并行采集与统一治理单一数据源始终有风险万一对方站点改版或者直接下线你的数据链就断了。更稳妥的做法是接入多个公开数据源同一个观测时刻的数据可以交叉验证。在实现上我用concurrent.futures.ThreadPoolExecutor控制并发度但这里有个非常重要的经验多站点并发不是线程越多越好。对同样的站点并发太多会被反爬对不同站点并发太多就失去了“模拟人手操作”的合理性。我自己的设置是线程池大小 3且同一个站点内部的请求还是保持串行加随机延时。多站点带来的另一个问题是数据结构不统一。有的站点给的是华氏温度有的给的是摄氏度有的风速给 m/s有的给 km/h。这些都需要在清洗模块里做归一化。我建议在数据库里增加两列source_site和raw_data存原始文本这样后续出现异常可以倒查原始数据而不是对着被清洗过的数字猜。6.2 从采集到可视化打通数据链路数据采集只是第一步采集之后的可视化和分析才是价值落地的环节。我现在会在采集完成后自动把当天的新数据追加到一个汇总 CSV然后用 Jupyter Notebook 做探索性分析画时间序列曲线、做相关性分析。这块使用的是 pandas 和 matplotlib代码量不大但收益很直观之前需要手动导数据、清洗、画图的流程现在全部自动化了。如果你有更复杂的分析需求还可以把 SQLite 的数据接入到 pandas 后直接跑 sklearn 模型比如用历史气象数据训练一个短时气温预测模型。到这里爬虫的角色就从“抓数据的小工具”变成了“数据管道的第一环”。6.3 数据质量评估与异常检测系统跑久了我还有个新体会长期采集的数据质量评估比数量积累更重要。数据链路里可能出现设备临时故障、站点维护、网络超时等导致的缺测也可能是站点本身的数据源出了问题。这些异常在单条记录里很难发现但在时间序列里一目了然。我现在做了一个简单但非常有效的数据质量日报每次采集并入库后按站点和日期统计缺失率、最大值、最小值、均值如果某个站点的某天气温最大最小值之差小于 1 度或者缺失率突增到 30% 以上就把这个站点标记为“待人工核查”并自动把这些异常汇总到每周邮件里。这套机制帮我发现过两次目标站点数据更新异常的问题比等到分析阶段再发现要省心得多。回过头来看这套基于 Python 网络爬虫的环境数据采集系统本质上是把一件“枯燥重复但逻辑清晰”的工作自动化了。真正的收获不在于写了几百行代码而在于想清楚了请求、解析、清洗、存储、调度、监控这一整条链路的每一步为什么这样设计。如果你也在搭类似的采集系统我的建议是先把“怎么不挂”想明白再考虑“怎么抓更多”。控制频率、做好日志、设计好幂等约束这三件事做好了系统基本就稳定了。