最近帮朋友做了一个小工具：把某图书平台的价格信息定时抓下来，落库存储，再导成表格给他做选品分析。做完之后发现这个需求挺典型的，很多做电商、做选品、做市场调研的朋友都需要类似的东西。索性把这套流程完整整理了，从零开始，不依赖框架，纯手写Python爬虫，把数据存进SQLite，同时导出CSV，整个过程跑通也就两百来行代码。
这个项目适合这些人看：刚入门Python想找个完整练手项目的；已经会写简单爬虫但不知道怎么优雅落地存储的；以及有价格监控、竞品分析、数据采集需求但预算有限不想买商业软件的。我会把每一步的设计思路、代码实现、踩坑点都讲清楚，你看完直接能照着搭一套自己的博文价格情报数据库。
1. 项目设计与技术选型
1.1 这个项目到底要解决什么问题
先聊聊需求本质。所谓“价格情报数据库”，说穿了就三件事：定时采集、历史沉淀、快速检索。采集是爬虫的活，沉淀和检索是数据库的活，缺一不可。
如果你只是临时查一次价格，那用浏览器打开页面人工记录就够了，根本不需要写程序。但当成体系来做就不一样了：几十上百本书的定价、折扣、上下架状态，人工一天都录不完，录完也容易出错。更关键的是，价格是变动的，今天打七折明天可能恢复原价，你手工记一次只能看到当前快照，而数据库可以让你回溯“这个商品过去三十天的价格曲线”——这才是“情报”二字的真正价值。
所以我给这个项目定了三个硬性目标：
- 能抓取指定书籍的商品标题、当前价格、原价、折扣、商品链接，以及抓取时间戳。
- 数据要能长期积累，重复抓取不产生脏数据，最好天然支持增量更新。
- 数据要能方便导出给非技术人员使用，CSV是Excel能直接打开的最通用格式。
围绕这三点，技术方案其实就呼之欲出了。
1.2 为什么不用Scrapy，而选requests加BeautifulSoup
很多人一提到Python爬虫就想到Scrapy框架。我不反对Scrapy，对于大规模分布式爬虫，Scrapy确实成熟稳定。但在这个项目里，我特意没用它，理由有三。
第一，Scrapy的学习曲线对新手不友好。项目里涉及Item Pipeline、Middleware、Selector、Twisted异步引擎，光理解框架的执行流程就得花不少时间。而我们的核心需求就一个页面列表加详情页，用requests发HTTP请求，用BeautifulSoup解析HTML，概念简单清晰，代码也好调试。
第二，requests加BeautifulSoup的组合足够灵活。价格情报这类场景，目标站点通常不多，请求频率也不会高到需要异步IO的程度。同步请求每秒抓个三五页，完全够用。真要以后规模上去了，把核心解析逻辑抽取出来，迁移到Scrapy也只是换层皮的事。
第三，双存储的需求用标准库就能优雅解决。Python自带的csv和sqlite3，一个导出表格，一个落库查询，不需要引入SQLAlchemy或Pandas这些重依赖。整个项目只需要额外安装requests和beautifulsoup4两个包，环境极其干净。
1.3 CSV和SQLite同时用，是不是重复了
不少朋友看这个标题会疑惑：既然都存SQLite了，为什么还要导CSV？这不是脱裤子放屁吗？
还真不是。两种存储的定位完全不同，属于互补关系。
CSV的价值在于“交换”。它是个纯文本的扁平表格，任何操作系统上都能打开，Excel能直接处理，用WPS也能处理。你做数据分析、做报表、把数据发给同事，CVS格式是最低门槛的通行证。缺点是它是个文件，不是数据库，没法做并发查询，也没法高效做条件过滤和聚合统计。
SQLite的价值在于“积累”。它是一个单文件的关系型数据库，支持SQL标准查询，建个索引之后，千百万行数据的查询也是毫秒级。你可以随意写SELECT title, price FROM books WHERE price < 50 ORDER BY price LIMIT 10，这在CSV文件里得靠代码遍历才能实现，麻烦得多。
所以我的建议是：程序运行的核心存储用SQLite，保证数据结构和查询能力；同时每次采集完成后自动导出一份快照CSV，方便人直接打开看。两条腿走路，各干各的，谁都舒服。
2. 核心细节与前置准备
2.1 数据模型设计：字段要精细到什么程度
这是整个项目的地基。字段设计得不好，后面写SQL都难受。
我一开始做过一个粗糙版本，只存了书名、价格、抓取时间三个字段。后来发现根本不够用：同一本书不同页面有不同促销信息，价格旁边还可能有划线原价，链接丢了没法回源，书号没有就没法和已有商品库关联。所以后来我重新设计了表结构，代码如下。
CREATE TABLE IF NOT EXISTS books ( id INTEGER PRIMARY KEY AUTOINCREMENT, book_id TEXT UNIQUE NOT NULL, title TEXT NOT NULL, price REAL NOT NULL, original_price REAL, discount REAL, detail_url TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL );逐个字段说说设计意图。
book_id是业务唯一键，不是自增主键。它用来标识“同一本书”，防止重复入库，后面所有去重逻辑都依赖它。price和original_price为什么要用REAL而不是TEXT？因为只有在SQL层面把价格存成数值类型，你才能跑AVG(price)、MIN(price)这类聚合函数，存成字符串就是给自己挖坑。
created_at和updated_at是审计字段，记录数据的首次入库时间和最后更新时间。后面做价格趋势分析，全靠这两个字段区分不同批次的数据。
顺便提一句，折扣discount这个字段，我建议也存成浮点数，比如0.85表示八五折。因为价格信息页上显示的可能是“85折”这种中文文案，清洗的时候要统一换算。
2.2 抓取规范和反爬意识
既然是爬虫教程，就绕不开反爬这个话题。我的原则很简单：做“有礼貌的爬虫”，遵守robots.txt协议，控制请求频率，不攻击、不绕过、不拖垮目标站点。
怎么理解“有礼貌”？抓取两页之间至少间隔2到5秒，模拟人类浏览的节奏；设置合理的User-Agent请求头，不用默认的Python-requests标识；如果目标站点明确在robots.txt里禁止了某个路径，那就不抓那个路径，换一个数据源或者主动放弃。
别觉得这么做效率低。做价格情报讲究的是持续稳定，你今天暴力抓取把IP封了，明天数据就断档，反而捡了芝麻丢西瓜。再说了，正规网站的页面结构也不是让你高并发爬的，低频小请求配合恰当的解析逻辑，数据质量反而更高。
2.3 环境准备：三步搞定开发环境
以下环境都是在Windows 10测试的，Mac和Linux同理，就是包管理命令略有差异。
Python版本推荐3.9以上。太老版本的Python在类型注解和部分语法上会有兼容性问题，没必要折磨自己。没装Python的话，直接去官网下载安装包，勾选“Add Python to PATH”选项，一路Next就行。
然后创建项目目录，建议用虚拟环境把依赖隔离起来。
mkdir book-price-tracker cd book-price-tracker python -m venv venv在Windows上激活虚拟环境：
venv\Scripts\activate在Mac/Linux上激活虚拟环境：
source venv/bin/activate最后安装两个第三方库：
pip install requests beautifulsoup4搞定。整个环境就这么简单，不需要配置数据库服务，因为SQLite是Python标准库自带的，内置了sqlite3模块，零安装直接import。
3. 从零实现：完整代码实操
3.1 先来搭爬虫骨架：请求与解析
我习惯把代码拆成几个模块各司其职，就算是个小项目也别把几百行代码塞进一个文件里。下面逐步拆解。
项目文件结构：
book-price-tracker/ ├── crawler.py # 核心爬虫逻辑 ├── database.py # SQLite高值化操作 ├── exporter.py # CSV导出 ├── config.py # 全局配置 └── main.py # 主入口先看config.py，把目标地址、请求头、抓取间隔这些配置集中管理，方便后续调整。
# config.py BASE_URL = "https://example-books.com/list?page={}" HEADERS = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " "(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8", "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8", } REQUEST_DELAY = (2, 5) # 随机延迟，防止请求频率过高 DB_PATH = "books.db" CSV_PATH = "books_export.csv"这里的REQUEST_DELAY用元组表示延迟的区间，实际使用random.uniform(2, 5)生成随机秒数。随机化请求间隔比固定间隔更贴近人类行为，也更容易规避基于频率的简单反爬策略。
再看crawler.py中的请求封装。这里我加了三层保护：超时设置、重试机制、状态码检查。
# crawler.py import logging import random import time import requests from bs4 import BeautifulSoup import config logger = logging.getLogger(__name__) def fetch_page(page_num: int, max_retries: int = 3) -> str | None: url = config.BASE_URL.format(page_num) for attempt in range(max_retries): try: resp = requests.get( url, headers=config.HEADERS, timeout=10, ) resp.raise_for_status() resp.encoding = resp.apparent_encoding or resp.encoding return resp.text except requests.RequestException as e: logger.warning("第 %s 页请求失败（%d/3）：%s", page_num, attempt + 1, e) time.sleep(2 ** attempt) # 指数退避，第一次等2秒，第二次等4秒 logger.error("第 %s 页请求超过最大重试次数", page_num) return None几个关键点解释一下。
timeout=10必须设置，不然requests可能一直挂着不动，程序卡死都不知道原因。resp.raise_for_status()会在返回4xx或5xx时直接抛出异常，省得你手动判断状态码。apparent_encoding是requests根据页面内容智能推断的编码，中文网站经常会在charset上标错编码，用这个可以避开乱码问题。
重试时的指数退避思路：第一次失败等2秒，第二次等4秒，第三次等8秒，给目标服务器一个缓冲时间，也避免自己陷入死循环式的重试风暴。
3.2 解析页面：为什么推荐BeautifulSoup而不是正则
拿到HTML源码之后，下一步就是从一堆标签里把书名和价格抠出来。这一步常见的技术方案有正则表达式、XPath、BeautifulSoup、lxml，我推荐用BeautifulSoup加CSS选择器。
正则表达式最大的问题是脆弱。页面稍微改一下标签结构，正则就废了，而且写复杂的正则表达式很费眼神，可读性差。BeautifulSoup能把HTML解析成一棵标签树，你用选择器定位元素，逻辑清晰，容错性也好。
在解析之前，我建议先在浏览器里按F12打开开发者工具，用“选择元素”的小箭头点一下目标位置，确认书名、价格、链接分别在哪个标签的哪一层。下面是一个简化的解析函数，真实项目中选择器需要按你目标站点的实际结构调整。
def parse_books(html: str) -> list[dict]: soup = BeautifulSoup(html, "html.parser") books = [] for item in soup.select("div.book-item"): title_tag = item.select_one("h3.book-title a") if not title_tag: continue # 结构不完整，跳过错漏项 title = title_tag.get_text(strip=True) detail_url = title_tag.get("href", "") book_id = extract_book_id(detail_url) # 从URL里提取唯一ID price_node = item.select_one("span.book-price") original_price_node = item.select_one("span.original-price") discount_node = item.select_one("span.book-discount") price = parse_price(price_node.get_text(strip=True) if price_node else "") original_price = parse_price(original_price_node.get_text(strip=True) if original_price_node else "") discount = parse_discount(discount_node.get_text(strip=True) if discount_node else "") books.append({ "book_id": book_id, "title": title, "price": price, "original_price": original_price, "discount": discount, "detail_url": detail_url, }) return books写解析函数时，最核心的习惯是每个字段都要做空值保护。select_one可能返回None，get_text拿到的字符串可能为空，直接调用下一步处理就会抛AttributeError或者ValueError。我习惯先判断元素是否存在，再决定要不要取值。
extract_book_id这个函数的作用是从商品链接里提取稳定的ID，比如/books/9787123456789，那就把9787123456789截出来。这个ID用来做后续去重，没有稳定ID的话，就只能靠标题文本去重了，那是会出问题的。
3.3 数据处理层：清洗价格和折扣
网页上抓下来的价格长什么样？形形色色都有。
"¥ 88.00""88元""原价：120.00""折扣：7.5折""7.5折"或"75折"- 甚至可能是
"undefined"这样的乱值
所以必须有一层清洗函数，把乱七八糟的文本转成规范的数值。
import re def parse_price(text: str) -> float: """从价格文本中提取浮点数，提取失败返回0.0""" if not text: return 0.0 match = re.search(r"\d+\.?\d*", text.replace(",", "")) return float(match.group()) if match else 0.0 def parse_discount(text: str) -> float: """把'7.5折'、'75折'、'0.75'统一转为0.75格式""" if not text: return 0.0 match = re.search(r"(\d+\.?\d*)", text) if not match: return 0.0 value = float(match.group(1)) if value > 10: # 75折被写成75的情况 value = value / 10 elif value > 1: # 1.5折这种写法 value = value / 10 # 确保最终值域在0到1之间，异常数据直接归零 return round(value, 2) if 0 < value <= 1 else 0.0parse_price为什么要把逗号去掉？因为有些站点千位分隔符写作"1,299.00"，re.search(r"\d+\.?\d*")匹配到1就停了，直接提取出错误结果。先去掉逗号，再提取数字，稳妥得多。
parse_discount里的值域判断是个细节活。不同站点的折扣写法差异很大，75折你没法直接判断它到底意味着0.75还是75。我这里加了一个值域归一化的逻辑：大于10的除以10，大于1的再除以10，最后保证值落在0到1之间，异常数据直接返回0。后续写SQL做price < original_price * discount这类判断时就不会被脏数据带偏。
3.4 存储层之一：CSV导出的正确姿势
CSV导出看似简单，实际有坑，最常见的就是中文乱码。直接用Python的csv模块写中文，然后拿Excel打开，满屏都是“鐢垫棤宓屻�”这种鬼东西。原因在于Excel默认用ANSI编码读取CSV，而Python默认写入的是UTF-8。
解决方案是用utf-8-sig编码写入。
# exporter.py import csv import config def export_to_csv(books: list[dict]) -> str: if not books: return config.CSV_PATH fieldnames = ["book_id", "title", "price", "original_price", "discount", "detail_url", "created_at"] with open(config.CSV_PATH, "w", newline="", encoding="utf-8-sig") as f: writer = csv.DictWriter(f, fieldnames=fieldnames) writer.writeheader() for book in books: writer.writerow(book) return config.CSV_PATHnewline=""是必须传的参数。不传的话，在Windows上每写完一行，文件里会多出一个空行，数据隔行显示，看着非常别扭。这是Python官方文档都明确提示的坑。
另外，fieldnames的顺序就是CSV表头的顺序。我在导出的时候会把数据库查询出来的结果原样传给这个函数，数据库里面哪个字段先出来就按哪个字段排列，所以fieldnames的顺序要和SQL查询的字段顺序保持一致。
如果数据量大，比如一次导出一万条，用csv.DictWriter一条条写也能撑得住。真要导几十万条，建议改用pandas的to_csv，效率会高很多，但这个小项目用标准库就足够了，不用为了快一点去引入一个几百MB的依赖。
3.5 存储层之二：SQLite持久化的细节
SQLite这个模块是Python标准库自带的，无需安装，直接import sqlite3就能用。我用一个类把它包起来，方便管理和复用。
# database.py import sqlite3 import config class BookDatabase: def __init__(self, db_path: str = config.DB_PATH): self.db_path = db_path self.conn = sqlite3.connect(db_path) self.conn.row_factory = sqlite3.Row # 让查询结果支持列名访问 self._init_table() def _init_table(self): with self.conn: self.conn.execute(""" CREATE TABLE IF NOT EXISTS books ( id INTEGER PRIMARY KEY AUTOINCREMENT, book_id TEXT UNIQUE NOT NULL, title TEXT NOT NULL, price REAL NOT NULL, original_price REAL, discount REAL, detail_url TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL ) """)row_factory = sqlite3.Row这个设置很实用。默认情况下查询结果是元组，你需要靠位置下标row[0]、row[1]去取数据，字段一多就眼花。设置成Row后，可以直接用row["title"]这种列名方式访问，代码可读性一下就上来了。
写入数据时，我用了INSERT OR IGNORE配合UNIQUE约束来做去重保护。
def upsert_books(self, books: list[dict]) -> int: """插入或更新书籍数据，返回影响行数""" affected = 0 now = time.strftime("%Y-%m-%d %H:%M:%S") with self.conn: for book in books: cur = self.conn.execute(""" INSERT INTO books (book_id, title, price, original_price, discount, detail_url, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(book_id) DO UPDATE SET title=excluded.title, price=excluded.price, original_price=excluded.original_price, discount=excluded.discount, detail_url=excluded.detail_url, updated_at=excluded.updated_at """, ( book["book_id"], book["title"], book["price"], book["original_price"], book["discount"], book["detail_url"], now, now, )) affected += cur.rowcount return affected这个ON CONFLICT(book_id) DO UPDATE是SQLite的UPSERT语法。它的逻辑是：如果book_id在表中不存在，就执行插入；如果已存在，就更新价格、标题这些可变字段，同时刷新updated_at时间戳。created_at保持不变——第一次入库的时间永远留下来，这就是历史追踪的基础。
affected变量累加每次操作影响的行数，返回出去方便后面打日志判断这次抓取是新增多还是更新多。
再说两个数据库层面的性能优化细节。
第一个是用with self.conn:管理事务。with退出时自动提交事务，如果在with块内抛出异常，事务自动回滚，数据不会出现写了一半的脏状态。比自己手动conn.commit()安全得多。
第二是查询场景增加索引。如果以后要频繁按价格排序、按更新时间筛选，可以加索引：
CREATE INDEX IF NOT EXISTS idx_books_price ON books(price); CREATE INDEX IF NOT EXISTS idx_books_updated ON books(updated_at);索引不是越多越好，写操作会变慢，所以只给高频查询字段建索引就够了。
3.6 主流程编排与增量更新实现
每个模块写好了，主流程其实就是把它们串起来。逻辑不复杂，我直接贴main.py。
# main.py import logging import random import time import config from crawler import fetch_page, parse_books from database import BookDatabase from exporter import export_to_csv def main(): logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", ) logger = logging.getLogger(__name__) db = BookDatabase() all_books = [] for page_num in range(1, 6): # 先抓前5页做演示 logger.info("开始抓取第 %s 页", page_num) html = fetch_page(page_num) if html is None: logger.warning("第 %s 页抓取失败，跳过", page_num) continue books = parse_books(html) logger.info("第 %s 页解析到 %s 条数据", page_num, len(books)) if not books: continue affected = db.upsert_books(books) logger.info("入库影响 %s 条记录", affected) all_books.extend(books) time.sleep(random.uniform(*config.REQUEST_DELAY)) export_to_csv(all_books) logger.info("CSV 导出完成，共 %s 条数据", len(all_books)) logger.info("全流程结束") if __name__ == "__main__": main()这段主流程有几个细节值得说。
我在每次请求之后用time.sleep(random.uniform(*config.REQUEST_DELAY))控制节奏。这个随机延迟区间在config.py里配置成了2到5秒，实际抓5页大概需要十几秒，完全在可接受范围内。如果你在运行过程中被反爬拦截了，把间隔调大到5到10秒再试试。
all_books列表在内存里逐渐累积所有页的数据，最后一次性传给export_to_csv。这里有个小问题：如果抓10万条数据，全部累积在内存里可能占用几百MB内存。更好的做法是不保存在内存，而是每次从数据库里SELECT出来再导出。想简单的话，可以改一下导出逻辑，直接从数据库读：
def export_from_db(db: BookDatabase, path: str) -> int: rows = db.conn.execute("SELECT * FROM books ORDER BY updated_at DESC").fetchall() books = [dict(row) for row in rows] export_to_csv(books) return len(books)这样导出的永远是全量数据，且不占额外内存，我实际项目中用的就是这个方案。
增量更新的逻辑其实已经内含在upsert_books里了。第一次运行，所有书籍都是新增；第二次运行相同的book_id再出现，就走更新分支，价格变动被记录下来，updated_at被刷新。想要查看某本书的价格历史，只需要在表里增加一个price_history表，每次更新前把旧价格插入历史表即可。这个扩展方案我在文章结尾再细说。
4. 常见问题与排查技巧实录
4.1 请求总是被403拦截，怎么办
这是爬虫会遇到频率最高的问题。403意味着服务器识别出你是程序而不是真人，拒绝了请求。
排查顺序我建议先做四步检查。
第一步，检查User-Agent。requests默认的python-requests/x.x.x标识太明显了，在headers里换成浏览器的UA基本能解决六成问题。
第二步，检查请求频率。把两次请求的间隔加到5秒以上，再加一个随机抖动。很多站点的反爬阈值就是“固定时间内的请求次数”，你把速率降下来就能绕过去。
第三步，检查Cookie。有些站点第一次请求会下发Cookie，后续请求要带上才能继续访问。用requests.Session()会自动维护Cookie，建议始终用Session对象发请求。
第四步，检查页面是否需要登录才能看到价格。如果需要登录，那就得加入登录流程。具体做法是先用requests模拟提交用户名密码获取登录态Cookie，再带着这个Cookie去请求商品页面。
如果以上四步都做了还被封，大概率是你的IP被临时拉黑了。此时只有两个选择：等一段时间自动解封，或者换代理IP。代理这块水很深，免费代理大多不稳定，收费代理需要额外成本，小项目建议直接降低抓取频率，等封禁解除再继续。
4.2 页面抓下来了，但解析出来全是空列表
这个问题通常出现在CSS选择器失效的场景。最典型的原因是目标站点改版了，HTML结构变了。你周一写的选择器，周三就不灵了。
排查方法很简单：把抓到的HTML存成一个文件，放在项目目录里，然后用BeautifulSoup单独跑一个小脚本反复调试选择器。
# debug_selector.py with open("debug.html", "r", encoding="utf-8") as f: html = f.read() from bs4 import BeautifulSoup soup = BeautifulSoup(html, "html.parser") # 反复试不同的选择器 items = soup.select("div.book-item") print("数量:", len(items)) if items: print(items[0].select_one("h3.book-title a"))把网页结构变了这个问题排除掉之后，再检查你的选择器匹配到的子节点是否存在。比如h3.book-title a匹配不到，可能是书名不是写在a标签里而是写在了span里，改一下选择器就好。
另外，动态加载也要注意。如果商品列表是通过JavaScript异步加载的，直接requests.get拿到的HTML里根本没有商品条目，那再怎么解析都是空的。这时候你只有两个选择：找页面里是否有能直接返回JSON数据的API接口，或者换用Selenium这类可以执行JS的自动化工具。优先找API接口，速度快、效率高，Selenium是最后的兜底方案。
4.3 数据库里出现重复数据，怎么避免
如果你严格按照我给出的建表语句，设置了book_id TEXT UNIQUE NOT NULL，那重复数据基本进不来，ON CONFLICT会把重复操作转成更新。
但有一种情况会导致重复：你的book_id提取逻辑不稳定。第一次抓取时extract_book_id从URL里截取了9787123456789，第二次页面改版URL里没有这个ID了，extract_book_id返回值变了，数据库就认为这是两条不同的数据。
所以，你在开发完解析逻辑之后，一定要单独验证book_id的稳定性。可以把同一个商品抓三遍，打印出三次的book_id，确认完全一致再放心去跑全量。如果同一商品在不同场景下的URL前缀不同、参数不同，那就需要对URL做归一化处理，只截取关键的ID部分。
已经产生了重复数据怎么办？可以用下面的SQL清理：
DELETE FROM books WHERE id NOT IN ( SELECT MIN(id) FROM books GROUP BY book_id );这条SQL的作用是保留每个book_id对应最小id的那条记录，把其余重复的删掉。执行前记得先备份数据库文件。
4.4 SQLite在运行时报“database is locked”
SQLite是轻量级数据库，但它的并发能力有限。如果你的爬虫一边往数据库里写数据，另一边有人用SQLite浏览器打开了同一个数据库文件进行查看，写入操作就可能报database is locked。
解决思路有三个。
一是给连接加超时参数。sqlite3.connect(db_path, timeout=10)，这样当数据库被锁时会等待10秒而不是直接报错。
二是开启WAL模式。SQLite的WAL（Write-Ahead Logging）模式允许读写并发执行，读操作不阻塞写操作，写操作也不阻塞读操作。在初始化数据库后执行：
self.conn.execute("PRAGMA journal_mode=WAL")三是批量提交而不是单条提交。如果你一次抓了1000条数据，不要每插入一条就commit一次，放到一个事务里批量提交，锁冲突的概率会大幅降低。
还有一个我不太建议但你可能会用到的做法：多进程同时写入同一个SQLite文件。SQLite对多进程并发写入的限制比较严格，两个进程同时写很容易互相锁住。如果真有这个需求，应该改成“子进程只负责爬取和解析，把数据通过队列交给主进程统一入库”，不要多个进程直接各写各的。
4.5 CSV文件中文乱码问题完整记录
这个问题我调试过很多次，总结出血泪经验：用utf-8-sig编码写CSV是最稳妥的解决方案，它的原理是在文件开头多写入一个BOM头，Excel识别到BOM会用UTF-8解码，就不会乱码了。
如果你是老代码用encoding="utf-8"写的CSV已经乱码了，还有一个抢救办法：用记事本打开CSV文件，另存为时把编码改成“带BOM的UTF-8”，保存后再用Excel打开就好了。不过已经乱码的数据可能救不回来，因为编码错乱是不可逆的，只能重新导出。
另外，CSV里如果某个字段本身包含逗号、换行符或双引号，直接用逗号拼接字符串就会把表格结构弄坏。用csv.DictWriter写数据时它内部会自动处理这些特殊字符的转义，所以你永远应该用csv模块写CSV，别自己手动拼字符串。
5. 我的实际使用体验和几个扩展建议
这个项目写完到现在，我在本地挂了一个定时任务，每天早上九点自动跑一次，把某个图书平台几个分类页的数据抓下来存库。跑了一个多月，数据库里积累了几万条价格历史记录，查起来依然很快。
实际使用中有两个体会特别深。
第一个是“懒人逻辑”的价值。当初在设计时坚持用UNIQUE约束加UPSERT，虽然写的时候多花了点心思，但之后的日常维护省了太多事。每天定时跑任务，不需要担心重复数据，不需要手工清理，跑完看日志就行。
第二个是“异常处理要留足冗余”。一开始我写的代码重试次数只有两次，超时时间只有5秒。结果有一次目标网站做活动，响应速度变慢，少量请求超时导致数据缺漏，当天导出的CSV里少了十几条记录。后来改成超时10秒、重试3次并且指数退避，这类问题再没出现过。爬虫这东西，跑得慢不可怕，跑漏了才是真麻烦。
如果你想把这套项目进一步扩展，我建议按优先级尝试三个方向。
一是增加价格历史表，记录每次抓取的价格快照。当前设计方案只保留了最新价格，查询“某本书过去三十天的价格变化”还做不到。加一张price_history表，在upsert_books执行更新前把旧价格写入历史表，就能做完整的趋势分析。
二是增加邮件或IM通知。当某本书的价格低于你设定的阈值时，自动发送通知。实现方式就是在upsert_books之后加一个判断，把价格低于阈值的书筛选出来，走SMTP发邮件。
三是增加调度逻辑。当前是交给系统的计划任务（Windows的“任务计划程序”或Linux的cron）来定时运行的。如果你不喜欢依赖系统调度，可以用APScheduler库在Python进程内写定时任务，配置文件里指定运行频率。
这套代码的核心价值不在于爬虫本身，而在于它把“采集、清洗、存储、导出”四个环节完整串了起来，让你在拿到原始数据后能真正组合出有价值的情报。数据这东西，抓下来只是第一步，沉淀下来并持续追踪，才是真正的财富。
