“花自飘零水自流”这句词出自李清照的《一剪梅》看上去只是古诗词里的一句经典名句但如果你换一个视角把它当成一条需要清洗、存储、检索、分析和展示的文本数据整个技术链路就很有意思了。这次我们就把这句词作为切入点做一套完整的古诗词语料处理合集从采集、清洗、分词、情感分析到构建本地检索 API再到可视化展示和批量任务封装。整个过程不依赖 GPU普通笔记本就能跑重点不是复现某个大型模型而是把古诗词文本从“能读”变成“能算、能查、能调用”。这篇文章会直接给出可复制的 Python 示例讲清楚每一步的输入输出、预期效果、资源占用和常见坑。如果你手里正好有一批古诗词文本想做整理或者想给自己做一个诗词检索工具又或者想给后续的诗词生成、知识图谱、大模型微调准备一套干净的语料管线这篇合集可以直接收藏照着做。1. 核心能力速览先给一张规格表把整个合集能做的事情和选型列清楚方便你判断值不值得往下看。能力模块技术方案难度主要用途语料采集requests parsel / BeautifulSoup低从公开页面抓取古诗词文本数据清洗pandas 正则表达式低去重、去空、统一格式分词与词频jieba低提取关键词、统计高频词情感分析snownlp / 自建词典中分析诗词情感倾向本地存储SQLite低轻量数据库单文件部署检索 APIFlask中提供 HTTP 接口给其他系统调用可视化ECharts / pyecharts中词云、朝代分布、情感分布批量任务脚本循环 日志低批量导入、批量分词、批量导出整个合集不需要显卡也不需要 Docker只要本机能装 Python 3.9 以上版本就能跑。数据量在几万条以内SQLite 完全够用内存占用通常不超过 300MB。如果你想后续接入大模型做语义检索或诗词续写也可以把清洗后的文本导出为 JSON 或 CSV作为微调集的预处理结果。2. 适用场景与使用边界这个技术合集适合四类人。第一类是古诗词爱好者想把自己收藏的诗句整理成可检索的私人数据库按作者、朝代、词牌名、关键字快速查找。第二类是语文老师或教育研究者需要对教材诗词做词频、意象、情感倾向的量化分析比如统计“花”“月”“愁”这些高频意象观察不同作者用词差异。第三类是开发者需要给自己做的 App、公众号、知识库项目提供一个诗词查询接口本地起一个 Flask 服务就能用。第四类是算法工程师正在做大模型微调或 RAG 项目需要一批干净的结构化古诗词数据本文的清洗和导出流程可以直接复用。使用边界必须说清楚。古诗词文本可能涉及版权问题尤其是近现代作品。实际使用时优先选择已进入公有领域的古典诗词采集公开页面时要遵守目标站点的 robots 协议控制请求频率不用于商业变现。不建议大规模爬取私人整理或有明确版权声明的数据库。本文所有代码示例只用于本地学习研究发布或商用前要自行确认版权授权。另外情感分析模型对古汉语的适配性有限输出结果只能作为辅助参考不能直接当成学术结论。3. 环境准备与前置条件本文所有操作都以 Python 3.9 为基础建议使用虚拟环境隔离依赖。操作系统不限Windows、macOS、Linux 都可以但命令稍有差异下面以 Windows 和 Linux 通用写法为主。先创建项目目录和虚拟环境mkdir poetry_analysis cd poetry_analysis python -m venv venv # Windows venv\Scripts\activate # macOS / Linux source venv/bin/activate然后安装依赖pip install requests parsel pandas jieba snownlp flask flask-cors pyecharts如果你打算把词云图保存为 HTML需要 pyecharts如果只是数据分析不装也可以。下面是依赖清单和作用说明依赖库用途requests发送 HTTP 请求抓取网页parselCSS/XPath 提取网页字段pandas表格化处理、去重、统计jieba中文分词snownlp情感分析flask构建 API 服务flask-cors解决前后端跨域问题pyecharts生成交互式图表磁盘占用方面Python 环境和依赖加起来约 500MB语料库按 2 万条诗词计算JSON 文件约为 15MB 到 40MBSQLite 单文件更小通常不超过 10MB。不需要额外下载模型文件不涉及 CUDA 或显卡驱动。4. 古诗词语料采集与清洗这一节解决一个问题手里没有现成的语料怎么把“花自飘零水自流”这类诗句批量整理成结构化数据。4.1 采集思路先说明通用方法不针对具体站点做过于激进的抓取。假设目标页面结构包含“诗词标题”“作者”“朝代”“正文”四类字段我们可以用 parsel 提取。下面给出一个爬虫模板重点是控制采集节奏和解析逻辑实际使用时需要根据站点结构调整 CSS 选择器。import requests import parsel import time import random headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36 } def fetch_poem(url, title_selector, author_selector, content_selector): resp requests.get(url, headersheaders, timeout10) resp.encoding resp.apparent_encoding sel parsel.Selector(resp.text) title sel.css(title_selector).get().strip() author sel.css(author_selector).get().strip() content_list sel.css(content_selector).getall() content \n.join([c.strip() for c in content_list if c.strip()]) return { title: title, author: author, content: content } if __name__ __main__: # 示例链接和选择器实际项目需要按目标站点替换 poem fetch_poem( https://example.com/poem/1, h1::text, .author::text, .content p::text ) print(poem) time.sleep(random.uniform(1, 3))这里强调几个点。第一resp.encoding要优先用apparent_encoding很多古诗词站点是 GBK 编码UTF-8 解码会乱码。第二请求之间必须加延时避免给对方服务器造成压力。第三正文的多行内容要用 join 合并方便后续分词。4.2 清洗与去重采集回来的数据往往包含空白字符、标点混乱、重复项等问题。用 pandas 处理比较高效。import pandas as pd import re def clean_content(text): if not isinstance(text, str): return # 去掉多余空白 text re.sub(r\s, , text) # 去掉特殊字符保留中文、英文、数字、常见标点 text re.sub(r[^\u4e00-\u9fff0-9a-zA-Z。、\\《》【】\s], , text) return text.strip() df pd.DataFrame([ {title: 一剪梅, author: 李清照, content: 花自飘零水自流。一种相思两处闲愁。}, {title: 一剪梅, author: 李清照, content: 花自飘零 水自流。一种相思两处闲愁。}, ]) df[content] df[content].apply(clean_content) df df.drop_duplicates(subset[title, author, content]) print(df)输出结果中两条内容重复的数据会被去重空白字符也会被清理。实际项目中建议保留原始采集结果清洗后的数据单独存放方便以后调整清洗规则。4.3 数据存储清洗后的数据可以存成 JSON、CSV 或 SQLite。对于后续 API 服务SQLite 更合适。import sqlite3 import json conn sqlite3.connect(poetry.db) cursor conn.cursor() cursor.execute( CREATE TABLE IF NOT EXISTS poems ( id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT NOT NULL, author TEXT, content TEXT NOT NULL, source_url TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ) def insert_poems_batch(poem_list): rows [(p[title], p.get(author, ), p[content], p.get(source_url, )) for p in poem_list] cursor.executemany( INSERT INTO poems (title, author, content, source_url) VALUES (?, ?, ?, ?), rows ) conn.commit() def export_json(pathpoems.json): cursor.execute(SELECT title, author, content FROM poems) rows cursor.fetchall() with open(path, w, encodingutf-8) as f: json.dump([{title: r[0], author: r[1], content: r[2]} for r in rows], f, ensure_asciiFalse, indent2) if __name__ __main__: insert_poems_batch([]) export_json()这段代码提供了批量插入和 JSON 导出两个函数。落地到项目时你可以从 CSV 读取数据集循环调用insert_poems_batch分批写入每批 1000 条即可。5. 分词、词频与情感分析有了干净的语料下一步就是把“整句”拆成“词”再从词层面观察规律。5.1 分词与高频词统计jieba 是最常见的中文分词工具用法很直接。from collections import Counter import jieba import pandas as pd texts [ 花自飘零水自流。一种相思两处闲愁。, 红藕香残玉簟秋。轻解罗裳独上兰舟。, 云中谁寄锦书来雁字回时月满西楼。 ] stop_words {。, , , , 、, 的, 了, 在} def tokenize(text): return [w for w in jieba.lcut(text) if w.strip() and w not in stop_words] word_counter Counter() for text in texts: word_counter.update(tokenize(text)) for word, count in word_counter.most_common(15): print(word, count)运行后可以看到像“花”“水”“相思”“两处”“闲愁”“玉簟秋”这些词会进入高频词列表。分词结果和停用词表关系很大建议根据语料特点维护一份领域停用词把虚词和标点滤掉后词频统计才有意义。5.2 情感倾向分析snownlp 是一个轻量中文情感分析库分数范围在 0 到 1 之间越接近 1 情感越正向。这里用它做参考分析。from snownlp import SnowNLP poems [ 花自飘零水自流。一种相思两处闲愁。, 此情无计可消除才下眉头却上心头。, 云中谁寄锦书来雁字回时月满西楼。, 大江东去浪淘尽千古风流人物。 ] for text in poems: s SnowNLP(text) sentiment s.sentiments print(f{text[:20]}... - 情感分数: {sentiment:.4f})你大概率会看到“花自飘零水自流”这类离愁词句的情感分数偏低而“大江东去”这类豪放词略高。snownlp 本质上面向现代汉语对文言文和古诗词的拟合效果一般所以只能做相对趋势参考。如果要做更严格的诗词情感分析建议构建自定义情感词典比如把“愁”“残”“泪”“断肠”标记为负向词再统计加权分数。5.3 把分词结果导出为表格在实际项目中分词结果往往要落到文件里方便后续做可视化或机器学习特征工程。results [] for text in texts: tokens tokenize(text) results.append({ 原文: text, 分词结果: .join(tokens), 词数: len(tokens) }) df pd.DataFrame(results) df.to_csv(seg_result.csv, indexFalse, encodingutf-8-sig) print(df)utf-8-sig编码能让 Excel 直接打开看到中文不会乱码。这一步属于常用工程细节第一次做文本分析时很容易漏掉。6. 构建本地检索 API如果你只是想自己看一眼结果上一节就够了。但如果要做工具、App 或公众号你需要一个稳定的 HTTP 接口把数据暴露给其他程序。这一节用 Flask 实现一个轻量 API 服务。6.1 编写 Flask 服务from flask import Flask, request, jsonify from flask_cors import CORS import sqlite3 app Flask(__name__) CORS(app) DB_PATH poetry.db def query_db(sql, args()): conn sqlite3.connect(DB_PATH) conn.row_factory sqlite3.Row cur conn.cursor() cur.execute(sql, args) rows [dict(row) for row in cur.fetchall()] conn.close() return rows app.route(/api/poems/search, methods[GET]) def search_poem(): keyword request.args.get(keyword, ).strip() if not keyword: return jsonify({error: keyword is required}), 400 like_keyword f%{keyword}% sql SELECT id, title, author, content FROM poems WHERE title LIKE ? OR author LIKE ? OR content LIKE ? LIMIT 20 rows query_db(sql, (like_keyword, like_keyword, like_keyword)) return jsonify({total: len(rows), items: rows}) app.route(/api/poems/random, methods[GET]) def random_poem(): conn sqlite3.connect(DB_PATH) conn.row_factory sqlite3.Row cur conn.cursor() cur.execute(SELECT id, title, author, content FROM poems ORDER BY RANDOM() LIMIT 1) row dict(cur.fetchone() or {}) conn.close() return jsonify(row) app.route(/api/poems/stats, methods[GET]) def stats(): total query_db(SELECT COUNT(*) AS cnt FROM poems)[0][cnt] authors query_db(SELECT COUNT(DISTINCT author) AS cnt FROM poems)[0][cnt] return jsonify({total_poems: total, total_authors: authors}) if __name__ __main__: app.run(host0.0.0.0, port8080, debugFalse)这个服务提供三个接口接口方法参数返回/api/poems/searchGETkeyword符合条件的诗词列表/api/poems/randomGET无随机一条诗词/api/poems/statsGET无总数和作者数启动后命令行访问curl http://127.0.0.1:8080/api/poems/search?keyword花自飘零6.2 Python 客户端调用示例import requests BASE_URL http://127.0.0.1:8080 def search(keyword): resp requests.get(f{BASE_URL}/api/poems/search, params{keyword: keyword}, timeout10) resp.raise_for_status() return resp.json() if __name__ __main__: result search(花自飘零) for item in result[items]: print(item[title], item[author]) print(item[content]) print(---)这个接口设计适合本地小量级使用。如果查询量很大建议给 SQLite 加FTS5全文索引或者直接换 MySQL。搜索逻辑目前是 LIKE 模糊匹配对少量数据够用但数据量到十万条以上时全文检索性能会明显下降可以考虑引入 Elasticsearch 或 MeiliSearch但这不是本合集的第一阶段重点。7. 可视化与知识库构建数据只有变成图表才更容易看出规律。这一节用 pyecharts 生成词云和作者分布图逻辑上也可以扩展成前端页面的 JSON 接口。7.1 生成词云from collections import Counter from pyecharts.charts import WordCloud from pyecharts.globals import SymbolType from pyecharts import options as opts # 假设已经把所有诗词内容加载到 texts 列表 texts [花自飘零水自流, 一种相思两处闲愁, 红藕香残玉簟秋] counter Counter() stop_words {自, 飘零, 一种, 两处} for text in texts: words [w for w in jieba.lcut(text) if w not in stop_words] counter.update(words) data_pairs [(word, count) for word, count in counter.items()] wc WordCloud() wc.add(, data_pairs, word_size_range[20, 80], shapeSymbolType.DIAMOND) wc.render(wordcloud.html) print(WordCloud generated: wordcloud.html)打开生成的 HTML可以看到“花”“水”“流”“相思”“愁”等词被放大显示。词云适合快速直观地发现语料中的核心意象但不适合精确统计比较。7.2 作者/朝代分布图如果你的数据表里有朝代字段可以按朝代统计数量生成柱状图。from pyecharts.charts import Bar # 假设取得了 [(朝代, 数量), ...] data [(宋, 1200), (唐, 950), (清, 420), (明, 300)] bar Bar() bar.add_xaxis([d[0] for d in data]) bar.add_yaxis(诗词数量, [d[1] for d in data]) bar.set_global_opts(title_optsopts.TitleOpts(title古诗词朝代分布)) bar.render(dynasty_bar.html) print(Bar chart generated: dynasty_bar.html)如果后续想结合大模型做知识图谱可以把“作者—作品—名句”抽取成三元组存成 JSON 后导入 Neo4j 或图数据库。不过对本地单机项目来说先用 SQLite 足够不要过早引入太重的基础设施。8. 资源占用与性能观察整套流程占用资源很低你不需要为性能担心太多但要了解瓶颈在哪。启动 Flask 服务后内存占用大约 50MB 到 90MB这部分不随语料量线性增长。分词和分析阶段处理 1 万条诗词内存峰值可能在 200MB 到 500MB主要取决于是否一次性把所有文本加载到列表中。清洗和分词是最耗费 CPU 的两个环节但通常几分钟内跑完。如果你在 Windows 上还能打开任务管理器查看进程占用重点关注 Python 进程的内存和 CPU 占比。在 Linux 上可以用htop或nvidia-smi观察但本合集不涉及 GPU所以正常看不到显存占用也不需要担心显卡要求。批量导入时建议分批提交不要一次性把几十万条记录塞进一个事务。数据量较大时SQLite 写入速度会下降可以每 500 到 1000 条 commit 一次。分词同理边处理边写文件不要把所有结果全部装在内存里再统一导出。9. 常见问题与排查方法下面整理了一份高频问题排查表实际跑流程时可以直接对照。问题现象可能原因排查方式解决方案requests 请求返回乱码页面编码非 UTF-8打印resp.apparent_encoding设置resp.encoding resp.apparent_encoding爬虫请求被拒绝缺少 User-Agent 或频率过高查看响应状态码和日志完善请求头增加延时使用代理要遵守合规要求jieba 分词结果不准不是领域词典查看 example 区分词性加载自定义词典jieba.load_userdict(dict.txt)SQLite 插入很慢每行单独 commit检查代码中事务频率使用executemany 分批 commitFlask 接口启动后无法访问端口被占用或防火墙拦截检查启动日志使用netstat -ano查看端口换一个端口如port8081或关闭防火墙测试搜索接口返回 400缺少关键词参数检查请求 URL 是否带 keyword调用/api/poems/search?keyword花自飘零前端跨域被拦截未启用 CORS检查浏览器控制台错误安装并注册flask_cors.CORS(app)情感分析结果都接近 0.5snownlp 对古汉语不敏感抽样人工校验构建领域情感词典改用大模型做情感分类还有一类很常见的问题明明运行了脚本但数据库里没有数据。大概率是采集阶段选择器写错解析结果为空。这里建议在解析后立刻打印title和content的原始长度如果长度为 0就说明选择器需要调整不要再继续往下清洗。10. 最佳实践与使用建议把整个流程跑通之后有几条工程经验值得沉淀。第一语料采集要分阶段不要一次抓太多。先抓 100 条做通全流程再决定是否需要扩大到全量。第二清洗规则要版本化管理。删诗、替换标点这类规则会因为数据格式变化而失效建议把规则写进独立函数加单元测试。第三原始数据、中间数据、最终数据分目录存放。比如raw/、processed/、output/避免后期误改原数据。第四API 服务要控制访问范围如果只在本机使用host设置为127.0.0.1即可不要直接用0.0.0.0暴露到局域网。合规方面再做一次提醒不要采集有明确版权声明的现代诗词不要将语料库用于商业化或公开二次分发。如果只是本地个人学习、技术验证公开领域的古典诗词没有版权风险但引用原文时建议保留作者出处。输出到公众号或博客时也要注意引用规范。批量任务这块建议增加日志。import logging logging.basicConfig( filenamepipeline.log, levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s ) def process_poem(poem_id, content): try: words tokenize(content) logging.info(poem %s processed, %d words, poem_id, len(words)) return words except Exception as exc: logging.error(poem %s failed: %s, poem_id, exc) return []失败重试可以按poem_id记录断点重启后跳过已处理的数据。建议在你自己的工程里保留一套最小可运行配置包括一个包含 5 条诗词的测试集、一个启动脚本、一个 README这样任何环境出问题时都能快速定位。11. 总结与下一步这次我们从“花自飘零水自流”这一句词出发完整走了一条古诗词技术处理链路采集、清洗、分词、情感分析、建库、API、可视化。整套方案没有使用重型框架也没有引入 GPU 依赖适合作为个人知识库项目或教育分析项目的地基。接下来最值得做的三件事一是先把本地语料库扩充到几千条观察分词和搜索的稳定性二是把/api/poems/search接口接进自己的前端页面或者公众号后端三是尝试把清洗后的数据导出成微调格式用开源中文大模型做一个古诗词续写或风格模仿的 Demo。最容易踩的坑集中在采集阶段的编码问题和分词阶段的领域词典缺失这两个问题优先解决。后续如果你想继续深入可以考虑加入向量检索、知识图谱和大模型微调但前提都是有一份干净、结构化、可复用的古诗词语料。建议先把本文的完整流程跑一遍再决定是否扩展成更大的项目。