从零到接单 11：数据存储——CSV、Excel、JSON 与 SQLite
pythontutorial爬虫自动化接单数据存储CSVExcelSQLite
系列目录：本文是「从零到接单：Python 自动化与爬虫实战」系列的第 11 篇。前面的文章我们已经能把网页数据抓下来了，但数据散在内存里，关了程序就没了。这篇我们系统学习如何把数据存储为 CSV、Excel、JSON 和 SQLite 数据库——这些是接单交付时客户最常要的格式。
一、选择哪种存储？
| 格式 | 适合场景 | 优点 | 缺点 | |------|----------|------|------| | CSV | 表格数据、数据交换 | 通用、简单、Excel 能打开 | 不支持多 sheet、嵌套数据 | | Excel | 需要格式、多 sheet、图表 | 所见即所得、客户最爱 | 库较重、大文件慢 | | JSON | API 返回数据、配置、嵌套数据 | 灵活、支持嵌套 | 非技术人员不熟悉 | | SQLite | 大量数据、需要查询 | 零配置、支持 SQL | 需要 SQL 知识 |
二、CSV：最通用的表格格式
基础写入
import csv
data = [
{"姓名": "张三", "年龄": 25, "城市": "北京", "薪资": 15000},
{"姓名": "李四", "年龄": 30, "城市": "上海", "薪资": 20000},
{"姓名": "王五", "年龄": 28, "城市": "广州", "薪资": 18000},
]
# 写入
with open("employees.csv", "w", newline="", encoding="utf-8-sig") as f:
writer = csv.DictWriter(f, fieldnames=["姓名", "年龄", "城市", "薪资"])
writer.writeheader()
writer.writerows(data)
# 逐行写入
with open("employees.csv", "w", newline="", encoding="utf-8-sig") as f:
writer = csv.DictWriter(f, fieldnames=["姓名", "年龄", "城市", "薪资"])
writer.writeheader()
for row in data:
writer.writerow(row)
print(f"写入：{row['姓名']}")
# 追加写入
with open("employees.csv", "a", newline="", encoding="utf-8-sig") as f:
writer = csv.DictWriter(f, fieldnames=["姓名", "年龄", "城市", "薪资"])
writer.writerow({"姓名": "赵六", "年龄": 35, "城市": "深圳", "薪资": 25000})
💡
encoding="utf-8-sig"让 Excel 打开 CSV 时不乱码（加 BOM 头）。
读取
# 字典方式读取
with open("employees.csv", "r", encoding="utf-8-sig") as f:
reader = csv.DictReader(f)
for row in reader:
print(f"{row['姓名']} - {row['薪资']}元")
# 转为列表
with open("employees.csv", "r", encoding="utf-8-sig") as f:
reader = csv.DictReader(f)
data = list(reader)
# 数据分析
salaries = [int(row["薪资"]) for row in data]
print(f"平均薪资：{sum(salaries) / len(salaries):.0f}元")
print(f"最高薪资：{max(salaries)}元")
print(f"最低薪资：{min(salaries)}元")
原生 csv 模块 vs pandas
# 如果没有 pandas 或数据量小，用原生 csv
import csv
# 如果装了 pandas 且数据量大，用 pandas
# import pandas as pd
# df = pd.read_csv("employees.csv")
# df.to_excel("employees.xlsx", index=False)
三、Excel：客户最爱的格式
openpyxl：读写 .xlsx
pip install openpyxl
from openpyxl import Workbook
from openpyxl.styles import Font, Alignment, PatternFill, Border, Side
from openpyxl.utils import get_column_letter
# 创建 Excel 文件
wb = Workbook()
ws = wb.active
ws.title = "员工信息"
# 表头
headers = ["姓名", "年龄", "城市", "薪资"]
for col, header in enumerate(headers, 1):
cell = ws.cell(row=1, column=col, value=header)
cell.font = Font(bold=True, color="FFFFFF")
cell.fill = PatternFill(start_color="4472C4", end_color="4472C4", fill_type="solid")
cell.alignment = Alignment(horizontal="center")
# 写入数据
data = [
["张三", 25, "北京", 15000],
["李四", 30, "上海", 20000],
["王五", 28, "广州", 18000],
["赵六", 35, "深圳", 25000],
]
for row_idx, row_data in enumerate(data, 2):
for col_idx, value in enumerate(row_data, 1):
ws.cell(row=row_idx, column=col_idx, value=value)
# 添加公式
ws.cell(row=len(data) + 3, column=1, value="平均薪资")
ws.cell(row=len(data) + 3, column=4, value=f"=AVERAGE(D2:D{len(data)+1})")
# 调整列宽
for col in range(1, len(headers) + 1):
ws.column_dimensions[get_column_letter(col)].width = 15
# 保存
wb.save("employees.xlsx")
print("Excel 文件已生成 ✅")
读取 Excel
from openpyxl import load_workbook
wb = load_workbook("employees.xlsx")
ws = wb.active
# 逐行读取
for row in ws.iter_rows(min_row=2, values_only=True):
name, age, city, salary = row
print(f"{name} | {age}岁 | {city} | {salary}元")
# 转为字典列表
headers = [cell.value for cell in ws[1]]
data = []
for row in ws.iter_rows(min_row=2, values_only=True):
data.append(dict(zip(headers, row)))
多 Sheet 工作簿
wb = Workbook()
# Sheet 1：汇总
ws1 = wb.active
ws1.title = "汇总"
ws1.append(["日期", "抓取数量", "成功", "失败"])
ws1.append(["2026-07-16", 500, 485, 15])
# Sheet 2：详细数据
ws2 = wb.create_sheet("详细数据")
ws2.append(["标题", "价格", "链接", "抓取时间"])
# Web 开发 Sheet
ws3 = wb.create_sheet("Web 开发")
ws4 = wb.create_sheet("移动开发")
wb.save("report.xlsx")
四、JSON：最灵活的数据交换格式
前面第 06 篇已经介绍了 JSON 基础操作，这里补充进阶用法：
import json
# ── 处理复杂嵌套数据 ──
crawl_result = {
"meta": {
"source": "example.com",
"crawl_time": "2026-07-16 15:30:00",
"total": 1200
},
"items": [
{
"title": "Python 爬虫入门",
"url": "https://example.com/1",
"tags": ["python", "爬虫", "入门"],
"price": None
},
# ... 更多条目
]
}
# 美化写入
with open("result.json", "w", encoding="utf-8") as f:
json.dump(crawl_result, f, ensure_ascii=False, indent=2)
# ── 自定义 JSON 编码器 ──
from datetime import datetime
class CustomEncoder(json.JSONEncoder):
def default(self, obj):
if isinstance(obj, datetime):
return obj.isoformat()
return super().default(obj)
data = {"time": datetime.now(), "value": 42}
json_str = json.dumps(data, cls=CustomEncoder)
print(json_str) # {"time": "2026-07-16T15:30:00", "value": 42}
# ── 大 JSON 流式读取 ──
# 逐条读取（适合非常大的 JSON 数组文件）
def read_json_lines(filepath):
"""读取 JSON Lines 格式（每行一条 JSON）"""
with open(filepath, "r", encoding="utf-8") as f:
for line in f:
yield json.loads(line)
# 写入 JSON Lines
with open("data.jsonl", "w", encoding="utf-8") as f:
for item in crawl_result["items"]:
f.write(json.dumps(item, ensure_ascii=False) + "\n")
五、SQLite：零配置数据库
SQLite 不需要安装服务器，一个文件就是一个数据库。适合中小型数据。
基础操作
import sqlite3
# 连接数据库（文件不存在则自动创建）
conn = sqlite3.connect("crawler.db")
cursor = conn.cursor()
# 创建表
cursor.execute("""
CREATE TABLE IF NOT EXISTS articles (
id INTEGER PRIMARY KEY AUTOINCREMENT,
title TEXT NOT NULL,
url TEXT UNIQUE,
author TEXT,
publish_date TEXT,
views INTEGER DEFAULT 0,
crawl_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
# 插入数据
cursor.execute("""
INSERT INTO articles (title, url, author, publish_date, views)
VALUES (?, ?, ?, ?, ?)
""", ("Python 爬虫入门", "https://example.com/1", "张三", "2026-07-01", 1280))
# 批量插入
articles = [
("自动化脚本实战", "https://example.com/2", "李四", "2026-07-05", 2560),
("RPA 入门指南", "https://example.com/3", "张三", "2026-07-10", 960),
]
cursor.executemany("""
INSERT OR IGNORE INTO articles (title, url, author, publish_date, views)
VALUES (?, ?, ?, ?, ?)
""", articles)
conn.commit()
查询
# 查询所有
cursor.execute("SELECT * FROM articles")
for row in cursor.fetchall():
print(row)
# 条件查询
cursor.execute("SELECT title, views FROM articles WHERE views > ?", (1000,))
for row in cursor.fetchall():
print(f"{row[0]} - {row[1]}阅读")
# 聚合查询
cursor.execute("""
SELECT author, COUNT(*) as count, SUM(views) as total_views
FROM articles
GROUP BY author
""")
for author, count, total in cursor.fetchall():
print(f"{author}: {count}篇, 总阅读{total}")
# 分页查询
page = 1
page_size = 10
offset = (page - 1) * page_size
cursor.execute(f"SELECT * FROM articles LIMIT {page_size} OFFSET {offset}")
# 字典形式读取（推荐）
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
cursor.execute("SELECT * FROM articles")
for row in cursor.fetchall():
print(dict(row))
conn.close()
爬虫专用存储类
import sqlite3
from datetime import datetime
class CrawlerDB:
"""爬虫数据库工具"""
def __init__(self, db_path="crawler.db"):
self.conn = sqlite3.connect(db_path)
self.conn.row_factory = sqlite3.Row
self._init_tables()
def _init_tables(self):
self.conn.execute("""
CREATE TABLE IF NOT EXISTS items (
id INTEGER PRIMARY KEY AUTOINCREMENT,
source_url TEXT,
title TEXT,
content TEXT,
price REAL,
category TEXT,
crawl_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
UNIQUE(source_url, title)
)
""")
self.conn.execute("""
CREATE TABLE IF NOT EXISTS crawl_log (
id INTEGER PRIMARY KEY AUTOINCREMENT,
url TEXT,
status TEXT,
items_count INTEGER,
error TEXT,
start_time TIMESTAMP,
end_time TIMESTAMP
)
""")
self.conn.commit()
def insert_items(self, items):
cursor = self.conn.cursor()
count = 0
for item in items:
try:
cursor.execute("""
INSERT OR IGNORE INTO items (source_url, title, content, price, category)
VALUES (?, ?, ?, ?, ?)
""", (item.get("source_url"), item.get("title"),
item.get("content"), item.get("price"), item.get("category")))
if cursor.rowcount > 0:
count += 1
except sqlite3.Error as e:
print(f"插入失败：{e}")
self.conn.commit()
return count
def log_crawl(self, url, status, items_count, error=None):
self.conn.execute("""
INSERT INTO crawl_log (url, status, items_count, error, start_time, end_time)
VALUES (?, ?, ?, ?, ?, ?)
""", (url, status, items_count, error, datetime.now(), datetime.now()))
self.conn.commit()
def get_stats(self):
cursor = self.conn.cursor()
cursor.execute("SELECT COUNT(*) FROM items")
total_items = cursor.fetchone()[0]
cursor.execute("SELECT COUNT(DISTINCT source_url) FROM items")
total_sources = cursor.fetchone()[0]
return {"total_items": total_items, "total_sources": total_sources}
def export_to_csv(self, filename="export.csv"):
import csv
cursor = self.conn.cursor()
cursor.execute("SELECT * FROM items")
rows = cursor.fetchall()
if not rows:
return 0
headers = rows[0].keys()
with open(filename, "w", newline="", encoding="utf-8-sig") as f:
writer = csv.DictWriter(f, fieldnames=headers)
writer.writeheader()
writer.writerows([dict(r) for r in rows])
return len(rows)
def close(self):
self.conn.close()
六、数据存储选择决策流程图
数据量 < 1 万条？
├─ 是 → 客户要求 Excel？
│ ├─ 是 → openpyxl 写 .xlsx
│ └─ 否 → CSV（够简单、通用）
└─ 否 → SQLite（支持 SQL 查询）
└─ 数据量 > 100 万？→ MySQL / PostgreSQL
总结
| 存储方式 | 核心库 | 写操作 | 读操作 |
|---------|--------|--------|--------|
| CSV | csv | csv.DictWriter | csv.DictReader |
| Excel | openpyxl | Workbook() → cell.value = | load_workbook() → iter_rows() |
| JSON | json | json.dump() | json.load() |
| SQLite | sqlite3 | cursor.execute("INSERT...") | cursor.execute("SELECT...") |
数据存储是交付的"最后一公里"。下一篇我们进入办公自动化——用 openpyxl、python-docx 等库处理 Word/Excel/PDF，这是自动化接单的核心利润点。
练习：改造第 09 篇的 BlogCrawler，让它在爬取文章列表后，同时保存为 CSV 和 SQLite 两种格式。SQLite 中记录爬取日志（URL、抓取时间、成功条数）。
Comments
Sign in to leave a comment.