Python断点续爬案例如何实现断点续爬取

本文目录导读：

1. 核心思路
2. 数据库版断点续爬
3. 高级特性示例
4. 关键要点总结

我来为你详细介绍Python断点续爬的实现方法,包含完整案例代码。

核心思路

断点续爬的核心是记录爬取进度，在中断后从上次位置继续爬取。

import requests
import json
import os
import time
from typing import Set, Dict, List
from datetime import datetime
class ResumeCrawler:
    """断点续爬爬虫类"""
    def __init__(self, checkpoint_file: str = "checkpoint.json"):
        self.checkpoint_file = checkpoint_file
        self.crawled_urls: Set[str] = set()  # 已爬取的URL
        self.failed_urls: Set[str] = set()    # 失败的URL
        self.pending_urls: List[str] = []     # 待爬取的URL
        self.current_index: int = 0           # 当前进度索引
        self.total_urls: int = 0              # 总URL数量
        # 加载断点信息
        self._load_checkpoint()
    def _load_checkpoint(self):
        """加载断点信息"""
        if os.path.exists(self.checkpoint_file):
            try:
                with open(self.checkpoint_file, 'r', encoding='utf-8') as f:
                    data = json.load(f)
                    self.crawled_urls = set(data.get('crawled_urls', []))
                    self.failed_urls = set(data.get('failed_urls', []))
                    self.pending_urls = data.get('pending_urls', [])
                    self.current_index = data.get('current_index', 0)
                    self.total_urls = data.get('total_urls', 0)
                print(f"[恢复] 发现断点文件，已爬取 {len(self.crawled_urls)} 个URL")
            except Exception as e:
                print(f"[警告] 断点文件加载失败: {e}")
    def _save_checkpoint(self):
        """保存断点信息"""
        data = {
            'crawled_urls': list(self.crawled_urls),
            'failed_urls': list(self.failed_urls),
            'pending_urls': self.pending_urls,
            'current_index': self.current_index,
            'total_urls': self.total_urls,
            'update_time': datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        }
        try:
            with open(self.checkpoint_file, 'w', encoding='utf-8') as f:
                json.dump(data, f, ensure_ascii=False, indent=2)
        except Exception as e:
            print(f"[错误] 保存断点失败: {e}")
    def fetch_url(self, url: str) -> bool:
        """爬取单个URL"""
        try:
            headers = {
                'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
            }
            response = requests.get(url, headers=headers, timeout=10)
            if response.status_code == 200:
                # 保存爬取结果
                self._save_result(url, response.text)
                return True
            else:
                print(f"[失败] {url} - 状态码: {response.status_code}")
                return False
        except Exception as e:
            print(f"[异常] {url} - 错误: {str(e)}")
            return False
    def _save_result(self, url: str, content: str):
        """保存爬取结果到文件"""
        # 创建保存目录
        save_dir = "crawled_data"
        os.makedirs(save_dir, exist_ok=True)
        # 生成文件名
        filename = f"{save_dir}/page_{len(self.crawled_urls)}.html"
        # 保存内容
        with open(filename, 'w', encoding='utf-8') as f:
            f.write(content)
        print(f"[成功] 已保存: {filename}")
    def crawl(self, urls: List[str], batch_size: int = 10):
        """执行爬取任务"""
        # 初始化任务
        if not self.pending_urls:
            self.pending_urls = [url for url in urls if url not in self.crawled_urls]
            self.total_urls = len(self.pending_urls)
            self.current_index = 0
        else:
            # 从断点恢复时，过滤掉已爬取的URL
            self.pending_urls = [url for url in self.pending_urls if url not in self.crawled_urls]
        total = len(self.pending_urls)
        print(f"[开始] 总共需要爬取 {total} 个URL")
        # 分批处理
        for i in range(0, total, batch_size):
            batch = self.pending_urls[i:i+batch_size]
            print(f"\n[批次] 处理第 {i+1}-{min(i+batch_size, total)} 个URL")
            # 爬取批次中的每个URL
            for url in batch:
                print(f"[爬取] {self.current_index + 1}/{total}: {url}")
                success = self.fetch_url(url)
                if success:
                    self.crawled_urls.add(url)
                else:
                    self.failed_urls.add(url)
                self.current_index += 1
                # 每处理一个URL保存一次断点
                self._save_checkpoint()
                # 延时避免请求过快
                time.sleep(1)
            # 打印进度
            progress = (self.current_index / total) * 100
            print(f"[进度] 已完成 {self.current_index}/{total} ({progress:.1f}%)")
            print(f"[统计] 成功: {len(self.crawled_urls)}, 失败: {len(self.failed_urls)}")
        # 爬取完成，清理断点文件
        if self.current_index >= total:
            self._cleanup_checkpoint()
            print("\n[完成] 所有URL爬取完成！")
    def _cleanup_checkpoint(self):
        """清理断点文件"""
        if os.path.exists(self.checkpoint_file):
            os.remove(self.checkpoint_file)
            print("[清理] 断点文件已删除")
# 使用示例
def main():
    """主函数示例"""
    # 生成示例URL列表
    base_url = "https://httpbin.org/delay/"
    urls = [f"{base_url}{i}" for i in range(1, 21)]  # 20个示例URL
    # 创建爬虫实例
    crawler = ResumeCrawler("my_crawler_checkpoint.json")
    print("=" * 50)
    print("断点续爬爬虫")
    print("=" * 50)
    try:
        # 执行爬取
        crawler.crawl(urls, batch_size=5)
    except KeyboardInterrupt:
        print("\n[中断] 用户中断爬取！")
        print(f"[保存] 断点已保存到: {crawler.checkpoint_file}")
        print(f"下次运行将从第 {crawler.current_index + 1} 个URL继续")
    except Exception as e:
        print(f"\n[错误] 发生异常: {e}")
        print("[保存] 已保存断点信息")
if __name__ == "__main__":
    main()

数据库版断点续爬

使用SQLite数据库存储进度更可靠：

import sqlite3
import requests
from datetime import datetime
class DatabaseResumeCrawler:
    """使用数据库的断点续爬"""
    def __init__(self, db_path: str = "crawler.db"):
        self.db_path = db_path
        self._init_database()
    def _init_database(self):
        """初始化数据库"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        # 创建任务表
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS crawl_tasks (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                url TEXT UNIQUE NOT NULL,
                status TEXT DEFAULT 'pending',
                retry_count INTEGER DEFAULT 0,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        ''')
        # 创建结果表
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS crawl_results (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                url TEXT NOT NULL,
                content TEXT,
                crawled_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        ''')
        conn.commit()
        conn.close()
    def add_tasks(self, urls: List[str]):
        """添加爬取任务"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        for url in urls:
            try:
                cursor.execute(
                    "INSERT OR IGNORE INTO crawl_tasks (url) VALUES (?)",
                    (url,)
                )
            except:
                pass
        conn.commit()
        conn.close()
        print(f"[添加] 已添加 {len(urls)} 个任务")
    def get_pending_tasks(self) -> List[tuple]:
        """获取待爬取任务"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute(
            "SELECT id, url FROM crawl_tasks WHERE status = 'pending' ORDER BY id"
        )
        tasks = cursor.fetchall()
        conn.close()
        return tasks
    def update_task_status(self, task_id: int, status: str):
        """更新任务状态"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute(
            "UPDATE crawl_tasks SET status = ?, updated_at = ? WHERE id = ?",
            (status, datetime.now(), task_id)
        )
        conn.commit()
        conn.close()
    def save_result(self, url: str, content: str):
        """保存爬取结果"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute(
            "INSERT INTO crawl_results (url, content) VALUES (?, ?)",
            (url, content)
        )
        conn.commit()
        conn.close()
    def crawl(self, batch_size: int = 10):
        """执行爬取"""
        while True:
            # 获取待爬取任务
            tasks = self.get_pending_tasks()
            if not tasks:
                print("[完成] 所有任务已完成！")
                break
            print(f"[信息] 剩余 {len(tasks)} 个待爬取任务")
            # 分批处理
            batch = tasks[:batch_size]
            for task_id, url in batch:
                print(f"[爬取] {url}")
                try:
                    # 模拟爬取
                    response = requests.get(url, timeout=10)
                    if response.status_code == 200:
                        self.save_result(url, response.text)
                        self.update_task_status(task_id, 'success')
                        print(f"[成功] {url}")
                    else:
                        self.update_task_status(task_id, 'failed')
                        print(f"[失败] {url} - 状态码: {response.status_code}")
                except Exception as e:
                    self.update_task_status(task_id, 'failed')
                    print(f"[异常] {url} - {str(e)}")
                time.sleep(0.5)
            # 显示统计信息
            self.show_statistics()
    def show_statistics(self):
        """显示统计信息"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute("""
            SELECT 
                status, 
                COUNT(*) as count 
            FROM crawl_tasks 
            GROUP BY status
        """)
        stats = cursor.fetchall()
        print("\n[统计]")
        for status, count in stats:
            print(f"  {status}: {count}")
        print()
        conn.close()
# 数据库版本使用示例
def db_crawler_example():
    crawler = DatabaseResumeCrawler("resume_crawler.db")
    # 添加任务
    urls = [f"https://httpbin.org/get?id={i}" for i in range(1, 21)]
    crawler.add_tasks(urls)
    # 执行爬取
    crawler.crawl(batch_size=5)
if __name__ == "__main__":
    # 运行文件版
    # main()
    # 运行数据库版
    db_crawler_example()

高级特性示例

class AdvancedResumeCrawler:
    """高级版断点续爬，支持更多特性"""
    def __init__(self):
        self.checkpoint = {
            'version': '1.0',
            'crawled': [],
            'failed': [],
            'pending': [],
            'metadata': {
                'start_time': None,
                'last_update': None,
                'total_urls': 0,
                'success_count': 0,
                'fail_count': 0,
                'avg_speed': 0
            }
        }
        self.max_retries = 3
        self.download_speed = []
    def retry_failed(self):
        """重试失败的URL"""
        failed_urls = self.checkpoint['failed'].copy()
        self.checkpoint['failed'] = []
        for url in failed_urls:
            success = self.crawl_single(url)
            if not success:
                # 如果还是失败，添加到失败列表
                self.checkpoint['failed'].append(url)
        self._save_checkpoint()
    def merge_checkpoints(self, checkpoint_files: List[str]):
        """合并多个断点文件"""
        merged_crawled = set(self.checkpoint['crawled'])
        merged_failed = set(self.checkpoint['failed'])
        for file in checkpoint_files:
            try:
                with open(file, 'r') as f:
                    data = json.load(f)
                    merged_crawled.update(data.get('crawled', []))
                    merged_failed.update(data.get('failed', []))
            except:
                pass
        self.checkpoint['crawled'] = list(merged_crawled)
        self.checkpoint['failed'] = list(merged_failed)
        # 更新待爬取列表
        all_urls = set(self.checkpoint['pending'])
        self.checkpoint['pending'] = list(
            all_urls - merged_crawled - merged_failed
        )
        self._save_checkpoint()

关键要点总结

核心机制

• 进度记录：实时保存已爬取/失败/待爬取的URL
• 断点恢复：启动时检查并加载断点文件
• 容错处理：网络异常、程序中断时的数据保护

实现技巧

1. 频繁保存：每处理一个URL就更新断点
2. 原子操作：确保断点文件写入完整
3. 状态追踪：记录每个URL的状态（待爬取/成功/失败）
4. 批量处理：分批爬取，避免内存溢出
5. 重试机制：失败URL自动重试

注意事项

• 合理设置请求间隔,避免被封IP
• 定期清理断点文件,避免数据冗余
• 对于大型爬取任务,使用数据库存储更可靠
• 做好日志记录,便于问题排查

这个实现可以处理网络中断、程序崩溃、手动中断等各种情况，确保爬虫任务可以从上次中断处继续执行。