Python网络爬虫入门：从原理到实践
网络爬虫（Web Spider）是自动获取互联网数据的程序。从搜索引擎到价格监控，爬虫技术无处不在。本教程将系统介绍Python爬虫的原理和实践方法，帮助你掌握数据抓取的核心技能。
一、HTTP协议基础
理解HTTP协议是学习爬虫的前提。HTTP（超文本传输协议）是客户端和服务器之间通信的规则。
HTTP请求与响应
# http_basics.py - HTTP协议演示
"""
HTTP请求的基本组成：
1. 请求方法：GET（获取数据）、POST（提交数据）、PUT、DELETE等
2. 请求URL：目标资源的地址
3. 请求头（Headers）：包含User-Agent、Cookie、Referer等信息
4. 请求体（Body）：POST请求时携带的数据
HTTP响应的基本组成：
1. 状态码：200成功、404未找到、500服务器错误等
2. 响应头：包含内容类型、编码等信息
3. 响应体：实际的数据内容（HTML、JSON、图片等）
"""
# 常见HTTP状态码
status_codes = {
200: "OK - 请求成功",
301: "Moved Permanently - 永久重定向",
302: "Found - 临时重定向",
304: "Not Modified - 资源未修改（使用缓存）",
400: "Bad Request - 请求参数错误",
401: "Unauthorized - 未授权",
403: "Forbidden - 禁止访问",
404: "Not Found - 资源不存在",
429: "Too Many Requests - 请求过于频繁",
500: "Internal Server Error - 服务器内部错误",
503: "Service Unavailable - 服务不可用"
}
for code, desc in status_codes.items():
print(f"{code}: {desc}")二、requests库入门
requests是Python中最流行的HTTP库，API设计优雅，使用简单。
# 安装requests
pip install requests# requests_basics.py - requests库基础用法
import requests
# ===== GET请求 =====
# 最基本的GET请求
response = requests.get('https://httpbin.org/get')
print(f"状态码: {response.status_code}")
print(f"响应内容: {response.text[:200]}")
# 带查询参数的GET请求
params = {
'q': 'Python',
'page': 1,
'limit': 10
}
response = requests.get('https://httpbin.org/get', params=params)
print(f"请求URL: {response.url}")
# 带请求头的GET请求
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
'Accept': 'text/html,application/xhtml+xml',
'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8'
}
response = requests.get('https://httpbin.org/get', headers=headers)
# ===== POST请求 =====
# 表单数据提交
data = {
'username': 'admin',
'password': '123456'
}
response = requests.post('https://httpbin.org/post', data=data)
# JSON数据提交
json_data = {
'title': '爬虫教程',
'content': 'Python网络爬虫入门'
}
response = requests.post('https://httpbin.org/post', json=json_data)
# ===== Session会话管理 =====
# Session可以保持Cookie，适合需要登录的场景
session = requests.Session()
# 设置全局请求头
session.headers.update({
'User-Agent': 'Mozilla/5.0 (My-Crawler/1.0)',
'Accept-Language': 'zh-CN,zh;q=0.9'
})
# 使用Session发送请求（自动管理Cookie）
response = session.get('https://httpbin.org/cookies/set?name=value')
response = session.get('https://httpbin.org/cookies') # 自动携带上次的Cookie
print(f"Cookies: {response.json()}")
# ===== 超时设置 =====
try:
# 连接超时5秒，读取超时10秒
response = requests.get('https://httpbin.org/delay/3',
timeout=(5, 10))
print("请求成功")
except requests.exceptions.Timeout:
print("请求超时")
except requests.exceptions.ConnectionError:
print("连接失败")
except requests.exceptions.RequestException as e:
print(f"请求异常: {e}")三、HTML解析（BeautifulSoup）
获取到网页HTML后，需要从中提取数据。BeautifulSoup是最流行的HTML解析库。
# 安装BeautifulSoup和解析器
pip install beautifulsoup4 lxml# beautifulsoup_basics.py - BeautifulSoup基础
from bs4 import BeautifulSoup
# 示例HTML
html_doc = """
<html>
<head><title>示例网页</title></head>
<body>
<div class="container">
<h1 id="main-title">欢迎来到爬虫教程</h1>
<p class="intro">这是一段介绍文字</p>
<ul class="article-list">
<li class="item">
<a href="/article/1" class="link">第一篇文章</a>
<span class="date">2024-01-15</span>
</li>
<li class="item">
<a href="/article/2" class="link">第二篇文章</a>
<span class="date">2024-01-16</span>
</li>
<li class="item featured">
<a href="/article/3" class="link">第三篇文章</a>
<span class="date">2024-01-17</span>
</li>
</ul>
<table id="data-table">
<tr><th>姓名</th><th>年龄</th></tr>
<tr><td>张三</td><td>25</td></tr>
<tr><td>李四</td><td>30</td></tr>
</table>
</div>
</body>
</html>
"""
# 创建BeautifulSoup对象
soup = BeautifulSoup(html_doc, 'lxml')
# ===== 基本选择方法 =====
# 获取标题
print("标题:", soup.title.string)
# 获取h1标签（通过id）
h1 = soup.find('h1', id='main-title')
print("H1:", h1.string)
# 获取第一个p标签
p = soup.find('p', class_='intro')
print("P:", p.string)
# 获取所有class为item的li标签
items = soup.find_all('li', class_='item')
print(f"\n找到 {len(items)} 个列表项:")
for item in items:
link = item.find('a')
date = item.find('span', class_='date')
print(f" 标题: {link.string}, 链接: {link['href']}, 日期: {date.string}")
# 获取同时有item和featured两个class的标签
featured = soup.find('li', class_='featured')
print(f"\n特色文章: {featured.find('a').string}")
# ===== 解析表格 =====
table = soup.find('table', id='data-table')
rows = table.find_all('tr')
print("\n表格数据:")
for row in rows:
cells = row.find_all(['th', 'td'])
print(" | ".join(cell.string for cell in cells))四、CSS选择器提取数据
CSS选择器是提取网页数据的强大工具，BeautifulSoup完美支持。
# css_selectors.py - CSS选择器详解
from bs4 import BeautifulSoup
html_doc = """
<div id="content">
<article class="post" data-id="1">
<h2 class="title"><a href="/p/1">文章1</a></h2>
<p class="excerpt">文章摘要1</p>
<span class="author">作者A</span>
<span class="views">100</span>
</article>
<article class="post" data-id="2">
<h2 class="title"><a href="/p/2">文章2</a></h2>
<p class="excerpt">文章摘要2</p>
<span class="author">作者B</span>
<span class="views">200</span>
</article>
<article class="post featured" data-id="3">
<h2 class="title"><a href="/p/3">文章3</a></h2>
<p class="excerpt">文章摘要3</p>
<span class="author">作者C</span>
<span class="views">300</span>
</article>
</div>
"""
soup = BeautifulSoup(html_doc, 'lxml')
# ===== CSS选择器示例 =====
# 1. 标签选择器
articles = soup.select('article')
print(f"文章总数: {len(articles)}")
# 2. 类选择器
titles = soup.select('.title')
for title in titles:
print(f"标题: {title.a.string}")
# 3. ID选择器
content = soup.select_one('#content')
print(f"内容区存在: {content is not None}")
# 4. 后代选择器（空格）
links = soup.select('article .title a')
for link in links:
print(f"链接: {link['href']} - {link.string}")
# 5. 子元素选择器（>）
direct_h2 = soup.select('article > h2')
print(f"直接子元素h2: {len(direct_h2)}")
# 6. 属性选择器
featured = soup.select('article.featured')
print(f"特色文章数: {len(featured)}")
# 按data-id属性选择
article_2 = soup.select_one('article[data-id="2"]')
print(f"data-id=2的文章: {article_2.h2.a.string}")
# 7. 组合选择器
authors = soup.select('article.post .author')
for author in authors:
print(f"作者: {author.string}")
# 8. 选择第一个/最后一个
first_post = soup.select_one('article.post')
last_post = soup.select('article.post')[-1]
print(f"第一篇: {first_post.h2.a.string}")
print(f"最后一篇: {last_post.h2.a.string}")
# 9. 提取完整数据
print("\n=== 所有文章信息 ===")
for article in soup.select('article.post'):
data = {
'id': article.get('data-id'),
'title': article.select_one('.title a').string,
'url': article.select_one('.title a')['href'],
'excerpt': article.select_one('.excerpt').string,
'author': article.select_one('.author').string,
'views': article.select_one('.views').string,
'is_featured': 'featured' in article.get('class', [])
}
print(data)五、分页爬取
大多数网站的数据分布在多个页面，需要实现分页爬取。
# pagination.py - 分页爬取示例
import requests
from bs4 import BeautifulSoup
import time
class PaginationCrawler:
"""分页爬取器"""
def __init__(self):
self.session = requests.Session()
self.session.headers.update({
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) '
'AppleWebKit/537.36 (KHTML, like Gecko) '
'Chrome/120.0.0.0 Safari/537.36'
})
self.base_url = "https://books.toscrape.com/"
def get_page_url(self, page_num):
"""构造分页URL"""
if page_num == 1:
return self.base_url + "catalogue/page-1.html"
return self.base_url + f"catalogue/page-{page_num}.html"
def parse_page(self, html):
"""解析单页数据"""
soup = BeautifulSoup(html, 'lxml')
books = []
# 找到所有书籍的article标签
articles = soup.select('article.product_pod')
for article in articles:
book = {
'title': article.select_one('h3 a')['title'],
'price': article.select_one('.price_color').text,
'availability': article.select_one('.availability').text.strip(),
'rating': article.select_one('.star-rating')['class'][1],
'url': self.base_url + 'catalogue/' +
article.select_one('h3 a')['href'].replace('../../../', '')
}
books.append(book)
return books
def get_total_pages(self, html):
"""获取总页数"""
soup = BeautifulSoup(html, 'lxml')
pager = soup.select_one('.pager')
if pager:
last_page = pager.select('li')[-1].text.strip()
if last_page.isdigit():
return int(last_page)
return 1
def crawl(self, max_pages=None, delay=1):
"""
爬取多页数据
Args:
max_pages: 最大爬取页数，None表示爬取全部
delay: 每页之间的延时（秒）
"""
all_books = []
# 获取第一页以确定总页数
first_url = self.get_page_url(1)
response = self.session.get(first_url)
if response.status_code != 200:
print("获取第一页失败")
return all_books
total_pages = self.get_total_pages(response.text)
if max_pages:
total_pages = min(total_pages, max_pages)
print(f"总共 {total_pages} 页")
# 爬取每一页
for page in range(1, total_pages + 1):
print(f"正在爬取第 {page}/{total_pages} 页...", end=" ")
url = self.get_page_url(page)
try:
response = self.session.get(url, timeout=10)
if response.status_code == 200:
books = self.parse_page(response.text)
all_books.extend(books)
print(f"获取 {len(books)} 本书")
else:
print(f"状态码: {response.status_code}")
except Exception as e:
print(f"错误: {e}")
# 礼貌性延时
if page < total_pages:
time.sleep(delay)
return all_books
# 使用示例
if __name__ == "__main__":
crawler = PaginationCrawler()
# 只爬取前3页作为演示
books = crawler.crawl(max_pages=3, delay=1)
print(f"\n总共获取 {len(books)} 本书")
print("\n前5本书的信息:")
for book in books[:5]:
print(f" 《{book['title'][:30]}》 - {book['price']} "
f"({book['rating']}星)")六、数据存储（CSV/JSON）
爬取的数据需要持久化存储。CSV和JSON是最常用的两种格式。
# data_storage.py - 数据存储
import csv
import json
import os
class DataStorage:
"""数据存储工具类"""
@staticmethod
def save_to_csv(data, filename, encoding='utf-8-sig'):
"""
保存数据到CSV文件
Args:
data: 字典列表
filename: 文件名
encoding: 编码，utf-8-sig可解决Excel中文乱码
"""
if not data:
print("没有数据可保存")
return
# 获取所有字段名
fieldnames = list(data[0].keys())
with open(filename, 'w', newline='', encoding=encoding) as f:
writer = csv.DictWriter(f, fieldnames=fieldnames)
writer.writeheader()
writer.writerows(data)
print(f"已保存 {len(data)} 条记录到 {filename}")
@staticmethod
def save_to_json(data, filename, encoding='utf-8'):
"""
保存数据到JSON文件
Args:
data: 要保存的数据
filename: 文件名
encoding: 编码
"""
with open(filename, 'w', encoding=encoding) as f:
json.dump(data, f, ensure_ascii=False, indent=2)
print(f"已保存数据到 {filename}")
@staticmethod
def load_from_json(filename, encoding='utf-8'):
"""从JSON文件加载数据"""
with open(filename, 'r', encoding=encoding) as f:
return json.load(f)
@staticmethod
def append_to_csv(data, filename, encoding='utf-8-sig'):
"""追加数据到CSV文件（增量爬取时使用）"""
if not data:
return
fieldnames = list(data[0].keys())
file_exists = os.path.exists(filename)
with open(filename, 'a', newline='', encoding=encoding) as f:
writer = csv.DictWriter(f, fieldnames=fieldnames)
if not file_exists:
writer.writeheader()
writer.writerows(data)
print(f"已追加 {len(data)} 条记录到 {filename}")
# 使用示例
if __name__ == "__main__":
# 示例数据
books = [
{"title": "Python编程", "price": "¥59.00",
"author": "张三", "rating": "Five"},
{"title": "数据分析", "price": "¥69.00",
"author": "李四", "rating": "Four"},
{"title": "机器学习", "price": "¥89.00",
"author": "王五", "rating": "Five"},
]
storage = DataStorage()
# 保存为CSV
storage.save_to_csv(books, 'books.csv')
# 保存为JSON
storage.save_to_json(books, 'books.json')
# 从JSON读取
loaded = storage.load_from_json('books.json')
print(f"\n从JSON加载了 {len(loaded)} 条记录")七、反爬应对策略
网站为了保护数据，会采取各种反爬措施。以下是常见的应对策略。
# anti_crawl.py - 反爬应对策略
import requests
import time
import random
from fake_useragent import UserAgent
# ===== 1. 设置请求头 =====
# 很多网站会检查User-Agent，拒绝非浏览器请求
# 方法1：手动设置常见浏览器UA
USER_AGENTS = [
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:121.0) Gecko/20100101 Firefox/121.0',
'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',
'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
]
def get_random_headers():
"""获取随机请求头"""
return {
'User-Agent': random.choice(USER_AGENTS),
'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',
'Accept-Encoding': 'gzip, deflate, br',
'Connection': 'keep-alive',
'Referer': 'https://www.google.com/',
}
# 方法2：使用fake_useragent库（需要安装 pip install fake-useragent）
def get_fake_ua():
try:
ua = UserAgent()
return ua.random
except:
return random.choice(USER_AGENTS)
# ===== 2. 请求延时 =====
# 避免请求过于频繁
def polite_delay(min_delay=1, max_delay=3):
"""随机延时，模拟人类行为"""
delay = random.uniform(min_delay, max_delay)
time.sleep(delay)
return delay
# ===== 3. 使用代理IP =====
class ProxyManager:
"""代理IP管理器"""
def __init__(self, proxy_list=None):
# 示例代理格式：http://ip:port 或 http://user:pass@ip:port
self.proxies = proxy_list or []
self.current_index = 0
def get_proxy(self):
"""轮询获取代理"""
if not self.proxies:
return None
proxy = self.proxies[self.current_index]
self.current_index = (self.current_index + 1) % len(self.proxies)
return {'http': proxy, 'https': proxy}
def test_proxy(self, proxy, test_url='https://httpbin.org/ip'):
"""测试代理是否可用"""
try:
proxies = {'http': proxy, 'https': proxy}
response = requests.get(test_url, proxies=proxies, timeout=5)
if response.status_code == 200:
return True
except:
pass
return False
# ===== 4. Cookie管理 =====
class CookieManager:
"""Cookie管理器"""
def __init__(self):
self.session = requests.Session()
def login_and_get_cookies(self, login_url, credentials):
"""模拟登录获取Cookie"""
response = self.session.post(login_url, data=credentials)
if response.status_code == 200:
print("登录成功，Cookie已保存")
return self.session.cookies
else:
print("登录失败")
return None
# ===== 5. 综合反爬策略示例 =====
class SmartCrawler:
"""综合反爬策略爬虫"""
def __init__(self, max_retries=3):
self.session = requests.Session()
self.max_retries = max_retries
self.proxy_manager = ProxyManager()
def request_with_retry(self, url, method='GET', **kwargs):
"""
带重试机制的请求
Args:
url: 请求URL
method: 请求方法
**kwargs: 传递给requests的额外参数
"""
for attempt in range(self.max_retries):
try:
# 设置随机请求头
kwargs.setdefault('headers', get_random_headers())
# 设置超时
kwargs.setdefault('timeout', 15)
# 尝试使用代理
proxy = self.proxy_manager.get_proxy()
if proxy:
kwargs.setdefault('proxies', proxy)
# 发送请求
if method.upper() == 'GET':
response = self.session.get(url, **kwargs)
else:
response = self.session.post(url, **kwargs)
# 检查是否被限流
if response.status_code == 429:
wait_time = (attempt + 1) * 5
print(f"被限流，等待 {wait_time} 秒后重试...")
time.sleep(wait_time)
continue
response.raise_for_status()
return response
except requests.exceptions.RequestException as e:
print(f"第 {attempt + 1} 次尝试失败: {e}")
if attempt < self.max_retries - 1:
wait = (attempt + 1) * 2
print(f"等待 {wait} 秒后重试...")
time.sleep(wait)
print(f"已达到最大重试次数 {self.max_retries}")
return None
def crawl(self, url):
"""爬取页面"""
# 礼貌延时
delay = polite_delay(1, 2)
print(f"延时 {delay:.1f} 秒")
response = self.request_with_retry(url)
if response:
return response.text
return None八、robots.txt规范
robots.txt是网站告知爬虫哪些页面可以抓取、哪些不可以的规范。虽然不是强制性的，但遵守robots.txt是爬虫的基本道德。
# robots_check.py - 检查robots.txt
import requests
from urllib.robotparser import RobotFileParser
def check_robots_txt(url):
"""
检查URL是否允许爬取
Args:
url: 要检查的URL
"""
# 解析域名
from urllib.parse import urlparse
parsed = urlparse(url)
robots_url = f"{parsed.scheme}://{parsed.netloc}/robots.txt"
# 获取robots.txt
rp = RobotFileParser()
rp.set_url(robots_url)
try:
rp.read()
except Exception as e:
print(f"无法读取robots.txt: {e}")
return True # 如果无法读取，默认允许
# 检查是否允许爬取
user_agent = '*'
can_fetch = rp.can_fetch(user_agent, url)
print(f"网站: {parsed.netloc}")
print(f"robots.txt: {robots_url}")
print(f"URL: {url}")
print(f"允许爬取: {'是' if can_fetch else '否'}")
# 获取爬取延时建议
crawl_delay = rp.crawl_delay(user_agent)
if crawl_delay:
print(f"建议延时: {crawl_delay} 秒")
return can_fetch
# 使用示例
if __name__ == "__main__":
# 检查books.toscrape.com是否允许爬取
url = "https://books.toscrape.com/catalogue/page-1.html"
check_robots_txt(url)
print()
# 检查另一个网站
url2 = "https://httpbin.org/get"
check_robots_txt(url2)九、实战案例：爬取图书信息
综合运用以上知识，我们来爬取books.toscrape.com（一个专门用于爬虫练习的网站）的图书信息。
# book_crawler.py - 完整的图书爬虫
import requests
from bs4 import BeautifulSoup
import csv
import json
import time
import random
from urllib.parse import urljoin
class BookCrawler:
"""
图书信息爬虫
目标网站：https://books.toscrape.com
"""
BASE_URL = "https://books.toscrape.com/"
# 星级评分映射
RATING_MAP = {
'One': 1, 'Two': 2, 'Three': 3,
'Four': 4, 'Five': 5
}
def __init__(self, delay_range=(1, 2)):
self.session = requests.Session()
self.delay_range = delay_range
# 设置请求头
self.session.headers.update({
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) '
'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': 'en-US,en;q=0.9',
})
self.all_books = []
def _polite_delay(self):
"""礼貌延时"""
delay = random.uniform(*self.delay_range)
time.sleep(delay)
def _get_page(self, url):
"""获取页面内容"""
try:
response = self.session.get(url, timeout=15)
response.raise_for_status()
return response.text
except requests.exceptions.RequestException as e:
print(f"获取页面失败: {e}")
return None
def parse_book_detail(self, url):
"""解析图书详情页"""
html = self._get_page(url)
if not html:
return None
soup = BeautifulSoup(html, 'lxml')
# 提取详情信息
details = {}
# 产品描述
description = soup.select_one('#product_description ~ p')
details['description'] = description.string if description else ''
# 产品信息表格
table = soup.select_one('.table table-striped')
if table:
for row in table.select('tr'):
th = row.select_one('th')
td = row.select_one('td')
if th and td:
details[th.string] = td.string
return details
def parse_list_page(self, html):
"""解析列表页，提取图书信息"""
soup = BeautifulSoup(html, 'lxml')
books = []
articles = soup.select('article.product_pod')
for article in articles:
book = {}
# 标题和链接
title_link = article.select_one('h3 a')
book['title'] = title_link['title']
book['url'] = urljoin(self.BASE_URL,
title_link['href'])
# 价格
price_elem = article.select_one('.price_color')
book['price'] = price_elem.string if price_elem else ''
# 库存状态
availability = article.select_one('.availability')
if availability:
# 提取库存数量
instock_text = availability.text.strip()
book['availability'] = instock_text
# 尝试提取数字
import re
match = re.search(r'\((\d+) available\)', instock_text)
book['stock'] = int(match.group(1)) if match else 0
else:
book['availability'] = ''
book['stock'] = 0
# 星级评分
rating_elem = article.select_one('.star-rating')
if rating_elem and rating_elem.get('class'):
rating_class = rating_elem['class'][1] # 第二个class是星级
book['rating'] = self.RATING_MAP.get(rating_class, 0)
else:
book['rating'] = 0
# 封面图片
img_elem = article.select_one('img')
if img_elem:
img_src = img_elem.get('src', '')
book['image_url'] = urljoin(self.BASE_URL, img_src)
else:
book['image_url'] = ''
books.append(book)
return books
def get_next_page_url(self, html):
"""获取下一页的URL"""
soup = BeautifulSoup(html, 'lxml')
next_btn = soup.select_one('li.next a')
if next_btn:
return urljoin(self.BASE_URL, next_btn['href'])
return None
def crawl(self, max_pages=None):
"""
爬取图书数据
Args:
max_pages: 最大页数限制
"""
print("=" * 60)
print("图书爬虫启动")
print(f"目标网站: {self.BASE_URL}")
print("=" * 60)
current_url = urljoin(self.BASE_URL, 'catalogue/page-1.html')
page_num = 1
while current_url:
print(f"\n正在爬取第 {page_num} 页: {current_url}")
html = self._get_page(current_url)
if not html:
print(f"第 {page_num} 页获取失败，跳过")
break
# 解析当前页
books = self.parse_list_page(html)
self.all_books.extend(books)
print(f" 获取到 {len(books)} 本图书")
# 检查页数限制
if max_pages and page_num >= max_pages:
print(f" 已达到最大页数限制 {max_pages}")
break
# 获取下一页
current_url = self.get_next_page_url(html)
page_num += 1
# 礼貌延时
if current_url:
self._polite_delay()
print(f"\n爬取完成！共获取 {len(self.all_books)} 本图书")
def save_to_csv(self, filename='books.csv'):
"""保存为CSV"""
if not self.all_books:
print("没有数据可保存")
return
fieldnames = ['title', 'price', 'rating', 'stock',
'availability', 'url', 'image_url']
with open(filename, 'w', newline='', encoding='utf-8-sig') as f:
writer = csv.DictWriter(f, fieldnames=fieldnames)
writer.writeheader()
writer.writerows(self.all_books)
print(f"已保存 {len(self.all_books)} 条记录到 {filename}")
def save_to_json(self, filename='books.json'):
"""保存为JSON"""
if not self.all_books:
print("没有数据可保存")
return
with open(filename, 'w', encoding='utf-8') as f:
json.dump(self.all_books, f, ensure_ascii=False, indent=2)
print(f"已保存 {len(self.all_books)} 条记录到 {filename}")
def print_statistics(self):
"""打印统计信息"""
if not self.all_books:
print("没有数据")
return
print("\n" + "=" * 60)
print("统计信息")
print("=" * 60)
# 总数
print(f"图书总数: {len(self.all_books)}")
# 平均价格
prices = []
for book in self.all_books:
try:
price = float(book['price'].replace('£', ''))
prices.append(price)
except:
pass
if prices:
print(f"平均价格: £{sum(prices)/len(prices):.2f}")
print(f"最高价格: £{max(prices):.2f}")
print(f"最低价格: £{min(prices):.2f}")
# 评分分布
rating_dist = {}
for book in self.all_books:
rating = book['rating']
rating_dist[rating] = rating_dist.get(rating, 0) + 1
print("\n评分分布:")
for rating in sorted(rating_dist.keys()):
count = rating_dist[rating]
bar = '█' * (count // 5)
print(f" {rating}星: {count}本 {bar}")
# 库存统计
in_stock = sum(1 for b in self.all_books if b['stock'] > 0)
print(f"\n有库存: {in_stock}本")
print(f"无库存: {len(self.all_books) - in_stock}本")
# 运行爬虫
if __name__ == "__main__":
crawler = BookCrawler(delay_range=(0.5, 1.5))
# 爬取前3页作为演示（完整爬取去掉max_pages参数）
crawler.crawl(max_pages=3)
# 保存数据
crawler.save_to_csv()
crawler.save_to_json()
# 打印统计信息
crawler.print_statistics()
# 显示前10本书
print("\n前10本书:")
for i, book in enumerate(crawler.all_books[:10], 1):
print(f"{i}. 《{book['title'][:40]}》")
print(f" 价格: {book['price']} | 评分: {book['rating']}星 "
f"| 库存: {book['stock']}本")总结
本教程系统介绍了Python网络爬虫的核心知识：
- 理解了HTTP协议的基础概念，包括请求方法、状态码等
- 掌握了requests库的使用，包括GET/POST请求、Session管理
- 学会了使用BeautifulSoup解析HTML
- 掌握了CSS选择器提取数据的方法
- 实现了分页爬取功能
- 学会了将数据保存为CSV和JSON格式
- 了解了反爬策略：请求头、延时、代理IP、Cookie等
- 知道了robots.txt规范的重要性
- 通过图书爬虫实战综合运用了所有知识
爬虫道德与法律提醒：
- 遵守robots.txt规范
- 控制爬取频率，不要对服务器造成压力
- 不要爬取和个人隐私相关的数据
- 尊重版权，爬取的数据仅用于学习研究
- 商业使用前请确认相关法律法规