Python脚本爬虫增量抓取如何实现

wen 实用脚本 2026-07-19 4

本文目录导读：

Python脚本爬虫增量抓取如何实现

基于时间戳的增量抓取
基于ID的增量抓取
基于哈希校验的增量抓取
数据库记录方案
综合方案（带调度）

实现Python爬虫的增量抓取，核心思路是记录已抓取的状态，只抓取新增或更新的内容,以下是几种常见的实现方案：

基于时间戳的增量抓取

最简单的增量方式,记录上次抓取时间。

import requests
import json
from datetime import datetime
import time
class TimestampIncrementalCrawler:
    def __init__(self, state_file='crawler_state.json'):
        self.state_file = state_file
        self.last_crawl_time = self.load_state()
    def load_state(self):
        """加载上次抓取时间"""
        try:
            with open(self.state_file, 'r') as f:
                state = json.load(f)
                return state.get('last_crawl_time')
        except FileNotFoundError:
            return None
    def save_state(self, crawl_time):
        """保存当前抓取时间"""
        state = {'last_crawl_time': crawl_time}
        with open(self.state_file, 'w') as f:
            json.dump(state, f)
    def crawl(self):
        current_time = datetime.now().isoformat()
        # 构建请求参数，只获取最新数据
        params = {}
        if self.last_crawl_time:
            params['since'] = self.last_crawl_time
        try:
            response = requests.get('https://api.example.com/posts', params=params)
            data = response.json()
            # 处理新数据
            for item in data['items']:
                self.process_item(item)
            # 更新状态
            self.save_state(current_time)
            print(f"成功抓取 {len(data['items'])} 条新数据")
        except Exception as e:
            print(f"抓取失败: {e}")
    def process_item(self, item):
        """处理每条数据"""
        print(f"处理: {item['id']} - {item['title']}")

基于ID的增量抓取

适用于有自增ID或唯一标识的数据源。

class IDBasedIncrementalCrawler:
    def __init__(self, state_file='crawler_state.json'):
        self.state_file = state_file
        self.last_processed_id = self.load_state()
    def load_state(self):
        try:
            with open(self.state_file, 'r') as f:
                state = json.load(f)
                return state.get('last_id', 0)
        except FileNotFoundError:
            return 0
    def save_state(self, last_id):
        state = {'last_id': last_id}
        with open(self.state_file, 'w') as f:
            json.dump(state, f)
    def crawl(self):
        current_max_id = self.last_processed_id
        # 从上次处开始抓取
        page = 1
        while True:
            params = {
                'page': page,
                'per_page': 100,
                'min_id': self.last_processed_id + 1
            }
            response = requests.get('https://api.example.com/items', params=params)
            items = response.json()
            if not items:
                break
            for item in items:
                self.process_item(item)
                current_max_id = max(current_max_id, item['id'])
            page += 1
        # 保存最新ID
        if current_max_id > self.last_processed_id:
            self.save_state(current_max_id)
            print(f"已处理到ID: {current_max_id}")

基于哈希校验的增量抓取

适用于需要检测内容变化的场景。

import hashlib
import pickle
class HashBasedIncrementalCrawler:
    def __init__(self, state_file='content_hashes.pkl'):
        self.state_file = state_file
        self.content_hashes = self.load_hashes()
    def load_hashes(self):
        try:
            with open(self.state_file, 'rb') as f:
                return pickle.load(f)
        except FileNotFoundError:
            return {}
    def save_hashes(self):
        with open(self.state_file, 'wb') as f:
            pickle.dump(self.content_hashes, f)
    def calculate_hash(self, content):
        """计算内容的哈希值"""
        return hashlib.md5(str(content).encode()).hexdigest()
    def crawl(self):
        response = requests.get('https://example.com/page')
        soup = BeautifulSoup(response.text, 'html.parser')
        # 获取所有文章
        articles = soup.find_all('article')
        changed_items = []
        new_items = []
        for article in articles:
            item_id = article['data-id']
            content = article.get_text()
            content_hash = self.calculate_hash(content)
            if item_id in self.content_hashes:
                # 检查内容是否变化
                if self.content_hashes[item_id] != content_hash:
                    changed_items.append(item_id)
                    self.content_hashes[item_id] = content_hash
            else:
                # 新内容
                new_items.append(item_id)
                self.content_hashes[item_id] = content_hash
        self.save_hashes()
        print(f"新内容: {len(new_items)}, 变化内容: {len(changed_items)}")

数据库记录方案

使用数据库记录抓取状态,更可靠。

import sqlite3
from datetime import datetime
class DatabaseIncrementalCrawler:
    def __init__(self, db_path='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_records (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                url TEXT UNIQUE,
                content_hash TEXT,
                last_crawled TIMESTAMP,
                is_updated BOOLEAN DEFAULT 0
            )
        ''')
        conn.commit()
        conn.close()
    def is_crawled(self, url):
        """检查URL是否已抓取"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute('SELECT content_hash, last_crawled FROM crawl_records WHERE url = ?', (url,))
        result = cursor.fetchone()
        conn.close()
        return result
    def save_record(self, url, content_hash):
        """保存抓取记录"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute('''
            INSERT OR REPLACE INTO crawl_records 
            (url, content_hash, last_crawled, is_updated)
            VALUES (?, ?, ?, ?)
        ''', (url, content_hash, datetime.now(), 0))
        conn.commit()
        conn.close()
    def crawl(self, urls):
        for url in urls:
            # 检查是否已抓取
            existing = self.is_crawled(url)
            if existing:
                # 如果已存在，判断是否需要更新
                response = requests.get(url)
                new_hash = hashlib.md5(response.text.encode()).hexdigest()
                if existing[0] != new_hash:
                    print(f"内容更新: {url}")
                    self.save_record(url, new_hash)
                else:
                    print(f"内容未变化: {url}")
            else:
                # 新URL，直接抓取
                print(f"新URL: {url}")
                response = requests.get(url)
                content_hash = hashlib.md5(response.text.encode()).hexdigest()
                self.save_record(url, content_hash)

综合方案（带调度）

结合定时任务和多种增量策略。

import schedule
import time
from datetime import datetime, timedelta
class AdvancedIncrementalCrawler:
    def __init__(self):
        self.state = {
            'last_crawl_time': None,
            'last_ids': {},
            'url_hashes': {}
        }
        self.load_state()
    def load_state(self):
        """加载完整状态"""
        try:
            with open('crawler_state.json', 'r') as f:
                self.state = json.load(f)
        except FileNotFoundError:
            pass
    def save_state(self):
        with open('crawler_state.json', 'w') as f:
            json.dump(self.state, f, indent=2)
    def crawl_with_multiple_strategies(self):
        """组合多种增量策略"""
        print(f"开始增量抓取: {datetime.now()}")
        # 策略1: 基于时间
        if self.state['last_crawl_time']:
            print(f"上次抓取时间: {self.state['last_crawl_time']}")
        # 策略2: 基于ID
        for source, last_id in self.state['last_ids'].items():
            print(f"{source} 最后ID: {last_id}")
        # 策略3: 基于Hash
        for url, hash_val in self.state['url_hashes'].items():
            print(f"URL: {url}, Hash: {hash_val}")
        # 执行实际抓取逻辑
        self.do_crawl()
        # 更新状态
        self.state['last_crawl_time'] = datetime.now().isoformat()
        self.save_state()
    def do_crawl(self):
        # 实际抓取逻辑
        pass
    def run_scheduled(self):
        """定时运行"""
        # 每小时执行一次
        schedule.every().hour.do(self.crawl_with_multiple_strategies)
        while True:
            schedule.run_pending()
            time.sleep(60)
# 使用示例
if __name__ == "__main__":
    # 简单时间戳方案
    crawler = TimestampIncrementalCrawler()
    crawler.crawl()
    # 或使用数据库方案
    db_crawler = DatabaseIncrementalCrawler()
    db_crawler.crawl(['https://example.com/page1', 'https://example.com/page2'])

状态持久化：将抓取状态保存到文件或数据库
唯一标识：使用ID、URL或时间戳作为唯一标识
去重机制：避免重复处理相同数据
错误恢复：支持断点续传
性能优化：只处理新增和变化的数据

选择哪种方案取决于你的具体需求：

有时间戳的API：使用时间戳方案
自增ID的数据：使用ID方案可能变化：使用Hash方案
需要持久化：使用数据库方案