尧图网站建设 
最近帮朋友做了一个小工具把某图书平台的价格信息定时抓下来落库存储再导成表格给他做选品分析。做完之后发现这个需求挺典型的很多做电商、做选品、做市场调研的朋友都需要类似的东西。索性把这套流程完整整理了从零开始不依赖框架纯手写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/xhtmlxml,application/xml;q0.9,image/webp,*/*;q0.8, Accept-Language: zh-CN,zh;q0.9,en;q0.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, headersconfig.HEADERS, timeout10, ) 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几个关键点解释一下。timeout10必须设置不然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(stripTrue) 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(stripTrue) if price_node else ) original_price parse_price(original_price_node.get_text(stripTrue) if original_price_node else ) discount parse_discount(discount_node.get_text(stripTrue) 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可能返回Noneget_text拿到的字符串可能为空直接调用下一步处理就会抛AttributeError或者ValueError。我习惯先判断元素是否存在再决定要不要取值。extract_book_id这个函数的作用是从商品链接里提取稳定的ID比如/books/9787123456789那就把9787123456789截出来。这个ID用来做后续去重没有稳定ID的话就只能靠标题文本去重了那是会出问题的。3.3 数据处理层清洗价格和折扣网页上抓下来的价格长什么样形形色色都有。¥ 88.0088元原价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.00re.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, encodingutf-8-sig) as f: writer csv.DictWriter(f, fieldnamesfieldnames) 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 titleexcluded.title, priceexcluded.price, original_priceexcluded.original_price, discountexcluded.discount, detail_urlexcluded.detail_url, updated_atexcluded.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( levellogging.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, encodingutf-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, timeout10)这样当数据库被锁时会等待10秒而不是直接报错。二是开启WAL模式。SQLite的WALWrite-Ahead Logging模式允许读写并发执行读操作不阻塞写操作写操作也不阻塞读操作。在初始化数据库后执行self.conn.execute(PRAGMA journal_modeWAL)三是批量提交而不是单条提交。如果你一次抓了1000条数据不要每插入一条就commit一次放到一个事务里批量提交锁冲突的概率会大幅降低。还有一个我不太建议但你可能会用到的做法多进程同时写入同一个SQLite文件。SQLite对多进程并发写入的限制比较严格两个进程同时写很容易互相锁住。如果真有这个需求应该改成“子进程只负责爬取和解析把数据通过队列交给主进程统一入库”不要多个进程直接各写各的。4.5 CSV文件中文乱码问题完整记录这个问题我调试过很多次总结出血泪经验用utf-8-sig编码写CSV是最稳妥的解决方案它的原理是在文件开头多写入一个BOM头Excel识别到BOM会用UTF-8解码就不会乱码了。如果你是老代码用encodingutf-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进程内写定时任务配置文件里指定运行频率。这套代码的核心价值不在于爬虫本身而在于它把“采集、清洗、存储、导出”四个环节完整串了起来让你在拿到原始数据后能真正组合出有价值的情报。数据这东西抓下来只是第一步沉淀下来并持续追踪才是真正的财富。 纯手写Python爬虫：SQLite+CSV构建图书价格情报数据库
最近帮朋友做了一个小工具把某图书平台的价格信息定时抓下来落库存储再导成表格给他做选品分析。做完之后发现这个需求挺典型的很多做电商、做选品、做市场调研的朋友都需要类似的东西。索性把这套流程完整整理了从零开始不依赖框架纯手写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/xhtmlxml,application/xml;q0.9,image/webp,*/*;q0.8, Accept-Language: zh-CN,zh;q0.9,en;q0.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, headersconfig.HEADERS, timeout10, ) 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几个关键点解释一下。timeout10必须设置不然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(stripTrue) 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(stripTrue) if price_node else ) original_price parse_price(original_price_node.get_text(stripTrue) if original_price_node else ) discount parse_discount(discount_node.get_text(stripTrue) 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可能返回Noneget_text拿到的字符串可能为空直接调用下一步处理就会抛AttributeError或者ValueError。我习惯先判断元素是否存在再决定要不要取值。extract_book_id这个函数的作用是从商品链接里提取稳定的ID比如/books/9787123456789那就把9787123456789截出来。这个ID用来做后续去重没有稳定ID的话就只能靠标题文本去重了那是会出问题的。3.3 数据处理层清洗价格和折扣网页上抓下来的价格长什么样形形色色都有。¥ 88.0088元原价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.00re.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, encodingutf-8-sig) as f: writer csv.DictWriter(f, fieldnamesfieldnames) 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 titleexcluded.title, priceexcluded.price, original_priceexcluded.original_price, discountexcluded.discount, detail_urlexcluded.detail_url, updated_atexcluded.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( levellogging.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, encodingutf-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, timeout10)这样当数据库被锁时会等待10秒而不是直接报错。二是开启WAL模式。SQLite的WALWrite-Ahead Logging模式允许读写并发执行读操作不阻塞写操作写操作也不阻塞读操作。在初始化数据库后执行self.conn.execute(PRAGMA journal_modeWAL)三是批量提交而不是单条提交。如果你一次抓了1000条数据不要每插入一条就commit一次放到一个事务里批量提交锁冲突的概率会大幅降低。还有一个我不太建议但你可能会用到的做法多进程同时写入同一个SQLite文件。SQLite对多进程并发写入的限制比较严格两个进程同时写很容易互相锁住。如果真有这个需求应该改成“子进程只负责爬取和解析把数据通过队列交给主进程统一入库”不要多个进程直接各写各的。4.5 CSV文件中文乱码问题完整记录这个问题我调试过很多次总结出血泪经验用utf-8-sig编码写CSV是最稳妥的解决方案它的原理是在文件开头多写入一个BOM头Excel识别到BOM会用UTF-8解码就不会乱码了。如果你是老代码用encodingutf-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进程内写定时任务配置文件里指定运行频率。这套代码的核心价值不在于爬虫本身而在于它把“采集、清洗、存储、导出”四个环节完整串了起来让你在拿到原始数据后能真正组合出有价值的情报。数据这东西抓下来只是第一步沉淀下来并持续追踪才是真正的财富。