资讯动态

Python爬虫实战:从请求到存储的工程化采集与合规避坑指南

发布时间:2026/10/16 20:29:33 来源:尧图企业网站定制
简介这份资源面向具备一定编程基础、希望系统掌握网络数据采集的开发者与学习者围绕Python爬虫从基础概念到实战应用展开帮助读者理解自动化抓取工具的工作原理并落地到实际项目。内容涵盖发送请求、解析网页、数据存储三大核心环节重点讲解requests、BeautifulSoup、lxml与Scrapy等常用库的适用场景并延伸至动态加载处理、反爬应对与效率优化等进阶技巧同时强调法律法规、网站协议与用户隐私保护等合规要求。资源包共1个docx文件约16KB以文档形式系统梳理知识点与示例代码便于按章节查阅与对照练习。目前已有628人学习适合需要批量获取公开数据的研究人员、构建数据挖掘或信息监测系统的开发者以及希望将爬虫技术应用于个人小项目的学习者参考。1. 从一次被反爬拦到凌晨三点的采集任务说起很多人对 Python 爬虫技术的第一印象还停留在requests.get()加BeautifulSoup就能走天下直到某天接了一个需要持续采集几万条结构化数据的活儿才发现真正卡住你的不是解析 HTML而是请求被静默丢弃、返回内容变成验证页、IP 被限速、字段对不上。这篇笔记面向两类人刚学完语法想跑通第一个真实采集任务的新手以及做过一些脚本但没系统梳理过工程化与合规边界的熟手。我会按「先立住概念和选型 → 再动手跑通最小闭环 → 再拆并发、解析、存储、反爬应对 → 最后落到避坑和进阶技巧」的顺序讲每一步都给可复现的代码和参数说明。核心不是教你绕过什么而是教你用规范、可维护、对目标站点友好的方式把网络数据采集这件事做稳、做久、做对。2. 先想清楚再动手采集目标、请求模型与合规边界2.1 采集前必须回答的三个问题在写第一行代码之前我一般会先逼自己回答三个问题这三个问题决定了后面所有技术选型。第一数据是公开可访问的吗如果目标页面需要登录才能看到内容那采集行为就涉及账号权限和用户协议必须谨慎评估。公开页面、公开 API、站点主动提供的 RSS 或 sitemap才是优先考虑的对象。第二采集频率会不会影响对方服务一个单线程脚本每秒发几十个请求对小型站点来说就是压力。常见做法是主动限速把并发控制在个位数并在请求头里带上能表明身份的信息而不是伪装成浏览器疯狂刷。第三数据拿来做什么如果只是本地分析、学术研究、个人学习和商业再分发合规要求完全不同。涉及个人信息、版权内容、非公开数据时必须停下来确认授权。提示很多采集任务翻车不是因为技术不行而是因为一开始就没想清楚边界后面越写越复杂最后既不敢跑也不敢停。2.2 请求模型同步、异步与任务队列怎么选Python 爬虫的请求模型大致分三档选错了后面会非常痛苦。同步阻塞模型最简单requests逐个发请求代码直观、调试容易适合几十到几百个页面、对速度没要求的场景。缺点是等待 IO 的时间全浪费了。异步模型用aiohttp或httpx配合asyncio在单线程内并发处理大量请求适合页面数量大、单页响应快、需要控制内存占用的场景。代价是调试门槛高异常栈不直观同步库不能直接混用。任务队列模型把「待采集 URL」放进队列由多个 worker 消费适合需要断点续采、失败重试、分布式扩展的长期任务。常见组合是 Redis 做队列、Celery 或自写 worker 做消费。我的经验是先写同步版本跑通全流程确认字段和解析逻辑没问题再按需换成异步或队列。一上来就上分布式往往连数据长什么样都没看清。2.3 合规采集的四个硬约束合规不是一句口号落到代码里有四个硬约束。一是遵守robots.txt。用urllib.robotparser解析目标站点的规则至少做到不采集明确禁止的路径。二是控制请求频率。在请求之间加time.sleep()或用令牌桶限速别让脚本变成压测工具。三是设置合理的超时和重试。超时不是越长越好连接超时和读取超时分开设重试次数控制在 2 到 3 次避免失败请求无限堆积。四是保留可追溯的日志。记录每次请求的时间、URL、状态码、耗时出问题时能快速定位也方便向对方说明采集行为。import time import requests from urllib.robotparser import RobotFileParser HEADERS { User-Agent: MyDataCollector/1.0 (contact: exampleexample.com), Accept: text/html,application/xhtmlxml, } def can_fetch(base_url: str, target_url: str) - bool: rp RobotFileParser() rp.set_url(base_url.rstrip(/) /robots.txt) try: rp.read() except Exception: # 读不到 robots.txt 时不默认放行按保守策略处理 return False return rp.can_fetch(HEADERS[User-Agent], target_url) def fetch(url: str, timeout(5, 15), retries: int 2) - str | None: for attempt in range(retries 1): try: resp requests.get(url, headersHEADERS, timeouttimeout) if resp.status_code 200: return resp.text # 429 表示被限速退避后重试 if resp.status_code 429: time.sleep(2 ** attempt) except requests.RequestException: time.sleep(1 attempt) return None这段代码里timeout(5, 15)表示连接超时 5 秒、读取超时 15 秒分开设置能避免慢响应把整个脚本拖死。retries2配合指数退避是应对临时网络抖动和限速的常见做法。can_fetch在读取失败时返回False属于保守策略宁可少采也不越界。3. 跑通最小闭环请求、解析、存储一条龙3.1 用 requests parsel 搭一个可复用的采集骨架最小闭环只需要三件事发请求拿到 HTML、从 HTML 里提取字段、把字段存下来。解析库我常用parsel它同时支持 CSS 选择和 XPath比单独用BeautifulSoup更灵活也比纯正则稳。import csv import time import requests from parsel import Selector BASE https://example.com/list?page{} HEADERS {User-Agent: MyDataCollector/1.0} def parse_list(html: str) - list[dict]: sel Selector(texthtml) items [] # 每个条目是一个 article 节点字段用 CSS 选择器定位 for node in sel.css(article.item): items.append({ title: node.css(h2.title::text).get(default).strip(), link: node.css(a::attr(href)).get(default).strip(), date: node.css(time::attr(datetime)).get(default).strip(), }) return items def crawl(pages: int 3) - list[dict]: all_items [] for page in range(1, pages 1): url BASE.format(page) try: resp requests.get(url, headersHEADERS, timeout(5, 15)) resp.raise_for_status() except requests.RequestException as exc: print(fpage {page} failed: {exc}) continue all_items.extend(parse_list(resp.text)) time.sleep(1) # 主动限速别把对方站点当压测目标 return all_items def save_csv(rows: list[dict], path: str output.csv) - None: if not rows: return with open(path, w, newline, encodingutf-8) as f: writer csv.DictWriter(f, fieldnamesrows[0].keys()) writer.writeheader() writer.writerows(rows) if __name__ __main__: data crawl(pages3) save_csv(data) print(fcollected {len(data)} rows)parse_list里用get(default)是为了防止某个字段缺失时直接抛异常采集任务里字段缺失是常态能容错就容错。time.sleep(1)是主动限速别小看这一秒它决定了你的脚本是「可持续采集」还是「跑十分钟就被封」。save_csv先判断rows是否为空避免空列表写文件时报错。3.2 字段对不上时怎么排查新手最常见的翻车是「代码没报错但字段全是空的」。排查顺序我一般这样走。先看请求是否真的拿到了目标 HTML。把resp.text写到本地文件用编辑器打开确认页面结构和你以为的一致。很多站点对无头请求返回的是简化页或验证页。再看选择器是否匹配。在浏览器开发者工具里复制选择器和代码里的对比。注意class名可能带动态后缀比如item-3f2a这种就不能写死。最后看编码。requests有时猜错编码导致中文变乱码字段看起来「对不上」。显式设置resp.encoding resp.apparent_encoding或按站点实际编码指定。3.3 存储选型CSV、SQLite 还是数据库小规模、一次性任务用 CSV 就够了简单、可直接用表格软件打开。需要反复查询、去重、增量更新时SQLite 是单机场景下最省事的选择不用装服务一个文件搞定。数据量大、多任务并发写入时再考虑 PostgreSQL 或 MySQL。import sqlite3 def save_sqlite(rows: list[dict], db: str data.db) - None: conn sqlite3.connect(db) cur conn.cursor() cur.execute( CREATE TABLE IF NOT EXISTS items ( id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT, link TEXT UNIQUE, date TEXT ) ) for row in rows: # link 加 UNIQUE 约束重复数据用 INSERT OR IGNORE 跳过 cur.execute( INSERT OR IGNORE INTO items (title, link, date) VALUES (?, ?, ?), (row[title], row[link], row[date]), ) conn.commit() conn.close()link字段加UNIQUE约束配合INSERT OR IGNORE天然实现去重。增量采集时重复跑同一批 URL 不会产生脏数据。这是我在长期任务里最常用的存储模式。4. 并发、解析与反爬应对把采集效率提上来4.1 异步采集的最小可用模板当页面数量到几千以上同步脚本的等待时间会变得不可接受。异步模板的核心是把「发请求」和「解析」解耦用信号量控制并发数。import asyncio import aiohttp from parsel import Selector HEADERS {User-Agent: MyDataCollector/1.0} SEM asyncio.Semaphore(5) # 同时最多 5 个请求在飞 async def fetch(session: aiohttp.ClientSession, url: str) - str | None: async with SEM: try: async with session.get(url, headersHEADERS, timeoutaiohttp.ClientTimeout(total20)) as resp: if resp.status 200: return await resp.text() if resp.status 429: await asyncio.sleep(2) except (aiohttp.ClientError, asyncio.TimeoutError): return None return None async def worker(urls: list[str]) - list[dict]: results [] async with aiohttp.ClientSession() as session: tasks [fetch(session, u) for u in urls] for html in await asyncio.gather(*tasks): if not html: continue sel Selector(texthtml) results.append({ title: sel.css(h1::text).get(default).strip(), }) return results if __name__ __main__: urls [fhttps://example.com/detail/{i} for i in range(1, 51)] data asyncio.run(worker(urls)) print(len(data))Semaphore(5)是关键参数它决定同时有多少请求在飞。设太大容易被限速设太小又失去异步的意义。我的经验是从 5 开始试观察状态码里 429 的比例再往上调。ClientTimeout(total20)是整体超时比单独设连接和读取超时更省心。4.2 解析策略CSS、XPath 与正则的取舍CSS 选择器可读性最好适合结构稳定的页面。XPath 表达能力更强能按文本内容、节点位置定位适合结构不规则或需要向上向下遍历的场景。正则只建议用在从文本里抠固定格式的片段比如从一段描述里提取日期或编号不要用正则解析整个 HTML。一个实用技巧是「先定位容器再在容器内取字段」。比如先选中所有article再在每个article里取标题和链接。这样即使页面其他地方有同名 class也不会串数据。4.3 反爬应对的正确姿势反爬机制常见的有几类请求头校验、频率限制、动态渲染、验证码。应对思路不是硬碰硬而是降低请求的「异常度」。请求头方面至少带上合理的User-Agent、Accept、Accept-Language让请求看起来像正常访问。频率方面主动限速、加随机抖动、错峰采集。动态渲染方面先确认数据是否在接口里直接返回很多页面其实是前端调 JSON 接口渲染的直接请求接口比渲染页面高效得多。验证码方面遇到就停不要试图自动化绕过这既不可靠也不合规。注意如果目标站点明确通过技术手段阻止采集继续对抗性采集可能违反其服务条款。此时正确做法是寻找官方 API、申请授权或放弃该数据源。5. 避坑与排查那些让我重跑整晚的坑5.1 坑一状态码 200 但内容是验证页现象脚本不报错解析结果全空把 HTML 存下来一看是「请完成验证」页面。原因站点识别出请求异常返回了验证页但状态码仍是 200。解决在解析前加内容校验比如检查页面里是否包含预期关键词或 HTML 长度是否低于阈值。不满足就标记为异常不要当成正常数据入库。5.2 坑二并发调太高导致大面积超时现象异步脚本刚跑时正常几分钟后大量请求超时成功率骤降。原因并发数超过对方承载能力触发限速或连接被丢弃。解决把Semaphore调小加入失败退避并监控成功率。成功率低于 90% 就该降并发而不是加机器。5.3 坑三字段选择器随页面改版失效现象昨天还能跑的脚本今天字段全空。原因目标页面结构调整class 名或层级变了。解决把选择器集中配置不要散落在代码各处。加一层「字段缺失告警」当某字段连续多页为空时主动提示而不是默默写入空值。5.4 坑四编码问题导致中文乱码现象标题里的中文变成测试这类字符。原因响应编码识别错误requests默认按头部猜测猜错就乱码。解决显式设置resp.encoding resp.apparent_encoding或按站点实际编码写死。存文件时统一用utf-8。5.5 坑五没有断点续采失败就得从头再来现象跑了两小时的任务在最后阶段失败重跑又要两小时。原因没有记录已完成的 URL状态全在内存里。解决把待采集 URL 和完成状态落到 SQLite 或 Redis每次启动先读状态跳过已完成的。这是长期任务的基本功。6. 进阶技巧让采集任务跑得更稳的几个习惯第一个习惯是给每个采集任务写一份「运行契约」。内容包括目标站点、采集字段、请求频率上限、超时和重试策略、存储位置、失败处理方式。这份契约不用很长但写下来之后换人接手或自己隔周再看都能快速恢复上下文。第二个习惯是把「请求层」和「解析层」彻底分开。请求层只负责拿到 HTML 并落盘或入缓存解析层只负责从 HTML 里提字段。这样页面改版时只动解析层网络策略调整时只动请求层互不干扰。我一般会把原始 HTML 按 URL 哈希存到本地目录解析失败时可以反复重放不用重新请求。第三个习惯是给采集结果加「质量校验」。比如检查必填字段的空值率、检查链接是否可访问、检查日期格式是否统一。校验不通过的数据单独存放不要混进主表。这一步能省掉后面大量清洗时间。第四个习惯是控制单次任务的规模。与其写一个跑一整天的大任务不如拆成多个小批次每批完成后落盘、记录进度。这样即使中途出问题损失也可控。最后一个习惯是定期回看日志里的状态码分布。200 占比、429 占比、超时占比这三个数字能告诉你任务健不健康。429 升高就降速超时升高就查网络或调超时参数。采集这件事稳比快重要得多。希望帮到你。本文还有配套的精品资源点击获取

读完文章,也想定制专属网站?

尧图设计师 24 小时内与您沟通定制方案

免费获取报价 →
↑