1. 项目缘起与整体设计思路做网络小说数据采集这件事我在不同阶段反复折腾过好几轮。最早是想给自己做一个离线书架把喜欢的小说按分类、按字数、按更新时间整理成一张表方便挑书后来帮朋友做阅读偏好分析又需要批量拉取榜单数据做统计。几次下来我总结出一套相对稳定的采集思路这里就以一个模拟的“小说站点采集项目”为例把整套流程拆开讲清楚。需要先说明的是本文讨论的是公开可访问的书籍元信息采集也就是书名、作者、分类、字数、更新时间、简介这类展示在列表页和详情页上的公开数据。采集行为本身要遵守目标站点的服务条款与相关规范控制请求频率、不采集需要登录才能访问的内容、不绕过任何访问限制这是做任何数据采集工作的前提。我下面讲的所有技巧都建立在这个前提之上。这个项目能解决什么问题简单说就是把人工翻页、复制粘贴的重复劳动变成一段可复用的脚本。你给它一个榜单入口它自动翻页、解析、去重、落库最后输出一张干净的结构化表格。适合谁来参考有一定 Python 基础、想入门数据采集的开发者需要做书籍数据分析的运营或研究者以及想给自己搭一个本地书库索引的普通读者。哪怕你之前没写过采集脚本跟着思路走一遍也能上手。整体设计上我把它拆成四层请求层、解析层、存储层、调度层。请求层负责发请求、控频、重试解析层负责从 HTML 里把结构化字段抠出来存储层负责去重和落库调度层负责翻页和任务编排。这样分层的好处是任何一层出问题都能单独替换——比如站点改版了只需要改解析层想换存储介质只动存储层就行。下面逐层展开。2. 请求层把“稳定拿回页面”这件事做扎实2.1 请求头与身份伪装的基本逻辑很多人写采集脚本第一步就卡在“请求被拒”。原因往往很简单默认的请求头里 User-Agent 是 python-requests 之类的标识服务器一眼就认出来这不是普通浏览器。所以第一件事是构造一套看起来正常的请求头。import requests 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,*/*;q0.8, Accept-Language: zh-CN,zh;q0.9, Connection: keep-alive, }这里有个细节值得说Accept-Language 要设成中文。有些站点会根据语言返回不同版本的页面如果你用默认的英文头可能拿到的是精简版或者跳转页解析规则就对不上了。我踩过这个坑明明本地浏览器打开结构正常脚本抓回来却少了一大块排查半天才发现是语言头的问题。注意伪装请求头只是为了拿到和普通浏览器一致的页面不是用来绕过任何访问控制。如果站点明确要求登录或限制访问应当尊重其规则。2.2 超时、重试与连接复用请求层最容易忽视的是超时设置。默认情况下 requests 是不设超时的一旦某个请求卡住整个脚本就挂在那里。我的习惯是连接超时 5 秒、读取超时 10 秒配合重试机制。from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry def build_session(): session requests.Session() retry Retry( total3, backoff_factor0.8, status_forcelist[429, 500, 502, 503, 504], allowed_methods[GET], ) adapter HTTPAdapter(max_retriesretry, pool_connections10, pool_maxsize10) session.mount(http://, adapter) session.mount(https://, adapter) session.headers.update(HEADERS) return sessionbackoff_factor0.8意味着第一次重试等 0.8 秒第二次 1.6 秒第三次 3.2 秒指数退避。这个参数不要设太小否则遇到限流时重试反而加重服务器负担也不要太大否则整体采集速度被拖垮。0.5 到 1.0 之间是我实测比较舒服的区间。pool_connections和pool_maxsize控制连接池大小。单线程采集设成 10 足够多线程的话要相应调大否则会出现连接排队等待的情况反而比单线程还慢。2.3 请求频率控制别把人家服务器打趴下这是我最想强调的一点。采集脚本写得再漂亮如果请求频率失控轻则被封 IP重则给目标站点造成实际压力这是很不负责任的做法。我的做法是固定间隔 随机抖动。import time import random def polite_sleep(base1.2, jitter0.6): time.sleep(base random.uniform(0, jitter))固定间隔保证平均速率可控随机抖动避免请求呈现明显的机器规律。base 设多少合适我的经验是不低于 1 秒如果站点响应较慢或者页面较大调到 2 到 3 秒更稳妥。你可能觉得慢但采集几百页也就多花几分钟换来的是稳定不被中断这笔账很划算。提示如果目标站点提供了公开的数据接口或榜单文件优先用那些比逐页抓 HTML 高效得多也更友好。3. 解析层从 HTML 里精准抠出结构化字段3.1 解析库选型lxml 还是 BeautifulSoup解析 HTML 主流有两个选择lxml和BeautifulSoup。我的建议是用 lxml 做主力BeautifulSoup 做兜底。原因很直接lxml 基于 C 实现解析速度快支持 XPath定位精准BeautifulSoup 胜在容错性强遇到结构混乱的页面不容易崩。from lxml import etree def parse_list_page(html): tree etree.HTML(html) items tree.xpath(//div[classbook-list]//li) results [] for item in items: title item.xpath(.//h3/a/text()) author item.xpath(.//p[classauthor]/text()) results.append({ title: title[0].strip() if title else , author: author[0].strip() if author else , }) return results这里有个关键习惯每个字段都要做空值判断。列表页里经常有广告位、推荐位混在里面它们的结构可能和正常书籍项相似但字段缺失。如果不判断脚本跑到某一页就会因为IndexError崩掉。我早期写的脚本就是没做这个采到第 37 页突然报错前面全白跑了。3.2 XPath 定位的实战技巧XPath 写得好不好直接决定解析的稳定性。几个我常用的技巧优先用语义化的 class 或 id 定位比如//div[idcontent]比//div[3]/div[2]这种纯位置定位稳得多。位置定位在页面加一个广告位之后就全乱了。用contains处理多 class 情况比如//div[contains(class, book-item)]因为一个元素往往有多个 class精确匹配容易漏。文本提取用text()配合strip()因为 HTML 里的换行和空格会混进结果。# 处理多 class 和嵌套文本 title item.xpath(string(.//h3[contains(class,title)])).strip()string()函数会把节点下所有文本拼起来比text()更适合处理标题里带span标签的情况。3.3 详情页字段的补充采集列表页通常只有书名、作者、分类字数、更新时间、简介这些要去详情页拿。这就涉及二级页面采集。我的做法是列表页先收集详情页链接再逐个请求详情页。def parse_detail_page(html): tree etree.HTML(html) def pick(xp): r tree.xpath(xp) return r[0].strip() if r else return { word_count: pick(//div[classinfo]//span[contains(text(),字数)]/text()), update_time: pick(//div[classinfo]//span[contains(text(),更新)]/text()), intro: pick(//div[classintro]//text()), }详情页采集的请求量是列表页的 N 倍所以频率控制要更保守。我一般会把详情页请求间隔设到 2 秒以上宁可慢一点。4. 存储层去重、落库与数据清洗4.1 用集合做内存去重采集过程中重复数据很常见尤其是榜单页同一本书可能出现在多个分类里。最简单的去重是用集合记录已采集的书籍标识。seen set() def is_duplicate(book_id): if book_id in seen: return True seen.add(book_id) return False书籍标识优先用详情页 URL 里的数字 ID比书名可靠——书名可能重复ID 不会。如果页面没有明显 ID可以用书名加作者的组合做哈希。4.2 落库选型CSV、SQLite 还是 JSON小规模采集几千条以内用 CSV 就够了方便用 Excel 打开查看。中等规模几万到几十万条我推荐 SQLite单文件、免安装、支持 SQL 查询还能建索引加速去重。import sqlite3 def init_db(pathbooks.db): conn sqlite3.connect(path) conn.execute( CREATE TABLE IF NOT EXISTS books ( book_id TEXT PRIMARY KEY, title TEXT, author TEXT, category TEXT, word_count TEXT, update_time TEXT, intro TEXT ) ) conn.commit() return conn用book_id做主键插入时用INSERT OR IGNORE天然去重不用在代码里维护集合。def save_book(conn, book): conn.execute( INSERT OR IGNORE INTO books (book_id, title, author, category, word_count, update_time, intro) VALUES (?, ?, ?, ?, ?, ?, ?) , ( book[book_id], book[title], book[author], book[category], book[word_count], book[update_time], book[intro], )) conn.commit()4.3 数据清洗的几个细节原始数据里往往有脏东西落库前要处理字数页面显示“123万字”要转成数字 1230000 才能做统计。写个转换函数处理“万”字。更新时间可能是“3天前”“昨天”这种相对时间也可能是绝对日期统一转成标准格式。简介去掉多余空白和换行截断过长的内容。import re def parse_word_count(text): text text.replace(,, ).strip() m re.search(r(\d(?:\.\d)?)\s*万, text) if m: return int(float(m.group(1)) * 10000) m re.search(r(\d), text) return int(m.group(1)) if m else 0这个函数我改过好几版因为不同站点的字数格式五花八门有“123万”“123万字”“1234567字”各种写法正则要写得宽容一点。5. 调度层翻页逻辑与任务编排5.1 翻页的两种模式翻页分两种URL 规律翻页和下一页链接翻页。前者适合 URL 里带页码参数的站点比如?page2直接构造 URL 循环就行后者要解析页面里的“下一页”链接更通用但多一步解析。def crawl_by_page(session, base_url, max_pages50): all_books [] for page in range(1, max_pages 1): url f{base_url}?page{page} resp session.get(url, timeout(5, 10)) if resp.status_code ! 200: break books parse_list_page(resp.text) if not books: break # 没有数据了说明到末页 all_books.extend(books) polite_sleep() return all_books关键在终止条件不能只靠 max_pages还要判断当前页是否还有数据。如果某页解析出来是空的说明已经翻到末尾直接 break。我见过有人只设固定页数结果站点只有 20 页脚本硬翻到 50 页后面全是重复或空页。5.2 断点续采的设计采集任务跑得久中途可能因为网络或程序问题中断。如果每次都从头开始太浪费。我的做法是记录已完成的页码重启时跳过。import json import os def load_progress(pathprogress.json): if os.path.exists(path): with open(path, r, encodingutf-8) as f: return json.load(f) return {done_pages: []} def save_progress(progress, pathprogress.json): with open(path, w, encodingutf-8) as f: json.dump(progress, f, ensure_asciiFalse)每采完一页就把页码写进去重启时读出来跳过。这个小设计在采大站点时能省下大量时间。5.3 多线程的取舍要不要上多线程我的建议是先单线程跑通再考虑并发。单线程逻辑简单、调试容易、对服务器友好。如果确实需要提速用线程池控制在 3 到 5 个并发配合更保守的间隔。from concurrent.futures import ThreadPoolExecutor def crawl_concurrent(session, urls, workers3): results [] with ThreadPoolExecutor(max_workersworkers) as executor: for books in executor.map(lambda u: fetch_and_parse(session, u), urls): results.extend(books) return results并发数不是越高越好。我实测过超过 5 个并发之后很多站点会开始返回 429 或直接拒绝连接反而要花更多时间重试。3 个并发配合 1 秒间隔是速度和稳定性的平衡点。6. 常见问题与排查技巧实录采集脚本跑起来之后问题基本集中在几类。我把踩过的坑整理成一张速查表遇到问题先对照排查。现象可能原因排查与解决返回 403请求头被识别检查 User-Agent、Referer 是否完整返回 429请求频率过高加大间隔加随机抖动降低并发解析结果为空页面结构变了或拿到的是跳转页打印原始 HTML 确认结构检查语言头中文乱码编码判断错误用resp.apparent_encoding或手动指定脚本中途崩溃某字段缺失导致索引越界所有字段提取加空值判断数据大量重复未去重或翻页逻辑有误用主键去重检查翻页终止条件6.1 编码问题的处理中文站点编码问题很常见。requests 默认用响应头里的编码但有些站点声明不准导致中文变乱码。稳妥的做法是resp session.get(url, timeout(5, 10)) resp.encoding resp.apparent_encoding # 让 chardet 猜 html resp.textapparent_encoding会调用字符集检测库来猜编码比盲目信任响应头可靠。不过它也有猜错的时候如果发现乱码可以手动指定resp.encoding gbk或utf-8试试。6.2 页面结构变化的应对站点改版是采集脚本的“天敌”。我的应对策略是把 XPath 集中管理不要散落在代码各处。SELECTORS { list_item: //div[classbook-list]//li, title: .//h3/a/text(), author: .//p[classauthor]/text(), }这样改版时只需要改这个字典不用满代码找。另外解析函数里加日志记录每个字段的提取成功率一旦某个字段突然大量为空就能第一时间发现是结构变了。6.3 我踩过的几个典型坑坑一以为翻页参数是 page结果是 p。不同站点翻页参数名不一样有的是page有的是p有的是pageNum。写之前先在浏览器里翻几页看 URL 怎么变的别想当然。坑二忽略了列表页的懒加载。有些站点列表页只渲染前几条剩下的靠滚动加载。这种情况直接抓 HTML 只能拿到前几条需要找到真正的数据接口或者用能执行 JS 的工具。我一般优先找接口比模拟滚动省事。坑三详情页链接是相对路径。列表页里的链接可能是/book/12345这种相对路径直接请求会失败要用urljoin拼成完整 URL。from urllib.parse import urljoin full_url urljoin(base_url, relative_url)这个坑很隐蔽因为相对路径拼出来的 URL 看起来“差不多对”但请求会 404排查起来费时间。坑四没处理重定向。有些详情页会重定向到新地址requests 默认会跟随重定向但如果你用allow_redirectsFalse就会拿到 302 空页面。除非有特殊需求保持默认跟随就好。7. 采集之外的几点经验做这类项目技术只是一部分更重要的是边界意识。我给自己定了几个规矩只采公开数据不碰需要登录的内容频率控制在合理范围宁可慢采到的数据只用于个人分析不二次分发。这些规矩看起来是限制实际上是让项目能长期稳定跑下去的前提。另外采集脚本的可维护性比一次性跑通更重要。我现在的习惯是把配置入口 URL、间隔、并发数、存储路径全部抽到一个配置文件里代码只负责逻辑。这样换一个站点采集改配置就行不用动代码。解析规则也尽量用配置化的 XPath配合日志监控字段提取率站点一改版就能快速定位。最后分享一个实用的小技巧先用小样本验证再全量跑。我一般先设max_pages3跑一遍检查数据质量、字段完整性、去重是否生效确认没问题再放开全量。这个习惯帮我避免过好几次“跑了两小时才发现字段全错”的尴尬。采集这件事慢就是快稳就是省。