# Technical Documentation: NanmiCoder/CrawlerTutorial > ℹ️ **Provenance:** Multi-Tier Ingestion · [CodeWiki Reference](https://codewiki.google/github.com/NanmiCoder/CrawlerTutorial) ### Overview (`README.md`) # 关于作者 大家好，我是程序员阿江-Relakkes，近期我会给大家出一些爬虫方面的教程，爬虫入门、进阶、高级都有，有需要的朋友，star仓库并持续关注本仓库的更新。 - [Github万星开源自媒体爬虫仓库MediaCrawler作者](https://github.com/NanmiCoder/MediaCrawler) - 全栈程序员，熟悉Python、Golang、JavaScript，工作中主要用Golang。 - 曾经主导并参与过百万级爬虫采集系统架构设计与编码 - 爬虫是一种技术兴趣爱好，参与爬虫有一种对抗的感觉，越难越兴奋。 ## 查看教程 在线链接：https://nanmicoder.github.io/CrawlerTutorial/ 对应的视频链接近期也会同步更新出来，[查看B站合集地址](https://space.bilibili.com/434377496/channel/collectiondetail?sid=4035213&ctype=0) ## 爬虫入门 - [✔] [01_为什么要写这个爬虫教程](docs/爬虫入门/01_为什么要写这个爬虫教程.md) - [✔] [02_个人学会爬虫能赚钱吗](docs/爬虫入门/02_个人学会爬虫能赚钱吗.md) - [✔] [03_网络爬虫到底是什么](docs/爬虫入门/03_网络爬虫到底是什么.md) - [✔] [04_爬虫的基本工作原理](docs/爬虫入门/04_爬虫的基本工作原理.md) - [✔] [05_常用的抓包工具有那些](docs/爬虫入门/05_常用的抓包工具有那些.md) - [✔] [06_为什么说用Python写爬虫有天生优势](docs/爬虫入门/06_为什么说用Python写爬虫有天生优势.md) - [✔] [07_Python常见的网络请求库](docs/爬虫入门/07_Python常见的网络请求库.md) - [✔] [08_爬虫入门实战1_静态网页数据提取](docs/爬虫入门/08_爬虫入门实战1_静态网页数据提取.md) - [✔] [09_爬虫入门实战2_动态数据提取](docs/爬虫入门/09_爬虫入门实战2_动态数据提取.md) - [✔] [10_爬虫入门实战3_数据存储实现](docs/爬虫入门/10_爬虫入门实战3_数据存储实现.md) - [✔] [11_爬虫入门实战4_高效率的爬虫实现](docs/爬虫入门/11_爬虫入门实战4_高效率的爬虫实现.md) ## 爬虫进阶 - [✔] [01_工程化爬虫开发规范](docs/爬虫进价/01_工程化爬虫开发规范.md) - [✔] [02_反爬虫对抗基础_请求伪装](docs/爬虫进价/02_反爬虫对抗基础_请求伪装.md) - [✔] [03_代理IP的使用与管理](docs/爬虫进价/03_代理IP的使用与管理.md) - [✔] [04_Playwright浏览器自动化入门](docs/爬虫进价/04_Playwright浏览器自动化入门.md) - [✔] [05_Playwright进阶_反检测与性能优化](docs/爬虫进价/05_Playwright进阶_反检测与性能优化.md) - [✔] [06_登录认证_Cookie与Session管理](docs/爬虫进价/06_登录认证_Cookie与Session管理.md) - [✔] [07_登录认证_扫码与短信登录实现](docs/爬虫进价/07_登录认证_扫码与短信登录实现.md) - [✔] [08_验证码识别与处理](docs/爬虫进价/08_验证码识别与处理.md) - [✔] [09_数据清洗与预处理](docs/爬虫进价/09_数据清洗与预处理.md) - [✔] [10_数据分析与可视化](docs/爬虫进价/10_数据分析与可视化.md) - [✔] [11_进阶综合实战项目](docs/爬虫进价/11_进阶综合实战项目.md) ## 高级爬虫 - [✖] 待更新... ## 爬虫交流群 扫码加作者企微拉进群，备注来自github爬虫教程 企微二维码 ## 免责声明 >本仓库的所有内容仅供学习和参考之用，禁止用于商业用途。任何人或组织不得将本仓库的内容用于非法用途或侵犯他人合法权益。本仓库所涉及的爬虫技术仅用于学习和研究，不得用于对其他平台进行大规模爬虫或其他非法行为。对于因使用本仓库内容而引起的任何法律责任，本仓库不承担任何责任。使用本仓库的内容即表示您同意本免责声明的所有条款和条件。 ## Star History [](https://star-history.com/#NanmiCoder/CrawlerTutorial&Date) --- ### Overview (`docs/爬虫进价/README.md`) # Python 爬虫进阶教程 > 本教程是「Python 爬虫入门教程」的延续，面向已掌握基础爬虫技术的学习者。教程将从工程化实践入手，逐步深入反爬虫对抗、浏览器自动化、登录认证等进阶主题，最终通过综合实战项目巩固所学知识。 ## 适合人群 - 已完成入门教程 11 篇内容的学习者 - 了解 Python 异步编程基础（async/await） - 希望提升爬虫工程化能力和反爬对抗能力的开发者 ## 技术栈概览 | 类别 | 技术/工具 | |------|----------| | HTTP 客户端 | httpx (异步) | | 浏览器自动化 | Playwright | | 数据验证 | Pydantic | | 日志系统 | loguru | | 配置管理 | pydantic-settings | | 代理管理 | 自研代理池 | | 验证码识别 | ddddocr / 第三方打码平台 | | 数据分析 | pandas / wordcloud / jieba | --- ## 教程目录 ### [01_工程化爬虫开发规范](./01_工程化爬虫开发规范.md) **学习目标：** - 理解爬虫项目的工程化重要性 - 掌握 loguru 日志库的使用和日志分级策略 - 学会使用 pydantic-settings 进行配置管理 - 实现统一的异常处理和错误重试机制 - 了解项目目录结构的最佳实践 **核心内容：** - **日志系统设计**：loguru 基本使用、日志分级（DEBUG/INFO/WARNING/ERROR）、日志轮转和持久化、异步日志写入 - **配置管理**：环境变量 vs 配置文件、pydantic-settings 实现配置验证、敏感信息的安全管理（.env 文件）、多环境配置切换 - **异常处理**：自定义异常类设计、全局异常捕获、重试装饰器实现（tenacity 库）、优雅降级策略 - **项目结构规范**：分层架构设计、模块划分原则、代码复用策略 **实战案例：** 将入门教程第 10 章的数据存储爬虫进行工程化改造，添加完善的日志系统、配置管理和异常处理机制。 --- ### [02_反爬虫对抗基础_请求伪装](./02_反爬虫对抗基础_请求伪装.md) **学习目标：** - 理解常见的反爬虫检测手段 - 掌握 User-Agent 轮换策略 - 学会构造完整的请求头伪装 - 理解 Cookie 和 Session 的工作原理 - 实现基础的速率控制 **核心内容：** - **反爬虫机制概述**：基于请求特征的检测、基于行为特征的检测、基于内容的检测、常见的封禁策略 - **User-Agent 策略**：真实浏览器 UA 收集、UA 随机轮换实现、移动端 vs 桌面端 UA 选择、fake-useragent 库的使用 - **请求头完整伪装**：必要的请求头字段（Accept、Accept-Language、Accept-Encoding 等）、Referer 的正确设置、请求头一致性维护、使用 curl_cffi 模拟浏览器指纹 - **速率控制**：固定延迟 vs 随机延迟、自适应速率调整、令牌桶算法实现、asyncio.Semaphore 控制并发数 **实战案例：** 爬取一个具有基础反爬的电商网站商品列表，实现完整的请求伪装和智能速率控制。 --- ### [03_代理IP的使用与管理](./03_代理IP的使用与管理.md) **学习目标：** - 理解代理 IP 的工作原理和类型 - 学会评估代理 IP 的质量指标 - 掌握代理池的设计与实现 - 实现代理的自动检测和淘汰机制 - 理解隧道代理和 API 代理的使用 **核心内容：** - **代理 IP 基础**：HTTP 代理 vs HTTPS 代理 vs SOCKS5 代理、透明代理 vs 匿名代理 vs 高匿代理、免费代理 vs 付费代理的取舍、代理提供商选择指南 - **代理池设计**（参考 MediaCrawler ProxyIpPool 设计）：代理池抽象接口设计、代理获取和存储、代理有效性检测、代理评分和淘汰机制、ProxyRefreshMixin 刷新机制 - **代理使用模式**：每次请求更换代理、按 IP 绑定代理（Session 粘性）、隧道代理的使用、API 提取型代理的封装 - **代理与异步请求集成**：httpx 设置代理、代理超时和重试策略、代理切换的优雅实现 **实战案例：** 实现一个完整的代理池管理类，支持从多个代理源获取代理、自动检测有效性、按需分配代理，并与爬虫主程序集成。 --- ### [04_Playwright浏览器自动化入门](./04_Playwright浏览器自动化入门.md) **学习目标：** - 理解浏览器自动化的应用场景 - 掌握 Playwright 的安装和基本配置 - 学会页面导航、元素定位和交互操作 - 理解同步 API 和异步 API 的选择 - 掌握截图、PDF 导出等实用功能 **核心内容：** - **Playwright 概述**：与 Selenium/Puppeteer 的对比、支持的浏览器（Chromium/Firefox/WebKit）、同步 vs 异步 API 选择、安装和浏览器驱动管理 - **页面基础操作**：Browser/Context/Page 三层模型、页面导航和等待策略、元素定位方式（CSS/XPath/Text/Role）、点击、输入、选择等交互操作 - **等待策略**（重要）：自动等待机制、waitForSelector / waitForLoadState、自定义等待条件、超时处理 - **页面内容提取**：获取元素文本和属性、执行 JavaScript 获取数据、拦截网络请求获取 API 数据、截图和 PDF 导出 - **浏览器配置**：有头模式 vs 无头模式、视口大小设置、用户数据目录持久化、代理设置 **实战案例：** 使用 Playwright 爬取一个需要 JavaScript 渲染的 SPA 单页应用，提取动态加载的数据列表。 --- ### [05_Playwright进阶_反检测与性能优化](./05_Playwright进阶_反检测与性能优化.md) **学习目标：** - 理解浏览器指纹检测原理 - 掌握 stealth.min.js 反检测注入 - 学会 CDP 模式的使用场景 - 掌握多浏览器实例管理和资源优化 - 实现浏览器上下文复用策略 **核心内容：** - **浏览器指纹检测**：navigator 属性检测、WebGL/Canvas 指纹、WebDriver 特征检测、行为特征检测 - **stealth.js 反检测**（MediaCrawler 核心技术）：stealth.min.js 的工作原理、注入时机和方式、常见检测点的绕过、自定义反检测脚本 - **CDP 模式深入**：什么是 Chrome DevTools Protocol、标准模式 vs CDP 模式对比、连接已有浏览器实例、CDP 命令直接调用 - **性能优化**：禁用图片/CSS/字体加载、浏览器上下文复用、多页面并发管理、内存和资源监控、优雅关闭和资源释放 - **异常处理**：页面崩溃检测和恢复、超时重试策略、浏览器进程管理 **实战案例：** 使用 Playwright + stealth.js 爬取一个具有严格反爬检测的网站，对比有无反检测的效果差异。 --- ### [06_登录认证_Cookie与Session管理](./06_登录认证_Cookie与Session管理.md) **学习目标：** - 深入理解 Cookie 和 Session 的工作机制 - 掌握 Cookie 的提取、存储和注入 - 学会检测登录状态和 Cookie 有效性 - 实现 Cookie 的自动刷新机制 - 理解不同登录场景的处理策略 **核心内容：** - **认证机制深入**：Cookie 的属性详解（Domain/Path/Expires/HttpOnly/Secure）、Session 的服务端实现、Token 认证的工作流程、多设备登录和踢出策略 - **Cookie 提取与存储**：从浏览器 DevTools 手动提取、使用 Playwright 自动提取、Cookie 的序列化格式（JSON/Netscape）、加密存储敏感 Cookie - **Cookie 注入与使用**：httpx 中设置 Cookie、Playwright 中注入 Cookie、Cookie 的作用域管理、多账号 Cookie 轮换 - **登录状态管理**：登录状态检测方法、Cookie 过期检测、自动重新登录机制、登录状态持久化 **实战案例：** 实现一个完整的 Cookie 管理模块，支持 Cookie 的存储、加载、验证和自动刷新，可用于需要登录的爬虫场景。 --- ### [07_登录认证_扫码与短信登录实现](./07_登录认证_扫码与短信登录实现.md) **学习目标：** - 理解扫码登录的技术原理 - 掌握使用 Playwright 实现扫码登录流程 - 理解短信验证码登录的实现方式 - 学会登录状态的监控和回调机制 - 实现多种登录方式的统一封装 **核心内容：** - **扫码登录原理**（参考 MediaCrawler）：二维码生成和轮询机制、长轮询 vs WebSocket 状态推送、扫码状态（待扫描/已扫描/已确认/已过期）、登录成功后的 Cookie 获取 - **Playwright 实现扫码登录**：定位二维码元素、二维码截图和保存、终端显示二维码（可选）、轮询检测登录状态、登录成功后 Cookie 提取 - **短信验证码登录**：手机号输入和验证码发送、验证码输入（手动/API 接收）、滑块验证码的处理、登录失败的重试策略 - **登录模块封装**：登录方式抽象（工厂模式）、统一的登录接口设计、登录状态回调机制、异常处理和超时控制 **实战案例：** 参考 MediaCrawler 的登录实现，完成一个支持扫码登录和 Cookie 注入的双模式登录模块。 --- ### [08_验证码识别与处理](./08_验证码识别与处理.md) **学习目标：** - 了解常见验证码类型和破解思路 - 掌握图片验证码的 OCR 识别 - 学会滑块验证码的轨迹模拟 - 了解第三方打码平台的使用 - 理解验证码处理的合规边界 **核心内容：** - **验证码类型概览**：图片字符验证码、数学运算验证码、滑块拼图验证码、点选验证码、行为验证码（如 reCAPTCHA） - **图片验证码识别**：ddddocr 库的使用、图片预处理（二值化/降噪）、本地模型 vs 云端 API、识别率优化策略 - **滑块验证码处理**：缺口位置识别（OpenCV/图像差异）、人类拖拽轨迹模拟、速度和加速度曲线、Playwright 实现拖拽操作 - **第三方打码平台**：平台选择和价格对比、API 调用封装、超时和失败重试、成本控制策略 - **合规与伦理**：验证码绕过的法律边界、合理使用原则、替代方案考虑 **实战案例：** 实现一个验证码处理模块，支持图片验证码 OCR 识别和简单滑块验证码的自动处理。 --- ### [09_数据清洗与预处理](./09_数据清洗与预处理.md) **学习目标：** - 掌握常见的数据清洗技术 - 学会使用正则表达式提取和清洗数据 - 理解数据去重和合并策略 - 掌握数据格式标准化方法 - 学会处理缺失值和异常值 **核心内容：** - **数据清洗概述**：脏数据的常见类型、数据质量评估指标、清洗流程设计 - **文本清洗**：HTML 标签移除、空白字符处理、特殊字符清理、编码问题处理（Unicode 归一化） - **正则表达式高级应用**：复杂模式匹配、命名分组提取、替换和转换、性能优化 - **数据去重与合并**：精确去重 vs 模糊去重、相似度计算（编辑距离/余弦相似度）、多数据源合并策略、冲突解决规则 - **数据标准化**：日期时间格式统一、数值单位换算、枚举值映射、字段命名规范 **实战案例：** 对前几章爬取的数据进行完整的清洗流程，包括文本清洗、格式标准化、去重处理，并输出干净的数据集。 --- ### [10_数据分析与可视化](./10_数据分析与可视化.md) **学习目标：** - 掌握 pandas 进行数据统计分析 - 学会生成词云展示文本数据 - 了解常用的数据可视化库（matplotlib/pyecharts） - 实现简单的数据报告生成 - 理解爬取数据的分析价值 **核心内容：** - **pandas 数据分析**：DataFrame 基础操作、数据聚合与分组统计、数据透视表、时间序列分析 - **词云生成**（参考 MediaCrawler）：jieba 中文分词、停用词过滤、wordcloud 库使用、自定义词云形状和配色 - **数据可视化**：matplotlib 基础图表（折线图/柱状图/饼图）、pyecharts 交互式图表、数据仪表盘设计、图表导出和嵌入 - **分析报告生成**：Markdown 报告模板、自动化报告生成、图表嵌入、定期报告推送 **实战案例：** 对爬取的社交媒体评论数据进行分析，生成词云、情感倾向统计图表，并输出一份完整的数据分析报告。 --- ### [11_进阶综合实战项目](./11_进阶综合实战项目.md) **学习目标：** - 综合运用进阶教程所学的所有技术 - 实现一个完整的中等复杂度爬虫项目 - 掌握爬虫项目的完整开发流程 - 学会项目文档编写和部署 **核心内容：** - **项目需求分析**：目标网站分析、数据需求定义、技术方案选型、风险评估 - **项目架构设计**：模块划分、类图设计、数据流设计、接口定义 - **核心功能实现**：登录模块（Cookie 注入 + 扫码登录）、爬取模块（Playwright + 反检测）、数据存储（多后端支持）、代理池集成、日志和监控 - **项目打包与部署**：代码结构整理、依赖管理（requirements.txt / pyproject.toml）、Docker 容器化部署、定时任务配置 **实战案例：** 完成一个类似 MediaCrawler 简化版的社交媒体数据采集工具，支持： - 多种登录方式 - 反检测浏览器自动化 - 代理 IP 轮换 - 多格式数据存储 - 词云生成 --- ## 源代码目录结构 ``` 源代码/爬虫进阶/ ├── 01_工程化爬虫开发规范/ │ ├── config/ │ ├── logger_demo.py │ ├── exception_demo.py │ └── refactored_crawler/ ├── 02_反爬虫对抗基础_请求伪装/ │ ├── ua_rotator.py │ ├── headers_builder.py │ └── rate_limiter.py ├── 03_代理IP的使用与管理/ │ ├── proxy_pool/ │ ├── proxy_checker.py │ └── proxy_demo.py ├── 04_Playwright浏览器自动化入门/ │ ├── basic_operations.py │ ├── wait_strategies.py │ └── spa_crawler.py ├── 05_Playwright进阶_反检测与性能优化/ │ ├── stealth_demo.py │ ├── cdp_mode.py │ └── performance_optimization.py ├── 06_登录认证_Cookie与Session管理/ │ ├── cookie_manager.py │ ├── session_demo.py │ └── login_state_checker.py ├── 07_登录认证_扫码与短信登录实现/ │ ├── qrcode_login.py │ ├── sms_login.py │ └── login_factory.py ├── 08_验证码识别与处理/ │ ├── ocr_captcha.py │ ├── slider_captcha.py │ └── captcha_service.py ├── 09_数据清洗与预处理/ │ ├── text_cleaner.py │ ├── deduplication.py │ └── data_normalizer.py ├── 10_数据分析与可视化/ │ ├── pandas_analysis.py │ ├── wordcloud_generator.py │ └── chart_demo.py └── 11_进阶综合实战项目/ ├── config/ ├── core/ ├── login/ ├── crawler/ ├── store/ ├── proxy/ ├── analysis/ └── main.py ``` --- ## 教程特色 1. **与入门教程无缝衔接**：延续异步编程风格，逐步提升难度 2. **源于实战项目**：核心技术均来自 MediaCrawler 项目的生产实践 3. **每章都有代码**：提供完整可运行的示例代码和实战项目 4. **注重工程规范**：从第一章就强调代码质量和可维护性 5. **循序渐进**：从请求伪装到浏览器自动化，再到综合实战 --- ## 学习建议 1. **按顺序学习**：每章内容都有前后依赖，建议按顺序学习 2. **动手实践**：每章的实战案例务必亲自完成 3. **阅读源码**：参考 MediaCrawler 项目源码加深理解 4. **遵守法规**：爬虫技术仅用于学习研究，请遵守相关法律法规 --- ### Overview (`源代码/爬虫进阶/README.md`) # Python 爬虫进阶教程 - 源代码 本目录包含11个进阶章节的完整示例代码，每个章节都是独立的 uv 项目。 ## 🚀 快速开始 每个章节目录都配置了独立的 `pyproject.toml`，可以使用 uv 快速安装依赖： ```bash # 进入任意章节目录 cd 01_工程化爬虫开发规范 # 安装依赖 uv sync # 运行示例代码 uv run python main.py ``` ## 📚 章节目录 ### [01_工程化爬虫开发规范](./01_工程化爬虫开发规范/) **技术栈**：httpx, pydantic, parsel, loguru **内容**：日志系统、配置管理、异常处理、项目结构规范 ### [02_反爬虫对抗基础_请求伪装](./02_反爬虫对抗基础_请求伪装/) **技术栈**：httpx, loguru, fake-useragent **内容**：User-Agent轮换、请求头伪装、速率控制、TLS指纹 ### [03_代理IP的使用与管理](./03_代理IP的使用与管理/) **技术栈**：httpx, loguru **内容**：代理池设计、代理检测、多URL测试、代理轮换 **新增**：✨ 添加了 ipify.org 和 ip-api.com 多URL测试 ### [04_Playwright浏览器自动化入门](./04_Playwright浏览器自动化入门/) **技术栈**：playwright, loguru **内容**：页面操作、元素定位、等待策略、SPA爬取 **目标网站**：quotes.toscrape.com ### [05_Playwright进阶_反检测与性能优化](./05_Playwright进阶_反检测与性能优化/) **技术栈**：playwright, loguru **内容**：stealth.js注入、CDP模式、资源拦截、上下文池 **测试网站**：bot.sannysoft.com ### [06_登录认证_Cookie与Session管理](./06_登录认证_Cookie与Session管理/) **技术栈**：httpx, loguru **内容**：Cookie管理、登录状态检测、Cookie轮换 **新增**：✨ 添加了 quotes.toscrape.com 真实登录演示 ### [07_登录认证_扫码与短信登录实现](./07_登录认证_扫码与短信登录实现/) **技术栈**：playwright, loguru **内容**：扫码登录框架、短信登录框架、登录工厂模式 **说明**：⚠️ 技术框架示例，需根据实际网站适配 ### [08_验证码识别与处理](./08_验证码识别与处理/) **技术栈**：loguru, pillow, ddddocr(可选), opencv(可选) **内容**：图片验证码OCR、滑块验证码、人类轨迹生成 ### [09_数据清洗与预处理](./09_数据清洗与预处理/) **技术栈**：Python标准库 **内容**：文本清洗、数据标准化、精确/模糊去重 ### [10_数据分析与可视化](./10_数据分析与可视化/) **技术栈**：pandas, jieba, wordcloud, matplotlib **内容**：数据统计分析、词云生成、图表绘制 ### [11_进阶综合实战项目](./11_进阶综合实战项目/) ⭐ **技术栈**：playwright, httpx, pydantic, pandas, wordcloud **内容**：完整的电商数据采集工具 **目标网站**：✅ books.toscrape.com（专门的爬虫练习网站） **更新**：✨ 已替换为真实可用的练习网站 ## 📦 依赖安装 ### 基础用法 ```bash # 1. 进入章节目录 cd 01_工程化爬虫开发规范 # 2. 安装依赖 uv sync # 3. 运行代码 uv run python main.py ``` ### 可选依赖 某些章节提供了可选功能，可以按需安装： ```bash # 章节02 - 安装TLS指纹伪装 uv sync --extra advanced # 章节08 - 安装所有验证码识别功能 uv sync --extra all # 章节10 - 安装交互式图表功能 uv sync --extra interactive ``` ### Playwright 特别说明 使用 Playwright 的章节（04, 05, 07, 11）需要额外安装浏览器驱动： ```bash uv sync uv run playwright install chromium ``` ## 🎯 重点更新说明 ### ✨ 章节03 - 代理IP管理 新增了多URL测试功能（`demo_multi_url_test()`）： - httpbin.org/ip - 通用HTTP测试 - api.ipify.org - 专门IP服务 - ip-api.com/json/ - IP地理位置信息 ### ✨ 章节06 - Cookie管理 新增了真实登录演示（`demo_real_login()`）： - 演示 quotes.toscrape.com 完整登录流程 - Cookie保存和复用 - CookieManager集成 ### ✨ 章节11 - 综合实战项目 **重大更新**：将 example.com 替换为真实可用的练习网站： - **新目标**：http://books.toscrape.com - **特点**：50页书籍数据，1000+条记录 - **字段**：标题、价格、评分、库存、链接 - **状态**：✅ 代码可直接运行 ## 📖 使用建议 1. **按顺序学习**：章节之间有递进关系，建议按顺序学习 2. **动手实践**：每章都提供了可运行的示例代码 3. **查看文档**：每个章节的 README.md 提供了详细说明 4. **遵守规范**：爬虫技术仅用于学习研究，请遵守法律法规 ## 🛠️ 系统要求 - **Python**：>= 3.11 - **uv**：最新版本（用于依赖管理） - **操作系统**：支持 macOS / Linux / Windows ## 📝 常见问题 ### Q: 如何安装 uv？ ```bash # macOS/Linux curl -LsSf https://astral.sh/uv/install.sh | sh # Windows powershell -c "irm https://astral.sh/uv/install.ps1 | iex" ``` ### Q: Playwright 浏览器安装失败？ ```bash # 单独安装 Chromium uv run playwright install chromium # 如果网络问题，可以设置镜像 export PLAYWRIGHT_DOWNLOAD_HOST=https://npmmirror.com/mirrors/playwright/ uv run playwright install chromium ``` ### Q: 某些库无法安装？ 部分库可能需要系统级依赖，请查看各章节 README 的说明。 ## 🤝 贡献 欢迎提交 Issue 和 Pull Request 改进教程代码！ ## 📄 许可证 本教程代码仅供学习参考使用。 --- ### Overview (`源代码/爬虫进阶/01_工程化爬虫开发规范/README.md`) # 第01章：工程化爬虫开发规范 展示日志系统、配置管理、异常处理等工程化实践。 ## 快速开始 ### 使用 uv 安装依赖 ```bash # 进入本章目录 cd 01_工程化爬虫开发规范 # 安装依赖 uv sync # 运行示例 uv run python logger_demo.py uv run python exception_demo.py uv run python refactored_crawler/main.py ``` ### 核心依赖 - `httpx` - 异步HTTP客户端 - `pydantic` - 数据验证 - `parsel` - HTML解析 - `loguru` - 日志系统 ## 主要文件 - `logger_demo.py` - 日志系统演示 - `exception_demo.py` - 异常处理演示 - `refactored_crawler/` - 工程化改造后的完整爬虫项目 --- ### Overview (`源代码/爬虫进阶/02_反爬虫对抗基础_请求伪装/README.md`) # 第02章：反爬虫对抗基础 - 请求伪装 展示User-Agent轮换、请求头伪装、速率控制等反爬技术。 ## 快速开始 ### 使用 uv 安装依赖 ```bash cd 02_反爬虫对抗基础_请求伪装 # 安装基础依赖 uv sync # 安装高级功能（TLS指纹伪装，可选） uv sync --extra advanced # 运行示例 uv run python ua_rotator.py uv run python headers_builder.py uv run python rate_limiter.py uv run python anti_detection_crawler.py ``` ### 核心依赖 - `httpx` - 异步HTTP客户端 - `loguru` - 日志系统 - `fake-useragent` - User-Agent生成 - `curl-cffi`（可选）- TLS指纹伪装 ## 主要文件 - `ua_rotator.py` - User-Agent轮换器 - `headers_builder.py` - 请求头构建器 - `rate_limiter.py` - 速率限制器 - `anti_detection_crawler.py` - 完整反检测爬虫示例 --- ### Overview (`源代码/爬虫进阶/03_代理IP的使用与管理/README.md`) # 第03章：代理IP的使用与管理 展示代理池设计、代理检测、多URL测试等代理管理功能。 ## 快速开始 ```bash cd 03_代理IP的使用与管理 uv sync uv run python proxy_demo.py ``` ### 核心依赖 - `httpx` - HTTP客户端 - `loguru` - 日志系统 ## 测试URL - `httpbin.org/ip` - IP检测 - `api.ipify.org` - IP服务 - `ip-api.com/json/` - IP地理位置 --- ### Overview (`源代码/爬虫进阶/04_Playwright浏览器自动化入门/README.md`) # 第04章：Playwright浏览器自动化入门 展示Playwright基本操作、元素定位、等待策略等。 ## 快速开始 ```bash cd 04_Playwright浏览器自动化入门 uv sync uv run playwright install chromium # 安装浏览器 uv run python basic_operations.py uv run python wait_strategies.py uv run python spa_crawler.py ``` ### 目标网站 - **quotes.toscrape.com** - 静态版本 - **quotes.toscrape.com/js/** - SPA版本（需要JavaScript渲染） --- ### Overview (`源代码/爬虫进阶/05_Playwright进阶_反检测与性能优化/README.md`) # 第05章：Playwright进阶 - 反检测与性能优化 展示stealth.js注入、CDP模式、浏览器指纹伪装、性能优化等高级技术。 ## 快速开始 ```bash cd 05_Playwright进阶_反检测与性能优化 uv sync uv run playwright install chromium uv run python stealth_demo.py uv run python cdp_mode.py uv run python performance_optimization.py ``` ### 测试网站 - **bot.sannysoft.com** - 最佳的反检测测试网站 ### 核心技术 - stealth.js注入 - 隐藏自动化特征 - CDP模式 - 直接调用Chrome DevTools Protocol - 资源拦截 - 阻止不必要的资源加载 - 浏览器上下文池 - 复用提升性能 --- ### Overview (`源代码/爬虫进阶/06_登录认证_Cookie与Session管理/README.md`) # 第06章：登录认证 - Cookie与Session管理 展示Cookie管理、登录状态检测、真实网站登录演示。 ## 快速开始 ```bash cd 06_登录认证_Cookie与Session管理 uv sync uv run python cookie_manager.py uv run python login_state_checker.py uv run python session_demo.py # 包含 quotes.toscrape.com 真实登录演示 ``` ### 新增功能 ✨ **真实登录演示**：`session_demo.py` 中的 `demo_real_login()` 函数演示了完整的 quotes.toscrape.com 登录流程。 ### 核心依赖 - `httpx` - HTTP客户端 - `loguru` - 日志系统 - `cryptography`（可选）- Cookie加密 --- ### Overview (`源代码/爬虫进阶/07_登录认证_扫码与短信登录实现/README.md`) # 第07章：登录认证 - 扫码与短信登录实现 展示扫码登录和短信登录的技术框架和实现思路。 ## 快速开始 ```bash cd 07_登录认证_扫码与短信登录实现 # 安装基础依赖 uv sync # 安装二维码功能（可选） uv sync --extra qrcode # 安装浏览器 uv run playwright install chromium # 运行示例 uv run python qrcode_login.py uv run python sms_login.py uv run python login_factory.py ``` ### 重要说明 ⚠️ 本章代码提供的是**技术框架**而非特定网站实战，因为没有公开的练习网站提供真实扫码/短信登录功能。 代码展示了： - 扫码登录的技术原理和流程 - 短信登录的实现模式 - 登录工厂模式设计 实际使用时需要根据目标网站适配具体的： - 二维码选择器 - 状态轮询机制 - Cookie获取方式 --- ### Overview (`源代码/爬虫进阶/08_验证码识别与处理/README.md`) # 第08章：验证码识别与处理 展示图片验证码OCR识别、滑块验证码处理、人类轨迹生成等。 ## 快速开始 ```bash cd 08_验证码识别与处理 # 安装基础依赖 uv sync # 安装OCR功能 uv sync --extra ocr # 安装图像处理功能 uv sync --extra cv # 安装验证码生成功能 uv sync --extra generate # 安装轨迹可视化功能 uv sync --extra viz # 或安装所有功能 uv sync --extra all # 运行示例 uv run python ocr_captcha.py uv run python slider_captcha.py ``` ### 可选依赖说明 - `ddddocr` - OCR识别引擎 - `opencv-python` - 图像处理（滑块验证码） - `captcha` - 本地验证码生成 - `matplotlib` - 轨迹可视化 --- ### Overview (`源代码/爬虫进阶/09_数据清洗与预处理/README.md`) # 第09章：数据清洗与预处理 展示文本清洗、数据标准化、去重算法等数据处理技术。 ## 快速开始 ```bash cd 09_数据清洗与预处理 # 安装依赖（本章主要使用Python标准库） uv sync # 安装SimHash去重功能（可选） uv sync --extra simhash # 运行示例 uv run python text_cleaner.py uv run python data_normalizer.py uv run python deduplication.py ``` ### 主要功能 - **文本清洗** - HTML标签移除 - 空白字符处理 - 特殊字符清理 - 编码问题修复 - **数据标准化** - 日期时间格式统一 - 数值单位换算 - 文本规范化 - **去重算法** - 精确去重 - 模糊去重（编辑距离、Jaccard相似度） - SimHash去重（大规模文本） ### 依赖说明 本章主要使用Python标准库，`simhash` 为可选依赖，用于大规模文本去重。 --- ### Overview (`源代码/爬虫进阶/10_数据分析与可视化/README.md`) # 第10章：数据分析与可视化 展示Pandas数据分析、词云生成、图表制作等数据处理功能。 ## 快速开始 ```bash cd 10_数据分析与可视化 # 安装基础依赖 uv sync # 安装交互式图表功能（可选） uv sync --extra interactive # 运行示例 uv run python pandas_analysis.py uv run python wordcloud_generator.py uv run python chart_demo.py ``` ### 核心依赖 - `pandas` - 数据分析 - `jieba` - 中文分词 - `wordcloud` - 词云生成 - `matplotlib` - 图表绘制 - `pyecharts`（可选）- 交互式图表 --- ### Overview (`源代码/爬虫进阶/11_进阶综合实战项目/README.md`) # 第11章：进阶综合实战项目 完整的书籍电商数据采集工具，爬取 **books.toscrape.com**（专门的爬虫练习网站）。 ## 快速开始 ### 使用 uv 安装依赖 ```bash cd 11_进阶综合实战项目 # 安装依赖 uv sync # 安装可选功能（配置管理增强） uv sync --extra all # 安装Playwright浏览器驱动 uv run playwright install chromium # 运行项目 uv run python main.py ``` ### 目标网站 - **网站**：http://books.toscrape.com - **类型**：专门用于爬虫练习的合法网站 - **特点**：电商结构完整，50页书籍数据，无需登录 ### 核心依赖 - `playwright` - 浏览器自动化 - `httpx` - HTTP客户端 - `pydantic` - 数据验证 - `loguru` - 日志系统 - `pandas` - 数据分析 - `jieba` + `wordcloud` - 词云生成 ## 项目结构 ``` 11_进阶综合实战项目/ ├── config/ # 配置模块 ├── core/ # 核心模块（浏览器管理） ├── login/ # 登录模块 ├── crawler/ # 爬虫模块 ├── store/ # 存储模块 ├── proxy/ # 代理池模块 ├── analysis/ # 分析模块 └── main.py # 主程序入口 ``` ## 功能特性 - ✅ 浏览器自动化采集 - ✅ 反检测技术（stealth.js） - ✅ 多格式数据存储（JSON/CSV） - ✅ 词云和统计报告生成 - ✅ 代理池支持（可选） ## 运行结果 运行成功后会在 `output/` 目录下生成： - `data_*.json` 或 `data_*.csv` - 采集的数据 - `report.md` - 数据分析报告 - `wordcloud.png` - 词云图片 --- ### Index (`docs/index.md`) # 关于作者 大家好，我是程序员阿江-Relakkes，近期我会给大家出一些爬虫方面的教程，爬虫入门、进阶、高级都有，有需要的朋友，star仓库并持续关注本仓库的更新。 - [Github万星开源自媒体爬虫仓库MediaCrawler作者](https://github.com/NanmiCoder/MediaCrawler) - 全栈程序员，熟悉Python、Golang、JavaScript，工作中主要用Golang。 - 曾经主导并参与过百万级爬虫采集系统架构设计与编码 - 爬虫是一种技术兴趣爱好，参与爬虫有一种对抗的感觉，越难越兴奋。 ## 爬虫入门 ### 爬虫入门教程目录大纲 - [✔] [01_为什么要写这个爬虫教程](爬虫入门/01_为什么要写这个爬虫教程.md) - [✔] [02_个人学会爬虫能赚钱吗](爬虫入门/02_个人学会爬虫能赚钱吗.md) - [✔] [03_网络爬虫到底是什么](爬虫入门/03_网络爬虫到底是什么.md) - [✔] [04_爬虫的基本工作原理](爬虫入门/04_爬虫的基本工作原理.md) - [✔] [05_常用的抓包工具有那些](爬虫入门/05_常用的抓包工具有那些.md) - [✔] [06_Python写爬虫的优势](爬虫入门/06_为什么说用Python写爬虫有天生优势.md) - [✔] [07_Python常见的网络请求库](爬虫入门/07_Python常见的网络请求库.md) - [✔] [08_入门实战1_静态网页数据提取](爬虫入门/08_爬虫入门实战1_静态网页数据提取.md) - [✔] [09_入门实战2_动态数据提取](爬虫入门/09_爬虫入门实战2_动态数据提取.md) - [✔] [10_入门实战3_数据存储实现](爬虫入门/10_爬虫入门实战3_数据存储实现.md) - [✔] [11_入门实战4_高效率的爬虫实现](爬虫入门/11_爬虫入门实战4_高效率的爬虫实现.md) - [✖] [12_入门实战5_编写易于维护的爬虫代码](爬虫入门/12_爬虫入门实战5_编写易于维护的爬虫代码.md) ## 爬虫进阶 ### 爬虫进阶教程目录大纲 - [✔] [01_工程化爬虫开发规范](爬虫进价/01_工程化爬虫开发规范.md) - [✔] [02_反爬虫对抗基础_请求伪装](爬虫进价/02_反爬虫对抗基础_请求伪装.md) - [✔] [03_代理IP的使用与管理](爬虫进价/03_代理IP的使用与管理.md) - [✔] [04_Playwright浏览器自动化入门](爬虫进价/04_Playwright浏览器自动化入门.md) - [✔] [05_Playwright进阶_反检测与性能优化](爬虫进价/05_Playwright进阶_反检测与性能优化.md) - [✔] [06_登录认证_Cookie与Session管理](爬虫进价/06_登录认证_Cookie与Session管理.md) - [✔] [07_登录认证_扫码与短信登录实现](爬虫进价/07_登录认证_扫码与短信登录实现.md) - [✔] [08_验证码识别与处理](爬虫进价/08_验证码识别与处理.md) - [✔] [09_数据清洗与预处理](爬虫进价/09_数据清洗与预处理.md) - [✔] [10_数据分析与可视化](爬虫进价/10_数据分析与可视化.md) - [✔] [11_进阶综合实战项目](爬虫进价/11_进阶综合实战项目.md) ## 高级爬虫 ### 高级爬虫教程目录大纲 - [✖] 待更新... ## 爬虫交流群 扫码加作者企微拉进群，备注来自github爬虫教程 企微二维码 ## 免责声明 >本仓库的所有内容仅供学习和参考之用，禁止用于商业用途。任何人或组织不得将本仓库的内容用于非法用途或侵犯他人合法权益。本仓库所涉及的爬虫技术仅用于学习和研究，不得用于对其他平台进行大规模爬虫或其他非法行为。对于因使用本仓库内容而引起的任何法律责任，本仓库不承担任何责任。使用本仓库的内容即表示您同意本免责声明的所有条款和条件。 ## Star History [](https://star-history.com/#NanmiCoder/CrawlerTutorial&Date) --- ### 爬虫入门 01 为什么要写这个爬虫教程 (`docs/爬虫入门/01_为什么要写这个爬虫教程.md`) # 为什么写这个爬虫教程？ - 满足需求：我的[自媒体平台爬虫](https://github.com/NanmiCoder/MediaCrawler)爆火了之后，越来越多人私信我都是问想要学习爬虫时，我意识到了有很多人对这个爬虫领域感兴趣，但却不知道从何入手。因此决定撰写这个爬虫教程。 - 分享经验：首先我这个人虽然技术一般，但是真的很喜欢分享，平时工作中我学到了什么新知识都迫不及待的想要跟身边同事去分享，所以既然大家有需要，那么我会分享自己在爬虫领域的一些经验知识和见解。 - 激发兴趣：希望通过我的教程学习能够激发更多人对编程和数据获取的兴趣，让大家探索更多的可能。 ## 爬虫是一个综合性的技能 爬虫工程师需要具备广泛的技能，包括前端开发（HTML、JavaScript、浏览器和APP抓包）、HTTP协议知识，简单数据挖掘技能（数据结构化、清洗、排重等工作）以及数据存储方面的知识。 熟练掌握这些技能可以让你成为一名合格的爬虫工程师，能够处理数百万个网页数据。然而，当需要处理上千万个网页数据时，你将需要不断优化存储方法、内存调配方式和抓取策略。在对抓取数据的实时性和数据量有极高要求时，爬虫工程师又需要像解决社会工程问题一样，处理大量分布的IP和账号问题。 ## 对初学者的一些建议 学爬虫不要被网络上的一堆名词吓到。如果你是一个初学者，你只需要多动手练习。从最简单的爬虫开始写起。 在编码和调试的过程中，你会遇到各种各样的问题。这时是你最好的学习和进阶时刻。通过解决这些问题，你会摸索出需要掌握哪些技巧。
记住，编程是一个不断学习和成长的过程，每个错误都是一个宝贵的教训，每次解决问题都是一次进步。
不要害怕犯错，勇敢地面对挑战，你会发现自己的能力在不断提升。坚持不懈，相信自己，你一定能成为优秀的程序员！ --- ### 爬虫入门 02 个人学会爬虫能赚钱吗 (`docs/爬虫入门/02_个人学会爬虫能赚钱吗.md`) # 个人学会爬虫能赚钱吗？ **先说结论** >肯定可以赚钱，关键在于挣得多还是少。掌握任何一门技术，都有赚钱的机会，因为技术需求一直存在。 > >举个例子，我在22年前后，专门在某鱼平台发布了许多Python爬虫相关的商品，吸引了很多咨询者。最高峰时，这些小项目带来的额外收入一个月超过1万元，但也带来了疲劳，每天工作到凌晨1-2点才能休息。 > >可能有人会认为我在吹嘘，但事实上，我确实通过某鱼平台挣了一些钱。现在我用的MacBook Air M1电脑就是当时接Python项目挣来的。如果感兴趣，可以访问我的某鱼主页，了解我的100%好评率，以及我发布的商品或已售出的Python项目。 > >总的来说，我想强调一点：无论是爬虫还是其他技术，只要你能满足他人需求，解决他们无法解决的问题，你就能赚钱。 爬虫挣钱的分类我基于我个人的经验总结了一下，大致分为三类，`兼职小活`、`长期雇佣关系`、`专职爬虫`，下面对这三类都做一个简单介绍 ## 1、兼职小活 我们经常能在各种卖爬虫课直播间看到一些主播晒关于接单成交截图之类，其实这些截图我认为有真有假，可能大部分都是真的，因为市场是一个供需关系来的，有需求自然就有提供方，爬虫这个也不例外。 在现如今这个数据为王的时代。做事儿都需要有强有力的数据做支撑然后去做决策，而不是盲目的去做。我打个比方： 假如你是一个饮食类电商选品的负责人，现在你想要去选品然后上架到抖音或者淘宝之类的电商平台，那么你是不是会去先了解下竞品的销售数据情况、评论情况等等 那需求不就来了吗，有的人可能不了解有一些专门做电商数据的平台，他心里内心就自然而然的诞生一个想法，找个`IT高手`帮我去抓取一下我想关注的商品数据、 > 这类活我们可以把它统称为：兼职小活,特点：`客单价低`，`大多是一次性需求`，`花的时间相对少`。 ## 2、长期雇佣关系 长期雇佣关系这个从字面意思就比较好理解了，基于某一类固定需求，长期给需求方提供服务，这里我举一个我之前兼职遇到的客户吧 她是一个女生，她们公式是xxx相关的业务，她接到了领导的一个需求，需要定期出一份她们公司所在省份的xxx业务数据。 有目标站点，只要定期去目标站点上拉取数据，然后与本地数据做diff，只要增量数据，就这个需求，听起来很简单是不是？ 确实很简单，但这是针对于我们这一类会编程的人，那假如她不会编程呢，该怎么办，是不是要手动一个一个的去记录到Excel，然后自己check更新的部分？ 这个需求我就和她保持了长期的雇佣关系，每一次费用大概是100-300不等。（读到这里千万不要想歪了） 还有一种是你已经给他做了一些爬虫类的兼职小活，从效率、服务等各方面都觉得你还不错，那么下次有同样的需求的时候会优先找到你，所以做任何一行都要服务好～ > 这类活我们可以把它统称为：长期雇佣关系, 特点：`客单价有高有低`，`大多为数据导出清洗需求`，`回购 or 复购`。 ## 3、专职爬虫 这一类就比较直观了，就是找一个`爬虫工程师`的工作，工作内容一般为：数据采集、数据清洗、爬虫系统设计与架构。 薪水的话拿一线城市举个例子吧，初级的话大概在8-14k 中高级 15 - 30k，当然这是一个参考区间，有的可能更高，有的可能稍微少点。 > 这一类专职爬虫，通常对技术要求比较高，需要对技术知识积累覆盖面光，前端、后端、数据库、安全、逆向、网络都要去涉及。 > 当你覆盖了不同的技术知识积累，在面对不对的case时都能很快的给出解决方案，那么恭喜你你的薪水可能就会涨的很快。 --- ### 爬虫入门 03 网络爬虫到底是什么 (`docs/爬虫入门/03_网络爬虫到底是什么.md`) # 网络爬虫到底是什么？ ## 1、看看维基百科的定义 > 网络爬虫（英语：web crawler），也叫网络蜘蛛（spider），是一种用来自动浏览万维网的网络机器人。其目的一般为编纂网络索引。
> 网络搜索引擎等站点通过爬虫软件更新自身的网站内容或其对其他网站的索引。网络爬虫可以将自己所访问的页面保存下来，以便搜索引擎事后生成索引供用户搜索。
> 爬虫访问网站的过程会消耗目标系统资源。不少网络系统并不默许爬虫工作。因此在访问大量页面时，爬虫需要考虑到规划、负载，还需要讲“礼貌”。 不愿意被爬虫访问、被爬虫主人知晓的公开站点可以使用robots.txt文件之类的方法避免访问。这个文件可以要求机器人只对网站的一部分进行索引，或完全不作处理。
> 互联网上的页面极多，即使是最大的爬虫系统也无法做出完整的索引。因此在公元2000年之前的万维网出现初期，搜索引擎经常找不到多少相关结果。现在的搜索引擎在这方面已经进步很多，能够即刻给出高素质结果。
> 爬虫还可以验证超链接和HTML代码，用于网络抓取 ## 2、Google和百度搜索引擎是如何工作的呢？ - 首先，网络爬虫持续抓取网站内容，将其存储在搜索引擎的数据库中。 - 紧接着，索引程序对数据库中的网页进行整理，创建倒排索引。 - 最后，当用户输入查询关键词时，搜索程序会在索引中查找相关内容，并通过排序算法（例如Pagerank）将最相关的结果展现给用户。 这一过程虽然看似简单，却构建了一个复杂的搜索引擎系统。网络爬虫作为系统的基础部分，对数据的完整性和丰富性起着决定性作用。简而言之，网络爬虫是用于自动获取互联网上公开数据的工具。 重要的是要明确，网络爬虫抓取的是`公开数据`，并不涉及非法手段获取非公开数据。 所谓`公开数据`，指的是网站上公开供用户浏览和获取的信息。虽然这些数据是公开的，但当个人或机构通过大规模收集这些数据来获利时，可能会引起数据提供方的不满，甚至导致法律问题。例如，Google在早期因此曾面临诉讼。 为了解决这一问题，搜索引擎和网站之间形成了一种默契——robots.txt协议。网站通过这个文件指明哪些内容可以被爬虫抓取，哪些不可以；同时，搜索引擎在访问网站时会通过User-Agent标识自己的身份（如Googlebot、Baiduspider），以此保持双方的和平共处和互利共赢。 ## 3、爬虫Crawler的职责 ### 抓取页面（Fetching） 网络爬虫会按照一定的规则和算法，访问网站上的页面并下载页面内容。这个过程需要考虑页面的深度、频率、并发请求数量等因素，以确保高效地获取数据。 ### 解析页面（Parsing） 爬虫需要解析下载的页面内容，提取其中的文本、链接、图像等信息。通过解析页面，爬虫可以识别页面结构、内容特征以及与其他页面的关联。 ### 处理链接（Link Handling） 爬虫在解析页面时会提取页面中的链接，然后根据一定的策略处理这些链接。这包括去重、筛选、调度等操作，以确保爬虫系统能够高效地覆盖目标网站的内容。 ### 存储数据（Storing） 爬虫需要将抓取到的数据进行存储，通常存储在数据库或索引中。这样可以为后续的数据处理、索引建立和搜索提供支持。 ### 遵守规则（Respect Robots.txt） ：爬虫需要遵守网站所有者制定的规则，比如robots.txt文件中定义的爬取限制。爬虫需要尊重网站的爬取策略，以避免对目标网站造成不必要的干扰。 ## 4、总结 网络爬虫在信息检索系统中扮演着数据搜集和处理的关键角色。通过抓取、解析、处理链接、存储数据、更新索引等一系列操作，网络爬虫为搜索引擎提供了高效的数据支持，帮助用户快速准确地获取所需信息。 --- ### 爬虫入门 04 爬虫的基本工作原理 (`docs/爬虫入门/04_爬虫的基本工作原理.md`) # 网络爬虫工作原理详解 网络爬虫，也称为网页蜘蛛或网页爬虫，是一种自动浏览互联网并从中提取信息的程序。爬虫广泛应用于搜索引擎索引、数据挖掘、市场分析等领域。本文旨在深入探讨网络爬虫的工作原理，并介绍相关的技术和工具。 ## 爬虫的基本流程 > 近几年随着大家互联网冲浪的需求不断攀升，传统的那种后端渲染前端页面的方式已经满足不了快速迭代变化的需求了，所以基于前后端分离的应用越来越多.
> 大多数我们写的爬虫代码的目标站点都是这种前后端分离的应用，我建议大家可以去网上稍微瞄一下前后端分离的一些技术实现，看个大概就行，对这个前后端分离的这种应用有一些理解，在你后续写爬虫有一些帮助。 现代的网络爬虫工作流程大致可以分为以下几个步骤： 1. **识别入口点**：确定需要抓取数据的API入口点。这些入口点是爬虫开始抓取数据的起点，通常是一些返回JSON数据的URL。 2. **构造请求**：根据API文档或通过分析网络请求，构造出正确的HTTP/HTTPS请求。这包括正确的请求方法（GET、POST等）、请求头、以及必要的参数。 3. **发送请求**：爬虫对API入口点发起请求，等待服务器响应。对于需要认证的API，可能还需要处理登录和会话管理。 4. **解析响应**：服务器返回JSON或其他格式的数据。爬虫需要解析这些数据，提取有价值的信息。 5. **数据处理**：将提取出的数据进行清洗、转换和存储。数据可以存储在数据库、文件或其他存储系统中。 6. **遍历与递归**：从返回的数据中提取出新的URL或API入口点，重复上述过程，直至抓取到所需的全部数据。 ## 关键技术点 ### HTTP/HTTPS协议 > 不知道http和https协议的可以看下[MDN WEB对于HTTP的讲解](https://developer.mozilla.org/zh-CN/docs/Web/HTTP) 理解HTTP和HTTPS协议是开发网络爬虫的基础。这些协议定义了客户端和服务器之间如何交换数据。 ### 请求库 - **Requests**：Python的Requests库是处理HTTP请求的同步库，简单易用，适合初学者。 - **Aiohttp**：Aiohttp是一个支持异步请求的库，可以在处理大量并发连接时提高效率。 - **HTTPX**：HTTPX支持同步和异步请求，是一个现代化的网络请求库，提供了丰富的功能。 ### 解析库 > 自从我用了Parsel之后，我解析网页的库再也没用过其他的了，可以基于css选择器，也可以基于xpath，是从Scrapy框架中的解析库做了二次封装得来的，所以我把它推荐给你、 - **Parsel**：Parsel基于lxml，专为HTML/XML解析设计，简化了数据提取的流程。 ### 浏览器自动化测试工具 > 优先推荐Playwright，因为现在Python的异步编程很流行，那基于异步的爬虫代码也很多，Playwright它也是支持异步调用， 并且微软开源的，更新迭代速度还行，这也是我推荐的。 - **Playwright**：Playwright是一个由Microsoft开发的现代化浏览器自动化库。它支持所有主流浏览器和多种语言，特别适合高效率的自动化测试和爬虫开发。 - **Selenium**：Selenium是一个浏览器自动化工具，可以模拟用户操作浏览器。它支持多种编程语言和浏览器，适合处理JavaScript渲染的页面。 ## 结论 网络爬虫技术涉及广泛，从基本的HTTP/HTTPS协议到复杂的页面解析和浏览器自动化。掌握这些技术和工具，可以帮助开发者有效地从互联网上抓取数据。随着技术的发展，还会有更多先进的工具出现，为网络爬虫开发提供更多可能性。 --- ### 爬虫入门 05 常用的抓包工具有那些 (`docs/爬虫入门/05_常用的抓包工具有那些.md`) # 利用各种工具抓包网络请求 在进行网络爬虫开发时，抓包是一个非常重要的步骤。它可以帮助我们了解客户端和服务器之间的通信过程，包括请求的发送和响应的接收。这对于分析和模拟网络请求尤其关键。本教程将介绍几种常用的抓包工具，包括Chrome的开发者工具、Charles和Fiddler。
- Chrome抓包Web应用 - Charles和Fiddler既可以抓包APP也可以抓包Web ## 使用Chrome的开发者工具 > 我之前写了一篇博客微博爬虫的示例，其中有关于Chrome F12抓包的文章，感兴趣的可以去看下：[微博帖子评论爬取教程](https://blog.csdn.net/weixin_43252709/article/details/135431751) Chrome浏览器内置的开发者工具是最直接便捷的抓包方式之一，特别适合前后端分离的网站分析。 1. 打开Chrome浏览器，访问目标网站。 2. 按`F12`或右键点击页面，选择“检查”打开开发者工具。 3. 切换到“Network”（网络）标签页。此时可能需要刷新页面以捕获所有网络请求。 4. 浏览网络请求列表，点击任一请求查看详细信息，包括请求头、响应头、响应体等。 5. 可以通过过滤器筛选特定类型的请求，例如XHR（Ajax请求）。 6. 使用Chrome开发者工具的优点在于无需安装额外软件，操作简单，适合快速查看和分析HTTP/HTTPS请求。 ## 使用Charles > 当我有抓取APP或者小程序等需求的时候，我用charles比较多一些，因为它对Macos支持比较友好。
> 具体安装和使用示例可以看这篇文章，写的比较细：[charles安装入门使用示例](https://zhuanlan.zhihu.com/p/140942687) Charles是一款广受欢迎的跨平台抓包工具，它可以作为代理服务器运行，监控和修改进出电脑的所有HTTP和HTTPS请求。 1. 下载并安装Charles。 2. 启动Charles，它会自动开始捕获网络请求。 3. 配置浏览器或设备使用Charles为代理服务器。这通常涉及到设置代理服务器地址为127.0.0.1（本机地址），端口为Charles显示的端口（默认8888）。 4. 访问目标网站或应用，Charles会显示通过它的所有请求和响应。 5. 双击任一请求或响应以查看详细内容。 Charles强大之处在于它能够修改请求或响应，实现更深入的测试和分析。 ## 使用Fiddler > 这款工具在windows平台很火，我之前在上一家公司做一些爬虫的小需求有用过，整体尚可，但是好像已经开始收费了，免费版的功能又受限。
> 具体安装和使用示例可以看这篇文章，写的比较细：[Fiddler安装入门使用示例](https://blog.csdn.net/FourAu/article/details/136479512) Fiddler同样是一款功能强大的网络请求分析工具，它也可以捕获计算机上的HTTP/HTTPS请求。 1. 下载并安装Fiddler。 2. 启动Fiddler，它默认开始捕获网络请求。 3. 在“Web Sessions”窗口中，可以看到通过Fiddler的所有HTTP/HTTPS请求和响应。 4. 点击任一条目查看详细的请求和响应信息。 Fiddler提供了广泛的自定义选项，包括断点设置、请求编辑、性能测试等高级功能。 ## 总结 抓包是网络爬虫开发中不可或缺的一环。通过使用Chrome开发者工具、Charles、Fiddler等工具，我们可以有效地分析和理解客户端与服务器之间的通信过程。这些工具各有特点，选择合适的工具可以大大提高爬虫开发的效率和质量。 --- ### 爬虫入门 06 为什么说用Python写爬虫有天生优势 (`docs/爬虫入门/06_为什么说用Python写爬虫有天生优势.md`) # 为什么说用Python写爬虫有天生的优势 在网络爬虫的开发过程中，我们经常遇到的问题是：昨天还运行良好的爬虫，今天可能就因为目标网站的改版或封锁策略而出现问题。这种变幻莫测的情况要求我们能够快速定位问题并进行修复。在这个过程中，Python以其独特的优势成为了许多开发者的首选语言。
Python在网络爬虫领域的流行并非偶然。它以其简洁的语法、丰富的库支持、广泛的社区资源以及高效的开发效率，在众多编程语言中脱颖而出。以下是使用Python编写网络爬虫具有天生优势的几个关键原因。 > 我0基础入门计算机时，最最开始也是在使用Python写一些爬虫程序，那个时候python学了点基础，正好就拿爬虫练练手了，因为写爬虫程序比较直观，方便上手，容易看到结果，久而久之的慢慢的就熟悉python的语法了。我觉得爬虫真的算是是一个入门python编程的这门语言的一个最好炼金石了。 ## 简洁易学的语法 Python的语法简洁明了，极大地降低了编程的学习门槛，使得即使是编程新手也能快速入门。对于爬虫开发者来说，这意味着能够更加专注于逻辑实现而非语言细节，从而提高开发效率。 ## 丰富的库支持 Python提供了广泛的库来支持网络爬虫的开发： - **Requests**、**Aiohttp**、**Httpx**等HTTP库简化了网络请求的发送与处理。 - **BeautifulSoup**、**lxml**和**Parsel**等解析库提供了强大的文档解析能力。 - **Scrapy**框架为构建复杂的爬虫应用提供了一站式解决方案。 - **Pandas**等数据分析库方便对抓取到的数据进行处理和分析。 - **自动化测试工具**：如**Playwright**和**Selenium**，可以模拟浏览器行为，对付JavaScript渲染的页面。 这些库的存在极大地加速了爬虫项目的开发进程。 ## 广泛的社区资源 Python拥有一个庞大且活跃的社区。无论是遇到技术难题还是寻找特定的库，你都可以在社区中找到解答或者得到灵感。这种资源的丰富性为Python爬虫开发者提供了极大的便利。 ## 高效的开发效率 得益于简洁的语法和丰富的库支持，使用Python进行爬虫开发可以显著提高编码效率。从原型设计到最终实现，Python都能够帮助开发者快速迭代，缩短项目周期。 ## 易于维护和扩展 Python代码的可读性和简洁性使得维护和扩展变得更加容易。当爬虫项目需要增加新功能或者优化时，清晰的代码结构可以让其他开发者快速理解项目架构，有效地进行协作开发。 ## 跨平台性 Python是一种跨平台的编程语言，可以在多种操作系统上运行，如Windows、Linux和macOS。这意味着用Python编写的爬虫可以轻松适应不同的运行环境，增强了项目的可移植性。 总结来说，Python在网络爬虫领域之所以受到广泛欢迎，不仅因为它简单易学、开发高效，还因为它拥有强大的库支持、活跃的社区和良好的跨平台性能。这些因素共同构成了Python在网络爬虫开发中不可替代的地位。 --- ### 爬虫入门 07 Python常见的网络请求库 (`docs/爬虫入门/07_Python常见的网络请求库.md`) # 1. 常用的网络请求库 在Python中，进行网络请求的库主要分为同步和异步两大类。 - **同步请求库**: - `urllib`: Python的标准库之一，提供了一系列用于操作URL的功能。 - `requests`: 第三方库，提供了更加方便的API来发送HTTP请求，是最受欢迎的HTTP客户端库之一。 - **异步请求库**: - `aiohttp`: 支持异步请求的库，使用`asyncio`进行网络通信，适合处理高并发需求。 - `httpx`: 是一个全功能的HTTP客户端，支持HTTP/1.1和HTTP/2，并且同时支持同步和异步接口。 ## 2. 优缺点及适用场景 - `urllib`: - **优点**: 标准库，不需要额外安装。 - **缺点**: API相对繁琐。 - **适用场景**: 简单的应用，或不想引入外部依赖时。 - `requests`: - **优点**: API简单易用，社区支持强大。 - **缺点**: 不支持异步。 - **适用场景**: 大多数HTTP请求场景，尤其是对性能要求不是非常高的同步程序。 - `aiohttp`: - **优点**: 支持异步，适合高并发场景。 - **缺点**: API相对复杂。 - **适用场景**: 需要处理大量并发连接的应用。 - `httpx`: - **优点**: 同时支持同步和异步API，支持HTTP/2。 - **缺点**: 相对较新，社区支持和稳定性正在增强中。 - **适用场景**: 需要同时使用同步和异步请求，或需要HTTP/2支持的应用。 ## 3. Requests和httpx的使用 > headers、cookies、auth、proxy这几种是我们日常爬虫过程中，经常需要使用的，下面分别基于request和httpx来展示如何使用 - **Requests**: - **Headers**: ```python import requests response = requests.get('https://httpbin.org/get', headers={'User-Agent': 'My App'}) ``` - **Cookies**: ```python response = requests.get('https://httpbin.org/cookies', cookies={'session_id': '12345'}) ``` - **认证**: ```python response = requests.get('https://httpbin.org/basic-auth/user/passwd', auth=('user', 'passwd')) ``` - **SSL证书验证**: ```python response = requests.get('https://httpbin.org/get', verify='/path/to/certfile') ``` - **httpx**: - **Headers**: ```python import httpx response = httpx.get('https://httpbin.org/get', headers={'User-Agent': 'My App'}) ``` - **Cookies**: ```python client = httpx.Client() client.cookies.set('session_id', '12345') response = client.get('https://httpbin.org/cookies') ``` - **认证**: ```python response = httpx.get('https://httpbin.org/basic-auth/user/passwd', auth=('user', 'passwd')) ``` - **SSL证书验证**: ```python response = httpx.get('https://httpbin.org/get', verify='/path/to/certfile') ``` 细心的你可能发现了，httpx和request的用法大差不差，是的没错。 httpx 的设计灵感来源于 requests，因此两者在用法上有很多相似之处。这是因为 httpx 的开发者希望提供一个类似于 requests 的简洁、易用的接口，同时又能够支持更多的功能和特性，比如对异步请求的支持以及对 HTTP/2 的原生支持。因此，如果您熟悉 requests 的用法，那么学习和使用 httpx 会变得非常容易和顺畅。 ## 4. 入门示例 - **Requests示例**: ```python import requests response = requests.get('https://httpbin.org/get') print(response.json()) ``` - **httpx示例**: ```python import httpx response = httpx.get('https://httpbin.org/get') print(response.json()) ``` ## 5. 实际业务逻辑示例 假设我们需要实现一个功能：向`httpbin.org/post`发送POST请求，提交一些数据，并接收响应。 - **使用Requests**: ```python import requests data = {'name': '程序员阿江','email':'relakkes@gmail.com'} response = requests.post('https://httpbin.org/post', data=data) print(response.json()) ``` - **使用httpx（异步）**: ```python import httpx import asyncio async def post_data(): data = {'name': '程序员阿江','email':'relakkes@gmail.com'} async with httpx.AsyncClient() as client: response = await client.post('https://httpbin.org/post', data=data) print(response.json()) asyncio.run(post_data()) ``` ## 6. 异步爬虫的趋势 Python3.7之后，随着`asyncio`库的成熟和普及，异步编程在Python中变得更加容易实现。异步爬虫可以同时发起和管理成百上千的网络请求，而不会阻塞主线程。这使得编写高性能的爬虫代码不再是难事，尤其是在数据采集、实时数据处理等领域，异步爬虫将会成为一种趋势。 ## 7. 总结 在选择合适的网络请求库时，应考虑实际应用的需求：对于简单或不频繁的网络请求，可以选择`urllib`或`requests`；而在需要处理大量并发连接的场景下，则应考虑使用`aiohttp`或`httpx`。随着Python异步编程的发展，未来异步爬虫无疑会在性能和效率上带来更多优势。 --- ### 爬虫入门 08 爬虫入门实战1 静态网页数据提取 (`docs/爬虫入门/08_爬虫入门实战1_静态网页数据提取.md`) # 爬虫静态网页数据提取 > 为了照顾一些新入门的朋友，本篇的内容html内容解析会用两个库来完成，一个是`BeautifulSoup` 另一个是我比较喜欢用的`parsel`. 大多数新入门朋友可能学习爬虫的时候，都是从BeautifulSoup这个库开始的。 ## 什么是静态网页 静态网页是指内容固定不变的网页，它的内容是直接写在 HTML 文件中的，不会因为用户的请求或者其他因素而改变。静态网页的内容通常由 HTML、CSS 和 JavaScript 组成，服务器只需要将这些文件发送给浏览器，浏览器就可以直接解析并显示网页内容。 ## 静态网页工作原理 >当用户在浏览器中输入一个静态网页的 URL 时，浏览器会向服务器发送一个 HTTP 请求，请求获取该 URL 对应的 HTML 文件。服务器接收到请求后，会在服务器上查找对应的 HTML 文件，并将其内容发送给浏览器。浏览器接收到 HTML 文件后，会解析其中的 HTML、CSS 和 JavaScript 代码，并根据这些代码渲染出网页内容。 ## 爬取静态网页一般需要那些技术 - 会一点点前端的三件套（html、css、js）不会的朋友可以去菜鸟教程上面看一看，只需要简单的入门，知道html标签的一个结构，css选择器的简单用法，js的话暂时不太需要。 - 会使用网络请求库，比如requests、httpx等 - 会使用html解析库，比如BeautifulSoup、parsel等 - 会查找静态网页一个规律 - 存储方面的话看自己需求，如果需要存db这些，就需要自己去了解一些db方面的知识（可选） ## 实战示例 > 下面开始爬虫入门教程系列的第一次代码实战，前面7讲都在将一些理论知识，我们来看看如何将这些理论知识用于实践当中。
> 我的教程都会给大家写两个版本，一个同步请求版本，一个异步请求版本，可能大家在很多别人教程里看见的大多数都是使用requests + BeautifulSoup这一套。
> 我这边给大家再写一套异步的，我为什么很喜欢写爬虫代码喜欢用异步？ > 之前大家如果看过MediaCrawler的源码实现的话，可以看到我整个实现全是基于异步，从发请求、操作数据库、操作db，只要能异步化我都异步化了。 > - 1、这是一种趋势，python一些流行的web框架现在都在往异步方面靠，我们提前用爬虫代码练练手，为后续你可能从事python方面的后端工程师做一些准备 > - 2、性能真的很不错，之前那种多线程爬取的效率有点低，使用异步能在单进程单线内把资源利用发挥到极致。 ### 任务需求描述 > 由于合规信息要求，我们的案例大多都会选择一些不在国内的站点来作为爬虫目标站点，技术的原理是相通的。 今天我要爬取的是一个BBS论坛网站的股票讨论部分，目标站点地址：https://www.ptt.cc/bbs/Stock/index.html 需要采集前N页的信息，具体采集内容如下： - 前N页的帖子列表汇总 - 前N页的每一个帖子内容信息信息 - 前N页的每一个帖子的推文信息（可以理解为评论信息）获取 ### 技术可行性分析 #### 1、如何获取前N页中的最新分页Number？ 需求中说的是前N页帖子，那么我们是不是要从最新的帖子往前推N页就可以了，理论上我们只需要找出它最新的分页Number就可以了。 我们打开 `https://www.ptt.cc/bbs/Stock/index.html` 并点击`上一页` 按钮，从页面URL `https://www.ptt.cc/bbs/Stock/index7083.html` 可以得出 7083可能是分页Number 我们再点一次`上一页` 按钮，可以发现URL变为了：`https://www.ptt.cc/bbs/Stock/index7082.html`, 那么我们可以初步断定，这个网站的分页模式就是从高到低递减了。
我们如何知道7083这个分页数字？
静态网页一般找这个数字都不难，Chrome浏览器，F12，选中上一页按钮，从html文档中的elements就能看见这个按钮是一个a标签，其中的href属性放着点击该按钮之后要跳转的URL地址。 所以我们只需要使用解析库把这个数字解析出来就可以了 #### 2、html结构分析 > 我们拿帖子列表来做一个简单分析，帖子详情页的类似 同样F12进入控制台，鼠标选择其中一个帖子，查看右边Chrome调试工具的Eelements，可以看到每一个帖子的一块区域所对应的html代码都是由一个`div calss='r-ent'`包裹的. 这种结构化的网页是我们最喜欢看见的，有规律可循，所以下一步就是按需提取信息了。 下面我贴出一个帖子的html代码，然后分别基于两个解析库`BeautifulSoup`、`parsel`提取我们想要的信息 ```html
11
[新聞] 童子賢：用稅收補貼電費非長久之計 應共
addy7533967
⋯
搜尋同標題文章
搜尋看板內 addy7533967 的文章
3/27
``` ``` /* Detailed source-code truncated for AI context efficiency. */ ``` ### 简易流程图 > 一般像这一类比较简单的爬虫需求，我几乎不化流程图，但是为了让大家更清楚的知道，我在代码编写前都会给大家画一下，养成一个coding前画图的习惯，其实对于自己提升和代码容错都有一定的帮助，画图的过程中就在思考代码流程了。 ### 代码实现 #### 依赖库安装 ```shell pip3 install requests pip3 install beautifulsoup4 pip3 install lxml pip3 install httpx pip3 install parsel ``` #### requests + BeautifulSoup 同步版本 > 代码路径：源代码/爬虫入门/08_爬虫入门实战1_静态网页数据提取/002_源码实现_同步版本.py ``` /* Detailed source-code truncated for AI context efficiency. */ ``` #### httpx + parsel 异步版本 > 代码路径：源代码/爬虫入门/08_爬虫入门实战1xxx ``` /* Detailed source-code truncated for AI context efficiency. */ ``` ### 存储实现 > 存储实现我们留在第10讲再去实现吧 ### 源代码 [08_爬虫入门实战1_静态网页数据提取 - 章节源代码地址](https://github.com/NanmiCoder/CrawlerTutorial/tree/main/%E6%BA%90%E4%BB%A3%E7%A0%81/%E7%88%AC%E8%99%AB%E5%85%A5%E9%97%A8/08_%E7%88%AC%E8%99%AB%E5%85%A5%E9%97%A8%E5%AE%9E%E6%88%981_%E9%9D%99%E6%80%81%E7%BD%91%E9%A1%B5%E6%95%B0%E6%8D%AE%E6%8F%90%E5%8F%96) ### 其他 > 不知不觉的这一篇教程从晚上9点开始的，写到了晚上2.27，存储实现还没写玩，之前写前几篇帖子没什么感觉，到了实战帖子之后，感觉花费的时间多了很多很多。
> 并且这还是一个很简单很简单的爬虫需求，想要把完整的内容思路用图文表达出来，确实不是那么容易。。。。想到后续的进阶和高级爬虫，有点社死，可能时间严重不够。 --- ### 爬虫入门 09 爬虫入门实战2 动态数据提取 (`docs/爬虫入门/09_爬虫入门实战2_动态数据提取.md`) # 爬虫入门实战2：动态数据提取 在爬虫入门实战1中，我们学习了如何从静态网页中提取数据。这一节，我们将探索动态网页的数据提取，这是爬虫技术中更为高级也更为常见的一个环节。 ## 什么是动态网页 动态网页与静态网页不同，其内容是可以根据用户操作、请求参数或者是服务器端的数据变化而变化的。动态网页的内容通常是通过客户端的JavaScrip发起异步请求，由服务端动态返回的数据（json、html） ## 动态网页工作原理 动态网页的数据加载通常有两种方式： - **客户端渲染**：当用户访问一个网页时，服务器只发送HTML框架和JavaScript代码。浏览器执行JavaScript代码，这些代码会向服务器发送请求，获取数据，然后动态地更新页面内容。 - **服务器端渲染**：服务器根据请求的不同，动态生成HTML内容发送给客户端。客户端接收到完整的HTML页面，直接渲染显示。 ## 爬取动态网页需要的技术 - 理解AJAX和API请求：动态网页往往通过AJAX请求获取数据，了解这一点对于数据提取至关重要。 - 使用浏览器开发者工具：通过分析网络请求，找出数据加载的具体过程。 - 使用适合动态网页的请求库，如requests、httpx、aiohttp等。 - 使用适合动态网页的驱动库，如Selenium、Puppeteer、Playwirght等，模拟浏览器行为获取数据。 - 学习JavaScript基础，有助于理解网页是如何动态加载数据的。 ## 实战示例1：爬取雅虎财经网站的数字加密货币数据 ## 任务需求描述 目标站点URL：https://finance.yahoo.com/crypto 爬取目标站点所有数字货币的数据，请使用API请求，不要用模拟浏览器驱动的方式 数据保存示例 ## 技术可行性分析 > 这是一个典型的异步加载方式，当点击翻页时，浏览器整个页面并没有完整刷新，而是加密货币这个表格动态更新（局部页面变化）， > 那么这一种就是上面说的动态数据提取的应用示例，接下来一步步的去研究这种页面我们该如何去分析与抓取。 ### 如何找到动态数据请求的API 先开F12，选中`XHR`, 打开上面的目标页面 https://finance.yahoo.com/crypto 找到表格，点击下面的`Next`翻页，随便找一个表格中的数据，在搜索栏目搜索结果 点击搜出来的结果，找到响应信息与表格中的数据对比，我们就能找到这个表格发起的API请求了。 **请求信息** **响应信息** **接口信息** API URL ：https://query1.finance.yahoo.com/v1/finance/screener?crumb=UllRf10isbP&lang=en-US®ion=US&formatted=true&corsDomain=finance.yahoo.com API Payload参数： ```json { "offset": 25, "size": 25, "sortType": "DESC", "sortField": "intradaymarketcap", "quoteType": "CRYPTOCURRENCY", "query": { "operator": "and", "operands": [ { "operator": "eq", "operands": [ "currency", "USD" ] }, { "operator": "eq", "operands": [ "exchange", "CCC" ] } ] }, "userId": "", "userIdType": "guid" } ``` API JSON 响应数据： ``` /* Detailed source-code truncated for AI context efficiency. */ ``` 也可以在chrome浏览器复制该请求的curl命令： ``` /* Detailed source-code truncated for AI context efficiency. */ ``` 可以在谷歌搜索curl to python request工具，可以把上面的curl命令直接转成python代码，方便调试 转成的python request 代码请求示例 ``` /* Detailed source-code truncated for AI context efficiency. */ ``` ### 请求参数和响应数据分析 **请求参数** ```json {"offset":0,"size":25,"sortType":"DESC","sortField":"intradaymarketcap","quoteType":"CRYPTOCURRENCY","query":{"operator":"and","operands":[{"operator":"eq","operands":["currency","USD"]},{"operator":"eq","operands":["exchange","CCC"]}]},"userId":"","userIdType":"guid"} ``` 主要看`offset`、`size`这两个参数，这在`ajax`动态分页的请求中是最常见的，一般情况下 `offset`表示数据的分页的起始位置，`size`表示分页的每一次数据（步长）。
我们在页面中点击下一页，看看这两个参数有什么变化。从下面的截图中可以得知，如果我们想要下一页的数据，offset + 25， size不变 变化情况 offset size
第一次请求 0 - 25
第二次请求 25 - 25
第三次请求 50 - 25
... 相当于新的起始位置参数offset 等于 上一次的offset + size.
**响应数据** 我们要的是每个币种的数据，从下面的响应json中能看见币中数据存在了quotes中 那现在基本就完事了，分页参数有了，响应数据有了 ### 如何提取我们要的目标数据 这一块原本我是想给大家推荐一个`jsonpath`的提取工具，但是发现会增加大家的学习成本，就先不讲了，感兴趣的朋友可以去了解一下。 这个提取`jsonpath` 提取json数据很方便。你可以理解把它为xpath一样的工具，xpath主要是解析xml/html, jsonpath主要是提取json。 好了，言归正传，我们还是用python的内置类型来提取把。 响应数据是json格式，如何转变成python里面的数据格式呢，大多数人都知道用json.loads()方法，但是这个一步到底是在干什么？ json.loads()方法的核心作用是将JSON格式的字符串“翻译”为Python中的数据结构，使得程序可以方便地访问和操作这些数据。这个过程包括了解析、类型转换和构建数据结构等步骤，是数据处理和Web开发中不可或缺的一环。 这一过程叫做`json数据的反序列化`, 什么是序列化什么是反序列化，大家可以在问下GPT助手，或者谷歌搜索一下。 ### 简易流程图 ### 代码实现 #### 依赖库安装 ```shell # 同步代码安装依赖 pip install requests # 异步代码安装依赖 pip install httpx pip install aiofiles ``` #### requests版本 > 源代码在 源代码/爬虫入门/09_爬虫入门实战2_动态数据提取/002_源码实现_同步版本.py ``` /* Detailed source-code truncated for AI context efficiency. */ ``` #### httpx + aiofiles版本 > > 源代码在 源代码/爬虫入门/09_爬虫入门实战2_动态数据提取/003_源码实现_异步.py ``` /* Detailed source-code truncated for AI context efficiency. */ ``` ### 源代码地址 [09_爬虫入门实战2_动态数据提取 - 章节源代码地址](https://github.com/NanmiCoder/CrawlerTutorial/tree/main/%E6%BA%90%E4%BB%A3%E7%A0%81/%E7%88%AC%E8%99%AB%E5%85%A5%E9%97%A8/09_%E7%88%AC%E8%99%AB%E5%85%A5%E9%97%A8%E5%AE%9E%E6%88%982_%E5%8A%A8%E6%80%81%E6%95%B0%E6%8D%AE%E6%8F%90%E5%8F%96) --- ### 爬虫入门 10 爬虫入门实战3 数据存储实现 (`docs/爬虫入门/10_爬虫入门实战3_数据存储实现.md`) # 爬虫入门实战3 数据存储实现 数据存储，顾名思义，将我们爬取的数据存储下来，这一章将带着大家一步步的实现一个数据存储的过程 我们将会使用到两种存储方式，一种是存储到本地文件，另一种是存储到数据库中。 本地文件又可以分为2种，一种是存储到csv文件，另一种是存储到json文件。 > 你不妨思考一下，如果是你，你会怎么设计你的爬虫程序支持不同的存储方式呢？ ## 前置准备 在正式讨论数据存储之前，我们先来回顾一下在 [动态数据这一章](09_爬虫入门实战2_动态数据提取.md) 是如何获取动态数据。 在动态数据获取这一章，我们是不是爬取了一个雅虎财经方面的加密货币数据，然后将数据保存到CSV文件中了，当时对数据存储没有做过的介绍。 这一章，我们将会详细的讲解数据存储的过程，以及如何将数据存储到数据库中。 回顾上一章我们存储动态数据的容器类定义： ```python from typing import List class SymbolContent: symbol: str = "" name: str = "" price: str = "" # 价格（盘中） change_price: str = "" # 跌涨价格 change_percent: str = "" # 跌涨幅 market_price: str = "" # 市值 @classmethod def get_fields(cls) -> List[str]: return [key for key in cls.__dict__.keys() if not key.startswith('__') and key != "get_fields"] def __str__(self): return f""" Symbol: {self.symbol} Name: {self.name} Price: {self.price} Change Price: {self.change_price} Change Percent: {self.change_percent} Market Price: {self.market_price} """ ``` 从上面的容器类定义可以看见，我们已经将爬取到的数据有意识的转换成一个python结构化数据了，但是呢，这个数据是存储在内存中的，我们需要将这个数据存储到文件或者数据库中，这样我们才能更好的利用这些数据。 那么问题来了，当前的SymbolContent类，要支持存储到文件或db中，要做一转换工作，打个比方，如何转成json格式，如何转成python的dict，这些都是我们需要考虑的问题。 幸好的是，在python这边，有一个库，它非常适合做这件事情，那就是[Pydantic](https://docs.pydantic.dev/latest/)库，这个库可以帮助我们将python的数据结构转换成json格式，dict格式，有了dict和json，其实存储CSV、存储数据库都是非常方便的了。 ## Pydantic的基本使用介绍 [Pydantic](https://docs.pydantic.dev/latest/)是一个数据验证和序列化库，它可以帮助我们定义数据模型，然后将数据模型转换成json格式，dict格式，这样我们就可以很方便的将数据存储到文件或者数据库中。 使用Pydantic最简单的demo ```python # 导入Pydantic库的BaseModel基类 from pydantic import BaseModel # 定义你的模型类 class User(BaseModel): id: int name: str age: int # 实例化你的模型类 user = User(id=1, name="小明", age=18) # 将模型类转换成dict user_dict = user.model_dump() print(type(user_dict), user_dict) # {'id': 1, 'name': '小明', 'age': 18} # 将模型类转换成json user_json = user.model_dump_json() print(type(user_json), user_json) # {"id": 1, "name": "小明", "age": 18} ``` 相信你从上面例子中可以看出，Pydantic的使用是非常简单的，只需要定义你的模型类，然后实例化你的模型类，然后调用model_dump()方法就可以将模型类转换成dict格式，调用model_dump_json()方法就可以将模型类转换成json格式。 pydantic真的非常非常好用，尤其是在python这门动态类型的脚本语言，使用pydantic能够加强你代码的类型检查，让你的代码更加健壮。更多的使用方法，可以参考[Pydantic官方文档](https://docs.pydantic.dev/latest/) 如果python是你的主力语言的话， 强力推荐！ ## 学习目标 - 设计一个数据存储抽象类 - 将数据存储到CSV文件 - 将数据存储到JSON文件 - 封装一个mysql数据库操作类 - 将数据存储到数据库 - 改造动态数据章节的存储代码，支持不同类型存储 ## 设计原理 ## 流程设计 ## 存储层类图设计 > 类图是软件工程的统一建模语言一种静态结构图，该图描述了系统的类集合，类的属性和类之间的关系。
> > 类图是面向对象式的建模。他们一般都被用于概念建模（conceptual modelling）的系统分类的应用程序，并可将模型建模转译成代码。
> > 一个类有三个区域 > - 最上面是类名称 > - 中间部分包含类的属性 > - 底部部分包含类的方法 > - 为了进一步描述系统的行为，这些类图可以辅之以状态图或UML状态机。 ## 代码实现过程 > 后面源码都用python的异步编程来写吧，这样可以更好的提高代码的执行效率，提高爬虫的效率。而且我比较擅长异步编程，哈哈哈 ## SymbolContent类转换成Pydantic模型类 ```python from pydantic import BaseModel, Field class SymbolContent(BaseModel): symbol: str = Field(default="", title="Symbol") name: str = Field(default="", title="Name") price: str = Field(default="", title="价格盘中") change_price: str = Field(default="", title="跌涨价格") change_percent: str = Field(default="", title="跌涨幅") market_price: str = Field(default="", title="市值") ``` 可以看到 `SymbolContent` 模型类，相较于原来的变化不是很大，只是增加了一些Pydantic的Field字段，这些字段可以帮助我们定义字段的默认值，字段的标题等信息。 另外差异比较大的是现在的 `SymbolContent` 类，继承了 `BaseModel` 类，这样我们就可以使用Pydantic的一些方法了。 ## 设计存储抽象类 > 抽象的好处： > > 一个好的抽象可以将大量实现细节隐藏在一个干净，简单易懂的外观下面。一个好的抽象也可以广泛用于各类不同应用。比起重复造很多轮子，重用抽象不仅更有效率，而且有助于开发高质量的软件。抽象组件的质量改进将使所有使用它的应用受益。 > > 例如，高级编程语言是一种抽象，隐藏了机器码、CPU 寄存器和系统调用。SQL 也是一种抽象，隐藏了复杂的磁盘 / 内存数据结构、来自其他客户端的并发请求、崩溃后的不一致性。当然在用高级语言编程时，我们仍然用到了机器码；只不过没有 直接（directly） 使用罢了，正是因为编程语言的抽象，我们才不必去考虑这些实现细节。 > > 抽象可以帮助我们将系统的复杂度控制在可管理的水平，不过，找到好的抽象是非常困难的。在分布式系统领域虽然有许多好的算法，但我们并不清楚它们应该打包成什么样抽象。 > > from ddia 从类图的我们很轻松就能把存储抽象类给定义出来，代码定义如下： ```python from abc import ABC, abstractmethod from common import SymbolContent class AbstractStore(ABC): @abstractmethod async def save(self, save_item: SymbolContent): """ 存储数据 :param save_item: :return: """ raise NotImplementedError ``` 其中save方法就是抽象方法，我们需要在子类中实现这个方法，这样我们就可以将数据存储到不同的地方了。 这里多说一点关于python这边的抽象类， - python这边提供了一个abc模块，我们可以使用这个模块来定义抽象类，这样我们就可以在子类中实现抽象方法了。 - @abstractmethod装饰器是一个抽象方法的标志，如果一个类中有抽象方法，那么这个类就是一个抽象类，抽象类不能被实例化，只能被继承。 - raise NotImplementedError是一个异常，如果子类没有实现抽象方法，那么就会抛出这个异常。 ## 存储到CSV文件 首先，从类图看，我们需要实现一个存储到CSV文件的类，这个类需要继承我们的抽象类，然后实现抽象方法，代码如下： ```python import csv import pathlib import time from typing import Dict import aiofiles from abstract_store import AbstractStore from common import SymbolContent class CsvStoreImpl(AbstractStore): def __init__(self): self.csv_store_path = "data/csv" def make_save_file_name(self) -> str: """ make save file name :return: """ return f"{self.csv_store_path}/symbol_content_{int(time.time())}.csv" async def save(self, save_item: SymbolContent): """ save data to csv :param save_item: :return: """ pathlib.Path(self.csv_store_path).mkdir(parents=True, exist_ok=True) save_file_name = self.make_save_file_name() async with aiofiles.open(save_file_name, mode='a+', encoding="utf-8-sig", newline="") as f: f.fileno() writer = csv.writer(f) save_item_dict: Dict = save_item.model_dump() if await f.tell() == 0: await writer.writerow(save_item_dict.keys()) await writer.writerow(save_item_dict.values()) ``` 这个类的实现非常简单，首先我们需要实现save方法，这个方法是抽象方法，我们需要将数据存储到CSV文件中，这里我们使用了aiofiles库来异步写文件，这样可以提高写文件的效率。 另外可以看到函数签名中`save_item`的这个参数的类型是`SymbolContent`，这个是我们定义的数据模型类，这个类是我们爬取到的数据，我们需要将这个数据存储到CSV文件中。 我们是用pydantic的model_dump()方法将数据转换成dict格式，然后使用csv库将数据写入到CSV文件中。使用起来非常的方便。 ## 存储到JSON文件 存储到JSON文件的实现和存储到CSV文件的实现非常类似，只是存储的格式不同，代码如下： ```python import json import os import pathlib import time import aiofiles from abstract_store import AbstractStore from common import SymbolContent class JsonStoreImpl(AbstractStore): def __init__(self): self.json_store_path = "data/json" def make_save_file_name(self) -> str: """ make save file name :return: """ return f"{self.json_store_path}/symbol_content_{int(time.time())}.json" async def save(self, save_item: SymbolContent): """ save data to json :param save_item: :return: """ pathlib.Path(self.json_store_path).mkdir(parents=True, exist_ok=True) save_file_name = self.make_save_file_name() save_data_list = [] # todo 如果这里涉及并发写入，需要加锁, 可以查看MediaCrawler项目中的实现方式 # 先判断文件是否存在，如果存在则读取文件内容放到save_data_list中，然后再将新的数据添加到save_data_list中 if os.path.exists(save_file_name): async with aiofiles.open(save_file_name, 'r', encoding='utf-8') as file: save_data_list = json.loads(await file.read()) save_data_list.append(save_item.model_dump()) # 将数据写入到文件中 async with aiofiles.open(save_file_name, 'w', encoding='utf-8') as file: await file.write(json.dumps(save_data_list, ensure_ascii=False)) ``` 存储数据到json的实现思路是每一次存储数据的时候，我们都将数据读取出来，然后将新的数据添加到数据中，然后再将数据写入到文件中，这样可以保证数据不会丢失。 （因为json特殊的数据结构的原因，它无法像存储csv那样可以在文件末尾行追加写入） ## 存储到数据库 ### 封装一个mysql数据库操作类 > 与mysql数据库交互，我们需要使用到一个库，这个库就是aiomysql，这个库是一个异步的mysql库，可以帮助我们异步的操作mysql数据库，这样可以提高数据库的操作效率。 > > 由于aiomyql提供的方法比较偏底层，并且在我们连接数据库查询到数据之后，我们还要进行部分数据转换，所以为了使用方面，我们可以封装一些基础的数据库操作方法，这样我们在存储数据的时候，就可以直接调用这些方法，而不用关心底层的实现。 > > 主要是增、删、改、查，下面给出我用了几年的一个封装aiomysql代码，供大家参考。 ``` /* Detailed source-code truncated for AI context efficiency. */ ``` 从`AsyncMysqlDB`这个类的构造函数__init__看，它接收一个aiomysql.Pool对象，这个对象是一个连接池对象，我们可以通过这个连接池对象来获取数据库连接，然后执行我们的sql语句。 这样做的好处是，我们创建了一个连接池，可以减少数据库连接的开销，提高数据库的操作效率。 这个类中有几个方法，分别是query、get_first、item_to_table、update_table、execute，这几个方法分别是查询、查询第一个、插入、更新、执行sql语句的方法，这几个方法是我们在存储数据的时候经常会用到的方法，我们可以直接调用这些方法，而不用关心底层的实现。 `MysqlConnect` 这个类是一个单例模式，它的作用是创建一个数据库连接对象，我们可以通过这个对象来操作数据库，这样可以减少数据库连接的开销。 如果想查看AsyncMysqlDB的更多用法，可以查看我在这里写的使用示例：[test_mysql_async_db.py](https://github.com/NanmiCoder/python_common_libs/blob/main/test/test_mysql_async_db.py) ### DB存储数据实现 存储到数据库的实现和存储到文件的实现有一些不同，我们需要先将数据存储到内存中(save_item这个模型类)，然后再将数据存储到数据库中，代码如下： ```python from typing import Optional from abstract_store import AbstractStore from async_db import MysqlConnect, AsyncMysqlDB from common import SymbolContent class DbStoreImpl(AbstractStore): def __init__(self): self.db: Optional[AsyncMysqlDB] = None async def save(self, save_item: SymbolContent): """ save data to db :param save_item: :return: """ self.db = (await MysqlConnect().async_init()).get_db() from sqls import (insert_symbol_content, query_symbol_content_by_symbol, update_symbol_content) # 查询是否存在 exist_item = await query_symbol_content_by_symbol(self.db, save_item.symbol) if exist_item.symbol: # 更新 await update_symbol_content(self.db, save_item) else: # 插入 await insert_symbol_content(self.db, save_item) ``` 从 `DbStoreImpl` 类的save方法的逻辑看，其实非常的简单，就是先查询数据库中是否存在这个数据，如果存在则更新，如果不存在则插入，这样就可以保证数据不会重复。 sqls文件定义： ```python # -*- coding: utf-8 -*- # @Author : relakkes@gmail.com # @Name : 程序员阿江-Relakkes # @Time : 2024/6/7 17:09 # @Desc : from async_db import AsyncMysqlDB from common import SymbolContent async def insert_symbol_content(db: AsyncMysqlDB, symbol_content: SymbolContent) -> int: """ 插入数据 :param db: :param symbol_content: :return: """ item = symbol_content.model_dump() return await db.item_to_table("symbol_content", item) async def update_symbol_content(db: AsyncMysqlDB, symbol_content: SymbolContent) -> int: """ 更新数据 :param db: :param symbol_content: :return: """ item = symbol_content.model_dump() return await db.update_table("symbol_content", item, "symbol", symbol_content.symbol) async def query_symbol_content_by_symbol(db: AsyncMysqlDB, symbol: str) -> SymbolContent: """ 查询数据 :param db: :param symbol: :return: """ sql = f"select * from symbol_content where symbol = '{symbol}'" rows = await db.query(sql) if len(rows) > 0: return SymbolContent(**rows[0]) return SymbolContent() ``` 存储数据到数据库的好处太多了，比如数据的持久化，数据的查询，数据的分析等等，这些都是存储到文件无法实现的。 所以在我们平时做数据爬取的时候，我建议大家将数据存储到数据库中，这样可以更好的利用数据。 ## 动态数据爬虫章节主代码修改 > 修改动态数据章节的代码，支持将加密货币数据存储到不同的地方 代码改动在`run_crawler`中, 我们从网页中抓取到数据后，遍历每一个数据，然后将数据存储到指定的存储介质中，代码如下 ```python async def run_crawler(data_save_type: str) -> None: """ 爬虫主流程 :param data_save_type: 数据存储的类型，支持csv、json、db :return: """ # step1 获取最大数据总量 max_total: int = await get_max_total_count() # step2 遍历每一页数据并解析存储到数据容器中 data_list: List[SymbolContent] = await fetch_currency_data_list(max_total) # step3 将数据保存到指定存储介质中 for data_item in data_list: await StoreFactory.get_store(data_save_type).save(data_item) if __name__ == '__main__': _data_save_type = "csv" # 可选配置（csv、json、db） asyncio.run(run_crawler(_data_save_type)) ``` 从使用层面看，我们指定了存储类型，根据存储工厂类的设计，我们可以很方便的切换存储类型，这样我们就可以将数据存储到不同的地方了。 下面给出存储工厂类的实现（设计模式中的简单工厂）： ```python class StoreFactory: @staticmethod def get_store(store_type: str) -> AbstractStore: if store_type == "csv": return CsvStoreImpl() elif store_type == "json": return JsonStoreImpl() elif store_type == "db": return DbStoreImpl() else: raise ValueError(f"Unknown store type: {store_type}") ``` 大家有没有发现一问题，从09章动态数据提取到这一章，我们的代码是不是越来越规范了，越来越好维护了 这就是我们学习的意义，良好的抽象能力，良好的设计能力，能够让我们的代码更加的优雅，更加的易维护。 ## 依赖安装 & 代码运行 ### 依赖安装 ```shell # 依赖安装，进入到`10_爬虫入门实战3_数据存储实现`目录下，运行以下命令 # 创建python虚拟环境 python3 -m venv venv # 激活虚拟环境 source venv/bin/activate # 安装依赖 pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple ``` ### 代码运行 ```shell # 如果是保存的mysql中的话，需要创建数据库和表结构，可以查看源代码目录下的sql文件 python main.py ``` - 如果是以`csv`的方式存储数据，可以查看`data/csv`目录下的文件， - 如果是`json`的方式存储数据，可以查看`data/json`目录下的文件， - 如果是`db`的方式存储数据，可以查看数据库中的数据。 ### 源代码地址 [10_爬虫入门实战3_数据存储实现 - 章节源代码地址](https://github.com/NanmiCoder/CrawlerTutorial/tree/main/%E6%BA%90%E4%BB%A3%E7%A0%81/%E7%88%AC%E8%99%AB%E5%85%A5%E9%97%A8/10_%E7%88%AC%E8%99%AB%E5%85%A5%E9%97%A8%E5%AE%9E%E6%88%983_%E6%95%B0%E6%8D%AE%E5%AD%98%E5%82%A8%E5%AE%9E%E7%8E%B0) --- ### 爬虫入门 11 爬虫入门实战4 高效率的爬虫实现 (`docs/爬虫入门/11_爬虫入门实战4_高效率的爬虫实现.md`) # 爬虫入门实战4_高效率的爬虫实现 在现代爬虫开发中，提高爬取效率是一个永恒的话题。本文将深入探讨Python中的多进程、多线程和协程三种并发编程方式，分析它们在爬虫开发中的应用，并通过实际案例比较它们的性能表现。 ## 1. 并发编程概述 ### 1.1 多进程（Multiprocessing） 多进程是指在操作系统中同时运行多个独立的进程。每个进程都有自己的内存空间和系统资源。 优点： - 可以充分利用多核CPU - 绕过Python的全局解释器锁（GIL） - 进程间内存隔离，更安全 缺点： - 进程创建和切换开销大 - 进程间通信相对复杂 - 占用较多系统资源 ### 1.2 多线程（Multithreading） 多线程是在同一进程内创建多个线程，共享进程的内存空间。 优点： - 资源占用相对较少 - 线程间共享内存，通信方便 - 适合I/O密集型任务 缺点： - 受Python GIL限制，难以充分利用多核CPU - 需要考虑线程安全问题 - 调试相对困难 ### 1.3 协程（Coroutine） 协程是一种用户态的轻量级线程，通过协作式多任务实现并发。 优点： - 极低的系统开销 - 高效处理I/O密集型任务 - 编程模型简单，易于理解 缺点： - 不适合CPU密集型任务 - 需要特定的库支持（如asyncio） - 对于长时间运行的I/O操作可能会阻塞事件循环 ## 2. 并发编程的演变 > 最新的3.13已经支持编译一个无gil版本的python，后面可能python真的要起飞🛫了 Python并发编程的演变历程： 1. 早期：单线程同步编程 2. Python 2.x：引入threading模块，支持多线程 3. Python 2.6+：引入multiprocessing模块，支持多进程 4. Python 3.4+：引入asyncio模块，支持协程 5. Python 3.5+：引入async/await语法，简化协程编写 这种演变反映了开发者对更高效、更易用的并发编程方式的不懈追求。 ## 3. 最简单的基本示例 ### 3.1 多进程示例 ```python import multiprocessing import time def worker(num): print(f"Worker {num} started") time.sleep(2) print(f"Worker {num} finished") if __name__ == "__main__": processes = [] for i in range(5): p = multiprocessing.Process(target=worker, args=(i,)) processes.append(p) p.start() for p in processes: p.join() print("All processes completed") ``` ### 3.2 多线程示例 ```python import threading import time def worker(num): print(f"Thread {num} started") time.sleep(2) print(f"Thread {num} finished") threads = [] for i in range(5): t = threading.Thread(target=worker, args=(i,)) threads.append(t) t.start() for t in threads: t.join() print("All threads completed") ``` ### 3.3 协程示例 ```python import asyncio async def worker(num): print(f"Coroutine {num} started") await asyncio.sleep(2) print(f"Coroutine {num} finished") async def main(): tasks = [asyncio.create_task(worker(i)) for i in range(5)] await asyncio.gather(*tasks) asyncio.run(main()) print("All coroutines completed") ``` 上述的三个示例，在时间上都是2秒，但是在内存上有所不同，多进程的内存占用最大，多线程次之，协程最小。 ## 4. 爬虫中的应用 在爬虫开发中，这三种并发方式各有其适用场景： - 多进程：适合需要绕过GIL、利用多核CPU的场景，如大规模数据处理。 - 多线程：适合I/O密集型任务，如同时爬取多个网页。 - 协程：最适合大量并发I/O操作，如高并发的网络请求。 ## 5. 实战对比 我们将使用多进程、多线程和协程三种方式实现同一个爬虫任务：爬取Yahoo Finance的加密货币数据。我们会比较它们的性能表现。 如果不熟悉 Yahoo Finance 的加密货币数据 可以回过头去看 [09_爬虫入门实战2_动态数据提取.md](09_爬虫入门实战2_动态数据提取.md) 章节 ### 5.1 多进程版本实现 ``` /* Detailed source-code truncated for AI context efficiency. */ ``` ### 5.2 多线程版本 ``` /* Detailed source-code truncated for AI context efficiency. */ ``` ### 5.3 协程版本实现 ``` /* Detailed source-code truncated for AI context efficiency. */ ``` > 上述源代码路径：[11_爬虫入门实战4_高效率的爬虫实现](https://github.com/NanmiCoder/CrawlerTutorial/tree/main/%E6%BA%90%E4%BB%A3%E7%A0%81/%E7%88%AC%E8%99%AB%E5%85%A5%E9%97%A8/11_%E7%88%AC%E8%99%AB%E5%85%A5%E9%97%A8%E5%AE%9E%E6%88%984_%E9%AB%98%E6%95%88%E7%8E%87%E7%9A%84%E7%88%AC%E8%99%AB%E5%AE%9E%E7%8E%B0) ### 5.4 执行耗时 #### 5.4.1 多线程执行耗时 开始获取最大的币种数量
获取到 9967 种币种
总共发起: 100 次网络请求
多线程执行程序耗时: 7.992658853530884 秒
#### 5.4.2 多进程执行耗时 开始获取最大的币种数量
获取到 9967 种币种
总共发起: 100 次网络请求
多进程执行程序耗时: 17.447596073150635 秒
#### 5.4.3 协程执行耗时 开始获取最大的币种数量
获取到 9967 种币种
总共发起: 100 次网络请求
asyncio调度协程执行程序耗时: 4.690491199493408 秒
## 6. 上述代码总结 > 我比较喜欢使用异步协程，因为编程风格很像同步代码，并且还能带来高效率的表现。 ### 6.1. 多线程（run_crawler_multi_thread.py） 适用场景：适合I/O密集型任务，如网络请求，因为线程在等待I/O操作（如网络响应）时可以让出CPU给其他线程。 实现逻辑： - 使用ThreadPoolExecutor来管理线程池。 - 将任务（获取单页货币数据）分配给线程池中的线程执行。 - 使用executor.map来并行处理多个页面的数据获取，这个方法会自动处理任务的分配和结果的收集。 ### 6.2. 多进程（run_crawler_multi_process.py） 适用场景：适合CPU密集型任务，但在这个案例中，它用于处理I/O密集型任务，这通常不是最佳选择，因为进程间通信成本较高。 实现逻辑： - 使用multiprocessing.Pool来创建进程池。 - 类似于多线程，使用pool.map来并行处理多个页面的数据获取。 - 进程间的数据传递通过序列化和反序列化实现，这可能会引入额外的开销。 ### 6.3. 协程（run_crawler_multi_coroutine.py） 适用场景：非常适合I/O密集型任务，如网络请求。协程通过事件循环和非阻塞I/O操作提高程序的执行效率。 实现逻辑： - 使用asyncio库来管理协程。 - 使用httpx.AsyncClient进行异步HTTP请求，这允许在等待网络响应时不阻塞程序的其他部分。 - 使用asyncio.gather来并发执行多个协程，这些协程分别处理不同页面的数据获取。 总结 - 多线程和多进程都可以处理并发任务，但在处理大量的网络I/O操作时，它们可能不如协程高效。 - 协程提供了最高的效率和最佳的资源利用率，特别是在处理网络I/O密集型任务时。 - 在选择并发策略时，应考虑任务的类型（CPU密集型还是I/O密集型）、系统的资源（如CPU核心数）以及程序的复杂性。 --- ### 爬虫进价 01 工程化爬虫开发规范 (`docs/爬虫进价/01_工程化爬虫开发规范.md`) # 工程化爬虫开发规范 > 在入门教程中，我们学会了如何写一个能用的爬虫。但在实际生产环境中，一个"能用"的爬虫远远不够。本章将带你从工程化的角度重新审视爬虫开发，让你的爬虫代码更加健壮、可维护、易于调试。 > **学习目标**：掌握日志系统、配置管理、异常处理和项目结构设计，为后续进阶学习打下坚实基础。 ## 为什么需要工程化 当你的爬虫从简单的脚本演变为需要长期运行、多人协作的项目时，以下问题会逐渐浮现： - **调试困难**：程序出错了，但不知道错在哪里 - **配置混乱**：API密钥、数据库连接等硬编码在代码中 - **脆弱性高**：网络波动就导致整个程序崩溃 - **可读性差**：几个月后自己都看不懂当初写的代码 工程化开发规范正是为了解决这些问题而存在的。本章我们将重点学习： 1. 使用 `loguru` 构建专业的日志系统 2. 使用 `pydantic-settings` 进行配置管理 3. 实现统一的异常处理和重试机制 4. 项目目录结构的最佳实践 ### 工程化核心模块 ```mermaid graph TB subgraph 工程化核心 log[日志系统
loguru] config[配置管理
pydantic-settings] exception[异常处理
自定义异常+重试] structure[项目结构
模块化设计] end subgraph 解决的问题 p1[调试困难] p2[配置混乱] p3[脆弱性高] p4[可读性差] end log -->|解决| p1 config -->|解决| p2 exception -->|解决| p3 structure -->|解决| p4 ``` --- ## 日志系统设计 ### 为什么需要专业的日志系统 在入门教程中，我们大量使用 `print()` 来输出信息。这种方式在调试时很方便，但存在以下问题： - 无法区分信息的重要程度（调试信息和错误信息混在一起） - 无法持久化保存日志 - 无法在生产环境中动态调整日志级别 - 无法记录时间、文件位置等上下文信息 `loguru` 是一个优秀的 Python 日志库，它用极简的 API 解决了以上所有问题。 ### loguru 基本使用 首先安装 loguru： ```bash pip install loguru ``` 最简单的使用方式： ```python from loguru import logger logger.debug("这是一条调试信息") logger.info("这是一条普通信息") logger.warning("这是一条警告信息") logger.error("这是一条错误信息") logger.critical("这是一条严重错误信息") ``` 运行后你会看到彩色的、带时间戳和文件位置的日志输出： ``` 2024-03-28 10:30:00.123 | DEBUG | __main__::3 - 这是一条调试信息 2024-03-28 10:30:00.124 | INFO | __main__::4 - 这是一条普通信息 2024-03-28 10:30:00.125 | WARNING | __main__::5 - 这是一条警告信息 2024-03-28 10:30:00.126 | ERROR | __main__::6 - 这是一条错误信息 2024-03-28 10:30:00.127 | CRITICAL | __main__::7 - 这是一条严重错误信息 ``` ### 日志分级策略 在爬虫项目中，建议按以下方式使用日志级别： ```mermaid graph LR subgraph 日志级别从低到高 DEBUG[DEBUG
调试信息] --> INFO[INFO
正常信息] INFO --> WARNING[WARNING
警告信息] WARNING --> ERROR[ERROR
错误信息] ERROR --> CRITICAL[CRITICAL
致命错误] end subgraph 爬虫使用场景 d1[请求参数/响应片段] -.-> DEBUG d2[爬取第N页/保存N条] -.-> INFO d3[请求超时重试] -.-> WARNING d4[单页解析失败] -.-> ERROR d5[数据库连接失败] -.-> CRITICAL end ``` | 级别 | 使用场景 | 爬虫示例 | |------|----------|----------| | DEBUG | 详细的调试信息，如请求参数、响应内容片段 | 请求URL、响应状态码、响应内容片段 | | INFO | 正常的运行信息，如"开始爬取第N页"、"成功保存N条数据" | 爬取进度、数据保存成功 | | WARNING | 可恢复的异常，如"请求超时，准备重试" | 请求超时、触发频率限制、需要重试 | | ERROR | 严重错误但程序可继续，如"单个页面解析失败" | 单个页面解析失败、数据格式异常 | | CRITICAL | 致命错误，程序无法继续，如"数据库连接失败" | 数据库连接失败、认证失效 | ### 日志持久化和轮转 在生产环境中，我们需要将日志保存到文件以便后续分析。loguru 提供了强大的日志文件管理功能： ```python from loguru import logger import sys # 移除默认的控制台输出（可选） logger.remove() # 添加控制台输出，只显示 INFO 及以上级别 logger.add( sys.stderr, level="INFO", format="{time:YYYY-MM-DD HH:mm:ss} | {level: <8} | {name}:{function}:{line} - {message}" ) # 添加文件输出，记录所有级别，按日期轮转 logger.add( "logs/crawler_{time:YYYY-MM-DD}.log", rotation="00:00", # 每天午夜轮转 retention="7 days", # 保留7天的日志 compression="zip", # 旧日志压缩 level="DEBUG", encoding="utf-8" ) # 单独的错误日志文件 logger.add( "logs/error_{time:YYYY-MM-DD}.log", rotation="00:00", retention="30 days", level="ERROR", encoding="utf-8" ) ``` ### 在爬虫中使用日志的最佳实践 #### 通用示例 ```python from loguru import logger async def fetch_page(client, url: str) -> str: """获取页面内容""" logger.debug(f"准备请求: {url}") try: response = await client.get(url) response.raise_for_status() logger.info(f"请求成功: {url}, 状态码: {response.status_code}") return response.text except httpx.TimeoutException: logger.warning(f"请求超时: {url}") raise except httpx.HTTPStatusError as e: logger.error(f"HTTP错误: {url}, 状态码: {e.response.status_code}") raise except Exception as e: logger.exception(f"未知错误: {url}") # exception() 会自动记录堆栈信息 raise ``` #### 爬虫日志实战示例 以下是一个名言网站爬虫的日志使用示例： ```python # quotes_crawler.py - 名言网站爬虫日志示例 from loguru import logger import httpx async def fetch_quotes(client: httpx.AsyncClient, page: int = 1): """ 获取名言列表 展示在实际爬虫中如何使用日志 """ url = f"https://quotes.toscrape.com/page/{page}/" logger.debug(f"[QuotesCrawler] 请求参数: page={page}") try: response = await client.get(url) # HTTP状态码处理 if response.status_code == 200: logger.info(f"[QuotesCrawler] 获取成功: 第{page}页") return response.text elif response.status_code == 404: logger.warning(f"[QuotesCrawler] 页面不存在: 第{page}页") return None elif response.status_code == 429: logger.warning(f"[QuotesCrawler] 触发频率限制，需要等待") raise Exception("触发频率限制") else: logger.error(f"[QuotesCrawler] HTTP错误: status_code={response.status_code}") return None except httpx.TimeoutException: logger.warning(f"[QuotesCrawler] 请求超时: {url}") raise except Exception as e: logger.exception(f"[QuotesCrawler] 未知错误: {e}") raise ``` > **日志前缀约定**：在爬虫项目中，我们使用 `[模块名]` 作为日志前缀，便于在日志文件中快速定位问题来源，如 `[QuotesCrawler]`、`[DataParser]`、`[DataStore]`。 --- ## 配置管理 ### 为什么需要配置管理 硬编码的配置散落在代码各处是一种糟糕的实践： ```python # 反面示例 - 不要这样做 client = httpx.AsyncClient(timeout=30) db_url = "mysql://root:password123@localhost/crawler" API_KEY = "sk-xxxxxxxxxxxx" ``` 这种做法的问题： - 敏感信息（密码、API密钥）暴露在代码中 - 不同环境（开发、测试、生产）需要修改代码 - 配置分散，难以统一管理 ### pydantic-settings 简介 `pydantic-settings` 是 Pydantic 的扩展，专门用于处理应用配置。它支持： - 从环境变量读取配置 - 从 `.env` 文件读取配置 - 配置值的类型验证 - 敏感信息的安全处理 安装： ```bash pip install pydantic-settings ``` ### 配置管理流程 ```mermaid flowchart LR subgraph 配置来源 env[环境变量
CRAWLER_XXX] dotenv[.env 文件
敏感信息] default[代码默认值] end subgraph pydantic-settings validate[类型验证] merge[配置合并] end subgraph 使用 settings[settings 单例] app[应用程序] end env --> validate dotenv --> validate default --> validate validate --> merge merge --> settings settings --> app ``` ### 基本使用 创建配置类： ```python # config/settings.py from pydantic_settings import BaseSettings from pydantic import Field from typing import Optional class CrawlerSettings(BaseSettings): """爬虫配置""" # 基础配置 debug: bool = Field(default=False, description="调试模式") log_level: str = Field(default="INFO", description="日志级别") # 请求配置 request_timeout: int = Field(default=30, description="请求超时时间(秒)") max_retries: int = Field(default=3, description="最大重试次数") retry_delay: float = Field(default=1.0, description="重试延迟(秒)") # 并发配置 max_concurrent: int = Field(default=10, description="最大并发数") # 数据库配置 db_host: str = Field(default="localhost", description="数据库主机") db_port: int = Field(default=3306, description="数据库端口") db_user: str = Field(default="root", description="数据库用户") db_password: str = Field(default="", description="数据库密码") db_name: str = Field(default="crawler", description="数据库名") # 代理配置 proxy_url: Optional[str] = Field(default=None, description="代理地址") class Config: env_file = ".env" env_file_encoding = "utf-8" # 环境变量前缀，如 CRAWLER_DEBUG=true env_prefix = "CRAWLER_" # 全局配置实例 settings = CrawlerSettings() ``` ### 爬虫配置实战示例 以下是名言网站爬虫的配置示例，展示如何组织配置文件： ```python # config/settings.py - 爬虫配置 from pydantic_settings import BaseSettings from pydantic import Field from typing import Optional from enum import Enum class StorageType(str, Enum): """存储类型""" JSON = "json" CSV = "csv" class CrawlerSettings(BaseSettings): """爬虫配置""" # 基础配置 app_name: str = "QuotesCrawler" debug: bool = False log_level: str = "INFO" # 请求配置 request_timeout: int = 30 max_retries: int = 3 retry_delay: float = 1.0 # 爬虫配置 base_url: str = "https://quotes.toscrape.com" max_pages: int = 10 max_concurrency: int = 3 crawl_delay_min: float = 0.5 crawl_delay_max: float = 1.5 # 存储配置 storage_type: StorageType = StorageType.JSON output_dir: str = "./output" class Config: env_file = ".env" env_prefix = "CRAWLER_" # 全局配置实例 settings = CrawlerSettings() ``` ```python # config/constants.py - 常量配置 """爬虫常量配置""" # 默认请求头 DEFAULT_HEADERS = { "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " "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", } # HTTP 状态码 HTTP_OK = 200 HTTP_NOT_FOUND = 404 HTTP_TOO_MANY_REQUESTS = 429 HTTP_SERVER_ERROR = 500 ``` ### 使用 .env 文件 创建 `.env` 文件（记得添加到 `.gitignore`）： ```env # .env CRAWLER_DEBUG=true CRAWLER_LOG_LEVEL=DEBUG CRAWLER_REQUEST_TIMEOUT=60 CRAWLER_DB_PASSWORD=your_secret_password CRAWLER_PROXY_URL=http://127.0.0.1:7890 ``` 在代码中使用配置： ```python from config.settings import settings from loguru import logger # 根据配置设置日志级别 logger.add("logs/app.log", level=settings.log_level) # 使用配置创建客户端 client = httpx.AsyncClient( timeout=settings.request_timeout, proxies=settings.proxy_url ) logger.info(f"爬虫启动，调试模式: {settings.debug}") ``` ### 多环境配置 对于需要区分开发、测试、生产环境的项目，可以这样组织： ```python # config/settings.py import os from pydantic_settings import BaseSettings from pydantic import Field class BaseConfig(BaseSettings): """基础配置""" env: str = Field(default="development", description="运行环境") debug: bool = False log_level: str = "INFO" class Config: env_file = ".env" env_prefix = "CRAWLER_" class DevelopmentConfig(BaseConfig): """开发环境配置""" debug: bool = True log_level: str = "DEBUG" class ProductionConfig(BaseConfig): """生产环境配置""" debug: bool = False log_level: str = "WARNING" def get_settings() -> BaseConfig: """根据环境变量返回对应的配置""" env = os.getenv("CRAWLER_ENV", "development") config_map = { "development": DevelopmentConfig, "production": ProductionConfig, } config_class = config_map.get(env, DevelopmentConfig) return config_class() settings = get_settings() ``` --- ## 异常处理与重试机制 ### 异常处理流程 ```mermaid flowchart TD request[发起请求] --> response{响应状态} response -->|成功| parse[解析数据] response -->|超时| retry{重试?} response -->|HTTP错误| check_code{检查错误码} retry -->|是| request retry -->|否| fail[记录失败] check_code -->|429| rate_limit[触发限流
等待后重试] check_code -->|412| risk_control[触发风控
切换IP/降速] check_code -->|401/403| auth_error[认证失败
重新登录] check_code -->|其他| fail parse --> save[保存数据] rate_limit --> request risk_control --> request ``` ### 自定义异常类 为爬虫项目定义专门的异常类，便于区分和处理不同类型的错误： ```python # exceptions.py class CrawlerException(Exception): """爬虫基础异常类""" pass class RequestException(CrawlerException): """请求相关异常""" pass class ParseException(CrawlerException): """解析相关异常""" pass class StorageException(CrawlerException): """存储相关异常""" pass class RateLimitException(RequestException): """触发速率限制""" pass class IPBlockedException(RequestException): """IP被封禁""" pass class LoginRequiredException(CrawlerException): """需要登录""" pass ``` ### 爬虫异常定义实战 以下是通用爬虫的异常类设计示例： ```python # exceptions/crawler_exceptions.py """爬虫异常类""" class CrawlerException(Exception): """爬虫基础异常""" pass class RequestException(CrawlerException): """请求相关异常""" pass class ParseException(CrawlerException): """解析相关异常""" pass class StorageException(CrawlerException): """存储相关异常""" pass class RateLimitException(RequestException): """触发速率限制""" pass class IPBlockedException(RequestException): """IP被封禁""" pass class PageNotFoundException(RequestException): """页面不存在""" pass # HTTP 状态码映射 HTTP_ERROR_MESSAGES = { 400: "请求参数错误", 401: "未授权访问", 403: "禁止访问", 404: "页面不存在", 429: "请求过于频繁", 500: "服务器内部错误", 502: "网关错误", 503: "服务暂时不可用", } ``` ### 使用 tenacity 实现重试 `tenacity` 是一个强大的重试库，比手写重试逻辑更加优雅和可靠。 安装： ```bash pip install tenacity ``` 基本使用： ```python from tenacity import ( retry, stop_after_attempt, wait_exponential, retry_if_exception_type, before_sleep_log ) from loguru import logger import httpx from exceptions import RequestException, RateLimitException @retry( stop=stop_after_attempt(3), # 最多重试3次 wait=wait_exponential(multiplier=1, max=10), # 指数退避，最长等待10秒 retry=retry_if_exception_type(( # 只对特定异常重试 httpx.TimeoutException, httpx.ConnectError, RequestException )), before_sleep=before_sleep_log(logger, "WARNING") # 重试前记录日志 ) async def fetch_with_retry(client: httpx.AsyncClient, url: str) -> str: """带重试的请求函数""" response = await client.get(url) # 检查是否触发速率限制 if response.status_code == 429: raise RateLimitException("触发速率限制") response.raise_for_status() return response.text ``` ### 更复杂的重试策略 ```python from tenacity import ( retry, stop_after_attempt, wait_random_exponential, retry_if_exception_type, RetryError ) def create_retry_decorator(max_attempts: int = 3, max_wait: int = 60): """创建可配置的重试装饰器""" return retry( stop=stop_after_attempt(max_attempts), wait=wait_random_exponential(multiplier=1, max=max_wait), retry=retry_if_exception_type(( httpx.TimeoutException, httpx.ConnectError, httpx.HTTPStatusError, )), reraise=True # 重试用尽后重新抛出异常 ) # 使用 @create_retry_decorator(max_attempts=5, max_wait=30) async def fetch_important_data(client, url): """重要数据获取，使用更多重试次数""" response = await client.get(url) response.raise_for_status() return response.json() ``` ### 全局异常处理 在爬虫主程序中实现全局异常处理： ```python import asyncio from loguru import logger from exceptions import CrawlerException, IPBlockedException async def run_crawler(): """爬虫主程序""" try: # 爬虫逻辑 await crawl_all_pages() except IPBlockedException as e: logger.critical(f"IP被封禁，程序终止: {e}") # 可以在这里触发告警通知 raise except CrawlerException as e: logger.error(f"爬虫异常: {e}") raise except asyncio.CancelledError: logger.warning("任务被取消") raise except Exception as e: logger.exception(f"未预期的异常: {e}") raise finally: logger.info("爬虫任务结束，执行清理工作...") # 清理资源 await cleanup() if __name__ == "__main__": try: asyncio.run(run_crawler()) except KeyboardInterrupt: logger.info("用户中断程序") except Exception as e: logger.critical(f"程序异常退出: {e}") exit(1) ``` --- ## 项目目录结构 ### 项目架构图 ```mermaid graph TB subgraph 入口层 main[main.py
程序入口] end subgraph 核心模块 config[config/
配置管理] crawler[crawler/
爬虫逻辑] store[store/
数据存储] end subgraph 功能模块 login[login/
登录认证] client[client/
API客户端] analysis[analysis/
数据分析] end subgraph 基础模块 core[core/
浏览器管理] tools[tools/
工具函数] models[models/
数据模型] exceptions[exceptions/
异常定义] end main --> config main --> crawler main --> store main --> analysis crawler --> login crawler --> client login --> core client --> tools tools --> models ``` ### 推荐的目录结构 一个工程化的爬虫项目应该有清晰的目录结构： ``` my_crawler/ ├── config/ # 配置模块 │ ├── __init__.py │ └── settings.py # 配置定义 ├── core/ # 核心模块 │ ├── __init__.py │ ├── client.py # HTTP客户端封装 │ └── retry.py # 重试策略 ├── crawler/ # 爬虫模块 │ ├── __init__.py │ ├── base.py # 爬虫基类 │ └── xxx_crawler.py # 具体爬虫实现 ├── parser/ # 解析模块 │ ├── __init__.py │ └── xxx_parser.py # 页面解析器 ├── store/ # 存储模块 │ ├── __init__.py │ ├── base.py # 存储基类 │ ├── mysql.py # MySQL存储 │ └── json_store.py # JSON文件存储 ├── models/ # 数据模型 │ ├── __init__.py │ └── xxx_model.py # Pydantic模型定义 ├── exceptions/ # 异常定义 │ ├── __init__.py │ └── crawler_exceptions.py ├── utils/ # 工具函数 │ ├── __init__.py │ └── helpers.py ├── logs/ # 日志目录 ├── data/ # 数据输出目录 ├── tests/ # 测试目录 │ └── test_xxx.py ├── .env # 环境变量（不提交到git） ├── .env.example # 环境变量示例 ├── .gitignore ├── requirements.txt # 依赖列表 ├── main.py # 程序入口 └── README.md # 项目说明 ``` ### 模块划分原则 1. **单一职责**：每个模块只负责一件事 2. **高内聚低耦合**：相关代码放在一起，模块间依赖最小化 3. **抽象与实现分离**：定义基类/接口，方便扩展和替换 --- ## 实战案例：工程化改造 让我们将入门教程中的爬虫进行工程化改造。 ### 改造前（入门教程版本） ```python # 入门教程的简单版本 import httpx from parsel import Selector async def crawl(): async with httpx.AsyncClient() as client: response = await client.get("https://example.com") print(f"状态码: {response.status_code}") # 解析... # 存储... ``` ### 改造后（工程化版本） 项目结构： ``` refactored_crawler/ ├── config/ │ ├── __init__.py │ └── settings.py ├── exceptions.py ├── logger_config.py ├── client.py ├── crawler.py └── main.py ``` 核心代码示例： ```python # logger_config.py import sys from loguru import logger from config.settings import settings def setup_logger(): """配置日志系统""" # 移除默认处理器 logger.remove() # 控制台输出 logger.add( sys.stderr, level=settings.log_level, format="{time:YYYY-MM-DD HH:mm:ss} | " "{level: <8} | " "{name}:{function}:{line} - " "{message}" ) # 文件输出 logger.add( f"logs/crawler_{settings.env}.log", rotation="00:00", retention="7 days", level="DEBUG", encoding="utf-8" ) return logger ``` ```python # client.py import httpx from tenacity import retry, stop_after_attempt, wait_exponential from loguru import logger from config.settings import settings from exceptions import RequestException class CrawlerClient: """封装的HTTP客户端""" def __init__(self): self.client = httpx.AsyncClient( timeout=settings.request_timeout, headers={ "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36" } ) @retry( stop=stop_after_attempt(settings.max_retries), wait=wait_exponential(multiplier=1, max=10) ) async def get(self, url: str) -> httpx.Response: """发送GET请求，带重试""" logger.debug(f"请求: {url}") response = await self.client.get(url) response.raise_for_status() logger.info(f"请求成功: {url}") return response async def close(self): """关闭客户端""" await self.client.aclose() logger.debug("HTTP客户端已关闭") ``` ```python # main.py import asyncio from loguru import logger from logger_config import setup_logger from crawler import BBSCrawler from config.settings import settings async def main(): """主程序入口""" # 初始化日志 setup_logger() logger.info(f"爬虫启动 - 环境: {settings.env}, 调试模式: {settings.debug}") crawler = BBSCrawler() try: await crawler.run() except Exception as e: logger.exception(f"爬虫运行异常: {e}") raise finally: await crawler.cleanup() logger.info("爬虫运行结束") if __name__ == "__main__": try: asyncio.run(main()) except KeyboardInterrupt: logger.info("用户中断") ``` --- ## 本章小结 本章我们学习了爬虫工程化开发的核心内容： 1. **日志系统**：使用 loguru 替代 print，实现分级日志、日志轮转和持久化 2. **配置管理**：使用 pydantic-settings 统一管理配置，支持环境变量和 .env 文件 3. **异常处理**：自定义异常类，使用 tenacity 实现优雅的重试机制 4. **项目结构**：遵循单一职责原则，合理划分模块 这些工程化实践将贯穿整个进阶教程，是后续学习的基础。 --- ## 下一章预告 下一章我们将学习「反爬虫对抗基础：请求伪装」。主要内容包括： - 常见的反爬虫检测手段 - User-Agent 随机轮换 - 完整的请求头伪装 - 使用 curl_cffi 模拟浏览器指纹 - 智能速率控制 这将是我们与反爬虫斗争的第一步，让我们的爬虫不再那么容易被识别和封禁。 --- ### 爬虫进价 02 反爬虫对抗基础 请求伪装 (`docs/爬虫进价/02_反爬虫对抗基础_请求伪装.md`) # 反爬虫对抗基础：请求伪装 > 当你开始爬取一些有价值的网站时，很快就会发现：不是所有网站都欢迎爬虫。反爬虫与反反爬虫的对抗，是爬虫工程师必须面对的课题。本章我们将学习最基础也最重要的反爬对抗技术——请求伪装。 > **学习目标**：掌握 User-Agent 轮换、请求头伪装、TLS 指纹模拟和速率控制等核心技术，让你的爬虫不易被识别。 ## 反爬虫机制概述 ### 反爬检测流程 ```mermaid flowchart TD request[爬虫发起请求] --> check1{User-Agent检测} check1 -->|异常UA| block1[直接拒绝] check1 -->|正常UA| check2{请求头检测} check2 -->|缺少必要头| block2[返回错误] check2 -->|请求头正常| check3{频率检测} check3 -->|频率过高| block3[触发限流 429] check3 -->|频率正常| check4{Cookie检测} check4 -->|无Cookie/过期| block4[要求登录 401] check4 -->|Cookie有效| check5{其他检测} check5 -->|检测失败| block5[返回 403] check5 -->|检测通过| success[返回数据] style block1 fill:#f66 style block2 fill:#f66 style block3 fill:#f96 style block4 fill:#f96 style block5 fill:#f66 style success fill:#6f6 ``` ### 为什么网站要反爬虫 在开始学习反爬技术之前，我们需要理解网站为什么要反爬虫： 1. **保护数据资产**：数据是有价值的，网站不希望被批量获取 2. **保护服务器资源**：爬虫会消耗服务器带宽和计算资源 3. **防止恶意行为**：如价格监控、竞品分析、数据倒卖等 4. **合规要求**：某些数据有法律保护要求 ### 常见的反爬虫检测手段 ```mermaid graph LR subgraph 低难度 ua[UA检测] headers[请求头检测] end subgraph 中难度 freq[频率检测] cookie[Cookie检测] sign[API签名检测] end subgraph 高难度 tls[TLS指纹检测] js[JS环境检测] captcha[验证码] end ua --> headers --> freq --> cookie --> sign --> tls --> js --> captcha ``` | 检测类型 | 检测方式 | 难度 | 常见应对 | |---------|---------|------|---------| | 请求特征检测 | User-Agent、请求头完整性 | 低 | 伪装请求头 | | 行为特征检测 | 访问频率、访问路径 | 中 | 速率控制、随机延迟 | | API签名检测 | 参数签名验证 | 中 | 逆向签名算法 | | Cookie检测 | 登录状态验证 | 中 | 登录获取Cookie | | 浏览器指纹检测 | JS 环境、Canvas、WebGL | 高 | 使用真实浏览器 | | 验证码检测 | 图片验证码、滑块验证码 | 高 | OCR、打码平台 | 本章主要聚焦于**请求特征检测**的对抗，这是最基础的反爬手段。 --- ## User-Agent 策略 ### 什么是 User-Agent User-Agent（简称 UA）是 HTTP 请求头中的一个字段，用于标识发起请求的客户端类型。例如： ``` Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36 ``` 这个字符串包含了： - 浏览器类型（Chrome） - 浏览器版本（120.0.0.0） - 操作系统（Windows 10 64位） - 渲染引擎（AppleWebKit/537.36） ### 为什么要轮换 User-Agent 如果你的爬虫始终使用同一个 UA，会有以下风险： 1. **特征明显**：Python 默认的 UA 是 `python-requests/2.x.x`，一眼就能识别 2. **容易被追踪**：同一 UA 的大量请求会被关联分析 3. **容易被封禁**：一旦被识别，可以按 UA 封禁 ### 实现 User-Agent 轮换 #### 方法一：手动维护 UA 列表 ``` /* Detailed source-code truncated for AI context efficiency. */ ``` #### 方法二：使用 fake-useragent 库 ```bash pip install fake-useragent ``` ```python from fake_useragent import UserAgent # 创建 UserAgent 对象 ua = UserAgent() # 获取随机 UA print(ua.random) # 完全随机 print(ua.chrome) # 随机 Chrome UA print(ua.firefox) # 随机 Firefox UA print(ua.safari) # 随机 Safari UA ``` #### 实现 UA 轮换器 ``` /* Detailed source-code truncated for AI context efficiency. */ ``` --- ## 请求头完整伪装 ### 为什么仅有 User-Agent 不够 很多网站不仅检测 UA，还会检测其他请求头字段。真实浏览器的请求头是非常丰富的： ```http GET /api/data HTTP/1.1 Host: www.example.com User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Accept: text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8 Accept-Language: zh-CN,zh;q=0.9,en;q=0.8 Accept-Encoding: gzip, deflate, br Connection: keep-alive Upgrade-Insecure-Requests: 1 Sec-Fetch-Dest: document Sec-Fetch-Mode: navigate Sec-Fetch-Site: none Sec-Fetch-User: ?1 Cache-Control: max-age=0 ``` 而 Python 默认的请求头非常简陋，很容易被识别。 ### 构建完整的请求头 ``` /* Detailed source-code truncated for AI context efficiency. */ ``` ### Referer 的正确设置 Referer 表示当前请求是从哪个页面发起的。正确设置 Referer 很重要： ```python async def crawl_with_referer(client, list_url: str, detail_urls: list): """演示正确设置 Referer""" headers_builder = HeadersBuilder() # 访问列表页 list_headers = headers_builder.build() response = await client.get(list_url, headers=list_headers) # 访问详情页时，Referer 应该是列表页 for detail_url in detail_urls: detail_headers = headers_builder.build(referer=list_url) response = await client.get(detail_url, headers=detail_headers) ``` ### 请求头实战配置 以下是一个通用的请求头构建器，可以根据目标网站进行定制： ``` /* Detailed source-code truncated for AI context efficiency. */ ``` > **请求头配置要点**： > 1. **Referer**：很多网站会检查，应设置为合理的来源页面 > 2. **Sec-Fetch 系列**：现代浏览器标准头，建议保持完整 > 3. **Accept**：页面请求和 API 请求的 Accept 头不同 --- ## 使用 curl_cffi 模拟浏览器指纹 ### 什么是 TLS 指纹 除了 HTTP 请求头，服务器还可以通过 TLS（HTTPS 握手）的特征来识别客户端。不同的客户端（浏览器、Python requests、curl 等）在 TLS 握手时会展现不同的特征： - 支持的加密套件顺序 - 支持的 TLS 扩展 - 椭圆曲线参数 Python 的 `requests` 和 `httpx` 使用的 TLS 指纹与真实浏览器差异很大，容易被识别。 ### curl_cffi 简介 `curl_cffi` 是一个可以模拟各种浏览器 TLS 指纹的 HTTP 客户端库。 安装： ```bash pip install curl_cffi ``` ### 基本使用 ```python from curl_cffi import requests # 模拟 Chrome 浏览器 response = requests.get( "https://tls.browserleaks.com/json", impersonate="chrome120" # 模拟 Chrome 120 ) print(response.json()) # 支持的浏览器指纹 # chrome99, chrome100, chrome101, ..., chrome120 # chrome99_android # edge99, edge101 # safari15_3, safari15_5, safari17_0 ``` ### 异步使用 ```python from curl_cffi.requests import AsyncSession async def fetch_with_curl_cffi(): """使用 curl_cffi 的异步请求""" async with AsyncSession(impersonate="chrome120") as session: response = await session.get("https://httpbin.org/headers") print(response.json()) ``` ### 封装 curl_cffi 客户端 ``` /* Detailed source-code truncated for AI context efficiency. */ ``` --- ## 速率控制 ### 为什么需要速率控制 即使你的请求伪装得再好，如果访问频率过高，也会触发反爬机制。人类的浏览行为是有一定节奏的，而机器的请求往往过于规律或过于密集。 ### 基本延迟策略 ```python import asyncio import random async def crawl_with_delay(urls: list, min_delay: float = 1.0, max_delay: float = 3.0): """ 带随机延迟的爬取 Args: urls: URL 列表 min_delay: 最小延迟（秒） max_delay: 最大延迟（秒） """ for url in urls: # 发送请求 response = await fetch(url) # 随机延迟 delay = random.uniform(min_delay, max_delay) await asyncio.sleep(delay) ``` ### 令牌桶限速器 令牌桶算法是一种更精确的限速方式，它允许一定程度的突发请求，同时保持长期平均速率。 ``` /* Detailed source-code truncated for AI context efficiency. */ ``` ### 使用 asyncio.Semaphore 控制并发 ```python import asyncio class ConcurrencyLimiter: """并发限制器""" def __init__(self, max_concurrent: int = 10): """ 初始化并发限制器 Args: max_concurrent: 最大并发数 """ self.semaphore = asyncio.Semaphore(max_concurrent) async def run_with_limit(self, coro): """在并发限制下执行协程""" async with self.semaphore: return await coro async def crawl_concurrent(urls: list, max_concurrent: int = 5): """ 带并发限制的批量爬取 Args: urls: URL 列表 max_concurrent: 最大并发数 """ semaphore = asyncio.Semaphore(max_concurrent) async def fetch_with_limit(url): async with semaphore: return await fetch(url) # 并发执行，但同时最多 max_concurrent 个请求 tasks = [fetch_with_limit(url) for url in urls] results = await asyncio.gather(*tasks, return_exceptions=True) return results ``` --- ## HTTP 错误处理 ### HTTP 状态码处理 在爬取网站时，需要处理各种 HTTP 状态码： ```mermaid flowchart LR response[HTTP响应] --> check{检查状态码} check -->|200| success[成功
正常处理数据] check -->|401| login[需要认证
登录获取凭证] check -->|403| forbidden[禁止访问
检查请求头] check -->|429| ratelimit[频率限制
降低速率] check -->|404| notfound[资源不存在] login --> relogin[获取认证凭证] ratelimit --> wait[等待后重试] forbidden --> fixheaders[修复请求头] ``` ```python # error_handler.py - HTTP错误处理 from loguru import logger async def handle_response(response, url: str): """ 处理 HTTP 响应 Args: response: HTTP 响应对象 url: 请求的 URL """ status_code = response.status_code if status_code == 200: logger.info(f"请求成功: {url}") return response elif status_code == 401: logger.warning(f"需要认证: {url}") raise Exception("需要登录或认证凭证已过期") elif status_code == 403: logger.error(f"禁止访问: {url}") raise Exception("访问被禁止，请检查请求头配置") elif status_code == 429: logger.warning(f"触发频率限制: {url}") raise Exception("请求过于频繁，请降低访问速率") elif status_code == 404: logger.warning(f"资源不存在: {url}") return None else: logger.error(f"HTTP错误 {status_code}: {url}") raise Exception(f"HTTP错误: {status_code}") ``` --- ## 实战案例：完整的请求伪装爬虫 让我们把本章学到的技术整合成一个完整的通用爬虫示例： ``` /* Detailed source-code truncated for AI context efficiency. */ ``` --- ## 本章小结 本章我们学习了反爬虫对抗的基础技术——请求伪装： 1. **User-Agent 轮换**：使用真实浏览器 UA，随机轮换避免被追踪 2. **请求头完整伪装**：构建与真实浏览器一致的完整请求头 3. **TLS 指纹模拟**：使用 curl_cffi 模拟浏览器的 TLS 指纹 4. **速率控制**：使用随机延迟和令牌桶算法控制请求频率 5. **HTTP 错误处理**：正确处理各种 HTTP 状态码 这些技术可以应对大部分基于请求特征的反爬检测。 --- ## 下一章预告 下一章我们将学习「代理 IP 的使用与管理」。主要内容包括： - 代理 IP 的类型和选择 - 代理池的设计与实现 - 代理的有效性检测和淘汰机制 - 代理与爬虫的集成 代理 IP 是突破 IP 封禁的重要手段，也是大规模爬虫必不可少的基础设施。 --- ### 爬虫进价 03 代理IP的使用与管理 (`docs/爬虫进价/03_代理IP的使用与管理.md`) # 代理 IP 的使用与管理 > 当你的爬虫遭遇 IP 封禁时，代理 IP 就成了必不可少的工具。本章将深入讲解代理 IP 的原理、类型选择，以及如何设计和实现一个实用的代理池管理系统。 ## 代理 IP 基础 ### 什么是代理 IP 代理服务器（Proxy Server）是一种位于客户端和目标服务器之间的中间服务器。当你通过代理发送请求时： 1. 你的请求先发送到代理服务器 2. 代理服务器代替你向目标服务器发送请求 3. 目标服务器将响应返回给代理服务器 4. 代理服务器再将响应转发给你 这样，目标服务器看到的是代理服务器的 IP，而不是你的真实 IP。 ### 代理类型详解 #### 按协议分类 | 类型 | 特点 | 使用场景 | |------|------|----------| | HTTP 代理 | 只支持 HTTP 协议 | 普通网页爬取 | | HTTPS 代理 | 支持 HTTPS 协议 | 加密网站爬取 | | SOCKS4 代理 | 支持 TCP 连接 | 需要更底层控制 | | SOCKS5 代理 | 支持 TCP/UDP，可认证 | 最灵活的代理类型 | #### 按匿名度分类 | 类型 | 特点 | 识别难度 | |------|------|----------| | 透明代理 | 目标服务器能看到你的真实 IP | 容易被识别 | | 匿名代理 | 隐藏真实 IP，但暴露代理身份 | 中等 | | 高匿代理 | 完全隐藏真实 IP 和代理身份 | 较难识别 | 对于爬虫来说，**高匿代理**是首选。 #### 按来源分类 | 类型 | 优点 | 缺点 | |------|------|------| | 免费代理 | 成本为零 | 不稳定、速度慢、可用率低 | | 付费代理 | 稳定、速度快、可用率高 | 需要成本 | | 自建代理 | 完全可控 | 需要服务器资源 | ### 代理提供商选择指南 选择代理提供商时需要考虑： 1. **IP 质量**：是否是高匿代理，是否已被目标网站封禁 2. **IP 数量**：IP 池的规模 3. **地理分布**：是否覆盖目标地区 4. **稳定性**：连接成功率和响应速度 5. **价格**：按流量计费还是按 IP 数计费 6. **API 支持**：是否提供便捷的 API 获取代理 常见的代理类型： - **API 提取型**：通过 API 获取代理 IP 列表 - **隧道代理型**：固定入口，自动轮换 IP - **动态转发型**：每次请求自动更换 IP --- ## 代理池设计 ### 为什么需要代理池 直接使用单个代理存在以下问题： - 代理可能随时失效 - 单个 IP 容易被封禁 - 无法动态切换和管理 代理池可以解决这些问题： - 统一管理多个代理 - 自动检测代理有效性 - 智能分配和轮换代理 - 记录代理质量评分 ### 代理池架构设计 ```mermaid graph TB subgraph ProxyPoolManager["代理池管理器"] direction TB subgraph Components["核心组件"] direction LR Fetcher["代理获取器
(Fetcher)"] Checker["代理检测器
(Checker)"] Allocator["代理分配器
(Allocator)"] end Storage[("代理存储
(内存 / Redis)")] Fetcher --> Storage Checker --> Storage Allocator --> Storage end style ProxyPoolManager fill:#e8f4f8,stroke:#0288d1 style Storage fill:#fff3e0,stroke:#ff9800 ``` **代理池工作流程**： ```mermaid flowchart LR subgraph 获取阶段 API["代理API"] --> Fetcher["获取器"] Free["免费代理"] --> Fetcher end subgraph 检测阶段 Fetcher --> Checker["检测器"] Checker -->|有效| Pool[("代理池")] Checker -->|无效| Discard["丢弃"] end subgraph 分配阶段 Pool --> Allocator["分配器"] Allocator --> Crawler["爬虫请求"] Crawler -->|成功| Score["评分+1"] Crawler -->|失败| Penalty["评分-1"] Score --> Pool Penalty --> Pool end style Pool fill:#c8e6c9,stroke:#4caf50 style Discard fill:#ffcdd2,stroke:#f44336 ``` ### 核心接口设计 ``` /* Detailed source-code truncated for AI context efficiency. */ ``` --- ## 代理获取器实现 ### 免费代理获取（仅供学习） ```python import httpx from typing import List from loguru import logger class FreeProxyFetcher(IProxyFetcher): """ 免费代理获取器 注意：免费代理质量较差，仅供学习测试使用 """ async def fetch(self) -> List[ProxyInfo]: """从免费代理网站获取代理""" proxies = [] # 示例：从 API 获取（这里用一个示例 API） try: async with httpx.AsyncClient(timeout=10) as client: # 这是一个示例 URL，实际使用时替换为真实的代理 API response = await client.get( "https://api.proxyscrape.com/v2/" "?request=getproxies&protocol=http&timeout=10000&country=all" ) if response.status_code == 200: lines = response.text.strip().split("\n") for line in lines: try: host, port = line.strip().split(":") proxies.append(ProxyInfo( host=host, port=int(port), protocol=ProxyProtocol.HTTP )) except ValueError: continue logger.info(f"获取到 {len(proxies)} 个免费代理") except Exception as e: logger.error(f"获取免费代理失败: {e}") return proxies ``` ### API 代理获取器 ```python class APIProxyFetcher(IProxyFetcher): """ API 代理获取器 从付费代理服务商的 API 获取代理 """ def __init__( self, api_url: str, api_key: Optional[str] = None, count: int = 10 ): """ 初始化 API 代理获取器 Args: api_url: API 地址 api_key: API 密钥 count: 每次获取数量 """ self.api_url = api_url self.api_key = api_key self.count = count async def fetch(self) -> List[ProxyInfo]: """从 API 获取代理""" proxies = [] try: params = {"num": self.count} if self.api_key: params["key"] = self.api_key async with httpx.AsyncClient(timeout=10) as client: response = await client.get(self.api_url, params=params) data = response.json() # 根据实际 API 返回格式解析 # 这里假设返回 {"data": [{"ip": "x.x.x.x", "port": 8080}, ...]} for item in data.get("data", []): proxies.append(ProxyInfo( host=item["ip"], port=item["port"], protocol=ProxyProtocol(item.get("protocol", "http")) )) logger.info(f"从 API 获取到 {len(proxies)} 个代理") except Exception as e: logger.error(f"从 API 获取代理失败: {e}") return proxies ``` --- ## 代理检测器实现 ``` /* Detailed source-code truncated for AI context efficiency. */ ``` --- ## 代理池实现 ``` /* Detailed source-code truncated for AI context efficiency. */ ``` --- ## 代理与爬虫集成 ### 使用 httpx 设置代理 ```python import httpx async def fetch_with_proxy(url: str, proxy_url: str) -> str: """使用代理发送请求""" async with httpx.AsyncClient(proxies=proxy_url, timeout=30) as client: response = await client.get(url) return response.text # 使用示例 proxy = "http://user:pass@127.0.0.1:8080" content = await fetch_with_proxy("https://httpbin.org/ip", proxy) ``` ### 集成代理池的爬虫 ```python class ProxiedCrawler: """ 集成代理池的爬虫 自动管理代理的获取、轮换和报告 """ def __init__(self, proxy_pool: ProxyPool): self.proxy_pool = proxy_pool async def fetch(self, url: str) -> Optional[str]: """使用代理获取页面""" proxy = await self.proxy_pool.get_proxy() if not proxy: logger.warning("无可用代理") return None try: async with httpx.AsyncClient( proxies=proxy.url, timeout=30 ) as client: response = await client.get(url) response.raise_for_status() # 报告成功 await self.proxy_pool.return_proxy(proxy, success=True) return response.text except Exception as e: logger.warning(f"请求失败: {url} - {e}") # 报告失败 await self.proxy_pool.return_proxy(proxy, success=False) return None ``` --- ## 隧道代理使用 隧道代理是一种特殊的代理模式，你只需连接到固定的代理入口，每次请求自动分配不同的 IP。 ```python class TunnelProxyClient: """ 隧道代理客户端 特点：固定入口，自动轮换 IP """ def __init__( self, host: str, port: int, username: str, password: str ): self.proxy_url = f"http://{username}:{password}@{host}:{port}" async def get(self, url: str, **kwargs) -> httpx.Response: """发送请求（自动使用隧道代理）""" async with httpx.AsyncClient( proxies=self.proxy_url, timeout=30 ) as client: return await client.get(url, **kwargs) # 使用示例 tunnel = TunnelProxyClient( host="tunnel.example.com", port=12345, username="your_username", password="your_password" ) # 每次请求自动使用不同 IP response1 = await tunnel.get("https://httpbin.org/ip") response2 = await tunnel.get("https://httpbin.org/ip") ``` --- ## 代理使用最佳实践 ### IP 封禁机制分析 大多数网站都有反爬虫的 IP 封禁机制，常见的触发条件和处理方式： ```mermaid flowchart TD Request["发起请求"] --> RateCheck{"频率检测"} RateCheck -->|正常| UACheck{"UA检测"} RateCheck -->|过快| Block429["429 限流"] UACheck -->|正常| BehaviorCheck{"行为检测"} UACheck -->|异常| Block412["412 风控"] BehaviorCheck -->|正常| Success["正常响应"] BehaviorCheck -->|异常| Block403["403 封禁"] Block429 --> IPMark["IP标记"] Block412 --> IPMark Block403 --> IPMark IPMark --> BlackList["IP黑名单"] style Success fill:#c8e6c9,stroke:#4caf50 style Block429 fill:#fff3e0,stroke:#ff9800 style Block412 fill:#ffcdd2,stroke:#f44336 style Block403 fill:#ffcdd2,stroke:#f44336 style BlackList fill:#ffcdd2,stroke:#f44336 ``` **常见 IP 封禁特点**： | 触发条件 | 响应码 | 封禁时长 | 解封方式 | |---------|--------|---------|---------| | 请求频率过高 | 429 | 几分钟~1小时 | 降低频率后自动解封 | | 风控检测触发 | 403 | 数小时~1天 | 需更换IP | | 严重违规 | 403/IP拉黑 | 数天~永久 | 需更换IP | ### 代理有效性检测器 使用 httpbin.org 等测试服务来验证代理的可用性： ``` /* Detailed source-code truncated for AI context efficiency. */ ``` ### 代理爬虫完整示例 下面展示一个完整的代理爬虫示例，使用 httpbin.org 作为测试目标： ``` /* Detailed source-code truncated for AI context efficiency. */ ``` ### 代理使用最佳实践 代理使用的一些通用建议： ```mermaid graph LR subgraph 代理选择 A1["高匿代理"] --> A2["住宅IP优先"] A2 --> A3["国内节点"] end subgraph 请求策略 B1["控制频率"] --> B2["随机延迟"] B2 --> B3["失败轮换"] end subgraph 风控规避 C1["完整请求头"] --> C2["Cookie携带"] C2 --> C3["行为模拟"] end A3 --> B1 B3 --> C1 style A1 fill:#e3f2fd,stroke:#2196f3 style B1 fill:#fff3e0,stroke:#ff9800 style C1 fill:#e8f5e9,stroke:#4caf50 ``` **关键建议**： 1. **代理类型**：大型网站对代理检测严格，推荐使用高匿住宅代理 2. **请求频率**：单IP建议 0.5-1 秒/请求，避免触发频率限制 3. **完整请求头**：必须携带 User-Agent、Accept 等头信息 4. **Cookie 携带**：部分 API 需要登录态，代理请求也要携带 Cookie 5. **失败处理**：遇到 403/429 立即切换代理，避免 IP 被永久封禁 --- ## 本章小结 本章我们学习了代理 IP 的完整知识体系： 1. **代理基础**：代理类型、匿名度、来源选择 2. **代理池设计**：获取器、检测器、分配器的接口设计 3. **核心实现**：代理获取、有效性检测、智能分配 4. **爬虫集成**：httpx 代理设置、自动轮换、隧道代理 5. **最佳实践**：代理检测、错误处理、使用建议 代理 IP 是大规模爬虫的基础设施，合理使用可以有效应对 IP 封禁。 --- ## 下一章预告 下一章我们将学习「Playwright 浏览器自动化入门」。主要内容包括： - Playwright 的安装和基本使用 - 页面导航和元素定位 - 等待策略和超时处理 - 截图和 PDF 导出 - 爬取 JavaScript 渲染的页面 浏览器自动化是应对复杂反爬的利器，让我们一起探索！ --- ### 爬虫进价 04 Playwright浏览器自动化入门 (`docs/爬虫进价/04_Playwright浏览器自动化入门.md`) # Playwright 浏览器自动化入门 > 当网页内容需要 JavaScript 渲染，或者需要模拟复杂的用户交互时，传统的 HTTP 请求方式就无能为力了。这时候，浏览器自动化工具就成了我们的利器。本章将带你入门 Playwright——一个现代、强大的浏览器自动化库。 ## 为什么选择 Playwright ### 浏览器自动化工具对比 | 特性 | Selenium | Puppeteer | Playwright | |------|----------|-----------|------------| | 语言支持 | 多语言 | JavaScript/TypeScript | 多语言 | | 浏览器支持 | 多浏览器 | 仅 Chromium | Chromium/Firefox/WebKit | | 自动等待 | 手动 | 部分自动 | 完全自动 | | 性能 | 较慢 | 快 | 快 | | API 设计 | 较旧 | 现代 | 最现代 | | 维护状态 | 活跃 | 活跃 | 微软维护，非常活跃 | ### Playwright 的优势 1. **跨浏览器支持**：一套代码支持 Chromium、Firefox、WebKit 2. **自动等待机制**：智能等待元素可交互，减少 flaky 测试 3. **网络拦截**：可以拦截和修改网络请求 4. **移动端模拟**：内置多种设备配置 5. **同步/异步 API**：Python 版本同时支持同步和异步模式 --- ## 安装与配置 ### 安装 Playwright ```bash pip install playwright ``` ### 安装浏览器 ```bash # 安装所有支持的浏览器 playwright install # 或者只安装 Chromium playwright install chromium # 安装 Firefox playwright install firefox # 安装 WebKit（Safari 内核） playwright install webkit ``` ### 验证安装 ```python from playwright.sync_api import sync_playwright with sync_playwright() as p: browser = p.chromium.launch(headless=False) page = browser.new_page() page.goto("https://example.com") print(page.title()) browser.close() ``` --- ## 核心概念 ### Browser/Context/Page 三层模型 ```mermaid graph TB subgraph Browser["Browser 浏览器实例"] direction TB subgraph Context1["Browser Context 1
(登录态环境)"] Page1["Page 1
网站首页"] Page2["Page 2
详情页面"] end subgraph Context2["Browser Context 2
(匿名环境)"] Page3["Page 3
搜索页面"] end end style Browser fill:#e3f2fd,stroke:#1976d2 style Context1 fill:#e8f5e9,stroke:#4caf50 style Context2 fill:#fff3e0,stroke:#ff9800 style Page1 fill:#f3e5f5,stroke:#9c27b0 style Page2 fill:#f3e5f5,stroke:#9c27b0 style Page3 fill:#f3e5f5,stroke:#9c27b0 ``` **三层架构说明**： ```mermaid flowchart LR subgraph Browser层 B["Browser
浏览器进程"] end subgraph Context层 C1["Context
Cookie/Storage"] C2["Context
独立环境"] end subgraph Page层 P1["Page
页面操作"] P2["Page
DOM/Network"] end B --> C1 & C2 C1 --> P1 C2 --> P2 style B fill:#e3f2fd,stroke:#2196f3 style C1 fill:#e8f5e9,stroke:#4caf50 style C2 fill:#e8f5e9,stroke:#4caf50 style P1 fill:#fff3e0,stroke:#ff9800 style P2 fill:#fff3e0,stroke:#ff9800 ``` - **Browser**：浏览器实例，对应一个浏览器进程 - **Browser Context**：独立的浏览器环境，有独立的 Cookie、localStorage 等 - **Page**：一个标签页，在这里进行实际的页面操作 > **应用场景**：可以用不同 Context 分别处理登录态和匿名访问，互不干扰。 ### 同步 vs 异步 API Playwright Python 版本提供两套 API： ```python # 同步 API from playwright.sync_api import sync_playwright with sync_playwright() as p: browser = p.chromium.launch() page = browser.new_page() page.goto("https://example.com") browser.close() # 异步 API from playwright.async_api import async_playwright import asyncio async def main(): async with async_playwright() as p: browser = await p.chromium.launch() page = await browser.new_page() await page.goto("https://example.com") await browser.close() asyncio.run(main()) ``` 对于爬虫项目，推荐使用**异步 API**，可以更好地与其他异步代码配合。 --- ## 页面导航 ### 基本导航 ```python # 打开页面 await page.goto("https://example.com") # 等待特定状态 await page.goto("https://example.com", wait_until="domcontentloaded") # DOM 加载完成 await page.goto("https://example.com", wait_until="load") # 页面完全加载 await page.goto("https://example.com", wait_until="networkidle") # 网络空闲 # 刷新页面 await page.reload() # 前进/后退 await page.go_back() await page.go_forward() ``` ### wait_until 参数说明 | 值 | 说明 | 使用场景 | |---|------|---------| | `domcontentloaded` | DOM 解析完成 | 快速获取页面结构 | | `load` | 所有资源加载完成 | 需要图片等资源 | | `networkidle` | 网络空闲（500ms 无请求） | SPA 应用，动态加载 | | `commit` | 收到第一个响应字节 | 最快，不常用 | --- ## 元素定位 Playwright 提供多种元素定位方式，推荐优先使用语义化的定位器。 ### 推荐：使用 Locator ```python # 通过文本定位 page.get_by_text("登录") page.get_by_text("登录", exact=True) # 精确匹配 # 通过角色定位（推荐用于按钮、链接等） page.get_by_role("button", name="提交") page.get_by_role("link", name="首页") page.get_by_role("textbox", name="用户名") # 通过标签定位 page.get_by_label("用户名") page.get_by_label("密码") # 通过占位符定位 page.get_by_placeholder("请输入用户名") # 通过 test-id 定位（需要在 HTML 中添加 data-testid 属性） page.get_by_test_id("submit-button") ``` ### CSS 选择器 ```python # CSS 选择器 page.locator("div.container") page.locator("#login-form") page.locator("input[type='text']") page.locator("div.item:nth-child(2)") ``` ### XPath ```python # XPath page.locator("//div[@class='container']") page.locator("//button[contains(text(), '提交')]") page.locator("//input[@id='username']") ``` ### 组合定位 ```python # 链式定位 page.locator("div.container").locator("button.submit") # 过滤 page.locator("div.item").filter(has_text="特价") page.locator("div.item").filter(has=page.get_by_role("button")) # 第 N 个元素 page.locator("div.item").nth(0) # 第一个 page.locator("div.item").first # 第一个 page.locator("div.item").last # 最后一个 ``` --- ## 交互操作 ### 点击 ```python # 基本点击 await page.click("button.submit") await page.locator("button.submit").click() # 双击 await page.dblclick("div.item") # 右键点击 await page.click("div.item", button="right") # 点击位置 await page.click("div.map", position={"x": 100, "y": 200}) # 强制点击（忽略可见性检查） await page.click("button.hidden", force=True) ``` ### 输入 ```python # 输入文本 await page.fill("input#username", "myuser") await page.locator("input#password").fill("mypassword") # 逐字符输入（模拟真实打字） await page.type("input#search", "hello world", delay=100) # 清空后输入 await page.fill("input#search", "") await page.fill("input#search", "new text") ``` ### 选择 ```python # 下拉选择（select 元素） await page.select_option("select#country", "china") await page.select_option("select#country", value="cn") await page.select_option("select#country", label="中国") # 多选 await page.select_option("select#tags", ["tag1", "tag2"]) # 复选框/单选框 await page.check("input#agree") await page.uncheck("input#newsletter") await page.set_checked("input#agree", True) ``` ### 键盘操作 ```python # 按键 await page.keyboard.press("Enter") await page.keyboard.press("Tab") await page.keyboard.press("Control+A") await page.keyboard.press("Control+C") # 输入文本 await page.keyboard.type("Hello World") # 组合键 await page.keyboard.down("Shift") await page.keyboard.press("ArrowDown") await page.keyboard.up("Shift") ``` ### 鼠标操作 ```python # 移动鼠标 await page.mouse.move(100, 200) # 点击 await page.mouse.click(100, 200) await page.mouse.dblclick(100, 200) # 拖拽 await page.mouse.move(100, 200) await page.mouse.down() await page.mouse.move(300, 400) await page.mouse.up() # 悬停 await page.hover("div.dropdown") ``` --- ## 等待策略 ### 自动等待 Playwright 的一大优势是**自动等待**。当你执行操作时，它会自动等待元素满足条件： ```python # 点击时自动等待元素可见、可交互 await page.click("button.submit") # fill 时自动等待元素可编辑 await page.fill("input#name", "value") ``` ### 显式等待 有时需要显式等待某些条件： ```python # 等待元素出现 await page.wait_for_selector("div.content") await page.wait_for_selector("div.content", state="visible") await page.wait_for_selector("div.content", state="hidden") # 等待页面状态 await page.wait_for_load_state("networkidle") await page.wait_for_load_state("domcontentloaded") # 等待 URL 变化 await page.wait_for_url("**/success") await page.wait_for_url(lambda url: "success" in url) # 等待特定时间 await page.wait_for_timeout(1000) # 等待 1 秒（尽量避免使用） # 等待函数返回 True await page.wait_for_function("document.querySelector('.loaded') !== null") ``` ### 超时设置 ```python # 全局超时 browser = await p.chromium.launch() context = await browser.new_context() page = await context.new_page() page.set_default_timeout(30000) # 30 秒 # 单次操作超时 await page.click("button.submit", timeout=5000) await page.wait_for_selector("div.result", timeout=10000) ``` --- ## 页面内容提取 ### 获取文本 ```python # 获取元素文本 text = await page.locator("h1.title").text_content() texts = await page.locator("div.item").all_text_contents() # 获取可见文本 visible_text = await page.locator("div.content").inner_text() ``` ### 获取属性 ```python # 获取属性 href = await page.locator("a.link").get_attribute("href") src = await page.locator("img.avatar").get_attribute("src") # 获取多个元素的属性 hrefs = await page.locator("a").evaluate_all("els => els.map(el => el.href)") ``` ### 获取 HTML ```python # 获取元素 HTML html = await page.locator("div.content").inner_html() # 获取整个页面 HTML full_html = await page.content() ``` ### 执行 JavaScript ```python # 执行 JS 并获取返回值 title = await page.evaluate("document.title") # 传递参数 result = await page.evaluate("(x, y) => x + y", 1, 2) # 在元素上执行 JS text = await page.locator("div.item").evaluate("el => el.textContent") ``` --- ## 网络请求拦截 Playwright 可以拦截和修改网络请求，这对于爬虫非常有用。 ### 监听请求 ```python # 监听请求 def on_request(request): print(f"Request: {request.method} {request.url}") page.on("request", on_request) # 监听响应 def on_response(response): print(f"Response: {response.status} {response.url}") page.on("response", on_response) # 等待特定请求 async with page.expect_request("**/api/data") as request_info: await page.click("button.load") request = await request_info.value # 等待特定响应 async with page.expect_response("**/api/data") as response_info: await page.click("button.load") response = await response_info.value data = await response.json() ``` ### 拦截请求 ```python # 拦截并修改请求 async def handle_route(route): # 修改请求头 headers = {**route.request.headers, "X-Custom": "value"} await route.continue_(headers=headers) await page.route("**/*", handle_route) # 阻止某些资源加载（提高性能） await page.route("**/*.{png,jpg,jpeg,gif}", lambda route: route.abort()) await page.route("**/analytics.js", lambda route: route.abort()) # 返回模拟响应 async def mock_api(route): await route.fulfill( status=200, content_type="application/json", body='{"data": "mocked"}' ) await page.route("**/api/data", mock_api) ``` --- ## 截图与 PDF ### 截图 ```python # 页面截图 await page.screenshot(path="screenshot.png") # 全页面截图 await page.screenshot(path="full.png", full_page=True) # 元素截图 await page.locator("div.chart").screenshot(path="chart.png") # 截图选项 await page.screenshot( path="screenshot.png", type="png", # png 或 jpeg quality=80, # jpeg 质量（0-100） full_page=True, # 全页面 clip={"x": 0, "y": 0, "width": 800, "height": 600} # 裁剪区域 ) ``` ### 导出 PDF ```python # 导出 PDF（仅 Chromium 支持） await page.pdf(path="page.pdf") # PDF 选项 await page.pdf( path="page.pdf", format="A4", print_background=True, margin={"top": "1cm", "bottom": "1cm"} ) ``` --- ## 浏览器配置 ### 有头/无头模式 ```python # 有头模式（显示浏览器窗口） browser = await p.chromium.launch(headless=False) # 无头模式（不显示窗口，默认） browser = await p.chromium.launch(headless=True) ``` ### 视口设置 ```python # 设置视口大小 context = await browser.new_context( viewport={"width": 1920, "height": 1080} ) # 模拟移动设备 from playwright.async_api import async_playwright async with async_playwright() as p: iphone = p.devices["iPhone 13"] browser = await p.webkit.launch() context = await browser.new_context(**iphone) ``` ### 代理设置 ```python # 设置代理 browser = await p.chromium.launch( proxy={ "server": "http://proxy.example.com:8080", "username": "user", "password": "pass" } ) ``` ### 用户数据目录 ```python # 使用持久化的用户数据目录（保存 Cookie 等） context = await p.chromium.launch_persistent_context( user_data_dir="./user_data", headless=False ) ``` --- ## Playwright 实战演练 ### 为什么需要浏览器自动化 当网页使用 JavaScript 动态渲染内容时，传统的 HTTP 请求无法获取渲染后的数据。Playwright 可以完整模拟浏览器行为，等待页面渲染完成后再提取数据。 ```mermaid flowchart TD subgraph 适用场景 SPA["SPA应用
动态路由"] Lazy["懒加载
滚动触发"] JS["JS渲染
动态内容"] end subgraph Playwright优势 Wait["智能等待
自动判断"] Interact["模拟交互
点击/输入"] Network["网络拦截
获取API数据"] end SPA --> Wait Lazy --> Interact JS --> Network style SPA fill:#e3f2fd,stroke:#2196f3 style Lazy fill:#fff3e0,stroke:#ff9800 style JS fill:#e8f5e9,stroke:#4caf50 ``` ### 实战：爬取 JS 渲染的名言网站 quotes.toscrape.com/js 是一个专门用于学习的网站，其内容通过 JavaScript 动态渲染，非常适合作为 Playwright 入门练习。 ``` /* Detailed source-code truncated for AI context efficiency. */ ``` ### 实战：分页爬取与滚动加载 ``` /* Detailed source-code truncated for AI context efficiency. */ ``` ### 实战：拦截网络请求 Playwright 的网络拦截功能可以直接获取 API 返回的 JSON 数据，比解析 DOM 更高效： ``` /* Detailed source-code truncated for AI context efficiency. */ ``` ### Playwright 使用注意事项 ```mermaid graph LR subgraph 反检测 A1["隐藏自动化特征"] A2["模拟真实UA"] A3["设置合理视口"] end subgraph 性能优化 B1["禁用图片加载"] B2["拦截无关资源"] B3["复用浏览器实例"] end subgraph 稳定性 C1["适当等待时间"] C2["处理弹窗"] C3["异常重试机制"] end A1 --> B1 --> C1 style A1 fill:#e3f2fd,stroke:#2196f3 style B1 fill:#fff3e0,stroke:#ff9800 style C1 fill:#e8f5e9,stroke:#4caf50 ``` **关键注意点**： 1. **反自动化检测**：部分网站会检测 `navigator.webdriver`，需要使用反检测技术（下一章详解） 2. **请求频率**：避免频繁刷新页面，建议间隔 1-2 秒 3. **Cookie 管理**：登录态通过 Context 的 `storage_state` 保存和恢复 4. **资源优化**：禁用图片/字体加载可大幅提升速度 5. **网络拦截**：直接拦截 API 响应比解析 DOM 更稳定高效 --- ## 本章小结 本章我们学习了 Playwright 浏览器自动化的基础知识： 1. **核心概念**：Browser/Context/Page 三层模型，同步/异步 API 2. **页面导航**：goto、等待策略、导航控制 3. **元素定位**：语义化定位器、CSS、XPath 4. **交互操作**：点击、输入、选择、键盘、鼠标 5. **等待策略**：自动等待、显式等待、超时设置 6. **内容提取**：文本、属性、HTML、执行 JavaScript 7. **网络拦截**：监听请求响应、拦截修改请求 8. **实战演练**：爬取 JS 渲染页面、分页处理、资源拦截 --- ## 下一章预告 下一章我们将学习「Playwright 进阶：反检测与性能优化」。主要内容包括： - 浏览器指纹检测原理 - stealth.min.js 反检测注入 - CDP 模式的使用 - 性能优化技巧（禁用资源、复用上下文） - 异常处理和资源管理 这些进阶技巧将帮助你的爬虫更好地应对严格的反爬检测。 --- ### 爬虫进价 05 Playwright进阶 反检测与性能优化 (`docs/爬虫进价/05_Playwright进阶_反检测与性能优化.md`) # Playwright 进阶：反检测与性能优化 > 掌握了 Playwright 的基础操作后，你会发现在实际爬虫场景中，很多网站能够检测到自动化行为并进行封禁。本章将深入讲解浏览器指纹检测原理，以及如何使用 stealth.js 等技术绕过检测，同时介绍性能优化技巧。 ## 浏览器指纹检测原理 ### 什么是浏览器指纹 浏览器指纹是通过收集浏览器的各种特征（如 User-Agent、屏幕分辨率、插件列表、字体列表等）来唯一标识一个浏览器的技术。即使没有 Cookie，网站也可以通过指纹追踪用户。 ### 自动化浏览器的检测点 ```mermaid flowchart TD subgraph 检测层级 L1["Level 1: 基础检测"] L2["Level 2: 特征检测"] L3["Level 3: 行为检测"] end subgraph 基础检测项 D1["navigator.webdriver"] D2["window.chrome"] D3["navigator.plugins"] end subgraph 特征检测项 D4["Canvas指纹"] D5["WebGL指纹"] D6["Audio指纹"] end subgraph 行为检测项 D7["鼠标轨迹"] D8["键盘节奏"] D9["页面停留"] end L1 --> D1 & D2 & D3 L2 --> D4 & D5 & D6 L3 --> D7 & D8 & D9 style L1 fill:#e8f5e9,stroke:#4caf50 style L2 fill:#fff3e0,stroke:#ff9800 style L3 fill:#ffebee,stroke:#f44336 ``` | 检测类型 | 检测方法 | 说明 | |---------|---------|------| | WebDriver 标志 | `navigator.webdriver` | Playwright 默认为 true | | Chrome 特征 | `window.chrome` | 自动化环境可能缺失 | | 插件检测 | `navigator.plugins` | 自动化环境插件列表异常 | | 语言检测 | `navigator.languages` | 配置不当可能暴露 | | 权限 API | `navigator.permissions` | 行为与真实浏览器不同 | | Canvas 指纹 | `canvas.toDataURL()` | 渲染结果可能有差异 | | WebGL 指纹 | WebGL 参数 | 可能暴露虚拟化环境 | | 时区/语言 | 系统设置 | 与声称的地区不符 | ### 常见的检测脚本 ```javascript // 检测 WebDriver if (navigator.webdriver) { console.log("检测到自动化浏览器"); } // 检测 Chrome 特征 if (!window.chrome) { console.log("可能是自动化环境"); } // 检测 plugins if (navigator.plugins.length === 0) { console.log("插件列表为空，可能是自动化"); } // 检测 permissions navigator.permissions.query({name: "notifications"}).then(result => { if (result.state === "prompt") { // 正常行为 } }); ``` --- ## stealth.js 反检测技术 ### 什么是 stealth.js `stealth.js` 是 puppeteer-extra-plugin-stealth 项目提取出的反检测脚本，它通过修改浏览器的各种属性来伪装自动化浏览器，使其看起来像真实用户操作的浏览器。 这也是 MediaCrawler 项目使用的核心反检测技术。 ### stealth.min.js 的工作原理 stealth.js 主要做以下修改： 1. **隐藏 WebDriver 标志**：将 `navigator.webdriver` 设置为 `undefined` 2. **模拟 Chrome 特征**：添加 `window.chrome` 对象 3. **修改 plugins**：模拟正常的插件列表 4. **修改 permissions**：使权限 API 行为正常 5. **修复 iframe 检测**：处理 contentWindow 问题 6. **其他特征修复**：语言、时区等 ### 在 Playwright 中使用 stealth.js ```python import asyncio from playwright.async_api import async_playwright # stealth.min.js 脚本内容（实际使用时从文件加载） STEALTH_JS = """ // 隐藏 webdriver 标志 Object.defineProperty(navigator, 'webdriver', { get: () => undefined }); // 模拟 chrome 对象 window.chrome = { runtime: {} }; // 模拟 plugins Object.defineProperty(navigator, 'plugins', { get: () => [1, 2, 3, 4, 5] }); // 模拟 languages Object.defineProperty(navigator, 'languages', { get: () => ['zh-CN', 'zh', 'en'] }); // 修复 permissions const originalQuery = window.navigator.permissions.query; window.navigator.permissions.query = (parameters) => ( parameters.name === 'notifications' ? Promise.resolve({ state: Notification.permission }) : originalQuery(parameters) ); """ async def create_stealth_page(browser): """创建带反检测的页面""" context = await browser.new_context() # 在每个新页面创建时注入 stealth 脚本 await context.add_init_script(STEALTH_JS) page = await context.new_page() return page async def main(): async with async_playwright() as p: browser = await p.chromium.launch(headless=True) page = await create_stealth_page(browser) # 测试反检测效果 await page.goto("https://bot.sannysoft.com/") await page.screenshot(path="stealth_test.png", full_page=True) await browser.close() asyncio.run(main()) ``` ### 完整的 stealth.min.js 实际项目中，建议使用完整的 stealth.min.js 文件。你可以从以下来源获取： 1. [puppeteer-extra-plugin-stealth](https://github.com/berstend/puppeteer-extra/tree/master/packages/puppeteer-extra-plugin-stealth) 2. MediaCrawler 项目中的 stealth.min.js 加载方式： ```python async def create_stealth_context(browser, stealth_js_path: str): """创建带完整反检测的上下文""" context = await browser.new_context() # 从文件加载 stealth.js with open(stealth_js_path, "r") as f: stealth_js = f.read() await context.add_init_script(stealth_js) return context ``` --- ## CDP 模式 ### 什么是 CDP CDP（Chrome DevTools Protocol）是 Chrome 浏览器提供的调试协议，允许程序直接与浏览器通信。Playwright 内部就是使用 CDP 与 Chromium 通信的。 ### 标准模式 vs CDP 模式 | 特性 | 标准模式 | CDP 模式 | |------|---------|---------| | API | Playwright 封装的 API | 原始 CDP 命令 | | 功能 | 常用功能 | 完整的浏览器控制 | | 复杂度 | 简单 | 较复杂 | | 使用场景 | 大多数场景 | 需要底层控制时 | ### 使用 CDP 连接已有浏览器 ```python import asyncio from playwright.async_api import async_playwright async def connect_existing_browser(): """连接到已经运行的浏览器""" async with async_playwright() as p: # 首先启动一个带调试端口的浏览器 # chrome --remote-debugging-port=9222 # 通过 CDP 连接 browser = await p.chromium.connect_over_cdp( "http://localhost:9222" ) # 获取所有页面 contexts = browser.contexts if contexts: pages = contexts[0].pages if pages: page = pages[0] print(f"当前页面: {page.url}") await browser.close() ``` ### 使用 CDP 命令 ```python async def use_cdp_commands(page): """直接使用 CDP 命令""" # 获取 CDP session client = await page.context.new_cdp_session(page) # 执行 CDP 命令 # 例如：获取性能指标 metrics = await client.send("Performance.getMetrics") print(metrics) # 例如：模拟网络条件 await client.send("Network.emulateNetworkConditions", { "offline": False, "downloadThroughput": 1000000, # 1 MB/s "uploadThroughput": 500000, "latency": 100 }) # 例如：截取完整页面 result = await client.send("Page.captureScreenshot", { "format": "png", "captureBeyondViewport": True }) ``` --- ## 性能优化 ### 禁用不必要的资源加载 加载图片、字体、CSS 等资源会消耗大量时间和带宽。对于只需要获取数据的爬虫，可以禁用这些资源： ```python async def create_optimized_context(browser): """创建优化性能的上下文""" context = await browser.new_context() # 拦截并阻止不必要的资源 await context.route("**/*.{png,jpg,jpeg,gif,svg,ico}", lambda route: route.abort()) await context.route("**/*.{woff,woff2,ttf,otf}", lambda route: route.abort()) await context.route("**/*.css", lambda route: route.abort()) # 阻止分析和追踪脚本 await context.route("**/analytics.js", lambda route: route.abort()) await context.route("**/gtag/**", lambda route: route.abort()) await context.route("**/facebook.com/**", lambda route: route.abort()) return context async def selective_blocking(page): """选择性阻止资源""" async def handle_route(route): resource_type = route.request.resource_type # 只允许 document、script、xhr、fetch if resource_type in ["document", "script", "xhr", "fetch"]: await route.continue_() else: await route.abort() await page.route("**/*", handle_route) ``` ### 浏览器上下文复用 创建新的浏览器上下文比创建新页面消耗更多资源。合理复用上下文可以提高性能： ```python class BrowserPool: """浏览器上下文池""" def __init__(self, browser, pool_size: int = 5): self.browser = browser self.pool_size = pool_size self._contexts = [] self._available = asyncio.Queue() async def initialize(self): """初始化上下文池""" for _ in range(self.pool_size): context = await self.browser.new_context() self._contexts.append(context) await self._available.put(context) async def get_context(self): """获取可用上下文""" return await self._available.get() async def return_context(self, context): """归还上下文""" # 清理 cookies 和 storage await context.clear_cookies() await self._available.put(context) async def close_all(self): """关闭所有上下文""" for context in self._contexts: await context.close() ``` ### 多页面并发管理 ```python import asyncio class ConcurrentCrawler: """并发爬虫""" def __init__(self, browser, max_concurrent: int = 5): self.browser = browser self.max_concurrent = max_concurrent self.semaphore = asyncio.Semaphore(max_concurrent) async def crawl_url(self, context, url: str) -> dict: """爬取单个 URL""" async with self.semaphore: page = await context.new_page() try: await page.goto(url, wait_until="domcontentloaded") title = await page.title() return {"url": url, "title": title, "success": True} except Exception as e: return {"url": url, "error": str(e), "success": False} finally: await page.close() async def crawl_batch(self, urls: list) -> list: """批量爬取""" context = await self.browser.new_context() try: tasks = [self.crawl_url(context, url) for url in urls] return await asyncio.gather(*tasks) finally: await context.close() ``` ### 内存和资源监控 ```python async def monitor_resources(browser): """监控浏览器资源使用""" # 获取所有上下文 contexts = browser.contexts print(f"活跃上下文数: {len(contexts)}") total_pages = 0 for ctx in contexts: pages = ctx.pages total_pages += len(pages) for page in pages: print(f" 页面: {page.url[:50]}...") print(f"总页面数: {total_pages}") async def cleanup_stale_pages(context, max_age_seconds: int = 300): """清理闲置页面""" # 这里需要自己记录页面的创建时间 # 示例仅展示逻辑 for page in context.pages: # 关闭空白页或闲置页 if page.url == "about:blank": await page.close() ``` --- ## 异常处理 ### 页面崩溃检测和恢复 ```python async def safe_navigate(page, url: str, max_retries: int = 3): """安全的页面导航，带重试""" for attempt in range(max_retries): try: await page.goto(url, timeout=30000) return True except Exception as e: error_msg = str(e).lower() if "crash" in error_msg or "target closed" in error_msg: # 页面崩溃，需要重新创建 print(f"页面崩溃，尝试恢复 (尝试 {attempt + 1}/{max_retries})") # 这里应该创建新页面 continue elif "timeout" in error_msg: # 超时，可以重试 print(f"超时，重试 (尝试 {attempt + 1}/{max_retries})") continue else: # 其他错误 print(f"导航错误: {e}") raise return False ``` ### 浏览器进程管理 ```python import signal import atexit class BrowserManager: """浏览器进程管理器""" def __init__(self): self._browser = None self._playwright = None # 注册清理函数 atexit.register(self._cleanup_sync) signal.signal(signal.SIGTERM, self._signal_handler) signal.signal(signal.SIGINT, self._signal_handler) async def start(self, playwright): """启动浏览器""" self._playwright = playwright self._browser = await playwright.chromium.launch(headless=True) return self._browser async def stop(self): """停止浏览器""" if self._browser: await self._browser.close() self._browser = None def _cleanup_sync(self): """同步清理""" if self._browser: # 强制关闭 try: import asyncio loop = asyncio.get_event_loop() loop.run_until_complete(self.stop()) except Exception: pass def _signal_handler(self, signum, frame): """信号处理""" print(f"收到信号 {signum}，正在清理...") self._cleanup_sync() exit(0) ``` --- ## 反检测实战 ### 网站的反自动化检测机制 大型网站通常有较为完善的反自动化检测： ```mermaid flowchart LR subgraph 检测机制 Check1["WebDriver检测"] Check2["请求头验证"] Check3["行为分析"] Check4["频率限制"] end subgraph 检测结果 Pass["正常访问"] Block["触发风控"] Captcha["触发验证码"] end Check1 -->|通过| Check2 Check1 -->|失败| Block Check2 -->|通过| Check3 Check2 -->|异常| Block Check3 -->|正常| Pass Check3 -->|异常| Captcha Check4 -->|超限| Block style Pass fill:#c8e6c9,stroke:#4caf50 style Block fill:#ffcdd2,stroke:#f44336 style Captcha fill:#fff3e0,stroke:#ff9800 ``` ### 完整反检测配置 ``` /* Detailed source-code truncated for AI context efficiency. */ ``` ### 性能优化爬虫配置 ``` /* Detailed source-code truncated for AI context efficiency. */ ``` ### 反检测效果验证流程 ```mermaid flowchart LR Start["启动浏览器"] --> Inject["注入stealth.js"] Inject --> Visit["访问测试网站"] Visit --> Check{"检测验证"} Check -->|webdriver=undefined| Pass1["✓ 通过"] Check -->|chrome存在| Pass2["✓ 通过"] Check -->|plugins正常| Pass3["✓ 通过"] Pass1 & Pass2 & Pass3 --> Result["反检测成功"] Result --> Crawl["正常爬取"] style Result fill:#c8e6c9,stroke:#4caf50 style Crawl fill:#e3f2fd,stroke:#2196f3 ``` --- ## 本章小结 本章我们学习了 Playwright 的进阶技术： 1. **浏览器指纹检测**：了解网站如何检测自动化浏览器 2. **stealth.js 反检测**：使用脚本注入绕过检测 3. **CDP 模式**：直接使用 Chrome DevTools Protocol 4. **性能优化**：禁用资源加载、上下文复用、并发管理 5. **异常处理**：页面崩溃恢复、资源清理 6. **实战演练**：反检测配置、性能优化爬虫 --- ## 下一章预告 下一章我们将学习「登录认证：Cookie 与 Session 管理」。主要内容包括： - Cookie 和 Session 的深入理解 - Cookie 的提取、存储和注入 - 登录状态检测和自动刷新 - 多账号 Cookie 轮换 这些技术是爬取需要登录的网站的基础。 --- ### 爬虫进价 06 登录认证 Cookie与Session管理 (`docs/爬虫进价/06_登录认证_Cookie与Session管理.md`) # 06_登录认证：Cookie 与 Session 管理 在爬虫开发中，很多数据需要登录后才能获取。本章将深入讲解 Cookie 和 Session 的工作机制，以及如何在爬虫中高效管理登录状态。 ## 一、认证机制深入理解 ### 1.1 Cookie 的工作原理 Cookie 是服务器发送给浏览器的小型数据片段，浏览器会保存并在后续请求中自动携带。 ```mermaid sequenceDiagram participant Client as 客户端 participant Server as 服务器 Client->>Server: HTTP 请求 (首次访问) Server-->>Client: Set-Cookie: session=xxx Note over Client: 保存 Cookie 到本地 Client->>Server: HTTP 请求 + Cookie: session=xxx Server-->>Client: 返回用户数据 Note over Server: 识别用户身份 ``` ### 1.2 Cookie 属性详解 每个 Cookie 都有多个属性，理解这些属性对于正确使用 Cookie 至关重要： | 属性 | 说明 | 爬虫影响 | |-----|------|---------| | **Name/Value** | Cookie 的名称和值 | 核心数据，必须正确提取 | | **Domain** | Cookie 生效的域名 | 决定请求哪些域名时携带 | | **Path** | Cookie 生效的路径 | 决定请求哪些路径时携带 | | **Expires/Max-Age** | 过期时间 | 决定 Cookie 何时失效 | | **HttpOnly** | 禁止 JavaScript 访问 | 只能通过 HTTP 响应获取 | | **Secure** | 仅 HTTPS 传输 | 注意协议匹配 | | **SameSite** | 跨站请求限制 | 影响第三方请求 | ```python # Cookie 属性示例 cookie_example = { "name": "session_id", "value": "abc123xyz", "domain": ".example.com", # 包含子域名 "path": "/", # 所有路径 "expires": 1735689600, # Unix 时间戳 "httpOnly": True, # 不能通过 JS 访问 "secure": True, # 仅 HTTPS "sameSite": "Lax" # 跨站限制 } ``` ### 1.3 Session 机制 Session 是服务端的会话状态存储机制，通常通过 Cookie 中的 Session ID 关联： ```python # 服务端 Session 工作流程（伪代码） class SessionFlow: """ 1. 用户登录 -> 服务器创建 Session 2. 服务器返回 Session ID (通过 Set-Cookie) 3. 浏览器携带 Session ID 访问 4. 服务器根据 Session ID 查找用户状态 """ def login(self, username, password): if self.verify(username, password): session_id = generate_session_id() self.sessions[session_id] = { "user_id": user.id, "login_time": time.time(), "expires": time.time() + 3600 * 24 } return session_id return None ``` ### 1.4 Token 认证机制 现代应用越来越多使用 Token 认证（如 JWT），与 Cookie/Session 有所不同： | 特性 | Cookie/Session | Token (JWT) | |-----|----------------|-------------| | 状态存储 | 服务端 | 客户端（Token 自包含） | | 跨域支持 | 受限（SameSite） | 天然支持 | | 传输方式 | Cookie 头 | Authorization 头 | | 服务端开销 | 需要存储 Session | 无状态 | ```python # Token 认证示例 import httpx headers = { "Authorization": "Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9..." } async with httpx.AsyncClient() as client: resp = await client.get("https://api.example.com/data", headers=headers) ``` ## 二、Cookie 的提取与存储 ### 2.1 从浏览器手动提取 最简单的方式是从浏览器开发者工具中手动复制 Cookie： 1. 打开浏览器，登录目标网站 2. 按 F12 打开开发者工具 3. 切换到 Application/Storage -> Cookies 4. 复制需要的 Cookie 值 ```python # 手动提取的 Cookie 使用示例 import httpx cookies = { "session_id": "abc123", "user_token": "xyz789", "preferences": "theme=dark" } async with httpx.AsyncClient(cookies=cookies) as client: resp = await client.get("https://example.com/user/profile") ``` ### 2.2 使用 Playwright 自动提取 Playwright 可以自动化登录流程并提取 Cookie： ```python from playwright.async_api import async_playwright import json async def extract_cookies_after_login(): """登录后自动提取 Cookie""" async with async_playwright() as p: browser = await p.chromium.launch(headless=False) # 有头模式便于手动登录 context = await browser.new_context() page = await context.new_page() # 导航到登录页 await page.goto("https://example.com/login") # 等待用户手动登录（或自动填写表单） # await page.fill("#username", "your_username") # await page.fill("#password", "your_password") # await page.click("#login-button") # 等待登录成功（检测特定元素或 URL 变化） await page.wait_for_url("**/dashboard**", timeout=120000) # 提取所有 Cookie cookies = await context.cookies() # 保存到文件 with open("cookies.json", "w") as f: json.dump(cookies, f, indent=2) print(f"提取了 {len(cookies)} 个 Cookie") await browser.close() return cookies ``` ### 2.3 Cookie 序列化格式 Cookie 可以用多种格式存储，常见的有 JSON 和 Netscape 格式： ```python import json from http.cookiejar import MozillaCookieJar from datetime import datetime class CookieSerializer: """Cookie 序列化工具""" @staticmethod def to_json(cookies: list, filepath: str): """保存为 JSON 格式""" with open(filepath, "w", encoding="utf-8") as f: json.dump(cookies, f, indent=2, ensure_ascii=False) @staticmethod def from_json(filepath: str) -> list: """从 JSON 加载""" with open(filepath, "r", encoding="utf-8") as f: return json.load(f) @staticmethod def to_netscape(cookies: list, filepath: str): """保存为 Netscape 格式（兼容 curl/wget）""" lines = ["# Netscape HTTP Cookie File"] for c in cookies: # 格式: domain, flag, path, secure, expiry, name, value domain = c.get("domain", "") flag = "TRUE" if domain.startswith(".") else "FALSE" path = c.get("path", "/") secure = "TRUE" if c.get("secure", False) else "FALSE" expiry = str(int(c.get("expires", 0))) name = c.get("name", "") value = c.get("value", "") lines.append(f"{domain}\t{flag}\t{path}\t{secure}\t{expiry}\t{name}\t{value}") with open(filepath, "w") as f: f.write("\n".join(lines)) @staticmethod def to_dict(cookies: list) -> dict: """转换为简单字典格式（name: value）""" return {c["name"]: c["value"] for c in cookies} ``` ### 2.4 Cookie 加密存储 对于敏感的 Cookie，建议加密存储： ```python import json import base64 from cryptography.fernet import Fernet class SecureCookieStorage: """加密 Cookie 存储""" def __init__(self, key: bytes = None): # 如果没有提供密钥，生成新密钥 self.key = key or Fernet.generate_key() self.cipher = Fernet(self.key) def save_key(self, filepath: str): """保存密钥（请妥善保管）""" with open(filepath, "wb") as f: f.write(self.key) @classmethod def load_key(cls, filepath: str) -> "SecureCookieStorage": """从文件加载密钥""" with open(filepath, "rb") as f: return cls(f.read()) def encrypt_cookies(self, cookies: list, filepath: str): """加密并保存 Cookie""" data = json.dumps(cookies).encode("utf-8") encrypted = self.cipher.encrypt(data) with open(filepath, "wb") as f: f.write(encrypted) def decrypt_cookies(self, filepath: str) -> list: """解密并加载 Cookie""" with open(filepath, "rb") as f: encrypted = f.read() decrypted = self.cipher.decrypt(encrypted) return json.loads(decrypted.decode("utf-8")) ``` ## 三、Cookie 注入与使用 ### 3.1 httpx 中使用 Cookie httpx 提供了灵活的 Cookie 管理方式： ```python import httpx # 方式1: 直接传入字典 cookies_dict = {"session": "abc123", "token": "xyz"} async with httpx.AsyncClient(cookies=cookies_dict) as client: resp = await client.get("https://example.com/api") # 方式2: 使用 Cookies 对象 from httpx import Cookies cookies = Cookies() cookies.set("session", "abc123", domain="example.com") cookies.set("token", "xyz", domain="example.com") async with httpx.AsyncClient(cookies=cookies) as client: resp = await client.get("https://example.com/api") # 方式3: 从响应中自动获取 async with httpx.AsyncClient() as client: # 登录请求会设置 Cookie login_resp = await client.post( "https://example.com/login", data={"username": "user", "password": "pass"} ) # 后续请求自动携带 Cookie profile_resp = await client.get("https://example.com/profile") ``` ### 3.2 Playwright 中注入 Cookie Playwright 可以在访问页面前注入 Cookie： ```python from playwright.async_api import async_playwright import json async def inject_cookies_and_browse(): """注入 Cookie 并访问页面""" async with async_playwright() as p: browser = await p.chromium.launch(headless=True) context = await browser.new_context() # 从文件加载 Cookie with open("cookies.json", "r") as f: cookies = json.load(f) # 注入 Cookie await context.add_cookies(cookies) # 现在可以直接访问需要登录的页面 page = await context.new_page() await page.goto("https://example.com/dashboard") # 检查是否登录成功 if await page.locator(".user-avatar").is_visible(): print("Cookie 注入成功，已登录") else: print("Cookie 可能已过期") await browser.close() ``` ### 3.3 跨工具共享 Cookie 将 Playwright 提取的 Cookie 转换为 httpx 可用的格式： ```python def playwright_cookies_to_httpx(playwright_cookies: list) -> dict: """ 将 Playwright 格式的 Cookie 转换为 httpx 格式 Playwright 格式: [{"name": "x", "value": "y", "domain": "...", ...}, ...] httpx 格式: {"name": "value", ...} """ return {c["name"]: c["value"] for c in playwright_cookies} def httpx_cookies_to_playwright(cookies_dict: dict, domain: str) -> list: """ 将简单字典格式转换为 Playwright 格式 """ return [ { "name": name, "value": value, "domain": domain, "path": "/" } for name, value in cookies_dict.items() ] ``` ### 3.4 多账号 Cookie 轮换 在需要多账号爬取的场景，可以实现 Cookie 轮换： ``` /* Detailed source-code truncated for AI context efficiency. */ ``` ## 四、登录状态管理 ### 4.1 登录状态检测 检测 Cookie 是否仍然有效是关键步骤： ``` /* Detailed source-code truncated for AI context efficiency. */ ``` ### 4.2 Cookie 过期检测 监控 Cookie 的过期时间： ```python import time from datetime import datetime, timedelta from typing import List, Tuple class CookieExpiryMonitor: """Cookie 过期监控""" @staticmethod def check_expiry(cookies: list) -> Tuple[bool, List[dict]]: """ 检查 Cookie 是否过期 Returns: (是否全部有效, 已过期的 Cookie 列表) """ now = time.time() expired = [] for cookie in cookies: expires = cookie.get("expires", 0) if expires > 0 and expires < now: expired.append(cookie) return len(expired) == 0, expired @staticmethod def get_earliest_expiry(cookies: list) -> datetime: """获取最早过期的时间""" expiry_times = [ c.get("expires", float("inf")) for c in cookies if c.get("expires", 0) > 0 ] if not expiry_times: return datetime.max return datetime.fromtimestamp(min(expiry_times)) @staticmethod def will_expire_soon(cookies: list, threshold_hours: int = 1) -> bool: """检查 Cookie 是否即将过期""" earliest = CookieExpiryMonitor.get_earliest_expiry(cookies) return earliest < datetime.now() + timedelta(hours=threshold_hours) ``` ### 4.3 完整的 Cookie 管理器 整合以上功能的完整管理器： ``` /* Detailed source-code truncated for AI context efficiency. */ ``` ## 五、B站 Cookie 管理实战 ### 5.1 B站 Cookie 结构分析 B站使用多个关键 Cookie 来管理用户登录状态： ```mermaid graph TB subgraph B站核心Cookie SESSDATA["SESSDATA
会话凭证"] DedeUserID["DedeUserID
用户ID"] bili_jct["bili_jct
CSRF Token"] end subgraph 辅助Cookie buvid3["buvid3
设备标识"] buvid4["buvid4
设备标识"] sid["sid
短会话ID"] end SESSDATA --> API["API请求认证"] DedeUserID --> API bili_jct --> POST["POST请求CSRF"] style SESSDATA fill:#e8f5e9,stroke:#4caf50 style DedeUserID fill:#e8f5e9,stroke:#4caf50 style bili_jct fill:#fff3e0,stroke:#ff9800 ``` | Cookie 名称 | 作用 | 有效期 | 说明 | |------------|------|-------|------| | **SESSDATA** | 会话凭证 | ~1个月 | 最重要的登录凭证 | | **DedeUserID** | 用户ID | ~1个月 | 用于部分API请求 | | **bili_jct** | CSRF Token | ~1个月 | POST请求必需 | | buvid3 | 设备标识 | ~1年 | 追踪设备 | | buvid4 | 设备标识 | ~1年 | 追踪设备 | | sid | 短会话 | 会话 | 短期会话标识 | ### 5.2 B站 Cookie 提取器 ``` /* Detailed source-code truncated for AI context efficiency. */ ``` ### 5.3 B站 Cookie 使用示例 ``` /* Detailed source-code truncated for AI context efficiency. */ ``` ### 5.4 B站 Cookie 与 Playwright 集成 ```python # -*- coding: utf-8 -*- """ B站 Cookie 与 Playwright 集成 """ import asyncio from playwright.async_api import async_playwright from loguru import logger async def bilibili_with_playwright(): """使用 Playwright 注入 B站 Cookie""" # 加载 Cookie manager = BilibiliCookieManager("data/bilibili_cookies.json") await manager.load() if not manager.cookies or not manager.cookies.is_valid(): logger.error("Cookie 无效") return async with async_playwright() as p: browser = await p.chromium.launch(headless=True) context = await browser.new_context( viewport={"width": 1920, "height": 1080}, locale="zh-CN" ) # 注入 Cookie await context.add_cookies( manager.cookies.to_playwright_cookies() ) page = await context.new_page() # 访问个人主页 await page.goto("https://space.bilibili.com/", wait_until="networkidle") # 检查是否登录成功 avatar = page.locator(".bili-avatar") if await avatar.count() > 0: logger.info("Cookie 注入成功，已登录") # 获取用户名 username = await page.locator(".h-name").text_content() print(f"当前用户: {username}") else: logger.warning("Cookie 可能已过期") await browser.close() if __name__ == "__main__": asyncio.run(bilibili_with_playwright()) ``` --- ## 六、完整的登录会话管理 下面是一个完整的实战示例，展示如何在爬虫中管理登录状态： ``` /* Detailed source-code truncated for AI context efficiency. */ ``` ## 六、最佳实践 ### 6.1 Cookie 安全存储建议 1. **不要硬编码 Cookie**：将 Cookie 存储在配置文件或环境变量中 2. **敏感 Cookie 加密**：使用加密存储敏感的登录凭证 3. **设置合适的文件权限**：Cookie 文件应限制访问权限 4. **定期轮换**：定期重新登录获取新 Cookie ### 6.2 登录状态管理建议 1. **检测频率控制**：不要频繁检测，设置合理的检测间隔 2. **优雅降级**：Cookie 失效时应有备用方案 3. **多账号轮换**：大规模爬取时使用多账号 4. **遵守频率限制**：避免触发反爬机制 ### 6.3 常见问题排查 | 问题 | 可能原因 | 解决方案 | |-----|---------|---------| | Cookie 无效 | 已过期/被踢出 | 检查过期时间，重新登录 | | 请求被拒绝 | 缺少必要 Cookie | 检查 Cookie 完整性 | | 跨域问题 | Domain 不匹配 | 检查 Cookie 的 Domain 属性 | | HTTPS 问题 | Secure 属性 | 确保使用 HTTPS 协议 | ## 本章小结 本章深入讲解了 Cookie 和 Session 的工作机制，以及在爬虫中如何有效管理登录状态： 1. **理解认证机制**：Cookie 属性、Session 原理、Token 认证 2. **Cookie 提取存储**：手动提取、自动化提取、序列化格式、加密存储 3. **Cookie 注入使用**：httpx 和 Playwright 中的使用方式、跨工具共享 4. **登录状态管理**：状态检测、过期监控、自动刷新机制 5. **B站实战**：Cookie 结构分析、专用管理器、验证与使用 掌握这些技术后，你就能够处理大多数需要登录的爬虫场景。 --- ## 与第11章实战项目的关联 本章 Cookie 管理技术在第11章 B站综合实战项目中有核心应用： | 本章内容 | 第11章对应实现 | 文件位置 | |---------|--------------|---------| | BilibiliCookies 数据类 | Cookie 模型定义 | `models/cookies.py` | | BilibiliCookieManager | 登录管理器 | `login/auth.py` | | Cookie 验证逻辑 | 登录状态检测 | `login/auth.py` | | Playwright 集成 | 扫码登录实现 | `login/auth.py` | ```mermaid graph LR subgraph 本章知识点 A1["Cookie结构"] A2["Cookie管理器"] A3["登录验证"] end subgraph 第11章实战应用 B1["扫码登录"] B2["登录态保持"] B3["API请求认证"] end A1 --> B1 A2 --> B2 A3 --> B3 style A1 fill:#e3f2fd,stroke:#2196f3 style A2 fill:#e3f2fd,stroke:#2196f3 style A3 fill:#e3f2fd,stroke:#2196f3 style B1 fill:#c8e6c9,stroke:#4caf50 style B2 fill:#c8e6c9,stroke:#4caf50 style B3 fill:#c8e6c9,stroke:#4caf50 ``` **学习建议**： 1. 理解 B站三大核心 Cookie（SESSDATA、DedeUserID、bili_jct）的作用 2. 掌握 Cookie 的提取、存储、验证完整流程 3. 在第11章中，`login/auth.py` 是 Cookie 管理的核心实现 ## 下一章预告 下一章我们将学习更复杂的登录方式——**扫码登录和短信验证码登录**的实现。我们会参考 MediaCrawler 的登录模块设计，学习如何处理二维码扫码流程、短信验证码接收等高级场景。 --- ### 爬虫进价 07 登录认证 扫码与短信登录实现 (`docs/爬虫进价/07_登录认证_扫码与短信登录实现.md`) # 07_登录认证：扫码与短信登录实现 上一章我们学习了 Cookie 和 Session 的基础管理，本章将深入探讨更复杂的登录场景——扫码登录和短信验证码登录。这些登录方式在社交媒体、电商等平台中广泛使用，也是 MediaCrawler 等项目的核心技术之一。 ## 一、扫码登录原理 ### 1.1 扫码登录流程 扫码登录是一种安全便捷的认证方式，其工作流程如下： ```mermaid sequenceDiagram participant PC as PC浏览器 participant Server as 服务器 participant APP as 手机APP PC->>Server: 1. 请求二维码 Server-->>PC: 2. 返回二维码+UUID loop 轮询登录状态 PC->>Server: 3. 查询状态 Server-->>PC: 状态: 待扫描 end APP->>Server: 4. 用户扫码 Server-->>PC: 5. 状态: 已扫描 APP->>Server: 6. 用户确认 Server-->>PC: 7. 返回登录凭证(Cookie) Note over PC: 8. 登录成功! ``` ### 1.2 关键技术点 1. **二维码生成**：服务器生成唯一 UUID，编码到二维码中 2. **状态轮询**：PC 端定时查询登录状态（长轮询或 WebSocket） 3. **状态流转**：待扫描 → 已扫描 → 已确认 → 已过期 4. **凭证下发**：登录成功后通过 Cookie 或 Token 返回凭证 ### 1.3 常见扫码登录状态 ```python from enum import Enum class QRCodeStatus(Enum): """二维码状态枚举""" WAITING = "waiting" # 等待扫描 SCANNED = "scanned" # 已扫描，等待确认 CONFIRMED = "confirmed" # 已确认登录 EXPIRED = "expired" # 二维码已过期 CANCELED = "canceled" # 用户取消 ``` ## 二、Playwright 实现扫码登录 ### 2.1 基础框架 ``` /* Detailed source-code truncated for AI context efficiency. */ ``` ### 2.2 完整扫码登录实现 以下是一个通用的扫码登录实现示例： ``` /* Detailed source-code truncated for AI context efficiency. */ ``` ### 2.3 终端显示二维码 为了在无头模式下也能扫码，可以在终端显示二维码： ```python try: import qrcode HAS_QRCODE = True except ImportError: HAS_QRCODE = False def display_qrcode_in_terminal(data: str): """在终端显示二维码""" if not HAS_QRCODE: print("提示: 安装 qrcode 库可在终端显示二维码: pip install qrcode") return qr = qrcode.QRCode( version=1, error_correction=qrcode.constants.ERROR_CORRECT_L, box_size=1, border=1 ) qr.add_data(data) qr.make(fit=True) # 使用字符绘制 qr.print_ascii(invert=True) def display_qrcode_image_in_terminal(image_path: str): """ 将图片二维码转换为终端可显示的文本 需要安装: pip install pillow """ try: from PIL import Image img = Image.open(image_path) img = img.convert('L') # 转为灰度 # 缩小图片 width = 60 ratio = width / img.width height = int(img.height * ratio * 0.5) # 0.5 补偿终端字符高宽比 img = img.resize((width, height)) # 转换为 ASCII chars = " .:-=+*#%@" pixels = img.getdata() ascii_img = "" for i, pixel in enumerate(pixels): if i > 0 and i % width == 0: ascii_img += "\n" ascii_img += chars[pixel * len(chars) // 256] print(ascii_img) except ImportError: print(f"二维码已保存到: {image_path}") ``` ## 三、短信验证码登录 ### 3.1 短信登录流程 ```mermaid sequenceDiagram participant User as 用户 participant Frontend as 前端 participant Server as 服务器 User->>Frontend: 1. 输入手机号 Frontend->>Server: 2. 请求发送验证码 Server-->>Frontend: 3. 返回发送结果 User->>Frontend: 4. 输入验证码 Frontend->>Server: 5. 提交登录请求 Server-->>Frontend: 6. 返回登录结果+凭证 Note over User,Server: 登录成功，获取Cookie ``` ### 3.2 Playwright 实现短信登录 ``` /* Detailed source-code truncated for AI context efficiency. */ ``` ### 3.3 验证码获取方式 ```python import asyncio async def get_code_from_user() -> str: """从控制台获取用户输入的验证码""" print("\n请输入收到的验证码: ", end="", flush=True) # 在异步环境中等待用户输入 loop = asyncio.get_event_loop() code = await loop.run_in_executor(None, input) return code.strip() class SMSCodeReceiver: """ 短信接码平台接口（示意） 注意：实际使用需要接入具体的接码平台 API """ def __init__(self, api_key: str, api_url: str): self.api_key = api_key self.api_url = api_url async def get_phone_number(self) -> str: """获取手机号""" # 调用接码平台 API 获取手机号 raise NotImplementedError("需要实现具体的接码平台接口") async def wait_for_code(self, phone: str, timeout: int = 60) -> Optional[str]: """等待接收验证码""" # 轮询接码平台获取验证码 raise NotImplementedError("需要实现具体的接码平台接口") async def release_phone(self, phone: str): """释放手机号""" raise NotImplementedError("需要实现具体的接码平台接口") ``` ## 四、登录模块统一封装 ### 4.1 使用工厂模式 ```python from abc import ABC, abstractmethod from typing import Optional, Callable, Awaitable from enum import Enum class LoginMethod(Enum): """登录方式枚举""" QRCODE = "qrcode" # 扫码登录 SMS = "sms" # 短信验证码 PASSWORD = "password" # 账号密码 COOKIE = "cookie" # Cookie 注入 class LoginResult: """登录结果""" def __init__( self, success: bool, cookies: list = None, error: str = None ): self.success = success self.cookies = cookies or [] self.error = error class BaseLogin(ABC): """登录基类""" @abstractmethod async def login(self) -> LoginResult: """执行登录""" pass @abstractmethod async def close(self): """清理资源""" pass class LoginFactory: """登录工厂""" _registry = {} @classmethod def register(cls, method: LoginMethod): """注册登录实现""" def decorator(login_class): cls._registry[method] = login_class return login_class return decorator @classmethod def create( cls, method: LoginMethod, **kwargs ) -> BaseLogin: """创建登录实例""" if method not in cls._registry: raise ValueError(f"不支持的登录方式: {method}") return cls._registry[method](**kwargs) ``` ### 4.2 注册具体实现 ``` /* Detailed source-code truncated for AI context efficiency. */ ``` ### 4.3 统一登录管理器 ``` /* Detailed source-code truncated for AI context efficiency. */ ``` ## 五、实战示例 ### 5.1 完整的扫码登录示例 ```python import asyncio from playwright.async_api import async_playwright import json async def qrcode_login_demo(): """扫码登录完整示例""" # 二维码准备好后的回调 async def on_qrcode_ready(path: str): print(f"\n{'='*40}") print(f"请使用手机扫描二维码: {path}") print(f"{'='*40}\n") # 可以在这里显示二维码到终端 display_qrcode_image_in_terminal(path) async with async_playwright() as p: # 创建扫码登录实例（以示例网站为例） qr_login = QRCodeLogin( login_url="https://example.com/login", qrcode_selector="img.qrcode", # 二维码选择器 success_url_pattern="/dashboard", timeout=120 ) await qr_login.start(p, headless=False) try: cookies = await qr_login.login(on_qrcode_ready) if cookies: # 保存 Cookie with open("login_cookies.json", "w") as f: json.dump(cookies, f, indent=2) print(f"登录成功！获取到 {len(cookies)} 个 Cookie") else: print("登录失败或超时") finally: await qr_login.close() if __name__ == "__main__": asyncio.run(qrcode_login_demo()) ``` ### 5.2 多平台登录管理 ```python async def multi_platform_demo(): """多平台登录管理示例""" # 配置多个平台 platforms = { "platform_a": { "cookie_path": "data/platform_a_cookies.json", "login_url": "https://a.example.com/login", "qrcode_selector": "#qrcode-img", "success_url": "/home" }, "platform_b": { "cookie_path": "data/platform_b_cookies.json", "login_url": "https://b.example.com/login", "qrcode_selector": ".login-qrcode", "success_url": "/dashboard" } } for name, config in platforms.items(): print(f"\n处理平台: {name}") manager = LoginManager( platform=name, cookie_path=config["cookie_path"], preferred_method=LoginMethod.QRCODE ) # 定义二维码回调 async def on_qrcode(path: str): print(f"[{name}] 请扫描二维码: {path}") success = await manager.ensure_login( login_url=config["login_url"], qrcode_selector=config["qrcode_selector"], success_url_pattern=config["success_url"], on_qrcode_ready=on_qrcode ) if success: cookies = manager.get_cookies_dict() print(f"[{name}] 登录成功，Cookie 数量: {len(cookies)}") ``` ## 六、最佳实践 ### 6.1 安全建议 1. **不要频繁登录**：频繁登录可能触发风控 2. **保护登录凭证**：Cookie 应加密存储 3. **遵守服务条款**：了解平台的使用限制 4. **处理敏感信息**：手机号等信息不要硬编码 ### 6.2 稳定性建议 1. **超时处理**：设置合理的超时时间 2. **重试机制**：登录失败时适当重试 3. **状态监控**：监控登录状态，及时发现问题 4. **优雅降级**：提供多种登录方式作为备选 ### 6.3 代码组织建议 ``` login/ ├── __init__.py ├── base.py # 基类定义 ├── factory.py # 登录工厂 ├── qrcode.py # 扫码登录实现 ├── sms.py # 短信登录实现 ├── cookie.py # Cookie 登录实现 └── manager.py # 统一管理器 ``` ## 七、B站扫码登录实战 本节以 B站 为实战平台，演示完整的扫码登录实现。B站 是国内最大的二次元视频平台，其登录系统具有代表性。 ### 7.1 B站扫码登录流程分析 B站扫码登录涉及以下API： ```mermaid sequenceDiagram participant PC as PC浏览器 participant Server as B站服务器 participant APP as B站APP PC->>Server: 1. GET /qrcode/generate Server-->>PC: 2. 返回 qrcode_key + url Note over PC: 生成二维码图片 loop 轮询登录状态 (间隔2秒) PC->>Server: 3. GET /qrcode/poll?qrcode_key=xxx Server-->>PC: 4. 返回状态码 Note right of Server: 86101=未扫描
86090=已扫描
86038=已过期
0=已确认 end APP->>Server: 5. 用户扫码 Server-->>PC: 6. 状态: 86090 已扫描 APP->>Server: 7. 用户确认登录 Server-->>PC: 8. 状态: 0 + Set-Cookie Note over PC: 9. 登录成功！保存Cookie ``` ### 7.2 B站登录状态码 | 状态码 | 含义 | 说明 | |-------|------|------| | 0 | 成功 | 登录成功，响应中包含Cookie | | 86101 | 未扫描 | 等待用户扫描二维码 | | 86090 | 已扫描 | 用户已扫描，等待确认 | | 86038 | 已过期 | 二维码已过期，需重新获取 | ### 7.3 B站扫码登录完整实现 ``` /* Detailed source-code truncated for AI context efficiency. */ ``` ### 7.4 使用已登录Cookie访问B站API ```python import httpx import json from pathlib import Path async def use_bilibili_cookies(): """使用已保存的Cookie访问B站API""" # 1. 加载Cookie cookie_path = Path("data/bilibili_cookies.json") if not cookie_path.exists(): print("请先执行扫码登录获取Cookie") return with open(cookie_path) as f: cookies = json.load(f) # 2. 创建带Cookie的客户端 async with httpx.AsyncClient( cookies=cookies, headers={ "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/120.0.0.0 Safari/537.36", "Referer": "https://www.bilibili.com/", } ) as client: # 3. 获取用户信息 resp = await client.get( "https://api.bilibili.com/x/web-interface/nav" ) data = resp.json() if data["code"] == 0: user_info = data["data"] print(f"登录用户: {user_info['uname']}") print(f"用户ID: {user_info['mid']}") print(f"等级: LV{user_info['level_info']['current_level']}") print(f"硬币: {user_info['money']}") else: print(f"获取用户信息失败: {data['message']}") if __name__ == "__main__": import asyncio asyncio.run(use_bilibili_cookies()) ``` ### 7.5 Cookie有效性验证 ``` /* Detailed source-code truncated for AI context efficiency. */ ``` --- ## 八、与第11章的关联 本章介绍的B站扫码登录技术在第11章综合实战项目中有完整应用： ### 代码位置 - **登录模块**：`源代码/爬虫进阶/11_进阶综合实战项目/login/auth.py` - **Cookie管理**：`源代码/爬虫进阶/11_进阶综合实战项目/client/bilibili_client.py` ### 技术要点对应 | 本章内容 | 第11章实现 | |---------|-----------| | `BilibiliQRCodeLogin` 类 | `login/auth.py` 中的登录逻辑 | | `BilibiliCookies` 数据类 | `models/` 中的数据模型 | | Cookie有效性验证 | `client/` 中的请求拦截器 | | 状态码枚举 | `config/bilibili_config.py` 中的常量定义 | ### 学习路径 ```mermaid graph LR A[第06章
Cookie管理] --> B[第07章
扫码登录] B --> C[第08章
验证码处理] C --> D[第11章
综合实战] style B fill:#e1f5fe style D fill:#fff9c4 ``` 掌握本章的扫码登录技术后，你已经具备了B站爬虫的核心认证能力，可以直接应用到第11章的综合项目中。 --- ## 本章小结 本章深入讲解了扫码登录和短信验证码登录的实现： 1. **扫码登录原理**：二维码生成、状态轮询、凭证下发的完整流程 2. **Playwright 实现**：使用浏览器自动化实现扫码登录 3. **短信登录实现**：手机号输入、验证码获取、登录提交 4. **统一封装**：使用工厂模式封装多种登录方式 5. **B站实战**：完整的B站扫码登录实现，包括二维码生成、状态轮询、Cookie提取 掌握这些技术后，你可以应对大多数需要登录的爬虫场景。 ## 下一章预告 下一章我们将学习**验证码识别与处理**，包括图片验证码 OCR 识别、滑块验证码轨迹模拟等技术。这些技术在登录和爬取过程中经常遇到，是爬虫进阶的重要内容。 --- ### 爬虫进价 08 验证码识别与处理 (`docs/爬虫进价/08_验证码识别与处理.md`) # 08_验证码识别与处理 验证码是网站用于区分人类和机器的重要手段。本章将介绍常见的验证码类型，以及如何在爬虫中处理这些验证码。 ## 一、验证码类型概览 ### 1.1 常见验证码类型 | 类型 | 特点 | 难度 | 常见场景 | |-----|------|------|---------| | 图片字符验证码 | 数字/字母/汉字 | ⭐⭐ | 登录、注册 | | 数学运算验证码 | 简单算术题 | ⭐ | 评论、提交 | | 滑块拼图验证码 | 拖动滑块到缺口 | ⭐⭐⭐ | 登录、支付 | | 点选验证码 | 点击指定图片/文字 | ⭐⭐⭐⭐ | 登录、敏感操作 | | 行为验证码 | 分析用户行为 | ⭐⭐⭐⭐⭐ | 全站防护 | ### 1.2 验证码的工作原理 ```mermaid sequenceDiagram participant User as 用户 participant Frontend as 前端 participant Server as 服务器 User->>Frontend: 1. 请求验证码 Frontend->>Server: 2. 请求验证码 Server-->>Frontend: 3. 返回验证码图片+ID Frontend-->>User: 4. 显示验证码 User->>Frontend: 5. 输入验证码 Frontend->>Server: 6. 提交验证 Server-->>Frontend: 7. 返回验证结果 Note over User,Server: 验证通过后继续操作 ``` ## 二、图片验证码识别 ### 2.1 ddddocr 库介绍 ddddocr 是一个开源的 OCR 库，专门用于识别验证码： ```bash pip install ddddocr ``` ### 2.2 基本使用 ```python import ddddocr # 创建 OCR 实例 ocr = ddddocr.DdddOcr() # 读取验证码图片 with open("captcha.png", "rb") as f: image_bytes = f.read() # 识别 result = ocr.classification(image_bytes) print(f"识别结果: {result}") ``` ### 2.3 图片预处理 对于复杂验证码，预处理可以提高识别率： ``` /* Detailed source-code truncated for AI context efficiency. */ ``` ### 2.4 封装的 OCR 识别器 ``` /* Detailed source-code truncated for AI context efficiency. */ ``` ## 三、滑块验证码处理 滑块验证码需要识别缺口位置并模拟人类拖拽行为。 ### 3.1 缺口位置识别 ``` /* Detailed source-code truncated for AI context efficiency. */ ``` ### 3.2 人类轨迹模拟 模拟人类拖拽行为是绕过行为检测的关键： ``` /* Detailed source-code truncated for AI context efficiency. */ ``` ### 3.3 Playwright 实现滑块拖拽 ``` /* Detailed source-code truncated for AI context efficiency. */ ``` ## 四、第三方打码平台 对于复杂验证码，可以使用第三方打码平台。 ### 4.1 打码平台接口封装 ``` /* Detailed source-code truncated for AI context efficiency. */ ``` ### 4.2 带重试的验证码处理 ``` /* Detailed source-code truncated for AI context efficiency. */ ``` ## 五、合规与伦理 ### 5.1 法律边界 1. **遵守服务条款**：了解目标网站的使用条款 2. **合理使用**：仅用于学习研究或授权的测试 3. **不用于恶意目的**：不进行攻击、欺诈等行为 4. **保护用户隐私**：不获取、存储他人隐私信息 ### 5.2 替代方案 在某些场景下，可以考虑替代方案： 1. **官方 API**：很多平台提供开放 API，无需处理验证码 2. **登录态复用**：使用已登录的 Cookie，减少登录次数 3. **降低频率**：控制请求频率，避免触发验证码 4. **联系网站**：对于数据研究，可联系网站获取授权 ### 5.3 成本控制 使用打码平台时的成本控制建议： ```python class CostController: """打码成本控制器""" def __init__( self, daily_budget: float, cost_per_captcha: float = 0.01 ): self.daily_budget = daily_budget self.cost_per_captcha = cost_per_captcha self._daily_spent = 0.0 self._last_reset = None def can_use_service(self) -> bool: """是否可以使用打码服务""" self._check_reset() return self._daily_spent < self.daily_budget def record_usage(self): """记录一次使用""" self._daily_spent += self.cost_per_captcha def _check_reset(self): """检查是否需要重置""" from datetime import date today = date.today() if self._last_reset != today: self._daily_spent = 0.0 self._last_reset = today @property def remaining_budget(self) -> float: """剩余预算""" self._check_reset() return max(0, self.daily_budget - self._daily_spent) ``` ## 六、实战示例 ### 6.1 完整的验证码处理流程 ```python import asyncio from playwright.async_api import async_playwright async def demo_captcha_solving(): """验证码处理完整演示""" # 初始化 OCR 识别器 ocr_solver = OCRCaptchaSolver(preprocess=True) async with async_playwright() as p: browser = await p.chromium.launch(headless=False) page = await browser.new_page() await page.goto("https://example.com/login") # 获取验证码图片 captcha_element = page.locator("#captcha-image") captcha_bytes = await captcha_element.screenshot() # 识别验证码 code = ocr_solver.solve(captcha_bytes) print(f"识别结果: {code}") # 输入验证码 if code: await page.fill("#captcha-input", code) await page.click("#submit-btn") await asyncio.sleep(3) await browser.close() ``` ## 七、验证码处理实战 大型网站在特定场景下会触发验证码，本节介绍常见验证码的特点和处理方法。 ### 7.1 验证码触发场景 ```mermaid flowchart TD request[发起请求] --> check{风控检测} check -->|正常| success[返回数据] check -->|异常| trigger{触发验证码} trigger -->|高频请求| slider[滑块验证码] trigger -->|IP异常| geetest[极验验证码] trigger -->|敏感操作| click[点选验证码] slider --> verify{验证} geetest --> verify click --> verify verify -->|成功| success verify -->|失败| block[临时封禁] ``` ### 7.2 常见验证码类型 | 场景 | 验证码类型 | 触发条件 | 处理难度 | |-----|-----------|---------|---------| | 登录保护 | 滑块验证码 | 异地登录、频繁登录 | ⭐⭐⭐ | | 接口防护 | 极验验证码 | 请求频率过高 | ⭐⭐⭐⭐ | | 敏感操作 | 点选验证码 | 短时间大量操作 | ⭐⭐⭐⭐⭐ | | 批量操作 | 简单确认 | 批量提交 | ⭐ | ### 7.3 滑块验证码处理 ``` /* Detailed source-code truncated for AI context efficiency. */ ``` ### 7.4 避免触发验证码的策略 在爬虫开发中，预防优于处理： ```mermaid flowchart LR subgraph 预防策略 A[控制请求频率] --> B[模拟真实行为] B --> C[使用登录态] C --> D[IP轮换] end subgraph 建议配置 E[每分钟<30请求] F[随机延迟2-5秒] G[保持Cookie有效] H[高匿代理池] end A --> E B --> F C --> G D --> H ``` ``` /* Detailed source-code truncated for AI context efficiency. */ ``` --- ## 本章小结 本章介绍了验证码识别与处理的核心技术： 1. **验证码类型**：图片字符、滑块、点选等多种类型 2. **OCR 识别**：使用 ddddocr 进行图片验证码识别 3. **滑块处理**：缺口检测和人类轨迹模拟 4. **打码平台**：第三方服务的接入和成本控制 5. **合规考虑**：法律边界和替代方案 6. **实战演练**：验证码触发场景和处理策略 验证码处理是爬虫进阶的重要技能，但务必在合法合规的前提下使用。 ## 下一章预告 下一章我们将学习**数据清洗与预处理**，包括文本清洗、正则表达式应用、数据去重等技术。爬取的数据往往需要清洗才能使用，这是数据处理流程中的关键环节。 --- ### 爬虫进价 09 数据清洗与预处理 (`docs/爬虫进价/09_数据清洗与预处理.md`) # 09_数据清洗与预处理 爬取的原始数据往往包含噪声、冗余和格式不统一等问题。本章将介绍数据清洗的核心技术，帮助你将"脏数据"转换为可用的"干净数据"。 ## 一、数据清洗概述 ### 1.1 为什么需要数据清洗 爬取的原始数据常见问题： | 问题类型 | 示例 | 影响 | |---------|------|------| | HTML 残留 | `

文本

` | 数据不纯净 | | 空白字符 | 多余空格、换行 | 格式混乱 | | 编码问题 | 乱码、特殊字符 | 无法正常显示 | | 重复数据 | 相同内容多次出现 | 数据冗余 | | 格式不统一 | 日期格式各异 | 难以分析 | | 缺失值 | 空字段 | 数据不完整 | ### 1.2 数据清洗流程 ```mermaid flowchart LR raw[原始数据] --> format[格式清洗] format --> content[内容清洗] content --> dedup[去重处理] dedup --> normalize[标准化] normalize --> clean[干净数据] subgraph 格式清洗 html[HTML标签移除] encode[编码修复] end subgraph 内容清洗 whitespace[空白处理] special[特殊字符] end ``` ### 1.3 数据质量评估 在清洗前后，评估数据质量： ```python from dataclasses import dataclass from typing import List, Dict, Any @dataclass class DataQualityReport: """数据质量报告""" total_records: int valid_records: int duplicate_count: int missing_fields: Dict[str, int] format_issues: Dict[str, int] @property def validity_rate(self) -> float: """有效率""" return self.valid_records / self.total_records if self.total_records > 0 else 0 @property def duplicate_rate(self) -> float: """重复率""" return self.duplicate_count / self.total_records if self.total_records > 0 else 0 def __str__(self) -> str: return f""" 数据质量报告: 总记录数: {self.total_records} 有效记录: {self.valid_records} ({self.validity_rate:.1%}) 重复记录: {self.duplicate_count} ({self.duplicate_rate:.1%}) 缺失字段: {self.missing_fields} 格式问题: {self.format_issues} """.strip() ``` ## 二、文本清洗 ### 2.1 HTML 标签移除 ```python import re from typing import Optional class HTMLCleaner: """HTML 清洗器""" # 需要完全移除的标签（包括内容） REMOVE_TAGS = ['script', 'style', 'head', 'meta', 'link'] @staticmethod def remove_tags(html: str) -> str: """移除所有 HTML 标签""" # 先移除特定标签及其内容 for tag in HTMLCleaner.REMOVE_TAGS: pattern = f'<{tag}[^>]*>.*?' html = re.sub(pattern, '', html, flags=re.DOTALL | re.IGNORECASE) # 移除所有标签 html = re.sub(r'<[^>]+>', '', html) return html @staticmethod def remove_tags_keep_text(html: str) -> str: """移除标签但保留文本内容""" # 处理常见的块级元素，添加换行 html = re.sub(r'', '\n', html, flags=re.IGNORECASE) # 移除其他标签 html = re.sub(r'<[^>]+>', '', html) return html @staticmethod def decode_entities(text: str) -> str: """解码 HTML 实体""" import html return html.unescape(text) # 使用 BeautifulSoup（更可靠） def clean_html_with_bs4(html: str) -> str: """使用 BeautifulSoup 清洗 HTML""" try: from bs4 import BeautifulSoup soup = BeautifulSoup(html, 'html.parser') # 移除脚本和样式 for script in soup(['script', 'style']): script.decompose() # 获取文本 text = soup.get_text(separator='\n') return text except ImportError: return HTMLCleaner.remove_tags(html) ``` ### 2.2 空白字符处理 ```python class WhitespaceCleaner: """空白字符清洗器""" @staticmethod def normalize_whitespace(text: str) -> str: """标准化空白字符""" # 将所有空白字符转为普通空格 text = re.sub(r'[\t\r\f\v]', ' ', text) # 合并多个空格 text = re.sub(r' +', ' ', text) # 合并多个换行 text = re.sub(r'\n+', '\n', text) # 去除首尾空白 return text.strip() @staticmethod def remove_all_whitespace(text: str) -> str: """移除所有空白字符""" return re.sub(r'\s+', '', text) @staticmethod def trim_lines(text: str) -> str: """去除每行首尾空白""" lines = text.split('\n') return '\n'.join(line.strip() for line in lines) @staticmethod def remove_empty_lines(text: str) -> str: """移除空行""" lines = text.split('\n') return '\n'.join(line for line in lines if line.strip()) ``` ### 2.3 特殊字符清理 ``` /* Detailed source-code truncated for AI context efficiency. */ ``` ### 2.4 编码问题处理 ```python import chardet class EncodingFixer: """编码问题修复器""" @staticmethod def detect_encoding(data: bytes) -> str: """检测编码""" result = chardet.detect(data) return result['encoding'] or 'utf-8' @staticmethod def fix_encoding(text: str, source_encoding: str = None) -> str: """修复编码问题""" try: # 如果是乱码，尝试重新编码 if source_encoding: return text.encode('latin1').decode(source_encoding) else: # 尝试常见编码 for encoding in ['utf-8', 'gbk', 'gb2312', 'big5']: try: return text.encode('latin1').decode(encoding) except (UnicodeDecodeError, UnicodeEncodeError): continue except Exception: pass return text @staticmethod def safe_decode(data: bytes, fallback: str = 'utf-8') -> str: """安全解码""" # 检测编码 detected = EncodingFixer.detect_encoding(data) try: return data.decode(detected) except (UnicodeDecodeError, TypeError): return data.decode(fallback, errors='ignore') ``` ## 三、正则表达式高级应用 ### 3.1 常用提取模式 ```python class RegexPatterns: """常用正则表达式模式""" # 基础模式 CHINESE = r'[\u4e00-\u9fa5]+' # 中文 EMAIL = r'[\w.+-]+@[\w-]+\.[\w.-]+' # 邮箱 PHONE = r'1[3-9]\d{9}' # 中国手机号 URL = r'https?://[^\s<>"{}|\\^`\[\]]+' # URL IP = r'\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}' # IP 地址 # 日期时间模式 DATE_YMD = r'\d{4}[-/年]\d{1,2}[-/月]\d{1,2}日?' # 年月日 TIME_HMS = r'\d{1,2}:\d{2}(:\d{2})?' # 时分秒 DATETIME = r'\d{4}[-/]\d{1,2}[-/]\d{1,2}\s+\d{1,2}:\d{2}(:\d{2})?' # 数值模式 INTEGER = r'-?\d+' FLOAT = r'-?\d+\.?\d*' PRICE = r'[¥$￥]\s*\d+\.?\d*' # 价格 PERCENTAGE = r'\d+\.?\d*%' # 百分比 class RegexExtractor: """正则表达式提取器""" @staticmethod def extract_all(text: str, pattern: str) -> list: """提取所有匹配""" return re.findall(pattern, text) @staticmethod def extract_first(text: str, pattern: str) -> str: """提取第一个匹配""" match = re.search(pattern, text) return match.group() if match else '' @staticmethod def extract_groups(text: str, pattern: str) -> dict: """提取命名分组""" match = re.search(pattern, text) return match.groupdict() if match else {} @staticmethod def extract_between(text: str, start: str, end: str) -> list: """提取两个标记之间的内容""" pattern = f'{re.escape(start)}(.*?){re.escape(end)}' return re.findall(pattern, text, re.DOTALL) ``` ### 3.2 高级替换 ```python class RegexReplacer: """正则表达式替换器""" @staticmethod def replace_with_callback( text: str, pattern: str, callback ) -> str: """使用回调函数替换""" return re.sub(pattern, callback, text) @staticmethod def mask_sensitive(text: str) -> str: """脱敏处理""" # 手机号脱敏 text = re.sub( r'(1[3-9]\d)\d{4}(\d{4})', r'\1****\2', text ) # 邮箱脱敏 text = re.sub( r'([\w.+-]{1,3})[\w.+-]*(@[\w-]+\.[\w.-]+)', r'\1***\2', text ) # 身份证脱敏 text = re.sub( r'(\d{6})\d{8}(\d{4})', r'\1********\2', text ) return text @staticmethod def clean_url_params(url: str, keep_params: list = None) -> str: """清理 URL 参数""" from urllib.parse import urlparse, parse_qs, urlencode, urlunparse parsed = urlparse(url) params = parse_qs(parsed.query) if keep_params: params = {k: v for k, v in params.items() if k in keep_params} else: params = {} new_query = urlencode(params, doseq=True) return urlunparse(parsed._replace(query=new_query)) ``` ## 四、数据去重 ### 4.1 精确去重 ```python from typing import List, Dict, Any, Callable import hashlib class ExactDeduplicator: """精确去重器""" @staticmethod def dedupe_list(items: List[str]) -> List[str]: """列表去重（保持顺序）""" seen = set() result = [] for item in items: if item not in seen: seen.add(item) result.append(item) return result @staticmethod def dedupe_dicts( items: List[Dict], key_field: str ) -> List[Dict]: """字典列表去重""" seen = set() result = [] for item in items: key = item.get(key_field) if key not in seen: seen.add(key) result.append(item) return result @staticmethod def dedupe_by_hash( items: List[Dict], fields: List[str] ) -> List[Dict]: """ 根据多个字段计算哈希去重 Args: items: 数据列表 fields: 用于计算哈希的字段 Returns: 去重后的列表 """ seen = set() result = [] for item in items: # 计算哈希 key_str = '|'.join(str(item.get(f, '')) for f in fields) key_hash = hashlib.md5(key_str.encode()).hexdigest() if key_hash not in seen: seen.add(key_hash) result.append(item) return result ``` ### 4.2 模糊去重 ``` /* Detailed source-code truncated for AI context efficiency. */ ``` ## 五、数据标准化 ### 5.1 日期时间标准化 ``` /* Detailed source-code truncated for AI context efficiency. */ ``` ### 5.2 数值标准化 ```python class NumberNormalizer: """数值标准化器""" @staticmethod def parse_number(text: str) -> float: """ 解析数字（支持中文单位） Examples: "1.5万" -> 15000 "3.2亿" -> 320000000 "1,234.56" -> 1234.56 """ text = text.strip() # 中文单位映射 units = { '万': 10000, '亿': 100000000, 'k': 1000, 'K': 1000, 'm': 1000000, 'M': 1000000, 'b': 1000000000, 'B': 1000000000, } multiplier = 1 for unit, value in units.items(): if unit in text: multiplier = value text = text.replace(unit, '') break # 移除逗号 text = text.replace(',', '') # 提取数字 match = re.search(r'-?\d+\.?\d*', text) if match: return float(match.group()) * multiplier return 0.0 @staticmethod def format_number( value: float, precision: int = 2, use_units: bool = True ) -> str: """格式化数字""" if not use_units: return f'{value:.{precision}f}' if value >= 100000000: return f'{value/100000000:.{precision}f}亿' elif value >= 10000: return f'{value/10000:.{precision}f}万' else: return f'{value:.{precision}f}' ``` ### 5.3 文本标准化 ```python class TextNormalizer: """文本标准化器""" @staticmethod def normalize(text: str) -> str: """完整的文本标准化流程""" # 1. Unicode 标准化 text = unicodedata.normalize('NFKC', text) # 2. 全角转半角 text = SpecialCharCleaner.to_halfwidth(text) # 3. 移除控制字符 text = SpecialCharCleaner.remove_control_chars(text) # 4. 标准化空白 text = WhitespaceCleaner.normalize_whitespace(text) return text @staticmethod def normalize_punctuation(text: str) -> str: """标准化标点符号""" # 中文标点转英文 mapping = { '，': ', ', '。': '. ', '！': '! ', '？': '? ', '；': '; ', '：': ': ', '"': '"', '"': '"', ''': "'", ''': "'", } for cn, en in mapping.items(): text = text.replace(cn, en) return text ``` ## 六、综合数据清洗器 ``` /* Detailed source-code truncated for AI context efficiency. */ ``` ## 七、实战示例 ``` /* Detailed source-code truncated for AI context efficiency. */ ``` ## 八、数据清洗实战 本节以网页数据为例，演示完整的数据清洗流程。 ### 8.1 数据特点分析 ```mermaid flowchart TD subgraph 原始数据问题 title_em["标题含HTML标签
(搜索高亮)"] view_unit["数值含单位
(1.5万、3.2亿)"] duration_format["时长格式不一
(02:30、1:23:45)"] pubdate_format["发布时间多样
(刚刚、3天前、2024-01-15)"] desc_html["描述含HTML
(换行、链接)"] end subgraph 清洗目标 title_clean[纯文本标题] view_num[数值] duration_sec[秒数时长] pubdate_std[标准日期] desc_plain[纯文本描述] end title_em --> title_clean view_unit --> view_num duration_format --> duration_sec pubdate_format --> pubdate_std desc_html --> desc_plain ``` ### 8.2 数据清洗器 ``` /* Detailed source-code truncated for AI context efficiency. */ ``` ### 8.3 数据去重 ```python import hashlib from typing import List, Set class DataDeduplicator: """数据去重器""" def __init__(self): self._seen_bvids: Set[str] = set() self._seen_hashes: Set[str] = set() def is_duplicate_by_bvid(self, bvid: str) -> bool: """通过BV号判断是否重复""" if bvid in self._seen_bvids: return True self._seen_bvids.add(bvid) return False def is_duplicate_by_content(self, title: str, owner_mid: int) -> bool: """ 通过内容哈希判断是否重复 （用于检测同一UP主发布的相似标题视频） """ content = f"{title}|{owner_mid}" content_hash = hashlib.md5(content.encode()).hexdigest() if content_hash in self._seen_hashes: return True self._seen_hashes.add(content_hash) return False def dedupe_videos( self, videos: List[WebContent] ) -> List[WebContent]: """ 去重视频列表 优先使用BV号去重，同时检测内容重复 """ results = [] for video in videos: # BV号去重 if self.is_duplicate_by_bvid(video.bvid): continue # 内容去重（可选） if self.is_duplicate_by_content(video.title, video.owner_mid): continue results.append(video) return results def reset(self): """重置去重状态""" self._seen_bvids.clear() self._seen_hashes.clear() ``` ### 8.4 数据质量报告 ``` /* Detailed source-code truncated for AI context efficiency. */ ``` --- ## 本章小结 本章介绍了数据清洗与预处理的核心技术： 1. **文本清洗**：HTML 移除、空白处理、特殊字符、编码修复 2. **正则表达式**：常用模式、提取与替换、脱敏处理 3. **数据去重**：精确去重和模糊去重 4. **数据标准化**：日期、数值、文本的统一格式 5. **综合实战**：视频数据清洗、播放量解析、时长转换、发布时间标准化 数据清洗是数据处理流程中的关键环节，干净的数据才能产生有价值的分析结果。 ## 下一章预告 下一章我们将学习**数据分析与可视化**，包括使用 pandas 进行数据统计、生成词云、绑制图表等技术。这些技术可以帮助我们从爬取的数据中提取有价值的洞察。 --- ### 爬虫进价 10 数据分析与可视化 (`docs/爬虫进价/10_数据分析与可视化.md`) # 第十章：数据分析与可视化 > 本章将学习如何使用 Python 数据分析和可视化工具处理爬取的数据。我们将掌握 pandas 进行数据统计分析、使用词云展示文本数据、利用 matplotlib 和 pyecharts 生成各类图表，最终实现自动化的数据分析报告生成。 ## 10.1 数据分析概述 ### 为什么需要数据分析 爬虫获取的原始数据只是第一步，真正的价值在于从数据中提取有用的信息和洞察： - **发现数据规律**：了解数据的分布、趋势和特征 - **验证假设**：用数据验证业务假设和决策 - **支持决策**：为业务决策提供数据依据 - **数据可视化**：将复杂数据转化为直观的图表 ### 数据分析流程 ```mermaid flowchart LR subgraph 数据准备 A[原始数据] --> B[数据清洗] B --> C[数据转换] end subgraph 数据分析 C --> D[统计分析] C --> E[文本分析] C --> F[时间序列] end subgraph 结果输出 D --> G[图表可视化] E --> H[词云生成] F --> G G --> I[分析报告] H --> I end ``` ### 主要工具介绍 | 工具 | 用途 | 特点 | |------|------|------| | pandas | 数据处理和分析 | 功能强大，生态完善 | | jieba | 中文分词 | 准确率高，易于使用 | | wordcloud | 词云生成 | 可定制性强 | | matplotlib | 静态图表 | 功能全面，输出清晰 | | pyecharts | 交互式图表 | 效果炫酷，支持 Web | ## 10.2 pandas 数据分析 ### DataFrame 基础操作 pandas 是 Python 数据分析的核心库，DataFrame 是其最重要的数据结构： ```python import pandas as pd # 从爬取的数据创建 DataFrame data = [ {"title": "Python 教程", "views": 15000, "likes": 320, "date": "2024-01-15"}, {"title": "爬虫入门", "views": 12000, "likes": 280, "date": "2024-01-16"}, {"title": "数据分析", "views": 18000, "likes": 450, "date": "2024-01-17"}, ] df = pd.DataFrame(data) # 基础信息 print(df.info()) # 列类型和非空数量 print(df.describe()) # 数值列统计摘要 # 数据选择 print(df['title']) # 选择单列 print(df[['title', 'views']]) # 选择多列 print(df[df['views'] > 14000]) # 条件筛选 # 排序 print(df.sort_values('views', ascending=False)) # 新增列 df['engagement'] = df['likes'] / df['views'] * 100 ``` ### 数据聚合与分组统计 分组统计是数据分析中最常用的操作： ```python # 假设有更多数据 data = [ {"category": "技术", "title": "Python 教程", "views": 15000}, {"category": "技术", "title": "爬虫入门", "views": 12000}, {"category": "生活", "title": "美食推荐", "views": 8000}, {"category": "生活", "title": "旅行日记", "views": 10000}, {"category": "技术", "title": "数据分析", "views": 18000}, ] df = pd.DataFrame(data) # 按分类分组统计 grouped = df.groupby('category') print(grouped['views'].sum()) # 各分类总浏览量 print(grouped['views'].mean()) # 各分类平均浏览量 print(grouped['views'].agg(['sum', 'mean', 'max', 'min'])) # 多个统计量 # 多列聚合 print(df.groupby('category').agg({ 'views': 'sum', 'title': 'count' }).rename(columns={'title': 'article_count'})) ``` ### 数据透视表 数据透视表可以快速进行多维度分析： ```python # 更复杂的数据 data = [ {"date": "2024-01", "category": "技术", "platform": "PC", "views": 15000}, {"date": "2024-01", "category": "技术", "platform": "Mobile", "views": 8000}, {"date": "2024-01", "category": "生活", "platform": "PC", "views": 6000}, {"date": "2024-02", "category": "技术", "platform": "PC", "views": 18000}, {"date": "2024-02", "category": "生活", "platform": "Mobile", "views": 9000}, ] df = pd.DataFrame(data) # 创建透视表：按日期和分类统计各平台浏览量 pivot = pd.pivot_table( df, values='views', index='date', columns='category', aggfunc='sum', fill_value=0 ) print(pivot) # 带边距的透视表 pivot_with_margins = pd.pivot_table( df, values='views', index='date', columns='category', aggfunc='sum', margins=True, margins_name='总计' ) print(pivot_with_margins) ``` ### 时间序列分析 爬虫数据通常包含时间维度，时间序列分析很有价值： ```python # 创建时间序列数据 dates = pd.date_range('2024-01-01', periods=30, freq='D') views = [1000 + i * 50 + (i % 7) * 200 for i in range(30)] df = pd.DataFrame({'date': dates, 'views': views}) df.set_index('date', inplace=True) # 滚动统计（7天移动平均） df['rolling_avg'] = df['views'].rolling(window=7).mean() # 按周统计 weekly = df.resample('W').agg({ 'views': ['sum', 'mean'] }) print(weekly) # 环比增长率 df['growth_rate'] = df['views'].pct_change() * 100 ``` ## 10.3 词云生成 词云是展示文本数据的直观方式，在分析评论、标题等文本数据时非常有用。 ### jieba 中文分词 jieba 是最流行的中文分词库： ```python import jieba import jieba.analyse # 基础分词 text = "Python爬虫教程帮助你快速入门数据采集技术" words = jieba.lcut(text) print(words) # ['Python', '爬虫', '教程', '帮助', '你', '快速', '入门', '数据', '采集', '技术'] # 提取关键词（TF-IDF） keywords = jieba.analyse.extract_tags( text, topK=5, withWeight=True ) for word, weight in keywords: print(f"{word}: {weight:.4f}") # 提取关键词（TextRank） keywords_tr = jieba.analyse.textrank( text, topK=5, withWeight=True ) ``` ### 停用词过滤 停用词是指对文本分析没有实际意义的词（如"的"、"是"、"在"等）： ```python # 常用中文停用词 STOPWORDS = { '的', '是', '在', '了', '和', '与', '或', '有', '个', '人', '这', '那', '就', '都', '也', '为', '对', '到', '从', '把', '被', '让', '给', '向', '往', '于', '及', '以', '等', '不', '很', '会', '能', '可', '要', '我', '你', '他', '她', '它', } def filter_stopwords(words: list) -> list: """过滤停用词和单字""" return [ w for w in words if w not in STOPWORDS and len(w) > 1 ] # 使用 text = "这是一个关于Python爬虫的教程，帮助你快速入门" words = jieba.lcut(text) filtered = filter_stopwords(words) print(filtered) # ['Python', '爬虫', '教程', '帮助', '快速', '入门'] ``` ### 词云生成 使用 wordcloud 库生成词云： ``` /* Detailed source-code truncated for AI context efficiency. */ ``` ### 自定义词云形状 可以使用图片作为词云的形状蒙版： ```python import numpy as np from PIL import Image from wordcloud import WordCloud, ImageColorGenerator def generate_shaped_wordcloud( text: str, mask_image_path: str, output_path: str, font_path: str = None, use_mask_colors: bool = True ): """ 生成自定义形状的词云 Args: text: 词语文本 mask_image_path: 形状蒙版图片路径 output_path: 输出路径 font_path: 字体路径 use_mask_colors: 是否使用蒙版图片的颜色 """ # 读取蒙版图片 mask = np.array(Image.open(mask_image_path)) # 创建词云 wc = WordCloud( mask=mask, background_color="white", font_path=font_path, max_words=500, max_font_size=80, random_state=42, contour_width=1, contour_color='steelblue' ) wc.generate(text) # 使用蒙版颜色 if use_mask_colors: image_colors = ImageColorGenerator(mask) wc.recolor(color_func=image_colors) wc.to_file(output_path) return output_path ``` ## 10.4 数据可视化 ### matplotlib 基础图表 matplotlib 是 Python 最基础的可视化库： ``` /* Detailed source-code truncated for AI context efficiency. */ ``` ### pyecharts 交互式图表 pyecharts 可以生成交互式的 HTML 图表： ``` /* Detailed source-code truncated for AI context efficiency. */ ``` ## 10.5 数据分析报告生成 ### Markdown 报告模板 自动生成 Markdown 格式的分析报告： ``` /* Detailed source-code truncated for AI context efficiency. */ ``` ### 完整分析流程示例 ``` /* Detailed source-code truncated for AI context efficiency. */ ``` ## 10.6 实战案例：社交媒体评论分析 让我们综合运用本章所学，对爬取的社交媒体评论数据进行完整分析： ``` /* Detailed source-code truncated for AI context efficiency. */ ``` ## 10.7 视频数据分析实战 本节以视频数据为例，演示完整的数据分析流程。 ### 数据分析目标 ```mermaid flowchart TD subgraph 输入数据 videos[视频数据] comments[弹幕/评论数据] end subgraph 分析维度 stat[播放量/互动统计] up[UP主分析] time[发布时间分析] keyword[标题关键词] end subgraph 输出 rank[热门视频排行] trend[发布趋势图] cloud[标题词云] report[分析报告] end videos --> stat --> rank videos --> up --> rank videos --> time --> trend videos --> keyword --> cloud rank --> report trend --> report cloud --> report ``` ### 视频数据分析器 ``` /* Detailed source-code truncated for AI context efficiency. */ ``` --- ## 本章小结 本章我们学习了数据分析与可视化的核心技术： 1. **pandas 数据分析** - DataFrame 是数据分析的核心数据结构 - groupby 实现分组聚合统计 - pivot_table 进行多维分析 - 时间序列支持滚动统计和重采样 2. **词云生成** - jieba 实现中文分词 - 停用词过滤提升分析质量 - wordcloud 库生成静态词云 - 支持自定义形状和颜色 3. **数据可视化** - matplotlib 生成静态图表 - pyecharts 生成交互式图表 - 选择合适的图表类型展示数据 4. **自动化报告** - Markdown 格式便于阅读和分享 - 模板化生成提高效率 - 图表嵌入增强可读性 5. **综合实战** - 视频数据多维统计分析 - 发布时间规律分析 - UP主活跃度排名 - 标题关键词词云 **关键要点：** - 数据分析的目标是发现洞察，而非展示技术 - 选择合适的可视化方式，让数据"说话" - 自动化报告生成可以大大提高效率 - 中文分析需要注意字体和编码问题 --- ## 下一章预告 在最后一章「进阶综合实战项目」中，我们将综合运用整个进阶教程所学的所有技术，实现一个完整的社交媒体数据采集与分析工具。该项目将包含： - 多种登录方式支持 - 反检测浏览器自动化 - 代理 IP 轮换 - 数据清洗和存储 - 词云和图表分析 敬请期待！ --- ### 爬虫进价 11 进阶综合实战项目 (`docs/爬虫进价/11_进阶综合实战项目.md`) # 第十一章：进阶综合实战项目 > 本章将综合运用整个进阶教程所学的所有技术，以 **B站（bilibili.com）** 为目标，实现一个完整的视频数据采集与分析工具。该项目将包含登录认证、API 签名、浏览器自动化、数据存储、分析报告等完整功能链路。 ## 11.1 项目概述 ### 项目目标 构建一个类似 [MediaCrawler](https://github.com/NanmiCoder/MediaCrawler) 简化版的视频数据采集工具，以 **B站** 为目标平台，具备以下能力： - **登录认证**：支持扫码登录和 Cookie 登录 - **API 签名**：实现 B站 WBI 签名算法 - **视频搜索**：按关键词搜索视频 - **视频详情**：获取完整视频信息（播放量、点赞、收藏等） - **数据存储**：支持 JSON、CSV 两种存储方式 - **数据分析**：自动生成词云和统计报告 > **目标网站说明：** > - 网站：https://www.bilibili.com > - 类型：国内最大的视频社区平台 > - 特点：需要登录获取完整数据，API 有 WBI 签名保护 > - 数据：视频标题、UP主、播放量、点赞、收藏、弹幕数等 ### 整体架构图 在开始编码之前，让我们先从宏观角度理解整个项目的架构： ```mermaid graph TB subgraph 入口层 main["main.py
程序入口"] end subgraph 核心模块 config["config 配置
settings.py
bilibili_config.py"] crawler["crawler 爬虫
spider.py
核心调度"] store["store 存储
backend.py
JSON/CSV"] end subgraph 功能模块 login["login 登录
auth.py
扫码/Cookie"] client["client 客户端
bilibili_client.py
API请求"] analysis["analysis 分析
report.py
词云/统计"] end subgraph 基础模块 core["core 浏览器
browser.py
Playwright封装"] tools["tools 工具
sign.py
WBI签名"] models["models 模型
bilibili.py
数据结构"] end main --> config main --> crawler main --> store main --> analysis crawler --> login crawler --> client login --> core client --> tools tools --> models config -.->|配置注入| crawler crawler -->|数据| store ``` ### 爬虫执行流程 整个爬虫从启动到完成的完整流程如下： ```mermaid flowchart TD start([程序启动 main.py]) --> step1 subgraph step1 [步骤1: 初始化浏览器] browser["BrowserManager.start()
启动 Playwright
创建 BrowserContext"] end step1 --> step2 subgraph step2 [步骤2: 登录认证] check{检查登录状态} check -->|已登录| skip[跳过登录] check -->|未登录| login_flow subgraph login_flow [执行登录] qrcode["扫码登录: 显示二维码-等待扫码-获取Cookie"] cookie["Cookie登录: 注入Cookie-验证有效性"] end end step2 --> step3 subgraph step3 [步骤3: 初始化API客户端] init["同步Cookie到httpx
获取WBI签名密钥
初始化签名器"] end step3 --> step4 subgraph step4 [步骤4: 执行爬取任务] mode{爬取模式} mode -->|SEARCH| search["关键词搜索-翻页获取-获取详情"] mode -->|DETAIL| detail[直接获取指定视频详情] search --> request detail --> request request["每次请求:
构造参数-WBI签名-发送请求-解析响应-随机延迟"] end step4 --> step5 subgraph step5 [步骤5: 数据存储] save["BilibiliVideo对象列表
转换为字典
保存为 JSON/CSV"] end step5 --> step6 subgraph step6 [步骤6: 生成分析报告] report["统计分析-词频统计
生成词云-输出Markdown报告"] end step6 --> step7 subgraph step7 [步骤7: 清理资源] cleanup["关闭浏览器
保存登录状态
程序退出"] end step7 --> finish([完成]) ``` ### 数据流向图 理解数据在各模块之间如何流转： ```mermaid flowchart LR subgraph input [用户输入] keyword["关键词
配置文件"] end subgraph process [系统处理] search["搜索API
(WBI签名)"] detail["详情API
(获取详情)"] validate["Pydantic数据验证
BilibiliVideo"] end subgraph output [最终输出] json["JSON文件
(结构化)"] csv["CSV文件
(表格化)"] report["分析报告
(Markdown)"] wordcloud["词云图片
(PNG)"] end keyword --> search search -->|视频列表| detail detail --> validate validate --> json validate --> csv validate --> report report --> wordcloud ``` ### 参考项目 本项目参考 [MediaCrawler](https://github.com/NanmiCoder/MediaCrawler) 的 B站实现： | 文件 | 说明 | |------|------| | `media_platform/bilibili/core.py` | 爬虫核心逻辑 | | `media_platform/bilibili/client.py` | API 客户端 | | `media_platform/bilibili/login.py` | 登录认证 | | `media_platform/bilibili/help.py` | WBI 签名算法 | ### 技术栈 | 模块 | 技术选型 | 作用 | |------|----------|------| | 配置管理 | pydantic-settings | 类型安全的配置，支持环境变量 | | 日志系统 | loguru | 优雅的日志记录和轮转 | | 浏览器自动化 | Playwright | 处理登录、获取Cookie和签名密钥 | | HTTP 客户端 | httpx | 异步HTTP请求，高性能 | | 数据验证 | Pydantic | 数据模型定义和验证 | | 数据分析 | pandas + jieba + wordcloud | 统计分析和可视化 | ### 项目结构 ``` 11_进阶综合实战项目/ ├── config/ # 配置模块 │ ├── __init__.py │ ├── settings.py # 通用配置 │ └── bilibili_config.py # B站特定配置 ├── core/ # 核心模块 │ ├── __init__.py │ └── browser.py # 浏览器管理 ├── login/ # 登录模块 │ ├── __init__.py │ └── auth.py # B站登录认证 ├── client/ # API 客户端模块 │ ├── __init__.py │ └── bilibili_client.py # B站 API 客户端 ├── crawler/ # 爬虫模块 │ ├── __init__.py │ └── spider.py # B站爬虫实现 ├── store/ # 存储模块 │ ├── __init__.py │ └── backend.py # 存储后端 ├── proxy/ # 代理模块（可选） │ ├── __init__.py │ └── pool.py # 代理池 ├── models/ # 数据模型模块 │ ├── __init__.py │ └── bilibili.py # B站数据模型 ├── tools/ # 工具模块 │ ├── __init__.py │ └── sign.py # WBI 签名工具 ├── analysis/ # 分析模块 │ ├── __init__.py │ └── report.py # 报告生成 └── main.py # 入口文件 ``` ## 11.2 配置模块设计 ### 通用配置 使用 pydantic-settings 实现类型安全的配置管理： ```python # config/settings.py from pydantic_settings import BaseSettings from pydantic import Field from typing import Optional, List from enum import Enum class StorageType(str, Enum): """存储类型""" JSON = "json" CSV = "csv" class LoginType(str, Enum): """登录类型""" COOKIE = "cookie" QRCODE = "qrcode" class CrawlerType(str, Enum): """爬取类型""" SEARCH = "search" # 关键词搜索 DETAIL = "detail" # 指定视频详情 class Settings(BaseSettings): """项目配置""" # 基础配置 app_name: str = "BilibiliCrawler" debug: bool = False # 浏览器配置 browser_headless: bool = False # B站扫码登录需要显示浏览器 browser_timeout: int = 30000 browser_user_data_dir: Optional[str] = "./browser_data" save_login_state: bool = True # 登录配置 login_type: LoginType = LoginType.QRCODE cookie_str: str = "" # 爬虫配置 crawler_type: CrawlerType = CrawlerType.SEARCH keywords: str = "Python教程" # 搜索关键词，多个用逗号分隔 specified_id_list: List[str] = [] # 指定视频列表 max_video_count: int = 20 max_concurrency: int = 3 crawl_delay_min: float = 1.0 crawl_delay_max: float = 3.0 # 存储配置 storage_type: StorageType = StorageType.JSON storage_output_dir: str = "./output" class Config: env_file = ".env" env_prefix = "CRAWLER_" # 全局配置实例 settings = Settings() ``` ### B站特定配置 ```python # config/bilibili_config.py """B站 API 配置""" # API 地址 SEARCH_URL = "https://api.bilibili.com/x/web-interface/wbi/search/type" VIDEO_INFO_URL = "https://api.bilibili.com/x/web-interface/view" NAV_URL = "https://api.bilibili.com/x/web-interface/nav" # 请求配置 SEARCH_PAGE_SIZE = 20 REQUEST_TIMEOUT = 30 # 默认请求头 DEFAULT_HEADERS = { "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/120.0.0.0 Safari/537.36", "Referer": "https://www.bilibili.com", "Origin": "https://www.bilibili.com", } # WBI 签名密钥混淆表 WBI_MIXIN_KEY_ENC_TAB = [ 46, 47, 18, 2, 53, 8, 23, 32, 15, 50, 10, 31, 58, 3, 45, 35, 27, 43, 5, 49, 33, 9, 42, 19, 29, 28, 14, 39, 12, 38, 41, 13, 37, 48, 7, 16, 24, 55, 40, 61, 26, 17, 0, 1, 60, 51, 30, 4, 22, 25, 54, 21, 56, 59, 6, 63, 57, 62, 11, 36, 20, 34, 44, 52, ] # 登录相关 LOGIN_BUTTON_SELECTOR = "xpath=//div[@class='right-entry__outside go-login-btn']//div" QRCODE_SELECTOR = "//div[@class='login-scan-box']//img" LOGIN_COOKIE_KEYS = ["SESSDATA", "DedeUserID", "bili_jct"] ``` ## 11.3 WBI 签名算法 B站使用 WBI 签名保护 API 请求，需要实现签名算法。 ### WBI 签名原理 WBI（Web Bilibili Interface）签名是 B站用来保护 API 接口的一种机制，防止接口被恶意调用。 **为什么需要签名？** - 防止请求被篡改 - 防止接口被滥用 - 增加爬虫难度 ### WBI 签名流程图 ```mermaid flowchart TD subgraph step1 ["第一步: 获取签名密钥"] api["调用 /x/web-interface/nav API"] api --> response["响应中包含 wbi_img:
{img_url, sub_url}"] response --> extract["提取文件名作为密钥:
img_key, sub_key"] end step1 --> step2 subgraph step2 ["第二步: 生成盐值 Salt"] concat["raw_key = img_key + sub_key"] concat --> mixin["使用混淆表重排字符:
WBI_MIXIN_KEY_ENC_TAB"] mixin --> salt["salt = 重排后取前32位"] end step2 --> step3 subgraph step3 ["第三步: 计算签名"] params["原始参数:
{keyword, page}"] params --> addwts["添加时间戳 wts"] addwts --> sort["按 key 排序并URL编码"] sort --> append["拼接盐值: query + salt"] append --> md5["计算 MD5 得到 w_rid"] md5 --> final["最终参数:
{...原参数, wts, w_rid}"] end ``` > **💡 关于 JS 逆向** > > 你可能会好奇：这个 WBI 签名算法是怎么逆向分析出来的？混淆表 `WBI_MIXIN_KEY_ENC_TAB` 又是从哪里找到的？ > > 别着急！这部分涉及到 **JavaScript 逆向**技术，我会在后面的 **「高级爬虫 - JS 逆向」** 章节中详细讲解。届时会带你一步步分析 B站的前端代码，找出签名算法的实现细节。 > > 本章的重点是让你理解**如何使用**这个签名算法，以及整个项目的工程化架构。签名算法的逆向分析过程，我们后面再深入探讨。 ### 签名器实现 ``` /* Detailed source-code truncated for AI context efficiency. */ ``` ## 11.4 数据模型定义 使用 Pydantic 定义视频数据模型： ``` /* Detailed source-code truncated for AI context efficiency. */ ``` ## 11.5 登录认证模块 登录认证是爬虫获取完整数据的关键步骤。B站对未登录用户有很多数据限制，登录后可以获取更多信息。 ### 为什么需要登录？ | 数据项 | 未登录 | 已登录 | |--------|--------|--------| | 搜索结果 | 有限制 | 完整 | | 视频详情 | 基础信息 | 完整信息 | | 用户数据 | 部分隐藏 | 可见 | | API调用频率 | 严格限制 | 相对宽松 | ### 登录方式对比 | 方式 | 优点 | 缺点 | 适用场景 | |------|------|------|----------| | 扫码登录 | 安全、无需处理复杂逻辑 | 需要手机APP配合 | 首次登录、开发调试 | | Cookie登录 | 快速、可自动化 | Cookie会过期 | 批量部署、定时任务 | ### 扫码登录流程图 ```mermaid sequenceDiagram participant PC as PC浏览器 participant Server as B站服务器 participant APP as B站APP PC->>Server: 1. 访问B站首页 PC->>Server: 2. 点击登录按钮 PC->>Server: 3. 请求二维码+UUID Server-->>PC: 4. 返回二维码图片 Note over PC: 5. 显示二维码 loop 轮询登录状态 PC->>Server: 6. 检查登录状态 APP->>Server: 7. 用户扫描二维码 Server-->>PC: 8. 状态: 已扫描 PC->>Server: 9. 继续轮询 APP->>Server: 10. 用户点击确认 Server-->>PC: 11. 返回登录凭证(Set-Cookie) end Note over PC: 12. 保存Cookie
登录成功! ``` ### Cookie 登录流程图 ```mermaid flowchart LR subgraph input [用户输入] cookie["用户提供Cookie
(从浏览器复制)"] end subgraph process [处理流程] parse["解析Cookie字符串
提取键值对"] inject["注入到BrowserContext
add_cookies()"] visit["访问B站首页
加载页面"] check{"检查关键Cookie
SESSDATA存在?"} end subgraph result [结果] success["登录成功!
开始爬取"] fail["Cookie已过期
需要重新登录"] end cookie --> parse --> inject --> visit --> check check -->|存在| success check -->|不存在| fail ``` **关键Cookie说明：** | Cookie名称 | 说明 | |-----------|------| | SESSDATA | 会话凭证，最重要的登录标识 | | DedeUserID | 用户ID | | bili_jct | CSRF Token，某些操作需要 | ### 登录实现代码 ``` /* Detailed source-code truncated for AI context efficiency. */ ``` ## 11.6 API 客户端 API 客户端是爬虫与 B站服务器交互的核心模块，负责发送请求、处理签名、解析响应。 ### 客户端职责 ```mermaid graph TB subgraph 核心功能 cookie["Cookie管理
• 从浏览器同步
• 注入到请求头
• 验证有效性"] wbi["WBI签名
• 获取密钥
• 参数签名
• 自动刷新"] http["HTTP请求
• GET/POST请求
• 超时处理
• 错误重试"] end subgraph API方法 api["BilibiliClient API
• search_video_by_keyword
• get_video_info
• pong (登录检查)"] end cookie --> api wbi --> api http --> api ``` ### B站 API 列表 | API | 地址 | 功能 | 是否需要签名 | |-----|------|------|-------------| | 用户信息 | `/x/web-interface/nav` | 获取登录用户信息和WBI密钥 | 否 | | 视频搜索 | `/x/web-interface/wbi/search/type` | 按关键词搜索视频 | **是** | | 视频详情 | `/x/web-interface/view` | 获取视频完整信息 | 否 | ### 客户端实现代码 ``` /* Detailed source-code truncated for AI context efficiency. */ ``` ## 11.7 爬虫模块 爬虫模块是整个项目的核心调度器，负责协调浏览器、登录、API客户端等组件完成数据采集任务。 ### 爬虫类设计 ```mermaid graph LR subgraph BilibiliCrawler subgraph 属性 attr1["browser_manager"] --> desc1["管理Playwright浏览器"] attr2["browser_context"] --> desc2["浏览器上下文(Cookie容器)"] attr3["context_page"] --> desc3["页面实例"] attr4["bili_client"] --> desc4["API客户端"] attr5["_results"] --> desc5["爬取结果列表"] end subgraph 方法 m1["start()"] --> d1["启动爬虫(主入口)"] m2["_init_browser()"] --> d2["初始化浏览器"] m3["_do_login()"] --> d3["执行登录"] m4["_init_client()"] --> d4["初始化API客户端"] m5["search_by_keywords()"] --> d5["关键词搜索"] m6["get_specified_videos()"] --> d6["获取指定视频"] m7["close()"] --> d7["清理资源"] end end ``` ### 两种爬取模式 #### SEARCH 模式（关键词搜索视频） ```mermaid flowchart LR subgraph input [关键词列表] k1["Python"] k2["教程"] k3["数据分析"] end subgraph search [搜索分页] p1["第1页"] p2["第2页"] p3["..."] end subgraph result [视频列表] v1["BV1xxx"] v2["BV2xxx"] v3["BV3xxx"] end detail["获取每个视频详情
(完整播放量等)"] input --> search --> result --> detail ``` #### DETAIL 模式（获取指定视频详情） ```mermaid flowchart LR subgraph input [指定BV号列表] bv1["BV1abc"] bv2["BV2def"] bv3["BV3ghi"] end api["遍历列表
逐个调用详情API"] input --> api ``` ### 反爬策略 为了避免被 B站 封禁，爬虫采用了以下策略： | 策略 | 实现方式 | 配置项 | |------|----------|--------| | 随机延迟 | 每次请求后随机等待 1-3 秒 | `crawl_delay_min`, `crawl_delay_max` | | 频率控制 | 限制最大爬取数量 | `max_video_count` | | 登录态 | 使用真实登录Cookie | `login_type` | | 完整请求头 | User-Agent、Referer等 | `DEFAULT_HEADERS` | ### 爬虫实现代码 ``` /* Detailed source-code truncated for AI context efficiency. */ ``` ## 11.8 数据存储模块 数据存储模块负责将爬取到的数据持久化保存，支持多种存储格式。 ### 存储架构设计 采用**策略模式**设计，方便扩展新的存储方式： ```mermaid graph TB subgraph manager [存储管理器] sm["StorageManager
• save(data)
• load()
• filepath"] end subgraph base [抽象基类] bs["BaseStorage (ABC)
• save() 抽象方法
• load() 抽象方法"] end subgraph impl [具体实现] json["JSONStorage
• 保存为JSON文件
• 保持数据结构
• 便于程序处理"] csv["CSVStorage
• 保存为CSV文件
• 适合Excel打开
• 便于数据分析"] end sm -->|根据配置选择| json sm -->|根据配置选择| csv json -->|继承| bs csv -->|继承| bs ``` ### 存储格式对比 | 格式 | 优点 | 缺点 | 适用场景 | |------|------|------|----------| | JSON | 保持嵌套结构、程序易读取 | 文件较大、不便人工查看 | 后续程序处理、API接口 | | CSV | Excel可打开、便于分析 | 无法保存嵌套结构 | 数据分析、报表制作 | ### 存储实现代码 ``` /* Detailed source-code truncated for AI context efficiency. */ ``` ## 11.9 分析报告模块 分析报告模块负责对爬取的数据进行统计分析，生成可视化报告。 ### 分析功能概览 ```mermaid flowchart LR subgraph input [输入数据] videos["BilibiliVideo
对象列表"] end subgraph process [分析处理] stats["视频指标统计
• 播放量统计
• 点赞数统计
• 收藏数统计"] top["热门视频排名
TOP 10"] up["UP主分布统计
词频分析
(jieba分词)"] end subgraph output [输出结果] md["Markdown 报告
• 数据表格
• TOP排名
• UP主分布"] img["词云图片
(PNG)"] end videos --> stats --> md stats --> top --> img top --> up ``` ### 报告内容结构 生成的 Markdown 报告包含以下章节： | 章节 | 内容 | 分析维度 | |------|------|----------| | 视频指标统计 | 播放量、点赞、投币等指标的汇总统计 | 总计、平均、最高、最低 | | 热门视频 TOP 10 | 按播放量排序的前10个视频 | 标题、UP主、播放量、点赞 | | UP主分布 TOP 10 | 出现频率最高的UP主 | UP主名称、视频数量 | | 标题热词 TOP 20 | 视频标题中出现最多的词汇 | 词汇、出现频次 | | 标题词云 | 可视化展示热门词汇 | 词云图片 | ### 可选依赖说明 分析模块使用了可选依赖，即使没有安装也不会报错： ```mermaid graph TB subgraph deps [可选依赖] jieba["jieba
中文分词
(词频统计)"] wc["wordcloud
词云生成
(可视化)"] pd["pandas
数据处理
(可选)"] end subgraph fallback [未安装时的降级策略] note["如果未安装:
• jieba 未安装 - 跳过词频统计和词云生成
• wordcloud 未安装 - 跳过词云生成
• pandas 未安装 - 使用纯Python实现统计"] end jieba --> note wc --> note pd --> note ``` ### 分析实现代码 ``` /* Detailed source-code truncated for AI context efficiency. */ ``` ## 11.10 主程序入口 整合所有模块： ``` /* Detailed source-code truncated for AI context efficiency. */ ``` ## 11.11 运行与测试 ### 依赖安装 ```bash # 安装依赖 pip install playwright httpx pydantic pydantic-settings loguru # 安装 Playwright 浏览器 playwright install chromium # 可选：数据分析依赖 pip install pandas jieba wordcloud ``` ### 配置修改 修改 `config/settings.py` 中的配置： ```python # 爬取类型 crawler_type = CrawlerType.SEARCH # 或 CrawlerType.DETAIL # 搜索关键词 keywords = "Python教程,数据分析" # 最大爬取数量 max_video_count = 20 # 登录方式 login_type = LoginType.QRCODE # 首次使用扫码登录 ``` ### 运行程序 ```bash cd 源代码/爬虫进阶/11_进阶综合实战项目 python main.py ``` 首次运行会弹出浏览器窗口，使用 B站 APP 扫码登录后，程序会自动开始爬取。 ## 本章小结 本章我们完成了一个完整的 B站视频数据采集与分析项目，综合运用了： 1. **登录认证** - 扫码登录 - Cookie 登录 - 登录状态持久化 2. **API 签名** - WBI 签名算法 - 签名密钥获取 3. **数据爬取** - 关键词搜索 - 视频详情获取 - 并发控制 4. **数据处理** - Pydantic 数据模型 - JSON/CSV 存储 - 数据分析报告 5. **工程化设计** - 模块化架构 - 配置管理 - 日志系统 **关键要点：** - B站 API 需要 WBI 签名保护 - 登录状态可以持久化，避免重复扫码 - 注意控制爬取频率，避免触发限制 - 做好异常处理和日志记录 - 遵守 B站使用条款和法律法规 --- ## 教程总结 恭喜你完成了 Python 爬虫进阶教程的全部内容！在这 11 章的学习中，你掌握了： 1. **工程化开发**：日志、配置、异常处理 2. **反爬对抗**：请求伪装、代理 IP、浏览器指纹 3. **浏览器自动化**：Playwright 基础和进阶 4. **登录认证**：Cookie 管理、扫码登录 5. **验证码处理**：OCR 识别、滑块验证 6. **数据处理**：清洗、去重、分析、可视化 这些技术都源自 [MediaCrawler](https://github.com/NanmiCoder/MediaCrawler) 等实际项目的生产实践，希望能帮助你在爬虫开发领域更进一步。 **最后的建议：** - 持续关注反爬技术的演进 - 遵守法律法规和网站规则 - 参与开源项目，与社区共同成长 - 将爬虫作为数据获取的手段，关注数据本身的价值 祝你在数据采集的道路上越走越远！ --- --- METRICS --- - Files Extracted: 37 - Estimated Token Budget: ~49001 tokens - Canonical Reference: https://codewiki.google/github.com/NanmiCoder/CrawlerTutorial