三段式爬虫管道设计：列表-详情-附件解耦采集架构

做采集项目这些年我踩过最深的坑不是反爬严也不是解析难而是把“抓列表”“抓详情”“下附件”全塞在一个脚本里几百行代码串成一坨跑到一半报错从头再来。后来我把这套流程重构成“列表-详情-附件”三段式解耦管道才真正体会到什么叫可持续的爬虫工程。今天这篇就把这套设计思路、数据模型、核心代码和排坑经验完整拆给你。这篇文章适合两类人一类是刚学完Python基础、想写第一个正经爬虫项目的初学者另一类是已经在用脚本抓数据、但感觉维护成本越来越高、想重构采集代码的开发者。不管你是抓新闻、抓商品数据、抓行业报告还是抓资料站的附件这套三段式管道都能直接复用。核心就一句话——把“发现链接”“提取元数据”“下载文件”三件事彻底拆开各管各的中间用队列和数据结构传输消息。1. 为什么非要做成三段式管道1.1 单脚本爬虫的痛点先说说我以前是怎么写爬虫的。很典型一个main函数里面先for循环翻列表页拿到详情链接后马上请求详情页用BeautifulSoup解析出标题、正文、发布时间再在页面里找附件链接顺手把附件下载下来。看着挺顺手但写长了就出问题。比如列表页翻到第50页突然超时前面的详情页数据已经在内存里了但你不知道哪些存了、哪些没存。重新跑一遍又全部请求一遍白白消耗对方服务器的资源也容易被封IP。更麻烦的是改需求。今天只抓标题明天要加一个作者字段后天要加一个图片下载。每当这种时候你就得在一大段耦合的代码里找哪里解析了详情、哪里拼接了路径改一处可能牵动三处。尤其附件下载下载失败还要重试如果和详情解析混在一起逻辑会非常混乱。我曾经一个项目光处理重试和断点续传就把主流程写成了面条代码后来谁都不敢动那套脚本。1.2 三段式管道解决了什么所谓“三段式”就是把一次完整的采集任务拆成三个阶段列表采集只负责翻页、解析列表、提取详情页链接和简略信息产出“待处理的详情任务”。详情采集消费详情任务请求详情页解析出结构化元数据并找出附件链接产出“待下载的附件任务”。附件下载消费附件任务按链接把文件拉到本地并校验文件大小、类型和完整性。三个阶段之间不直接调用函数而是通过队列或消息机制传递任务。列表采集不关心详情页长什么样详情采集不关心附件怎么保存附件下载也不关心元数据里有多少个字段。需要扩展时比如新增一个视频号源只需要多写一个列表解析器详情和下载的代码一行都不用改。这种松耦合的架构写起来前期会多一点设计成本但后期维护的体验完全不一样。1.3 解耦到底带来了什么有人觉得解耦是概念听起来虚。我用几个实际收益来说明第一是可平行调试。我以前遇到详情页解析报错只能在整个爬虫运行中打日志。现在三段独立我可以在命令行单独跑详情采集模块传入一个链接就能复现问题不需要启动整个管道。第二是可断点续采。因为每一段都消费“任务”这些任务如果持久化到本地那么中途断了重启后直接消费未完成的任务就行不需要从头抓列表。这个能力在采集上万级数据时特别重要。第三是可局部替换。比如今天我想把详情页解析从BeautifulSoup换成lxml或者把附件下载从requests换成httpx只要保证输入输出不变其他模块不受影响。数据还可以换存储从SQLite换到MySQL只改数据访问层就够。2. 技术选型与数据模型设计2.1 技术栈怎么选这套管道对技术栈的要求是“轻量、稳定、易调试”不需要一上来就上Scrapy或分布式框架。我的选择是Python 3.10用dataclass定义数据模型比字典强太多。requests处理HTTP请求配合Session复用连接。BeautifulSoup lxml做HTML解析。lxml做解析器性能稳定兼容大部分奇怪页面。SQLite做元数据和任务状态的存储。单机场景完全够用不用额外部署数据库。queue.Queue做内存队列配合ThreadPoolExecutor做并发。如果你采集量很大后续可以考虑把内存队列替换成Redis List或RabbitMQ但理念完全一样。我建议先跑通单机版再考虑分布式。2.2 定义三个核心数据模型解耦的第一步就是定义清楚“每一段之间传输什么”。我用dataclass定义三个模型分别对应三个阶段的任务。from dataclasses import dataclass, field from datetime import datetime dataclass class ListTask: 列表页采集产出一个待抓取的详情任务 detail_url: str title: str source: str # 来源站点标识 extra: dict field(default_factorydict) # 列表页能拿到的其他信息 status: str pending # pending / success / failed retry_count: int 0 dataclass class AttachmentTask: 详情页采集产出一个待下载的附件任务 file_url: str filename: str article_url: str # 关联的详情页便于追溯 file_type: str # pdf / zip / docx ... save_path: str status: str pending retry_count: int 0然后是详情页解析的结果也就是“元数据”。字段可以根据业务扩展但建议保留几个公共字段方便统计dataclass class ArticleMeta: 详情页元数据 url: str title: str author: str publish_time: str content_text: str tags: list field(default_factorylist) attachments: list field(default_factorylist) # 附件链接列表 crawled_at: str field(default_factorylambda: datetime.now().isoformat()) status: str success这里有个关键习惯每个任务都带status和retry_count。这不是可有可无的而是断点续采和失败重试的基础。没有状态标记你就无法知道哪些任务处理成功、哪些处理失败只能靠日志猜很不靠谱。2.3 状态标记与任务去重设计任务去重是爬虫里最容易被忽略又最容易出问题的一环。列表页重复翻页、详情页有多个入口都会导致重复采集。我这里的策略是维护一个已处理URL集合处理前先检查。import sqlite3 class TaskStore: 用SQLite存储任务状态和元数据 def __init__(self, db_pathcrawler.db): self.conn sqlite3.connect(db_path) self._init_tables() def _init_tables(self): cur self.conn.cursor() cur.execute( CREATE TABLE IF NOT EXISTS articles ( url TEXT PRIMARY KEY, title TEXT, author TEXT, publish_time TEXT, tags TEXT, content TEXT, crawled_at TEXT, status TEXT ) ) cur.execute( CREATE TABLE IF NOT EXISTS attachments ( id INTEGER PRIMARY KEY AUTOINCREMENT, article_url TEXT, file_url TEXT UNIQUE, filename TEXT, saved_path TEXT, file_size INTEGER, status TEXT ) ) self.conn.commit() def is_article_done(self, url: str) - bool: cur self.conn.execute(SELECT 1 FROM articles WHERE url ? AND status success, (url,)) return cur.fetchone() is not None def save_article(self, meta: ArticleMeta): self.conn.execute( INSERT OR REPLACE INTO articles VALUES (?,?,?,?,?,?,?,?), (meta.url, meta.title, meta.author, meta.publish_time, ,.join(meta.tags), meta.content_text, meta.crawled_at, meta.status) ) self.conn.commit()去重逻辑放在列表采集端产出ListTask之前先查一下详情URL是否已经在articles表里。如果已经采集过直接跳过不重复入队。这样即使列表页重复翻页也不会产生重复的详情请求。3. 核心实现三阶段管线3.1 列表页采集模块列表采集段的任务很纯粹输入一个列表页URL输出一批ListTask。为了通用我封装一个BaseListParser基类具体站点只需要实现解析方法。import time import requests from bs4 import BeautifulSoup from urllib.parse import urljoin class BaseListParser: 列表页解析基类子类覆盖 parse_list_html 即可 def __init__(self, task_store: TaskStore): self.store task_store self.session requests.Session() self.session.headers.update({ User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 }) def fetch_list_page(self, url: str) - str: resp self.session.get(url, timeout10) resp.raise_for_status() resp.encoding resp.apparent_encoding return resp.text def parse_list_html(self, html: str, base_url: str) - list[ListTask]: raise NotImplementedError def run(self, start_url: str, max_pages: int 10): tasks [] for page in range(1, max_pages 1): url start_url if page 1 else f{start_url}?page{page} html self.fetch_list_page(url) page_tasks self.parse_list_html(html, start_url) for t in page_tasks: # 去重已经入库的详情页跳过 if not self.store.is_article_done(t.detail_url): tasks.append(t) time.sleep(0.5) # 基本限速 return tasks一个实际子类示例class DemoListParser(BaseListParser): def parse_list_html(self, html, base_url): soup BeautifulSoup(html, lxml) tasks [] # 假设每条列表项在 .item-title a 里 for a in soup.select(.item-title a): href urljoin(base_url, a.get(href, )) title a.get_text(stripTrue) if href: tasks.append(ListTask(detail_urlhref, titletitle)) return tasks这里有几个细节值得注意。一是resp.encoding resp.apparent_encoding很多中文站点的编码声明不规范用apparent_encoding能减少乱码。二是urljoin详情页的链接往往是相对路径不join一下直接请求必挂。三是列表页解析里不要做详情页的逻辑比如不要在列表页提取正文——一旦混进去解耦就破功了。3.2 详情页元数据提取详情采集是管道中最核心的一段。它消费ListTask请求详情页解析出ArticleMeta同时识别页面里的附件链接生成AttachmentTask。class DetailParser: def __init__(self, task_store: TaskStore): self.store task_store self.session requests.Session() self.session.headers.update({ User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 }) def parse(self, list_task: ListTask) - ArticleMeta: resp self.session.get(list_task.detail_url, timeout10) resp.raise_for_status() resp.encoding resp.apparent_encoding soup BeautifulSoup(resp.text, lxml) meta ArticleMeta( urllist_task.detail_url, titleself._extract_title(soup, list_task.title), authorself._extract_author(soup), publish_timeself._extract_time(soup), content_textself._extract_content(soup), tagsself._extract_tags(soup) ) # 提取附件任务 for link in self._extract_attachment_links(soup): meta.attachments.append(link) return meta def _extract_title(self, soup, default_title): h1 soup.select_one(h1) return h1.get_text(stripTrue) if h1 else default_title解析方法可以单独定义成_extract_xxx形式每个方法只负责一个字段。好处是当某个站点的字段规则变了你只需要改对应方法不会污染其他解析逻辑。这其实是“解耦”在函数层面的体现。附件链接提取是详情段的关键能力。注意不是所有a标签都值得下载。我通常通过扩展名或链接特征来筛选import re def _extract_attachment_links(self, soup): 根据文件扩展名筛选附件链接 ext_pattern re.compile(r\.(pdf|zip|rar|7z|docx?|xlsx?|pptx?)$, re.I) links [] for a in soup.find_all(a, hrefTrue): href a[href] if ext_pattern.search(href): full_url urljoin(self.store.base_url, href) links.append(full_url) return links这里有个经验文件扩展名判断只适用于一部分站点有些下载链接不带扩展名比如/download?id123。对这种情况可以配合链接文本里的文件名来判断或者先请求一次HEAD看响应头里的Content-Type和Content-Disposition。这个后面在常见问题里展开。详情解析完别忘了做两件事一是保存元数据到库二是把附件任务推入下载队列。顺序很重要——先保存元数据再入下载队列。万一附件下载挂了元数据还在可以单独补附件。3.3 附件下载与文件命名策略附件下载是这个管道里最容易踩坑的环节。文件可能很大、响应可能很慢、链接可能失效下载策略必须稳健。class AttachmentDownloader: def __init__(self, download_dirdownloads): self.download_dir Path(download_dir) self.download_dir.mkdir(parentsTrue, exist_okTrue) self.session requests.Session() self.session.headers.update({ User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 }) def download(self, task: AttachmentTask) - Path: resp self.session.get(task.file_url, streamTrue, timeout(5, 30)) resp.raise_for_status() # 优先使用服务端提供的文件名 filename self._resolve_filename(resp, task.filename) safe_name self._safe_filename(filename) save_path self.download_dir / safe_name # 流式写入 with open(save_path, wb) as f: for chunk in resp.iter_content(chunk_size8192): if chunk: f.write(chunk) return save_path def _resolve_filename(self, resp, default_name): cd resp.headers.get(Content-Disposition, ) match re.search(rfilename?([^;])?, cd) if match: return match.group(1) if default_name: return default_name # fallback: 从URL取最后一段 return resp.url.split(/)[-1] def _safe_filename(self, name: str) - str: 去掉Windows和Linux下不安全的字符 name name.strip().replace(\\, _).replace(/, _) name re.sub(r[:|?*], _, name) name re.sub(r\s, _, name) return name关于下载我想强调三点注意下载大文件时一定要用streamTrue否则requests会一次性把文件读入内存一个几百MB的压缩包就能让程序内存爆掉。iter_content的chunk_size选择8KB到64KB都可以太小影响速度太大会占用内存。实测8KB在多数网络环境下表现均衡。注意文件名必须安全化。很多服务端返回的Content-Disposition头里带有filename*UTF-8...这种RFC 5987格式直接用会得到一串百分号编码。更常见的坑是文件名里带斜杠和冒号在Windows下直接写入就报错。所以_safe_filename这步不能省。文件名冲突是另一个容易被忽略的问题。两个不同链接可能指向同名文件如果直接覆盖写入前面的数据就丢了。我的方案是在文件名冲突时自动加序号def _unique_path(self, save_path: Path) - Path: if not save_path.exists(): return save_path stem save_path.stem suffix save_path.suffix counter 1 while True: new_path save_path.with_name(f{stem}_{counter}{suffix}) if not new_path.exists(): return new_path counter 1这个“重复自动改名”的思路同样适用于元数据存储——如果你的数据主键设计得不好后面就只能靠这种办法补救。4. 节点协作队列与调度编排4.1 用队列实现阶段解耦三段模块写好了要串起来还得有个“管道”。我的选择是Python内置queue.Queue双队列结构一个详情任务队列一个附件任务队列。import queue from concurrent.futures import ThreadPoolExecutor class Pipeline: def __init__(self, list_parser: BaseListParser, detail_parser: DetailParser, downloader: AttachmentDownloader): self.list_parser list_parser self.detail_parser detail_parser self.downloader downloader self.detail_queue: queue.Queue queue.Queue() self.download_queue: queue.Queue queue.Queue() def stage1_collect_tasks(self, start_url: str, max_pages: int): 阶段一列表采集产出详情任务 tasks self.list_parser.run(start_url, max_pages) for t in tasks: self.detail_queue.put(t) print(f[stage1] 共采集到 {len(tasks)} 个详情任务) def stage2_fetch_details(self, worker_id: int): 阶段二详情采集消费详情任务产出附件任务 while True: try: task self.detail_queue.get(timeout5) except queue.Empty: return # 队列为空退出线程 try: meta self.detail_parser.parse(task) self.list_parser.store.save_article(meta) for link in meta.attachments: self.download_queue.put(AttachmentTask( file_urllink, article_urlmeta.url, filenamelink.split(/)[-1] )) print(f[stage2-{worker_id}] 完成: {task.detail_url}) except Exception as e: print(f[stage2-{worker_id}] 失败: {task.detail_url} - {e}) finally: self.detail_queue.task_done() def stage3_download_files(self, worker_id: int): 阶段三附件下载 while True: try: task self.download_queue.get(timeout5) except queue.Empty: return try: path self.downloader.download(task) print(f[stage3-{worker_id}] 下载: {path}) except Exception as e: print(f[stage3-{worker_id}] 失败: {task.file_url} - {e}) finally: self.download_queue.task_done() def run(self, start_url, max_pages, detail_workers4, download_workers3): self.stage1_collect_tasks(start_url, max_pages) # 详情和附件可以并行同一批工人同时处理两个阶段 with ThreadPoolExecutor(max_workersdetail_workers) as pool: futures [pool.submit(self.stage2_fetch_details, i) for i in range(detail_workers)] for f in futures: f.result() with ThreadPoolExecutor(max_workersdownload_workers) as pool: futures [pool.submit(self.stage3_download_files, i) for i in range(download_workers)] for f in futures: f.result()这里有个设计细节stage2和stage3我用了两个独立的ThreadPoolExecutor顺序执行而不是全部塞进一个池。这样做的原因是避免附件下载阻塞详情解析。如果附件下载线程和详情解析线程在同一个池里某个大文件的下载会长时间占住线程导致详情解析速度下降。分开池子两个阶段可以各跑各的互不干扰。4.2 并发与限速的平衡很多人一上来就把线程数调到20、30结果很快被对方站点封掉。我自己常用的原则是做一个礼貌的爬虫。单站点并发控制在4~6个线程以内。每次请求间隔0.3~1秒随机上下浮动。设置合理的超时时间连接超时3秒、读取超时10秒。开启重试机制但重试次数不超过3次且重试间隔递增。线程数可以调整但建议不要把间隔设成0。曾经我为赶工把一个站的间隔设成0.1秒抓了2000多个页面后对方的WAF直接把我整段IP封了前功尽弃。后来规规矩矩加0.5秒间隔虽然慢一点但整整一周没出过事。4.3 断点续采与异常兜底断点续采的实现不完全靠内存队列因为程序一重启内存就清了。我前面设计的TaskStore在这里派上用场详情任务在入队前通过is_article_done去重附件任务在下载前通过数据库的file_url唯一约束去重。还有一类情况要兜底详情页解析报错比如页面结构临时改了或者某个字段提取方法取不到值。如果直接跳过数据就丢了。我的做法是把失败的任务单独存到一个failed表或者输出到日志文件后续统一重试。def safe_extract(func, default): try: return func() except Exception: return default单个字段的提取建议都用safe_extract包装一层。一个字段解析挂了不应该导致整条数据报废这是工程经验不是代码洁癖。5. 常见问题与排查技巧实录5.1 详情页字段缺失或结构不统一做采集最常遇到的一个问题同站点的详情页90%结构一致但总有10%的页面少了某个字段或者嵌套层级不同。比如有的文章有作者有的没有有的发布时间的class是publish-time有的是time。解决思路是“多级回退匹配”。拿标题举例先找h1找不到就找h2再找不到就取URL最后的slug。拿到什么算什么宁可少一个字段也不能让整条任务失败。def _extract_title(self, soup, default_title): for selector in [h1, .article-title h1, .content-title, h2]: node soup.select_one(selector) if node and node.get_text(stripTrue): return node.get_text(stripTrue) return default_title5.2 详情链接是动态加载的现在很多列表页是JavaScript渲染的直接requests拿到的HTML里只有空壳没有详情链接。两个思路找接口用浏览器开发者工具观察Network面板找到返回JSON数据的XHR接口直接请求接口。风险小速度快是我最推荐的方式。用渲染工具Playwright或Selenium把页面完整渲染后再拿HTML。成本高还容易被检测能不用就不用。很多人一碰到动态网页就上Selenium但仔细看看请求记录很多站点的数据其实都是通过接口加载的。你只需要找到那个接口URL把列表解析器的输入从HTML改成JSON反而更稳定。5.3 附件下载失败与大小校验附件下载失败的原因五花八门链接过期、文件名编码错误、服务端断流、网络超时。我的经验是“分层校验”下载完成后对比本地文件大小与服务端Content-Length头。不一致说明下载不完整标记失败安排重试。HTTP状态码非200时直接按失败处理。def _verify_download(self, resp, save_path): file_size save_path.stat().st_size content_length resp.headers.get(Content-Length) if content_length and int(content_length) ! file_size: raise ValueError(f文件大小不匹配: expected {content_length}, got {file_size})如果服务端没返回Content-Length就只能靠文件扩展名和后处理逻辑校验。比如PDF可以尝试读取文件头部的%PDF标记来判断是否完整。注意最坑的一种情况是服务器返回了一个200状态码但内容是一个错误提示页。比如下载链接已失效服务器重定向到了登录页。这种时候Content-Length对不上或者文件大小只有几KB而正常的PDF应该是几百KB。所以我的经验是给文件加一个最小大小阈值低于阈值的文件一律视为下载失败。5.4 反爬与限流关于反爬我的态度比较明确不要和站点硬碰硬。你的目标是拿数据不是为了证明爬虫技术有多强。遇到明确反爬的站点正确做法是降低请求频率模拟真人的浏览节奏。加载一个列表页后停几秒再加载下一页。用Session保持连接很多站点对不带Cookie的请求特别敏感。设置Referer头直接在headers里带上详情页自身URL能减少一部分拦截。还有一个容易忽略的点爬虫的requests库默认TLS指纹和浏览器不一致部分高防护站点会通过TLS握手特征直接拒绝。如果遇到这种情况可以试试httpx或curl_cffi它们的指纹模拟更接近浏览器。但这属于进阶话题常规站点用requests加合理限速就足够了。5.5 断点续采时队列里的任务丢失最后说一个我踩过的真实坑第一次写完管道stage1把1000个任务放进了内存队列stage2跑了一半程序崩了重启后队列是空的全部重新采集又担心被封。后来我把“入队”这一步也落盘。具体做法是stage1采集到的ListTask不是直接put进内存队列而是先写入SQLite的tasks表再启动stage2时从tasks表里读取status为pending的记录入队。这样即使程序中途崩溃重启后只需要从数据库恢复未完成的任务不用重新抓列表。这是一次把“解耦”做彻底后的额外收获——阶段之间不再通过“直接在内存里调用”耦合而是通过持久化的任务状态耦合。虽然增加了写库的开销但换来的可靠性和恢复能力在高价值、大批量采集场景下非常值。关于这套三段式管道我最后还想补充一点设计模式不是越复杂越好。如果你只是临时抓一次性数据直接在脚本里写循环就够了。但如果你觉得自己要长期维护这个采集项目、数据量会持续增长、需求可能变化那花费半天时间按“列表-详情-附件”三层解耦来构建绝对划得来。我自己现在写新爬虫不管大小都会顺手把这三个阶段分开写因为这样写的代码过两个月自己回头看还能看懂别人接手也容易上手。工程化的核心不是代码多漂亮而是出了问题你能快速定位、改了一处不影响其他地方这就是解耦带来的最大价值。