本文目录导读：
当Python爬虫失败时，记录断点（断点续爬）有几种常见方法,我来介绍几种实用的方案：
使用JSON文件记录断点（最简单）
import json
import os
from datetime import datetime
class CrawlerWithCheckpoint:
def __init__(self, checkpoint_file='checkpoint.json'):
self.checkpoint_file = checkpoint_file
self.checkpoint = self.load_checkpoint()
def load_checkpoint(self):
"""加载断点记录"""
if os.path.exists(self.checkpoint_file):
with open(self.checkpoint_file, 'r', encoding='utf-8') as f:
return json.load(f)
return {
'last_index': 0,
'completed_urls': [],
'failed_urls': [],
'timestamp': None
}
def save_checkpoint(self, current_index, url, success=True):
"""保存断点"""
self.checkpoint['last_index'] = current_index
self.checkpoint['timestamp'] = datetime.now().isoformat()
if success:
if url not in self.checkpoint['completed_urls']:
self.checkpoint['completed_urls'].append(url)
else:
if url not in self.checkpoint['failed_urls']:
self.checkpoint['failed_urls'].append(url)
with open(self.checkpoint_file, 'w', encoding='utf-8') as f:
json.dump(self.checkpoint, f, ensure_ascii=False, indent=2)
def crawl(self, urls):
"""爬虫主逻辑"""
start_index = self.checkpoint['last_index']
for i, url in enumerate(urls[start_index:], start=start_index):
try:
# 模拟爬取
print(f"爬取: {url}")
# 模拟随机失败
if i == 3: # 假设第3个URL失败
raise Exception("网络错误")
# 爬取成功，记录断点
self.save_checkpoint(i + 1, url, success=True)
except Exception as e:
print(f"爬取失败: {url}, 错误: {e}")
self.save_checkpoint(i, url, success=False)
break # 或继续尝试其他URL
# 使用示例
urls = [f"https://example.com/page/{i}" for i in range(10)]
crawler = CrawlerWithCheckpoint()
crawler.crawl(urls)
使用SQLite数据库（适合大量数据）
import sqlite3
from datetime import datetime
class DatabaseCheckpoint:
def __init__(self, db_file='crawler_checkpoint.db'):
self.conn = sqlite3.connect(db_file)
self.create_tables()
def create_tables(self):
"""创建表结构"""
cursor = self.conn.cursor()
# URL状态表
cursor.execute('''
CREATE TABLE IF NOT EXISTS url_status (
id INTEGER PRIMARY KEY AUTOINCREMENT,
url TEXT UNIQUE,
status TEXT, -- pending, processing, completed, failed
error_message TEXT,
retry_count INTEGER DEFAULT 0,
created_at TIMESTAMP,
updated_at TIMESTAMP
)
''')
# 爬虫状态表
cursor.execute('''
CREATE TABLE IF NOT EXISTS crawler_status (
id INTEGER PRIMARY KEY AUTOINCREMENT,
batch_id TEXT,
current_position INTEGER,
total_urls INTEGER,
is_running BOOLEAN,
last_updated TIMESTAMP
)
''')
self.conn.commit()
def init_urls(self, urls, batch_id=None):
"""初始化URL列表"""
cursor = self.conn.cursor()
now = datetime.now()
for url in urls:
cursor.execute('''
INSERT OR IGNORE INTO url_status
(url, status, created_at, updated_at)
VALUES (?, 'pending', ?, ?)
''', (url, now, now))
# 更新爬虫状态
cursor.execute('''
INSERT OR REPLACE INTO crawler_status
(id, batch_id, current_position, total_urls, is_running, last_updated)
VALUES (1, ?, 0, ?, 1, ?)
''', (batch_id, len(urls), now))
self.conn.commit()
def update_url_status(self, url, status, error=None):
"""更新URL状态"""
cursor = self.conn.cursor()
now = datetime.now()
retry_count = 0
if status == 'failed':
cursor.execute('''
UPDATE url_status
SET status = ?, error_message = ?, retry_count = retry_count + 1, updated_at = ?
WHERE url = ?
''', (status, error, now, url))
else:
cursor.execute('''
UPDATE url_status
SET status = ?, updated_at = ?
WHERE url = ?
''', (status, now, url))
# 更新爬虫位置
cursor.execute('''
UPDATE crawler_status
SET current_position = current_position + 1, last_updated = ?
WHERE id = 1
''', (now,))
self.conn.commit()
def get_unprocessed_urls(self):
"""获取未处理的URL"""
cursor = self.conn.cursor()
cursor.execute('''
SELECT url FROM url_status
WHERE status IN ('pending', 'failed')
ORDER BY id
''')
return [row[0] for row in cursor.fetchall()]
def get_status_summary(self):
"""获取状态摘要"""
cursor = self.conn.cursor()
cursor.execute('''
SELECT status, COUNT(*) as count
FROM url_status
GROUP BY status
''')
return cursor.fetchall()
# 使用示例
db = DatabaseCheckpoint()
urls = [f"https://example.com/page/{i}" for i in range(100)]
db.init_urls(urls, batch_id="batch_001")
# 爬取逻辑
unprocessed = db.get_unprocessed_urls()
for url in unprocessed:
try:
# 模拟爬取
result = f"Crawled: {url}"
db.update_url_status(url, 'completed')
except Exception as e:
db.update_url_status(url, 'failed', str(e))
# 查看状态
summary = db.get_status_summary()
for status, count in summary:
print(f"{status}: {count}")
使用Pickle序列化（适合复杂对象）
import pickle
import os
from typing import Any, Dict
class PickleCheckpoint:
def __init__(self, filepath='checkpoint.pkl'):
self.filepath = filepath
def save(self, data: Dict[str, Any]):
"""保存断点数据"""
with open(self.filepath, 'wb') as f:
pickle.dump(data, f)
print(f"断点保存到: {self.filepath}")
def load(self) -> Dict[str, Any]:
"""加载断点数据"""
if os.path.exists(self.filepath):
with open(self.filepath, 'rb') as f:
return pickle.load(f)
return {}
def clear(self):
"""清除断点"""
if os.path.exists(self.filepath):
os.remove(self.filepath)
print("断点已清除")
# 使用示例
checkpoint = PickleCheckpoint()
# 保存断点
crawler_state = {
'current_page': 5,
'processed_items': ['item1', 'item2'],
'failed_items': ['item3'],
'cookies': {'session': 'abc123'},
'headers': {'User-Agent': 'Mozilla/5.0'}
}
checkpoint.save(crawler_state)
# 恢复断点
state = checkpoint.load()
print(f"从第 {state['current_page']} 页继续爬取")
使用装饰器自动记录断点
import functools
import json
import traceback
def auto_checkpoint(checkpoint_file='auto_checkpoint.json'):
"""自动记录断点的装饰器"""
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
# 加载已存在的断点
try:
with open(checkpoint_file, 'r') as f:
checkpoint_data = json.load(f)
except (FileNotFoundError, json.JSONDecodeError):
checkpoint_data = {'completed': [], 'failed': []}
# 执行原函数
try:
result = func(*args, **kwargs)
# 记录成功
url = kwargs.get('url', '')
if url:
checkpoint_data['completed'].append(url)
except Exception as e:
# 记录失败
url = kwargs.get('url', '')
if url:
checkpoint_data['failed'].append({
'url': url,
'error': str(e),
'traceback': traceback.format_exc()
})
print(f"爬取失败: {url}")
raise
finally:
# 保存断点
with open(checkpoint_file, 'w') as f:
json.dump(checkpoint_data, f, indent=2, ensure_ascii=False)
return result
return wrapper
return decorator
# 使用示例
@auto_checkpoint('crawler_checkpoint.json')
def crawl_single_page(url, **kwargs):
# 爬取逻辑
print(f"正在爬取: {url}")
# 模拟失败
if 'fail' in url:
raise Exception("模拟失败")
return f"成功爬取: {url}"
# 批量爬取
urls = [
'https://example.com/page/1',
'https://example.com/page/2',
'https://example.com/page/fail', # 这个会失败
'https://example.com/page/3'
]
for url in urls:
try:
crawl_single_page(url=url)
except:
continue
最佳实践建议
class RobustCrawler:
def __init__(self):
self.checkpoint_method = 'json' # 或 'db', 'pickle'
self.max_retries = 3
self.timeout = 30
def crawl_with_retry(self, url, max_retries=None):
"""带重试机制的爬取"""
max_retries = max_retries or self.max_retries
for attempt in range(max_retries):
try:
# 爬取逻辑
result = self.fetch_url(url)
return result
except Exception as e:
if attempt == max_retries - 1:
# 最后一次失败，记录断点
self.save_checkpoint(url, failed=True)
raise
else:
# 等待后重试
import time
time.sleep(2 ** attempt) # 指数退避
def batch_crawl(self, urls):
"""批量爬取"""
for i, url in enumerate(urls):
try:
# 检查是否已经爬取过
if self.is_completed(url):
continue
result = self.crawl_with_retry(url)
self.save_checkpoint(url, success=True)
# 每100个URL保存一次
if i % 100 == 0:
self.save_progress(i, len(urls))
except KeyboardInterrupt:
print("用户中断，保存断点...")
self.save_progress(i, len(urls))
break
except Exception as e:
print(f"爬取失败: {url}, 错误: {e}")
continue
-
选择合适的存储方式：
- 简单场景：JSON文件
- 大量数据：SQLite数据库
- 复杂对象：Pickle
-
记录关键信息：
- 已处理的URL
- 失败的URL及其错误信息
- 当前处理位置
- 时间戳
-
实现重试机制：
- 指数退避等待
- 最大重试次数限制
-
处理中断情况：
- 键盘中断（Ctrl+C）
- 网络断开
- 程序异常
这样即使爬虫中途失败，也能从断点处继续爬取,避免重复工作。
