什么是网络爬虫及其应用场景
网络爬虫(Web Crawler),也被称为网络蜘蛛(Web Spider),是一种按照一定规则自动浏览和抓取网络信息的程序。它模拟人类用户访问网页的行为,通过HTTP协议请求网页,然后解析返回的HTML内容,提取有价值的数据。
网络爬虫在现代互联网中有着广泛的应用场景:
- 搜索引擎:Google、Baidu等搜索引擎使用爬虫收集网页信息,建立索引数据库
- 价格监控:电商平台的价格监控系统,实时追踪竞争对手的价格变化
- 舆情分析:收集社交媒体、新闻网站的评论和报道,进行情感分析
- 学术研究:收集公开的学术论文、数据集用于研究分析
- 数据聚合:房产网站聚合多个平台的房源信息,为用户提供全面的房源数据
爬虫开发的法律和道德准则
在开发爬虫程序之前,必须了解并遵守以下准则:
1. 遵守robots.txt协议
robots.txt是网站根目录下的一个文本文件,用于告诉爬虫哪些页面可以访问,哪些不能访问。例如:
User-agent: *
Disallow: /admin/
Disallow: /private/
Allow: /public/
2. 尊重版权和隐私
- 不要抓取受版权保护的内容用于商业目的
- 不要抓取个人隐私信息
- 遵守GDPR等数据保护法规
3. 控制请求频率
避免对目标网站造成过大压力,应该:
- 设置合理的请求间隔(如1-3秒)
- 使用代理IP池轮换
- 遵循网站的API使用限制
Python爬虫开发环境准备
安装必要的Python库
# 安装requests库用于HTTP请求
pip install requests
# 安装BeautifulSoup4用于HTML解析
pip install beautifulsoup4
# 安装lxml作为解析器(可选,但速度更快)
pip install lxml
# 安装Selenium用于动态网页抓取
pip install selenium
# 安装pandas用于数据处理
pip install pandas
基础爬虫实战:抓取豆瓣电影Top250
1. 分析目标网站
豆瓣电影Top250页面结构分析:
- URL: https://movie.douban.com/top250
- 每页显示25部电影,共10页
- 需要提取:电影名称、评分、导演、主演、年份等信息
2. 编写基础爬虫代码
import requests
from bs4 import BeautifulSoup
import time
import random
import pandas as pd
class DoubanMovieCrawler:
def __init__(self):
self.headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'
}
self.base_url = 'https://movie.douban.com/top250'
self.movies = []
def get_page(self, url):
"""获取页面内容"""
try:
response = requests.get(url, headers=self.headers, timeout=10)
response.raise_for_status() # 检查请求是否成功
return response.text
except requests.RequestException as e:
print(f"请求失败: {e}")
return None
def parse_page(self, html):
"""解析页面内容"""
if not html:
return
soup = BeautifulSoup(html, 'lxml')
# 找到所有电影条目
items = soup.find_all('div', class_='item')
for item in items:
# 提取电影名称
title = item.find('span', class_='title').text.strip()
# 提取评分
rating = item.find('span', class_='rating_num').text.strip()
# 提取导演和主演信息
info = item.find('div', class_='bd').find('p', class_='').text.strip()
info_lines = info.split('\n')
director_actor = info_lines[0].strip()
year_country = info_lines[1].strip() if len(info_lines) > 1 else ""
# 提取评价人数
votes = item.find('div', class_='star').find_all('span')[-1].text[:-3]
# 提取短评
quote_tag = item.find('span', class_='inq')
quote = quote_tag.text if quote_tag else "暂无短评"
movie_data = {
'电影名称': title,
'评分': rating,
'导演/主演': director_actor,
'年份/国家': year_country,
'评价人数': votes,
'短评': quote
}
self.movies.append(movie_data)
def crawl(self, max_pages=10):
"""爬取多页数据"""
for page in range(max_pages):
url = f"{self.base_url}?start={page * 25}"
print(f"正在爬取第 {page + 1} 页: {url}")
html = self.get_page(url)
self.parse_page(html)
# 随机延迟1-3秒,避免请求过于频繁
time.sleep(random.uniform(1, 3))
print(f"已爬取 {len(self.movies)} 部电影")
def save_data(self, filename='douban_top250.csv'):
"""保存数据到CSV文件"""
if not self.movies:
print("没有数据可保存")
return
df = pd.DataFrame(self.movies)
df.to_csv(filename, index=False, encoding='utf-8-sig')
print(f"数据已保存到 {filename}")
def display_data(self):
"""展示爬取的数据"""
for i, movie in enumerate(self.movies[:5], 1):
print(f"\n{i}. {movie['电影名称']}")
print(f" 评分: {movie['评分']}")
print(f" 导演/主演: {20 * ' '}{movie['导演/主演']}")
print(f" 年份/国家: {20 * ' '}{movie['年份/国家']}")
print(f" 评价人数: {20 * ' '}{movie['评价人数']}人")
print(f" 短评: {20 * ' '}{movie['短评']}")
# 使用示例
if __name__ == '__main__':
crawler = DoubanMovieCrawler()
crawler.crawl(max_pages=3) # 先爬取3页测试
crawler.display_data()
crawler.save_data()
3. 代码详细说明
3.1 类结构设计
__init__:初始化请求头和基础URLget_page:负责HTTP请求,包含异常处理parse_page:使用BeautifulSoup解析HTML,提取数据crawl:主爬取逻辑,控制翻页save_data:使用pandas保存数据display_data:展示爬取结果
3.2 关键技术点
- 请求头设置:使用User-Agent模拟浏览器访问
- 异常处理:使用try-except捕获网络请求异常
- 随机延迟:使用random.uniform生成随机延迟时间
- HTML解析:使用CSS选择器定位元素
- 数据清洗:去除多余的空白字符和换行符
进阶爬虫:处理动态加载内容
1. 为什么需要Selenium
现代网站大量使用JavaScript动态加载内容,传统的requests库无法获取这些内容。Selenium可以模拟真实浏览器行为,执行JavaScript并获取完整页面内容。
2. Selenium基础使用
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.common.exceptions import TimeoutException
import time
class DynamicCrawler:
def __init__(self, headless=True):
"""初始化Chrome驱动"""
chrome_options = Options()
if headless:
chrome_options.add_argument('--headless') # 无头模式
chrome_options.add_argument('--no-sandbox')
chrome_options.add_argument('--disable-dev-shm-usage')
chrome_options.add_argument('--user-agent=Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36')
# 指定ChromeDriver路径(需要提前下载)
# service = Service('/path/to/chromedriver')
# self.driver = webdriver.Chrome(service=service, options=chrome_options)
# 或者使用webdriver-manager自动管理驱动
from webdriver_manager.chrome import ChromeDriverManager
self.driver = webdriver.Chrome(ChromeDriverManager().install(), options=chrome_options)
self.wait = WebDriverWait(self.driver, 10) # 显式等待
def scroll_to_bottom(self):
"""滚动到页面底部,触发懒加载"""
last_height = self.driver.execute_script("return document.body.scrollHeight")
while True:
self.driver.execute_script("window.scrollTo(0, document.body.scrollHeight);")
time.sleep(2)
new_height = self.driver.execute_script("return document.body.scrollHeight")
if new_height == last_height:
break
last_height = new_height
def crawl_dynamic_page(self, url):
"""爬取动态页面"""
try:
self.driver.get(url)
# 等待关键元素加载
self.wait.until(
EC.presence_of_element_located((By.CLASS_NAME, "movie-item"))
)
# 滚动加载所有内容
self.scroll_to_bottom()
# 获取页面源码
html = self.driver.page_source
# 使用BeautifulSoup解析
soup = BeautifulSoup(html, 'lxml')
# 提取数据...
movies = []
items = soup.find_all('div', class_='movie-item')
for item in items:
title = item.find('h3').text.strip()
movies.append(title)
return movies
except TimeoutException:
print("页面加载超时")
return []
finally:
self.driver.quit()
# 使用示例
# crawler = DynamicCrawler(headless=True)
# movies = crawler.crawl_dynamic_page('https://example-dynamic-site.com')
3. Selenium高级技巧
3.3.1 显式等待 vs 隐式等待
# 显式等待(推荐)- 等待特定条件
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.common.by import By
wait = WebDriverWait(driver, 10)
element = wait.until(EC.element_to_be_clickable((By.ID, "load-more-button")))
# 隐式等待 - 全局设置
driver.implicitly_wait(10)
3.3.2 处理弹窗和iframe
# 处理alert弹窗
try:
alert = driver.switch_to.alert
alert.accept() # 点击确定
# alert.dismiss() # 点击取消
except:
pass
# 处理iframe
iframe = driver.find_element(By.TAG_NAME, "iframe")
driver.switch_to.frame(iframe)
# 在iframe内操作
driver.switch_to.default_content() # 切换回主文档
3.3.3 模拟用户交互
from selenium.webdriver.common.action_chains import ActionChains
from selenium.webdriver.common.keys import Keys
# 模拟鼠标悬停
element = driver.find_element(By.ID, "menu")
ActionChains(driver).move_to_element(element).perform()
# 模拟键盘输入
input_box = driver.find_element(By.NAME, "search")
input_box.send_keys("Python爬虫")
input_box.send_keys(Keys.ENTER)
# 模拟拖拽
source = driver.find_element(By.ID, "drag-source")
target = driver.find_element(By.ID, "drop-target")
ActionChains(driver).drag_and_drop(source, target).perform()
反爬虫策略与应对方法
1. 常见反爬虫机制
1.1 请求头检测
网站会检查HTTP请求头,识别是否为爬虫。
# 应对:设置完整的请求头
headers = {
'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,*/*;q=0.8',
'Accept-Language': 'en-US,en;q=0.5',
'Accept-Encoding': 'gzip, deflate',
'Connection': 'keep-alive',
'Upgrade-Insecure-Requests': '1',
}
1.2 IP限制
网站会限制同一IP的访问频率。
# 应对:使用代理IP池
import random
proxy_pool = [
'http://123.45.67.89:8080',
'http://98.76.54.32:3128',
'http://111.222.333.44:8080',
]
def get_random_proxy():
return random.choice(proxy_pool)
proxies = {'http': get_random_proxy()}
response = requests.get(url, headers=headers, proxies=proxies)
1.3 Cookie和Session验证
# 应对:维护Session
session = requests.Session()
# 先访问一次获取cookie
session.get('https://example.com/login', headers=headers)
# 再带着cookie访问目标页面
response = session.get('https://example.com/data', headers=headers)
1.4 验证码识别
# 应对:使用OCR库或第三方打码平台
import pytesseract
from PIL import Image
def solve_captcha(image_path):
image = Image.open(image_path)
text = pytesseract.image_to_string(image)
return text
# 或者使用第三方服务(如超级鹰、云打码)
1.5 JavaScript混淆和加密
# 应对:使用PyExecJS执行JavaScript
import execjs
# 网页中有加密函数
js_code = """
function encrypt(data) {
return btoa(data);
}
"""
ctx = execjs.compile(js_code)
encrypted = ctx.call('encrypt', 'hello world')
print(encrypted) # aGVsbG8gd29ybGQ=
数据存储方案
1. CSV文件存储
import csv
def save_to_csv(data, filename):
with open(filename, 'w', newline='', encoding='utf-8') as f:
writer = csv.DictWriter(f, fieldnames=data[0].keys())
writer.writeheader()
writer.writerows(data)
2. JSON文件存储
import json
def save_to_json(data, filename):
with open(filename, 'w', encoding='utf-8') as f:
json.dump(data, f, ensure_ascii=False, indent=2)
3. SQLite数据库存储
import sqlite3
def save_to_sqlite(data, db_name='crawler.db'):
conn = sqlite3.connect(db_name)
cursor = conn.cursor()
# 创建表
cursor.execute('''
CREATE TABLE IF NOT EXISTS movies (
id INTEGER PRIMARY KEY AUTOINCREMENT,
title TEXT,
rating REAL,
director TEXT,
year TEXT,
votes INTEGER,
quote TEXT
)
''')
# 插入数据
for movie in data:
cursor.execute('''
INSERT INTO movies (title, rating, director, year, votes, quote)
VALUES (?, ?, ?, ?, ?, ?)
''', (
movie['电影名称'],
float(movie['评分']),
movie['导演/主演'].split('导演:')[0] if '导演:' in movie['导演/主演'] else movie['导演/主演'],
movie['年份/国家'].split('/')[0] if '/' in movie['年份/国家'] else movie['年份/国家'],
int(movie['评价人数']),
movie['短评']
))
conn.commit()
conn.close()
4. MongoDB存储
from pymongo import MongoClient
def save_to_mongodb(data, db_name='crawler', collection_name='movies'):
client = MongoClient('mongodb://localhost:27017/')
db = client[db_name]
collection = db[collection_name]
# 插入数据
for movie in data:
collection.insert_one(movie)
client.close()
完整项目:构建一个分布式爬虫系统
1. 系统架构设计
主控节点 (Master)
├── 任务分发器
├── URL去重器
└── 结果收集器
爬虫节点 (Worker)
├── 请求调度器
├── 页面下载器
├── 页面解析器
└── 数据存储器
消息队列 (Redis/RabbitMQ)
├── 待爬取URL队列
├── 爬取结果队列
└── 去重集合
2. 主控节点代码
import redis
import json
import hashlib
from urllib.parse import urlparse
class MasterNode:
def __init__(self, redis_host='localhost', redis_port=6379):
self.redis = redis.Redis(host=redis_host, port=redis_port, decode_responses=True)
self.url_queue_key = 'url_queue'
self.result_queue_key = 'result_queue'
self.seen_urls_key = 'seen_urls'
def add_urls(self, urls):
"""添加URL到任务队列"""
for url in urls:
# URL去重
url_hash = hashlib.md5(url.encode()).hexdigest()
if self.redis.sismember(self.seen_urls_key, url_hash):
continue
self.redis.sadd(self.seen_urls_key, url_hash)
self.redis.lpush(self.url_queue_key, url)
def get_result(self, timeout=1):
"""获取爬取结果"""
result = self.redis.brpop(self.result_queue_key, timeout=timeout)
if result:
return json.loads(result[1])
return None
def get_stats(self):
"""获取统计信息"""
url_count = self.redis.llen(self.url_queue_key)
result_count = self.redis.llen(self.result_queue_key)
seen_count = self.redis.scard(self.seen_urls_key)
return {
'pending_urls': url_count,
'completed_results': result_count,
'total_seen': seen_count
}
3. 爬虫节点代码
import requests
import json
import time
import redis
from bs4 import BeautifulSoup
class WorkerNode:
def __init__(self, worker_id, redis_host='localhost', redis_port=6379):
self.worker_id = worker_id
self.redis = redis.Redis(host=redis_host, port=redis_port, decode_responses=True)
self.url_queue_key = 'url_queue'
self.result_queue_key = 'result_queue'
self.headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
}
def crawl(self, url):
"""爬取单个URL"""
try:
response = requests.get(url, headers=self.headers, timeout=10)
response.raise_for_status()
soup = BeautifulSoup(response.text, 'lxml')
# 提取数据(示例)
title = soup.find('title')
data = {
'worker_id': self.worker_id,
'url': url,
'title': title.text if title else '',
'crawl_time': time.time(),
'status': 'success'
}
return data
except Exception as e:
return {
'worker_id': self.worker_id,
'url': url,
'error': str(e),
'status': 'failed'
}
def run(self):
"""工作循环"""
print(f"Worker {self.worker_id} started")
while True:
# 从队列获取URL
result = self.redis.brpop(self.url_queue_key, timeout=5)
if not result:
print(f"Worker {self.worker_id} queue empty, waiting...")
time.sleep(2)
continue
url = result[1]
print(f"Worker {self.worker_id} crawling: {url}")
# 爬取
data = self.crawl(url)
# 保存结果
self.redis.lpush(self.result_queue_key, json.dumps(data))
# 随机延迟
time.sleep(random.uniform(1, 3))
4. 运行示例
# 主控节点
master = MasterNode()
urls = [
'https://www.example.com/page1',
'https://www.example.com/page2',
'https://www.example.com/page3',
]
master.add_urls(urls)
# 爬虫节点(可以运行多个实例)
worker = WorkerNode(worker_id=1)
worker.run()
# 收集结果
while True:
result = master.get_result()
if result:
print(f"收到结果: {result}")
else:
break
爬虫性能优化
1. 异步爬虫(使用aiohttp)
import aiohttp
import asyncio
from bs4 import BeautifulSoup
class AsyncCrawler:
def __init__(self, max_concurrent=10):
self.max_concurrent = max_concurrent
self.semaphore = asyncio.Semaphore(max_concurrent)
async def fetch(self, session, url):
async with self.semaphore:
try:
async with session.get(url, timeout=10) as response:
return await response.text()
except Exception as e:
print(f"Error fetching {url}: {e}")
return None
async def parse(self, html):
if not html:
return []
soup = BeautifulSoup(html, 'lxml')
# 解析逻辑...
return []
async def crawl(self, urls):
async with aiohttp.ClientSession() as session:
tasks = [self.fetch(session, url) for url in urls]
htmls = await asyncio.gather(*tasks)
# 解析所有页面
results = []
for html in htmls:
data = await self.parse(html)
results.extend(data)
return results
# 使用示例
async def main():
crawler = AsyncCrawler(max_concurrent=20)
urls = [f'https://example.com/page{i}' for i in range(100)]
results = await crawler.crawl(urls)
print(f"爬取完成,共 {len(results)} 条数据")
# asyncio.run(main())
2. 使用Scrapy框架
Scrapy是Python最强大的爬虫框架,提供了完整的解决方案。
# spiders/movie_spider.py
import scrapy
from scrapy.crawler import CrawlerProcess
from scrapy.utils.project import get_project_settings
class MovieSpider(scrapy.Spider):
name = 'movie'
allowed_domains = ['douban.com']
start_urls = ['https://movie.douban.com/top250']
custom_settings = {
'DOWNLOAD_DELAY': 2,
'CONCURRENT_REQUESTS': 10,
'FEED_FORMAT': 'csv',
'FEED_URI': 'movies.csv',
}
def parse(self, response):
for item in response.css('div.item'):
yield {
'title': item.css('span.title::text').get(),
'rating': item.css('span.rating_num::text').get(),
'quote': item.css('span.inq::text').get(default=''),
}
# 翻页
next_page = response.css('a.next::attr(href)').get()
if next_page:
yield response.follow(next_page, self.parse)
# 运行爬虫
if __name__ == '__main__':
process = CrawlerProcess(get_project_settings())
process.crawl(MovieSpider)
process.start()
爬虫监控与日志
1. 日志配置
import logging
from logging.handlers import RotatingFileHandler
def setup_logging():
"""配置日志"""
logger = logging.getLogger('crawler')
logger.setLevel(logging.INFO)
# 文件处理器
file_handler = RotatingFileHandler(
'crawler.log', maxBytes=10*1024*1024, backupCount=5
)
file_handler.setLevel(logging.INFO)
# 控制台处理器
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO)
# 格式化器
formatter = logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
file_handler.setFormatter(formatter)
console_handler.setFormatter(formatter)
logger.addHandler(file_handler)
logger.addHandler(console_handler)
return logger
# 使用
logger = setup_logging()
logger.info("爬虫启动")
logger.warning("请求失败,重试中...")
logger.error("爬取失败: %s", error_msg)
2. 监控指标
class CrawlerMonitor:
def __init__(self):
self.stats = {
'total_requests': 0,
'successful_requests': 0,
'failed_requests': 0,
'total_time': 0,
'data_size': 0,
}
self.start_time = time.time()
def record_request(self, success, duration, data_size=0):
self.stats['total_requests'] += 1
if success:
self.stats['successful_requests'] += 1
else:
self.stats['failed_requests'] += 1
self.stats['total_time'] += duration
self.stats['data_size'] += data_size
def get_report(self):
elapsed = time.time() - self.start_time
return {
'运行时间': f"{elapsed:.2f}秒",
'总请求数': self.stats['total_requests'],
'成功数': self.stats['successful_requests'],
'失败数': self.stats['failed_requests'],
'成功率': f"{(self.stats['successful_requests'] / self.stats['total_requests'] * 100):.2f}%" if self.stats['total_requests'] > 0 else "0%",
'平均速度': f"{self.stats['total_requests'] / elapsed:.2f} 请求/秒",
'数据量': f"{self.stats['data_size'] / 1024:.2f} KB",
}
法律风险与合规建议
1. 必须遵守的法律条款
- 《中华人民共和国网络安全法》:禁止非法侵入他人网络、干扰网络正常功能
- 《数据安全法》:数据处理活动必须合法合规
- 《个人信息保护法》:不得非法收集个人信息
- 《反不正当竞争法》:禁止利用技术手段进行不正当竞争
2. 合规建议
- 获取授权:优先使用网站提供的官方API
- 遵守robots.txt:严格遵守爬虫协议
- 控制频率:设置合理的请求间隔,避免对服务器造成压力
- 数据使用:仅用于合法目的,不侵犯他人权益
- 用户代理:在爬虫中提供真实的联系方式
3. 风险规避清单
- [ ] 检查目标网站的robots.txt
- [ ] 阅读网站的使用条款
- [ ] 设置合理的User-Agent和请求间隔
- [ ] 不抓取个人隐私信息
- [ ] 不进行DDoS攻击式爬取
- [ ] 保留爬虫日志以备查验
- [ ] 咨询法律专业人士(商业用途)
常见问题与解决方案
Q1: 爬虫被封IP怎么办?
A:
- 使用代理IP池(如阿布云、蘑菇代理)
- 降低请求频率
- 使用Tor网络
- 联系网站管理员申请白名单
Q2: 如何处理验证码?
A:
- 使用OCR库(如pytesseract)
- 接入打码平台(如超级鹰)
- 人工介入处理
- 优化爬虫行为避免触发验证码
Q3: 动态内容无法抓取?
A:
- 使用Selenium或Playwright
- 分析API接口,直接调用数据接口
- 使用Splash等渲染服务
- 分析JavaScript代码,模拟执行
Q4: 数据量太大,内存不足?
A:
- 使用生成器逐条处理
- 分批写入文件
- 使用数据库存储
- 增加内存或优化数据结构
总结
网络爬虫是一项强大的技术,但也需要谨慎使用。本文从基础到进阶,详细介绍了Python爬虫的开发流程、技术要点和最佳实践。记住以下关键点:
- 合法合规:始终遵守法律法规和网站规则
- 技术选型:根据需求选择合适的工具(requests/Selenium/Scrapy)
- 性能优化:使用异步、分布式等技术提升效率
- 数据安全:妥善存储和使用爬取的数据
- 持续学习:爬虫技术不断发展,需要持续学习新技术
通过本文的示例代码和详细说明,你应该能够独立开发出功能完善、性能优良的爬虫程序。祝你爬虫开发顺利!
