简介：这份资源面向Python爬虫初学者与需要快速上手Scrapy框架的开发者，围绕BBS网站数据抓取这一典型场景，提供了一套可直接参考的实战代码。包内共14个文件，以6个py源码与6个pyc编译文件为主，另含1个docx说明文档和1个scrapy.cfg配置文件，压缩包约18KB，体积轻巧便于下载后直接运行调试。源码覆盖items.py、pipelines.py、settings.py等Scrapy核心模块，清晰展示了从数据定义、管道处理到项目配置的完整结构，docx文档则对项目背景与使用方式做了补充说明，有助于理解各模块之间的协作关系。目前已有5622人学习下载，说明该案例在爬虫入门群体中具有较高的参考价值。读者可借此掌握Scrapy项目的目录组织方式、爬虫编写流程与配置要点，并在此基础上迁移到其他网站的数据采集任务中，适合作为爬虫框架学习的第一份练手项目。
1. 爬虫抓取网页数据：从“能跑”到“跑得稳”的分水岭
很多人第一次写爬虫，三行代码就能把网页源码拉下来，于是觉得这事已经会了。可真到要抓几万条商品价格、每天定时增量更新、还要保证字段不丢不乱的时候，才发现“能跑”和“跑得稳”之间隔着一整套工程习惯。爬虫抓取网页数据这件事，本质上是把非结构化的 HTML 或接口响应，稳定地变成结构化记录，并且在这个过程中不被反爬策略打断、不把脏数据写进库。它适合做数据分析、竞品监控、内容聚合的从业者，也适合想系统补齐采集链路的后端同学。这篇笔记不讲空泛概念，而是按“先立住原理、再动手复现、最后讲清坑”的顺序，把一套可落地的采集方案拆开讲，参数怎么设、失败看哪里，都尽量给到。
2. 抓取前的三个选型判断：静态、动态还是接口
2.1 先判断页面是服务端渲染还是前端渲染
拿到一个目标页，第一件事不是写代码，而是右键查看网页源代码，搜索你要的字段。如果源码里能搜到，说明是服务端渲染，直接用 HTTP 请求加解析就能拿到；如果搜不到，但页面上肉眼可见，那多半是前端渲染，数据由 JavaScript 异步请求回来。这两种情况的抓取路径完全不同，选错了后面全是无用功。
判断方法很直接：打开浏览器开发者工具的 Network 面板，筛选 XHR 或 Fetch，刷新页面，看有没有返回 JSON 的请求。如果有，优先抓接口而不是抓 HTML，因为接口返回的是干净的结构化数据，解析成本低、字段稳定。常见做法是复制该请求的 URL、请求头和参数，用代码复现一次，确认能拿到同样的 JSON。
import requests # 直接请求接口，而不是解析渲染后的 HTML url = "https://example.com/api/list" params = { "page": 1, # 页码，翻页核心参数 "pageSize": 20, # 每页条数，别设太大，容易被限流 "category": "book" # 业务筛选条件 } headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)", "Referer": "https://example.com/list", # 部分站点校验来源 "Accept": "application/json" } resp = requests.get(url, params=params, headers=headers, timeout=10) data = resp.json() # 接口返回 JSON，直接取字段 print(data["data"]["list"][0])这段代码的关键在于params和headers。page和pageSize控制翻页，Referer和User-Agent是最常被校验的两个头。timeout必须设，否则网络卡住时程序会一直挂着。如果返回 403 或空数据，先检查请求头是否完整，再检查参数是否被加密或签名。
2.2 静态页面用解析库，动态页面用浏览器自动化
如果确认是服务端渲染，用requests加BeautifulSoup或lxml就够了，速度快、资源占用低。解析时优先用 CSS 选择器或 XPath 定位，不要用正则去匹配 HTML 结构，因为页面稍微改版正则就崩。定位到节点后，取文本用.get_text(strip=True)，取属性用.get("href")，这两个是最常用的取值方式。
动态页面如果找不到接口，才考虑用浏览器自动化。它的代价是慢、吃内存，但能处理复杂的登录态和交互。用的时候注意设置无头模式和合理的等待策略，不要用固定sleep，而是等待某个元素出现。
from playwright.sync_api import sync_playwright with sync_playwright() as p: browser = p.chromium.launch(headless=True) # 无头模式，省资源 page = browser.new_page() page.goto("https://example.com/list", timeout=30000) # 等待目标元素出现，而不是死等固定秒数 page.wait_for_selector(".item-title", timeout=15000) titles = page.eval_on_selector_all( ".item-title", "els => els.map(e => e.innerText)" ) print(titles) browser.close()wait_for_selector比sleep可靠得多，元素一出现就继续，不会浪费时间也不会提前抓空。eval_on_selector_all一次性取回所有匹配节点的文本，比循环取单个节点效率高。浏览器自动化适合作为兜底方案，能抓接口就优先抓接口。
2.3 并发方式的选择：线程、协程还是队列
单线程顺序抓几百个页面还能忍，上到几千个就必须并发。IO 密集型任务用协程最划算，aiohttp加asyncio能在单线程内跑出很高的并发。但并发不是越高越好，每个站点能承受的请求频率不同，盲目开大并发只会触发封禁。
我一般会先小批量测试，从并发 5 开始，观察响应时间和状态码，逐步加到 20 或 50。如果出现大量 429 或连接超时，就说明到顶了。生产环境更稳妥的做法是用队列加固定数量的 worker，把待抓 URL 放进队列，worker 从队列取任务，这样并发可控、失败可重试。
import asyncio import aiohttp async def fetch(session, url, sem): async with sem: # 信号量控制并发上限 try: async with session.get(url, timeout=10) as resp: return await resp.text() except Exception as e: print(f"失败 {url}: {e}") return None async def main(urls): sem = asyncio.Semaphore(10) # 同时最多 10 个请求 async with aiohttp.ClientSession() as session: tasks = [fetch(session, u, sem) for u in urls] return await asyncio.gather(*tasks) urls = [f"https://example.com/item/{i}" for i in range(1, 101)] results = asyncio.run(main(urls))Semaphore(10)是并发控制的核心，它保证同时只有 10 个请求在飞。gather把所有任务收集起来一起跑。异常必须捕获，否则一个失败会拖垮整批。这套结构比无脑开线程池更可控，也更容易加限速和重试逻辑。
3. 把数据稳定落库：解析、清洗与增量更新
3.1 解析字段的容错写法
页面结构不会永远不变，今天能取到的字段明天可能就空了。解析时不能假设节点一定存在，每个取值都要有兜底。用BeautifulSoup时，先find再判断是否为None，取属性时用.get()而不是[]，这样缺字段时返回None而不是抛异常。
from bs4 import BeautifulSoup html = "<div class='item'><h3 class='title'>示例书</h3><span class='price'>39.9</span></div>" soup = BeautifulSoup(html, "lxml") item = soup.find("div", class_="item") title = item.find("h3", class_="title") price = item.find("span", class_="price") record = { "title": title.get_text(strip=True) if title else None, "price": float(price.get_text(strip=True)) if price else None, } print(record)if title else None这种写法看起来啰嗦，但能避免整批任务因为一个缺失字段而中断。价格转float前最好先去掉货币符号和空格，否则转换会失败。字段清洗的原则是：能在解析层做的标准化，不要留到入库后再补。
3.2 增量更新靠唯一键和去重
每天定时抓取时，最怕重复写入。解决办法是给每条记录确定一个业务唯一键，比如商品 ID 或 URL 的哈希，入库时用INSERT ... ON DUPLICATE KEY UPDATE或先查后写。唯一键要稳定，不能依赖会变的字段。
-- 以 url_hash 作为唯一键，重复抓取时更新价格和更新时间 INSERT INTO items (url_hash, title, price, updated_at) VALUES (%s, %s, %s, NOW()) ON DUPLICATE KEY UPDATE title = VALUES(title), price = VALUES(price), updated_at = NOW();url_hash建议用 URL 的 MD5 或 SHA1，长度固定、索引效率高。updated_at每次更新，方便后续判断数据新鲜度。如果只想保留最新价格，这种写法就够了；如果要留历史价格曲线，就得单独建一张历史表，每次插入新记录而不是更新。
3.3 限速与重试的工程参数
限速是为了不把对方站点打挂，也是为了避免自己被封。最简单的做法是每个请求之间加随机间隔，进阶做法是用令牌桶控制速率。重试要区分错误类型：连接超时和 5xx 可以重试，403 和 404 重试没意义。
import time import random import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry session = requests.Session() retry = Retry( total=3, # 最多重试 3 次 backoff_factor=1, # 退避系数，间隔 1s、2s、4s status_forcelist=[500, 502, 503, 504], # 只对这些状态码重试 ) session.mount("https://", HTTPAdapter(max_retries=retry)) for url in urls: resp = session.get(url, timeout=10) time.sleep(random.uniform(0.5, 1.5)) # 随机间隔，降低规律性backoff_factor让重试间隔指数增长，避免短时间内反复冲击。status_forcelist明确哪些状态码值得重试。随机间隔比固定间隔更难被识别为机器行为。这些参数没有万能值，要根据目标站点的响应情况调整。
4. 爬虫抓取网页数据的避坑清单：五个血泪教训
4.1 现象：本地跑得好好的，服务器上全是空数据
原因通常是服务器 IP 被目标站点识别为数据中心 IP，或者缺少浏览器环境导致某些校验失败。解决方法是先确认请求头是否完整，再检查是否被返回了验证页。可以在服务器上用curl手动请求一次，对比本地和服务器返回的内容差异。如果确实是 IP 问题，就需要调整采集频率或使用合规的访问方式。
4.2 现象：翻页翻到一半，后面全是重复数据
原因是目标站点的分页参数有上限，超过后返回最后一页或空页，而代码没有判断终止条件。解决方法是每页抓完后检查返回条数，如果少于预期或与上一页完全重复，就停止翻页。不要用固定页数循环，要用数据特征判断结束。
4.3 现象：字段偶尔多一个空格或换行，入库后查询对不上
原因是 HTML 里的空白字符没有清理干净。解决方法是所有文本字段统一做strip()，必要时用正则把连续空白替换成单个空格。入库前做一次字段校验，长度异常或包含控制字符的记录先隔离，不要直接写库。
4.4 现象：程序跑了几小时突然卡死，没有任何报错
原因多半是某个请求没有设超时，或者连接池耗尽后一直等待。解决方法是所有网络请求必须设timeout，包括连接超时和读取超时。用requests时timeout=(5, 10)分别控制连接和读取。协程方案里也要给aiohttp设ClientTimeout。
4.5 现象：昨天还能抓的接口，今天返回 403
原因是目标站点加了签名参数或令牌校验。解决方法是重新抓包，对比请求参数的变化，看是否多了sign、timestamp、nonce之类的字段。这类参数通常由前端 JavaScript 生成，需要分析生成逻辑。如果逻辑复杂，可以考虑用浏览器自动化直接执行页面脚本，而不是硬逆向。
5. 进阶技巧：用断点续抓和监控让采集任务可运维
采集任务一旦上生产，就不能靠“跑完看结果”来发现问题。我一般会给任务加两个东西：断点续抓和基础监控。断点续抓的思路是把已完成的 URL 或页码记录到本地文件或数据库，任务重启时跳过已完成部分。这样即使中途失败，也不用从头再来。
import os import json STATE_FILE = "crawl_state.json" def load_state(): if os.path.exists(STATE_FILE): with open(STATE_FILE, "r", encoding="utf-8") as f: return json.load(f) return {"done_pages": []} def save_state(state): with open(STATE_FILE, "w", encoding="utf-8") as f: json.dump(state, f, ensure_ascii=False) state = load_state() for page in range(1, 101): if page in state["done_pages"]: continue # 跳过已完成页码 # ... 执行抓取 ... state["done_pages"].append(page) save_state(state) # 每页完成后立即持久化done_pages记录已完成的页码，每次抓完一页就写盘。写盘频率不能太高，否则 IO 成瓶颈；也不能太低，否则失败时丢进度。每页一次是折中方案。状态文件建议加时间戳字段，方便判断任务是否卡住。
监控方面，最少要记录四个指标：请求总数、成功数、失败数、平均响应时间。这些数据可以打到日志里，也可以写到一个简单的统计表。我习惯在任务结束时输出一行汇总，失败率超过 5% 就人工介入。另外，给任务加一个最大运行时间，超过就主动退出，避免僵死进程占着资源。
还有一个容易被忽略的点是数据质量校验。抓完后随机抽几条记录，人工核对字段是否和页面一致。这一步看起来笨，但能发现很多自动化检查发现不了的问题，比如价格单位错了、标题截断了。我踩过一次坑，价格字段把“起”字也抓进去了，导致后续计算出错，就是因为没做抽样核对。现在每次上线新规则，我都会先跑 20 条样本，确认无误再放量。
采集这件事，快不是目标，稳和准才是。把并发控制住、把异常兜住、把进度存住，剩下的就是耐心调参。希望帮到你。
本文还有配套的精品资源，点击获取
