面向对象思想的Python爬虫实例：从脚本到工程化设计

作为一个常年跟爬虫打交道的开发者我越来越觉得如果只会用脚本堆功能写到最后基本就是给自己挖坑。这期网络安全学习笔记我想认真聊聊“基于面向对象思想的Python爬虫实例”。换句话说不是教你怎么用requests抓个页面就完事而是把爬虫当成一个正经项目来设计——用类、继承、接口约束把请求、解析、存储拆得明明白白。这个思路我在自己的模拟项目X里实测过代码写到第三版基本零改动就能复用到同类站点排查问题也不像以前那样瞎猜。这篇文章适合谁一是写过几段爬虫但代码越改越乱的初学者二是在安全分析中需要快速复现数据采集逻辑的人三是想让自己写的工具能被团队其他人接手使用的开发者。我会从一个完整实例出发把类怎么分、方法怎么设计、异常怎么兜底、日志怎么埋点一步步讲清楚最后再整理几个真实踩过的坑。1. 为什么爬虫这种“临时脚本”也值得用面向对象去重构1.1 脚本式爬虫的“三宗罪”很多人入门爬虫都是这么写的一个py文件从上到下requests.get然后正则或者BeautifulSoup提取最后写CSV。这个写法对于一次性任务完全没问题但如果目标网站稍微改个版、请求频率变了、需要加代理你就要在一大段顺序代码里翻来覆去地找那个关键行。我把它总结成三个痛点。第一逻辑全部耦合在一起。请求超时要改解析规则要改存储字段要改全部堆在同一个函数里。你改一个字段的时候极容易动到别的逻辑而且自己毫不知情。第二复用几乎是零。今天抓A站明天抓B站哪怕两个站的数据结构高度相似你也得把代码复制一份改一改。复制一时爽维护火葬场如果哪天发现公共部分有bug所有副本都要手动修一遍。第三单元测试根本无从下手。脚本式写法里你很难单独测某一段解析逻辑因为解析依赖前一步的请求状态中间还夹着各种全局变量。有个很形象的比喻脚本式爬虫像你临时抓几根电线接在一起能亮面向对象爬虫像做了一个配电箱虽然前期布线要多花点时间但哪路开关控制哪个电器清清楚楚出了跳闸直接看空气开关就行。1.2 面向对象给爬虫带来的三个直接收益第一个收益是封装。每个类只干一件事数据模型类只管字段定义和数据校验请求类只管发请求、处理超时和重试解析类只管把HTML变成结构化数据。调用者根本不需要关心内部细节只需要知道传入什么、返回什么。第二个收益是继承。你可以把“发送请求 → 解析响应 → 清洗数据 → 存储结果”这个流程定义为基类方法子类只需要重写其中的解析部分其他步骤全部复用。这个收益在做多站点数据采集时极其明显我在模拟项目X里维护了三个不同的数据源页面结构完全不同但公共逻辑写了不到两遍。第三个收益是接口约束带来的安全感。通过抽象类定义好方法签名所有子类都必须实现某个方法否则类实例化直接报错。这样做的好处是哪怕这个类是三天前写的你回忆一下基类的方法约定就能知道从哪里入手。1.3 什么时候值得用类什么时候脚本也够用我并不是说所有爬虫都非要用面向对象。如果你只是从某个接口拉一次数据、写一次脚本跑完删掉那用类反而多余。这里我给一个简单的判断标准这段代码你预计会再运行三次以上或者可能被别人拿去改或者目标站点未来有变化那么请先写类设计再写实现。另外如果你在网络安全学习过程中需要做应急的数据取证或者写一个小的PoC验证面向对象的封装也能帮你把请求逻辑与证据解析解耦后续扩展成定期巡检工具更加顺手。但是要注意爬虫工具的使用必须限定在授权范围内不要对无关站点做高频请求也不要尝试破解任何访问控制机制。2. 爬虫类设计整体思路与职责划分2.1 先建数据模型用数据类固定字段写爬虫类之前我非常建议先写一个数据类来固定“我们要拿什么”。拿我之前做的一个图书信息采集项目举例我需要从页面上拿到书名、作者、出版社、出版日期、定价、ISBN号、内容简介、封面图片链接这8个字段。把这些字段写进一个数据类里有几个好处解析函数返回什么类型是确定的将来写存储模块的时候不用再猜字段名代码补全器也能直接提示。这里我使用Python的dataclasses模块避免手写一堆__init__和__repr__模板代码。数据类里我还会设计一个validate方法检查必填字段不为空检查价格是否能转成数字。这个思路其实是把数据校验往前移以免脏数据一路传到数据库才发现问题。2.2 请求类与解析类各管一段常见的爬虫设计有两种风格。一种是把请求和解析都放在同一个类里另一种是单独拆分请求客户端和解析器。我偏向于前者但做了微调基类中定义好骨架流程子类通过模板方法的方式只补充解析和字段映射逻辑。举个例子基类里我会写好fetch_page方法负责构造请求头、带上超时、捕获网络异常写好run方法负责把整个流程串起来。子类只需要提供一个parse_response方法接收响应对象或者HTML文本然后返回数据对象的列表。这样带来的直接好处是如果目标站点改了请求头校验我只在基类里加一次所有子类都生效。但如果是请求逻辑异常复杂的站点比如需要登录、加密参数、动态token我建议把请求逻辑单独拆成一个类因为这时候请求已经不是“通用技能”了它本身就成了一块核心业务。在这个实例中为了保持教程的通用性我统一在基类里维护请求逻辑再留一个build_headers方法供子类重写。2.3 存储模块接口设计比实现更关键数据拿到之后怎么存存CSV、存JSON、存SQLite还是推送数据库很多人在这一步就开始纠结选型。我的做法是定义好一个存储接口不在基类里写死。具体来说保存函数接收一个迭代器每个元素都是数据对象实例。至于存到哪里是CSV还是SQLite完全可以换。这种设计的核心思路是让存储方式的变化不影响爬虫主体逻辑。如果今天要改成存数据库只需要替换一个存储实现类而不需要改动任何解析代码。3. 手写一个书籍信息采集爬虫的完整实现3.1 准备工作与依赖这个实例我基于Python 3.10环境用到的主要依赖如下pip install requests beautifulsoup4 lxmlrequests负责HTTP请求BeautifulSoup负责HTML解析lxml作为解析器速度和容错性都优于Python内置的html.parser。此外还会用到dataclasses标准库、logging标准库、time标准库和sqlite3标准库。我在模拟项目X里使用的是https://books.example.com这个测试站点页面结构和字段都做过固定读者可以仿照这个结构去套自己的目标站点。3.2 数据类定义先把字段和校验写清楚from dataclasses import dataclass, field from typing import List, Optional import re dataclass class BookInfo: title: str authors: List[str] field(default_factorylist) publisher: str pub_date: str price: float 0.0 isbn: str summary: str cover_url: str def validate(self) - bool: if not self.title or not self.title.strip(): return False if not self.isbn or not re.match(r^[\dXx-]{10,17}$, self.isbn): return False try: float(self.price) except (TypeError, ValueError): return False return True这个数据类里的validate方法是我特别想强调的。它在数据被使用之前就拦截掉非法的ISBN和价格格式避免了后面入库时才发现问题。authors字段用field(default_factorylist)而不是[]也是为了防止同一个默认列表被多个实例共享。3.3 抽象基类固定流程模板import requests import logging from abc import ABC, abstractmethod from typing import List, Dict, Any, Iterator, Optional logger logging.getLogger(base_spider) class BaseSpider(ABC): BASE_URL REQUEST_HEADERS { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0 Safari/537.36 } REQUEST_TIMEOUT 15 def __init__(self, start_url: str , max_retry: int 3): self.current_url start_url or self.BASE_URL self.max_retry max_retry self.session requests.Session() self.session.headers.update(self.REQUEST_HEADERS) abstractmethod def parse_response(self, html: str) - Iterator[BookInfo]: 子类必须实现解析方法 pass def fetch_page(self, url: str, params: Optional[Dict] None) - requests.Response: for attempt in range(self.max_retry): try: resp self.session.get(url, paramsparams, timeoutself.REQUEST_TIMEOUT) resp.raise_for_status() resp.encoding resp.apparent_encoding logger.info(请求成功: %s, 状态码: %d, 耗时: %.2fs, url, resp.status_code, resp.elapsed.total_seconds()) return resp except requests.RequestException as e: logger.warning(第 %d 次请求 %s 失败: %s, attempt 1, url, e) if attempt self.max_retry - 1: raise time.sleep(2 ** attempt) raise RuntimeError(请求失败) def save_items(self, items: Iterator[BookInfo]) - int: 默认什么都不做由子类或外部传入存储策略 count 0 for item in items: count 1 return count def run(self): resp self.fetch_page(self.current_url) items self.parse_response(resp.text) total self.save_items(items) logger.info(本次采集结束, 共 %d 条数据, total) return total这段代码里我用了抽象方法parse_response强制子类实现解析逻辑同时把请求重试、日志、编码处理都封装在基类里。run方法作为一个模板方法定义好了算法骨架请求 → 解析 → 保存。子类只需要填解析这一步。其中重试时的退避逻辑是每次重试等待时间为2的幂次秒数第一次失败等2秒第二次失败等4秒。这样做是为了降低连续失败时对目标站点造成的压力。实际工作中这个数字可以根据目标站点的响应速度调整但至少保证重试不是死等也不是疯狂重连。3.4 子类实现真正干活的地方现在实现一个针对目标站点的爬虫子类。假设页面里每个图书条目都放在article classbook-item中字段分布是标题在h2.title下作者在.authors span出版社和出版日期在.meta里价格在.priceISBN在.isbn简介在.summary封面图片在img标签的>from bs4 import BeautifulSoup from typing import Iterator class BooksSpider(BaseSpider): BASE_URL https://books.example.com/catalog def parse_response(self, html: str) - Iterator[BookInfo]: soup BeautifulSoup(html, lxml) for article in soup.select(article.book-item): item BookInfo( titlearticle.select_one(h2.title).get_text(stripTrue) if article.select_one(h2.title) else , authors[a.get_text(stripTrue) for a in article.select(.authors span)], publisherarticle.select_one(.meta .publisher).get_text(stripTrue) if article.select_one(.meta .publisher) else , pub_datearticle.select_one(.meta .pub-date).get_text(stripTrue) if article.select_one(.meta .pub-date) else , priceself._parse_price(article), isbnarticle.select_one(.isbn).get_text(stripTrue) if article.select_one(.isbn) else , summaryarticle.select_one(.summary).get_text(stripTrue) if article.select_one(.summary) else , cover_urlarticle.select_one(img.book-cover).get(data-src, ) if article.select_one(img.book-cover) else , ) if item.validate(): yield item else: logger.warning(跳过无效条目: %s, item.title) staticmethod def _parse_price(article: BeautifulSoup) - float: price_node article.select_one(.price) if not price_node: return 0.0 text price_node.get_text(stripTrue) digits re.sub(r[^\d.], , text) try: return float(digits) * 100 if in text or 元 in text else float(digits) except ValueError: return 0.0这个子类中每个字段都用select_one(...)取出节点再调用get_text并且都做了空值保护。这里特别要说一下我的习惯所有对页面的字段解析必须带上空值判断不然某一条数据缺了一个节点整个列表推导式就会报AttributeError然后就断掉了。_parse_price这个方法做了一个很实际的兼容有些页面显示的是“49.00”有些显示的是“49元”有些直接是“49.00”。我先把非数字字符去掉再根据页面上的货币符号做统一换算。这里换算成“分”这个单位其实是有意为之——浮点数直接存储会导致精度问题所以我乘以100转成整数来规避。如果读者只是存成文本展示那这一步可以跳过。3.5 存储到SQLite替换掉默认存储策略基类里save_items只是随便计数现在我们要把真实数据落到本地。我单独写了一个存储函数并覆盖基类方法。import sqlite3 class BooksSpiderWithStorage(BooksSpider): DB_PATH books.db def __init__(self, start_url: str ): super().__init__(start_url) self._init_db() def _init_db(self): with sqlite3.connect(self.DB_PATH) as conn: conn.execute( CREATE TABLE IF NOT EXISTS books ( id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT NOT NULL, authors TEXT, publisher TEXT, pub_date TEXT, price INTEGER, isbn TEXT UNIQUE, summary TEXT, cover_url TEXT ) ) def save_items(self, items: Iterator[BookInfo]) - int: count 0 with sqlite3.connect(self.DB_PATH) as conn: for item in items: try: conn.execute( INSERT OR IGNORE INTO books (title, authors, publisher, pub_date, price, isbn, summary, cover_url) VALUES (?, ?, ?, ?, ?, ?, ?, ?) , ( item.title, ,.join(item.authors), item.publisher, item.pub_date, int(item.price), item.isbn, item.summary, item.cover_url )) count 1 except sqlite3.Error as e: logger.error(写入数据库失败: %s, ISBN: %s, e, item.isbn) return count这个实现里有几个细节值得讲。第一个是CREATE TABLE IF NOT EXISTS保证多次运行不会因为表已存在而报错。第二个是INSERT OR IGNORE因为表中定义了isbn字段为UNIQUE重复采集时会自动跳过实现天然的去重。第三个是with sqlite3.connect(DB_PATH) as conn这个上下文管理器在退出时会自动提交事务如果中途抛异常会自动回滚。不需要手动写conn.commit()代码也更干净。价格字段在SQLite里存成整数因为前面已经做过乘以100的处理。authors列表用逗号拼接成一个文本字段虽然这样在数据库里做作者级别的筛选不太方便但对一个采集工具来说足够用了。如果后续要做更细致的查询可以建单独的作者表那是后话。3.6 主程序把流程串起来import logging logging.basicConfig( levellogging.INFO, format%(asctime)s [%(name)s] %(levelname)s: %(message)s, datefmt%Y-%m-%d %H:%M:%S ) if __name__ __main__: spider BooksSpiderWithStorage() total spider.run() print(f采集完成入库 {total} 条记录)运行这个脚本后日志里会输出每次请求的状态码、耗时最后输出入库条数。如果网络正常、页面结构匹配你应该能在books.db里看到一张books表里面躺着采集回来的数据。这个过程中如果发现请求失败脚本会按指数退避重试重试三次后抛出异常。对于模拟项目X我设置的重试上限是3次因为那个站点偶尔会返回503但业务高峰期最多持续十几秒。如果目标站点的故障恢复时间更长可以把max_retry调到5甚至8。4. 避坑实录抓取过程的典型问题与排查思路4.1 编码问题比想象中更常见网页编码是爬虫新手最容易踩的坑之一。很多网站明明声明了charsetutf-8但页面里混入了GBK编码的片段还有些老站根本不给charset只靠meta标签里的信息来声明。我在基类里写了一句resp.encoding resp.apparent_encoding这个做法可以解决大部分编码问题。apparent_encoding是requests库基于页面内容自动检测出的编码相比resp.encoding直接从响应头取检测准确性更高。但这个方法不是万能的。如果页面内容非常短自动检测就可能猜错此时可以通过人工指定编码来覆盖resp.encoding gb18030。我个人建议在实际解析前先打印一下resp.text的前200个字符确认没有乱码再写解析逻辑。这一步虽然费时间但能省下后面排乱码问题的大把时间。4.2 解析器选错导致的白白折腾BeautifulSoup支持lxml和html.parser两种常见解析器。我几乎总是用lxml因为它在处理表格混乱、标签漏闭合的页面时容错性更好解析速度也快很多。html.parser虽然不需要额外装第三方库但遇到一些不规范页面时会直接放弃治疗。有一次我在模拟项目X上排查数据缺失问题重新跑了一遍解析发现article.book-item这条CSS选择器只能选中一半的条目。后来我把页面HTML保存下来仔细看才发现目标站点在部分页面里把book-item这个class写成了book-item featured。CSS选择器article.book-item仍然可以匹配但如果我当初用的是.book-item反而不会漏。这个教训告诉我解析规则尽量选择特征稳定、层级简单的选择器如果发现数据缺失先用len(soup.select(...))和页面上实际看到的条目数对比判断是选择器问题还是请求被截断。4.3 频率控制与请求头合规使用爬虫写爬虫这个事很多人一上来就关注“怎么绕过限制”但作为网络安全学习的笔记我必须强调爬虫的使用必须限制在授权范围内。对于你自己的测试站点或明确授权抓取的公开数据应该设置合理的请求头表明自己的客户端信息而不是伪造各种身份去试探目标服务器的底线。我在基类里设置了15秒超时和指数退避重试但整个脚本跑起来并没有对请求频率做任何限制。在正式项目中我会在两次请求之间加入一个随机延时比如time.sleep(random.uniform(1, 2))而不是完全不控制。这样虽然采集速度变慢了但会明显减少被目标站点限流的概率。之前有个同事的爬虫脚本因为没有做频率控制对目标测试站发了每秒十几次的请求结果站点WAF直接把这个IP封了半小时。后来加上随机延时之后采集任务顺利完成日志里再也没出现过429状态码。4.4 常见问题速查表现象可能原因解决方案请求返回403请求头缺少或有反爬校验补全User-Agent、Referer等常见头返回内容为空但状态码200页面由JavaScript动态渲染改为寻找接口地址或使用渲染工具数据大量缺失CSS选择器不匹配保存HTML局部文件逐条调试选择器ISBN字段为空页面里非必填字段逻辑上允许该字段为空但校验逻辑要控制重复入库重复运行脚本采集同一页面使用唯一索引INSERT OR IGNORE中文乱码编码识别错误先打印前几百字符再手动指定编码网络连接不稳定目标站点超时或被拒绝增加重试次数指数退避合理限速这张表是我把模拟项目X调试过程中遇到的问题整理出来的覆盖了我自己踩过的大多数坑。再次强调表格里的“反爬校验”指的是目标站点正常的访问控制机制我们不讨论突破访问控制的任何手段合法合规的爬虫一定要遵循站点规则。5. 一些其他可以考虑的扩展方向这个实例只是面向对象爬虫的一个最小可运行版本。如果继续往工程化方向走至少还有三个可以扩展的地方。第一把请求逻辑再抽象一层。比如带Cookie的登录态请求、需要签名参数的接口请求都可以作为独立的请求客户端类然后在爬虫类中组合使用。这样做虽然初期复杂一点但应对多个目标站点时收益非常明显。第二引入数据管道。把“解析后的数据”和“处理后的数据”分成两个阶段中间插入过滤、去重、转换逻辑。面向对象设计里可以做成多个处理器类的组合每个处理器负责一个字段或者一条规则这样数据清洗的规则也能复用和单独测试。第三与定时调度结合起来。爬虫本身不需要一直循环运行可以用系统计划任务或者在程序里嵌入调度器每隔一段时间跑一次爬虫类。由于类的设计已经让“单次采集”非常稳定调度器只需要按时间调用spider.run()然后记录运行结果和异常即可。我在实际使用中体会最深的一件事是面向对象思想不是玄学它解决的是“需求变化时改代码的成本”问题。爬虫领域最不缺的就是变化页面改版、字段调整、接口失效都是家常便饭。写一个灵活的类设计确保自己在三个月后还能看懂这个爬虫在干嘛这本身就是一种很实用的效率投资。最后再分享一个小技巧所有爬虫项目都建议在本地保留一份HTML快照目录每次解析不对的时候直接翻快照排查比重新请求目标站点要方便得多也大大降低了对目标站点的影响。希望这次的实例拆解能给你一些启发下次写爬虫的时候可以先在笔记本上画一个简单的类图再动手。