简介：这份资源面向希望入门 Python 爬虫与数据采集的开发者，围绕 Scrapy 框架展开，解决从 BBS 类网站批量抓取并提取结构化数据的实际问题，适合具备 Python 基础、想通过完整项目理解爬虫工程结构的学习者。压缩包共 14 个文件，以 py 源码与 pyc 编译文件为主，另含一份 docx 说明文档和 scrapy.cfg 配置文件，整体仅 18KB，轻量易读。源码中可见 spiders、items、pipelines、settings 等典型模块，覆盖请求调度、字段定义、数据管道与项目配置等核心环节，便于读者对照理解 Scrapy 的目录组织与运行机制。目前已有 5622 人学习下载，说明该案例在爬虫入门群体中具有一定参考价值。读者可借此掌握爬虫项目的搭建思路、结构化数据提取方式与常见模块职责划分，并在此基础上迁移到其他站点的采集任务中。
1. 爬虫抓取网页数据：为什么你写的选择器总在第二天失效
做过网页数据抓取的人大概都经历过这种崩溃：昨天跑得好好的脚本，今天一早起来发现抓下来的全是空值，或者干脆被目标站点拒之门外。这不是玄学，而是网页数据抓取这件事本身就在跟动态渲染、反爬策略和页面结构改版赛跑。所谓爬虫抓取网页数据，本质是用程序模拟浏览器请求，把 HTML 或接口返回的 JSON 解析成结构化字段，再落库或落文件。它适合做竞品价格监控、公开舆情汇总、行业数据采集这类需求，但前提是你得接受一个事实：任何写死 CSS 选择器的方案，生命周期都不会太长。真正能长期跑的抓取系统，靠的是分层设计——请求层管调度和伪装，解析层管容错和兜底，存储层管去重和增量。接下来我会按这个思路，把从最小可跑脚本到能扛住改版的工程化方案拆开讲，中间会重点说清楚请求头、限速、解析兜底和增量更新这几个最容易翻车的地方。
2. 请求层怎么搭：从单次请求到可调度抓取
2.1 先跑通最小请求闭环
不管后面要做多复杂的系统，第一步永远是让一个 URL 能稳定返回内容。很多人一上来就上框架，结果连请求头里少了什么字段都说不清。我一般先用最朴素的方式验证目标站点是否接受程序访问。
import requests # 目标页面地址，这里用占位域名示意 url = "https://example.com/list?page=1" # 请求头是绕开基础拦截的第一道门槛 headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/120.0.0.0 Safari/537.36", "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8", "Referer": "https://example.com/", } # 超时一定要设，否则网络抖动会把整个脚本挂死 resp = requests.get(url, headers=headers, timeout=10) resp.encoding = resp.apparent_encoding # 防止中文乱码 print(resp.status_code, len(resp.text))这段代码里最容易被忽略的是timeout和encoding。不设超时，一个慢响应就能让整个任务卡住；不处理编码，中文页面抓下来就是乱码。User-Agent和Referer是最基础的伪装字段，缺了它们，很多站点会直接返回 403 或空内容。跑通这一步之后，你才能判断问题是出在请求层还是解析层。
2.2 会话保持与请求频率控制
单次请求能通，不代表连续请求能通。目标站点通常会看你的请求间隔、Cookie 变化和 IP 行为。用requests.Session()可以自动保持 Cookie，模拟一个连续访问的会话。但更关键的是限速——我见过太多脚本因为请求太猛，前 50 条正常，第 51 条开始全部超时。
import time import random import requests session = requests.Session() session.headers.update({ "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/120.0.0.0 Safari/537.36", }) def fetch(url, retry=3): for i in range(retry): try: # 每次请求间隔 1.5 到 3.5 秒之间的随机值 time.sleep(random.uniform(1.5, 3.5)) resp = session.get(url, timeout=10) if resp.status_code == 200: return resp.text # 遇到 429 或 503 时拉长等待再重试 if resp.status_code in (429, 503): time.sleep(10 * (i + 1)) except requests.RequestException as e: print(f"第 {i+1} 次请求失败: {e}") return None这里的核心参数是random.uniform(1.5, 3.5)，固定间隔比随机间隔更容易被识别。重试次数建议控制在 3 次以内，超过 3 次还失败，大概率是 IP 或账号被限制了，继续重试只会加重封禁。429和503是典型的限流信号，遇到它们要主动退让，而不是硬刚。
2.3 代理池与请求头轮换的取舍
当单 IP 请求量上去之后，代理池几乎是绕不开的。但代理池不是越多越好，低质量代理会让你的失败率飙升，反而拖慢整体进度。我的经验是：先测代理可用性，再按目标站点的容忍度决定并发数。
| 方案 | 适用场景 | 维护成本 | 风险点 |
|---|---|---|---|
| 单 IP + 限速 | 小规模、低频采集 | 低 | 请求量一大就封 |
| 短效代理池 | 中等规模、多目标 | 中 | 代理质量参差，需实时校验 |
| 长效代理 + 会话 | 大规模、需登录态 | 高 | 成本高，配置复杂 |
请求头轮换方面，不要只换User-Agent，Accept-Language、Referer甚至sec-ch-ua这类字段也要跟着变。但要注意，请求头字段必须和你的 User-Agent 匹配，比如移动端 UA 配桌面端sec-ch-ua，反而更容易被识别。
3. 解析层怎么做：选择器、兜底与结构化
3.1 CSS 选择器与 XPath 的选用边界
解析 HTML 最常用的两种方式是 CSS 选择器和 XPath。CSS 选择器写起来简洁，适合结构清晰的页面；XPath 表达能力更强，能处理复杂的层级和文本匹配。我一般优先用 CSS 选择器，遇到需要按文本内容定位时才切 XPath。
from bs4 import BeautifulSoup html = """ <div class="list"> <div class="item">import json from datetime import datetime def parse_list(html): soup = BeautifulSoup(html, "html.parser") items = soup.select("div.list > div.item") results = [] for item in items: title_el = item.select_one("h3.title") price_el = item.select_one("span.price") if not title_el: continue # 标题缺失的条目直接跳过 results.append({ "title": title_el.get_text(strip=True), "price": price_el.get_text(strip=True) if price_el else "", }) # 命中率低于 50% 时保存快照，方便事后排查 if len(items) > 0 and len(results) / len(items) < 0.5: ts = datetime.now().strftime("%Y%m%d%H%M%S") with open(f"debug_{ts}.html", "w", encoding="utf-8") as f: f.write(html) return results保存快照这个动作看起来笨，但它是你事后定位问题的唯一后悔药。页面改版后，你至少能对比新旧 HTML，快速找到变化点。命中率阈值不要设太高，50% 到 70% 之间比较合理，太低会漏报，太高会误报。
3.3 结构化字段的清洗与类型转换
抓下来的文本往往带着空格、换行、货币符号，直接入库会很难用。清洗这一步不能省，尤其是价格、日期、数量这类字段。
import re def clean_price(text): if not text: return None # 去掉货币符号和千分位逗号，保留数字和小数点 num = re.sub(r"[^\d.]", "", text.replace(",", "")) try: return float(num) except ValueError: return None def clean_date(text): if not text: return None # 把 2024年1月5日 这类格式统一成 2024-01-05 m = re.search(r"(\d{4})年(\d{1,2})月(\d{1,2})日", text) if m: return f"{m.group(1)}-{int(m.group(2)):02d}-{int(m.group(3)):02d}" return text.strip()清洗函数要写成纯函数，输入输出明确，方便单独测试。价格字段用float之前一定要做异常捕获，因为有些页面会显示“面议”或“暂无报价”，直接转换会抛异常。日期格式尽量统一成 ISO 格式，后续做增量更新和时间排序会方便很多。
4. 存储与增量：别让重复数据淹了你的库
4.1 用唯一键做去重
抓取任务跑久了，重复数据是最大的噪音来源。去重最可靠的方式是在存储层加唯一约束，而不是在应用层用列表判断。
-- 以商品抓取为例，用外部 ID 加来源站点做联合唯一键 CREATE TABLE items ( id INTEGER PRIMARY KEY AUTOINCREMENT, source_site TEXT NOT NULL, external_id TEXT NOT NULL, title TEXT, price REAL, updated_at TEXT, UNIQUE(source_site, external_id) ); -- 插入时遇到重复就更新价格和更新时间 INSERT INTO items (source_site, external_id, title, price, updated_at) VALUES ('example', '101', '示例标题一', 29.90, '2024-01-05 10:00:00') ON CONFLICT(source_site, external_id) DO UPDATE SET price = excluded.price, updated_at = excluded.updated_at;UNIQUE(source_site, external_id)这个联合唯一键是关键，它保证同一个站点同一个商品只会有一条记录。ON CONFLICT ... DO UPDATE是 SQLite 和 PostgreSQL 都支持的语法，MySQL 里对应的是ON DUPLICATE KEY UPDATE。用数据库约束去重，比在 Python 里维护一个已抓取 ID 集合要可靠得多，因为后者在脚本重启后会丢失状态。
4.2 增量抓取的时间窗口设计
全量抓取只适合首次建库，日常运行必须做增量。增量策略有两种：一种是按更新时间过滤，只抓最近 N 天有变化的条目；另一种是按页码或游标推进，记录上次抓到的位置。
| 增量方式 | 实现难度 | 数据完整性 | 适用场景 |
|---|---|---|---|
| 时间窗口过滤 | 低 | 依赖站点提供时间字段 | 有明确更新时间的列表 |
| 游标记录 | 中 | 高 | 分页稳定的列表 |
| 全量对比 | 高 | 最高 | 数据量小、要求精确 |
我一般优先用时间窗口，因为实现简单，而且大多数列表页都支持按时间排序。如果站点不提供时间字段，就用游标记录上次抓取的最后一个 ID，下次从该 ID 之后开始抓。游标要持久化到文件或数据库里，不能只放在内存。
4.3 抓取日志与失败重试队列
一个能长期跑的抓取系统，必须有日志和失败队列。日志记录每次请求的 URL、状态码、耗时和解析结果数量；失败队列存下那些请求失败或解析为空的 URL，等主流程跑完后再统一重试。
import logging import sqlite3 logging.basicConfig( level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s", filename="crawler.log", ) def save_failed_url(conn, url, reason): conn.execute( "INSERT INTO failed_urls (url, reason, created_at) " "VALUES (?, ?, datetime('now'))", (url, reason), ) conn.commit() logging.warning("保存失败 URL: %s, 原因: %s", url, reason)失败队列的表结构很简单，但作用很大。它让你可以在不重跑全量的前提下，单独处理那些偶发失败的请求。重试时建议换一套请求头或换一个代理，避免用同样的特征再次触发拦截。
5. 避坑与排查：那些让你白跑一晚上的问题
5.1 返回 200 但内容是空的
现象：请求状态码是 200，但resp.text里没有目标数据，或者只有一个框架 HTML。原因通常是页面内容由 JavaScript 动态渲染，requests拿到的是初始 HTML。解决方式是抓接口而不是抓页面，打开浏览器开发者工具，在 Network 面板里找 XHR 或 Fetch 请求，直接请求返回 JSON 的接口。如果接口有签名或加密参数，就需要进一步分析，或者改用能执行 JS 的采集方案。
5.2 选择器昨天能用今天失效
现象：解析代码没改，但抓下来的字段全空。原因基本是目标页面改版，class 名或 DOM 结构变了。解决方式是平时就保存原始 HTML 快照，改版后对比新旧结构，快速定位变化点。长期方案是给选择器做多套备选，按优先级依次尝试，比如先试div.item，失败再试li.product。
5.3 请求频率一高就被封
现象：前几十条正常，之后全部超时或返回 403。原因是请求间隔太短或并发太高，触发了站点的限流。解决方式是降低并发、加大随机间隔、轮换请求头和代理。如果站点有Retry-After响应头，一定要按它给的时间等待，硬刚只会延长封禁时间。
5.4 中文乱码
现象：抓下来的中文显示为乱码或问号。原因是响应头里的编码声明和实际编码不一致。解决方式是优先用resp.apparent_encoding自动推断，如果还不对，就手动指定resp.encoding = "utf-8"或"gbk"。保存文件时也要统一用 UTF-8，避免二次乱码。
5.5 增量抓取漏数据
现象：明明有新数据，但增量任务没抓到。原因是时间窗口设得太窄，或者游标推进逻辑有 bug。解决方式是每次增量任务结束后，记录本次抓取的最大时间和最大 ID，下次从该位置往前多取一个批次做重叠校验。重叠部分用唯一键去重，不会产生重复数据，但能防止漏抓。
6. 进阶技巧：用断点续跑和字段级监控把抓取做稳
断点续跑是长期抓取任务里最值得投入的一个能力。它的核心思路是把“待抓取队列”和“已完成集合”持久化，脚本重启后能从上次中断的地方继续，而不是从头再来。实现上可以用 SQLite 存队列状态，也可以用 Redis 的 set 结构。我一般用 SQLite，因为不依赖额外服务，部署简单。
import sqlite3 def init_queue(conn, urls): conn.execute("CREATE TABLE IF NOT EXISTS queue (" "url TEXT PRIMARY KEY, status TEXT DEFAULT 'pending')") for url in urls: conn.execute("INSERT OR IGNORE INTO queue (url) VALUES (?)", (url,)) conn.commit() def next_pending(conn): row = conn.execute( "SELECT url FROM queue WHERE status = 'pending' LIMIT 1" ).fetchone() return row[0] if row else None def mark_done(conn, url): conn.execute("UPDATE queue SET status = 'done' WHERE url = ?", (url,)) conn.commit()这套逻辑跑起来之后，即使脚本因为网络或机器原因中断，重新启动也会自动从第一个pending的 URL 继续。INSERT OR IGNORE保证重复添加 URL 不会报错，status字段让你能清楚看到还有多少没抓。
字段级监控是另一个实用技巧。不要只监控“抓取是否成功”，而要监控每个关键字段的命中率。比如标题字段命中率突然从 98% 掉到 60%，说明页面结构可能变了，但整体请求还是成功的。实现方式是在解析完成后统计各字段的非空比例，低于阈值就发告警。
| 监控指标 | 建议阈值 | 触发动作 |
|---|---|---|
| 请求成功率 | 低于 90% | 检查代理和限速 |
| 标题命中率 | 低于 80% | 检查选择器 |
| 价格命中率 | 低于 70% | 检查页面改版 |
| 新增条目数 | 连续 3 次为 0 | 检查增量逻辑 |
最后说一个我自己的习惯：每次调整选择器或请求参数后，先拿 10 条数据做小批量验证，确认字段完整、类型正确、去重生效，再放开全量跑。这个习惯帮我省过很多次“跑了一晚上才发现字段全错”的后悔药。抓取这件事，稳比快重要，能持续跑三个月的脚本，比一天抓一百万条然后被封的脚本有价值得多。希望帮到你。
本文还有配套的精品资源，点击获取
