前言
看爬虫教程最容易产生一种错觉:每一段代码单独看都懂,合起来放到一个能跑的脚本里就乱了。「实例详解」要解决的就是这个落差——把一个真实可运行的小爬虫拆成几层,逐层说清它为什么这样写。
本文用一个具体例子把完整链路走一遍:发请求(带超时与重试)、解析列表页、识别分页、限速、把结果写成 CSV。为了让例子在没有任何第三方包的环境里也能跑,解析部分只用标准库的html.parser。但这同时带来一个要提前说清的局限:HTMLParser是事件式的,它不帮你把 HTML 修成规范的树,所以下面这个解析器是针对某一个具体页面结构写的。换一个站点、甚至同一个站点改版,都要重写解析逻辑,这是正常且必然的。
例子抓的是quotes.toscrape.com,一个专门为练习爬虫搭建的站点。选它不是因为「大站不能抓」,而是因为它结构稳定、明确用于教学,比拿商业网站练手合适。真实项目里,动手前请先读目标站的 robots.txt 与使用条款,并控制请求频率。
所有代码以 Python 3 为基准。Python 2.7 已于 2020 年 1 月 1 日停止维护,print语句、dict.iteritems()、xrange这些写法现在都不能用了。
一、先设计数据,再写代码
写爬虫最容易犯的错误是先动手抓,抓到一半才发现「这个字段我忘了取」。正确顺序是先定下你要的条目(item)长什么样。这个例子要的每条记录有三个字段:
| 字段 | 含义 | 类型 | 缺失时怎么办 |
|---|
text | 名言正文 | str | 该条丢弃 |
author | 作者 | str | 用空串占位 |
tags | 标签 | list[str] | 用空列表 |
用字典表示一条记录即可:{"text": ..., "author": ..., "tags": [...]}。字段定下来,解析器的任务就变成「把 HTML 里的片段映射到这几个键」,边界清晰。
二、请求层:超时与重试必须显式写
urllib.request.urlopen的timeout参数不写就不会超时——网络卡住时程序会一直挂着。所以每次请求都要给超时。重试也要分清对象:
- 404:页面不存在,重试没有意义,直接放弃这条。
- 403:对方明确拒绝,重试只会加重问题,应当停止并检查自己的行为。
- 429:请求过多,应当退避后重试,而不是立刻再来。
- 网络层错误(连接超时、DNS 失败):值得重试,但要等一会儿。
urllib.error里有两个关键异常:URLError是网络层错误的基类,HTTPError是它的子类,代表「服务器返回了错误状态码」——它同时也是类文件对象,可以读取响应体。退避不要用固定间隔,用指数退避:第一次等 2 秒,第二次等 4 秒,以此类推。
三、解析层:用状态机把片段归位
HTMLParser的用法是继承它、重写handle_starttag(tag, attrs)、handle_data(data)、handle_endtag(tag),然后feed(text)喂入文本。这里有两个细节必须记住:
attrs是「(属性名, 属性值) 二元组」组成的列表,不是字典。想按名字取值要先dict(attrs)。同一个属性也可能出现多次,用dict()会只留最后一个。feed()之后必须调close(),否则文档末尾的缓冲数据不会处理完。
这段 HTML 的骨架大致是「一个卡片 div,里面包含正文 span、作者 small、若干 tag 链接,卡片内还嵌套了一个标签 div」。嵌套的 div 是关键陷阱:如果一见到</div>就以为卡片结束了,会在内层标签 div 处提前收尾。解决办法是用一个深度计数器,进入卡片时置 1,每遇到一个内层 div 加 1,遇到</div>减 1,减到 0 才算卡片真正闭合。
四、分页与终止条件
列表页的「下一页」是一个链接。解析出它的href后,用urllib.parse.urljoin(base, href)拼成绝对地址——不要手工拼接,也不要假设href已经是完整的。终止条件不能只看「有没有下一页」:还要设一个页数上限,否则一旦页面链接出现环状结构,循环就不会停。
实战:一个完整的采集脚本
# 适用于 Python 3.8+
import csv
import time
import urllib.error
import urllib.parse
import urllib.request
import urllib.robotparser
from html.parser import HTMLParser
START = "https://quotes.toscrape.com/"
UA = "quote-walkthrough/0.1 (contact: walkthrough@example.com)"
DELAY = 1.0 # 每页之间的间隔(秒)
MAX_PAGES = 3 # 页数硬上限,防止无限翻页
class CardCollector(HTMLParser):
"""把列表页解析成若干条 {text, author, tags}。"""
def __init__(self):
super().__init__()
self.items = []
self.next_href = None
self._current = None # 当前卡片
self._tags = []
self._depth = 0 # 卡片内的 div 嵌套深度
self._field = None # 正在收集的字段
self._buf = []
self._in_next = False
def handle_starttag(self, tag, attrs):
classes = dict(attrs).get("class", "").split()
if tag == "li" and "next" in classes:
self._in_next = True
return
if tag == "a" and self._in_next:
href = dict(attrs).get("href")
if href:
self.next_href = href
return
if tag == "div":
if self._current is None and "quote" in classes:
self._current = {}
self._tags = []
self._depth = 1
elif self._current is not None:
self._depth += 1
return
if self._current is None:
return
if tag == "span" and "text" in classes:
self._field, self._buf = "text", []
elif tag == "small" and "author" in classes:
self._field, self._buf = "author", []
elif tag == "a" and "tag" in classes:
self._field, self._buf = "tag", []
def handle_data(self, data):
if self._field is not None:
self._buf.append(data)
def handle_endtag(self, tag):
if tag == "li":
self._in_next = False
return
if self._field is not None and tag in ("span", "small", "a"):
value = "".join(self._buf).strip()
if self._field == "text":
self._current["text"] = value
elif self._field == "author":
self._current["author"] = value
else:
self._tags.append(value)
self._field = None
return
if tag == "div" and self._current is not None:
self._depth -= 1
if self._depth == 0:
self._current["tags"] = self._tags
self.items.append(self._current)
self._current = None
def fetch(url, retries=3):
"""取回文本;对可重试的错误做指数退避。"""
last_error = None
for attempt in range(1, retries + 1):
try:
req = urllib.request.Request(url, headers={"User-Agent": UA})
with urllib.request.urlopen(req, timeout=15) as resp:
charset = resp.headers.get_content_charset() or "utf-8"
return resp.read().decode(charset, errors="replace")
except urllib.error.HTTPError as exc:
if exc.code in (403, 404): # 明确答复,重试无意义
raise
last_error = exc
except urllib.error.URLError as exc:
last_error = exc
if attempt < retries:
time.sleep(2 ** attempt) # 2 秒、4 秒……
raise last_error if last_error else RuntimeError("请求失败")
def robots_allowed(url):
parts = urllib.parse.urlsplit(url)
robots_url = urllib.parse.urlunsplit(
(parts.scheme, parts.netloc, "/robots.txt", "", "")
)
parser = urllib.robotparser.RobotFileParser()
parser.set_url(robots_url)
try:
parser.read()
except Exception:
return False # 拿不到 robots 时保守处理
return parser.can_fetch(UA, url)
def main():
if not robots_allowed(START):
print("robots.txt 不允许抓取,退出")
return
items = []
url = START
for page in range(1, MAX_PAGES + 1):
html = fetch(url)
parser = CardCollector()
parser.feed(html)
parser.close()
print("第 %d 页解析到 %d 条" % (page, len(parser.items)))
items.extend(parser.items)
if not parser.next_href:
break
url = urllib.parse.urljoin(url, parser.next_href)
time.sleep(DELAY) # 翻页前限速
with open("quotes.csv", "w", encoding="utf-8", newline="") as f:
writer = csv.writer(f)
writer.writerow(["text", "author", "tags"])
for item in items:
writer.writerow([item["text"], item["author"], "|".join(item["tags"])])
print("共写入", len(items), "条")
if __name__ == "__main__":
main()跑完之后得到一个quotes.csv。这里有几个刻意的设计:next_href为None时提前跳出;即使还有下一页,MAX_PAGES也会兜底;写文件用了encoding="utf-8"和newline=""——后者是csv模块官方文档明确要求的,少了它会在 Windows 上多出空行。
常见坑点
1.open()写 CSV 时忘了newline=""
❌open("out.csv", "w", encoding="utf-8")—— 在 Windows 上每行之间会多出一行空行。 ✅open("out.csv", "w", encoding="utf-8", newline=""),这正是csv文档的做法。
2.open()不写encoding
❌open("out.csv", "w")—— 用的是平台默认编码,Windows 上可能是 GBK,中文与跨平台读取都会出问题。 ✅ 显式写encoding="utf-8"。
3. 用dict(attrs)取重复属性
❌ 以为dict(attrs)能拿到全部属性 —— 同名属性只保留最后一个。 ✅ 需要严格处理时遍历attrs列表本身;多数场景dict(attrs)够用,但要清楚它的取舍。
4. 嵌套 div 导致卡片提前闭合
❌ 一遇到</div>就当作卡片结束,结果标签 div 一闭合就把半成品条目丢进结果里。 ✅ 用深度计数器,减到 0 才收尾。
5.feed()之后忘了close()
❌ 只parser.feed(html)就读parser.items。 ✅ 补上parser.close(),让缓冲数据被处理完。
6. 只判断「有没有下一页」,不设页数上限
❌while True:里跟着 next 链接走 —— 一旦链接成环就无限抓。 ✅ 加MAX_PAGES上限,并同时判断 next 链接是否存在。
7. 手工拼接分页 URL
❌url + "page/" + str(n) + "/"—— 少个斜杠、多个斜杠都会 404。 ✅urllib.parse.urljoin(url, href)。
8.timeout不写
❌urlopen(req)—— 连接卡死时程序永远挂着。 ✅urlopen(req, timeout=15),并按错误类型决定是否重试。
总结
| 层次 | 要做的事 | 关键写法 |
|---|
| 数据设计 | 先定字段,再写解析 | 字典 item 三字段 |
| 请求 | 超时 + 分类重试 | timeout=15、指数退避 |
| 解析 | 片段归位,处理嵌套 | 深度计数器 +close() |
| 分页 | 绝对化 + 硬上限 | urljoin+MAX_PAGES |
| 限速 | 每页之间留间隔 | time.sleep(DELAY) |
| 落盘 | UTF-8 且不留空行 | encoding="utf-8",newline="" |
| 合规 | 先查 robots | RobotFileParser.can_fetch |
一个爬虫实例的难点从来不在「怎么发请求」,而在边界处理:嵌套结构怎么收尾、异常怎么分类、循环怎么终止、编码与换行怎么统一。把这几处想清楚,代码自然就短。最后提醒一句:页面结构会变,别人站点的解析规则在你手里可能明天就失效,写解析器时留好「取不到字段」的分支,比写得漂亮更重要。