Python爬虫保存案例如何保存爬取数据

wen python案例 2026-06-30 255

本文目录导读：

基础文件保存方式
数据库保存方式
使用Pandas保存数据
综合案例：完整的爬虫数据保存
最佳实践建议

我来详细介绍Python爬虫保存数据的多种方式及完整案例：

基础文件保存方式

1 保存为JSON文件

import json
import requests
def save_to_json():
    # 爬取数据
    url = "https://api.example.com/data"
    response = requests.get(url)
    data = response.json()
    # 保存为JSON
    with open('data.json', 'w', encoding='utf-8') as f:
        json.dump(data, f, ensure_ascii=False, indent=4)
    print("数据已保存为JSON文件")
# 完整案例
def crawl_and_save_json():
    headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
    }
    # 假设爬取一些文章数据
    articles = [
        {'title': '文章1', 'content': '内容1', 'date': '2024-01-01'},
        {'title': '文章2', 'content': '内容2', 'date': '2024-01-02'}
    ]
    # 保存到JSON文件
    with open('articles.json', 'w', encoding='utf-8') as f:
        json.dump(articles, f, ensure_ascii=False, indent=2)
    print(f"成功保存 {len(articles)} 篇文章到 articles.json")
crawl_and_save_json()

2 保存为CSV文件

import csv
import requests
from bs4 import BeautifulSoup
def save_to_csv():
    # 示例数据
    data = [
        ['标题', '作者', '日期', '链接'],
        ['Python教程', '张三', '2024-01-01', 'http://example.com/1'],
        ['数据科学入门', '李四', '2024-01-02', 'http://example.com/2']
    ]
    # 保存为CSV
    with open('data.csv', 'w', newline='', encoding='utf-8-sig') as f:
        writer = csv.writer(f)
        writer.writerows(data)
    print("数据已保存为CSV文件")
# 完整爬取并保存案例
def crawl_and_save_csv():
    headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
    }
    # 模拟爬取的新闻数据
    news_data = []
    for i in range(10):
        news_data.append([
            f'新闻标题{i+1}',
            f'作者{i+1}',
            f'2024-01-{i+1:02d}',
            f'http://example.com/news/{i+1}'
        ])
    # 写入CSV
    with open('news.csv', 'w', newline='', encoding='utf-8-sig') as f:
        writer = csv.writer(f)
        writer.writerow(['标题', '作者', '日期', '链接'])  # 写入表头
        writer.writerows(news_data)  # 写入数据
    print(f"成功保存 {len(news_data)} 条新闻到 news.csv")
crawl_and_save_csv()

数据库保存方式

1 SQLite数据库

import sqlite3
import requests
from datetime import datetime
class DataSaver:
    def __init__(self, db_name='crawl_data.db'):
        self.conn = sqlite3.connect(db_name)
        self.cursor = self.conn.cursor()
        self.create_tables()
    def create_tables(self):
        # 创建文章表
        self.cursor.execute('''
            CREATE TABLE IF NOT EXISTS articles (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                title TEXT NOT NULL,
                content TEXT,
                author TEXT,
                publish_date TEXT,
                source_url TEXT UNIQUE,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        ''')
        self.conn.commit()
    def save_article(self, article):
        try:
            self.cursor.execute('''
                INSERT INTO articles 
                (title, content, author, publish_date, source_url)
                VALUES (?, ?, ?, ?, ?)
            ''', (
                article['title'],
                article.get('content', ''),
                article.get('author', ''),
                article.get('publish_date', ''),
                article['source_url']
            ))
            self.conn.commit()
            print(f"保存文章: {article['title']}")
        except sqlite3.IntegrityError:
            print(f"文章已存在: {article['title']}")
    def save_batch(self, articles):
        for article in articles:
            self.save_article(article)
    def close(self):
        self.conn.close()
# 使用示例
def crawl_and_save_to_db():
    # 模拟爬取数据
    articles = [
        {
            'title': 'Python爬虫入门',
            'content': '这是一个完整的爬虫教程...',
            'author': '张三',
            'publish_date': '2024-01-15',
            'source_url': 'http://example.com/1'
        },
        {
            'title': '数据分析基础',
            'content': '学习数据处理的技巧...',
            'author': '李四',
            'publish_date': '2024-01-16',
            'source_url': 'http://example.com/2'
        }
    ]
    # 保存到数据库
    saver = DataSaver()
    saver.save_batch(articles)
    saver.close()
    print("数据已保存到数据库")
crawl_and_save_to_db()

2 MySQL数据库

import pymysql
import requests
from typing import List, Dict
class MySQLDataSaver:
    def __init__(self, host='localhost', port=3306, user='root', 
                 password='password', database='crawl_data'):
        self.connection = pymysql.connect(
            host=host,
            port=port,
            user=user,
            password=password,
            database=database,
            charset='utf8mb4'
        )
        self.cursor = self.connection.cursor()
        self.create_table()
    def create_table(self):
        sql = '''
            CREATE TABLE IF NOT EXISTS products (
                id INT AUTO_INCREMENT PRIMARY KEY,
                name VARCHAR(255) NOT NULL,
                price DECIMAL(10, 2),
                description TEXT,
                category VARCHAR(100),
                source_url VARCHAR(500) UNIQUE,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
        '''
        self.cursor.execute(sql)
        self.connection.commit()
    def save_product(self, product: Dict):
        sql = '''
            INSERT INTO products (name, price, description, category, source_url)
            VALUES (%s, %s, %s, %s, %s)
            ON DUPLICATE KEY UPDATE
            price = VALUES(price),
            description = VALUES(description)
        '''
        try:
            self.cursor.execute(sql, (
                product['name'],
                product['price'],
                product['description'],
                product['category'],
                product['source_url']
            ))
            self.connection.commit()
        except Exception as e:
            print(f"保存出错: {e}")
            self.connection.rollback()
    def close(self):
        self.cursor.close()
        self.connection.close()
# 使用示例
def save_products_example():
    products = [
        {
            'name': '笔记本电脑',
            'price': 5999.00,
            'description': '高性能轻薄本',
            'category': '电子产品',
            'source_url': 'http://example.com/product/1'
        }
    ]
    saver = MySQLDataSaver()
    for product in products:
        saver.save_product(product)
    saver.close()
# save_products_example()  # 需要MySQL环境才能运行

使用Pandas保存数据

import pandas as pd
import requests
def save_with_pandas():
    # 爬取数据并转换为DataFrame
    data = {
        'title': ['文章1', '文章2', '文章3'],
        'author': ['张三', '李四', '王五'],
        'views': [100, 200, 150],
        'date': ['2024-01-01', '2024-01-02', '2024-01-03']
    }
    df = pd.DataFrame(data)
    # 保存为多种格式
    # CSV格式
    df.to_csv('data_pandas.csv', index=False, encoding='utf-8-sig')
    # Excel格式（需要安装openpyxl）
    df.to_excel('data_pandas.xlsx', index=False)
    # JSON格式
    df.to_json('data_pandas.json', orient='records', force_ascii=False)
    # HTML表格
    df.to_html('data_pandas.html', index=False)
    print("使用Pandas保存数据完成")
save_with_pandas()

综合案例：完整的爬虫数据保存

import requests
from bs4 import BeautifulSoup
import json
import csv
import sqlite3
from datetime import datetime
import os
class CrawlerDataSaver:
    """完整的爬虫数据保存器"""
    def __init__(self, name='crawler'):
        self.name = name
        self.data = []
        self.setup_directories()
    def setup_directories(self):
        """创建必要的目录"""
        for dir_name in ['json', 'csv', 'db', 'excel']:
            os.makedirs(dir_name, exist_ok=True)
    def add_data(self, item):
        """添加数据"""
        self.data.append(item)
    def save_to_json(self, filename=None):
        """保存为JSON"""
        if not filename:
            filename = f"json/{self.name}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
        with open(filename, 'w', encoding='utf-8') as f:
            json.dump(self.data, f, ensure_ascii=False, indent=2)
        print(f"保存JSON: {filename}")
        return filename
    def save_to_csv(self, filename=None):
        """保存为CSV"""
        if not filename:
            filename = f"csv/{self.name}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv"
        if not self.data:
            print("没有数据可保存")
            return
        # 获取字段名
        fields = self.data[0].keys()
        with open(filename, 'w', newline='', encoding='utf-8-sig') as f:
            writer = csv.DictWriter(f, fieldnames=fields)
            writer.writeheader()
            writer.writerows(self.data)
        print(f"保存CSV: {filename}")
        return filename
    def save_to_sqlite(self, table_name=None, db_name=None):
        """保存到SQLite"""
        if not db_name:
            db_name = f"db/{self.name}_{datetime.now().strftime('%Y%m%d')}.db"
        if not table_name:
            table_name = self.name
        conn = sqlite3.connect(db_name)
        cursor = conn.cursor()
        # 动态创建表
        if self.data:
            fields = self.data[0].keys()
            field_types = ', '.join([f"{field} TEXT" for field in fields])
            cursor.execute(f'''
                CREATE TABLE IF NOT EXISTS {table_name} (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    {field_types}
                )
            ''')
            # 插入数据
            placeholders = ', '.join(['?' for _ in fields])
            columns = ', '.join(fields)
            for item in self.data:
                values = [str(item.get(field, '')) for field in fields]
                cursor.execute(f'''
                    INSERT INTO {table_name} ({columns})
                    VALUES ({placeholders})
                ''', values)
            conn.commit()
        conn.close()
        print(f"保存到SQLite: {db_name}")
        return db_name
    def save_all_formats(self):
        """保存所有格式"""
        results = {}
        results['json'] = self.save_to_json()
        results['csv'] = self.save_to_csv()
        results['sqlite'] = self.save_to_sqlite()
        return results
# 使用示例
def complete_crawler_example():
    # 创建爬虫数据保存器
    saver = CrawlerDataSaver('news_articles')
    # 模拟爬取数据
    for i in range(5):
        saver.add_data({
            'title': f'新闻标题{i+1}',
            'content': f'这是第{i+1}条新闻的详细内容...',
            'author': f'作者{i+1}',
            'date': datetime.now().strftime('%Y-%m-%d'),
            'url': f'http://example.com/news/{i+1}'
        })
    # 保存所有格式
    saver.save_all_formats()
    print("\n爬虫数据保存完成！")
complete_crawler_example()

最佳实践建议

1 数据保存配置

import configparser
def save_config_example():
    config = configparser.ConfigParser()
    # 创建配置文件
    config['Database'] = {
        'type': 'sqlite',
        'path': './data',
        'backup': 'True'
    }
    config['Export'] = {
        'formats': 'json,csv,excel',
        'encoding': 'utf-8'
    }
    config['Crawler'] = {
        'delay': '1',
        'user_agent': 'Mozilla/5.0'
    }
    # 保存配置
    with open('config.ini', 'w') as f:
        config.write(f)
# save_config_example()

2 错误处理和重试机制

import time
from functools import wraps
def retry_on_failure(max_retries=3, delay=1):
    """重试装饰器"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_retries - 1:
                        raise e
                    print(f"保存失败，重试中... (尝试 {attempt + 1}/{max_retries})")
                    time.sleep(delay)
            return None
        return wrapper
    return decorator
class SafeDataSaver:
    def __init__(self):
        self.failed_items = []
    @retry_on_failure(max_retries=3)
    def save_item(self, item):
        # 模拟保存操作
        if not item.get('title'):
            raise ValueError("缺少标题")
        # 实际保存代码...
        print(f"保存成功: {item['title']}")
    def save_batch(self, items):
        for item in items:
            try:
                self.save_item(item)
            except Exception as e:
                print(f"保存失败: {e}")
                self.failed_items.append(item)
        # 处理失败的项
        if self.failed_items:
            print(f"有 {len(self.failed_items)} 项保存失败")
            # 保存失败项到单独的文件
            with open('failed_items.json', 'w', encoding='utf-8') as f:
                json.dump(self.failed_items, f, ensure_ascii=False)
# 使用示例
def safe_save_example():
    saver = SafeDataSaver()
    items = [
        {'title': '文章1', 'content': '内容1'},
        {'title': '', 'content': '内容2'},  # 会失败
        {'title': '文章3', 'content': '内容3'}
    ]
    saver.save_batch(items)
safe_save_example()

选择合适的保存方式取决于：