content.db 是一个 SQLite 数据库文件,每个文档对应一个独立的数据库文件,存储文档中的所有内容块(Block)。
tmp/
├── {user_id}/
│ ├── {YYYY-MM-DD}/
│ │ └── doc-abc123.doc # 导出的 Word 文件
│ └── sqlite/
│ └── doc-abc123.db # 文档内容数据库(SQLite)
└── default.json # 默认样式文件
路径说明:
tmp/{user_id}/{YYYY-MM-DD}/{filename}.doc(按日期分区)tmp/{user_id}/sqlite/{document_id}.db(统一存放)# documents 表字段
class Document(Base):
__tablename__ = "documents"
id: Mapped[str] = mapped_column(String(64), primary_key=True)
content_db_path: Mapped[str] = mapped_column(String(1024)) # SQLite 数据库路径
# ... 其他字段
| 特性 | 优势 |
|---|---|
| 结构化存储 | 原生支持,无需解析 |
| 查询性能 | O(log n) 索引查询 |
| 局部更新 | 直接 UPDATE,无需全量更新 |
| 全文搜索 | 支持 FTS5 全文索引 |
| 关系查询 | 原生 JOIN 支持 |
| 数据完整性 | 外键约束、事务支持 |
CREATE TABLE document_blocks (
-- 主键
id TEXT PRIMARY KEY, -- Block 唯一 ID,如 "block-h1-0"
-- 位置
block_order INTEGER NOT NULL, -- 在文档中的绝对位置(用于排序)
-- 类型
type TEXT NOT NULL, -- Block 类型:heading, paragraph, table, image
-- 标题信息
level INTEGER NOT NULL DEFAULT 0, -- 标题级别(1-6)或段落级别(0)
"index" INTEGER NOT NULL DEFAULT 0, -- 该级别在文档中的序号
-- 内容
content TEXT, -- 文本内容(标题、段落)或结构化数据(表格、图片)
-- 样式
word_style TEXT, -- Word 样式名称,如 "Heading 1", "Normal"
style TEXT, -- 颗粒度样式 JSON(自定义样式时使用)
-- 元数据
metadata TEXT -- 其他元数据 JSON
);
-- 索引
CREATE INDEX idx_block_order ON document_blocks(block_order);
CREATE INDEX idx_type ON document_blocks(type);
CREATE INDEX idx_level ON document_blocks(level);
字段说明:
| 字段 | 类型 | 说明 | 示例 |
|---|---|---|---|
id |
TEXT | Block 唯一标识 | "block-h1-0", "block-p-5", "block-table-0" |
block_order |
INTEGER | 文档中的绝对位置,用于排序 | 1, 2, 3, ... |
type |
TEXT | Block 类型 | "heading", "paragraph", "table", "image" |
level |
INTEGER | 标题级别(1-6)或 0(非标题) | 1 (H1), 2 (H2), 0 (段落) |
index |
INTEGER | 该级别标题的序号 | 第 2 个 H2 标题的 index 为 1 |
content |
TEXT | 内容存储(格式因类型而异) | 标题/段落:纯文本;表格:JSON;图片:Base64 |
word_style |
TEXT | Word 样式名 | "Heading 1", "Normal", "Table Grid" |
style |
TEXT | 自定义样式 JSON | "{}" 或 '{"bold": true, "color": "FF0000"}' |
metadata |
TEXT | 元数据 JSON | '{"parent_id": "block-h1-0"}' |
为了在插入或删除 Block 时避免大量更新 block_order、index 等字段,采用稀疏排序策略。
传统方案的问题:
1, 2, 3, 4, 53, 4, 5 全部更新为 4, 5, 64, 5 全部更新为 3, 4稀疏排序方案:
100, 200, 300, 400, 500(间隔 100)150,无需更新其他 Block125, 175 等,填充间隙-- 初始文档结构(间隔 100)
INSERT INTO document_blocks (id, block_order, type, level, "index", content, ...)
VALUES
('block-h1-0', 0, 'heading', 1, 0, '第一章', ...),
('block-p-0', 100, 'paragraph', 0, 0, '段落1', ...),
('block-p-1', 200, 'paragraph', 0, 0, '段落2', ...),
('block-table-0', 300, 'table', 0, 0, '{"rows": [...]}', ...),
('block-h1-1', 400, 'heading', 1, 1, '第二章', ...);
-- 在 block_order=200 后插入段落
-- 计算中间值:(200 + 300) / 2 = 250
INSERT INTO document_blocks (id, block_order, type, level, "index", content, ...)
VALUES ('block-p-new', 250, 'paragraph', 0, 0, '新段落', ...);
-- 结果:0, 100, 200, 250(新), 300, 400
-- 无需更新其他 Block
-- 继续在 200 和 250 之间插入
INSERT INTO document_blocks (...) VALUES (..., 225, ...);
-- 结果:0, 100, 200, 225(新), 250, 300, 400
当某个区间的间隙用尽时(例如无法再在 200-201 之间插入),触发局部重排:
场景示例:
原始序列:200, 201, 202, 300
需要在 200 和 201 之间插入 → 间隙不足(200.5 无法使用整数)
触发局部重排(重新分配 200-300 区间):
- 区间内有 3 个 Block(201, 202, 新Block),区间大小 = 100
- 重新分配:220, 240, 260
- 结果:200, 220(原201), 240(原202), 260(新), 300
继续可插入:200, 210(新), 220, 230(新), 240, 250(新), 260, 270(新), 300
重排策略:
block_order 差值 ≤ 1 时标题的 index 字段也可以采用稀疏排序:
-- 初始标题(间隔 100)
INSERT INTO document_blocks (id, type, level, "index", ...)
VALUES
('block-h2-0', 'heading', 2, 0, ...), -- index: 0
('block-h2-100', 'heading', 2, 100, ...), -- index: 100
('block-h2-200', 'heading', 2, 200, ...); -- index: 200
-- 在 index=0 和 index=100 之间插入新标题
INSERT INTO document_blocks (id, type, level, "index", ...)
VALUES ('block-h2-50', 'heading', 2, 50, ...);
-- 结果:index: 0, 50(新), 100, 200
| 操作 | 传统方案 | 稀疏排序方案 |
|---|---|---|
| 插入 | 更新所有后续 Block | 仅插入 1 条记录 |
| 删除 | 更新所有后续 Block | 仅删除 1 条记录 |
| 性能 | O(n) 更新操作 | O(1) 插入/删除 |
| 数据库压力 | 高(大量 UPDATE) | 低(单条 INSERT/DELETE) |
| 适用场景 | 小文档(<100 Block) | 大文档(1000+ Block) |
如果在 metadata 中同时存储 parent_id 和 children_ids,会导致严重的性能问题:
问题示例:
{
"parent_id": "block-h1-0",
"children_ids": ["block-h2-0", "block-h2-100", "block-h2-200"] // ❌ 不推荐
}
性能问题:
只存储 parent_id,不存储 children_ids:
{
"parent_id": "block-h1-0"
// ✓ 不存储 children_ids,按需动态计算
}
优势:
动态查询子节点:
-- 查询某标题的所有子标题
SELECT * FROM document_blocks
WHERE type = 'heading'
AND json_extract(metadata, '$.parent_id') = 'block-h1-0'
ORDER BY block_order;
简单文本(无富文本格式):
INSERT INTO document_blocks (id, block_order, type, level, "index", content, word_style, style, metadata)
VALUES (
'block-h1-0', -- id
100, -- block_order
'heading', -- type
1, -- level
0, -- index
'基本数据', -- content(纯文本)
'Heading 1', -- word_style
'{}', -- style(空表示使用 word_style)
'{"parent_id": null}' -- metadata
);
当标题或段落中部分文字需要不同样式时(如部分加粗、变色),使用 JSON 数组格式:
示例 1:标题中部分文字变色
INSERT INTO document_blocks (id, block_order, type, level, "index", content, word_style, style, metadata)
VALUES (
'block-h1-1',
200,
'heading',
1,
100,
'[
{"text": "基本", "style": {}},
{"text": "数据", "style": {"color": "FF0000", "bold": true}}
]', -- content(富文本 JSON)
'Heading 1',
'{}',
'{"parent_id": null}'
);
渲染结果: "基本数据"
示例 2:标题中多段样式
-- 标题:"第一章 概述与分析"("概述与分析"为红色)
INSERT INTO document_blocks (..., content, ...)
VALUES (...,
'[
{"text": "第一章 ", "style": {}},
{"text": "概述与分析", "style": {"color": "FF0000"}}
]',
...
);
示例 3:复杂富文本组合
-- 标题:"项目名称:东河塘油田"("东河塘油田"加粗且下划线)
INSERT INTO document_blocks (..., content, ...)
VALUES (...,
'[
{"text": "项目名称:", "style": {}},
{"text": "东河塘油田", "style": {"bold": true, "underline": true}}
]',
...
);
content 字段支持两种格式:
格式 1:纯文本字符串
content = '基本数据'
word_style + style(Block 级别)格式 2:富文本 JSON 数组
[
{
"text": "文本片段1",
"style": {} // 继承 Block 级别样式
},
{
"text": "文本片段2",
"style": {"color": "FF0000", "bold": true} // 覆盖特定属性
}
]
word_style + style样式文件 (word_style)
↓
Block 级别样式 (style)
↓
文本片段样式 (content[].style)
完整示例:
-- Block 定义
INSERT INTO document_blocks (id, block_order, type, level, "index", content, word_style, style, metadata)
VALUES (
'block-h1-2',
300,
'heading',
1,
200,
'[
{"text": "关于", "style": {}},
{"text": "重要", "style": {"color": "FF0000", "bold": true}},
{"text": "通知", "style": {"font_size": 16.0}}
]',
'Heading 1', -- 样式文件中定义:宋体、14pt、黑色
'{"font_name": "微软雅黑"}', -- Block 覆盖:改为微软雅黑
'{"parent_id": null}'
);
渲染逻辑:
{font_name: "宋体", font_size: 14.0, color: "000000"}{font_name: "微软雅黑"} → 字体变为微软雅黑word_style → Block styleword_style → Block style → 片段 style设计要点:
level + index 对应现有 API 的局部更新逻辑metadata 中的 parent_id 构建标题树,便于生成目录block_order 使用稀疏排序保证文档块的顺序content 支持纯文本或富文本 JSON 格式查询示例:
-- 获取所有一级标题
SELECT * FROM document_blocks
WHERE type = 'heading' AND level = 1
ORDER BY block_order;
-- 获取标题树(使用 parent_id)
SELECT
b1.id,
b1.level,
b1."index",
b1.content,
json_extract(b1.metadata, '$.parent_id') as parent_id
FROM document_blocks b1
WHERE b1.type = 'heading'
ORDER BY b1.block_order;
-- 获取某标题下的所有内容块
SELECT * FROM document_blocks
WHERE json_extract(metadata, '$.parent_heading_id') = 'block-h1-0'
ORDER BY block_order;
方式 1:简单文本(无富文本格式)
INSERT INTO document_blocks (id, block_order, type, level, "index", content, word_style, style, metadata)
VALUES (
'block-p-0', -- id
200, -- block_order
'paragraph', -- type
0, -- level(段落为 0)
0, -- index
'东河塘油田东河1区块DH1-10H井', -- content(纯文本)
'Normal', -- word_style
'{}', -- style
'{"parent_heading_id": "block-h1-0"}' -- metadata
);
方式 2:富文本 JSON 格式(推荐)
INSERT INTO document_blocks (id, block_order, type, level, "index", content, word_style, style, metadata)
VALUES (
'block-p-1',
300,
'paragraph',
0,
0,
'[
{"text": "东河塘油田", "style": {"bold": true}},
{"text": "东河1区块DH1-10H井", "style": {}}
]', -- content(富文本 JSON)
'Normal',
'{}',
'{"parent_heading_id": "block-h1-0"}'
);
示例 1:部分文字加粗和变色
-- 段落:"项目位置在陕西省,投资金额约5000万元"
-- 效果:"陕西省" 红色加粗,"5000万元" 蓝色加粗
INSERT INTO document_blocks (..., content, ...)
VALUES (...,
'[
{"text": "项目位置在", "style": {}},
{"text": "陕西省", "style": {"bold": true, "color": "FF0000"}},
{"text": ",投资金额约", "style": {}},
{"text": "5000万元", "style": {"bold": true, "color": "0000FF"}}
]',
...
);
示例 2:混合样式段落
-- 复杂格式:普通文字 + 加粗 + 斜体 + 下划线
INSERT INTO document_blocks (..., content, ...)
VALUES (...,
'[
{"text": "根据", "style": {}},
{"text": "《安全生产法》", "style": {"bold": true, "underline": true}},
{"text": "第23条规定,企业应当", "style": {}},
{"text": "定期检查", "style": {"italic": true, "color": "FF0000"}},
{"text": "设备运行状态。", "style": {}}
]',
...
);
| 格式 | 适用场景 | 优点 | 缺点 |
|---|---|---|---|
| 纯文本 | 统一样式段落 | 简洁、数据量小 | 无法部分样式调整 |
| 富文本 JSON | 复杂样式、精确控制 | 样式灵活、精确 | 数据量稍大 |
纯文本与富文本对照:
-- 纯文本格式(整段统一样式)
content = '东河塘油田东河1区块'
-- 富文本格式(部分文字加粗)
content = '[
{"text": "东河塘油田", "style": {"bold": true}},
{"text": "东河1区块", "style": {}}
]'
设计要点:
metadata.parent_heading_id 关联所属章节查询示例:
-- 搜索段落内容
SELECT * FROM document_blocks
WHERE type = 'paragraph'
AND content LIKE '%东河塘%'
ORDER BY block_order;
-- 获取某章节下的所有段落
SELECT * FROM document_blocks
WHERE type = 'paragraph'
AND json_extract(metadata, '$.parent_heading_id') = 'block-h2-0'
ORDER BY block_order;
样式层级:
表格样式(默认):存储在 Block 的 style 字段
单元格内容样式(可修改):存储在每个单元格的 style 字段
合并单元格与尺寸:存储在 metadata 和单元格属性中
rowspan、colspan:单元格合并table_width、col_widths:表格和列宽表格的 content 字段存储 JSON 格式的结构化数据:
简单表格示例:
{
"rows": [
{
"cells": [
{
"text": "设计单位名称",
"rowspan": 1,
"colspan": 1,
"style": {"bold": true} // ← Word 提取时检测到加粗
},
{
"text": "东河采油气管理区",
"rowspan": 1,
"colspan": 3,
"style": {}
}
]
},
{
"cells": [
{
"text": "审核意见",
"rowspan": 2,
"colspan": 1,
"style": {"bold": true, "valign": "middle"} // ← 加粗 + 垂直居中
},
{
"text": "说明:...",
"rowspan": 1,
"colspan": 1,
"style": {}
}
]
}
]
}
插入示例:
INSERT INTO document_blocks (id, block_order, type, level, "index", content, word_style, style, metadata)
VALUES (
'block-table-0', -- id
500, -- block_order
'table', -- type
0, -- level
0, -- index
'{"rows": [...]}', -- content(JSON 格式)
'Table Grid', -- word_style(表格样式名)
'{}', -- style(使用默认表格样式)
'{"cols": 4, "rows": 6, "table_width": 100, "table_width_unit": "percent", "col_widths": [25, 25, 25, 25], "parent_heading_id": "block-h2-0"}'
);
Metadata 字段说明:
| 字段 | 类型 | 说明 | 示例 |
|---|---|---|---|
cols |
number | 表格列数 | 4 |
rows |
number | 表格行数 | 6 |
table_width |
number | 表格宽度值 | 100 |
table_width_unit |
string | 表格宽度单位 | "percent", "cm", "inch" |
col_widths |
array | 每列宽度 | [25, 25, 25, 25] |
row_heights |
array | 每行高度(厘米) | [1.0, 0.8, 0.8] |
parent_heading_id |
string | 所属章节 ID | "block-h2-0" |
表格总宽度单位:
// 示例 1:百分比宽度(占页面宽度的 100%)
{
"table_width": 100,
"table_width_unit": "percent"
}
// 示例 2:厘米宽度(固定 16 厘米)
{
"table_width": 16,
"table_width_unit": "cm"
}
// 示例 3:英寸宽度(固定 6.5 英寸)
{
"table_width": 6.5,
"table_width_unit": "inch"
}
规则:
cols)table_width_unit 为 "percent" 时,值表示百分比table_width_unit 为 "cm" 或 "inch" 时,值表示绝对宽度示例 1:百分比列宽(等宽)
{
"cols": 4,
"table_width": 100,
"table_width_unit": "percent",
"col_widths": [25, 25, 25, 25] // 每列占 25%,总和 = 100%
}
示例 2:百分比列宽(不等宽)
{
"cols": 4,
"table_width": 100,
"table_width_unit": "percent",
"col_widths": [40, 30, 20, 10] // 总和 = 100%
}
示例 3:绝对列宽(厘米)
{
"cols": 3,
"table_width": 15,
"table_width_unit": "cm",
"col_widths": [5, 5, 5] // 每列 5 厘米,总和 = 15cm
}
规则:
rows)示例 1:固定行高
{
"rows": 3,
"row_heights": [1.5, 1.0, 1.0] // 第1行 1.5cm,第2、3行各 1.0cm
}
示例 2:不定义行高(自动适应)
{
"rows": 3
// 不定义 row_heights,行高根据内容自动调整
}
示例 3:混合使用
{
"rows": 5,
"row_heights": [2.0, 0.8, 0.8, 0.8, 1.5]
// 第1行(表头)2.0cm
// 第2-4行(数据)0.8cm
// 第5行(总计)1.5cm
}
{
"cols": 4,
"rows": 3,
"table_width": 100,
"table_width_unit": "percent",
"col_widths": [30, 25, 25, 20], // 列宽:30%, 25%, 25%, 20%
"row_heights": [1.2, 0.8, 0.8], // 行高:1.2cm, 0.8cm, 0.8cm
"parent_heading_id": "block-h2-0"
}
单元格的 text 字段支持两种格式:
格式 1:纯文本
{
"text": "设计单位名称",
"rowspan": 1,
"colspan": 1,
"style": {"bold": true} // 整个单元格统一样式
}
格式 2:富文本(单元格内部分样式)
{
"text": [
{"text": "设计单位", "style": {}},
{"text": "名称", "style": {"color": "FF0000"}}
],
"rowspan": 1,
"colspan": 1,
"style": {"bold": true} // 单元格基础样式
}
完整示例:表头加粗显示
INSERT INTO document_blocks (..., content, ...)
VALUES (...,
'{
"rows": [
{
"cells": [
{"text": "姓名", "rowspan": 1, "colspan": 1, "style": {"bold": true}},
{"text": "年龄", "rowspan": 1, "colspan": 1, "style": {"bold": true}},
{"text": "职位", "rowspan": 1, "colspan": 1, "style": {"bold": true}}
]
},
{
"cells": [
{"text": "张三", "rowspan": 1, "colspan": 1, "style": {}},
{"text": "30", "rowspan": 1, "colspan": 1, "style": {}},
{"text": "工程师", "rowspan": 1, "colspan": 1, "style": {}}
]
}
]
}',
...
);
说明:
style.boldstyle.bold 渲染加粗效果{
"text": string | array, // 单元格内容:纯文本 或 富文本数组
"rowspan": number, // 垂直合并单元格数(默认1)✅ 可修改
"colspan": number, // 水平合并单元格数(默认1)✅ 可修改
"style": { // 单元格样式
// ===== 文本内容样式(可修改)=====
"font_name": string, // 字体名称 ✅
"font_size": number, // 字号(磅)✅
"bold": boolean, // 是否粗体 ✅ ← Word 提取时检测
"italic": boolean, // 是否斜体 ✅
"underline": boolean, // 是否下划线 ✅
"color": string, // 文字颜色(十六进制)✅
"align": string, // 水平对齐:"left", "center", "right" ✅
"valign": string, // 垂直对齐:"top", "middle", "bottom" ✅
// ===== 单元格样式(目前不支持修改)=====
}
}
可修改的属性:
text)rowspan、colspan)暂不支持修改的属性:
bg_color)border_*)padding)注意: Word 提取时会检测并保存所有样式属性,但前端编辑器目前只支持修改文本内容样式和对齐方式。
注意: 单元格边框目前不支持修改,此定义仅供 Word 提取和系统内部使用。
-- 获取表格基本信息
SELECT
id,
json_extract(metadata, '$.cols') as cols,
json_extract(metadata, '$.rows') as rows,
json_extract(metadata, '$.table_width') as width
FROM document_blocks
WHERE type = 'table'
ORDER BY block_order;
-- 搜索表格中的内容(使用 JSON 函数)
SELECT
id,
block_order,
content
FROM document_blocks
WHERE type = 'table'
AND content LIKE '%东河%'
ORDER BY block_order;
-- 统计表格数量
SELECT COUNT(*) as table_count
FROM document_blocks
WHERE type = 'table';
-- 获取包含特定列数的表格
SELECT * FROM document_blocks
WHERE type = 'table'
AND json_extract(metadata, '$.cols') = 4
ORDER BY block_order;
Word 文档使用 XML 标记表示合并单元格:
垂直合并(rowspan)标记:
<!-- 合并起始单元格 -->
<w:tc>
<w:tcPr>
<w:vMerge w:val="restart"/> <!-- 标记合并开始 -->
</w:tcPr>
<w:p><w:r><w:t>井控风险级别划分</w:t></w:r></w:p>
</w:tc>
<!-- 被合并的单元格(在后续行中)-->
<w:tc>
<w:tcPr>
<w:vMerge/> <!-- 或 <w:vMerge w:val="continue"/> -->
</w:tcPr>
<w:p></w:p> <!-- 内容通常为空 -->
</w:tc>
水平合并(colspan)标记:
<w:tc>
<w:tcPr>
<w:gridSpan w:val="3"/> <!-- 横跨3列 -->
</w:tcPr>
<w:p><w:r><w:t>标题</w:t></w:r></w:p>
</w:tc>
<!-- 注意:被横跨的列在 XML 中不存在 -->
vMerge 属性值说明:
| vMerge 值 | 含义 | Python 解析结果 |
|---|---|---|
w:val="restart" |
垂直合并的起始单元格 | v_merge_val = 'restart' (字符串) |
w:val="continue" |
垂直合并的继续单元格(新格式) | v_merge_val = 'continue' (字符串) |
无 val 属性(<w:vMerge/>) |
垂直合并的继续单元格(旧格式) | v_merge_val = None |
关键注意事项:
v_merge_val 可能是字符串 'continue',而不是 None提取表格时采用两遍扫描策略,确保正确识别合并单元格:
第一遍:构建 vmerge_map
扫描所有单元格的 XML 元素,记录每列的所有垂直合并区间:
# vmerge_map 数据结构
vmerge_map = {
0: [ # 列0的合并区间列表
{'start_row': 1, 'end_row': 12}, # 行1-12 合并
{'start_row': 15, 'end_row': 15} # 行15 单独
],
1: [ # 列1的合并区间列表
{'start_row': 1, 'end_row': 8}, # 行1-8 合并
{'start_row': 9, 'end_row': 11}, # 行9-11 合并
{'start_row': 12, 'end_row': 12} # 行12 单独
]
}
检测逻辑:
if has_vmerge_restart:
# 开始新的行合并区间
merges.append({
'start_row': row_idx,
'end_row': row_idx
})
elif has_vmerge_continue:
# 扩展当前合并区间(只有明确标记为 continue 才扩展)
if merges:
merges[-1]['end_row'] = row_idx
# ⚠️ 不使用 is_empty 判断,避免将空单元格误认为合并
第二遍:提取单元格数据
从 XML 直接提取单元格,根据 vmerge_map 判断是否跳过:
for row_idx, tr in enumerate(trs):
for tc in tr.findall(qn('w:tc')):
# 1. 检查 vMerge 标记
v_merge_val = tc.find('.//w:vMerge').get(qn('w:val'))
# 2. 跳过被合并的单元格
if v_merge_val != 'restart': # 'continue' 或 None
col_offset += colspan
continue # 不提取被合并的单元格
# 3. 计算 rowspan
rowspan = 1
if col_offset in vmerge_map:
for merge in vmerge_map[col_offset]:
if merge['start_row'] == row_idx:
rowspan = merge['end_row'] - merge['start_row'] + 1
break
# 4. 提取单元格数据(从 XML)
cell_data = {
'text': extract_text_from_xml(tc),
'rowspan': rowspan,
'colspan': colspan,
'col_index': col_offset, # ← 记录绝对列位置
'style': extract_style_from_xml(tc)
}
cells_data.append(cell_data)
col_offset += colspan
关键字段:col_index
col_index 记录单元格在表格中的绝对列位置(0-based)导出时使用 col_index 字段确保单元格被放置到正确的列位置:
导出逻辑:
# 1. 构建列索引映射
cells_by_col = {}
for cell_data in cells_data:
col_idx = cell_data.get('col_index')
if col_idx is not None:
cells_by_col[col_idx] = cell_data
# 2. 遍历所有列(而非只遍历 cells_data)
occupied = {} # 记录被合并单元格占用的位置
for col_idx in range(num_cols):
# 3. 检查位置是否被占用
if occupied.get((r_idx, col_idx), False):
continue # 跳过被占用的位置
# 4. 检查是否有数据填充
if col_idx not in cells_by_col:
continue # 这个位置没有数据
cell_data = cells_by_col[col_idx]
# 5. 使用绝对列位置访问单元格
start_cell = table.rows[r_idx].cells[col_idx]
# 6. 处理合并
if colspan > 1 or rowspan > 1:
end_row = r_idx + rowspan - 1
end_col = col_idx + colspan - 1
end_cell = table.rows[end_row].cells[end_col]
start_cell.merge(end_cell)
# 7. 标记被合并的位置为已占用
for merge_r in range(r_idx, end_row + 1):
for merge_c in range(col_idx, end_col + 1):
if merge_r != r_idx or merge_c != col_idx:
occupied[(merge_r, merge_c)] = True
关键改进:
cells_by_col 映射,避免按顺序迭代occupied 字典跟踪被合并占用的位置col_index 而非 col_offset,确保位置正确表格效果:
┌──────────────┬─────────┬──────────┬──────┬──────┐
│井控风险级别 │ 一级 │ 预测地层 │ │ │
│划分 │ │ 压力 │ │ │
│(12行合并) │(8行合并)│≥105MPa │ │ │
├──────────────┼─────────┼──────────┼──────┼──────┤
│ │ │ 预测硫化 │ │ │
│ │ │ 氢含量 │ │ │
│ ├─────────┼──────────┼──────┼──────┤
│ │ 二级 │ ... │ │ │
│ │(3行合并)│ │ │ │
└──────────────┴─────────┴──────────┴──────┴──────┘
提取结果(JSON 数据):
{
"rows": [
{
"cells": [
{
"col_index": 0,
"rowspan": 12,
"colspan": 1,
"text": "井控风险级别划分",
"style": {}
},
{
"col_index": 1,
"rowspan": 8,
"colspan": 1,
"text": "一级",
"style": {}
},
{
"col_index": 2,
"rowspan": 1,
"colspan": 3,
"text": "预测地层压力≥105MPa",
"style": {}
},
{
"col_index": 5,
"rowspan": 1,
"colspan": 1,
"text": " ",
"style": {}
},
{
"col_index": 6,
"rowspan": 1,
"colspan": 1,
"text": " ",
"style": {}
}
]
},
{
"cells": [
// 注意:列0和列1被上一行占用,不在此行定义
{
"col_index": 2,
"rowspan": 1,
"colspan": 3,
"text": "预测硫化氢含量...",
"style": {}
},
{
"col_index": 5,
"rowspan": 1,
"colspan": 1,
"text": " ",
"style": {}
},
{
"col_index": 6,
"rowspan": 1,
"colspan": 1,
"text": " ",
"style": {}
}
]
},
{
"cells": [
// 列0仍被第1行占用,列1在第9行才重新开始
{
"col_index": 1,
"rowspan": 3,
"colspan": 1,
"text": "二级",
"style": {}
},
// ...
]
}
]
}
关键特征:
col_index 字段记录绝对列位置rowspan 表示垂直跨越的行数(从 vmerge_map 计算)colspan 表示水平跨越的列数(从 gridSpan 提取)提取阶段关键点:
row.cells:对于合并单元格,它会在多行返回同一个对象<w:tc> 元素导出阶段关键点:
table.rows[r_idx].cells[col_idx]常见错误与修复:
| 错误 | 现象 | 原因 | 修复 |
|---|---|---|---|
| 错误合并 | 空单元格被合并 | 使用 is_empty 判断扩展合并 |
只在明确标记 vMerge=continue 时扩展 |
| 重复单元格 | 被合并的单元格也被提取 | 没有跳过 vMerge=continue 单元格 |
检测并跳过 |
| 位置错误 | 单元格在错误的列 | 按顺序迭代而非使用 col_index | 使用 cells_by_col 映射 + 遍历所有列 |
| 应该合并的不合并 | rowspan 应该 >1 但为 1 | 第一遍扫描的判断逻辑不一致 | 统一使用 v_merge_val != 'restart' |
TABLE_MERGE_FIX_SUMMARY.md - 提取阶段修复详细说明TABLE_EXPORT_FIX_SUMMARY.md - 导出阶段修复详细说明MERGE_CELL_COMPLETE_FIX.md - 完整修复总结示例 1:基础表格(无合并)
{
"rows": [
{
"cells": [
{"text": "姓名", "rowspan": 1, "colspan": 1, "style": {"bold": true}},
{"text": "年龄", "rowspan": 1, "colspan": 1, "style": {"bold": true}},
{"text": "职位", "rowspan": 1, "colspan": 1, "style": {"bold": true}}
]
},
{
"cells": [
{"text": "张三", "rowspan": 1, "colspan": 1, "style": {}},
{"text": "30", "rowspan": 1, "colspan": 1, "style": {}},
{"text": "工程师", "rowspan": 1, "colspan": 1, "style": {}}
]
}
]
}
示例 2:合并单元格
表格效果:
┌────────────────────────────────┐
│ 标题 │ ← 第1行:1个单元格横跨3列
│ (colspan=3) │
├──────────┬───────────┬─────────┤
│ │ 子项1 │ 100 │ ← 第2行:"项目A"纵跨2行
│ 项目A ├───────────┼─────────┤
│(rowspan=2)│ 子项2 │ 200 │ ← 第3行
│ │ │ │
└──────────┴───────────┴─────────┘
JSON 数据结构:
{
"rows": [
{
"cells": [
{
"text": "标题",
"rowspan": 1,
"colspan": 3, // ← 横跨3列
"col_index": 0,
"style": {
"bold": true,
"align": "center",
"bg_color": "F0F0F0"
}
}
]
},
{
"cells": [
{
"text": "项目A",
"rowspan": 2, // ← 纵跨2行
"colspan": 1,
"col_index": 0,
"style": {
"valign": "middle"
}
},
{"text": "子项1", "rowspan": 1, "colspan": 1, "col_index": 1, "style": {}},
{"text": "100", "rowspan": 1, "colspan": 1, "col_index": 2, "style": {}}
]
},
{
"cells": [
// 注意:第1个单元格被上一行的"项目A"占据,所以这行只有2个单元格
{"text": "子项2", "rowspan": 1, "colspan": 1, "col_index": 1, "style": {}},
{"text": "200", "rowspan": 1, "colspan": 1, "col_index": 2, "style": {}}
]
}
]
}
关键点:
colspan: 横向合并,值为合并的列数rowspan: 纵向合并,值为合并的行数col_index: 单元格在表格中的绝对列位置(0-based)插入示例:
INSERT INTO document_blocks (id, block_order, type, level, "index", content, word_style, style, metadata)
VALUES (
'block-img-0', -- id
10, -- block_order
'image', -- type
0, -- level
0, -- index
'data:image/png;base64,iVBORw0KG...', -- content(Base64 数据)
'Normal', -- word_style
'{"width": 10.0, "height": 7.0, "unit": "cm", "align": "center"}', -- style
'{"alt": "示意图", "para_style": "Normal", "parent_heading_id": "block-h2-0"}' -- metadata
);
样式优先级:同标题块
设计要点:
content 存储 Base64 编码的图片数据(Data URL 格式)style 定义图片显示尺寸和对齐方式metadata 包含图片描述和所属章节查询示例:
-- 获取所有图片
SELECT
id,
block_order,
json_extract(metadata, '$.alt') as alt,
json_extract(style, '$.width') as width,
json_extract(style, '$.height') as height
FROM document_blocks
WHERE type = 'image'
ORDER BY block_order;
-- 获取某章节下的所有图片
SELECT * FROM document_blocks
WHERE type = 'image'
AND json_extract(metadata, '$.parent_heading_id') = 'block-h2-0'
ORDER BY block_order;
在 Word 中,标题有多种设置方式:
方式A:使用内置标题样式
Heading 1, Heading 2, Heading 3 等Heading 1 → level: 1,Heading 2 → level: 2方式B:使用自定义样式名称
level: 1(一级标题)level: 2(二级标题)方式C:直接格式化但未应用样式
from docx import Document
def identify_heading_level(paragraph):
"""识别段落的标题级别"""
style = paragraph.style
# 方法1:检查样式名称(内置样式)
if style.name.startswith('Heading'):
try:
level = int(style.name.split()[-1]) # "Heading 1" → 1
return level, style.name
except ValueError:
pass
# 方法2:检查大纲级别(自定义样式)
if hasattr(style.element, 'pPr') and style.element.pPr is not None:
outline_lvl = style.element.pPr.outlineLvl
if outline_lvl is not None:
level = outline_lvl.val + 1 # Word 大纲级别从 0 开始
return level, style.name
# 方法3:通过段落格式的大纲级别
if paragraph._element.pPr is not None:
outline_lvl = paragraph._element.pPr.outlineLvl
if outline_lvl is not None:
level = outline_lvl.val + 1
return level, style.name
# 不是标题,返回 None
return None, style.name
style(颗粒度样式)覆盖 word_style(样式文件)中的对应属性 > 默认样式(代码内置)
重要说明:
style 只在渲染时覆盖,不修改样式文件本身style 字段独立存储覆盖值场景 A:使用样式文件(推荐)
INSERT INTO document_blocks (..., word_style, style, ...)
VALUES (..., 'Normal', '{}', ...);
示例:
// 样式文件中的 "Normal" 定义
{
"font_name": "宋体",
"font_size": 12.0,
"color": "000000",
"align": "left"
}
// 最终应用的样式
{
"font_name": "宋体",
"font_size": 12.0,
"color": "000000",
"align": "left"
}
场景 B:样式覆盖(部分自定义)
INSERT INTO document_blocks (..., word_style, style, ...)
VALUES (..., 'Normal', '{"font_name": "黑体"}', ...);
style 中定义的属性覆盖 word_style 中的对应属性style 中未定义的属性继承自 word_style示例:
// 样式文件中的 "Normal" 定义
{
"font_name": "宋体", // ← 会被覆盖
"font_size": 12.0, // ← 继承
"color": "000000", // ← 继承
"align": "left" // ← 继承
}
// style 中的覆盖定义
{
"font_name": "黑体" // 只覆盖字体
}
// 最终应用的样式(合并结果)
{
"font_name": "黑体", // ✓ 来自 style(覆盖)
"font_size": 12.0, // ✓ 来自 word_style(继承)
"color": "000000", // ✓ 来自 word_style(继承)
"align": "left" // ✓ 来自 word_style(继承)
}
场景 C:完全自定义样式(前端编辑器)
INSERT INTO document_blocks (..., word_style, style, ...)
VALUES (..., 'Normal', '{"font_name": "微软雅黑", "font_size": 18.0, "bold": true, "color": "FF0000", "align": "center"}', ...);
style 中定义了所有必需属性时,完全使用自定义样式word_style 作为基准,但前端可以完全重写示例:
// 样式文件中的 "Normal" 定义(作为参考)
{
"font_name": "宋体",
"font_size": 12.0,
"color": "000000",
"align": "left"
}
// style 中的完全自定义
{
"font_name": "微软雅黑",
"font_size": 18.0,
"bold": true,
"color": "FF0000",
"align": "center"
}
// 最终应用的样式(完全自定义)
{
"font_name": "微软雅黑", // ✓ 来自 style
"font_size": 18.0, // ✓ 来自 style
"bold": true, // ✓ 来自 style
"color": "FF0000", // ✓ 来自 style
"align": "center" // ✓ 来自 style
}
def apply_style(block):
"""应用样式的合并逻辑"""
# 1. 从样式文件加载 word_style(只读,不修改文件)
base_style = load_style_from_file(block.word_style) or get_default_style()
# 2. 解析 block.style
custom_style = json.loads(block.style) if block.style else {}
# 3. 合并样式:custom_style 覆盖 base_style(仅在内存中)
final_style = {**base_style, **custom_style}
return final_style
# 注意:样式文件 (default.json) 始终保持不变
# 只有 block.style 字段会存储用户的自定义覆盖值
数据流示例:
┌─────────────────────┐
│ 样式文件 (只读) │
│ default.json │
│ { │
│ "Normal": { │
│ "font": "宋体" │
│ "size": 12 │
│ } │
│ } │
└──────────┬──────────┘
│
│ 读取 (不修改)
↓
┌─────────────────────┐ ┌──────────────────┐
│ 数据库 Block │ │ 渲染输出 │
│ word_style: "Normal"│ → │ font: "黑体" │
│ style: { │ 合并 │ size: 12 │
│ "font": "黑体" │ → │ (黑体覆盖宋体) │
│ } │ │ │
└─────────────────────┘ └──────────────────┘
样式文件依然是:
{
"Normal": {
"font": "宋体", ← 未改变
"size": 12
}
}
通用属性(所有 Block 类型)✅ 可自定义
{
"font_name": "宋体", // 字体名称 ✅
"font_size": 12.0, // 字号(磅)✅
"bold": true, // 是否粗体 ✅
"italic": false, // 是否斜体 ✅
"underline": false, // 是否下划线 ✅
"color": "000000", // 文字颜色(十六进制)✅
"align": "left" // 对齐方式:"left", "center", "right", "justify" ✅
}
段落特有属性 ❌ 暂不支持自定义
{
"line_spacing": 1.5, // 行距 ❌
"indent_first": 0, // 首行缩进(磅)❌
"indent_left": 0, // 左缩进(磅)❌
"indent_right": 0, // 右缩进(磅)❌
"space_before": 12.0, // 段前间距(磅)❌
"space_after": 6.0 // 段后间距(磅)❌
}
注意: 段落特有属性目前前端不能自定义,这些属性从 Word 提取时保存,渲染时使用,但用户暂时无法编辑。
表格特有属性(表格级)❌ 使用默认样式
{
"table_width": 100, // 表格宽度 ✅ 可修改
"table_width_unit": "percent",// 宽度单位 ✅ 可修改
"align": "left", // 表格对齐 ❌
"border_style": "single", // 边框样式 ❌
"border_width": 0.5, // 边框宽度 ❌
"border_color": "000000", // 边框颜色 ❌
"cell_padding": 2.0, // 单元格内边距 ❌
"header_bold": true, // 表头是否加粗 ❌
"header_bg_color": "F0F0F0" // 表头背景色 ❌
}
注意: 表格样式使用默认样式,用户可修改表格宽度和列宽(见 4.3.2 章节),但不能修改边框、内边距等样式。
图片特有属性 ✅ 可自定义
{
"width": 10.0, // 宽度 ✅
"height": 7.0, // 高度 ✅
"unit": "cm", // 单位:"cm", "inch", "px" ✅
"align": "center" // 对齐:"left", "center", "right" ✅
}
元数据存储(推荐): 不存储 children_ids,只存储 parent_id,由前端/API 动态计算
{
"parent_id": "block-h1-0"
// children_ids 不存储,按需动态计算
}
优点:
import sqlite3
import json
def build_toc_tree(db_path: str) -> list:
"""构建完整目录树(一次 O(n) 遍历)"""
conn = sqlite3.connect(db_path)
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
# 获取所有标题
cursor.execute("""
SELECT id, level, "index", content, metadata, block_order
FROM document_blocks
WHERE type = 'heading'
ORDER BY block_order
""")
headings = []
for row in cursor.fetchall():
metadata = json.loads(row['metadata'] or '{}')
headings.append({
'id': row['id'],
'level': row['level'],
'index': row['index'],
'content': row['content'],
'parent_id': metadata.get('parent_id'),
'children': []
})
# 构建映射
heading_map = {h['id']: h for h in headings}
# 构建树
tree = []
for h in headings:
parent_id = h['parent_id']
if parent_id and parent_id in heading_map:
heading_map[parent_id]['children'].append(h)
else:
tree.append(h)
conn.close()
return tree
async function loadDocumentTOC(documentId) {
// 1. 获取所有标题块
const response = await fetch(`/api/v1/documents/${documentId}/blocks?type=heading`);
const { blocks } = await response.json();
// 2. 构建树形结构
const map = new Map(blocks.map(h => [h.id, { ...h, children: [] }]));
const tree = [];
blocks.forEach(h => {
const node = map.get(h.id);
const parentId = h.metadata?.parent_id;
if (parentId && map.has(parentId)) {
map.get(parentId).children.push(node);
} else {
tree.push(node);
}
});
return tree;
}
// 渲染目录
function renderTOC(tree) {
return (
<ul className="toc">
{tree.map(item => (
<li key={item.id}>
<a href={`#${item.id}`} onClick={(e) => {
e.preventDefault();
scrollToBlock(item.id);
}}>
{item.content}
</a>
{item.children.length > 0 && renderTOC(item.children)}
</li>
))}
</ul>
);
}
-- 获取所有块(按顺序)
SELECT * FROM document_blocks ORDER BY block_order;
-- 按类型获取块
SELECT * FROM document_blocks WHERE type = 'heading' ORDER BY block_order;
-- 获取单个块
SELECT * FROM document_blocks WHERE id = 'block-h1-0';
-- 获取某标题下的所有内容
SELECT * FROM document_blocks
WHERE json_extract(metadata, '$.parent_heading_id') = 'block-h1-0'
ORDER BY block_order;
-- 搜索段落内容
SELECT * FROM document_blocks
WHERE type = 'paragraph' AND content LIKE '%关键词%'
ORDER BY block_order;
-- 搜索表格内容
SELECT * FROM document_blocks
WHERE type = 'table' AND content LIKE '%关键词%'
ORDER BY block_order;
-- 全文搜索(所有类型)
SELECT * FROM document_blocks
WHERE content LIKE '%关键词%'
ORDER BY block_order;
-- 搜索标题
SELECT * FROM document_blocks
WHERE type = 'heading' AND content LIKE '%关键词%'
ORDER BY block_order, level;
已创建的索引可以加速查询:
-- 已创建的索引
CREATE INDEX idx_block_order ON document_blocks(block_order); -- 顺序查询
CREATE INDEX idx_type ON document_blocks(type); -- 类型筛选
CREATE INDEX idx_level ON document_blocks(level); -- 标题级别
async def create_document(self, data: CreateDocumentRequest) -> Document:
# 1. 下载并解析 Word 文档
word_content = await _download_word(data.file_url)
# 2. 提取为结构化 Blocks
blocks = extract_blocks_from_word(word_content)
# 3. 创建 SQLite 数据库
db_path = f"tmp/{data.user_id}/sqlite/{doc_id}.db"
Path(db_path).parent.mkdir(parents=True, exist_ok=True)
# 4. 初始化数据库并写入 Blocks
self._init_content_db(db_path, blocks)
# 5. 创建数据库记录
doc = Document(
id=doc_id,
content_db_path=db_path,
user_id=data.user_id,
created_at=datetime.now(timezone.utc)
)
self.db.add(doc)
await self.db.commit()
return doc
async def get_document(self, document_id: str) -> DocumentResponse:
# 1. 查询文档记录
doc = await self.db.get(Document, document_id)
if not doc:
raise DocumentNotFoundError(document_id)
# 2. 从 SQLite 加载 blocks
blocks = self._load_blocks_from_db(doc.content_db_path)
return {
"id": doc.id,
"blocks": blocks,
"dbPath": doc.content_db_path
}
async def update_document(self, document_id: str, updates: dict) -> Document:
doc = await self.db.get(Document, document_id)
# 按 Block ID 更新
if 'block_updates' in updates:
self._update_blocks_by_id(doc.content_db_path, updates['block_updates'])
doc.updated_at = datetime.now(timezone.utc)
await self.db.commit()
return doc
async def export_to_word(self, document_id: str) -> str:
# 1. 获取文档
doc = await self.db.get(Document, document_id)
# 2. 从 SQLite 加载 blocks
blocks = self._load_blocks_from_db(doc.content_db_path)
# 3. 生成 Word 文档
word_path = f"tmp/{doc.user_id}/{date.today()}/{document_id}.docx"
generate_word_from_blocks(blocks, word_path)
return word_path
if doc.format == "sqlite":
self._update_blocks_by_level_index(doc.content_db_path, data.blocks)
else:
doc.content = self._apply_block_updates(doc.content, data.blocks)
elif hasattr(data, 'block_updates') and data.block_updates:
# 模式3:按 Block ID 精确更新(新增)
if doc.format == "sqlite":
self._update_blocks_by_id(doc.content_db_path, data.block_updates)
doc.updated_at = datetime.now(timezone.utc)
await self.db.commit()
await self.db.refresh(doc)
return doc
def _update_blocks_by_id(self, db_path: str, updates: list):
"""按 Block ID 精确更新"""
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
for update in updates:
block_id = update['id']
fields = []
values = []
if 'content' in update:
fields.append('content = ?')
values.append(update['content'])
if 'style' in update:
fields.append('style = ?')
values.append(json.dumps(update['style']))
if 'metadata' in update:
fields.append('metadata = ?')
values.append(json.dumps(update['metadata']))
if fields:
values.append(block_id)
cursor.execute(f"""
UPDATE document_blocks
SET {', '.join(fields)}
WHERE id = ?
""", values)
conn.commit()
conn.close()
## 九、API 端点扩展
### 9.1 新增 Blocks 操作端点
```python
# app/api/v1/document_blocks.py
from fastapi import APIRouter, Depends, Query
from typing import Optional
router = APIRouter(prefix="/documents/{documentId}/blocks", tags=["Document Blocks"])
@router.get("", summary="获取文档所有 Blocks")
async def list_blocks(
documentId: str,
type: Optional[str] = Query(None, description="按类型筛选:heading, paragraph, table, image"),
parentId: Optional[str] = Query(None, description="按父标题 ID 筛选"),
):
"""
获取文档的所有内容块
**查询参数:**
- `type`: 按类型筛选
- `parentId`: 获取某标题下的所有内容
"""
doc = await document_service.get_document(documentId)
if doc.format != "sqlite":
raise ValueError("Document format is not SQLite")
conn = sqlite3.connect(doc.content_db_path)
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
# 构建查询
where_clauses = []
params = []
if type:
where_clauses.append('type = ?')
params.append(type)
if parentId:
where_clauses.append('json_extract(metadata, "$.parent_heading_id") = ?')
params.append(parentId)
where_sql = f"WHERE {' AND '.join(where_clauses)}" if where_clauses else ""
cursor.execute(f"""
SELECT * FROM document_blocks
{where_sql}
ORDER BY block_order
""", params)
blocks = []
for row in cursor.fetchall():
block = dict(row)
block['style'] = json.loads(block['style'] or '{}')
block['metadata'] = json.loads(block['metadata'] or '{}')
if block['type'] == 'table':
block['content'] = json.loads(block['content'])
blocks.append(block)
conn.close()
return ok({"blocks": blocks})
@router.get("/{blockId}", summary="获取单个 Block")
async def get_block(documentId: str, blockId: str):
"""获取指定 ID 的 Block"""
doc = await document_service.get_document(documentId)
if doc.format != "sqlite":
raise ValueError("Document format is not SQLite")
conn = sqlite3.connect(doc.content_db_path)
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
cursor.execute("""
SELECT * FROM document_blocks
WHERE id = ?
""", (blockId,))
row = cursor.fetchone()
conn.close()
if not row:
raise BlockNotFoundError(blockId)
block = dict(row)
block['style'] = json.loads(block['style'] or '{}')
block['metadata'] = json.loads(block['metadata'] or '{}')
if block['type'] == 'table':
block['content'] = json.loads(block['content'])
return ok(block)
@router.put("/{blockId}", summary="更新单个 Block") async def update_block(
documentId: str,
blockId: str,
updates: dict
):
"""
更新指定 Block
**请求体示例:**
```json
{
"content": "更新后的内容",
"style": {"bold": true, "color": "FF0000"},
"metadata": {"parent_heading_id": "block-h1-0"}
}
```
"""
doc = await document_service.get_document(documentId)
if doc.format != "sqlite":
raise ValueError("Document format is not SQLite")
conn = sqlite3.connect(doc.content_db_path)
cursor = conn.cursor()
# 构建 UPDATE 语句
fields = []
values = []
if 'content' in updates:
fields.append('content = ?')
content = updates['content']
# 表格需要序列化
if isinstance(content, dict):
content = json.dumps(content)
values.append(content)
if 'style' in updates:
fields.append('style = ?')
values.append(json.dumps(updates['style']))
if 'metadata' in updates:
fields.append('metadata = ?')
values.append(json.dumps(updates['metadata']))
if not fields:
conn.close()
raise ValueError("No fields to update")
values.append(blockId)
cursor.execute(f"""
UPDATE document_blocks
SET {', '.join(fields)}
WHERE id = ?
""", values)
conn.commit()
conn.close()
if cursor.rowcount == 0:
raise BlockNotFoundError(blockId)
return ok({"blockId": blockId, "updatedAt": int(time.time() * 1000)})
@router.delete("/{blockId}", summary="删除单个 Block") async def delete_block(documentId: str, blockId: str):
"""删除指定 Block"""
doc = await document_service.get_document(documentId)
if doc.format != "sqlite":
raise ValueError("Document format is not SQLite")
conn = sqlite3.connect(doc.content_db_path)
cursor = conn.cursor()
cursor.execute("""
DELETE FROM document_blocks
WHERE id = ?
""", (blockId,))
conn.commit()
conn.close()
if cursor.rowcount == 0:
raise BlockNotFoundError(blockId)
return ok({"message": "Block deleted successfully"})
@router.get("/search", summary="搜索文档内容") async def search_blocks(
documentId: str,
q: str = Query(..., description="搜索关键词"),
limit: int = Query(50, ge=1, le=200)
):
"""全文搜索文档内容"""
doc = await document_service.get_document(documentId)
if doc.format != "sqlite":
raise ValueError("Document format is not SQLite")
conn = sqlite3.connect(doc.content_db_path)
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
cursor.execute("""
SELECT * FROM document_blocks
WHERE content LIKE ?
ORDER BY block_order
LIMIT ?
""", (f'%{q}%', limit))
results = []
for row in cursor.fetchall():
block = dict(row)
block['style'] = json.loads(block['style'] or '{}')
block['metadata'] = json.loads(block['metadata'] or '{}')
results.append(block)
conn.close()
return ok({"results": results, "total": len(results)})
@router.get("/toc", summary="获取文档目录树")
async def get_toc(documentId: str):
"""获取文档的目录树结构"""
doc = await document_service.get_document(documentId)
if doc.format != "sqlite":
raise ValueError("Document format is not SQLite")
conn = sqlite3.connect(doc.content_db_path)
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
# 获取所有标题
cursor.execute("""
SELECT id, level, "index", content, metadata, block_order
FROM document_blocks
WHERE type = 'heading'
ORDER BY block_order
""")
headings = []
for row in cursor.fetchall():
metadata = json.loads(row['metadata'] or '{}')
headings.append({
'id': row['id'],
'level': row['level'],
'index': row['index'],
'content': row['content'],
'parent_id': metadata.get('parent_id'),
'children': []
})
conn.close()
# 构建树
heading_map = {h['id']: h for h in headings}
tree = []
for h in headings:
parent_id = h['parent_id']
if parent_id and parent_id in heading_map:
heading_map[parent_id]['children'].append(h)
else:
tree.append(h)
return ok({"toc": tree})
@router.get("/stats", summary="获取文档统计信息")
async def get_stats(documentId: str):
"""获取文档的统计信息"""
doc = await document_service.get_document(documentId)
if doc.format != "sqlite":
raise ValueError("Document format is not SQLite")
conn = sqlite3.connect(doc.content_db_path)
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
# 统计各类型 Block 数量
cursor.execute("""
SELECT type, COUNT(*) as count
FROM document_blocks
GROUP BY type
""")
type_stats = {row['type']: row['count'] for row in cursor.fetchall()}
# 统计标题层级
cursor.execute("""
SELECT level, COUNT(*) as count
FROM document_blocks
WHERE type = 'heading'
GROUP BY level
ORDER BY level
""")
heading_stats = [dict(row) for row in cursor.fetchall()]
# 总块数
cursor.execute("SELECT COUNT(*) as total FROM document_blocks")
total_blocks = cursor.fetchone()['total']
conn.close()
return ok({
"typeStats": type_stats,
"headingStats": heading_stats,
"totalBlocks": total_blocks
})
// 1. 加载文档基本信息
const response = await fetch('/api/v1/documents/doc-123');
const { id, format, dbPath, content, blocks } = await response.json();
if (format === 'sqlite') {
// 2. 如果响应已包含 blocks,直接使用
if (blocks) {
renderDocument(blocks);
} else {
// 3. 否则单独获取 blocks
const blocksResp = await fetch(`/api/v1/documents/${id}/blocks`);
const { blocks } = await blocksResp.json();
renderDocument(blocks);
}
// 4. 获取目录树
const tocResp = await fetch(`/api/v1/documents/${id}/blocks/toc`);
const { toc } = await tocResp.json();
renderTOC(toc);
}
function renderDocument(blocks) {
const container = document.getElementById('document-content');
blocks.forEach(block => {
let element;
switch (block.type) {
case 'heading':
element = document.createElement(`h${block.level}`);
element.id = block.id;
// 标题支持富文本格式
if (Array.isArray(block.content)) {
element.innerHTML = renderRichText(block.content);
} else {
element.textContent = block.content;
}
break;
case 'paragraph':
element = document.createElement('p');
element.id = block.id;
// 段落支持富文本格式
if (Array.isArray(block.content)) {
element.innerHTML = renderRichText(block.content);
} else {
element.textContent = block.content; // 简单文本
}
break;
case 'table':
element = renderTable(block);
break;
case 'image':
element = document.createElement('img');
element.id = block.id;
element.src = block.content;
element.alt = block.metadata?.alt || '图片';
applyImageStyle(element, block.style);
break;
}
if (element) {
applyStyle(element, block.style, block.word_style);
container.appendChild(element);
}
});
}
function renderTable(block) {
const table = document.createElement('table');
table.id = block.id;
const tableData = block.content;
tableData.rows.forEach((row, rowIdx) => {
const tr = document.createElement('tr');
row.cells.forEach((cell, colIdx) => {
const td = document.createElement(rowIdx === 0 ? 'th' : 'td');
// 表格单元格也支持富文本格式
if (Array.isArray(cell.text)) {
td.innerHTML = renderRichText(cell.text);
} else {
td.textContent = cell.text;
}
if (cell.rowspan > 1) td.rowSpan = cell.rowspan;
if (cell.colspan > 1) td.colSpan = cell.colspan;
applyStyle(td, cell.style);
tr.appendChild(td);
});
table.appendChild(tr);
});
return table;
}
// 富文本渲染函数
function renderRichText(content) {
if (!Array.isArray(content)) {
return content; // 如果不是数组,直接返回文本
}
return content.map(segment => {
let text = segment.text || '';
const style = segment.style || {};
// 应用行内样式
let html = text;
if (style.bold) {
html = `<strong>${html}</strong>`;
}
if (style.italic) {
html = `<em>${html}</em>`;
}
if (style.underline) {
html = `<u>${html}</u>`;
}
if (style.color) {
html = `<span style="color: ${style.color}">${html}</span>`;
}
if (style.font_name || style.font_size) {
const inlineStyle = [];
if (style.font_name) inlineStyle.push(`font-family: ${style.font_name}`);
if (style.font_size) inlineStyle.push(`font-size: ${style.font_size}pt`);
html = `<span style="${inlineStyle.join('; ')}">${html}</span>`;
}
return html;
}).join('');
}
async function searchDocument(documentId, keyword) {
const response = await fetch(
`/api/v1/documents/${documentId}/blocks/search?q=${encodeURIComponent(keyword)}`
);
const { results } = await response.json();
// 渲染搜索结果
renderSearchResults(results);
// 高亮第一个结果
if (results.length > 0) {
scrollToBlock(results[0].id);
}
}
function renderSearchResults(results) {
const container = document.getElementById('search-results');
container.innerHTML = '';
results.forEach(block => {
const item = document.createElement('div');
item.className = 'search-result-item';
item.innerHTML = `
<div class="result-type">${block.type}</div>
<div class="result-content">${highlightKeyword(block.content)}</div>
<button onclick="scrollToBlock('${block.id}')">跳转</button>
`;
container.appendChild(item);
});
}
function renderTOC(toc) {
return (
<ul className="toc">
{toc.map(item => (
<li key={item.id}>
<a
href={`#${item.id}`}
onClick={(e) => {
e.preventDefault();
scrollToBlock(item.id);
}}
>
{item.content}
</a>
{item.children && item.children.length > 0 && (
renderTOC(item.children)
)}
</li>
))}
</ul>
);
}
function scrollToBlock(blockId) {
const element = document.getElementById(blockId);
if (element) {
element.scrollIntoView({
behavior: 'smooth',
block: 'start'
});
// 高亮显示
element.classList.add('highlight');
setTimeout(() => {
element.classList.remove('highlight');
}, 2000);
}
}
async function updateBlock(documentId, blockId, updates) {
const response = await fetch(
`/api/v1/documents/${documentId}/blocks/${blockId}`,
{
method: 'PUT',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify(updates)
}
);
const result = await response.json();
if (result.code === 0) {
console.log('Block updated successfully');
// 重新加载该 Block
reloadBlock(documentId, blockId);
}
}
// 使用示例
await updateBlock('doc-123', 'block-p-5', {
content: '更新后的段落内容',
style: { bold: true, color: 'FF0000' }
});
现有索引已足够应对大部分查询场景,如需进一步优化可添加复合索引:
-- 复合索引(按类型和位置查询)
CREATE INDEX idx_type_order ON document_blocks(type, block_order);
-- 按父标题查询优化
CREATE INDEX idx_parent_heading ON document_blocks(
(json_extract(metadata, '$.parent_heading_id')),
block_order
);
-- 使用事务批量更新
BEGIN TRANSACTION;
UPDATE document_blocks SET ... WHERE id = 'block-1';
UPDATE document_blocks SET ... WHERE id = 'block-2';
UPDATE document_blocks SET ... WHERE id = 'block-3';
COMMIT;
启用 WAL 模式提升并发性能:
PRAGMA journal_mode=WAL;
PRAGMA synchronous=NORMAL;
PRAGMA cache_size=-64000; -- 64MB 缓存
优势:
本设计实现了结构化的 SQLite 存储方案,满足以下核心需求:
✅ 区分内容类型:标题、正文、表格、图片分别存储,type 字段清晰标识
✅ 保留完整样式:word_style + style 双重机制,支持样式文件和自定义样式
✅ 记录层级关系:level + index + parent_id 构建完整的标题树
✅ 前端目录索引:提供专用 API 快速构建目录,支持跳转和导航
✅ 支持局部更新:按 Block ID 或 level+index 精确更新,性能优越
✅ 高效搜索:全文搜索、类型筛选、范围查询一应俱全
✅ 稀疏排序策略:避免频繁的 metadata 更新,提升插入和删除性能
文档版本:v1.0
创建日期:2026-07-02
关联文档: