尧图网站建设 
1. 从一个实际痛点说起为什么我要造这个轮子做这个项目不是一时兴起。大概半年前我经常需要在不同平台上查找电子书资源今天去这个网站搜明天去那个论坛翻收藏夹里存了几十个用得上的页面但每次真正要找书时仍然是一场灾难。不同平台的资源质量参差不齐有的更新快但格式单一有的资源全但搜索体验极差更别提有些站的检索接口响应慢得像在拨号上网。那段时间我手头正好在做一个个人知识库的整理工作需要把散落在各处、格式各异的电子书统一归档。手动一个个下载显然不现实我就想能不能写一个工具把这些平台的基础检索能力聚合起来输入一个书名关键词它在后台同时向多个数据源发起查询再把结果汇总去重、统一格式返回给我这就是多平台异步采集智能检索这套系统的最初原型。如果你和我一样面临的问题是电子书资源分散在多个平台逐个手动搜索效率太低各平台返回的数据格式不统一想要集中管理很麻烦用同步方式写爬虫请求量大时耗时太长等待过程让人抓狂想用异步爬虫但不知道从哪里下手也不想一上来就上重型框架那这篇文章就是为你准备的。我会把整个项目的架构设计、异步采集核心代码、多平台统一解析策略、以及一个轻量但够用的本地全文检索模块全部拆开揉碎讲清楚。所有代码基于Python 3.10只依赖少数几个库,爬虫部分用aiohttp、parsel、asyncio检索部分用SQLite自带的FTS5全文搜索不需要额外部署Elasticsearch这类重组件。先说明一点这是一个用于个人学习研究的技术项目所有数据源的选择都基于公开可访问的网站且我会严格控制采集频率遵守目标网站的robots协议和访问条款。搭建这套系统的目的是学习并发采集和检索架构的设计思路而不是为了搞一堆盗版资源。下面正式进入正题。2. 系统架构设计先想清楚再动手写代码2.1 整体流程拆解这套系统从功能上可以分成三个大模块调度采集层、解析清洗层、存储检索层。调度采集层是整个系统的心脏负责管理多个平台的并发请求。它接收一个搜索关键词把请求任务按平台拆分成多个子任务通过异步并发的方式同时发出请求然后等待所有结果返回。这里的关键不是发了多少请求而是如何控制并发又不把自己和对方的服务器搞挂。解析清洗层做的事情是把各个平台返回的HTML或JSON响应转换成结构统一的记录。不同站的页面结构千差万别有的站直接给JSON接口有的需要解析HTML里的嵌套标签还有的做了一些前端渲染导致真实数据藏在script标签里。这一层需要为每个平台单独写一个适配器最终输出统一的字段书名、作者、格式、大小、链接、来源、更新时间等。存储检索层负责把清洗后的数据落库并提供搜索接口。我选SQLite不是因为它功能有多强而是它足够轻FTS5全文搜索在几万条电子书元数据里做关键词检索响应时间基本在毫秒级完全够用。而且单文件数据库对个人项目的部署和维护都非常友好备份就是复制一份文件。2.2 为什么选异步协程而不是多线程这里我要先讨论一个很多初学者容易纠结的问题既然用requests加线程池也能做并发为什么非要上asyncio核心原因在于IO密集型任务的特点。爬虫的绝大多数时间消耗在网络请求的等待上CPU基本是空闲的。传统多线程方案的问题是每个线程都要占用一定的系统资源线程切换也由操作系统调度线程数量一多GIL带来的限制和上下文切换开销会越来越明显。当然线程池在简单场景下确实够用代码也好写但如果你的目标是跑几十上百个并发任务协程在资源占用方面优势明显——单线程内管理成千上万个轻量任务都不成问题。另一个现实因素是主流爬虫场景中很多反爬手段都是基于检测请求节奏的协程可以非常精确地控制每个任务的发起时间配合信号量做全局限流比多线程的发起-阻塞-切换模式更容易预测行为。下面是架构示意图的文字描述用户输入关键词 ↓ 异步任务调度器asyncio.Queue ↓ 平台适配器A 平台适配器B 平台适配器C requests策略 JSON接口 HTML解析 ↓ 统一结果清洗、去重 ↓ SQLite FTS5 索引 ↓ 本地Web检索接口 / 命令行查询2.3 目录结构与模块职责实际项目我按下面的结构组织代码每个模块职责单一调试时不用大海捞针ebook_searcher/ ├── main.py # 入口命令行交互 ├── config.py # 全局配置并发数、超时时间、User-Agent池 ├── scheduler.py # 异步任务调度器信号量限流 ├── adapters/ │ ├── base.py # 平台适配器抽象基类 │ ├── platform_a.py # 平台A适配器 │ ├── platform_b.py # 平台B适配器 │ └── platform_c.py # 平台C适配器 ├── parser.py # 通用页面解析工具函数 ├── models.py # 数据模型定义 ├── db.py # SQLite初始化与FTS5全文索引 └── search.py # 检索逻辑这样分层的好处是以后想接入新的数据源只需要实现一个base.py抽象类的子类注册进adapters列表其他模块完全不用动。我下面详细讲每个部分的设计和踩坑。3. 异步采集核心逻辑并发控制是爬虫的灵魂3.1 基于asyncio aiohttp的任务调度器实现任务调度器的代码是整个项目的核心我直接给出可复用的版本然后逐行解释关键点。你需要先安装依赖pip install aiohttp parsel下面是调度器的核心实现import asyncio import aiohttp from typing import List, Dict, Any from config import CONCURRENCY_LIMIT, TIMEOUT, USER_AGENT_POOL import random class AsyncScheduler: def __init__(self, concurrency: int CONCURRENCY_LIMIT): self.semaphore asyncio.Semaphore(concurrency) self.timeout aiohttp.ClientTimeout(totalTIMEOUT) self.session None async def __aenter__(self): # 所有并发请求共享同一个session复用TCP连接池 self.session aiohttp.ClientSession( timeoutself.timeout, headers{User-Agent: random.choice(USER_AGENT_POOL)} ) return self async def __aexit__(self, exc_type, exc, tb): if self.session: await self.session.close() async def fetch(self, url: str, params: Dict[str, str] None) - str: 异步抓取单页带信号量限流和重试 async with self.semaphore: for attempt in range(3): try: async with self.session.get(url, paramsparams) as resp: if resp.status 200: return await resp.text() elif resp.status in (429, 503): # 被限流等待后重试指数退避 wait_time 2 ** attempt await asyncio.sleep(wait_time) else: resp.raise_for_status() except (aiohttp.ClientError, asyncio.TimeoutError) as e: if attempt 2: # 第三次失败直接放弃记录日志 print(f[ERROR] 请求失败 {url}: {e}) else: await asyncio.sleep(0.5 * (attempt 1)) return async def gather_tasks(self, tasks: List[Dict[str, Any]]) - List[str]: 并发执行所有采集任务 results [] for task in tasks: results.append(await self.fetch(task[url], task.get(params))) return results代码里的几个地方值得特别注意。第一是Semaphore信号量。这是并发限流的核心手段。CONCURRENCY_LIMIT我通常设置在5到10之间。设置这个值的时候需要同时考虑两个因素自己的机器性能和目标网站的承受能力。太激进会被对方封IP太保守又体现不出异步的优势。我实际测试过并发5的时候在保证基本不被封的前提下采集速度大约是同步请求的4到5倍这个性价比已经很高了。第二是aiohttp.ClientSession的复用。很多初学者容易犯的错误是每个请求都新建一个Session那样连接池根本起不到作用频繁的TCP握手反而比同步还慢。正确做法是全局共享一个Session连接复用性能提升非常明显。第三是重试策略。直接用一个简单的循环重试3次遇到限流状态码429、503时采用指数退避方式等待。这里没有用装饰器或者第三方重试库是因为爬虫场景下需要精确控制何时重试、重试多久手动写反而更灵活。3.2 并发版本asyncio.gather与任务分组上面的gather_tasks其实还是串行的——虽然用的是async语法但一个接一个地await并没有真正并发。不过别急我先把真正的并发版本放出来。在实际项目中我采用的是分组并发策略。把所有平台的搜索请求包装成一个个协程任务然后用asyncio.gather一次并发执行async def run_search(self, keyword: str) - List[Dict[str, Any]]: 并发执行所有平台的搜索 tasks [] for adapter in self.adapters: task asyncio.create_task(adapter.search(self.session, keyword)) tasks.append(task) results await asyncio.gather(*tasks, return_exceptionsTrue) # 处理结果标记失败的平台 merged [] for adapter, result in zip(self.adapters, results): if isinstance(result, Exception): print(f[WARN] 平台 {adapter.name} 搜索失败: {result}) continue merged.extend(result) return mergedreturn_exceptionsTrue这个参数很重要如果某个平台宕机或IP被封导致协程抛异常不加这个参数的话整个gather都会崩溃加上之后单个平台的失败不会影响其他平台的正常返回。分组并发策略是这么考虑的把每个平台视为一个独立的分组同一平台内部的多个分页请求共享一个信号量限制避免对单个网站造成突发压力不同平台之间互补干扰。实际操作时我还会对同一平台内部的请求做一次小间隔比如0.5秒到1秒的随机延迟进一步降低被反爬系统标记的风险。3.3 请求头伪装与频率控制安全是第一原则说到被反爬这是每个爬虫开发者的必修课。我并不支持突破某些网站的高强度反爬体系但基础的请求头伪装是任何一个正常爬虫项目都绕不开的功课。我的config.py里维护了一个User-Agent池里面放了十几个常见浏览器的UA字符串Chrome、Firefox、Safari、Edge的不同版本。每次请求时随机取一个这个操作很简单却能躲掉很大一部分基于UA的简易反爬。另外还需要补上常见的请求头HEADERS { Accept: text/html,application/xhtmlxml,application/xml;q0.9,image/webp,*/*;q0.8, Accept-Language: zh-CN,zh;q0.9,en;q0.8, Accept-Encoding: gzip, deflate, Connection: keep-alive, Upgrade-Insecure-Requests: 1, }这里有一个很多教程不会提到的细节Accept-Encoding如果设置为gzip, deflate有些服务器会返回压缩后的内容aiohttp会自动解压但如果你用的是requests库且没有正确设置就可能拿到乱码。在aiohttp中响应文本通过resp.text()读取时已经处理过编码和压缩问题这个问题基本不会遇到但理解这个机制还是有必要的。频率控制方面除了信号量限制并发数我还在每个请求之间加入了随机延迟await asyncio.sleep(random.uniform(0.3, 0.8))这个延迟看起来不起眼却能让请求节奏更接近真人浏览行为。持续高频率、恒定间隔的请求是反爬系统最关注的模式之一随机化之后风险明显降低。同时我会在代码里硬性要求每个平台每秒钟最多只能发起2个请求这是任何负责任爬虫项目的底线。3.4 为什么没用Scrapy?聊聊异步方案选型的取舍很多人看到这个标题可能会问为什么不用ScrapyScrapy也支持异步啊还有现成的中间件、管道、内置的去重机制。我的回答是看场景。Scrapy确实很强大适合大型分布式爬虫项目。但本项目需要一个高度可定制的搜索接口调用系统而不是传统意义上的爬虫——每次运行时用户输入关键词系统即时向多个平台发起查询请求拿到结果直接返回。Scrapy是爬取-解析-入库的批处理模式它的异步框架Twisted和项目本身的工作流不太匹配。使用aiohttpasyncio的好处是我可以像写普通函数一样组织代码请求、解析、结果合并一气呵成调试非常直观。比如在IDE里设一个断点就能看到所有协程的执行状态这种透明度在开发调试阶段极其重要。当然如果有一天这个系统需要扩展到爬取全站资源、需要断点续爬、需要分布式调度我会毫不犹豫迁移到Scrapy。工具没有绝对的好坏只有适不适合当前场景的问题。4. 多平台接入实战适配器模式的解析层设计4.1 不同网站的数据结构差异从HTML到JSON的适配思路你可能会想直接发请求、解析、返回数据不就行了为什么要专门搞一个适配器层当你真正接触多个不同平台的页面结构后就会发现内容结构不统一是集成过程中最大的坑。拿我接入的几个数据源举例平台A的搜索结果直接就是结构良好的JSON接口请求某个带参数URL就能拿到干净的数据字段命名也很规范。这种最省心解析代码不超过10行。平台B的搜索结果虽然也是HTML但书籍条目都嵌在带有特定class属性的div中标题在h3a标签里下载链接在后面的按钮里。用parsel配合XPath就能准确提取。平台C做得最绝它的书籍列表数据根本不直接在HTML里而是作为一段JSON字符串嵌在script typeapplication/ldjson标签中。直接把HTML当XML解析会累死人正确做法是先用正则或XPath把这个JSON片段抠出来再交给json.loads处理。所以适配器层的核心抽象是每个平台一个类对外暴露统一接口方法async def search(session, keyword) - List[BookItem]。无论底层是JSON还是HTML最终返回的都是结构统一的BookItem对象。4.2 适配器基类与统一数据模型定义先定义统一数据模型from dataclasses import dataclass from datetime import datetime dataclass class BookItem: title: str # 书名 author: str # 作者 format: str # 格式如 PDF/EPUB/MOBI size: str # 文件大小如 12.5MB source: str # 来源平台名称 url: str # 详情页或下载页链接 updated_at: str # 更新时间然后定义适配器基类from abc import ABC, abstractmethod from typing import List, Dict, Any import aiohttp class BaseAdapter(ABC): 所有平台适配器的基类 name: str base_url: str # 每个平台专属的请求间隔避免同名适配器请求过快 request_interval: float 0.8 def __init__(self): self.last_request_time 0 async def _rate_limited_sleep(self): 平台级限速每个平台独立控制节奏 import asyncio, time now time.monotonic() elapsed now - self.last_request_time if elapsed self.request_interval: await asyncio.sleep(self.request_interval - elapsed) self.last_request_time time.monotonic() abstractmethod async def search(self, session: aiohttp.ClientSession, keyword: str) - List[BookItem]: 每个平台必须实现自己的搜索逻辑 pass_rate_limited_sleep这个小函数设计得很巧妙。它在每个适配器内部维护一个上次请求时间戳如果两次请求间的时间间隔小于设定的阈值就sleep补足。这样即使共享同一个网络连接池每个平台也能独立控制自己的请求节奏。4.3 实战JSON接口型数据源的解析平台A是最简单的JSON类型它的搜索接口大概长这样class PlatformAAdapter(BaseAdapter): name 平台A base_url https://api.example-a.com async def search(self, session, keyword): await self._rate_limited_sleep() url f{self.base_url}/search params { q: keyword, page: 1, size: 20, } async with session.get(url, paramsparams) as resp: if resp.status ! 200: return [] data await resp.json() books [] for item in data.get(data, {}).get(list, []) or []: book BookItem( titleitem.get(title, ), authoritem.get(author, ), formatitem.get(file_type, ), sizeitem.get(file_size, ), sourceself.name, urlitem.get(detail_url, ), updated_atitem.get(update_time, ) ) books.append(book) return books这种适配器写起来非常轻松但有一个容易被忽视的坑JSON字段的嵌套层级和命名风格每个平台都不一样有的用data.list有的用result.records有的字段名是bookName而不是title。所以每个适配器内都要单独写映射关系这是无法避免的工作量。4.4 真实网页中JSON-LD数据的提取细节第三个平台的处理要复杂不少。我看过大量网页源码后发现很多信息类网站会把结构化数据以JSON-LD格式嵌入页面头部方便搜索引擎理解和展示。这实际上给爬虫开发者提供了一个后门——数据就摆在script typeapplication/ldjson标签里等着取。用parsel提取这类数据非常方便import json from parsel import Selector class PlatformCAdapter(BaseAdapter): name 平台C base_url https://www.example-c.com async def search(self, session, keyword): await self._rate_limited_sleep() url f{self.base_url}/search params {keyword: keyword} async with session.get(url, paramsparams) as resp: content await resp.text() sel Selector(textcontent) scripts sel.xpath(//script[typeapplication/ldjson]/text()).getall() books [] for script in scripts: try: data json.loads(script) except json.JSONDecodeError: continue # JSON-LD数据可能是单对象也可能是对象数组 items data if isinstance(data, list) else [data] for item in items: if item.get(type) ! Book: continue books.append(BookItem( titleitem.get(name, ), author, .join(item.get(author, [])), formatitem.get(bookFormat, Unknown), sizeitem.get(fileSize, ), sourceself.name, urlitem.get(url, ), updated_atitem.get(datePublished, ) )) return books这里有一个细节要特别强调JSON-LD中author字段经常是一个对象数组每个对象里有type和name字段如果直接把它当字符串用就会得到一长串Python字典的字符串表示。必须先把author字段取出来再做类型判断和拼装。这种你以为取到的是字符串其实是列表的情况在真实爬虫开发中比比皆是。还有一个相关细节JSON-LD的bookFormat字段每个平台定义都不一样有的写Paperback纸质书有的写EBook电子书我在入库前会做一轮手工映射把不统一的格式值翻译成自己的规范值。4.5 CSS选择器与XPath的选择经验在写HTML型平台解析代码时选择器用XPath还是CSS我个人的经验是结构性较强的页面用XPathclass命名规范且富有语义化的页面用CSS更简洁。实际上parsel同时支持两种语法切换成本为零。遇到一个需要注意的场景某些网站为了前端渲染方便会把class名做混淆处理比如c1x8k3、_2Vd4F这种这种情况下用CSS老老实实匹配就很容易踩坑。正确做法是先用浏览器开发者工具的Copy as XPath功能拿到参考路径再人工改成更稳健的相对路径写法。我频繁使用的XPath模式# 提取所有书籍条目的容器 //div[contains(class, book-list)]/div[contains(class, item)] # 提取相对路径下的标题 .//h3[contains(class, title)]/a/text() # 提取链接 .//h3[contains(class, title)]/a/href # 提取作者和大小等元数据 .//span[contains(class, author)]/text()注意XPath只写了相对路径开头是.这个点代表从当前上下文节点开始查找在for循环遍历每个书籍条目时必须以当前条目为上下文否则会匹配到页面里其他无关元素。5. 智能检索模块不止是字符串匹配5.1 为什么选择SQLite FTS5做全文索引数据采集回来后接下来要解决的是检索问题。很多人第一反应是用SELECT * FROM books WHERE title LIKE %keyword%这套方案在数据量小的时候确实没问题但一旦数据量上万LIKE查询无法利用索引全表扫描的耗时就会变得不可接受。SQLite从3.9版本开始引入了FTS5全文搜索扩展它可以为文本数据建立倒排索引查询速度提升一两个数量级。更重要的是FTS5支持中文分词吗严格说它自带的分词器对中文支持并不好默认的unicode61分词器是按空格和标点切词的英文效果不错但中文一整句话都会被当成一个词。这会导致什么结果搜索三体时如果标题里存的是三体全集或三体黑暗森林FTS5默认分词会把整个三体全集当成一个token三体这个token在倒排索引里根本不存在搜索不出来。解决办法有两个方向方向一在应用层做中文分词把分词后的词组用空格拼接存在额外的列里搜索时也对query做同样处理。方向二使用SQLite官方提供的simple分词器配合tokenizeunicode61 remove_diacritics 0效果有限。我实际采用的是方向一的简化版在写入索引前用jieba对标题和作者字段做分词把分词结果用空格连接后存在FTS5虚拟表里以下几种情况就不会出现白屏问题了。5.2 FTS5虚拟表设计与关键词搜索实现建表语句CREATE VIRTUAL TABLE IF NOT EXISTS books_fts USING fts5( title, authorname, content, tokenize unicode61 ); -- 每次插入新书数据时同步写入分词后的内容 INSERT INTO books_fts (title, authorname, content) VALUES (?, ?, ?);用jieba分词的代码片段import jieba def tokenize_chinese(text: str) - str: 把中文文本分词用空格连接 words jieba.cut(text) return .join([w.strip() for w in words if w.strip()])实际插入数据时title_tokens tokenize_chinese(book.title) author_tokens tokenize_chinese(book.author) conn.execute( INSERT INTO books_fts (title, authorname, content) VALUES (?, ?, ?), (title_tokens, author_tokens, book.title book.author) )这里存了两份数据分词后的token串用于索引匹配原始文本用于展示。搜索查询时def search_books(query: str): FTS5全文检索用BM25算法做相关性排序 query_tokens tokenize_chinese(query) sql SELECT b.*, bm25(books_fts) AS score FROM books_fts JOIN books b ON books_fts.rowid b.id WHERE books_fts MATCH ? ORDER BY score LIMIT 50 return conn.execute(sql, (query_tokens,)).fetchall()bm25()是FTS5内置的相关性评分函数BM25算法是搜索引擎领域最经典的排序算法之一返回的score越小表示相关性越高负值趋近0最佳直接按它排序就能让最匹配的结果排在最前。你可能会问建表时只建了FTS5虚拟表那books表在哪实际的books表存的是去重后的完整元数据FTS5表通过rowid和books.id建立映射关系。两个表通过JOIN查询。这样设计的好处是FTS5专注做全文索引books表专注存结构化的完整信息互不干扰数据更新时分别操作两张表即可。5.3 智能排序策略把最可能想要的放在最前面有了相关性评分还不够实际使用中我发现只靠BM25排序会有一个问题同一个关键词搜索出来几百条结果有些是新书有些是几年前的有些资源格式齐全有些只有一个格式。用户真正想找的往往是资源最好、格式最新、大小合理的那个。于是我在搜索SQL里加入了排序权重ORDER BY score ASC, CASE WHEN b.format IN (EPUB, PDF) THEN 0 ELSE 1 END, b.updated_at DESC解释一下score ASC先按相关性排序BM25得分越小越相关。CASE WHEN format IN (EPUB, PDF)EPUB和PDF是最通用、兼容性最好的电子书格式优先展示这两种格式的结果。占第二位优先级的排序条件。b.updated_at DESC最后按更新时间倒序让最近更新的资源排前面。用户输入搜索词后实际返回的结果质量比单纯用关键词匹配好很多。我强烈建议你在自己的系统里根据实际数据情况调整这个排序规则比如如果你的目标用户更看重MOBI格式就把MOBI放在优先列表里。排序逻辑是整个检索体验中最容易被忽视的环节但也是性价比最高的优化之一。5.4 去重逻辑多平台结果合并的隐藏陷阱多平台采集有个绕不开的问题同一本书在多个网站都有收录如果直接合并用户会看到好几条完全相同的记录体验很差。去重逻辑不能只看书名是否完全一致因为不同平台的标题格式可能略有差别比如三体全集和三体全集或者有的站标题里带了作者名。我实现的去重策略是归一化后比较。方法很朴素但实用import re def normalize_title(title: str) - str: 归一化标题消除格式差异 title re.sub(r[\(\[].*?[\)\]], , title) # 去掉括号内容 title re.sub(r[ -_·], , title) # 去掉常见连接符 title title.lower().strip() # 统一小写 return title在实际处理时维护一个normalized_title - BookItem的字典如果多条记录的归一化标题相同就比较它们的格式完整度、更新时间只保留最优的一条。这里注意一个边界情况有些书的标题本身就带着括号比如三体全集珍藏版如果直接去掉括号再归一化会丢掉全集这个重要信息。所以我的方案是只去掉括号但用括号内容做一次加权比较比如带全集“精排”这些关键词的记录在去重时优先保留。数据清洗的细节实在太多了很多时候你只有在实际处理大量真实数据后才会意识到这些问题。6. 真实运行中的踩坑与调试经验6.1 编码问题你以为的UTF-8不是真正的UTF-8做爬虫的人几乎都遇到过编码问题。中文网站最常见的两个编码是UTF-8和GBK/GB2312如果搞错了轻则个别字显示异常重则整段内容乱码。aiohttp的resp.text()方法默认使用HTTP响应头里的Content-Type字段指定的编码来解码。但有些网站的响应头没有正确声明编码甚至故意不声明或者声明了错误的编码这时候就需要手动处理。# 获取二进制内容手动解码 content await resp.read() # 优先使用响应头中的编码猜测否则用 chardet 自动检测 try: charset resp.charset or utf-8 text content.decode(charset, errorsignore) except LookupError: import chardet detected chardet.detect(content) text content.decode(detected[encoding] or utf-8, errorsignore)我在实际项目中遇到过这样的情况一个平台返回的HTML头部里明确写着charsetgb2312实际内容中却夹杂着UTF-8编码的字符。用errorsignore参数可以在解码时跳过无法识别的字节不会因为个别异常字符导致整个页面解析失败。6.2 IP被封的常见信号与应对策略这个环节我很想多聊几句因为如何检测自己是不是被封了这个问题比被封了怎么办更值得关心。判断IP是否被封不能只看请求成功后解析出的内容是否为空。以下几个信号值得关注页面返回200但内容明显变短或者跳转到一个验证码页面所有请求突然都返回403或418状态码响应速度突然变得极慢且包含很多重定向返回的HTML里出现请输入验证码“安全检查”等关键词我在代码里加了一个简单的检查函数def is_blocked(content: str) - bool: 简单判断页面是否包含封锁特征 block_keywords [验证码, captcha, access denied, 安全验证] return any(keyword in content.lower() for keyword in block_keywords)每次解析前先跑一遍检查如果确认被封立即停止该平台的所有请求并记录日志到文件方便事后分析。等到本地IP解封后再继续运行。应对策略方面除了第3节提到的请求频率控制和UA池还可以加长两次请求之间的间隔、随机化延迟时间、为关键请求增加重试退避策略。我个人不建议一开始就上代理池因为高质量代理成本高、低质量代理反而更容易触发封禁。把单IP的请求频率控制到合理范围以内是性价比最高的方案。6.3 断点续跑异常中断后的数据恢复长时间运行中网络波动、内存问题、电脑休眠都可能导致程序中断。如果每次中断都要从头开始采集那体验实在太糟糕了。我实现的方案是每条记录插入数据库时带一个source_task_id字段表示它属于哪一次采集任务。程序启动时先查询最后一次任务ID和该任务中已成功采集的平台列表只重新采集失败的部分。def get_last_task_status(conn) - dict: 查询最近一次采集任务完成状态 row conn.execute( SELECT task_id, GROUP_CONCAT(DISTINCT source) as sources FROM books WHERE task_id (SELECT MAX(task_id) FROM books) GROUP BY task_id ).fetchone() if not row: return {} completed_sources set(row[sources].split(,)) return {task_id: row[task_id], completed_sources: completed_sources}启动时把这个状态读出来completed_sources里已经成功的平台就不重复采集了。这个优化听起来简单实际使用中能帮你省掉大量重新采集的时间。6.4 asyncio调试技巧如何优雅地排查死锁和超时异步编程的一个大坑是程序看起来卡死了但没报任何错误。最常见的原因是某个协程一直不返回整个事件循环被占住了。排查方法第一设置统一的超时时间避免某个请求无限制等待。我在调度器里已经用aiohttp.ClientTimeout(totalTIMEOUT)做了处理。第二给长时间运行的任务设置总超时限制try: async with asyncio.timeout(30): results await asyncio.gather(*tasks, return_exceptionsTrue) except asyncio.TimeoutError: print(任务总超时强制返回)第三利用asyncio.timeout上下文管理器Python 3.11或asyncio.wait_for旧版本在关键路径上设置兜底超时。实际运行中单个平台的崩溃往往会让整个gather等待很久才返回设置了总超时后最坏情况半分钟就能恢复响应。调试时还可以手动启动一个简单的事件循环测试单个适配器import asyncio async def debug_adapter(): adapter PlatformAAdapter() async with AsyncScheduler() as scheduler: books await adapter.search(scheduler.session, 三体) for book in books: print(f《{book.title}》 {book.author} | {book.format} | {book.url}) asyncio.run(debug_adapter())单独调试一个适配器排除了并发干扰后能更快定位问题。这个习惯帮我节省了大量排错时间。7. 项目效果实测与性能对比7.1 实测数据异步采集到底比同步快多少为了让大家对这套方案的性能提升有直观认识我专门做了一组对比实验。测试环境普通家用带宽100Mbps目标平台3个每个平台搜索同一关键词并采集前20条结果。同步方案用requests库逐一请求异步方案用我上面的代码并发数设置8。采集方式总耗时成功请求数失败/超时requests 同步循环12.6秒600asyncio并发限流34.8秒600asyncio并发限流82.9秒600asyncio并发限流152.2秒601可以看到并发数从3提升到8性能提升非常明显从8提升到15收益急剧缩小还开始出现超时失败。这说明并发量不是越大越好8这个数量级在处理这种中小规模的采集任务时已经是甜点区。7.2 资源覆盖情况与检索质量测试我把系统跑了几天累计采集了几千条电子书元数据覆盖了文学、社科、计算机、经济管理等主要分类。用几个不同类型的查询词做测试搜索三体返回12条结果前三名是不同格式的《三体》全集排序合理。搜索Python编程返回86条结果包含入门书籍、进阶书籍、实战项目筛选精度不错。搜索一个比较冷门的专业名词量子机器学习也能返回4条结果说明覆盖度还行。故意输入一个不存在的书名不存在的某本书xyz返回0条结果没有出现莫名其妙的误匹配说明FTS5的匹配逻辑比较可靠。从使用体验来看最明显的改进是搜索响应时间。之前用LIKE模糊查询几万条数据全表扫描大概需要1-2秒。换成FTS5之后响应时间基本在50毫秒以内快了两个数量级。7.3 系统资源占用情况运行期间监控了内存和CPU占用。内存常驻内存在180MB左右主要是因为Python解释器本身和aiohttp的连接池占用。单进程并发8时峰值内存不超过300MB。CPU由于是IO密集型任务CPU占用极低平均在5%左右。只有在解析大量HTML时才看到CPU有短暂飙升。磁盘数据库文件在采集5000条记录后大小约为8MB包含FTS5索引非常轻量。这个资源占用水平完全可以跑在低配的云服务器或树莓派上做个人知识库的配套服务很合适。8. 几个值得留意的工程细节8.1 日志记录多平台状态一目了然开发过程中我意识到信息类爬虫项目的日志不能只写print。因为多个平台并发执行时打印输出的顺序是混乱的稍纵即逝的报错信息很快就被后续日志吞没。我的做法是使用Python标准库的logging模块每一个适配器配置一个带名称的logger日志输出格式包含时间、平台名、日志级别import logging logging.basicConfig( levellogging.INFO, format%(asctime)s [%(levelname)s] [%(name)s] %(message)s, handlers[ logging.FileHandler(searcher.log, encodingutf-8), logging.StreamHandler() ] ) logger logging.getLogger(PlatformA) logger.info(开始搜索关键词: %s, keyword) logger.warning(平台响应超时第%d次重试, attempt 1) logger.error(该平台已被封禁: HTTP %d, resp.status)加上日志的好处不用多说出现问题时回溯现场非常方便。我建议从项目一开始就规范日志格式不要等到出问题了再补。8.2 配置文件独立改参数不用翻代码我在config.py里集中定义了所有可调参数# 并发请求数量建议5-10 CONCURRENCY_LIMIT 8 # 单个请求超时时间秒 TIMEOUT 10 # User-Agent池 USER_AGENT_POOL [ Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36, Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.0 Safari/605.1.15, # 省略更多 ] # 每个平台请求间隔秒默认0.8 PLATFORM_REQUEST_INTERVAL 0.8调整并发量、超时时间、UA池都不需要动核心代码。对于长期维护的项目这种配置与代码分离的设计虽然简单却能让维护体验好很多。8.3 数据库定期维护FTS索引会变大FTS5索引会随着数据量的增加逐渐膨胀而且频繁的插入和删除操作会产生内部碎片。建议定期执行一次优化def optimize_db(conn): 优化FTS5索引重建表结构 conn.execute(INSERT INTO books_fts(books_fts) VALUES(optimize)) conn.commit()这个optimize命令会重组索引压缩体积。我在采集大量数据后手动跑一次效果明显。另外SQLite本身支持VACUUM命令回收未使用的空间但我实际体验是FTS5的专用优化命令效果更精准对数据库的锁定时间也更短。定期备份数据库文件这个问题在个人项目中不会造成严重影响但提前处理总比事后补救好。9. 最后再分享一个实用技巧如何处理搜索结果的动态加载在接入了几个平台之后我注意到有些网站的数据不是服务端直出的而是前端通过JavaScript异步请求JSON接口再渲染。这就导致直接请求搜索页面HTML拿不到任何数据。我总结了两条可行的路线都不是通过无头浏览器这类重武器去硬碰硬而是用更轻量、更稳定的方案第一条路线直接找XHR接口。打开浏览器开发者工具的Network面板搜索关键词筛选XHR类型的请求你通常能看到一个以.json或.do、.api等结尾的接口地址。它往往带一些加密参数或动态token但这些参数通常就藏在页面源码的某个script标签里或者在前面的JS文件中生成。仔细挖掘一下就能找到规律。第二条路线有些平台提供了CORS开放的公共API只是没有在页面UI上做明显入口。这类接口返回的数据比HTML更干净、更稳定。在动手写适配器之前先花几分钟翻看一下目标网站的网络请求面板往往能直接找到比页面解析简单得多的方案。但我要提醒一句如果一个平台的前端JS代码做了非常复杂的参数加密且把所有内容都放在异步加载中这种高强度的反爬设计往往意味着网站的拥有者不愿意被第三方程序访问。对于这类情况我的原则是直接放弃接入。爬虫开发的世界里不是所有接口都该去碰的懂得哪些不该碰有时比懂得怎么碰更重要——这也是我一路踩坑后最深的体会。 基于aiohttp与FTS5的多平台电子书搜索系统：异步爬虫架构实战
1. 从一个实际痛点说起为什么我要造这个轮子做这个项目不是一时兴起。大概半年前我经常需要在不同平台上查找电子书资源今天去这个网站搜明天去那个论坛翻收藏夹里存了几十个用得上的页面但每次真正要找书时仍然是一场灾难。不同平台的资源质量参差不齐有的更新快但格式单一有的资源全但搜索体验极差更别提有些站的检索接口响应慢得像在拨号上网。那段时间我手头正好在做一个个人知识库的整理工作需要把散落在各处、格式各异的电子书统一归档。手动一个个下载显然不现实我就想能不能写一个工具把这些平台的基础检索能力聚合起来输入一个书名关键词它在后台同时向多个数据源发起查询再把结果汇总去重、统一格式返回给我这就是多平台异步采集智能检索这套系统的最初原型。如果你和我一样面临的问题是电子书资源分散在多个平台逐个手动搜索效率太低各平台返回的数据格式不统一想要集中管理很麻烦用同步方式写爬虫请求量大时耗时太长等待过程让人抓狂想用异步爬虫但不知道从哪里下手也不想一上来就上重型框架那这篇文章就是为你准备的。我会把整个项目的架构设计、异步采集核心代码、多平台统一解析策略、以及一个轻量但够用的本地全文检索模块全部拆开揉碎讲清楚。所有代码基于Python 3.10只依赖少数几个库,爬虫部分用aiohttp、parsel、asyncio检索部分用SQLite自带的FTS5全文搜索不需要额外部署Elasticsearch这类重组件。先说明一点这是一个用于个人学习研究的技术项目所有数据源的选择都基于公开可访问的网站且我会严格控制采集频率遵守目标网站的robots协议和访问条款。搭建这套系统的目的是学习并发采集和检索架构的设计思路而不是为了搞一堆盗版资源。下面正式进入正题。2. 系统架构设计先想清楚再动手写代码2.1 整体流程拆解这套系统从功能上可以分成三个大模块调度采集层、解析清洗层、存储检索层。调度采集层是整个系统的心脏负责管理多个平台的并发请求。它接收一个搜索关键词把请求任务按平台拆分成多个子任务通过异步并发的方式同时发出请求然后等待所有结果返回。这里的关键不是发了多少请求而是如何控制并发又不把自己和对方的服务器搞挂。解析清洗层做的事情是把各个平台返回的HTML或JSON响应转换成结构统一的记录。不同站的页面结构千差万别有的站直接给JSON接口有的需要解析HTML里的嵌套标签还有的做了一些前端渲染导致真实数据藏在script标签里。这一层需要为每个平台单独写一个适配器最终输出统一的字段书名、作者、格式、大小、链接、来源、更新时间等。存储检索层负责把清洗后的数据落库并提供搜索接口。我选SQLite不是因为它功能有多强而是它足够轻FTS5全文搜索在几万条电子书元数据里做关键词检索响应时间基本在毫秒级完全够用。而且单文件数据库对个人项目的部署和维护都非常友好备份就是复制一份文件。2.2 为什么选异步协程而不是多线程这里我要先讨论一个很多初学者容易纠结的问题既然用requests加线程池也能做并发为什么非要上asyncio核心原因在于IO密集型任务的特点。爬虫的绝大多数时间消耗在网络请求的等待上CPU基本是空闲的。传统多线程方案的问题是每个线程都要占用一定的系统资源线程切换也由操作系统调度线程数量一多GIL带来的限制和上下文切换开销会越来越明显。当然线程池在简单场景下确实够用代码也好写但如果你的目标是跑几十上百个并发任务协程在资源占用方面优势明显——单线程内管理成千上万个轻量任务都不成问题。另一个现实因素是主流爬虫场景中很多反爬手段都是基于检测请求节奏的协程可以非常精确地控制每个任务的发起时间配合信号量做全局限流比多线程的发起-阻塞-切换模式更容易预测行为。下面是架构示意图的文字描述用户输入关键词 ↓ 异步任务调度器asyncio.Queue ↓ 平台适配器A 平台适配器B 平台适配器C requests策略 JSON接口 HTML解析 ↓ 统一结果清洗、去重 ↓ SQLite FTS5 索引 ↓ 本地Web检索接口 / 命令行查询2.3 目录结构与模块职责实际项目我按下面的结构组织代码每个模块职责单一调试时不用大海捞针ebook_searcher/ ├── main.py # 入口命令行交互 ├── config.py # 全局配置并发数、超时时间、User-Agent池 ├── scheduler.py # 异步任务调度器信号量限流 ├── adapters/ │ ├── base.py # 平台适配器抽象基类 │ ├── platform_a.py # 平台A适配器 │ ├── platform_b.py # 平台B适配器 │ └── platform_c.py # 平台C适配器 ├── parser.py # 通用页面解析工具函数 ├── models.py # 数据模型定义 ├── db.py # SQLite初始化与FTS5全文索引 └── search.py # 检索逻辑这样分层的好处是以后想接入新的数据源只需要实现一个base.py抽象类的子类注册进adapters列表其他模块完全不用动。我下面详细讲每个部分的设计和踩坑。3. 异步采集核心逻辑并发控制是爬虫的灵魂3.1 基于asyncio aiohttp的任务调度器实现任务调度器的代码是整个项目的核心我直接给出可复用的版本然后逐行解释关键点。你需要先安装依赖pip install aiohttp parsel下面是调度器的核心实现import asyncio import aiohttp from typing import List, Dict, Any from config import CONCURRENCY_LIMIT, TIMEOUT, USER_AGENT_POOL import random class AsyncScheduler: def __init__(self, concurrency: int CONCURRENCY_LIMIT): self.semaphore asyncio.Semaphore(concurrency) self.timeout aiohttp.ClientTimeout(totalTIMEOUT) self.session None async def __aenter__(self): # 所有并发请求共享同一个session复用TCP连接池 self.session aiohttp.ClientSession( timeoutself.timeout, headers{User-Agent: random.choice(USER_AGENT_POOL)} ) return self async def __aexit__(self, exc_type, exc, tb): if self.session: await self.session.close() async def fetch(self, url: str, params: Dict[str, str] None) - str: 异步抓取单页带信号量限流和重试 async with self.semaphore: for attempt in range(3): try: async with self.session.get(url, paramsparams) as resp: if resp.status 200: return await resp.text() elif resp.status in (429, 503): # 被限流等待后重试指数退避 wait_time 2 ** attempt await asyncio.sleep(wait_time) else: resp.raise_for_status() except (aiohttp.ClientError, asyncio.TimeoutError) as e: if attempt 2: # 第三次失败直接放弃记录日志 print(f[ERROR] 请求失败 {url}: {e}) else: await asyncio.sleep(0.5 * (attempt 1)) return async def gather_tasks(self, tasks: List[Dict[str, Any]]) - List[str]: 并发执行所有采集任务 results [] for task in tasks: results.append(await self.fetch(task[url], task.get(params))) return results代码里的几个地方值得特别注意。第一是Semaphore信号量。这是并发限流的核心手段。CONCURRENCY_LIMIT我通常设置在5到10之间。设置这个值的时候需要同时考虑两个因素自己的机器性能和目标网站的承受能力。太激进会被对方封IP太保守又体现不出异步的优势。我实际测试过并发5的时候在保证基本不被封的前提下采集速度大约是同步请求的4到5倍这个性价比已经很高了。第二是aiohttp.ClientSession的复用。很多初学者容易犯的错误是每个请求都新建一个Session那样连接池根本起不到作用频繁的TCP握手反而比同步还慢。正确做法是全局共享一个Session连接复用性能提升非常明显。第三是重试策略。直接用一个简单的循环重试3次遇到限流状态码429、503时采用指数退避方式等待。这里没有用装饰器或者第三方重试库是因为爬虫场景下需要精确控制何时重试、重试多久手动写反而更灵活。3.2 并发版本asyncio.gather与任务分组上面的gather_tasks其实还是串行的——虽然用的是async语法但一个接一个地await并没有真正并发。不过别急我先把真正的并发版本放出来。在实际项目中我采用的是分组并发策略。把所有平台的搜索请求包装成一个个协程任务然后用asyncio.gather一次并发执行async def run_search(self, keyword: str) - List[Dict[str, Any]]: 并发执行所有平台的搜索 tasks [] for adapter in self.adapters: task asyncio.create_task(adapter.search(self.session, keyword)) tasks.append(task) results await asyncio.gather(*tasks, return_exceptionsTrue) # 处理结果标记失败的平台 merged [] for adapter, result in zip(self.adapters, results): if isinstance(result, Exception): print(f[WARN] 平台 {adapter.name} 搜索失败: {result}) continue merged.extend(result) return mergedreturn_exceptionsTrue这个参数很重要如果某个平台宕机或IP被封导致协程抛异常不加这个参数的话整个gather都会崩溃加上之后单个平台的失败不会影响其他平台的正常返回。分组并发策略是这么考虑的把每个平台视为一个独立的分组同一平台内部的多个分页请求共享一个信号量限制避免对单个网站造成突发压力不同平台之间互补干扰。实际操作时我还会对同一平台内部的请求做一次小间隔比如0.5秒到1秒的随机延迟进一步降低被反爬系统标记的风险。3.3 请求头伪装与频率控制安全是第一原则说到被反爬这是每个爬虫开发者的必修课。我并不支持突破某些网站的高强度反爬体系但基础的请求头伪装是任何一个正常爬虫项目都绕不开的功课。我的config.py里维护了一个User-Agent池里面放了十几个常见浏览器的UA字符串Chrome、Firefox、Safari、Edge的不同版本。每次请求时随机取一个这个操作很简单却能躲掉很大一部分基于UA的简易反爬。另外还需要补上常见的请求头HEADERS { Accept: text/html,application/xhtmlxml,application/xml;q0.9,image/webp,*/*;q0.8, Accept-Language: zh-CN,zh;q0.9,en;q0.8, Accept-Encoding: gzip, deflate, Connection: keep-alive, Upgrade-Insecure-Requests: 1, }这里有一个很多教程不会提到的细节Accept-Encoding如果设置为gzip, deflate有些服务器会返回压缩后的内容aiohttp会自动解压但如果你用的是requests库且没有正确设置就可能拿到乱码。在aiohttp中响应文本通过resp.text()读取时已经处理过编码和压缩问题这个问题基本不会遇到但理解这个机制还是有必要的。频率控制方面除了信号量限制并发数我还在每个请求之间加入了随机延迟await asyncio.sleep(random.uniform(0.3, 0.8))这个延迟看起来不起眼却能让请求节奏更接近真人浏览行为。持续高频率、恒定间隔的请求是反爬系统最关注的模式之一随机化之后风险明显降低。同时我会在代码里硬性要求每个平台每秒钟最多只能发起2个请求这是任何负责任爬虫项目的底线。3.4 为什么没用Scrapy?聊聊异步方案选型的取舍很多人看到这个标题可能会问为什么不用ScrapyScrapy也支持异步啊还有现成的中间件、管道、内置的去重机制。我的回答是看场景。Scrapy确实很强大适合大型分布式爬虫项目。但本项目需要一个高度可定制的搜索接口调用系统而不是传统意义上的爬虫——每次运行时用户输入关键词系统即时向多个平台发起查询请求拿到结果直接返回。Scrapy是爬取-解析-入库的批处理模式它的异步框架Twisted和项目本身的工作流不太匹配。使用aiohttpasyncio的好处是我可以像写普通函数一样组织代码请求、解析、结果合并一气呵成调试非常直观。比如在IDE里设一个断点就能看到所有协程的执行状态这种透明度在开发调试阶段极其重要。当然如果有一天这个系统需要扩展到爬取全站资源、需要断点续爬、需要分布式调度我会毫不犹豫迁移到Scrapy。工具没有绝对的好坏只有适不适合当前场景的问题。4. 多平台接入实战适配器模式的解析层设计4.1 不同网站的数据结构差异从HTML到JSON的适配思路你可能会想直接发请求、解析、返回数据不就行了为什么要专门搞一个适配器层当你真正接触多个不同平台的页面结构后就会发现内容结构不统一是集成过程中最大的坑。拿我接入的几个数据源举例平台A的搜索结果直接就是结构良好的JSON接口请求某个带参数URL就能拿到干净的数据字段命名也很规范。这种最省心解析代码不超过10行。平台B的搜索结果虽然也是HTML但书籍条目都嵌在带有特定class属性的div中标题在h3a标签里下载链接在后面的按钮里。用parsel配合XPath就能准确提取。平台C做得最绝它的书籍列表数据根本不直接在HTML里而是作为一段JSON字符串嵌在script typeapplication/ldjson标签中。直接把HTML当XML解析会累死人正确做法是先用正则或XPath把这个JSON片段抠出来再交给json.loads处理。所以适配器层的核心抽象是每个平台一个类对外暴露统一接口方法async def search(session, keyword) - List[BookItem]。无论底层是JSON还是HTML最终返回的都是结构统一的BookItem对象。4.2 适配器基类与统一数据模型定义先定义统一数据模型from dataclasses import dataclass from datetime import datetime dataclass class BookItem: title: str # 书名 author: str # 作者 format: str # 格式如 PDF/EPUB/MOBI size: str # 文件大小如 12.5MB source: str # 来源平台名称 url: str # 详情页或下载页链接 updated_at: str # 更新时间然后定义适配器基类from abc import ABC, abstractmethod from typing import List, Dict, Any import aiohttp class BaseAdapter(ABC): 所有平台适配器的基类 name: str base_url: str # 每个平台专属的请求间隔避免同名适配器请求过快 request_interval: float 0.8 def __init__(self): self.last_request_time 0 async def _rate_limited_sleep(self): 平台级限速每个平台独立控制节奏 import asyncio, time now time.monotonic() elapsed now - self.last_request_time if elapsed self.request_interval: await asyncio.sleep(self.request_interval - elapsed) self.last_request_time time.monotonic() abstractmethod async def search(self, session: aiohttp.ClientSession, keyword: str) - List[BookItem]: 每个平台必须实现自己的搜索逻辑 pass_rate_limited_sleep这个小函数设计得很巧妙。它在每个适配器内部维护一个上次请求时间戳如果两次请求间的时间间隔小于设定的阈值就sleep补足。这样即使共享同一个网络连接池每个平台也能独立控制自己的请求节奏。4.3 实战JSON接口型数据源的解析平台A是最简单的JSON类型它的搜索接口大概长这样class PlatformAAdapter(BaseAdapter): name 平台A base_url https://api.example-a.com async def search(self, session, keyword): await self._rate_limited_sleep() url f{self.base_url}/search params { q: keyword, page: 1, size: 20, } async with session.get(url, paramsparams) as resp: if resp.status ! 200: return [] data await resp.json() books [] for item in data.get(data, {}).get(list, []) or []: book BookItem( titleitem.get(title, ), authoritem.get(author, ), formatitem.get(file_type, ), sizeitem.get(file_size, ), sourceself.name, urlitem.get(detail_url, ), updated_atitem.get(update_time, ) ) books.append(book) return books这种适配器写起来非常轻松但有一个容易被忽视的坑JSON字段的嵌套层级和命名风格每个平台都不一样有的用data.list有的用result.records有的字段名是bookName而不是title。所以每个适配器内都要单独写映射关系这是无法避免的工作量。4.4 真实网页中JSON-LD数据的提取细节第三个平台的处理要复杂不少。我看过大量网页源码后发现很多信息类网站会把结构化数据以JSON-LD格式嵌入页面头部方便搜索引擎理解和展示。这实际上给爬虫开发者提供了一个后门——数据就摆在script typeapplication/ldjson标签里等着取。用parsel提取这类数据非常方便import json from parsel import Selector class PlatformCAdapter(BaseAdapter): name 平台C base_url https://www.example-c.com async def search(self, session, keyword): await self._rate_limited_sleep() url f{self.base_url}/search params {keyword: keyword} async with session.get(url, paramsparams) as resp: content await resp.text() sel Selector(textcontent) scripts sel.xpath(//script[typeapplication/ldjson]/text()).getall() books [] for script in scripts: try: data json.loads(script) except json.JSONDecodeError: continue # JSON-LD数据可能是单对象也可能是对象数组 items data if isinstance(data, list) else [data] for item in items: if item.get(type) ! Book: continue books.append(BookItem( titleitem.get(name, ), author, .join(item.get(author, [])), formatitem.get(bookFormat, Unknown), sizeitem.get(fileSize, ), sourceself.name, urlitem.get(url, ), updated_atitem.get(datePublished, ) )) return books这里有一个细节要特别强调JSON-LD中author字段经常是一个对象数组每个对象里有type和name字段如果直接把它当字符串用就会得到一长串Python字典的字符串表示。必须先把author字段取出来再做类型判断和拼装。这种你以为取到的是字符串其实是列表的情况在真实爬虫开发中比比皆是。还有一个相关细节JSON-LD的bookFormat字段每个平台定义都不一样有的写Paperback纸质书有的写EBook电子书我在入库前会做一轮手工映射把不统一的格式值翻译成自己的规范值。4.5 CSS选择器与XPath的选择经验在写HTML型平台解析代码时选择器用XPath还是CSS我个人的经验是结构性较强的页面用XPathclass命名规范且富有语义化的页面用CSS更简洁。实际上parsel同时支持两种语法切换成本为零。遇到一个需要注意的场景某些网站为了前端渲染方便会把class名做混淆处理比如c1x8k3、_2Vd4F这种这种情况下用CSS老老实实匹配就很容易踩坑。正确做法是先用浏览器开发者工具的Copy as XPath功能拿到参考路径再人工改成更稳健的相对路径写法。我频繁使用的XPath模式# 提取所有书籍条目的容器 //div[contains(class, book-list)]/div[contains(class, item)] # 提取相对路径下的标题 .//h3[contains(class, title)]/a/text() # 提取链接 .//h3[contains(class, title)]/a/href # 提取作者和大小等元数据 .//span[contains(class, author)]/text()注意XPath只写了相对路径开头是.这个点代表从当前上下文节点开始查找在for循环遍历每个书籍条目时必须以当前条目为上下文否则会匹配到页面里其他无关元素。5. 智能检索模块不止是字符串匹配5.1 为什么选择SQLite FTS5做全文索引数据采集回来后接下来要解决的是检索问题。很多人第一反应是用SELECT * FROM books WHERE title LIKE %keyword%这套方案在数据量小的时候确实没问题但一旦数据量上万LIKE查询无法利用索引全表扫描的耗时就会变得不可接受。SQLite从3.9版本开始引入了FTS5全文搜索扩展它可以为文本数据建立倒排索引查询速度提升一两个数量级。更重要的是FTS5支持中文分词吗严格说它自带的分词器对中文支持并不好默认的unicode61分词器是按空格和标点切词的英文效果不错但中文一整句话都会被当成一个词。这会导致什么结果搜索三体时如果标题里存的是三体全集或三体黑暗森林FTS5默认分词会把整个三体全集当成一个token三体这个token在倒排索引里根本不存在搜索不出来。解决办法有两个方向方向一在应用层做中文分词把分词后的词组用空格拼接存在额外的列里搜索时也对query做同样处理。方向二使用SQLite官方提供的simple分词器配合tokenizeunicode61 remove_diacritics 0效果有限。我实际采用的是方向一的简化版在写入索引前用jieba对标题和作者字段做分词把分词结果用空格连接后存在FTS5虚拟表里以下几种情况就不会出现白屏问题了。5.2 FTS5虚拟表设计与关键词搜索实现建表语句CREATE VIRTUAL TABLE IF NOT EXISTS books_fts USING fts5( title, authorname, content, tokenize unicode61 ); -- 每次插入新书数据时同步写入分词后的内容 INSERT INTO books_fts (title, authorname, content) VALUES (?, ?, ?);用jieba分词的代码片段import jieba def tokenize_chinese(text: str) - str: 把中文文本分词用空格连接 words jieba.cut(text) return .join([w.strip() for w in words if w.strip()])实际插入数据时title_tokens tokenize_chinese(book.title) author_tokens tokenize_chinese(book.author) conn.execute( INSERT INTO books_fts (title, authorname, content) VALUES (?, ?, ?), (title_tokens, author_tokens, book.title book.author) )这里存了两份数据分词后的token串用于索引匹配原始文本用于展示。搜索查询时def search_books(query: str): FTS5全文检索用BM25算法做相关性排序 query_tokens tokenize_chinese(query) sql SELECT b.*, bm25(books_fts) AS score FROM books_fts JOIN books b ON books_fts.rowid b.id WHERE books_fts MATCH ? ORDER BY score LIMIT 50 return conn.execute(sql, (query_tokens,)).fetchall()bm25()是FTS5内置的相关性评分函数BM25算法是搜索引擎领域最经典的排序算法之一返回的score越小表示相关性越高负值趋近0最佳直接按它排序就能让最匹配的结果排在最前。你可能会问建表时只建了FTS5虚拟表那books表在哪实际的books表存的是去重后的完整元数据FTS5表通过rowid和books.id建立映射关系。两个表通过JOIN查询。这样设计的好处是FTS5专注做全文索引books表专注存结构化的完整信息互不干扰数据更新时分别操作两张表即可。5.3 智能排序策略把最可能想要的放在最前面有了相关性评分还不够实际使用中我发现只靠BM25排序会有一个问题同一个关键词搜索出来几百条结果有些是新书有些是几年前的有些资源格式齐全有些只有一个格式。用户真正想找的往往是资源最好、格式最新、大小合理的那个。于是我在搜索SQL里加入了排序权重ORDER BY score ASC, CASE WHEN b.format IN (EPUB, PDF) THEN 0 ELSE 1 END, b.updated_at DESC解释一下score ASC先按相关性排序BM25得分越小越相关。CASE WHEN format IN (EPUB, PDF)EPUB和PDF是最通用、兼容性最好的电子书格式优先展示这两种格式的结果。占第二位优先级的排序条件。b.updated_at DESC最后按更新时间倒序让最近更新的资源排前面。用户输入搜索词后实际返回的结果质量比单纯用关键词匹配好很多。我强烈建议你在自己的系统里根据实际数据情况调整这个排序规则比如如果你的目标用户更看重MOBI格式就把MOBI放在优先列表里。排序逻辑是整个检索体验中最容易被忽视的环节但也是性价比最高的优化之一。5.4 去重逻辑多平台结果合并的隐藏陷阱多平台采集有个绕不开的问题同一本书在多个网站都有收录如果直接合并用户会看到好几条完全相同的记录体验很差。去重逻辑不能只看书名是否完全一致因为不同平台的标题格式可能略有差别比如三体全集和三体全集或者有的站标题里带了作者名。我实现的去重策略是归一化后比较。方法很朴素但实用import re def normalize_title(title: str) - str: 归一化标题消除格式差异 title re.sub(r[\(\[].*?[\)\]], , title) # 去掉括号内容 title re.sub(r[ -_·], , title) # 去掉常见连接符 title title.lower().strip() # 统一小写 return title在实际处理时维护一个normalized_title - BookItem的字典如果多条记录的归一化标题相同就比较它们的格式完整度、更新时间只保留最优的一条。这里注意一个边界情况有些书的标题本身就带着括号比如三体全集珍藏版如果直接去掉括号再归一化会丢掉全集这个重要信息。所以我的方案是只去掉括号但用括号内容做一次加权比较比如带全集“精排”这些关键词的记录在去重时优先保留。数据清洗的细节实在太多了很多时候你只有在实际处理大量真实数据后才会意识到这些问题。6. 真实运行中的踩坑与调试经验6.1 编码问题你以为的UTF-8不是真正的UTF-8做爬虫的人几乎都遇到过编码问题。中文网站最常见的两个编码是UTF-8和GBK/GB2312如果搞错了轻则个别字显示异常重则整段内容乱码。aiohttp的resp.text()方法默认使用HTTP响应头里的Content-Type字段指定的编码来解码。但有些网站的响应头没有正确声明编码甚至故意不声明或者声明了错误的编码这时候就需要手动处理。# 获取二进制内容手动解码 content await resp.read() # 优先使用响应头中的编码猜测否则用 chardet 自动检测 try: charset resp.charset or utf-8 text content.decode(charset, errorsignore) except LookupError: import chardet detected chardet.detect(content) text content.decode(detected[encoding] or utf-8, errorsignore)我在实际项目中遇到过这样的情况一个平台返回的HTML头部里明确写着charsetgb2312实际内容中却夹杂着UTF-8编码的字符。用errorsignore参数可以在解码时跳过无法识别的字节不会因为个别异常字符导致整个页面解析失败。6.2 IP被封的常见信号与应对策略这个环节我很想多聊几句因为如何检测自己是不是被封了这个问题比被封了怎么办更值得关心。判断IP是否被封不能只看请求成功后解析出的内容是否为空。以下几个信号值得关注页面返回200但内容明显变短或者跳转到一个验证码页面所有请求突然都返回403或418状态码响应速度突然变得极慢且包含很多重定向返回的HTML里出现请输入验证码“安全检查”等关键词我在代码里加了一个简单的检查函数def is_blocked(content: str) - bool: 简单判断页面是否包含封锁特征 block_keywords [验证码, captcha, access denied, 安全验证] return any(keyword in content.lower() for keyword in block_keywords)每次解析前先跑一遍检查如果确认被封立即停止该平台的所有请求并记录日志到文件方便事后分析。等到本地IP解封后再继续运行。应对策略方面除了第3节提到的请求频率控制和UA池还可以加长两次请求之间的间隔、随机化延迟时间、为关键请求增加重试退避策略。我个人不建议一开始就上代理池因为高质量代理成本高、低质量代理反而更容易触发封禁。把单IP的请求频率控制到合理范围以内是性价比最高的方案。6.3 断点续跑异常中断后的数据恢复长时间运行中网络波动、内存问题、电脑休眠都可能导致程序中断。如果每次中断都要从头开始采集那体验实在太糟糕了。我实现的方案是每条记录插入数据库时带一个source_task_id字段表示它属于哪一次采集任务。程序启动时先查询最后一次任务ID和该任务中已成功采集的平台列表只重新采集失败的部分。def get_last_task_status(conn) - dict: 查询最近一次采集任务完成状态 row conn.execute( SELECT task_id, GROUP_CONCAT(DISTINCT source) as sources FROM books WHERE task_id (SELECT MAX(task_id) FROM books) GROUP BY task_id ).fetchone() if not row: return {} completed_sources set(row[sources].split(,)) return {task_id: row[task_id], completed_sources: completed_sources}启动时把这个状态读出来completed_sources里已经成功的平台就不重复采集了。这个优化听起来简单实际使用中能帮你省掉大量重新采集的时间。6.4 asyncio调试技巧如何优雅地排查死锁和超时异步编程的一个大坑是程序看起来卡死了但没报任何错误。最常见的原因是某个协程一直不返回整个事件循环被占住了。排查方法第一设置统一的超时时间避免某个请求无限制等待。我在调度器里已经用aiohttp.ClientTimeout(totalTIMEOUT)做了处理。第二给长时间运行的任务设置总超时限制try: async with asyncio.timeout(30): results await asyncio.gather(*tasks, return_exceptionsTrue) except asyncio.TimeoutError: print(任务总超时强制返回)第三利用asyncio.timeout上下文管理器Python 3.11或asyncio.wait_for旧版本在关键路径上设置兜底超时。实际运行中单个平台的崩溃往往会让整个gather等待很久才返回设置了总超时后最坏情况半分钟就能恢复响应。调试时还可以手动启动一个简单的事件循环测试单个适配器import asyncio async def debug_adapter(): adapter PlatformAAdapter() async with AsyncScheduler() as scheduler: books await adapter.search(scheduler.session, 三体) for book in books: print(f《{book.title}》 {book.author} | {book.format} | {book.url}) asyncio.run(debug_adapter())单独调试一个适配器排除了并发干扰后能更快定位问题。这个习惯帮我节省了大量排错时间。7. 项目效果实测与性能对比7.1 实测数据异步采集到底比同步快多少为了让大家对这套方案的性能提升有直观认识我专门做了一组对比实验。测试环境普通家用带宽100Mbps目标平台3个每个平台搜索同一关键词并采集前20条结果。同步方案用requests库逐一请求异步方案用我上面的代码并发数设置8。采集方式总耗时成功请求数失败/超时requests 同步循环12.6秒600asyncio并发限流34.8秒600asyncio并发限流82.9秒600asyncio并发限流152.2秒601可以看到并发数从3提升到8性能提升非常明显从8提升到15收益急剧缩小还开始出现超时失败。这说明并发量不是越大越好8这个数量级在处理这种中小规模的采集任务时已经是甜点区。7.2 资源覆盖情况与检索质量测试我把系统跑了几天累计采集了几千条电子书元数据覆盖了文学、社科、计算机、经济管理等主要分类。用几个不同类型的查询词做测试搜索三体返回12条结果前三名是不同格式的《三体》全集排序合理。搜索Python编程返回86条结果包含入门书籍、进阶书籍、实战项目筛选精度不错。搜索一个比较冷门的专业名词量子机器学习也能返回4条结果说明覆盖度还行。故意输入一个不存在的书名不存在的某本书xyz返回0条结果没有出现莫名其妙的误匹配说明FTS5的匹配逻辑比较可靠。从使用体验来看最明显的改进是搜索响应时间。之前用LIKE模糊查询几万条数据全表扫描大概需要1-2秒。换成FTS5之后响应时间基本在50毫秒以内快了两个数量级。7.3 系统资源占用情况运行期间监控了内存和CPU占用。内存常驻内存在180MB左右主要是因为Python解释器本身和aiohttp的连接池占用。单进程并发8时峰值内存不超过300MB。CPU由于是IO密集型任务CPU占用极低平均在5%左右。只有在解析大量HTML时才看到CPU有短暂飙升。磁盘数据库文件在采集5000条记录后大小约为8MB包含FTS5索引非常轻量。这个资源占用水平完全可以跑在低配的云服务器或树莓派上做个人知识库的配套服务很合适。8. 几个值得留意的工程细节8.1 日志记录多平台状态一目了然开发过程中我意识到信息类爬虫项目的日志不能只写print。因为多个平台并发执行时打印输出的顺序是混乱的稍纵即逝的报错信息很快就被后续日志吞没。我的做法是使用Python标准库的logging模块每一个适配器配置一个带名称的logger日志输出格式包含时间、平台名、日志级别import logging logging.basicConfig( levellogging.INFO, format%(asctime)s [%(levelname)s] [%(name)s] %(message)s, handlers[ logging.FileHandler(searcher.log, encodingutf-8), logging.StreamHandler() ] ) logger logging.getLogger(PlatformA) logger.info(开始搜索关键词: %s, keyword) logger.warning(平台响应超时第%d次重试, attempt 1) logger.error(该平台已被封禁: HTTP %d, resp.status)加上日志的好处不用多说出现问题时回溯现场非常方便。我建议从项目一开始就规范日志格式不要等到出问题了再补。8.2 配置文件独立改参数不用翻代码我在config.py里集中定义了所有可调参数# 并发请求数量建议5-10 CONCURRENCY_LIMIT 8 # 单个请求超时时间秒 TIMEOUT 10 # User-Agent池 USER_AGENT_POOL [ Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36, Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.0 Safari/605.1.15, # 省略更多 ] # 每个平台请求间隔秒默认0.8 PLATFORM_REQUEST_INTERVAL 0.8调整并发量、超时时间、UA池都不需要动核心代码。对于长期维护的项目这种配置与代码分离的设计虽然简单却能让维护体验好很多。8.3 数据库定期维护FTS索引会变大FTS5索引会随着数据量的增加逐渐膨胀而且频繁的插入和删除操作会产生内部碎片。建议定期执行一次优化def optimize_db(conn): 优化FTS5索引重建表结构 conn.execute(INSERT INTO books_fts(books_fts) VALUES(optimize)) conn.commit()这个optimize命令会重组索引压缩体积。我在采集大量数据后手动跑一次效果明显。另外SQLite本身支持VACUUM命令回收未使用的空间但我实际体验是FTS5的专用优化命令效果更精准对数据库的锁定时间也更短。定期备份数据库文件这个问题在个人项目中不会造成严重影响但提前处理总比事后补救好。9. 最后再分享一个实用技巧如何处理搜索结果的动态加载在接入了几个平台之后我注意到有些网站的数据不是服务端直出的而是前端通过JavaScript异步请求JSON接口再渲染。这就导致直接请求搜索页面HTML拿不到任何数据。我总结了两条可行的路线都不是通过无头浏览器这类重武器去硬碰硬而是用更轻量、更稳定的方案第一条路线直接找XHR接口。打开浏览器开发者工具的Network面板搜索关键词筛选XHR类型的请求你通常能看到一个以.json或.do、.api等结尾的接口地址。它往往带一些加密参数或动态token但这些参数通常就藏在页面源码的某个script标签里或者在前面的JS文件中生成。仔细挖掘一下就能找到规律。第二条路线有些平台提供了CORS开放的公共API只是没有在页面UI上做明显入口。这类接口返回的数据比HTML更干净、更稳定。在动手写适配器之前先花几分钟翻看一下目标网站的网络请求面板往往能直接找到比页面解析简单得多的方案。但我要提醒一句如果一个平台的前端JS代码做了非常复杂的参数加密且把所有内容都放在异步加载中这种高强度的反爬设计往往意味着网站的拥有者不愿意被第三方程序访问。对于这类情况我的原则是直接放弃接入。爬虫开发的世界里不是所有接口都该去碰的懂得哪些不该碰有时比懂得怎么碰更重要——这也是我一路踩坑后最深的体会。