1. 从“手撸”到“OpenClaw”一个程序员的浪漫最近在技术社区里看到不少朋友在讨论“手撸”这个词。它早已超越了字面意思成了一种极客精神的代名词不依赖现成的重型框架从最核心的原理出发自己动手实现一个功能或工具。这过程就像亲手组装一台模型或者更接地气一点像在夜市大排档里亲手剥开一只小龙虾——过程充满挑战但最终收获的成就感和对“食材”的理解是直接享用成品无法比拟的。今天我们就来“手撸”一个名为OpenClaw的小工具。这个名字听起来有点酷又带点趣味。“Claw”是爪子在编程世界里它常常指代那些能“抓取”东西的工具比如网络爬虫。所以OpenClaw 本质上是一个轻量级、可扩展的网络爬虫框架。我们不用 Scrapy 这样的“工业级起重机”也不用 Puppeteer 这样的“自动化机械臂”而是用最基础的“螺丝刀”和“扳手”即 Python 标准库和少量第三方库从零开始搭建一个能抓取网页、解析数据、并保存结果的核心骨架。为什么要在 Scrapy 等成熟框架大行其道的今天还要做这件事原因有三。第一理解本质框架封装了复杂性也隐藏了细节。自己实现一遍你会彻底明白 HTTP 请求、HTML 解析、异步并发、去重策略这些爬虫核心组件是如何协同工作的。第二极致定制当你的需求非常特殊或者对性能、资源有极端要求时一个量身定做、没有冗余的“手撸”工具往往比改造一个庞大框架更高效。第三学习与乐趣这是最好的编程练习涉及网络、数据结构、并发编程、设计模式等多个领域完成后的成就感十足。接下来我们将用大约10分钟的核心讲解时间实际动手编码和思考的时间当然会更长一步步构建 OpenClaw 的雏形。目标是让它具备最核心的抓取能力结构清晰方便后续按需添加代理、分布式、反反爬虫等高级特性。无论你是想深入理解爬虫原理的新手还是需要为一个轻量级特定任务寻找解决方案的开发者这篇内容都能给你带来直接的参考价值。2. OpenClaw 的核心架构设计四两拨千斤在动手写代码之前我们必须先想清楚 OpenClaw 应该长什么样。一个好的设计能让后续的编码事半功倍也决定了这个工具的扩展上限。我们不追求大而全而是要精准命中爬虫工作流中最关键的几个环节。一个最基本的爬虫工作流可以抽象为以下几个步骤调度决定下一个要抓取的网址URL是什么。下载向目标网址发起 HTTP 请求获取网页的原始内容HTML、JSON 等。解析从下载的原始内容中提取出我们感兴趣的结构化数据如文章标题、价格、链接。处理与存储对提取的数据进行清洗、验证然后保存到文件或数据库。链接提取从已解析的页面中发现新的、需要抓取的网址并送回给调度器。基于这个流程我们可以为 OpenClaw 设计一个经典的生产者-消费者模型核心包含四个模块2.1 调度器任务队列的管理者调度器是爬虫的大脑负责管理待抓取的 URL 队列。我们需要考虑几个关键问题队列数据结构Python 的collections.deque双端队列是一个很好的起点它支持高效的头部弹出和尾部追加。对于更复杂的优先级调度可以使用heapq模块实现优先队列。去重绝对不能重复抓取同一个 URL。我们将在调度器内部维护一个“已访问集合”。一个简单有效的去重方法是使用set()存储 URL但对于海量 URL内存可能吃紧。因此我们的设计要留有接口未来可以轻松替换为基于布隆过滤器或数据库的方案。线程安全如果我们的下载器是多线程的那么对任务队列的存取操作必须是原子的。我们可以直接使用queue.Queue它天生就是线程安全的。在 OpenClaw 的初版中我们将采用queue.Queue作为任务队列并用一个set做内存去重实现简单直观。2.2 下载器与网络对话的工人下载器的唯一职责就是根据 URL 下载内容。这里面的门道不少请求库选择requests库是同步请求的绝佳选择简单易用。但对于高性能爬虫异步是必由之路。aiohttp是异步 HTTP 客户端的标杆。为了平衡难度和实用性我们的 OpenClaw 初版将使用requests但会以面向接口的方式编写下载器以便后续无缝切换为aiohttp。请求头模拟这是绕过基础反爬虫的第一关。一个没有User-Agent的请求就像没穿衣服上街异常显眼。我们需要让下载器能够接收并应用自定义的请求头。异常处理与重试网络世界充满不确定性连接超时、服务器错误、页面不存在。一个健壮的下载器必须包含异常捕获和重试机制。我们可以设计一个装饰器或直接在方法内部实现带指数退避的重试逻辑。延迟控制尊重目标网站避免请求过快被封 IP。下载器应该支持在每个请求之间插入随机延迟。我们将实现一个Downloader类它接收一个 URL 和请求头返回响应文本和状态码并内置简单的重试和延迟功能。2.3 解析器从混沌中提取秩序下载到的是混杂着标签、样式和脚本的 HTML“矿石”解析器的任务就是从中提炼出数据的“金属”。解析库选择BeautifulSoup4配合lxml解析引擎是 Python 界的事实标准写法接近自然语言非常友好。对于极致的性能可以直接使用lxml的 XPath 或parsel库Scrapy 在用。我们选择BeautifulSoup4作为起点因为它的学习曲线平缓足以应对 90% 的场景。解析规则抽象我们不能把解析逻辑硬编码在爬虫主体里。好的设计是将解析规则独立出来可能是一个函数、一个配置字典甚至一个类。这样同一个爬虫框架就能通过更换解析规则来抓取不同的网站。数据封装解析出的数据应该被封装成结构化的对象比如 Python 字典或dataclass。这有利于后续的数据处理和存储。我们将定义一个Parser基类要求子类实现一个parse方法。这个方法接收 HTML 文本和当前 URL返回两项内容一是提取到的结构化数据项列表二是从当前页面中发现的新 URL 列表。2.4 数据管道数据的终点站数据管道负责处理解析器产出的数据。它可能做很多事情数据清洗去除空白字符、转换格式、处理缺失值。数据验证检查必填字段是否存在数据格式是否正确。数据存储将数据保存到 CSV 文件、JSON 文件、MySQL、MongoDB 等。 和解析器一样数据管道也应该被设计成可插拔的组件。我们可以定义一个Pipeline基类子类实现process_item方法。有了以上设计OpenClaw 的主循环逻辑就清晰了调度器出队一个 URL - 下载器下载 - 解析器解析得到数据和新的URL- 新URL送回调度器入队 - 数据送入管道处理。这个循环直到任务队列为空或达到其他停止条件为止。3. 手把手实现 OpenClaw 核心模块理论说得够多了现在打开你的代码编辑器我们开始真正“手撸”。我们将按照架构逐个模块实现。请确保你已安装必要的库pip install requests beautifulsoup4。3.1 构建调度器简单而稳固的队列核心首先我们实现调度器Scheduler。它需要完成添加任务、获取任务、判断任务是否已存在等基本功能。import queue from typing import Set from urllib.parse import urljoin class Scheduler: def __init__(self): # 使用线程安全的队列存储待抓取URL self.task_queue queue.Queue() # 使用集合存储已看到过的URL用于去重 self.seen_urls: Set[str] set() def add_task(self, url: str, base_url: str None): 添加一个新任务到队列。 # 可选将相对URL转换为绝对URL if base_url and not url.startswith((http://, https://)): url urljoin(base_url, url) # 去重检查如果URL没被见过则加入队列和集合 if url not in self.seen_urls: self.seen_urls.add(url) self.task_queue.put(url) print(f[调度器] 新任务入队: {url}) def get_task(self): 从队列中获取一个任务。如果队列为空返回None。 try: # blockFalse 表示非阻塞获取队列空时立刻抛出Empty异常 return self.task_queue.get(blockFalse) except queue.Empty: return None def has_pending_tasks(self): 判断是否还有待处理的任务。 return not self.task_queue.empty()关键点解析urljoin的作用网页中的链接常常是相对路径如/article/123。urljoin能根据当前页面的base_url将其补全为绝对 URL如https://example.com/article/123这是爬虫能持续工作的基础。去重时机我们在add_task时就去重而不是在get_task时。这保证了队列中本身就没有重复项更清晰。seen_urls集合保存在内存中适用于中小规模抓取。队列选择queue.Queue的get(blockFalse)让我们可以在主循环中非阻塞地检查是否有新任务方便控制爬虫的启停。3.2 打造下载器稳健的网络请求客户端接下来是下载器Downloader。它要处理网络请求的所有细节。import time import random from typing import Optional import requests from requests.exceptions import RequestException class Downloader: def __init__(self, delay: float 1.0, max_retries: int 3): 初始化下载器。 :param delay: 请求之间的基础延迟秒用于礼貌爬取。 :param max_retries: 请求失败时的最大重试次数。 self.delay delay self.max_retries max_retries # 创建一个共享的Session可以复用TCP连接提升效率 self.session requests.Session() # 设置一个默认的User-Agent模拟常见浏览器 self.default_headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36 } def download(self, url: str, headers: Optional[dict] None) - Optional[str]: 下载给定URL的内容。 :return: 成功则返回网页文本失败则返回None。 # 合并默认头和自定义头 final_headers {**self.default_headers, **(headers or {})} # 重试逻辑 for attempt in range(self.max_retries): try: # 请求前等待避免过快 time.sleep(self.delay random.uniform(0, 0.5)) # 增加一点随机性 print(f[下载器] 尝试抓取 ({attempt1}/{self.max_retries}): {url}) response self.session.get(url, headersfinal_headers, timeout10) response.raise_for_status() # 如果状态码不是200抛出HTTPError异常 # 检查编码避免乱码 response.encoding response.apparent_encoding return response.text except RequestException as e: print(f[下载器] 请求失败: {e}) if attempt self.max_retries - 1: # 最后一次尝试也失败了 return None # 指数退避失败后等待更长时间再重试 wait_time 2 ** attempt print(f[下载器] 等待 {wait_time} 秒后重试...) time.sleep(wait_time) return None关键点解析与避坑指南使用 Sessionrequests.Session()可以保持 cookies并在多个请求间复用底层 TCP 连接显著减少网络开销这是提升性能的一个小技巧。response.raise_for_status()这个方法非常有用。requests.get()即使遇到 404、500 等错误状态码也不会抛出异常除非你设置raise_for_statusTrue。手动调用此方法可以确保只处理成功的响应将错误处理统一到except块中。编码处理网页编码千奇百怪。response.apparent_encoding是requests库基于内容分析出的可能编码比response.encoding来自HTTP头通常更可靠能有效避免中文乱码问题。延迟与随机性固定的延迟如time.sleep(1)模式容易被识别。加入随机扰动random.uniform(0, 0.5)能使爬虫行为更接近人类是基础的“反反爬虫”策略。指数退避重试网络请求失败是常态。指数退避策略等待 1秒、2秒、4秒...既能给服务器喘息之机也能提高在临时网络波动下最终成功的概率。3.3 实现解析器定义数据提取的规则解析器我们设计成抽象基类具体规则由用户实现。这里我们以一个抓取虚构新闻网站为例。from abc import ABC, abstractmethod from typing import List, Tuple from bs4 import BeautifulSoup class Parser(ABC): 解析器抽象基类。所有具体的解析规则都应继承此类。 abstractmethod def parse(self, html: str, current_url: str) - Tuple[List[dict], List[str]]: 解析HTML提取数据和新的链接。 :param html: 下载的HTML文本。 :param current_url: 当前页面的URL用于将相对链接转为绝对链接。 :return: (提取到的数据项列表, 新发现的URL列表) pass class NewsParser(Parser): 一个示例解析器用于解析一个简单的新闻列表页和详情页。 def parse(self, html: str, current_url: str) - Tuple[List[dict], List[str]]: data_items [] new_urls [] soup BeautifulSoup(html, lxml) # 情况1当前页面是列表页包含多个新闻链接 # 假设列表页中新闻链接在 a classnews-link 标签里 for link_tag in soup.find_all(a, class_news-link): href link_tag.get(href) if href: new_urls.append(href) # 这里先存相对链接调度器会处理 # 也可以在列表页预提取一些信息比如标题 title link_tag.get_text(stripTrue) if title: data_items.append({title: title, url: href, source_page: list}) # 情况2当前页面是详情页包含新闻正文 # 假设详情页中标题在 h1 idarticle-title正文在 div classarticle-content title_tag soup.find(h1, idarticle-title) content_div soup.find(div, class_article-content) if title_tag and content_div: # 如果找到这两个元素则认为这是详情页 title title_tag.get_text(stripTrue) content content_div.get_text(stripTrue, separator\n) # 用换行符保留段落感 data_items.append({ title: title, content: content, url: current_url, source_page: detail }) # 详情页可能也有相关文章链接可以继续提取 for related_link in soup.select(.related-news a): href related_link.get(href) if href: new_urls.append(href) # 情况3提取页面中所有其他符合条件的链接广度爬取 # 例如抓取分页链接。假设分页链接在 a classpage-link 里 for page_link in soup.find_all(a, class_page-link): href page_link.get(href) if href and href not in new_urls: # 简单去重 new_urls.append(href) return data_items, new_urls关键点解析抽象基类使用ABC和abstractmethod强制要求子类实现parse方法。这定义了框架和插件之间的契约使得 OpenClaw 的核心循环可以调用parser.parse(...)而无需关心具体实现。灵活的解析逻辑NewsParser展示了如何在一个解析器里处理多种页面类型列表页、详情页。通过检查 HTML 中是否存在特定元素来判断页面类型这是一种很实用的模式。BeautifulSoup方法选择find_all(): 查找所有符合条件的标签。find(): 查找第一个符合条件的标签。select(): 使用 CSS 选择器查找功能强大且写法简洁。get_text(stripTrue, separator\n): 获取标签内所有文本stripTrue去除首尾空白separator可以指定标签间文本的分隔符对于正文提取非常有用。链接去重在解析器内部对提取的新 URL 做一层简单的去重if href not in new_urls可以减少调度器的无效工作。但调度器的全局去重仍然是最终保障。3.4 组装数据管道数据的清洁工与仓库管理员数据管道负责处理解析后的数据。我们先实现一个最简单的控制台打印管道和一个 CSV 文件存储管道。from abc import ABC, abstractmethod import csv import json from typing import List class Pipeline(ABC): 数据管道抽象基类。 abstractmethod def process_item(self, item: dict): 处理一个数据项。 pass def close(self): 管道关闭时执行的操作如关闭文件句柄、数据库连接。 pass class ConsolePipeline(Pipeline): 将数据打印到控制台的管道。用于调试。 def process_item(self, item: dict): print([控制台管道] 抓到数据:, json.dumps(item, ensure_asciiFalse, indent2)) class CsvPipeline(Pipeline): 将数据存储到CSV文件的管道。 def __init__(self, filename: str output.csv): self.filename filename self.file None self.writer None self.fieldnames None # CSV文件的表头 def open_spider(self): 在爬虫开始时调用这里由主循环手动触发。 self.file open(self.filename, w, newline, encodingutf-8-sig) # utf-8-sig 支持Excel直接打开 # 先不创建writer等收到第一个item时根据keys确定表头 def process_item(self, item: dict): if self.writer is None: # 第一个item到达根据其keys创建CSV表头和写入器 self.fieldnames item.keys() self.writer csv.DictWriter(self.file, fieldnamesself.fieldnames) self.writer.writeheader() # 写入一行数据 self.writer.writerow(item) print(f[CSV管道] 数据已写入: {item}) def close(self): if self.file: self.file.close() print(f[CSV管道] 文件 {self.filename} 已关闭。)关键点解析延迟创建表头CsvPipeline采用了一种巧妙的设计——在process_item收到第一个数据项时才根据这个数据项的键来确定 CSV 文件的表头。这使得管道无需预先知道数据的具体结构非常灵活。无论解析器提取出什么字段都能自动适应。utf-8-sig编码使用utf-8-sig编码写入 CSV 文件会在文件开头加入一个 BOM字节顺序标记。这对于 Windows 系统上的 Excel 等软件非常友好能确保中文字符正确显示而不会出现乱码。管道组合我们可以很容易地创建多个管道实例比如一个打印到控制台一个保存到 CSV一个发送到数据库并在主循环中依次调用它们的process_item方法。这就是“管道”模式的威力数据处理流程清晰且可扩展。4. 编写主引擎与实战测试让 OpenClaw 动起来现在我们将上面散落的模块组装起来形成一个完整的爬虫引擎OpenClawEngine并写一个简单的main函数来测试它。4.1 构建核心引擎class OpenClawEngine: OpenClaw 的核心引擎负责协调各模块工作。 def __init__(self, downloader, parser, pipelinesNone): self.scheduler Scheduler() self.downloader downloader self.parser parser # pipelines 是一个管道实例的列表 self.pipelines pipelines or [] def add_start_url(self, url: str): 添加初始URL。 self.scheduler.add_task(url) def run(self): 启动爬虫主循环。 print([引擎] OpenClaw 启动) # 启动管道如果有需要初始化的 for pipeline in self.pipelines: if hasattr(pipeline, open_spider): pipeline.open_spider() try: while self.scheduler.has_pending_tasks(): # 1. 调度获取下一个任务 current_url self.scheduler.get_task() if not current_url: break print(f\n[引擎] 处理: {current_url}) # 2. 下载 html self.downloader.download(current_url) if html is None: print(f[引擎] 下载失败跳过: {current_url}) continue # 3. 解析 data_items, new_urls self.parser.parse(html, current_url) print(f[引擎] 解析出 {len(data_items)} 条数据 {len(new_urls)} 个新链接) # 4. 新链接送回调度器 for url in new_urls: self.scheduler.add_task(url, base_urlcurrent_url) # 5. 数据送入管道处理 for item in data_items: for pipeline in self.pipelines: pipeline.process_item(item) except KeyboardInterrupt: print(\n[引擎] 用户中断停止爬取。) finally: # 关闭管道释放资源 for pipeline in self.pipelines: pipeline.close() print([引擎] 爬取结束。)引擎逻辑详解清晰的流程run方法完美体现了我们之前设计的核心循环调度 - 下载 - 解析 - 链接反馈- 数据处理。代码就是设计文档。资源管理在try...finally块中我们确保了即使爬虫被用户强制中断KeyboardInterrupt管道也能被正确关闭文件句柄等资源得到释放。这是编写健壮程序的好习惯。可扩展性引擎通过构造函数接收downloader,parser,pipelines。这意味着我们可以轻松地替换其中任何一个组件。例如明天你想用aiohttp只需写一个AsyncDownloader类替换掉现在的Downloader实例即可引擎代码无需改动。4.2 编写测试用例与模拟服务器为了在不触及真实网络的情况下测试我们的 OpenClaw我们需要一个本地模拟服务器。Python 内置的http.server模块可以快速实现。首先创建一个名为test_server.py的文件并在同目录下创建一些模拟的 HTML 文件。模拟的列表页 (index.html):!DOCTYPE html html headtitle测试新闻站/title/head body h1新闻列表/h1 ul lia classnews-link href/article/1第一条测试新闻/a/li lia classnews-link href/article/2第二条测试新闻/a/li /ul div classpagination a classpage-link href/page/2下一页/a /div /body /html模拟的详情页 (article1.html):!DOCTYPE html html headtitle第一条测试新闻/title/head body h1 idarticle-title第一条测试新闻/h1 div classarticle-content p这是第一条测试新闻的详细内容。/p p这里包含了我们想抓取的所有正文信息。/p /div div classrelated-news h3相关新闻/h3 a href/article/3相关新闻一/a /div /body /html创建test_server.py:import http.server import socketserver import os PORT 8888 DIRECTORY . # 当前目录 class Handler(http.server.SimpleHTTPRequestHandler): def __init__(self, *args, **kwargs): super().__init__(*args, directoryDIRECTORY, **kwargs) def log_message(self, format, *args): # 重写此方法以减少控制台输出 pass with socketserver.TCPServer((, PORT), Handler) as httpd: print(f测试服务器启动于 http://localhost:{PORT}) print(f服务目录: {os.path.abspath(DIRECTORY)}) try: httpd.serve_forever() except KeyboardInterrupt: print(\n服务器关闭。)4.3 运行完整的集成测试现在在另一个终端或你的主脚本中启动 OpenClaw 来抓取我们刚创建的模拟网站。# main.py if __name__ __main__: # 1. 初始化组件 my_downloader Downloader(delay0.5) # 测试时延迟调低 my_parser NewsParser() my_pipelines [ ConsolePipeline(), CsvPipeline(news_data.csv) ] # 2. 创建引擎 engine OpenClawEngine( downloadermy_downloader, parsermy_parser, pipelinesmy_pipelines ) # 3. 添加种子URL我们的模拟服务器地址 start_url http://localhost:8888/index.html engine.add_start_url(start_url) # 4. 运行 engine.run()运行步骤在一个终端先运行python test_server.py启动本地测试服务器。在另一个终端运行python main.py启动 OpenClaw。观察控制台输出你会看到调度、下载、解析、存储的完整日志。同时当前目录下会生成一个news_data.csv文件里面包含了抓取到的数据。通过这个完整的测试你不仅验证了 OpenClaw 各个模块能协同工作也体验了一个爬虫从 URL 种子开始如何自动发现新链接、抓取数据并保存的全过程。这种“闭环”测试对于构建任何工具都至关重要。5. 性能优化与扩展方向从玩具到工具我们有了一个能跑起来的 OpenClaw但它目前还是单线程的“玩具”。要把它变成实用的“工具”我们需要考虑性能和扩展性。这里提供几个最关键的优化和扩展思路。5.1 引入并发从同步到异步同步的requests库在等待服务器响应时会阻塞整个线程。对于 I/O 密集型的网络爬虫这是巨大的性能瓶颈。解决方案是使用异步编程。方案一多线程/多进程优点概念相对简单利用concurrent.futures模块可以快速实现。缺点线程/进程切换有开销且 Python 的 GIL 限制了多线程的 CPU 并行能力但爬虫主要是 I/O 等待影响相对小。需要小心管理共享资源如调度器队列的线程安全。改造点将Downloader.download任务提交给线程池。调度器Scheduler必须使用线程安全的queue.Queue我们已经做了。需要实现一个工作线程函数从队列取任务下载解析再投递新任务。方案二异步 I/O (asyncio aiohttp)优点这是现代 Python 高性能网络编程的首选。单线程即可处理成千上万个并发连接资源利用率极高。缺点需要理解异步编程范式async/await代码改造幅度较大。改造点将Downloader重写为AsyncDownloader使用aiohttp.ClientSession。将Scheduler的方法改为async可能使用asyncio.Queue。主循环run需要改写成异步的并使用asyncio.gather或asyncio.create_task来并发执行多个下载任务。解析和管道处理如果是 CPU 密集型可以考虑放到线程池中执行避免阻塞事件循环。一个简单的多线程改造示意from concurrent.futures import ThreadPoolExecutor, as_completed class OpenClawEngine: # ... __init__ 等部分不变 ... def run(self, max_workers3): # 增加最大工作线程数参数 print([引擎] OpenClaw 启动) for pipeline in self.pipelines: if hasattr(pipeline, open_spider): pipeline.open_spider() with ThreadPoolExecutor(max_workersmax_workers) as executor: future_to_url {} try: # 初始提交一批任务 initial_tasks [] while len(initial_tasks) max_workers and self.scheduler.has_pending_tasks(): url self.scheduler.get_task() if url: initial_tasks.append(url) for url in initial_tasks: future executor.submit(self._process_one_url, url) future_to_url[future] url # 处理完成的任务并提交新任务 while future_to_url: done, _ concurrent.futures.wait(future_to_url.keys(), return_whenconcurrent.futures.FIRST_COMPLETED) for future in done: url future_to_url.pop(future) try: new_urls_from_task future.result() # 获取任务结果新URL列表 for new_url in new_urls_from_task: self.scheduler.add_task(new_url, url) except Exception as e: print(f处理 {url} 时出错: {e}) # 补充新任务到线程池 if self.scheduler.has_pending_tasks(): next_url self.scheduler.get_task() if next_url: new_future executor.submit(self._process_one_url, next_url) future_to_url[new_future] next_url except KeyboardInterrupt: executor.shutdown(waitFalse, cancel_futuresTrue) print(\n[引擎] 用户中断停止爬取。) finally: for pipeline in self.pipelines: pipeline.close() print([引擎] 爬取结束。) def _process_one_url(self, url): 处理单个URL的完整流程适配线程池调用。 html self.downloader.download(url) if html is None: return [] data_items, new_urls self.parser.parse(html, url) for item in data_items: for pipeline in self.pipelines: pipeline.process_item(item) return new_urls # 返回新URL列表供主线程加入调度器注意这是一个高度简化的多线程示例真实场景下需要更精细的任务调度、错误处理和队列管理。但它清晰地展示了将阻塞的下载任务放到线程池中的思路。5.2 应对反爬虫策略提升生存能力真实的网站会有反爬虫措施。我们的 OpenClaw 需要一些“盔甲”。User-Agent 轮换准备一个 User-Agent 列表每次请求随机选择一个。class SmartDownloader(Downloader): def __init__(self, delay1.0, max_retries3): super().__init__(delay, max_retries) self.user_agents [ Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 ..., Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 ..., # ... 更多UA ] def download(self, url, headersNone): final_headers {**self.default_headers, **(headers or {})} final_headers[User-Agent] random.choice(self.user_agents) # 随机选择 # ... 其余逻辑不变请求头完善模拟更真实的浏览器添加Accept、Accept-Language、Referer可设置为当前页面的来源页等头部信息。Cookie 管理对于需要登录的网站使用requests.Session()可以自动管理 Cookie。对于更复杂的场景可能需要从浏览器导出 Cookie 文件并加载。IP 代理池当单个 IP 请求过于频繁被封锁时需要使用代理 IP。可以集成第三方代理服务或者在下载器中加入代理切换逻辑。class ProxiedDownloader(Downloader): def __init__(self, proxy_listNone, **kwargs): super().__init__(**kwargs) self.proxy_list proxy_list or [] self.proxy_index 0 def get_proxy(self): if not self.proxy_list: return None proxy self.proxy_list[self.proxy_index] self.proxy_index (self.proxy_index 1) % len(self.proxy_list) return {http: proxy, https: proxy} def download(self, url, headersNone): # ... 在session.get中增加proxies参数 proxy self.get_proxy() response self.session.get(url, headersfinal_headers, timeout10, proxiesproxy) # ...请求频率控制与随机延迟我们已经实现了基础版本。更高级的可以针对不同域名设置不同的延迟策略。5.3 扩展数据管道与调度策略更多存储后端继承Pipeline基类可以轻松实现MongoDBPipeline、MySQLPipeline、JsonPipeline等将数据存入不同目的地。优先级调度修改Scheduler使用heapq实现优先队列。可以为不同的 URL 设置优先级例如详情页优先级高于列表页。断点续爬将Scheduler中的seen_urls集合和任务队列定期序列化到磁盘如用pickle保存。爬虫重启时加载回来可以从上次中断的地方继续。布隆过滤器去重当 URL 数量达到百万级以上时内存中的set会非常庞大。可以使用pybloom-live等库实现布隆过滤器用很小的内存和一定的误判率实现海量 URL 去重。通过以上这些扩展你的 OpenClaw 就能从一个教学演示项目进化成一个能够应对中等复杂度真实场景的定制化爬虫框架。这个过程本身就是对“手撸”精神最好的诠释从理解每一个齿轮开始最终组装成一台属于自己的、运转良好的机器。
