爬虫抓取网页数据：从选择器失效到工程化稳定方案

简介这份资源面向希望入门 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?page1 # 请求头是绕开基础拦截的第一道门槛 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/xhtmlxml,application/xml;q0.9,*/*;q0.8, Accept-Language: zh-CN,zh;q0.9,en;q0.8, Referer: https://example.com/, } # 超时一定要设否则网络抖动会把整个脚本挂死 resp requests.get(url, headersheaders, timeout10) 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, retry3): for i in range(retry): try: # 每次请求间隔 1.5 到 3.5 秒之间的随机值 time.sleep(random.uniform(1.5, 3.5)) resp session.get(url, timeout10) 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第 {i1} 次请求失败: {e}) return None这里的核心参数是random.uniform(1.5, 3.5)固定间隔比随机间隔更容易被识别。重试次数建议控制在 3 次以内超过 3 次还失败大概率是 IP 或账号被限制了继续重试只会加重封禁。429和503是典型的限流信号遇到它们要主动退让而不是硬刚。2.3 代理池与请求头轮换的取舍当单 IP 请求量上去之后代理池几乎是绕不开的。但代理池不是越多越好低质量代理会让你的失败率飙升反而拖慢整体进度。我的经验是先测代理可用性再按目标站点的容忍度决定并发数。方案适用场景维护成本风险点单 IP 限速小规模、低频采集低请求量一大就封短效代理池中等规模、多目标中代理质量参差需实时校验长效代理 会话大规模、需登录态高成本高配置复杂请求头轮换方面不要只换User-AgentAccept-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 classlist div classitem>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(stripTrue), price: price_el.get_text(stripTrue) 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(fdebug_{ts}.html, w, encodingutf-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( levellogging.INFO, format%(asctime)s %(levelname)s %(message)s, filenamecrawler.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 条数据做小批量验证确认字段完整、类型正确、去重生效再放开全量跑。这个习惯帮我省过很多次“跑了一晚上才发现字段全错”的后悔药。抓取这件事稳比快重要能持续跑三个月的脚本比一天抓一百万条然后被封的脚本有价值得多。希望帮到你。本文还有配套的精品资源点击获取