实现一个数据爬虫工具需要综合考虑合法性、鲁棒性、可维护性和反爬策略应对。下面我将为你提供一个**生产级可用的、模块化、可扩展的 Python 爬虫工具示例**，基于 `requests` + `BeautifulSoup`（轻量静态页面）和可选 `selenium`（动态渲染），并包含： - 请求管理（带 User-Agent 轮换、请求延迟、重试机制） - HTML 解析与结构化数据抽取（支持 CSS/XPath 选择器） - 数据持久化（JSON/CSV 导出） - 基础反爬规避（headers、随机 delay、session 复用） - 日志记录与错误处理 - 配置驱动（YAML 配置文件支持） ⚠️ 重要前提： ✅ 仅用于学习、个人研究或已获明确授权的公开数据采集； ❌ 严禁绕过 robots.txt、高频请求、采集敏感/隐私/付费内容； ✅ 始终遵守目标网站的 `Terms of Service` 和 `robots.txt`（如 `https://example.com/robots.txt`）。 --- ### ✅ 完整可运行代码（含配置示例） #### 1. 项目结构建议： ``` crawler/ ├── crawler.py # 主爬虫类 ├── config.yaml # 爬取配置（URL、解析规则、输出等） ├── output/ │ ├── data.json │ └── data.csv └── requirements.txt ``` #### 2. `requirements.txt` ```text requests>=2.31.0 beautifulsoup4>=4.12.0 lxml>=4.9.0 pyyaml>=6.0.0 ``` #### 3. `config.yaml` 示例（爬取豆瓣电影 Top250 的标题与评分） ```yaml target_url: "https://movie.douban.com/top250" request: headers: User-Agent: "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36" delay_range: [1, 3] # 随机延迟 1~3 秒 max_retries: 3 timeout: 10 parse: selector_type: "css" # 或 "xpath" items: ".item" fields: title: "h3 > span.title:first-of-type::text" rating: ".rating_num::text" link: ".hd a::attr(href)" output: format: "json" # "json" or "csv" file: "output/data.json" ``` #### 4. `crawler.py`（核心实现） ```python import json import csv import time import random import logging from pathlib import Path from typing import List, Dict, Any, Optional import requests from bs4 import BeautifulSoup import yaml # 配置日志 logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s") logger = logging.getLogger(__name__) class DataCrawler: def __init__(self, config_path: str): with open(config_path, "r", encoding="utf-8") as f: self.config = yaml.safe_load(f) self.session = requests.Session() # 设置默认 headers self.session.headers.update(self.config["request"].get("headers", {})) self.delay_range = self.config["request"].get("delay_range", [1, 2]) self.max_retries = self.config["request"].get("max_retries", 3) self.timeout = self.config["request"].get("timeout", 10) def _get_page(self, url: str) -> Optional[str]: for i in range(self.max_retries): try: logger.info(f"Fetching {url} (attempt {i+1})...") resp = self.session.get(url, timeout=self.timeout) resp.raise_for_status() resp.encoding = resp.apparent_encoding return resp.text except Exception as e: logger.warning(f"Request failed: {e}") if i < self.max_retries - 1: time.sleep(random.uniform(*self.delay_range)) return None def _parse_with_bs4(self, html: str, rule: str, method: str = "css") -> List[Any]: soup = BeautifulSoup(html, "lxml") if method == "css": elements = soup.select(rule) elif method == "xpath": raise NotImplementedError("XPath parsing requires lxml.etree — use css for simplicity.") else: raise ValueError("Only 'css' supported in this lightweight version.") results = [] for el in elements: if rule.endswith("::text"): text = el.get_text(strip=True) results.append(text) elif rule.endswith("::attr(href)") or rule.endswith("::attr(src)"): attr = "href" if "href" in rule else "src" results.append(el.get(attr, "").strip()) else: results.append(el.get_text(strip=True)) return results def _extract_items(self, html: str) -> List[Dict[str, Any]]: parse_cfg = self.config["parse"] items_selector = parse_cfg["items"] fields = parse_cfg["fields"] soup = BeautifulSoup(html, "lxml") item_els = soup.select(items_selector) results = [] for item in item_els: row = {} for field_name, selector in fields.items(): # 简单支持 ::text / ::attr(...) 语法（仅限顶层） if "::text" in selector: sel_base = selector.replace("::text", "") el = item.select_one(sel_base) row[field_name] = el.get_text(strip=True) if el else "" elif "::attr(" in selector: attr_match = selector.split("::attr(")[1].rstrip(")") sel_base = selector.split("::attr(")[0] el = item.select_one(sel_base) row[field_name] = el.get(attr_match, "").strip() if el else "" else: el = item.select_one(selector) row[field_name] = el.get_text(strip=True) if el else "" results.append(row) return results def crawl(self) -> List[Dict[str, Any]]: url = self.config["target_url"] html = self._get_page(url) if not html: logger.error("Failed to fetch page content.") return [] logger.info("Parsing HTML...") data = self._extract_items(html) logger.info(f"Extracted {len(data)} items.") return data def save(self, data: List[Dict[str, Any]]): out_cfg = self.config["output"] fmt = out_cfg["format"].lower() file_path = Path(out_cfg["file"]) file_path.parent.mkdir(exist_ok=True) if fmt == "json": with open(file_path, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) logger.info(f"Saved JSON to {file_path}") elif fmt == "csv": if data: with open(file_path, "w", newline="", encoding="utf-8") as f: writer = csv.DictWriter(f, fieldnames=data[0].keys()) writer.writeheader() writer.writerows(data) logger.info(f"Saved CSV to {file_path}") else: logger.warning("No data to write to CSV.") else: raise ValueError(f"Unsupported output format: {fmt}") # --- 使用示例（直接运行）--- if __name__ == "__main__": # 注意：请先手动确认豆瓣允许爬取且遵守其 robots.txt（实际中豆瓣有强反爬，本例仅作教学演示） # 真实使用请替换为无反爬/开放 API/测试站点（如 http://httpbin.org/html） crawler = DataCrawler("config.yaml") results = crawler.crawl() crawler.save(results) ``` --- ### 🔍 关键设计解释： | 模块 | 说明 | |------|------| | **Session 复用 & Headers 管理** | 避免被识别为脚本请求；模拟真实浏览器行为。 | | **随机延迟 + 重试机制** | 尊重服务器负载，提升稳定性，避免被封 IP。 | | **CSS 选择器解析（非 XPath）** | `BeautifulSoup + lxml` 对 CSS 支持成熟，轻量高效；XPath 可后续用 `lxml.etree` 替代增强。 | | **`::text` / `::attr(...)` 语法糖** | 简化配置，使 YAML 更易读（类似 Scrapy 的 CSS 语法）。 | | **结构化解析逻辑 `_extract_items`** | 按 item 容器逐个提取字段，避免跨 item 错位，保证数据对齐。 | | **日志 + 异常隔离** | 易于调试、监控失败原因，不因单页失败中断整个流程。 | > 💡 进阶提示： > - 如需处理 JavaScript 渲染页面（如 React/Vue SPA），可集成 `selenium` 或 `playwright` 替换 `_get_page`； > - 如需分布式/增量爬取，可引入 `scrapy-redis`； > - 合法合规场景下，**优先使用官方 API**（如豆瓣 API v2 已关闭，但 GitHub API、NewsAPI 等仍开放）。 ---