IT技术社区_专业开发者平台_最新编程教程与解决方案 -沐辰社区

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

wen 实用脚本 225

本文目录导读：

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

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

实现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'])
  1. 状态持久化：将抓取状态保存到文件或数据库
  2. 唯一标识：使用ID、URL或时间戳作为唯一标识
  3. 去重机制：避免重复处理相同数据
  4. 错误恢复：支持断点续传
  5. 性能优化：只处理新增和变化的数据

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

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

版权声明：足球直播 足球直播 足球直播 足球直播 足球直播 中超直播 中超直播 中超直播 中超直播 中超直播 英超直播 英超直播 英超直播 英超直播 英超直播 德甲直播 德甲直播 德甲直播 德甲直播 德甲直播 西甲直播 西甲直播 西甲直播 西甲直播 西甲直播 意甲直播 意甲直播 意甲直播 意甲直播 意甲直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 篮球直播 篮球直播 篮球直播 篮球直播 篮球直播 足球直播 足球直播 足球直播 足球直播 足球直播 中超直播 中超直播 中超直播 中超直播 中超直播 英超直播 英超直播 英超直播 英超直播 英超直播 德甲直播 德甲直播 德甲直播 德甲直播 德甲直播 西甲直播 西甲直播 西甲直播 西甲直播 西甲直播 意甲直播 意甲直播 意甲直播 意甲直播 意甲直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 篮球直播 篮球直播 篮球直播 篮球直播 篮球直播 足球直播 足球直播 足球直播 足球直播 足球直播 中超直播 中超直播 中超直播 中超直播 中超直播 英超直播 英超直播 英超直播 英超直播 英超直播 德甲直播 德甲直播 德甲直播 德甲直播 德甲直播 西甲直播 西甲直播 西甲直播 西甲直播 西甲直播 意甲直播 意甲直播 意甲直播 意甲直播 意甲直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 篮球直播 篮球直播 篮球直播 篮球直播 篮球直播

上一篇Python脚本爬虫反爬策略如何应对

下一篇Python脚本爬虫分布式如何协调

相关文章

抱歉，评论功能暂时关闭!

友情链接攻略窝 · 礼包窝 · 配置窝 · 换算窝 · 算法窝 · 查文窝 · 手册窝 · 问答窝 · 全文搜 · 手册库
相关站点礼包窝 · 查文窝