☰
Python爬虫实战:requests+BeautifulSoup抓取全国天气数据
2026/10/2 9:53:30 网站建设 项目流程

入行爬虫的朋友,我几乎都会推荐先拿天气网站练手。原因很简单:数据公开免费、页面结构相对规整、更新频率固定,既能把 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, headers=headers) # 发请求 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 class="daily-item">soup.select("div.daily-item")

如果同 class 在页面里出现了多次,再用ul.weather-list > li这种父子选择器缩小范围。少用那种又长又脆弱的绝对路径,页面改一个层级就会全线崩盘。

2.2 请求头伪装与反爬应对

天气网站的反爬强度总体不高,但也不是完全不设防。最常见的几个拦截点:

User-Agent(UA)检测。很多服务器会先看 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 = f"https://www.example-weather.com/weather/{city_id}.shtml" try: resp = requests.get(url, headers=HEADERS, timeout=10) 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_encoding

3.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']} ({i+1}/{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", encoding="utf-8") as f: for item in data: f.write(json.dumps(item, ensure_ascii=False) + "\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, headers=HEADERS, timeout=10) 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"第 {attempt+1} 次失败: {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 2>&1

Windows 下则用任务计划程序,触发器设为“每天”,操作里指向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 文件里,不要和主逻辑混在一起。后面不管是加城市还是配合外部配置文件生成任务清单,都会省很多事。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询