做采集项目这些年，我踩过最深的坑，不是反爬严，也不是解析难，而是把“抓列表”“抓详情”“下附件”全塞在一个脚本里，几百行代码串成一坨，跑到一半报错，从头再来。后来我把这套流程重构成“列表-详情-附件”三段式解耦管道，才真正体会到什么叫可持续的爬虫工程。今天这篇就把这套设计思路、数据模型、核心代码和排坑经验完整拆给你。
这篇文章适合两类人：一类是刚学完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_factory=dict) # 列表页能拿到的其他信息 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_factory=list) attachments: list = field(default_factory=list) # 附件链接列表 crawled_at: str = field(default_factory=lambda: datetime.now().isoformat()) status: str = "success"这里有个关键习惯：每个任务都带status和retry_count。这不是可有可无的，而是断点续采和失败重试的基础。没有状态标记，你就无法知道哪些任务处理成功、哪些处理失败，只能靠日志猜，很不靠谱。
2.3 状态标记与任务去重设计
任务去重是爬虫里最容易被忽略又最容易出问题的一环。列表页重复翻页、详情页有多个入口，都会导致重复采集。我这里的策略是维护一个已处理URL集合，处理前先检查。
import sqlite3 class TaskStore: """用SQLite存储任务状态和元数据""" def __init__(self, db_path="crawler.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, timeout=10) 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(strip=True) if href: tasks.append(ListTask(detail_url=href, title=title)) 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, timeout=10) resp.raise_for_status() resp.encoding = resp.apparent_encoding soup = BeautifulSoup(resp.text, "lxml") meta = ArticleMeta( url=list_task.detail_url, title=self._extract_title(soup, list_task.title), author=self._extract_author(soup), publish_time=self._extract_time(soup), content_text=self._extract_content(soup), tags=self._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(strip=True) 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", href=True): href = a["href"] if ext_pattern.search(href): full_url = urljoin(self.store.base_url, href) links.append(full_url) return links这里有个经验：文件扩展名判断只适用于一部分站点，有些下载链接不带扩展名，比如/download?id=123。对这种情况，可以配合链接文本里的文件名来判断，或者先请求一次HEAD看响应头里的Content-Type和Content-Disposition。这个后面在常见问题里展开。
详情解析完，别忘了做两件事：一是保存元数据到库，二是把附件任务推入下载队列。顺序很重要——先保存元数据，再入下载队列。万一附件下载挂了，元数据还在，可以单独补附件。
3.3 附件下载与文件命名策略
附件下载是这个管道里最容易踩坑的环节。文件可能很大、响应可能很慢、链接可能失效，下载策略必须稳健。
class AttachmentDownloader: def __init__(self, download_dir="downloads"): self.download_dir = Path(download_dir) self.download_dir.mkdir(parents=True, exist_ok=True) 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, stream=True, 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_size=8192): if chunk: f.write(chunk) return save_path def _resolve_filename(self, resp, default_name): cd = resp.headers.get("Content-Disposition", "") match = re.search(r'filename="?([^";]+)"?', 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关于下载，我想强调三点：
注意：下载大文件时一定要用stream=True，否则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(timeout=5) 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_url=link, article_url=meta.url, filename=link.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(timeout=5) 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_workers=4, download_workers=3): self.stage1_collect_tasks(start_url, max_pages) # 详情和附件可以并行，同一批工人同时处理两个阶段 with ThreadPoolExecutor(max_workers=detail_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_workers=download_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(strip=True): return node.get_text(strip=True) 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的记录入队。这样即使程序中途崩溃，重启后只需要从数据库恢复未完成的任务，不用重新抓列表。
这是一次把“解耦”做彻底后的额外收获——阶段之间不再通过“直接在内存里调用”耦合，而是通过持久化的任务状态耦合。虽然增加了写库的开销，但换来的可靠性和恢复能力，在高价值、大批量采集场景下非常值。
关于这套三段式管道，我最后还想补充一点：设计模式不是越复杂越好。如果你只是临时抓一次性数据，直接在脚本里写循环就够了。但如果你觉得自己要长期维护这个采集项目、数据量会持续增长、需求可能变化，那花费半天时间按“列表-详情-附件”三层解耦来构建，绝对划得来。我自己现在写新爬虫，不管大小，都会顺手把这三个阶段分开写，因为这样写的代码，过两个月自己回头看还能看懂，别人接手也容易上手。工程化的核心，不是代码多漂亮，而是出了问题你能快速定位、改了一处不影响其他地方，这就是解耦带来的最大价值。
