先说个真实经历。我最早写爬虫的时候思路特别直白requests 拿页面BeautifulSoup 提取字段写进 CSV然后再 for 循环下一个 URL。看起来没毛病一百多行的脚本跑得也挺欢。直到需求变了三次——先是换数据结构然后加了一个来源站最后要求对相同标题的文章做去重——同一个逻辑被复制粘贴改了三遍代码就成了一团浆糊。那时候我才意识到爬虫的难点从来不在“把数据拿下来”而在“拿下来的数据你能不能稳定地、持续地处理”。这是网络安全学习篇的第37篇这篇不聊具体攻防想认真拆一个很多人都会忽略的基础功用面向对象思想来组织爬虫代码。对于做安全研究的人来说无论是收集公开情报、梳理资产信息还是做合规的数据采集写一手能维护的爬虫几乎算半个必修课。而这篇面向的读者就是那些已经会用 requests 和 BeautifulSoup但还不知道怎么把代码写得更清楚的初学者。这篇文章会给你一套可以直接照抄的类设计也会告诉你为什么这么拆、哪些坑我替你踩过了。全文拿一个通用的文章抓取场景做演示URL 用占位域名你只需要把它替换成自己有权采集的站点。1. 从命令式脚本到对象模型的思维转换1.1 脚本式爬虫为什么越写越乱我见过太多爬虫项目的起点都是同一个模样一个 .py 文件从上到下依次写着导入库、定义 headers、发请求、解析 HTML、存文件、循环地址列表。一开始只有三五个页面这么写完全够用。但爬虫这玩意有个特点它天生是给“变化”准备的。目标站点的页面结构会改你需要抓的字段会加采集范围会从列表页变成详情页甚至一个需求没说清楚第二天就要换来源站。一旦变化来了脚本的麻烦就非常明显。要做两个站的采集最省事的办法是把解析代码复制一份改几个选择器。等到两个站点都调整了结构你要改的地方就翻倍。如果你想给请求统一加一个重试机制又得在几段重复代码里各改一遍。说实话这种代码不是不能跑而是每次改完都觉得自己在排雷不知道还有哪一处没同步到。我自己最崩溃的一次是线上任务忽然抓不到新数据。排查了半天发现原因仅仅是某个页面的 CSS 类名换了一个单词而那段解析逻辑在脚本里出现了两处我只改了其中一处。这种错误不算难修但它非常消耗耐心也会让人怀疑自己是不是不适合写代码。1.2 对象模型到底在解决什么问题面向对象不是某种神秘的编程仪式它的本质就一句话把数据和对数据的操作绑在一起并且把“会变化的部分”单独隔离出来。拿爬虫场景来说一次完整的采集里有几个角色天然就是要分开的。负责发请求的“网络客户端”负责从 HTML 里提东西的“解析器”负责保存结果的“数据管道”以及被采集的那条数据本身。如果这些角色各自成为一个对象那么站点结构变了你改的是解析器反爬策略变了你改的是网络客户端输出格式变了你改的是数据管道。它们之间通过明确的接口协作互相不知道对方内部怎么实现。用一个生活化的类比一个餐厅不可能让厨师既去前台收银又去后厨洗菜。分工明确以后每一环都能独立优化、独立换人。爬虫代码也是一样。你不需要把所有的逻辑都塞进一个大脑而是让每一类对象只操心自己那一摊事。1.3 什么情况下值得用面向对象重写我也要泼一点冷水不是所有爬虫都必须上对象模型。如果你只是临时抓一个接口、跑一次就丢那脚本完全没毛病强行封装反而显得笨重。我的判断标准有三个。第一这个爬虫要持续跑一周以上第二你预感到来源站或者字段会变第三这个代码可能不止你一个人维护。满足任意一条我都会直接按面向对象的思路去写。不满足那就怎么快怎么来。还有一种情况也值得重写你发现自己在频繁地复制粘贴代码。复制粘贴一次是偷懒复制粘贴三次就说明原来的抽象不到位这时把公共逻辑抽成类收益最大。2. 爬虫对象的职责划分四个核心角色与设计取舍2.1 四个核心类一张表讲清协作关系我把一个采集任务拆成四个核心对象外加一个调度入口。它们的职责边界是这样的类名职责依赖方向HttpClient负责发送 HTTP 请求、处理超时与重试、维护会话不依赖其他业务类ArticleParser负责把 HTML 解析成结构化的数据对象依赖 ArticleArticle数据模型承载一条文章的所有字段和校验逻辑无依赖Pipeline负责把 Article 持久化到文件或数据库依赖 ArticleCrawler调度器编排以上对象完成完整采集流程依赖所有类但只通过方法接口调用这么拆分以后每个类都能单独拎出来测试。HttpClient 可以在没有真实站点的情况下用 mock 数据做单元测试Pipeline 也可以单独塞一条假数据验证写文件逻辑。而 Crawler 这一层实际上非常薄它只做“循环、调用、异常兜底”这三件事。2.2 数据模型Article 类的设计细节很多初学者写爬虫时会忽略数据模型直接拿字典到处传。字典用起来确实方便但问题在于它太自由了。字段拼写错了不会有人提醒你类型不对也只能等程序跑挂了才知道。我的习惯是先定义一个 Article 类字段明确写出来import hashlib class Article: def __init__(self, title, author, publish_time, content, url, source): self.title title self.author author self.publish_time publish_time self.content content self.url url self.source source property def unique_id(self): raw f{self.source}:{self.url} return hashlib.md5(raw.encode(utf-8)).hexdigest() def to_dict(self): return { title: self.title, author: self.author, publish_time: self.publish_time, content: self.content, url: self.url, source: self.source, unique_id: self.unique_id, }重点说一下 unique_id 这个属性。爬虫经常要重复运行同一篇文章可能被抓到好几次。如果只用 URL 去重某些站点的 URL 后面带跟踪参数同一个文章会对应好几个 URL去重就失效。我拿 source 加 url 拼在一起做 MD5等于给每篇文章一个业务上的唯一指纹。后面做断点续爬、增量更新都靠这个字段。to_dict 方法则是为了后续序列化准备的。Pipeline 拿到 Article 对象以后调一下 to_dict 就能变成字典再输出去 JSON 或者数据库都方便。注意这里我没有把保存的逻辑写进 Article 类因为“怎么存”是管道的事不是数据模型的事。2.3 继承还是组合别急着建基类面对多个来源站时新手很容易想到“建一个 BaseParser然后让每个站的 Parser 继承它”。这个思路不算错但实际用起来经常会后悔因为不同站的 HTML 结构差异可能非常大强行抽象出来的父类要么塞满了 if else要么每个子类都在重写父类方法父类反而变成了装饰品。我更推荐的做法是组合加配置解析器只需要一套通用逻辑具体的选择器通过配置传进来。同一个 ArticleParser 类换一套 CSS 选择器就是另一个站的解析器。这样做的好处非常直接新增一个站点时不用写新的 Python 类加一份配置就够了。selectors { title: h1.post-title, author: .post-author, publish_time: time.post-time, content: div.post-content, }继承的问题在于它把“公共逻辑”和“差异逻辑”硬绑定在类层级里。而组合的思路是公共逻辑留在类里差异逻辑变成配置数据。爬虫这种场景差异几乎都集中在选择器和字段映射上用配置驱动显然更灵活。3. 核心代码实现网络层、解析层、数据层完全解耦3.1 HttpClientSession、重试与限流策略网络层是所有爬虫的地基。很多爬虫挂掉不是解析代码写得差而是请求环节没做好容错。一个像样的 HttpClient 至少要处理三件事会话保持、超时重试、异常转换。import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry class HttpClient: def __init__(self, headersNone, timeout10, max_retries3, backoff_factor0.5): self.session requests.Session() self.timeout timeout retry Retry( totalmax_retries, connectmax_retries, readmax_retries, backoff_factorbackoff_factor, status_forcelist[500, 502, 503, 504], allowed_methodsfrozenset([GET, POST]), ) adapter HTTPAdapter( max_retriesretry, pool_connections5, pool_maxsize10, ) self.session.mount(http://, adapter) self.session.mount(https://, adapter) default_headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36, Accept: text/html,application/xhtmlxml,application/xml;q0.9,*/*;q0.8, } default_headers.update(headers or {}) self.session.headers.update(default_headers) def get(self, url, paramsNone): resp self.session.get(url, paramsparams, timeoutself.timeout) resp.raise_for_status() return resp.text这里面有几个值得展开的细节。为什么要用 Session 而不是每次 requests.get因为 Session 会自动保存 Cookie并且底层复用 TCP 连接。采集大量页面时连接复用的收益非常明显握手的时间省掉以后整体速度能快不少。Retry 配置里那个 backoff_factor0.5表示第一次重试前等待 0.5 秒第二次等待 1 秒第三次等待 2 秒呈指数退避。这样做是为了在服务端短暂抖动时自动恢复同时又不会因为疯狂重试给目标站点造成压力。特别提醒一下429 Too Many Requests 这个状态码我没有放进 status_forcelist。原因是它本身就代表“你太快了请慢一点”。遇到这种情况正确做法是停下来等一等而不是让重试机制继续加大火力。把 429 加入自动重试是很多爬虫被对方封禁的常见原因。3.2 ArticleParser只对 HTML 负责解析器要做的只有一件事输入 HTML 字符串和 URL输出一个 Article 对象。它不应该关心请求怎么发的也不应该关心数据最后存到哪。from bs4 import BeautifulSoup class ArticleParser: def __init__(self, selectors): self.title_selector selectors[title] self.author_selector selectors.get(author) self.publish_time_selector selectors.get(publish_time) self.content_selector selectors[content] def parse(self, html, url, source): soup BeautifulSoup(html, lxml) title self._extract_text(soup, self.title_selector) author self._extract_text(soup, self.author_selector) publish_time self._extract_text(soup, self.publish_time_selector) content self._extract_text(soup, self.content_selector) return Article( titletitle, authorauthor, publish_timepublish_time, contentcontent, urlurl, sourcesource, ) def _extract_text(self, soup, selector): if not selector: return node soup.select_one(selector) return node.get_text(stripTrue) if node else 我用的是 CSS 选择器而不是 XPath。对我来说CSS 选择器在 BeautifulSoup 里可读性更好而且目标站点改版时浏览器开发者工具直接复制的也是 CSS 选择器路径调试效率更高。如果你更习惯 XPath把 select_one 换成 select_one 加 xpath 处理ArticleParser 的整体结构不用变只是 _extract_text 内部实现换掉而已。这里还有一个细节并不是每个页面都有作者字段。比如一些聚合页面可能只有一个标题和内容所以 author_selector 和 publish_time_selector 我用了 get 方法允许缺省。解析器对字段缺失要有容错能力不能让一条脏数据导致整个任务崩溃。3.3 Pipeline最后一步的持久化Pipeline 负责把 Article 写成文件、插入数据库或者推送到消息队列。在这里我做一个最简单的 JSON 文件管道把每次采集的结果集中写到一个数组里。import json class JsonPipeline: def __init__(self, filepath): self.filepath filepath self._items [] def save(self, article): self._items.append(article.to_dict()) def flush(self): with open(self.filepath, w, encodingutf-8) as f: json.dump(self._items, f, ensure_asciiFalse, indent2)我用了“先攒内存、最后一次性写文件”的策略。这种写法适合数据量不大、单次任务跑完再统一落盘的场景。如果任务要持续抓几万条我会改成每攒够 100 条就写一次或者直接换成 SQLite避免进程意外中断时损失太多数据。设计模式上这两者差别不大Pipeline 对于调用方藏掉了保存的细节想换存储方案就补一个新类。3.4 为什么每个类可以独立测试这套结构最大的好处是你不需要真的去爬一个网站也能把每个组件测清楚。我平时会写几个简单的测试脚本直接用假 HTML 喂给 ArticleParser检查解析出来的字段对不对也会用本地文件路径测 Pipeline确认 JSON 结构符合预期。一旦某个类出了问题排查范围被缩小到了非常明确的一层。页面结构变了不会影响到网络层网络超时了不会导致数据错乱。代码逻辑像流水线一样清晰这是脚本式爬虫很难达到的。4. 实例主流程Crawler 如何把三类对象编排起来4.1 调度器的完整实现Crawler 是整条流水线的装配工。它持有 HttpClient、ArticleParser、Pipeline 三个对象然后定义采集的完整流程。这里我会让它做四件事遍历 URL、逐个请求、解析、保存。import requests class Crawler: def __init__(self, http_client, parser, pipeline): self.http_client http_client self.parser parser self.pipeline pipeline def crawl(self, start_urls): for url in start_urls: try: html self.http_client.get(url) article self.parser.parse(html, url, sourceexample_site) self.pipeline.save(article) print(f[OK] {article.title}) except requests.RequestException as e: print(f[网络错误] {url} - {e}) except Exception as e: print(f[解析错误] {url} - {e}) self.pipeline.flush()主流程看起来非常简单因为复杂的逻辑都被拆到各个类里去了。Crawler 唯一要做好的事情就是异常隔离。4.2 异常链路的处理与资源释放我在 crawl 方法里把网络异常单独放在一个 except 分支这样做的目的是区分“这个页面是网络问题导致没抓到”和“页面拿到了但解析失败”。这两个问题的排查方向完全不同。不过上面的代码有一个隐患如果某个 URL 在解析阶段抛异常后续 URL 还是会继续跑但假设前面已经 save 了一些数据最后的 flush 依然会执行因为 flush 放在 for 循环外面。这算是不错的兜底。但更稳妥的方式是用 try/finally 保证 flush 一定会被调用def crawl(self, start_urls): try: for url in start_urls: try: html self.http_client.get(url) article self.parser.parse(html, url, sourceexample_site) self.pipeline.save(article) print(f[OK] {article.title}) except requests.RequestException as e: print(f[网络错误] {url} - {e}) except Exception as e: print(f[解析错误] {url} - {e}) finally: self.pipeline.flush()这样哪怕 for 循环内部出现了我没有预想到的异常已抓到的数据也能安全落盘。Py 开发者常说的“资源释放必须放进 finally”就是这个意思。Pipeline 的文件句柄可能不涉及显式 close但 flush 本身就是一种资源释放优先级最高。4.3 组装入口main 函数示例为了让整个例子可以直接跑起来我写一个入口if __name__ __main__: selectors { title: h1.post-title, author: .post-author, publish_time: time.post-time, content: div.post-content, } http_client HttpClient( headers{ User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 }, timeout10, max_retries3, ) parser ArticleParser(selectors) pipeline JsonPipeline(./output/articles.json) crawler Crawler(http_clienthttp_client, parserparser, pipelinepipeline) crawler.crawl( [ https://example.com/posts/1, https://example.com/posts/2, https://example.com/posts/3, ] )打开终端跑一遍正常输出大概是这样[OK] 使用Python快速处理常见文本格式 [OK] 爬虫项目中的代码组织与测试技巧 [网络错误] https://example.com/posts/3 - HTTPSConnectionPool(...)网络错误的那个页面不影响前两条数据的保存。任务结束后打开 output/articles.json里面就是结构化好的数据。5. 并发改造与系统鲁棒性类结构如何支撑项目成长5.1 加并发但不要动类的接口单线程爬虫的瓶颈通常在网络等待。一个页面从发起请求到拿到 HTML可能几百毫秒而 CPU 解析只需要几毫秒。多线程确实能带来明显加速但很多人一上来就改业务逻辑把请求、解析、保存混在一起结果线程安全问题和代码复杂度一起爆发。我更喜欢用 Python 内置的 concurrent.futures 来做因为需要改动的只有 Crawler 内部的一小段逻辑HttpClient、ArticleParser、Pipeline 的接口都不用动from concurrent.futures import ThreadPoolExecutor class Crawler: def __init__(self, http_client, parser, pipeline, workers4): self.http_client http_client self.parser parser self.pipeline pipeline self.workers workers def _fetch_and_parse(self, url): html self.http_client.get(url) return self.parser.parse(html, url, sourceexample_site) def crawl(self, start_urls): with ThreadPoolExecutor(max_workersself.workers) as executor: futures [executor.submit(self._fetch_and_parse, url) for url in start_urls] for future in futures: try: article future.result() self.pipeline.save(article) print(f[OK] {article.title}) except requests.RequestException as e: print(f[网络错误] {e}) except Exception as e: print(f[解析错误] {e})为什么能只改 Crawler因为每个类的职责足够单一。HttpClient 用 Session 发请求requests 的 Session 本身是线程安全的多个线程共享同一个 Session 不会有问题。ArticleParser 没有内部可变状态每调用一次 parse 就生成一个新的 Article天然线程安全。唯一要注意的是 Pipeline因为多个线程会同时调 save。5.2 Pipeline 的线程安全与幂等设计在我的 JsonPipeline 里_items 是一个列表多个线程同时 append 在 CPython 里受 GIL 保护不会出现列表损坏但顺序会乱而且更严谨的做法是显式加锁。我把 save 方法升级一下import threading class JsonPipeline: def __init__(self, filepath): self.filepath filepath self._items [] self._lock threading.Lock() def save(self, article): with self._lock: self._items.append(article.to_dict())加锁是必要的防御性编程。虽然 GIL 让单个 append 不会崩但当你把“读取已抓取集合”“判断是否重复”“加入集合”这几个动作放在一起时没有锁就会出重复数据。比如多线程同时抓到了同一篇文章两个线程都发现集合里没有这个 ID然后都把它保存了。去重逻辑也可以放在 Pipeline 里class JsonPipeline: def __init__(self, filepath): self.filepath filepath self._items [] self._seen_ids set() self._lock threading.Lock() def save(self, article): with self._lock: if article.unique_id in self._seen_ids: return self._seen_ids.add(article.unique_id) self._items.append(article.to_dict())这样即使两个线程同时处理同一篇文章第二个线程也会因为 unique_id 已经在集合里而跳过。这也是我在 2.2 里为什么强调唯一指纹的原因——没有它去重逻辑根本没法做。5.3 请求频率控制与合规边界并发提上去以后必须同时把限速做起来。我的经验是用一个小工具类控制两次请求的最小间隔import time import threading class RateLimiter: def __init__(self, min_interval1.0): self.min_interval min_interval self._last_time 0.0 self._lock threading.Lock() def wait(self): with self._lock: now time.time() delta now - self._last_time if delta self.min_interval: time.sleep(self.min_interval - delta) self._last_time time.time()HttpClient 在 get 方法里调用一下 limiter.wait()所有线程的请求频率就会被全局控制住。爬虫是采集任务不是压力测试。礼貌地把频率控制在对方服务器完全无感的范围这既是技术问题也是基本的合规意识。动手前看一眼目标站点的 robots.txt 也是起码的礼貌。本文示例里的 example.com 只是占位域名你换成真实来源站之前务必确认自己有权采集。6. 面向对象设计里常见的过度设计与反模式6.1 类爆炸把函数搬进类并不是封装我见过有人把一个几百行的爬虫拆成二十多个类每个类只有一个方法方法名还起得特别抽象。这其实是把函数塞进类里并没有真正理解封装的用意。封装的价值在于把“一组相关的数据和操作”放在一起而不是为了让代码看起来更“面向对象”。一个只有一个方法且没有内部状态的类本质上就是一个函数不如直接用普通函数。判断标准很简单这个类有没有需要长期维护的内部状态如果有类才有意义如果没有函数就够了。HttpClient 的内部状态是 SessionPipeline 的内部状态是待写入的列表它们自然应该做成类。而一个 Tools 类里躺着七八个互不相干的静态方法这就属于典型的类爆炸。6.2 可变共享状态Session 不是垃圾桶另一个很容易踩的坑为了让某些地方省事往 Session 里塞各种自定义属性比如当前请求次数、上一次请求时间、临时缓存。Session 本身是线程安全的但那是针对它自己的内部实现。你往里塞自定义的可变状态多线程一跑数据就乱了。我之前就用过这种写法把“某个网站需要特殊 Cookie”的信息挂在 Session 上结果线程一多A 站请求带着 B 站的 Cookie各种奇奇怪怪的 403。后来我教训很深Session 只负责 HTTP 层面的会话业务相关的状态要放在明确的业务对象里。HttpClient 的职责边界必须清晰不要把它当收纳盒。6.3 抽象错了比不抽象更痛苦新手容易犯的另一个错是“提前抽象”。假设你有两个来源站要抓二话不说先写一个 BaseSpider 基类把公共方法都往里放。结果写着写着发现两个站的分页方式完全不同一个靠 URL 参数一个靠点击加载于是你在基类里写了一堆 if 判断来源类型。这时你会发现这个基类并没有帮你减少任何重复反而让代码更绕。我更推荐的做法是先分别实现两个独立的具体类跑通之后再回头看哪些代码是真的重复然后从中提取公共部分。抽象应该是从具体需求里“长”出来的而不是一开始就“设计”出来的。等你有了两个成功案例再做父类那个父类的接口设计会合理得多。这个原则在我 2.3 里提过这里再强调一次组合优先于继承配置优先于多态。6.4 我的取舍建议最后给一个简单可行的路线。小型项目用轻量封装HttpClient、Parser、Pipeline 三个类加一个调度器就够了。项目变得复杂以后再考虑引入更重的框架比如支持插件机制的爬虫框架或者把 Pipeline 换成消息队列。类的设计要为变化留出空间但不要为想象中的变化过度设计。YAGNI 原则在爬虫场景里尤其适用没有真实需求就不要提前引入一层抽象。写在最后的个人体会回头看我自己的路径从脚本搬到对象模型最大的变化其实不是代码行数变多而是调试时间明显变短。以前出了 bug 要在两百行脚本里从头看到尾现在直接看异常堆栈几乎立刻就知道是哪一层的问题。这个体验对比非常直观。最后分享一个我一直在用的小技巧先把调用方的代码写出来再回头定义类的接口。比如先写 crawler.crawl(urls)再决定 Crawler 的构造函数需要接收什么。这种“从使用倒推设计”的方法比闭门造车设计接口要舒服得多也更容易写出贴合实际需求的类。理论的东西讲再多都不如自己跑一遍。建议你从下一个爬虫开始哪怕很简单也试着拆成 HttpClient、ArticleParser、Pipeline 三个类跑通之后你自然会理解为什么爬虫项目需要面向对象。