Python爬虫练习案例:从zip安全解压到并发升级实战

📅 发布时间:2026/9/14 2:16:31
Python爬虫练习案例:从zip安全解压到并发升级实战
简介这是一套面向爬虫初学者与毕业设计/大作业场景的Python爬虫练习案例合集围绕多个真实网站的数据采集与JS逆向展开如看准网、网易云评论、房天下、粉笔网、企名片、天翼云、巨潮资讯、新榜资讯、公共资源交易、欧科云链、得物等帮助读者理解反爬绕过、参数加密与数据接口分析思路。压缩包共229个文件以130个Python脚本和60个JavaScript文件为主另有操作文档、md说明、xml配置、txt备注及png截图等辅助材料整体大小约14.12MB便于本地对照练习。目前已吸引505人学习浏览适合正在准备课程设计或希望提升爬虫逆向能力的开发者。资源内包含chromedriver操作步骤等实战文档目录结构清晰代码注释与可运行实例能够支撑从请求、解析到存储的完整流程并配有部分网站的反爬绕过思路说明是一份能直接上手练手的实用素材。1. 一份“python爬虫练习案例.zip”到底在练什么很多人下载 zip 后第一件事就是解压、跑起来看到控制台刷出一堆数据就认为完成了。其实练习案例的价值不在那几十行代码而在于它刻意保留了真实爬虫会遇到的边界条件请求超时、状态码异常、页面结构变化、数据去重、并发控制以及磁盘上那个文件到底该不该信。zip 里装的是“练习”但练习对象不是某个网站而是你面对不确定网络 IO 时的决策能力。这个标题适合刚学完 Python 语法、想用 requests 写点真实代码的入门者也适合写过业务逻辑、但没系统整理过爬虫工程化边界的有经验工程师。读懂一份这样的 zip等于把“能跑”变成“能维护”。2. 用 zipfile 安全解压练习包先看懂目录再跑代码拿到“python爬虫练习案例.zip”第一件事不是双击解压而是在命令行里用一个 Python 脚本先读一遍条目。压缩包里的文件名并不可靠条目名可以包含../也可以写成绝对路径如果直接解压文件可能被写到目标目录之外。练习包来源越随意这个检查越值得做。2.1 解压前先做安全校验zipfile 读条目与路径穿越用 zipfile 模块先列出条目import zipfile from pathlib import Path src Path(python爬虫练习案例.zip) dest Path(./unpacked) with zipfile.ZipFile(src) as zf: for info in zf.infolist(): print(info.filename, info.file_size, info.is_dir())infolist()返回的是 ZipInfo 对象包含文件名、未压缩大小、目录标志比namelist()多出大小信息。先跑这一步能看到压缩包内是否有可疑路径。确认没有异常后再执行真正的解压。为了把路径穿越挡在门外我一般先用resolve()规范化目标路径再判断最终路径是否还在解压目录内with zipfile.ZipFile(src) as zf: for info in zf.infolist(): target (dest / info.filename).resolve() if not target.is_relative_to(dest.resolve()): raise RuntimeError(f路径越界: {info.filename}) zf.extractall(dest)resolve()会把a/../b这类路径展开为绝对路径is_relative_to判断规范化后的目标是否仍位于dest内。Python 3.9 以下没有is_relative_to可以改成str(target).startswith(str(dest.resolve()))注意加上路径分隔符后缀避免把unpacked_bad也算进去。extractall放在校验通过后执行这样不需要自己创建子目录。Windows 上还有一个隐蔽问题压缩包内文件名如果使用 GBK 编码解压后可能乱码因为 zipfile 默认按 UTF-8 读取。遇到乱码时可以读原始字节再手工解码raw info.filename.encode(cp437) name raw.decode(gbk, errorsreplace)cp437是 zipfile 在非 UTF-8 标志位下使用的默认编码把乱码字节还原后再用gbk或目标系统编码解析。这个处理常见于中文练习包直接改文件名为乱码再手工重命名成本更高。2.2 用探测脚本列出练习案例的目录结构与依赖解压后不要急着装环境先扫一遍目录。最直接的方式是递归列出所有文件和大小from pathlib import Path root Path(./unpacked) for path in sorted(root.rglob(*)): rel path.relative_to(root) if path.is_dir(): print(f[D] {rel}) else: size path.stat().st_size print(f[F] {rel} ({size} bytes))rglob(*)递归匹配所有层级relative_to(root)把输出转换成相对路径方便阅读。如果文件很多可以把输出重定向到文件再搜索。练习案例的典型结构大致如下目录/文件作用你应该重点看什么spiders/爬虫逻辑请求头、解析函数、数据清洗utils/公共工具日志、去重、网络重试data/抓取结果是否存在初始种子文件main.py入口参数解析、调度方式requirements.txt依赖清单版本约束是否合理README.md说明目标站点、运行方式、免责声明如果包里有README.md优先读它的“运行方式”小节多数练习包作者会写明 Python 版本和入口命令。如果连 README 都没有才需要从代码推断入口。2.3 创建 venv 并安装依赖锁定版本而非最新版练习包里的requirements.txt常写成requests2.25这种宽松约束。直接pip install -r requirements.txt会装到当前时间点的最新版可能和作者跑通的版本不一致最常见的坑是lxml与pydantic的兼容性报错。我一般这样处理python -m venv .venv source .venv/bin/activate pip install -r requirements.txt pip freeze requirements.lock.txtpython -m venv创建独立环境避免污染系统 Pythonsource .venv/bin/activate激活环境Windows 下对应.venv\Scripts\activatefreeze输出的是解析后的精确版本号锁定后可以提交到仓库别人用pip install -r requirements.lock.txt就能重建一致环境。如果没有requirements.txt先装最小依赖pip install requests beautifulsoup4 lxml再根据ModuleNotFoundError逐个补齐httpx、pandas等额外包。注意不要为了省事直接pip install -r requirements.txt --upgrade那会把本可复现的练习变成一次版本赌博。3. 把 requests 爬虫练习拆成可改的模板请求、解析、重试练习案例最常出现的代码是requests.get(url)后直接resp.text这在实验环境能通遇到超时、重定向、编码错乱就无解。下面给出一套可复用模板重构包内代码时按这三块去改。3.1 最小模板Session、headers、超时三件套用requests.Session代替裸get是练习第一处值得改的地方import requests from requests.adapters import HTTPAdapter session requests.Session() session.headers.update({ User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/120.0 Safari/537.36, Accept: text/html,application/xhtmlxml,application/xml;q0.9,*/*;q0.8, }) adapter HTTPAdapter(max_retries2, pool_connections10, pool_maxsize20) session.mount(http://, adapter) session.mount(https://, adapter) resp session.get(https://example.org/catalog, timeout(3.05, 10)) resp.raise_for_status()Session内部维护 cookie 和连接池连续请求同一个域名时能复用 TCP 连接headers.update设置了浏览器常见的 UA 和 Accept有些站点对缺失 UA 的请求直接返回 403。HTTPAdapter的max_retries控制连接失败时的自动重试次数注意它不覆盖业务状态码session.get里的timeout(3.05, 10)是(连接超时, 读取超时)元组连接超时略大于 3 秒是因为 TCP 握手的最小重传间隔读取超时给 10 秒足够。raise_for_status()在状态码为 4xx/5xx 时抛出异常让后续except统一处理。练习包里的原始代码如果直接print(resp.status_code)遇到 403 还会继续解析得到一堆空数据改成抛异常反而更容易暴露问题。3.2 用 CSS 选择器解析少用正则匹配很多练习包里解析 HTML 用的是re.findall(ra href(.*?))。正则写起来快但页面结构一调整就失效例如属性值里多一个空格、标签大小写变化都会让匹配结果漏掉一半。换成 BeautifulSoup 加 CSS 选择器可读性和改造成本都会好很多from bs4 import BeautifulSoup soup BeautifulSoup(resp.text, lxml) items [] for a in soup.select(div.list a.item): title a.get_text(stripTrue) href a.get(href) items.append({title: title, url: href})soup.select(div.list a.item)选出div.list下所有带itemclass 的链接get_text(stripTrue)会把标签内的空白字符和换行清理掉get(href)取属性如果属性不存在返回None。相比正则选择器表达的是“页面结构”而不是“字符串模式”后续维护时只要页面结构没变改动就小。如果响应内容是 JSON 接口直接用resp.json()遇到resp.text中文乱码先检查resp.encoding常见是网站没声明字符集。可以这样兜底if resp.encoding is None or resp.encoding.lower() not in (utf-8, gbk): resp.encoding resp.apparent_encodingapparent_encoding会基于内容做推断但推断速度慢不要在全站每个响应上都调用只在发现乱码时补一次。3.3 状态码与重试策略把练习包改成不裸奔练习代码通常只判断status_code 200把其他状态码全部当失败处理。实际爬虫应该按状态码区分策略常见处理如下状态码含义练习中怎么处理200成功解析并写入结果301/302跳转让 Session 自动跟随但记录最终 URL403拒绝访问检查 UA、Cookie降低频率404页面不存在记录后跳过不重试429请求过多读取Retry-After后等待500/502/503服务端错误指数退避重试最多 3 次一个带重试的封装import time def fetch_with_retry(session, url, max_retries3): for attempt in range(max_retries): try: resp session.get(url, timeout(3.05, 10)) if resp.status_code in (403, 429): wait int(resp.headers.get(Retry-After, 60 // (attempt 1))) time.sleep(wait) continue resp.raise_for_status() return resp except requests.RequestException: wait 2 ** attempt time.sleep(wait) return NoneRetry-After是服务端明确要求的等待秒数比随机 sleep 更值得尊重如果响应头里没有这个字段退化为60 // (attempt 1)即第一次等 60 秒、第二次等 30 秒。except requests.RequestException捕获包括超时、连接错误在内的网络异常2 ** attempt实现指数退避。练习包里如果只有sleep(1)可以对比这个版本体会“等待”本身也是爬虫策略的一部分。4. 爬虫并发设计从单线程到协程与分布式zip 案例怎么升级练习包跑通后下一步通常是提速。并发设计看起来选项很多到底哪个好取决于瓶颈在哪里。对 IO 密集的爬虫来说协程占用资源更少代码也更容易控制并发上限。下面分步说明。4.1 用 aiohttp 写并发爬虫配合 Semaphore 限流同步循环的问题在于大部分时间都在等网络回应。aiohttp 可以用一个事件循环同时发起多个请求但控制并发数量是最容易被忽略的一点import asyncio import aiohttp async def fetch_one(session, url, sem): async with sem: try: async with session.get(url, timeout10) as resp: text await resp.text() return url, resp.status, text except aiohttp.ClientError as exc: return url, 0, str(exc) async def crawl(urls, limit8): sem asyncio.Semaphore(limit) timeout aiohttp.ClientTimeout(total15) async with aiohttp.ClientSession(timeouttimeout) as session: tasks [fetch_one(session, u, sem) for u in urls] results await asyncio.gather(*tasks) return resultsSemaphore(limit)保证同一时刻最多limit个请求在途async with sem是获取和释放信号量的标准写法。aiohttp.ClientTimeout(total15)限制整个请求完成的最长时间它接受的是配置对象和requests的元组写法不同。gather会等待所有任务结束返回顺序与tasks传入顺序一致不依赖响应完成的先后。每个fetch_one里自己捕获ClientError这样单个失败不会让gather整体抛异常。并发数选多少没有绝对答案我一般按域名和目标站点反馈来调并发数适用场景风险1~3目标站点没有明确频率限制慢但安全4~8静态页面、响应延迟波动小较合理的默认值8~20内部 API、响应快需要观察失败率20分布式压测容易被封不建议在练习中使用如果某个并发数下失败率突然升高优先降低而不是增加并发。练习案例的urls列表最好从文件读取别在代码里硬编码。4.2 写文件要回到同步 IO不能把事件循环堵住协程里最容易犯的错是直接在异步函数里写文件。open().write()是阻塞调用放在fetch_one里会让事件循环等待磁盘等于把并发全废掉。正确做法是先收集结果等所有请求结束再统一落盘import csv from pathlib import Path def write_csv(results, output: Path): output.parent.mkdir(parentsTrue, exist_okTrue) with open(output, w, newline, encodingutf-8) as f: writer csv.writer(f) writer.writerow([url, status, html_len]) for url, status, text in results: writer.writerow([url, status, len(text)])output.parent.mkdir(parentsTrue, exist_okTrue)先建目录避免目标不存在时报错newline是 Python 在 Windows 下写 CSV 必不可少的参数不写的话行与行之间会多出空行encodingutf-8保证中文不乱码。write_csv是同步函数在asyncio.run(crawl(...))之后调用实际上就把网络等待和本地 IO 分隔开了。如果结果量大到内存放不下另一个常见做法是每个 worker 写自己的独立文件最后合并。这样虽然避免内存峰值但要处理多文件合并。练习包的数据量通常不到百万条统一内存写文件足够。4.3 分布式什么时候上单机够用就别上消息队列练习包里出现“分布式爬虫”字样实际代码却常常是本机多线程。真正的分布式要解决任务分发、去重、断点续跑三个问题通常需要消息队列和 Redis 同时配合已经超出“zip 练习”的范围。方案选择和落地成本对比如下方案适用场景落地成本requests 循环练习、几十个 URL几乎没有requests 线程池轻量并发低aiohttp 协程数百到数千 URL中消息队列 多机 worker超大规模、需要断点续跑高我建议除非这个 zip 里明确给出任务队列配置和分布式部署说明否则不要为了“分布式”强行引入一套消息队列。单机协程加上一个本地任务文件已经能覆盖绝大多数练习和中小规模采集场景。真正需要上分布式时更多是工程问题而不是爬虫问题——你要先解决任务怎么切分、worker 怎么发现新任务、失败任务怎么回收。还有一个常被忽略的并发问题日志输出在协程下是安全的但如果在多线程方案里让每个线程各自打印 status输出顺序会乱控制台一刷就不知道哪条对应哪个 URL。练习阶段可以把日志收敛为“每 50 个请求打印一次进度”别让 print 淹没关键状态。5. 给练习案例加一个可验证的收尾命令行入口与数据体检练习包常见的遗憾是只能从源码改成固定 URL 后运行。给它加一个命令行入口让同一套代码可以被反复复用是很划算的收尾。5.1 用 argparse 暴露常用参数import argparse import asyncio from pathlib import Path parser argparse.ArgumentParser(descriptionpython爬虫练习案例启动器) parser.add_argument(--url-file, typePath, requiredTrue, help包含目标 URL 的文件每行一个) parser.add_argument(--limit, typeint, default8, help并发上限) parser.add_argument(--output, typePath, defaultPath(data/result.csv)) args parser.parse_args() urls [line.strip() for line in args.url_file.read_text(encodingutf-8).splitlines() if line.strip()] results asyncio.run(crawl(urls, args.limit)) write_csv(results, args.output)--url-file是必填项requiredTrue强制用户在运行时指定read_text(encodingutf-8)明确编码避免 Windows 默认编码不一致splitlines()按行拆分后再用if line.strip()过滤掉空行。这样一改任何拿到这个 zip 的人都只需要准备一个 URL 列表文件不用打开源码找start_url。5.2 写 SQLite 做去重比 CSV 更耐查CSV 适合快速查看但反复跑同一份采集任务时会出现大量重复记录。SQLite 的一个小表就能解决去重问题import sqlite3 conn sqlite3.connect(data/result.db) conn.execute(CREATE TABLE IF NOT EXISTS items (url TEXT PRIMARY KEY, title TEXT, crawled_at TEXT)) def save_item(url, title): conn.execute( INSERT OR IGNORE INTO items (url, title, crawled_at) VALUES (?, ?, datetime(now)), (url, title), ) conn.commit()PRIMARY KEY加在url字段INSERT OR IGNORE让同一 URL 第二次写入时静默跳过不需要先查询再判断。datetime(now)由 SQLite 生成记录这次抓取的时间。如果练习包原先用的是 CSV这个表可以平滑替代查询时也能用SELECT COUNT(DISTINCT url)验证数据量。5.3 按响应耗时调整请求节奏而不是猜 sleep 值最后一个我想留给你的技巧把固定 sleep 换成对响应耗时的观测。很多反爬场景并不是请求被拒而是服务端故意把响应变慢逼迫爬虫自动放弃。import time def timed_get(session, url): start time.perf_counter() resp session.get(url, timeout10) cost time.perf_counter() - start recent.append(cost) return respperf_counter比time.time()精度更高适合测量短时间间隔。如果最近几次请求的耗时从 0.3 秒涨到 3 秒以上大概率是服务端开始降速这时应该停止重试把当前 URL 写回待处理队列等下一个周期再跑。实现时可以用sum(recent[-5:]) / len(recent[-5:])算移动均值阈值超过 2 秒就把--limit减半。这个自我调节逻辑比固定time.sleep(1)更接近真实爬虫的节奏也是从练习案例走向生产代码前最值得带走的一段。本文还有配套的精品资源点击获取