简介：这份Python爬虫技术学习文档，面向具备一定编程经验、希望利用Python高效采集网络公开数据的学习者与开发者。文档从爬虫作为自动化抓取工具的基本概念、核心优势与工作原理切入，逐步拆解发送请求、解析网页、数据存储三大技术环节，并结合具体代码示例，对比了requests、BeautifulSoup、lxml、Scrapy等常用库的适用场景。针对现代网站常见的AJAX动态加载、反爬虫策略与抓取效率瓶颈，也给出了Selenium、Pyppeteer模拟浏览器以及请求头伪装、数据清洗等实用解法。资源为docx格式，共1个文件，压缩包仅16KB，便于通读与查阅；目前已有628人学习。除技术细节外，文档还着重提醒遵守法律法规、网站协议与用户隐私保护，读者可据此合法合规地开发数据挖掘项目、信息监测系统或个人爬虫应用。
1. 先从一次“手动采集”说起：为什么爬虫不是跑通就算完
有位同事曾花一整周手工复制粘贴某个公开示例站的数据，汇总成表格发给业务方。第二周数据更新，他又得再来一遍，整个人接近崩溃。同期另一个小组拿 Python 写了不到一百行脚本，十分钟跑完全量，大家觉得“这事搞定了”。结果第二天，目标站点开始返回 418，访问直接受阻。这个落差就是本篇文章想聊透的：Python 爬虫技术真正难的不是“把页面的数据抓下来”，而是怎么在持续运行、目标站点规则变化、数据量快速膨胀之后，仍然保持高效、可维护、有边界。这里说的边界，既包括请求频率这类礼貌性规范，也包括 robots 协议和目的正当性。所以这篇笔记会把整套能力拆成六个环节：先从 requests 和 BeautifulSoup 把“单页抓取解析”跑通，再讲 User-Agent、限速、重试这些规范怎么写进代码，然后是线程池、异步、断点续爬这些提效手段，最后集中列出我在真实项目里踩过的高频坑。适合谁看？打算用 Python 做公开数据采集的开发者，以及团队里刚开始搭建采集模块、对“能跑就行”心有余悸的新人。
2. 从 requests 到 BeautifulSoup：把页面变成可处理的数据
2.1 最小请求代码：先看状态码、再决定要不要解析
爬虫的起点永远是“发一个请求，拿回一个响应”。新手最容易犯的错误是拿到响应就开始解析，完全不管状态码是不是 200。我习惯把请求封装成一个独立函数，统一处理超时、异常和状态码判断，这样后面所有采集任务都能复用同一个入口。
import requests from requests.exceptions import Timeout, RequestException def fetch_html(url, timeout=10, headers=None): """抓取页面 HTML，返回响应对象；请求失败或状态码异常时返回 None""" try: resp = requests.get(url, timeout=timeout, headers=headers) # 先看状态码，再决定后续逻辑 if resp.status_code != 200: print(f"状态码异常: {resp.status_code}, url: {url}") return None # 显式设置编码，避免后续解析出现乱码 resp.encoding = resp.apparent_encoding return resp except Timeout: print(f"请求超时: {url}") return None except RequestException as e: print(f"请求失败: {url}, 错误: {e}") return None这段逻辑的核心是“分阶段处理”。requests.get 返回的 Response 对象里带着状态码、响应头和正文，先判断状态码可以过滤掉大量无效页面，避免把 404 页面当正常内容解析。timeout 参数同时作用于连接建立和数据读取两个阶段，一个站点长时间无响应时，脚本不会死等。resp.encoding 是编码坑的补救手段，后面 5.3 会专门展开。
参数上，timeout 我一般从 10 秒起步，对响应特别慢的老站点可以放宽到 20 秒，但不要超过 30 秒，否则整个采集链路会被拖死。headers 建议从浏览器开发者工具里复制一份初始请求头，至少要带 User-Agent 和 Accept，这两个字段能在很大程度上决定目标站点把请求当成浏览器还是当成“程序在外面探头”。
2.2 用选择器定位数据：CSS 选择器和 XPath 的取舍
拿到 HTML 之后，下一步是从一堆标签里把目标数据抠出来。BeautifulSoup 是主流选择，它支持两种定位思路：CSS 选择器适合“按 class 或结构选”，XPath 适合“按层级关系定位”。先说结论：普通页面用 CSS 选择器即可，只有遇到结构特别深的嵌套，或者要根据某个兄弟节点的文本反查位置时，再切 XPath。
from bs4 import BeautifulSoup def parse_with_css(html, css_selector): """按 CSS 选择器提取文本内容，返回清洗后的列表""" soup = BeautifulSoup(html, "lxml") # select 返回所有匹配元素；切片限制数量防止误选到超长列表 items = soup.select(css_selector)[:50] # get_text 的 strip 参数去掉首尾空白，但保留元素内部结构 return [item.get_text(" ", strip=True) for item in items]解析器我固定用 lxml，比 Python 自带的 html.parser 快很多，适合批量采集。代价是 lxml 遇到极其畸形的 HTML 时可能报错，这时降级用 html.parser 就好，我在 5.3 会讲一个相关案例。select 和 select_one 的差别也要注意：select 返回列表，select_one 返回第一个匹配或 None，用来定位“唯一标题”这类场景更安全。
CSS 选择器里最常见的坑是 class 带空格。HTML 里 class="item active" 表示两个类名，CSS 选择器要写作 .item.active（不加空格），而不是 .item.active 写成 .item .active，后者表示“.item 的后代 .active”。这个细节能省下大量排查时间。
2.3 一个完整解析流程：从 HTML 到结构化列表
前两节是零件，这一节把它们拼成一条流水线。场景是抓取一个表格页面：第一列是名称，第二列是数值，采集结果存成 JSON 文件。这个例子后面会反复用到，并发、断点续爬都会在它的基础上扩展。
import json import requests from bs4 import BeautifulSoup def collect_table_data(url): """抓取表格页数据并落盘为 JSON""" resp = fetch_html(url) if not resp: return [] soup = BeautifulSoup(resp.text, "lxml") rows = soup.select("table tbody tr")[:100] result = [] for row in rows: cells = row.find_all("td") if not cells: continue # 逐单元格清洗：去换行、去多余空白 record = { "name": cells[0].get_text(" ", strip=True), "value": cells[1].get_text(" ", strip=True), } result.append(record) # ensure_ascii=False 让中文以原文写入，indent=2 方便人读 with open("data.json", "w", encoding="utf-8") as f: json.dump(result, f, ensure_ascii=False, indent=2) return result这条流水线的关键节点有三个。第一是“先定位行，再定位单元格”，拿到的 cells 顺序对应页面里的列顺序，清洗规则只针对单元格内容做。第二是清洗和解析同步进行，在采集端就把换行、多余空格处理掉，比落库后再补救省事得多。第三是 JSON 落盘时显式声明 encoding="utf-8" 和 ensure_ascii=False，否则中文字段会变成 \uXXXX 转义序列，既难读也对后续处理不友好。
这里的字段映射看着简单，实际踩坑率极高。比如某个页面表格列数不固定，或者某一行缺少某个单元格，依靠“cells[0]、cells[1]”这种位置索引就会整体错位，错误数据直到入库后才发现。所以我在循环里加了 if not cells 的跳过逻辑，但更稳妥的做法是先打印一行的完整 cells 内容核对列索引，确认无误后再批量跑。
3. 把“礼貌”变成代码：规范的请求头、频率与重试策略
3.1 User-Agent 与 Referer：让请求看起来像真实浏览器
“规范”这个词在爬虫语境里不是道德口号，而是实打实影响成败的技术参数。最基础的一项是 User-Agent（简称 UA），它告诉服务器“我是谁”。requests 默认的 UA 是 python-requests/版本号，目标站点一眼就能识别出这是程序请求，拒绝访问或返回空白页是常事。我一般维护一个小型 UA 列表，每次请求随机选一个。
import random USER_AGENTS = [ "Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/120.0.0.0", "Mozilla/5.0 (Macintosh; Intel Mac OS X 14_0) Firefox/121.0", ] def build_headers(): """构造模拟浏览器的请求头，每次调用随机更换 UA""" return { "User-Agent": random.choice(USER_AGENTS), "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", "Accept-Language": "zh-CN,zh;q=0.9", }随机更换 UA 的目的是打散特征，避免单一 UA 高频访问被特征命中。但我要提醒一句：UA 只是最表层的一层皮，很多站点真正识别的是请求节奏、点击路径、停留时长这些行为特征，所以这节内容必须和 3.2 的频率控制一起用才有效。Referer 字段作为第二道校验，部分站点会检查“你这个请求是从哪个页面发起的”，带上目标站首页或列表页地址即可。
3.2 请求频率控制：time.sleep 的替代方案与重试机制
控制请求频率最朴素的办法是 time.sleep(1)，每发一个请求等一秒。但这个写法有两个问题：固定间隔本身也是一种行为特征；另外，如果目标页面响应很慢，sleep 之后又会叠加上请求本身的耗时，整体节奏完全不可控。更工程化的做法是两种：一种是记录上一次请求时间的限速器，另一种是带退避策略的重试机制，一个管“频率上限”，一个管“失败恢复”。
import time from functools import wraps class RateLimiter: """最小间隔限速器：保证两次请求之间至少间隔 min_interval 秒""" def __init__(self, min_interval=2.0): self.min_interval = min_interval self.last_time = 0.0 def wait(self): elapsed = time.time() - self.last_time if elapsed < self.min_interval: time.sleep(self.min_interval - elapsed) self.last_time = time.time() def retry_with_backoff(max_retries=3, base_delay=1.5): """失败自动重试，重试间隔按倍数增长，避免集中重试加剧冲突""" def decorator(func): @wraps(func) def wrapper(*args, **kwargs): delay = base_delay for attempt in range(max_retries): try: return func(*args, **kwargs) except Exception as e: if attempt == max_retries - 1: raise print(f"请求异常，{delay:.1f} 秒后重试: {e}") time.sleep(delay) delay *= 2 return None return wrapper return decoratorRateLimiter 的思路是记录上一次请求的完成时间，如果距上次请求不足 min_interval，就补上剩余等待时间；如果已经超过，则立即放行。这样在高响应速度下能保持天然平滑的间隔，不会出现固定 sleep 叠加出来的“密集冲刺+密集等待”的锯齿状节奏。retry_with_backoff 解决的是另一种问题：单个请求偶发超时或连接重置，立即重试往往会被同一故障再次命中，指数退避让重试间隔逐渐拉大，把瞬时故障和短暂的服务端压力错开。
参数选择上，min_interval=2.0 是一个比较保守的起步值，适合中小站点；如果对方返回 429 Too Many Requests，说明你请求过密，把间隔放大到 5 秒甚至 10 秒。max_retries=3 是常规值，超过 3 次还失败，就该把这条 URL 记入失败日志，而不是继续硬闯。
3.3 robots.txt 与数据边界：哪些能爬、哪些该停
“规范”的另一个维度是主动权边界，具体落到代码里就是 robots 协议。绝大多数正规站点都会在根目录放一个 robots.txt，声明哪些路径允许抓取。Python 标准库提供了专用解析器，可以在请求前先查一次，避免踩进明确声明禁止访问的区域。
from urllib.robotparser import RobotFileParser def check_robots(base_url, path="/data/page.html", user_agent="MyBot/1.0"): """检查目标站点是否允许当前 UA 抓取指定路径""" rp = RobotFileParser() rp.set_url(base_url.rstrip("/") + "/robots.txt") rp.read() # can_fetch 返回 True 表示允许抓取 full_url = base_url.rstrip("/") + path return rp.can_fetch(user_agent, full_url)标准库里这个解析器会按 Robot Exclusion Protocol 处理 Allow 和 Disallow 规则，不需要自己写正则匹配，准确性有保证。注意点在于，有些站点的 robots.txt 对所有非主流浏览器 UA 一律拒绝，这时不建议换一个伪装成浏览器的 UA 去“骗过”规则——规则的存在本身就是边界，采集公开数据时尊重这个边界，比技术能力重要得多。另外，robots 协议是自律性规范而非强制法律，但从一个长期维护采集任务的人的角度看，遵守它意味着更低的被拦截风险，也让整个采集方案经得起推敲。
4. 高效采集的正确姿势：从单线程到并发与断点续爬
4.1 线程池并发：ThreadPoolExecutor 的正确打开方式
单线程逐页采集在大规模任务面前慢得让人着急，100 个页面每个请求要 1 秒，那就是 100 秒起步。线程池是第一个提效手段，它用若干个线程同时请求不同页面，把总耗时从“页面数×单页耗时”压缩到“页面数÷并发数×单页耗时”。Python 的 concurrent.futures 把线程池封装得很好，不用手动管理线程生命周期。
from concurrent.futures import ThreadPoolExecutor, as_completed def fetch_and_save(url): """示例任务：抓取页面并返回字节大小，实际项目中换成完整采集逻辑""" html = fetch_html(url, headers=build_headers()) return url, len(html.text) if html else 0 urls = ["https://example.com/list/1", "https://example.com/list/2"] with ThreadPoolExecutor(max_workers=5) as pool: futures = {pool.submit(fetch_and_save, u): u for u in urls} for future in as_completed(futures): url, size = future.result() print(f"完成: {url}, 大小: {size}")关键点在 as_completed 这个函数。它按“完成顺序”而不是“提交顺序”遍历结果，哪个请求先返回就先处理哪个，不会因为某个慢页面卡住整个循环。submit 返回的 Future 对象可以用 result() 拿到任务返回值，也可以捕获异常。如果某个任务反复失败，result() 会把异常抛到主线程，方便集中记录。
max_workers 这个参数是线程池里最重要的调优点。经验值是 5 到 10 起步，而不是贪多开到几十。因为采集是 IO 密集型任务，线程数量超过目标站点的承受能力后，收益不会线性上升，反而会大幅抬高被限流的概率。
4.2 异步与限速的平衡：aiohttp 的参数调优
线程池能解决一部分并发问题，但每开一个线程都有上下文切换开销，线程数开大后效率反而下降。异步基于事件循环，在单线程内用协程切换管理大量并发请求，更适合“几千个 URL 短时间内抓取”的场景。aiohttp 是社区里最常用的异步 HTTP 库，但它的用法和 requests 有明显差异，最大的坑是「并发不加节制」。
import asyncio import aiohttp async def bounded_fetch(session, url, semaphore): """带信号量限制的异步请求，防止一次性发出海量请求""" async with semaphore: async with session.get(url) as resp: if resp.status == 200: return await resp.text() return None async def run_async(urls, concurrency=10): """异步采集入口，concurrency 控制同时进行的请求数""" timeout = aiohttp.ClientTimeout(total=10) semaphore = asyncio.Semaphore(concurrency) async with aiohttp.ClientSession(timeout=timeout) as session: tasks = [bounded_fetch(session, u, semaphore) for u in urls] return await asyncio.gather(*tasks)semaphore 是整个模式的灵魂。没有它，asyncio.gather 会把所有 URL 一次性丢进事件循环，同时发出成百上千个连接，这不是“高效”，是主动制造攻击流量。用 Semaphore 把并发限制在 10 左右，既保留异步的高吞吐，又给目标站点留出呼吸空间。aiohttp 的编码处理也和 requests 不同：resp.text() 会按响应头里的 charset 解码，如果服务器没声明或声明错误，就用 resp.content.read() 拿到原始二进制再手动 decode，这个和 requests 的 apparent_encoding 体验差一截，异步场景下我更常写 resp.content.decode("utf-8", errors="ignore")。
4.3 断点续爬：把进度存下来，失败重启不白跑
线程池和异步解决了速度问题，但实战里还有一个比速度更现实的问题：任务中断。进程崩溃、网络断开、目标站点临时不可用，任何一个都可能导致全量重跑。断点续爬的思路是把“已完成的任务标记”实时存到磁盘，重新启动时先加载进度，跳过已完成的数据。
import json import os class ProgressStore: """基于 JSON 文件的进度存储，每次标记后立即落盘""" def __init__(self, path="progress.json"): self.path = path self.done = self._load() def _load(self): if os.path.exists(self.path): with open(self.path, "r", encoding="utf-8") as f: return set(json.load(f)) return set() def mark(self, key): """标记某个 URL 已完成，写入磁盘""" self.done.add(key) with open(self.path, "w", encoding="utf-8") as f: json.dump(list(self.done), f) def is_done(self, key): return key in self.done把进度存成 JSON 文件是成本最低的方案，单文件几百上千条没有任何压力。关键在于 mark 方法在每次标记后都立即落盘，虽然写文件频繁了一点，但换来的是进程在任何时刻崩溃都不会丢失超过一条进度的保障。对于几千页的采集任务，这个交换非常划算。
key 的选择也值得花心思。用完整 URL 做 key 是最简单的，但如果同一 URL 会有翻页参数变化，建议用“去除了无损参数的规范 URL”，避免同一个逻辑页面被当成多个任务。等整个采集跑完后，这个进度文件可以直接删掉，或者归档到日志目录，方便追溯这次任务实际跑过哪些页面。
5. 实战中高频踩坑的 6 个位置：反爬、编码、超时与数据异味
5.1 返回 418 或 403：不是没带 Cookie，是行为模式被识别
现象：代码没有任何语法问题，也带了 User-Agent，第一次请求成功，第二次开始返回 418 或 403，页面内容变成“检测到异常访问”之类的提示。
原因：目标站点通过行为特征识别出这是程序访问。固定请求间隔、无 Referer、请求密度突然升高、UA 和请求头搭配怪异，都会触发风控。它们看的不是单次请求内容，而是整体访问模式。
解决：把第 3 章的随机 UA、限速器、重试机制组合起来，人为打散请求节奏。另外，部分站点要求先访问首页拿到 Cookie 再访问数据页，我的做法是先用一个会话对象请求一次首页，把 Cookie 带进后续请求，会话内请求头保持一致。看到 418 时不要立刻调大并发硬闯，先降速观察恢复时间，这是最省钱的做法。
5.2 重定向与 JS 渲染：requests 拿不到的内容怎么办
现象：requests 返回 200，但解析出的 HTML 里没有预期数据，或者 response.url 指向了一个登录页或跳转页。
原因：页面内容由前端 JavaScript 动态渲染，初始 HTML 只有框架，数据在脚本执行后才出现；另一类情况是站点对非浏览器请求做 302 跳转。
解决：先打开浏览器开发者工具，切到“网络”面板，看页面实际请求里是不是有一个 XHR 或 Fetch 接口直接返回 JSON。如果有，直接把请求地址换成这个接口，响应更小、结构更规整，还没有渲染开销。如果数据必须等 JS 执行完才能拿到，再考虑无头浏览器方案，但无头浏览器内存开销大，并发能力弱，且更容易被识别，能走接口就走接口是我的长期习惯。
5.3 编码乱码：页面声明与实际编码不一致
现象：解析出来的中文全部变成“�”或者乱码字符，用浏览器打开却完全正常。
原因：response.headers 里的 charset 和 HTML meta 里声明的编码不一致，或者页面实际内容用的编码和声明不同。requests 默认用 headers 里的编码解码，一旦服务器没声明就默认 ISO-8859-1，中文必然乱码。
解决：在请求函数里用 resp.encoding = resp.apparent_encoding 做第一层修正。如果还是乱码，用 chardet 对原始字节做探测。
import chardet raw = resp.content detected = chardet.detect(raw) text = raw.decode(detected["encoding"], errors="ignore")这里 errors="ignore" 是兜底参数，遇到个别无法映射的字节直接跳过，保证整体文本可用。不过注意：探测编码有概率性，重要数据最好抽样人工核对，不要盲信自动判断。
5.4 数据异味：字段错位、空白字符和非法转义
现象：表格数据落库后，某一列的值整体串到了前一列；单元格里残留大量换行、制表符；JSON 文件里出现奇怪的转义符号。
原因：选择器定位到了外层容器而不是目标单元格，导致子元素文本被整体拼接；或者清洗时只做了 strip，没处理文本内部的 \r\n、\t 和连续空格。
解决：先单页断点打印一行的完整单元格列表，人工确认列索引指向。清洗时用正则统一处理内部空白。
import re def clean_text(value): """统一清洗：去首尾空白、压缩内部连续空格、替换换行制表符""" if not value: return "" value = value.replace("\r", " ").replace("\n", " ").replace("\t", " ") return re.sub(r"\s+", " ", value).strip()这个函数看似简单，但它在采集端解决的数据错位问题，远比在数据库里写一堆清洗 SQL 来得高效。字段错位的另一种来源是目标页面升级模板，导致列顺序变化，这时候固定索引的解析代码就会悄悄产出错误数据，所以定期对样本做校验是必要的。
5.5 重试机制放大重复采集：请求成功但响应超时
现象：日志里显示某个 URL 请求超时，触发重试后再次采集，最终落库数据出现重复记录。
原因：timeout 可能发生在“服务端已处理请求但响应返回超时”的情况下，第一次请求其实已经写入目标数据，但客户端认为失败并重试，同一份数据被采集两次。
解决：给每条数据一个业务主键（比如文章 URL 或详情页 ID），落库前先按主键查重。重试逻辑在插入前查一次进度表，发现已存在则跳过。这个设计和 4.3 的断点续爬天然配合，进度表同时承担“跳过已完成”和“去重”两个职责。
5.6 保存结构化数据：JSON、CSV、数据库怎么选
现象：用 CSV 保存中文数据，用 Excel 打开全部乱码；JSON 里嵌套了好多层，后续查询很别扭；数据量到几百万行时，单文件读写明显变慢。
原因：CSV 默认编码和 Excel 打开偏好不一致，JSON 适合嵌套结构但不利于随机查询，单文件在大数据量下自然遇到性能瓶颈。
解决：按使用场景选存储。采集阶段中间结果用 JSON，便于调试和断点续爬；最终交付给业务的表格用 CSV，但要加上 UTF-8 with BOM 或直接交付 Excel 兼容格式；数据量大且需要反复查询的，落 SQLite，轻量、单文件、支持 SQL。
6. 把采集任务沉淀成工具：一个多页面的规范采集实战
前五章的所有零件，最终要组装成一个能长期维护的小工具，而不是每次换项目都从零写。我习惯把任务拆成三个对象：调度器管流程，限速器管节奏，进度表管状态。调度器里只写“抓什么页面、怎么解析”，限速和断点这些公共能力由基类统一提供。
class CrawlJob: """采集任务基类：限速、查重、抓取、解析、标记 一条链路完成""" def __init__(self, url, headers, limiter, progress): self.url = url self.headers = headers self.limiter = limiter self.progress = progress def run(self): if self.progress.is_done(self.url): return self.limiter.wait() html = fetch_html(self.url, headers=self.headers) self.process(html) self.progress.mark(self.url) def process(self, html): """子类实现具体解析逻辑，返回清洗后的结构化数据""" raise NotImplementedError子类只需要实现 process 方法，把 HTML 变成结构化数据并写入存储，限速和去重由基类统一兜底。这样做的好处是：换一个采集目标时，唯一要改的是解析逻辑；限速策略、重试策略、进度保存这些最容易出错的公共环节，被固定在基类里，不会因为赶进度被省略。我以前也喜欢“一把梭”写一次性脚本，目标站点改个模板，脚本就废了，只能翻代码四处补丁。后来才明白，把规范内化成基类，比靠毅力“每次记得写限速”可靠得多。你最该抄的作业也不是某个华丽框架，而是把 RateLimiter、ProgressStore、重试装饰器这三个零件放进任何项目都能用的习惯。希望帮到你。
本文还有配套的精品资源，点击获取
