- 爬取网站地图；
- 遍历每个网页的数据库ID；
- 跟踪网页链接。
下载网页
要想爬取网页，我们首先需要将其下载下来。下面的示例脚本使用Python3的urllib.request模块下载URL。
import urllib.request
def download(url):
return urllib.request.urlopen(url).read()
if __name__ == '__main__':
当传入URL参数时，print会输出download函数获取的网址源码。不过，这个代码片段存在一个问题，即当下载网页时，我们可能会遇到一些无法控制的错误，比如请求的页面可能不存在。此时，urllib会抛出异常，然后退出脚本。安全起见，下面在给出一个更健壮的版本，可以捕获这些异常。
def download(url):
"""捕获错误的下载函数"""
print("Downloading:", url)
try:
html = urllib.request.urlopen(url).read()
except urllib.request.URLError as e:
print("download error:", e.reason)
html = None
return html
现在，当出现下载错误时，该函数能够捕获到异常，然后返回None。
重新下载
def download(url, num_retries=2):
"""下载函数，也会重试5xx错误。参数二为重试次数，默认为2次"""
print("Downloading", url)
try:
html = urllib.request.urlopen(url).read()
except urllib.request.URLError as e:
print("Download error:", e.reason)
html = None
if num_retries > 0:
if hasattr(e, 'code') and 500 <= e.code < 600:
#重试 5xx http错误
html = download3(url, num_retries-1)
return html
设置用户代理
def download4(url, user_agent='wswp', num_retries=2):
"""包括用户代理支持的下载函数"""
print("Downloading:", url)
headers = {'User-agent': user_agent}
request = urllib.request.Request(url, headers=headers)
try:
html = urllib.request.urlopen(request).read()
except urllib.request.URLError as e:
print("Download error:", e.reason)
html = None
if num_retries > 0:
if hasattr(e, 'code') and 500 <= e.code < 600:
# 重试 5xx http错误
html = download4(url, user_agent, num_retries-1)
return html
现在，我们拥有了一个灵活的下载函数，可以在后续示例中得到复用。该函数能够捕获异常、重试下载并设置用户代理。
网站地图爬虫
在第一个简单的爬虫中，我们将使用示例网站robot.txt文件中发现的网站地图来下载所有网页。为了解析网站地图，我们将会使用一个简单的正则表达式，从<loc>标签中提取出URL。
import re
from common import download
def crawl_sitemap(url):
#下载sitemap文件
sitemap = download(url).decode('utf-8')
#抓取站点地图链接
links = re.findall('<loc>(.*?)</loc>', sitemap)
# 下载每一个链接
for link in links:
html = download(link)
# scrape html here
# ...
现在，运行网站地图爬虫，从示例网站中下载所有国家页面。
>>>
ownloading:
Downloading:
Downloading:
Downloading:
Downloading:
...
可以看出，上述运行结果和我们的预期一致，不过正如前文所述，我们无法依靠Sitemap文件提供每个网页的链接。下一节中，我们将会介绍另一个简单的爬虫，该爬虫不在依赖于Sitemap文件。
ID遍历爬虫
本节中，我们将利用网站结构的弱点，更加轻松地访问所有内容。下面是一些示例国家的URL。
- Downloading:
- Downloading:
- Downloading:
可以看出，这些URL只是在结尾出有所区别。包括国家名（作为页面别名）和ID。在URL中包含页面别名是非常普遍的做法。可以对搜索引擎优化起到帮助作用。一般情况下，Web服务器会忽略这个字符串，只使用ID来匹配数据库中的相关记录。下面我们将其移除，加载，测试示例网站中的链接是否仍然可用。如果我们加载成功，证明该方法是有用的。下载，我们就可以忽略别名。只遍历ID来下载所有国家的页面。下面是使用了该技巧的代码片段。
# -×- coding: utf-8 -*-
import itertools
from common import download
def iteration():
for page in itertools.count(1):
url =
html = download(url)
if html is None:
# 尝试下载此网站时收到的错误
# 所以假设已达到最后一个国家ID，并可以停止下载
break
else:
# 成功 - 能够刮结果
# ...
pass
在这段代码中，我们对ID进行遍历，直到出现下载错误时停止，我们假设此时已到达最后一个国家的页面。不过，这种实现方式存在一个缺陷，那就是某些记录可能已被删除，数据库ID之间并不是连续的。此时，只要访问到某个间隔点，爬虫就会立即退出。下面是这段代码的改进版本，在该版本中连续发生多次下载错误后才退出程序。
def iteration2():
max_errors = 5 # 允许最大连续下载错误数
num_errors = 0 # 当前连续下载错误数
for page in itertools.count(1):
url =
html = download(url)
if html is None:
# 尝试下载此网页时出错
num_errors += 1
if num_errors == max_errors:
# 达到最大错误数时，退出
break
# 所以假设已达到最后一个ID，并可以停止下载
else:
# 成功 - 能够刮到结果
# ...
num_errors = 0
上面代码中实现的爬虫需要连续5次下载错误才会停止遍历，这样就很大程度上降低了遇到被删除记录时过早停止遍历的风险。
在爬取网站时，遍历ID是一个很便捷的方法，但是和网站地图爬虫一样，这种方法也无法保证始终可用。比如，一些网站会检查页面别名是否满足预期，如果不是，则会返回404 Not Found 错误。而另一些网站则会使用非连续大数作为ID，或是不使用数值作为ID，此时遍历就难以发挥作用了。例如，Amazon使用ISBN作为图书ID，这种编码包好至少10位数字。使用ID对Amazon的图书进行遍历需要测试数十亿次，因此这种方法肯定不是抓取该网站内容最高效的方法。
链接爬取
到目前为止，我们已经利用示例网站的结构特点实现了两个简单爬虫，用于下载所有的国家页面。只要这两种技术可用，就应当使用其进行爬取，因为这两种方法最小化了需要下载的网页数量。不过，对于另一些网站，我们需要让爬虫表现的更像普通用户，跟踪链接，访问感兴趣的内容。
通过跟踪所有链接的方式，我们可以很容易地下载种鸽网站的页面。但是，这种方法会下载大量我们并不需要的网页。例如，我们想要从一个在线论坛中抓取用户账号详情页，那么此时我们只需要下载账号页，而不需要下载讨论帖的页面。本节中的链接爬虫将使用正则表达式来确定需要下载那些页面。下面时这段代码的初始版本。
import re
from common import download
def link_crawler(seed_url, link_regex):
"""从指定的种子网址按照link_regex匹配的链接进行抓取"""
crawal_queue = [seed_url] # 要下载的URL队列
while crawal_queue:
url = crawal_queue.pop()
html = download(url).decode('utf-8')
# 使用过滤器来匹配我们的正则表达式
for link in get_links(html):
if re.match(link_regex, link):
# 将这个链接添加到爬网队列
crawal_queue.append(link)
def get_links(html):
"""从HTML返回一个链接列表"""
# 从网页提取所有链接的正则表达式
webpage_regex = re.IGNORECASE)
# 来自网页的所有链接的列表
return webpage_regex.findall(html)
要运行这段代码，只需要调用link_crawler函数，并传入两个参数，要爬取的网站URL和用于跟踪链接的正则表达式。对于示例网站，我们想要爬取的是国家列表索引页和国家页面。其中，索引页链接格式如下。
-
国家页链接格式如下。 -
因此，我们可以用/(index|view)/这个简单的正则表达式来匹配这两类网页。当爬虫使用这些输入参数运行时会发生什么呢？你会发现我们得到了如下的下载错误。
>>> '/(index|view)')
Downloading:
Downloading: /index/1
Traceback (most recent call last):
...
ValueError: unknown url type: '/index/1'
可以看出，问题处在下载/index/1时，该链接只有网页的路径部分，而没有协议和服务器部分，也就是说这是一个相对链接。由于浏览器知道你正在浏览哪个网页，所以在浏览器浏览时，相对链接是能够正常工作的。但是，urllib是无法获知上下文的。为了让urllib能够定位网页，我们需要将链接转换为决定链接的形式，以便包含定位网页的所有细节。如你所愿，Python中确实有用来实现这一功能的模块，该模块称为urlparse。下面是link_crawler的改进版本，使用了urlparse模块来创建绝对路径。
import urllib.parse
def link_crawler(seed_url, link_regex):
"""从指定的种子网址按照link_regex匹配的链接进行抓取"""
crawal_queue = [seed_url] # 要下载的URL队列
while crawal_queue:
url = crawal_queue.pop()
html = download(url).decode('utf-8')
for link in get_links(html):
# 检查链接是否匹配预期正则表达式
if re.match(link_regex, link):
# 形式绝对链接
link = urllib.parse.urljoin(seed_url, link)
crawal_queue.append(link)
当你运行这段代码时，会发现虽然网页下载没有出现错误，但是同样的地点总是会被不断下载到。这是因为这些地点相互之间存在链接。比如，澳大利亚链接到了南极洲，而南极洲也存在到澳大利亚的链接，此时爬虫就会在它们之间不断循环下去。要想避免重复爬取相同的链接，我们需要记录哪些链接已经被爬取过。下面是修改后的link_crawler函数，已具备存储已发现URL的功能，可以避免重复下载。
def link_crawler(seed_url, link_regex):
"""从指定的种子网址按照link_regex匹配的链接进行抓取"""
crawal_queue = [seed_url] # 要下载的URL队列
seen = set(crawal_queue) # 跟踪以前看过的URL
while crawal_queue:
url = crawal_queue.pop()
html = download(url).decode('utf-8')
for link in get_links(html):
# 检查链接是否匹配预期正则表达式
if re.match(link_regex, link):
# 形式绝对链接
link = urllib.parse.urljoin(seed_url, link)
# 检查是否已经看过该链接
if link not in seen:
seen.add(link)
crawal_queue.append(link)
当运行该脚本时，它会爬取所有地点，并且能够如期停止。最终，我们得到了一个可用的爬虫！
高级功能
现在，让我们为链接爬虫添加了一些功能，使其在爬取其它网站时更加有用。
解析robots.txt
首先，我们需要解析robots.txt文件，以避免下载禁止爬取的URL。使用Python自带的robotparser模块，就可以轻松完成这项工作，如下图的代码所示。
>>> import robotparser
>>> rp = robotparser.RobotFileParser()
>>>
>>> rp.read()
>>> url =
>>> user_agent = 'BadCrawler'
>>> rp.can_fetch(user_agent, url)
False
>>> user_agent = 'GoodCrawler'
>>> rp.can_fetch(user_agent, url)
True
rotbotparser模块首先加载robots.txt文件，然后通过can_fetch()函数确定指定的用户代理是否允许访问网页。在本例中，当用户代理设置为‘BadCrawler’时，robotparser模块会返回结果表明无法获取网页，这和示例网站robots.txt的定义一样。
为了将该功能集成到爬虫中，我们需要在crawl循环中添加该检查。
while crawal_queue:
url = crawal_queue.pop()
# 检查网址传递的robots.txt限制
if rp.can_fetch(user_agent, url):
...
else:
print("Blocked by robots.txt", url)
支持代理
proxy = ...
opener = urllib.request.build_opener()
proxy_params = {urllib.parse.urlparse(url).scheme: proxy}
opener.add_handler(urllib.request.ProxyHandler(proxy_params))
response = opener.open(request)
下面是集成了该功能的新版本 download 函数：
# -*- coding: utf-8 -*-
import urllib.request
import urllib.parse
def download5(url, user_agent='wswp',proxy=None, num_retries=2):
"""支持代理功能的下载函数"""
print("Downloading:", url)
headers = {'User-agent': user_agent}
request = urllib.request.Request(url, headers=headers)
opener = urllib.request.build_opener()
if proxy:
proxy_params = {urllib.parse.urlparse(url).scheme: proxy}
opener.add_handler(urllib.request.ProxyHandler(proxy_params))
try:
html = opener.open(request).read()
except urllib.request.URLError as e:
print("Download error:", e.reason)
html = None
if num_retries > 0:
if hasattr(e, 'code') and 500 <= e.code < 600:
# 重试 5xx HTTP 错误
html = download5(url, user_agent, proxy, num_retries-1)
return html
下载限速
如果我们爬取网站的速度过快，就会面临被封禁或是造成服务器过载的风险。为了降低这些风险，我们可以在两次下载之间添加延时，从而对爬虫限速。下面是实现了该功能的类的代码。
class Throttle:
def __init__(self, delay):
self.delay = delay
self.domains = {}
def wait(self, url):
domain =
last_accessed = self.domains.get(domain)
if self.delay > 0 and last_accessed is not None:
sleep_secs = self.delay - (datetime.now() - last_accessed).seconds
if sleep_secs > 0:
time.sleep(sleep_secs)
self.domains[domain] = datetime.now()
def get_links(html):
webpage_regex = re.IGNORECASE)
return webpage_regex.findall(html)
Throttle 类记录了每个域名上次访问的时间，如果当前时间距离上次访问时间小于指定延时，则执行睡眠操作。我们可以在每次下载之前调用Throttle对爬虫进行限速。
throttles = Throttle(delay)
...
throttle.wait(url)
result = download(url, headers, proxy = proxy, num_retries=num_retries)
避免爬虫陷阱
目前，我们的爬虫会跟踪所有之前没有访问过的链接。但是，一些网站会动态生成页面内容，这样就会出现无限多的网页。比如，网站有一个在线日历功能，提供了可以访问下个月和下一年的链接，那么下个月的页面中同样会包含访问下个月的链接，这样页面就会无止境地链接下去。这种情况被称为爬虫陷阱。
想要避免陷入爬虫陷阱，一个简单的方法是记录到达当前网页经过了多少个链接，也就是深度。当到达最大深度时，爬虫就不在向队列中添加该网页中的链接了。要实现这一功能，我们需要修改seen变量。该变量原先只记录访问过的网页链接，现在修改为一个字典，增加了页面深度的记录。
def link_crawler(.... max_depth=2):
max_depth = 2
seen = {}
...
depth = seen[url]
if depth != max_depth:
for link in links:
if link not in seen:
seen[link] = depth + 1
crawl_queue.append(link)
现在有了这一功能，我们就有信心爬虫最终一定能够完成。如果想要禁用该功能，只需将max_depth设为一个负数即可，此时当前深度永远不会与之相等。
最终版本
>>> seed_url =
>>> link_regex = '/(index|view)'
>>> link_crawler(seed_url, link_regex, user_agent='BadCrawlar')
Blocked by robots.txt:
现在，让我们使用默认的用户代理，并将最大深度设置为1，这样只有主页上的链接才会被下载。
>>> link_crawler(seed_url, link_regex, max_depth=1)
和预期一样，爬虫在下载完国家列表的第一页之后就停止了。
最终链接爬虫的代码如下：
import re
import urllib.parse
import urllib.request
import urllib.robotparser
import time
from datetime import datetime
import queue
def link_crawler(seed_url, link_regex=None, delay=5, max_depth=-1,
max_urls=-1, headers=None, user_agent='wswp', proxy=None, num_retries=1):
"""从指定的种子网址按照link_regex匹配的链接进行抓取"""
crawal_queue = queue.deque([seed_url]) # 仍然需要抓取的网址队列
seen = {seed_url: 0} # 已经看到的网址以及深度
num_urls = 0 # 跟踪已下载了多少个URL
rp = get_robots(seed_url)
throttle = Throttle(delay)
headers = headers or {}
if user_agent:
headers['User-agent'] = user_agent
while crawal_queue:
url = crawal_queue.pop()
# 检查网址传递的robots.txt限制
if rp.can_fetch(user_agent, url):
throttle.wait(url)
html = download(url, headers, proxy=proxy, num_retries=num_retries)
links = []
depth = seen[url]
if depth != max_depth:
# 仍然可以进一步爬行
if link_regex:
# 过滤符合我们的正则表达式的链接
links.extend(link for link in get_links(html) if re.match(link_regex, link))
for link in links:
link = normalize(seed_url, link)
# 检查是否已经抓取这个链接
if link not in seen:
seen[link] = depth + 1
# 检查链接在同一域内
if same_domain(seed_url, link):
# 成功！ 添加这个新链接到队列里
crawal_queue.append(link)
# 检查是否已达到下载的最大值
num_urls += 1
if num_urls == max_urls:
break
else:
print("Blocked by robots.txt:", url) # 链接已被robots.txt封锁
class Throttle:
"""Throttle通过睡眠在请求之间下载同一个域"""
def __init__(self, delay):
"""每个域的下载之间的延迟量"""
self.delay = delay
# 上次访问域时的时间戳
self.domains = {}
def wait(self, url):
domain =
last_accessed = self.domains.get(domain)
if self.delay > 0 and last_accessed is not None:
sleep_secs = self.delay - (datetime.now() - last_accessed).seconds
if sleep_secs > 0:
time.sleep(sleep_secs)
self.domains[domain] = datetime.now()
def download(url, headers, proxy, num_retries, data=None):
print("Downloading:", url)
request = urllib.request.Request(url, data, headers)
opener = urllib.request.build_opener()
if proxy:
proxy_params = {urllib.parse.urlparse(url).scheme: proxy}
opener.add_handler(urllib.request.ProxyHandler(proxy_params))
try:
response = opener.open(request)
html = response.read()
code = response.code
except urllib.request.URLError as e:
print("Download error:", e.reason)
html = ''
if hasattr(e, 'code'):
code = e.code
if num_retries > 0 and 500 <= code < 600:
# 重试 5xx HTTP 错误
return download(url, headers, proxy, num_retries-1, data)
else:
code = None
return html
def normalize(seed_url, link):
"""通过删除散列和添加域来规范化此URL"""
link, _ = urllib.parse.urldefrag(link) # 删除散列以避免重复
return urllib.parse.urljoin(seed_url, link)
def same_domain(url1, url2):
"""如果两个网址属于同一域，则返回True"""
return ==
def get_robots(url):
"""初始化此域的机器人解析器"""
rp = urllib.robotparser.RobotFileParser()
rp.set_url(urllib.parse.urljoin(url, '/robots.txt'))
rp.read()
return rp
def get_links(html):
"""从HTML返回一个链接列表"""
# 从网页提取所有链接的正则表达式
webpage_regex = re.IGNORECASE)
html = html.decode('utf-8')
# 来自网页的所有链接的列表
return webpage_regex.findall(html)