Python脚本爬虫失败如何记录断点

本文目录导读：

1. 使用JSON文件记录断点（最简单）
2. 使用SQLite数据库（适合大量数据）
3. 使用Pickle序列化（适合复杂对象）
4. 使用装饰器自动记录断点
5. 最佳实践建议

当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

1. 选择合适的存储方式：

   • 简单场景：JSON文件
   • 大量数据：SQLite数据库
   • 复杂对象：Pickle

2. 记录关键信息：

   • 已处理的URL
   • 失败的URL及其错误信息
   • 当前处理位置
   • 时间戳

3. 实现重试机制：

   • 指数退避等待
   • 最大重试次数限制

4. 处理中断情况：

   • 键盘中断（Ctrl+C）
   • 网络断开
   • 程序异常

这样即使爬虫中途失败，也能从断点处继续爬取,避免重复工作。