Logo

菜单

发布于 2026-01-06 1 阅读
0

精通 Python 网络爬虫：从零开始编写爬虫代码

精通 Python 网络爬虫：从零开始编写爬虫代码

你有没有尝试过抓取成千上万个网页？进一步扩大规模？处理和从系统故障中恢复？

在了解了如何从网站提取内容以及如何避免被屏蔽之后，我们将探讨抓取过程。要大规模获取数据，手动获取几个 URL 是行不通的。我们需要使用自动化系统来发现新页面并访问它们。

免责声明：实际使用时，请寻找合适的软件。以下提供更多相关信息。本指南旨在介绍爬虫过程的工作原理和基本操作，但其中还有大量细节需要讲解。

先决条件

要使代码运行，您需要安装 Python 3。有些系统已经预装了 Python 3。之后，运行以下命令安装所有必要的库pip install：

pip install requests beautifulsoup4 pandas
Enter fullscreen mode Exit fullscreen mode

如何获取页面上的所有链接

requests.get从本系列的第一篇文章中，我们知道使用 `get_data()`和 ` get_link()` 函数很容易从网页上获取数据。我们将首先在一个为测试网页抓取而准备的虚拟商店BeautifulSoup中查找链接。

获取内容的基本步骤相同。然后，​​我们获取分页器上的所有链接，并将这些链接添加到一个元素中set。我们选择使用 `<link>` 标签来避免重复。如您所见，我们对链接的选择器进行了硬编码，这意味着它并非通用解决方案。目前，我们将专注于当前页面。

import requests 
from bs4 import BeautifulSoup 

to_visit = set() 
response = requests.get('https://scrapeme.live/shop/page/1/') 
soup = BeautifulSoup(response.content, 'html.parser') 
for a in soup.select('a.page-numbers'): 
    to_visit.add(a.get('href')) 

print(to_visit) 
# {'https://scrapeme.live/shop/page/2/', '.../3/', '.../46/', '.../48/', '.../4/', '.../47/'}
Enter fullscreen mode Exit fullscreen mode

一次一个URL，按顺序

现在我们有了几个链接，但无法全部访问。我们需要某种循环来对每个可用的 URL 执行提取部分，从而解决这个问题。也许最直接的方法（尽管扩展性较差）是使用同一个循环。但在此之前，还有一个关键步骤需要完成：避免重复抓取同一个页面。

我们会在另一个文件中跟踪已访问过的链接set，并在每次请求前进行检查以避免重复。目前，该文件to_visit并未实际使用，仅用于演示目的。为了防止访问每个页面，我们还会添加一个max_visits变量。现在，我们暂时忽略该robots.txt文件，但我们必须保持礼貌和友好。

visited = set() 
to_visit = set() 
max_visits = 3 

def crawl(url): 
    print('Crawl: ', url) 
    response = requests.get(url) 
    soup = BeautifulSoup(response.content, 'html.parser') 
    visited.add(url) 
    for a in soup.select('a.page-numbers'): 
        link = a.get('href') 
        to_visit.add(link) 
        if link not in visited and len(visited) < max_visits: 
            crawl(link) 

crawl('https://scrapeme.live/shop/page/1/') 

print(visited) # {'.../3/', '.../1/', '.../2/'} 
print(to_visit) # { ... new ones added, such as pages 5 and 6 ... }
Enter fullscreen mode Exit fullscreen mode

这是一个递归函数，有两个退出条件：没有更多链接可访问，或者已达到最大访问次数。无论哪种情况，它都会退出并打印已访问的链接和待访问的链接。

需要注意的是，同一个链接可以多次添加，但只会被抓取一次。在大型项目中，可以设置一个定时器，每隔几天才请求一次同一个URL。

关注点分离

我们说过，这不是关于提取或解析内容的问题，而是在职责混淆之前需要将其分离。为此，我们将创建三个辅助函数：获取 HTML、提取链接和提取内容。顾名思义，它们分别执行网络爬虫的三个主要任务。

第一种方法会使用与之前相同的库从 URL 获取 HTML，但try为了安全起见，会将其包装在一个块中。

def get_html(url): 
    try: 
        return requests.get(url).content 
    except Exception as e: 
        print(e) 
        return ''
Enter fullscreen mode Exit fullscreen mode

第二个功能，即提取链接，将与之前一样正常工作。

def extract_links(soup): 
    return [a.get('href') for a in soup.select('a.page-numbers') 
        if a.get('href') not in visited]
Enter fullscreen mode Exit fullscreen mode

最后一个元素将作为占位符，用于提取我们需要的内容。由于我们正在简化这部分，它将从同一页面获取基本信息，无需进入详情页面。

为了证明我们可以提取一些内容，我们将打印每个产品的标题（宝可梦名称）。

def extract_content(soup): 
    for product in soup.select('.product'): 
        print(product.find('h2').text) 
 # Bulbasaur, Ivysaur, ...
Enter fullscreen mode Exit fullscreen mode

将所有部件组装在一起。

def crawl(url): 
    if not url or url in visited: 
        return 
    print('Crawl: ', url) 
    visited.add(url) 
    html = get_html(url) 
    soup = BeautifulSoup(html, 'html.parser') 
    extract_content(soup) 
    links = extract_links(soup) 
    to_visit.update(links)
Enter fullscreen mode Exit fullscreen mode

注意到有什么不同吗？爬虫逻辑与链接提取部分并不关联。每个辅助函数负责处理一个独立的部分。而该crawl函数则充当协调器的角色，调用这些辅助函数并应用结果。

随着项目的推进，所有这些部分都可以移到文件中，或者作为参数/回调函数传递。如果核心功能独立于所选页面和内容，我们就可以推广这些用例。

我们是不是漏掉了什么？🤔
我们需要添加第一个 URL 并调用爬虫函数。由于crawl它不再是递归的，我们将在一个单独的循环中处理它。

to_visit.add('https://scrapeme.live/shop/page/1/') 

while (len(to_visit) > 0 and len(visited) < max_visits): 
    crawl(to_visit.pop())
Enter fullscreen mode Exit fullscreen mode

并行请求

这里缺少一个重要环节：并行性。HTTP 请求处理程序大部分时间都处于空闲状态，等待响应返回。这意味着我们可以同时发送多个请求而不会使机器过载，然后在收到响应后立即进行处理。

值得注意的是，这种方法仅在顺序并非强制性的情况下才有效。但我们已经在使用集合，根据Python 的定义，“集合是一个无序且不包含重复元素的集合”。这意味着我们的流程从一开始就是无序的。

在深入探讨并行请求之前，我们必须了解几个概念：同步和队列。

同步队列

在多线程或并行计算中存在一个巨大的风险：不同线程可以修改相同的变量或数据结构。这意味着我们的两个请求都会向同一个集合中添加新的链接（例如，添加新链接to_visit）。由于数据结构没有受到保护，因此两个请求都可以像这样对其进行读写操作：

  • 两者都阅读了其内容，即(1, 2, 3)（简化版）
  • 帖子一添加了页面链接4, 5：(1, 2, 3, 4, 5)
  • 线程二添加了页面链接6, 7：(1, 2, 3, 6, 7)

这是怎么回事？当第二个线程写入新链接时，它将它们添加到了一个只有三个元素的集合中。
这只是一个非常简化的版本；更多信息请查看链接。

如何避免这些冲突？同步或加锁。文档中写道： “队列使用锁来暂时阻塞竞争线程。” 这意味着线程一会获取集合的锁，顺利地进行读写操作，然后自动释放锁。与此同时，线程二必须等待锁释放后才能进行读写操作。

import queue 

q = queue.Queue() 
q.put('https://scrapeme.live/shop/page/1/') 

def crawl(url): 
    ... 
    links = extract_links(soup) 
    for link in links: 
        if link not in visited: 
            q.put(link)
Enter fullscreen mode Exit fullscreen mode

目前还无法正常工作，请不必担心。对现有代码的改动非常小：我们to_visit用队列替换了原来的代码。但是队列需要处理器或工作进程来处理其内容。通过以上步骤，我们创建了一个队列并添加了一个项目（即原来的项目）。我们还修改了函数，crawl使其将链接放入队列，而不是更新之前的集合。

我们将使用线程模块创建一个工作进程来处理该队列。

from threading import Thread 

def queue_worker(i, q): 
    while True: 
        url = q.get() # Get an item from the queue, blocks until one is available 
        print('to process:', url) 
        q.task_done() # Notifies the queue that the item has been processed 

q = queue.Queue() 
Thread(target=queue_worker, args=(0, q), daemon=True).start() 

q.put('https://scrapeme.live/shop/page/1/') 
q.join() # Blocks until all items in the queue are processed and marked as done 
print('Done') 

# to process: https://scrapeme.live/shop/page/1/ 
# Done
Enter fullscreen mode Exit fullscreen mode

我们定义了一个新函数来处理队列中的项目。为此，我们进入了一个无限循环，该循环会在所有处理完成后停止。

然后会有get一个待处理项，它会阻塞直到有新项可用。我们会处理该待处理项；目前，只是将其打印出来以演示其工作原理。crawl稍后会调用它。

最后，我们通过调用来通知队列该项目已被处理task_done。

当队列中的所有项目都被获取并清空后，它将停止执行并结束无限循环。这就是该join函数的作用，“阻塞直到队列中的所有项目都被获取并处理完毕”。

现在我们还需要两件事：处理项目和创建更多线程（如果只有一个线程，就无法并行处理，对吧？）。

def queue_worker(i, q): 
    while True: 
        url = q.get() 
        if (len(visited) < max_visits and url not in visited): 
            crawl(url) 
        q.task_done() 

q = queue.Queue() 
num_workers = 4 
for i in range(num_workers): 
    Thread(target=queue_worker, args=(i, q), daemon=True).start()
Enter fullscreen mode Exit fullscreen mode

运行此脚本时务必小心，因为输入较大的数字num_workers会max_visits触发大量请求。如果脚本存在任何细微的错误，都可能在几秒钟内发出数百个请求。

表现

我们运行不同设置的基准测试仅作为参考。

  • 顺序请求：29.32秒
  • 只有一个工作进程的队列（num_workers = 1）：29.41秒
  • 包含两个工作进程的队列（num_workers = 2）：20.05秒
  • 包含五个工作进程的队列（num_workers = 5）：11.97秒
  • 包含十个工作进程的队列（num_workers = 10）：12.02秒

顺序请求和使用单个工作线程几乎没有区别。线程会带来一些开销，但在这里几乎察觉不到。需要更严苛的负载测试才能体现出来。一旦我们开始增加工作线程，这些开销就会被抵消。我们可以增加更多线程，但这不会影响结果，因为它们大部分时间都处于空闲状态。

分布式处理

我们不会讨论接下来的扩展步骤：将爬虫进程分布到多个服务器上。Python允许这样做，一些库（例如 Celery或Redis Queue ）可以提供帮助。这是一个巨大的步骤，我们今天的内容已经足够多了。

简单来说，其背后的理念与线程处理相同。每个数据项都会像之前那样进行处理，但这次是在不同的线程甚至不同的机器上运行相同的代码。理论上，这种方法可以实现更大的扩展性，没有上限。但实际上，总会存在瓶颈，通常是负责分发的中心节点。

扩大规模时需要考虑的因素

我们为了教学目的展示了一个简化的爬虫过程。要大规模应用这些方法，首先需要考虑以下几个方面。

自建 vs 购买 vs 开源

在编写自己的爬虫库之前，不妨先试试现有的方案。许多优秀的开源库都能实现这一目标，例如Scrapy、pyspider、node-crawler（Node.js）或Colly（Go）。此外，还有许多公司和服务提供爬虫解决方案。

避免被屏蔽

正如我们在之前的文章中看到的，我们可以采取几种措施来避免被屏蔽。其中一些措施包括使用代理和设置请求头。以下是一个简单的代码片段，展示了如何将这些添加到我们当前的代码中。
请注意，这些免费代理可能不适用于所有情况，因为它们的有效期很短。

proxies = { 
    'http': 'http://190.64.18.177:80', 
    'https': 'http://49.12.2.178:3128', 
} 

headers = { 
    'authority': 'httpbin.org', 
    'cache-control': 'max-age=0', 
    'sec-ch-ua': '"Chromium";v="92", " Not A;Brand";v="99", "Google Chrome";v="92"', 
    'sec-ch-ua-mobile': '?0', 
    'upgrade-insecure-requests': '1', 
    'user-agent': 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/92.0.4515.107 Safari/537.36', 
    'accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9', 
    'sec-fetch-site': 'none', 
    'sec-fetch-mode': 'navigate', 
    'sec-fetch-user': '?1', 
    'sec-fetch-dest': 'document', 
    'accept-language': 'en-US,en;q=0.9', 
} 

def get_html(url): 
    try: 
        response = requests.get(url, headers=headers, proxies=proxies) 
        return response.content 
    except Exception as e: 
        print(e) 
        return ''
Enter fullscreen mode Exit fullscreen mode

提取内容

这里我们不赘述细节，只提供一个简单的代码片段，用于提取每个商品的 ID、名称和价格。我们将所有信息存储在一个data数组中，这不是一个好做法，但对于演示来说已经足够了。

data = [] 

def extract_content(soup): 
    for product in soup.select('.product'): 
        data.append({ 
            'id': product.find('a', attrs={'data-product_id': True})['data-product_id'], 
            'name': product.find('h2').text, 
            'price': product.find(class_='amount').text 
        }) 

print(data) 
# [{'id': '759', 'name': 'Bulbasaur', 'price': '£63.00'}, {'id': '729', 'name': 'Ivysaur', 'price': '£87.00'}, ...]
Enter fullscreen mode Exit fullscreen mode

持久性

我们还没有持久化任何数据，这无法扩展。在实际应用中，我们应该存储内容，甚至包括 HTML 代码本身，以便后续处理。此外，还要存储所有已发现的 URL 及其时间戳。这一切似乎都表明我们需要一个数据库。根据具体需求，我们可以只存储实际内容，也可以存储完整的 URL、日期、HTML 代码等等。

礼服

链接提取部分并未考虑规范链接。一个页面可能拥有多个 URL：查询字符串或哈希值可能会对其进行修改。在我们的案例中，我们会抓取该页面两次。目前这并非问题，但值得考虑。

正确的做法是将规范 URL（如果存在）添加到已访问列表中。这样，即使我们从不同的源 URL 访问同一个页面，系统也会将其检测为重复访问。我们还可以使用url_query_cleaner 函数移除一些查询字符串参数。

Robots.txt

我们没有进行检查，因为我们使用的是专门用于抓取的测试网站。但请务必检查 robots.txt 文件，并在抓取实际目标网站时遵守其中的规定。此外，请勿造成超出网站处理能力的流量。再次强调，请文明礼貌地进行抓取 ;)

最终代码

import requests 
from bs4 import BeautifulSoup 
import queue 
from threading import Thread 

starting_url = 'https://scrapeme.live/shop/page/1/' 
visited = set() 
max_visits = 100 # careful, it will crawl all the pages 
num_workers = 5 
data = [] 

def get_html(url): 
    try: 
        response = requests.get(url) 
        # response = requests.get(url, headers=headers, proxies=proxies) 
        return response.content 
    except Exception as e: 
        print(e) 
        return '' 

def extract_links(soup): 
    return [a.get('href') for a in soup.select('a.page-numbers') 
            if a.get('href') not in visited] 

def extract_content(soup): 
    for product in soup.select('.product'): 
        data.append({ 
            'id': product.find('a', attrs={'data-product_id': True})['data-product_id'], 
            'name': product.find('h2').text, 
            'price': product.find(class_='amount').text 
        }) 

def crawl(url): 
    visited.add(url) 
    print('Crawl: ', url) 
    html = get_html(url) 
    soup = BeautifulSoup(html, 'html.parser') 
    extract_content(soup) 
    links = extract_links(soup) 
    for link in links: 
        if link not in visited: 
            q.put(link) 

def queue_worker(i, q): 
    while True: 
        url = q.get() # Get an item from the queue, blocks until one is available 
        if (len(visited) < max_visits and url not in visited): 
            crawl(url) 
        q.task_done() # Notifies the queue that the item has been processed 

q = queue.Queue() 
for i in range(num_workers): 
    Thread(target=queue_worker, args=(i, q), daemon=True).start() 

q.put(starting_url) 
q.join() # Blocks until all items in the queue are processed and marked as done 

print('Done') 
print('Visited:', visited) 
print('Data:', data)
Enter fullscreen mode Exit fullscreen mode

结论

我们希望您能分享以下三点：

  1. 将获取 HTML 和提取链接的过程与爬虫本身分开。
  2. 根据您的使用场景选择合适的系统：简单顺序系统、并行系统或分布式系统。
  3. 从零开始大规模构建可能会很困难。不妨考虑使用免费或付费的库/解决方案。

我们即将完成关于网络爬虫的系列文章。敬请期待下一篇，我们将探讨如何进一步扩展爬虫流程。

别忘了看看本系列的其他文章。

您觉得这些内容有用吗？请分享给更多人！👈

原文发表于https://www.zenrows.com

文章来源：https://dev.to/anderrv/mastering-web-scraping-in-python-crawling-from-scratch-1dgd
NodeJS：如何发送电子邮件 D...
今天我学会了如何创建常见问题解答折叠功能