ARTICLE DETAIL

资讯详情

深耕网站视觉设计与运营推广的一线实战洞察。

Python爬虫实战:requests+BeautifulSoup抓取全国天气数据

Python爬虫实战:requests+BeautifulSoup抓取全国天气数据 入行爬虫的朋友我几乎都会推荐先拿天气网站练手。原因很简单数据公开免费、页面结构相对规整、更新频率固定既能把 requests、BeautifulSoup、数据存储这一整套流程跑通又不会因为追着某个小众目标死磕而打击信心。这个项目表面上只是“拿全国城市七日天气”实际上把网络请求、HTML 解析、异常处理、数据落地这些爬虫核心环节全过了一遍做完以后独立抓任何结构化网页都有了底气。适合刚学完 Python 基础语法、想找一个完整项目串联知识点的同学也适合写了几个小脚本但对“单机爬虫怎么设计得稳”还没有体感的人。我把整个开发过程拆成四块讲思路设计、数据源分析、代码落地、问题排查。每块都会说清楚为什么这么做代码环节直接给出可以跑通的实现。1. 项目背景与核心思路1.1 为什么选天气网站当练手项目很多人一上来就盯电商、社交平台结果被验证码、登录态、风控系统劝退。天气网站不一样它有几个天然适合初学者的特征公开数据访问门槛低。天气预报本来就是公共服务不需要登录不需要复杂的 token 体系。数据维度清晰。城市、日期、天气现象、最高温度、最低温度、风力风向这几个字段是固定的完美对应结构化数据训练。页面结构规整。天气网站基本都是表格或卡片式布局DOM 结构重复性高非常适合练 select 定位。实时校验效果直观。抓完和手机天气 App 对着看数据对不对一眼就能判断排错路径非常短。它的复杂度又恰好卡在一个“跳一跳才够得着”的位置多城市调度、频率控制、编码处理、反爬应对这些生产环境才会遇到的问题在这个项目里都会出现但都不会难到让人崩溃。1.2 技术选型requests BeautifulSoup 还是 Scrapy这个项目我推荐用requests BeautifulSoup而不是 Scrapy。原因不是 Scrapy 不好而是阶段不匹配。Scrapy 的优势在于异步调度、内置去重、Pipeline 管理这些是构建大规模采集框架才需要的。七日天气预报的体量几百个城市撑死也就几千个请求用 Scrapy 属于给自行车装涡轮增压——跑得快但没必要而且它的异步机制和 Item Pipeline 概念会分散你对“请求-解析-存储”这条主线的注意力。requests BeautifulSoup 的链路最直观requests.get(url, headersheaders) # 发请求 BeautifulSoup(html, html.parser) # 解析 soup.select(div.temp) # 定位 data.to_csv(weather.csv) # 存储每一个环节都是 Python 爬虫最底层的肌肉记忆搞懂这一套以后切 Scrapy 会发现所有概念都能映射过来。至于数据量放大之后再考虑分布式、消息队列那些东西那是后话。注意这个项目默认运行在本地单机环境掉线重试、失败重采都设计在单进程内完成。如果你需要每天自动更新配合 cron 或 Windows 计划任务就够了不需要额外引入消息队列。2. 数据源分析与抓取策略2.1 目标页面结构拆解以国内常见的天气数据页面为例打开开发者工具F12切到 Network 面板找到返回当前城市天气数据的那个请求观察响应内容。你会发现大部分天气网站的 HTML 结构都有类似套路城市 ID 通常带在 URL 里例如路径中的101010100这种编码七日天气数据在一个容器节点下每一天的数据卡片具有相同 class温度、天气现象、风力分别由不同标签承载我用一个简化示例说明div classdaily-item>soup.select(div.daily-item)如果同 class 在页面里出现了多次再用ul.weather-list li这种父子选择器缩小范围。少用那种又长又脆弱的绝对路径页面改一个层级就会全线崩盘。2.2 请求头伪装与反爬应对天气网站的反爬强度总体不高但也不是完全不设防。最常见的几个拦截点User-AgentUA检测。很多服务器会先看 UA如果发现是爬虫库的默认标识就直接拒绝。requests 默认的 UA 是python-requests/x.x.x一眼假。最简单的处理方式是伪造一个浏览器 UA。我在项目里用的是一份真实的 Chrome UA实测下来稳定度很高。直接把这条复制进去不用额外安装 fake_useragent 库少一个依赖就少一个坑。Referer 校验。部分页面会检查请求来源直接访问 URL 没问题但如果带上了外部来源可能会被 403。解决办法很朴素把Referer设为同站首页。频率限制。这是最容易被忽略的。很多初学者循环抓几百个城市不加任何间隔跑几十个就被限流。我在批量调度里强制每两次请求之间 sleep 1 到 3 秒这是纯防御性设计——宁可多等几分钟也不要触发对方的封禁策略。动态渲染。有些天气站的核心数据是 JavaScript 异步加载的直接请求 HTML 拿不到天气信息。这种情况下的处理思路是切换数据源优先找返回 JSON 的接口JSON 解析比 HTML 解析省心得多也不怕页面改版。我实际项目里就是两种方案都做了优先 JSON 接口拿不到再退回到 HTML 解析。3. 核心代码实现与逐步拆解3.1 单城市七日天气抓取模块先写最核心的函数给定一个城市 ID返回该城市未来七天的天气列表。这个函数是整个项目的地基后面批量调度、存储、异常处理都建立在它上面。import requests from bs4 import BeautifulSoup import re from typing import List, Dict 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, Referer: https://www.example-weather.com/, } def fetch_weather(city_id: str) - List[Dict]: url fhttps://www.example-weather.com/weather/{city_id}.shtml try: resp requests.get(url, headersHEADERS, timeout10) resp.raise_for_status() except requests.RequestException as e: print(f[{city_id}] 请求失败: {e}) return [] # 解决编码问题天气站多为 utf-8但有的老页面是 gbk if resp.encoding and resp.encoding.lower() gbk: resp.encoding gbk else: resp.encoding utf-8 soup BeautifulSoup(resp.text, html.parser) items soup.select(div.daily-item) result [] for item in items[:7]: # 只要七天的数据 date_tag item.get(data-date) weather_tag item.select_one(span.weather) temp_tag item.select_one(span.temp) wind_tag item.select_one(span.wind) if not (date_tag and weather_tag and temp_tag): continue # 从 -2℃ / 8℃ 中拆分最高温和最低温 temp_text temp_tag.text.strip() match re.search(r(-?\d)\s*℃?\s*/\s*(-?\d)\s*℃?, temp_text) low, high match.groups() if match else (None, None) result.append({ city_id: city_id, date: date_tag, weather: weather_tag.text.strip(), low_temp: low, high_temp: high, wind: wind_tag.text.strip() if wind_tag else , }) return result几个细节值得说透。关于timeout。这个参数必须写。不设 timeout 的情况下如果网站响应异常慢程序会一直挂在那里批量跑的时候一个卡住全盘停滞。设了 10 秒超时直接跳过并记日志这才是“稳”的体现。关于raise_for_status()。这个方法会在返回状态码是 4xx 或 5xx 时抛出异常配合 try-except 可以捕捉到 404、503 这类错误。不要只看status_code然后手动判断那会多写很多无关代码。关于编码处理。天气站页面有 utf-8 也有 gbk直接resp.text可能会乱码。requests库会从响应头猜测编码但脚本不可靠。通用做法是先看resp.encoding如果 HTTP 头没声明编码就用apparent_encoding兜底if not resp.encoding or resp.encoding.lower() iso-8859-1: resp.encoding resp.apparent_encoding3.2 全国城市列表与调度逻辑单城市跑通之后下一步就是全国城市。这里有个策略选择手动维护城市 ID 列表还是自动爬取城市列表我的建议是手动维护一份核心城市列表。全国地级市有 300 多个但实际需要监控的通常是省会城市、计划单列市和热点旅游城市五十个以内足够覆盖大部分使用场景。手动维护的好处是可控性强不会因为城市列表页改版导致整个调度崩掉。城市 ID 数据格式如CITY_LIST [ {name: 北京, id: 101010100}, {name: 上海, id: 101020100}, {name: 广州, id: 101280101}, {name: 深圳, id: 101280601}, # ... 更多城市 ]批量调度代码核心是一个循环遍历城市列表、抓取、写入、休眠、异常兜底。import time import json from datetime import datetime def run_spider(cities: List[Dict], output_path: str weather_data.jsonl): all_results [] for i, city in enumerate(cities): print(f正在抓取 {city[name]} ({i1}/{len(cities)})) data fetch_weather(city[id]) if not data: print(f - {city[name]} 无数据跳过) time.sleep(2) continue all_results.extend(data) # 每完成一个城市增量写入文件防止程序中断丢数据 with open(output_path, a, encodingutf-8) as f: for item in data: f.write(json.dumps(item, ensure_asciiFalse) \n) # 请求间隔降低被封风险 time.sleep(1.5) print(f完成共抓取 {len(all_results)} 条记录)这里用 JSONL每行一条 JSON而不是一个大 JSON 数组是因为增量写入时 JSONL 更稳健——即使中途断掉已写入的行仍然可以读取。这也是生产环境里日志和事件流的通用做法。3.3 数据清洗与结构化存储抓下来的是原始数据直接存肯定不行。以温度为例不同页面可能返回“-2℃ / 8℃”“最低-2℃最高8℃”“-2~8℃”三种格式解析规则必须统一。我在上面用正则做了一次拆分但正则也有局限天气现象里的“多云转晴”如果后续要按维度统计还得做映射。实际项目里我设计了几个清洗步骤第一步字段标准化。把low_temp、high_temp转成整数日期统一成YYYY-MM-DD风力描述去掉多余空格。def clean_item(item: Dict) - Dict: if item.get(low_temp): item[low_temp] int(item[low_temp]) if item.get(high_temp): item[high_temp] int(item[high_temp]) item[date] item[date].strip() item[weather] item[weather].strip() return item第二步去重。如果某天程序中断重跑同一城市同一日期的数据会被写两次。用(city_id, date)作为去重键写入前检查已有数据。def deduplicate(items: List[Dict]) - List[Dict]: seen set() unique_items [] for item in items: key (item[city_id], item[date]) if key in seen: continue seen.add(key) unique_items.append(item) return unique_items第三步存储格式选择。我最终选了 SQLite 而不是 CSV。CSV 适合一次性分析但七日数据每天更新、需要按城市查询SQLite 更合适。import sqlite3 def init_db(db_path: str weather.db): conn sqlite3.connect(db_path) conn.execute( CREATE TABLE IF NOT EXISTS daily_weather ( city_id TEXT, date TEXT, weather TEXT, low_temp INTEGER, high_temp INTEGER, wind TEXT, PRIMARY KEY (city_id, date) ) ) conn.commit() return conn def insert_data(conn, items: List[Dict]): conn.executemany( INSERT OR REPLACE INTO daily_weather (city_id, date, weather, low_temp, high_temp, wind) VALUES (:city_id, :date, :weather, :low_temp, :high_temp, :wind), items ) conn.commit()INSERT OR REPLACE配合主键(city_id, date)天然实现了幂等写入重复跑不会产生脏数据。这个设计决策省了我后面非常多的排查力气。4. 常见问题与排查技巧实录4.1 编码乱码问题症状与正解爬虫新手最常撞见的就是“打印出来一堆Ã¥ÂÂ这种乱码”。这个问题的根源是响应的二进制数据用了某种编码但你按另一种编码去解码了。排查路径很固定先打印resp.encoding和resp.apparent_encoding再手动指定。print(resp.encoding) # HTTP 头声明的编码可能为空 print(resp.apparent_encoding) # 基于内容的编码探测结果实测中有些服务器返回的 HTTP 头里没有charset字段requests 会默认设为ISO-8859-1这也就是乱码的根源。解决办法是强制指定resp.encoding utf-8或者更稳妥的resp.encoding resp.apparent_encoding这个“探测再解码”的思路不仅适用于天气站所有中文爬虫项目通用。4.2 请求超时与偶发 503批量调度跑了一半突然连续报错这是典型的“访问频率过高被临时限流”。我踩过一次很深的坑某次优化觉得 1.5 秒间隔太保守改成 0.3 秒结果跑到第 70 个城市开始大面积 503最后被封了十几分钟。从那以后我的策略是默认 1.5 秒夜间空闲时段可放宽到 0.8 秒但绝不低于 0.5 秒。同时加一个指数退避重试逻辑from time import sleep def robust_fetch(city_id: str, max_retries: int 3): for attempt in range(max_retries): try: resp requests.get(url, headersHEADERS, timeout10) if resp.status_code 503: raise ValueError(服务暂时不可用) resp.raise_for_status() return resp except Exception as e: wait 2 ** attempt # 1s, 2s, 4s print(f第 {attempt1} 次失败: {e}, {wait} 秒后重试) sleep(wait) return None指数退避的意思就是每次重试都等更久第一次 1 秒第二次 2 秒第三次 4 秒。这比固定间隔重试更贴合服务器的限流逻辑。4.3 页面改版导致的选择器失效这是所有爬虫都逃不掉的宿命今天跑得好好的明天页面改了 class 名select(div.daily-item)返回空列表。我的处理经验是三层防线第一层解析结果数量校验。如果返回的列表长度为零立即告警而不是默默输出空文件。if len(items) 0: raise RuntimeError(f[{city_id}] 未找到天气数据节点页面结构可能已变化)第二层抓取前自动探测。项目启动时先抓一个测试城市用soup.find探测几个关键 class 是否存在。如果探测失败程序直接停止并提示人工检查避免用错误的解析规则跑完全部城市。第三层保留访问凭证。页面大改版时直接看浏览器里正常显示的页面对应的 DOM 节点比对旧代码里的 select 路径通常只改 class 前缀就够了。把解析规则集中在代码头部的一个配置区更利于快速修改。4.4 分布运行时常见的坑虽然这个项目单机就能跑但多开几个进程或扩展到服务器时会暴露两类问题线程安全问题。我最初跑多线程版本时每个线程往同一个 SQLite 连接写数据结果频繁报database is locked。解决办法是每个线程独享一个连接或者写入操作用一个队列串行化。对于天气这种数据量开 10 个线程收益有限2 到 3 个加一个共享写入锁就够用了。时区问题。服务器运行在 UTC 时区时datetime.now()存进数据库会比北京时间早 8 小时。应对办法是统一用date字段对应目标城市当地日期不依赖服务器时间。这也是为什么我在数据结构里强调date必须从页面获取而不是用datetime.now()拼一个。5. 项目扩展方向与进阶建议5.1 从“抓一次”到“自动更新”七日天气数据的价值在于持续更新。把主脚本包成一个main()入口然后用系统定时任务驱动是最快的升级路径。Linux 下的 cron 配置示例每天早晨 7 点 30 分更新30 7 * * * /usr/bin/python3 /path/to/weather_spider.py /var/log/weather_spider.log 21Windows 下则用任务计划程序触发器设为“每天”操作里指向python.exe和脚本路径。注意两点一是 Python 解释器最好用绝对路径二是脚本内部日志一定要输出到文件定时任务环境下控制台输出是看不到的。5.2 从“存数据”到“用数据”数据落地之后不要停在“存起来”这一步继续做消费才划算。我后来把 SQLite 里的数据接到两个场景趋势可视化。用 pandas 读取数据matplotlib 画一个“未来七日温度折线图”城市之间最高温对比一眼就能看出冷空气路径。这里有个 pandas 读取的小技巧SQLite 里的日期字符串直接用pd.to_datetime()转换温度列用astype(int)处理完再画图避免 Object 类型导致的绘图报错。异常天气提醒。简单规则匹配比如weather字段包含“暴雨”“暴雪”“台风”时触发邮件或企业微信机器人通知。这个用 Python 标准库smtplib就能实现不需要额外的推送框架。5.3 合规红线不能碰最后必须说一句写爬虫要有边界意识。这个天气项目数据本身是公开公共服务信息采集频率也控制在合理范围这是安全的。但如果把这个框架原样套到需要登录、有用户协议保护、接口有签名加密的网站风险完全不同。几个底线原则遵守 robots.txt虽然它没有强制法律效力但体现了访问意图控制频率不要让服务器感受到压力只采公开数据不去绕验证码、破解签名不用于商业牟利尤其是天气预报这类公共服务数据转卖可能涉及法律问题我在实际项目里还会在代码里加一层访问白名单只有授权任务能调用数据接口存储时对涉及个人信息的字段做脱敏处理。这不是形式主义是让自己养成“数据有主权”的敏感度。这个项目抓完以后我个人最大的收获不是“会写爬虫了”而是建立了一套调试思路URL 不对先看完整请求链接解析不到内容先看响应文本批量出问题先看日志规律。把天气网站跑顺了以后再去碰数据接口、异步渲染、分布式框架本质都是同一个解决问题的链路。最后再分享一个小技巧把每个城市的城市 ID 列表单独放在一个 Python 文件里不要和主逻辑混在一起。后面不管是加城市还是配合外部配置文件生成任务清单都会省很多事。
返回列表