sqlite-implementation-plan.md 40 KB
SQLite 内容存储实施方案
创建日期: 2026-07-02
关联文档: content-sqlite-design.md · text-editor-backend-api.md · document-management-design.md · export-doc-content-mapping.md
一、项目修改概览
根据 SQLite 内容存储设计，需要将当前的 Markdown 存储方案升级为结构化的 SQLite + Block 存储方案。主要修改包括：
1.1 核心改动
| 改动类别 | 影响范围 | 复杂度 |
|---|---|---|
| 数据模型修改 | documents 表删除 content 字段，添加 content_db_path | 低 |
| 新建 Block 存储 | 创建 ContentDB 类和 SQLite 操作逻辑 | 中 |
| Word 解析重构 | 直接解析为 Block 输出（不再生成 Markdown） | 高 |
| API 扩展 | 新增 Blocks 操作端点 | 中 |
| 导出服务重构 | 从 SQLite 读取 Blocks 并导出（删除 Markdown 处理） | 高 |
1.2 设计原则
- 纯 SQLite 存储: 完全移除 Markdown 相关代码和字段
- 结构化优先: 所有内容以 Block 形式存储
- 性能优化: 稀疏排序策略，减少 metadata 更新
- 富文本支持: 内容支持纯文本或 JSON 格式
二、需要修改的文件清单
2.1 数据库层
新建文件
app/services/content_db.py- SQLite 内容数据库操作类create_content_db()- 创建新的内容数据库insert_blocks()- 批量插入 blocksget_blocks()- 查询所有 blocksget_block_by_id()- 按 ID 查询单个 blockupdate_block()- 更新单个 blockdelete_block()- 删除单个 blocksearch_blocks()- 全文搜索get_toc()- 构建目录树
app/services/word_parser.py- Word 文档解析为 Blocksparse_word_to_blocks()- 主入口函数_extract_heading()- 提取标题块_extract_paragraph()- 提取段落块_extract_table()- 提取表格块_extract_image()- 提取图片块_build_rich_text()- 构建富文本格式_detect_style()- 检测样式
migrations/versions/004_remove_content_field.py- 数据库迁移脚本- 删除
content字段从 documents 表 - 添加
content_db_path字段（必填） - 删除
format字段（不再需要区分格式）
- 删除
修改文件
app/models/document.py- 文档模型 ```python删除字段：
content: Mapped[str] = mapped_column(Text, nullable=False, default="")
format: Mapped[str] = mapped_column(String(32), nullable=False, default="markdown")
# 添加字段： content_db_path: Mapped[str] = mapped_column(String(1024), nullable=False)
---
### 2.2 服务层
#### 修改文件
5. **`app/services/document_service.py`** - 文档服务
- `create_document()` - 修改逻辑：
- 下载 Word 文档
- 调用 `word_parser.parse_word_to_blocks()` 生成 blocks
- 创建 SQLite 数据库文件
- 插入 blocks 到 content.db
- 保存 content_db_path 到 documents 表
- `get_document()` - 修改响应：
- 从 content.db 读取 blocks 并返回
- `update_document()` - 修改逻辑：
- 更新 SQLite 中的 blocks
- **删除所有 Markdown 相关函数**：
- `_docx_to_markdown()` - 删除
- `_apply_block_updates()` - 删除
- 所有 Markdown 解析辅助函数 - 删除
6. **`app/services/export_service.py`** - 导出服务
- `export_doc()` - 修改逻辑：
- 从 content.db 读取 blocks（不再处理 Markdown）
- 将 blocks 转换为 Word 样式数据
- 使用新的 `blocks_to_docx_bytes()` 生成 .doc 文件
- **删除函数**：
- `markdown_to_docx_bytes()` - 删除
- `DocxRenderer` 中的 Markdown 解析逻辑 - 删除
- 新增函数：
- `blocks_to_docx_bytes()` - 将 blocks 列表转换为 Word 文档
- `render_heading_block()` - 渲染标题块
- `render_paragraph_block()` - 渲染段落块（支持富文本）
- `render_table_block()` - 渲染表格块
- `render_image_block()` - 渲染图片块
---
### 2.3 API 层
#### 新建文件
7. **`app/api/v1/blocks.py`** - Blocks 操作 API
- `GET /documents/{id}/blocks` - 获取文档的所有 blocks
- `GET /documents/{id}/blocks/{blockId}` - 获取单个 block
- `PUT /documents/{id}/blocks/{blockId}` - 更新单个 block
- `DELETE /documents/{id}/blocks/{blockId}` - 删除单个 block
- `POST /documents/{id}/blocks` - 插入新 block
- `GET /documents/{id}/blocks/search` - 搜索 blocks
- `GET /documents/{id}/blocks/toc` - 获取目录树
- `GET /documents/{id}/blocks/stats` - 获取统计信息
#### 修改文件
8. **`app/api/v1/documents.py`** - 文档 API
- `POST /documents` - 创建文档
```python
# 请求保持不变
{
"userId": "user-xyz",
"fileUrl": "https://example.com/report.docx",
"sessionId": "chat-session-123"
}
# 响应修改
{
"code": 0,
"data": {
"documentId": "doc-abc123",
"contentDbPath": "tmp/user-xyz/2026-07-02/doc-abc123.db", # 新增
"createdAt": 1680000000000
}
}
```
- `GET /documents/{id}` - 获取文档详情
```python
# 响应修改（删除 format 和 content）
{
"code": 0,
"data": {
"id": "doc-123",
"contentDbPath": "tmp/user-xyz/2026-07-02/doc-123.db",
"blocks": [ # 直接返回 blocks 列表
{
"id": "block-h1-0",
"blockOrder": 0,
"type": "heading",
"level": 1,
"index": 0,
"content": "文档标题",
"wordStyle": "Heading 1",
"style": {},
"metadata": {"parentHeadingId": null}
},
{
"id": "block-p-100",
"blockOrder": 100,
"type": "paragraph",
"level": 0,
"index": 0,
"content": "段落内容",
"wordStyle": "Normal",
"style": {},
"metadata": {}
}
],
"sessionId": "chat-session-123",
"userId": "user-xyz",
"createdAt": 1680000000000,
"updatedAt": 1680000100000
}
}
```
- `PUT /documents/{id}` - 更新文档
```python
# 请求修改（通过 blocks API 更新，此接口标记为废弃）
# 推荐使用: PUT /documents/{id}/blocks/{blockId}
```
- `DELETE /documents/{sessionId}` - 删除会话的所有文档
```python
# 保持不变，同时删除关联的 SQLite 文件
{
"code": 0,
"message": "Deleted 3 document(s) successfully"
}
```
- `GET /documents` - 获取文档列表
```python
# 响应修改（删除 content 字段）
{
"code": 0,
"data": {
"documents": [
{
"id": "doc-abc123",
"sessionId": "chat-session-123",
"createdAt": 1680000000000,
"updatedAt": 1680000100000
}
],
"pagination": {
"page": 1,
"pageSize": 20,
"total": 100,
"totalPages": 5
}
}
}
```
9. **`app/api/v1/export.py`** - 导出 API
- `POST /export/doc` - 导出 .doc 文件
```python
# 请求保持不变
{
"documentId": "doc-abc123",
"styleId": null
}
# 响应保持不变
{
"code": 0,
"data": {
"recordId": "rec-abc123",
"downloadUrl": "http://192.168.0.200:8000/api/v1/export/records/rec-abc123/download",
"fileName": "2026年Q2季度报告_1680000000000.doc",
"styleId": "default",
"warning": null
}
}
# 内部逻辑：从 SQLite 读取 blocks 而非 Markdown
```
10. **`app/main.py`** - 主应用
```python
# 注册新路由
from app.api.v1 import blocks
app.include_router(blocks.router, prefix="/api/v1")
```
#### API 端点完整列表
| 端点 | 方法 | 说明 | 状态 |
|------|------|------|------|
| `/documents` | POST | 创建文档 | 修改 |
| `/documents` | GET | 获取文档列表 | 修改 |
| `/documents/{id}` | GET | 获取文档详情 | 修改 |
| `/documents/{id}` | PUT | 更新文档 | 废弃 → 使用 Blocks API |
| `/documents/{sessionId}` | DELETE | 删除会话文档 | 修改 |
| `/documents/{id}/blocks` | GET | 获取所有 blocks | **新增** |
| `/documents/{id}/blocks` | POST | 插入 block | **新增** |
| `/documents/{id}/blocks/{blockId}` | GET | 获取单个 block | **新增** |
| `/documents/{id}/blocks/{blockId}` | PUT | 更新 block | **新增** |
| `/documents/{id}/blocks/{blockId}` | DELETE | 删除 block | **新增** |
| `/documents/{id}/blocks/search` | GET | 搜索 blocks | **新增** |
| `/documents/{id}/blocks/toc` | GET | 获取目录树 | **新增** |
| `/documents/{id}/blocks/stats` | GET | 获取统计信息 | **新增** |
| `/export/doc` | POST | 导出 .doc | 修改 |
| `/export/records` | GET | 获取导出记录 | 不变 |
| `/export/records/{id}/download` | GET | 下载文件 | 不变 |
| `/export/records/{id}` | DELETE | 删除记录 | 不变 |
---
### 2.4 Schema 层
#### 新建文件
10. **`app/schemas/block.py`** - Block 相关 Schema
```python
class BlockBase(BaseModel):
id: str
block_order: int
type: str # heading, paragraph, table, image
level: int
index: int
content: str | dict # 字符串或 JSON
word_style: str
style: dict
metadata: dict
class BlockUpdate(BaseModel):
content: Optional[str | dict] = None
style: Optional[dict] = None
class TOCItem(BaseModel):
id: str
level: int
content: str
children: list['TOCItem'] = []
class BlockSearchResult(BaseModel):
blocks: list[BlockBase]
total: int
```
#### 修改文件
11. **`app/schemas/document.py`** - 文档 Schema
```python
class DocumentResponse(BaseModel):
id: str
content_db_path: str # 必填
blocks: list[BlockBase] # 直接返回 blocks
session_id: str
user_id: str
created_at: datetime
updated_at: datetime
# 删除字段：
# content: Optional[str] # 删除
# format: str # 删除
```
---
## 三、详细修改步骤
### 阶段 1：基础架构（1-2天）
#### 步骤 1.1：数据库迁移
```bash
# 生成迁移文件
alembic revision -m "remove_content_field_add_content_db_path"
迁移脚本内容：
def upgrade():
# 删除 content 和 format 字段
op.drop_column('documents', 'content')
op.drop_column('documents', 'format')
# 添加 content_db_path 字段（必填）
op.add_column('documents', sa.Column('content_db_path', sa.String(1024), nullable=False))
def downgrade():
# 恢复原字段
op.add_column('documents', sa.Column('content', sa.Text(), nullable=False, server_default=''))
op.add_column('documents', sa.Column('format', sa.String(32), nullable=False, server_default='markdown'))
# 删除新字段
op.drop_column('documents', 'content_db_path')
步骤 1.2：创建 ContentDB 类
文件: app/services/content_db.py
import sqlite3
import json
from pathlib import Path
from typing import Optional
class ContentDB:
"""SQLite 内容数据库操作类"""
def __init__(self, db_path: str):
self.db_path = Path(db_path)
self.conn: Optional[sqlite3.Connection] = None
def connect(self):
"""连接数据库"""
self.conn = sqlite3.connect(str(self.db_path))
self.conn.row_factory = sqlite3.Row
return self
def close(self):
"""关闭连接"""
if self.conn:
self.conn.close()
self.conn = None
def create_tables(self):
"""创建 document_blocks 表"""
self.conn.execute("""
CREATE TABLE IF NOT EXISTS document_blocks (
id TEXT PRIMARY KEY,
block_order INTEGER NOT NULL,
type TEXT NOT NULL,
level INTEGER DEFAULT 0,
"index" INTEGER DEFAULT 0,
content TEXT NOT NULL,
word_style TEXT DEFAULT '',
style TEXT DEFAULT '{}',
metadata TEXT DEFAULT '{}'
)
""")
# 创建索引
self.conn.execute('CREATE INDEX IF NOT EXISTS idx_block_order ON document_blocks(block_order)')
self.conn.execute('CREATE INDEX IF NOT EXISTS idx_type ON document_blocks(type)')
self.conn.execute('CREATE INDEX IF NOT EXISTS idx_level ON document_blocks(level)')
self.conn.commit()
def insert_blocks(self, blocks: list[dict]):
"""批量插入 blocks"""
for block in blocks:
content = block['content']
if isinstance(content, (dict, list)):
content = json.dumps(content, ensure_ascii=False)
self.conn.execute("""
INSERT INTO document_blocks
(id, block_order, type, level, "index", content, word_style, style, metadata)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
""", (
block['id'],
block['block_order'],
block['type'],
block.get('level', 0),
block.get('index', 0),
content,
block.get('word_style', ''),
json.dumps(block.get('style', {}), ensure_ascii=False),
json.dumps(block.get('metadata', {}), ensure_ascii=False)
))
self.conn.commit()
def get_blocks(self) -> list[dict]:
"""查询所有 blocks"""
cursor = self.conn.execute("""
SELECT * FROM document_blocks
ORDER BY block_order
""")
rows = cursor.fetchall()
return [self._row_to_dict(row) for row in rows]
def _row_to_dict(self, row) -> dict:
"""将 sqlite3.Row 转换为字典"""
d = dict(row)
# 解析 JSON 字段
d['style'] = json.loads(d['style']) if d['style'] else {}
d['metadata'] = json.loads(d['metadata']) if d['metadata'] else {}
# content 可能是 JSON
try:
d['content'] = json.loads(d['content'])
except (json.JSONDecodeError, TypeError):
pass # 保持原字符串
return d
阶段 2：Word 解析改造（2-3天）
步骤 2.1：创建 Word Parser
文件: app/services/word_parser.py
from docx import Document as DocxDocument
from pathlib import Path
def parse_word_to_blocks(docx_path: Path) -> list[dict]:
"""将 Word 文档解析为 Block 列表"""
doc = DocxDocument(str(docx_path))
blocks = []
block_order = 0
# 标题计数器（按 level 分别计数）
heading_counters = {1: 0, 2: 0, 3: 0, 4: 0, 5: 0, 6: 0}
parent_stack = [] # 维护父标题栈
# 遍历段落和表格
for elem in doc.element.body:
tag = elem.tag
if tag.endswith('p'): # 段落
para = _find_paragraph(doc, elem)
if para is None:
continue
style_name = para.style.name if para.style else "Normal"
# 判断是否为标题
if style_name.startswith("Heading "):
try:
level = int(style_name.split()[-1])
except ValueError:
level = 1
# 更新计数器
index = heading_counters[level]
heading_counters[level] += 1
# 重置更深层级的计数器
for l in range(level + 1, 7):
heading_counters[l] = 0
# 更新父标题栈
while parent_stack and parent_stack[-1]['level'] >= level:
parent_stack.pop()
parent_id = parent_stack[-1]['id'] if parent_stack else None
# 提取内容（支持富文本）
content = _extract_rich_text(para)
block = {
'id': f'block-h{level}-{block_order}',
'block_order': block_order,
'type': 'heading',
'level': level,
'index': index,
'content': content,
'word_style': style_name,
'style': {},
'metadata': {
'parent_heading_id': parent_id
}
}
blocks.append(block)
parent_stack.append({'id': block['id'], 'level': level})
block_order += 1
elif style_name in ("Normal", "No Spacing") or not style_name:
# 普通段落
content = _extract_rich_text(para)
if content: # 跳过空段落
block = {
'id': f'block-p-{block_order}',
'block_order': block_order,
'type': 'paragraph',
'level': 0,
'index': 0,
'content': content,
'word_style': style_name,
'style': {},
'metadata': {}
}
blocks.append(block)
block_order += 1
elif tag.endswith('tbl'): # 表格
table = _find_table(doc, elem)
if table is None:
continue
table_content = _extract_table(table)
block = {
'id': f'block-table-{block_order}',
'block_order': block_order,
'type': 'table',
'level': 0,
'index': 0,
'content': table_content, # JSON 格式
'word_style': '',
'style': {},
'metadata': {}
}
blocks.append(block)
block_order += 1
return blocks
def _extract_rich_text(para) -> str | list:
"""提取段落的富文本内容"""
# 检查是否包含多种格式
has_format = any(
run.bold or run.italic or run.font.color.rgb
for run in para.runs
)
if not has_format:
# 简单文本
return para.text
# 富文本格式（JSON 数组）
segments = []
for run in para.runs:
if not run.text:
continue
style = {}
if run.bold:
style['bold'] = True
if run.italic:
style['italic'] = True
# ... 其他样式
segments.append({
'text': run.text,
'style': style
})
return segments if segments else para.text
---
### 阶段 3：服务层改造（2-3天）
#### 步骤 3.1：修改 DocumentService.create_document()
**修改前**：下载 Word → 解析为 Markdown → 存入 content 字段
**修改后**：下载 Word → 解析为 Blocks → 创建 SQLite → 存入 content_db_path
```python
async def create_document(self, data: CreateDocumentRequest) -> Document:
# 1. 下载 Word 文档
tmp_path = await _download_word(data.file_url)
# 2. 解析为 Blocks
from app.services.word_parser import parse_word_to_blocks
blocks = parse_word_to_blocks(tmp_path)
# 3. 创建 SQLite 数据库
user_id = data.user_id or "default-user"
today = date.today().strftime("%Y-%m-%d")
db_dir = Path(settings.temp_dir) / user_id / today
db_dir.mkdir(parents=True, exist_ok=True)
doc_id = f"doc-{uuid.uuid4().hex[:12]}"
db_path = db_dir / f"{doc_id}.db"
# 4. 写入 Blocks
from app.services.content_db import ContentDB
content_db = ContentDB(str(db_path))
content_db.connect()
content_db.create_tables()
content_db.insert_blocks(blocks)
content_db.close()
# 5. 创建文档记录（不再写入 content 字段）
doc = Document(
id=doc_id,
content_db_path=str(db_path), # 必填
session_id=data.session_id,
created_by=data.user_id,
)
self.db.add(doc)
await self.db.commit()
await self.db.refresh(doc)
# 6. 清理临时文件
tmp_path.unlink(missing_ok=True)
return doc
步骤 3.2：修改 DocumentService.get_document()
async def get_document(self, document_id: str) -> Document:
doc = await self._get_document_by_id(document_id)
# 直接从 SQLite 加载 blocks
from app.services.content_db import ContentDB
content_db = ContentDB(doc.content_db_path)
content_db.connect()
blocks = content_db.get_blocks()
content_db.close()
# 将 blocks 附加到 doc 对象（临时属性）
doc.blocks = blocks
return doc
阶段 4：API 层扩展（1-2天）
步骤 4.1：创建 Blocks API
文件: app/api/v1/blocks.py
from fastapi import APIRouter, Depends, Query
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.dependencies import get_db
from app.services.document_service import DocumentService
from app.services.content_db import ContentDB
router = APIRouter(prefix="/documents/{documentId}/blocks", tags=["Blocks"])
@router.get("", summary="获取文档的所有 blocks")
async def get_blocks(
documentId: str,
db: AsyncSession = Depends(get_db),
) -> dict:
svc = DocumentService(db)
doc = await svc.get_document(documentId)
content_db = ContentDB(doc.content_db_path)
content_db.connect()
blocks = content_db.get_blocks()
content_db.close()
return {"code": 0, "data": {"blocks": blocks}}
@router.get("/toc", summary="获取目录树")
async def get_toc(
documentId: str,
db: AsyncSession = Depends(get_db),
) -> dict:
svc = DocumentService(db)
doc = await svc.get_document(documentId)
content_db = ContentDB(doc.content_db_path)
content_db.connect()
# 获取所有标题块
cursor = content_db.conn.execute("""
SELECT id, level, "index", content, metadata
FROM document_blocks
WHERE type = 'heading'
ORDER BY block_order
""")
headings = [dict(row) for row in cursor.fetchall()]
content_db.close()
# 构建树形结构
toc = _build_toc_tree(headings)
return {"code": 0, "data": {"toc": toc}}
def _build_toc_tree(headings: list[dict]) -> list[dict]:
"""构建目录树"""
root = []
stack = []
for h in headings:
node = {
"id": h["id"],
"level": h["level"],
"content": h["content"],
"children": []
}
# 找到父节点
while stack and stack[-1]["level"] >= h["level"]:
stack.pop()
if stack:
stack[-1]["children"].append(node)
else:
root.append(node)
stack.append(node)
return root
阶段 5：导出服务适配（2-3天）
步骤 5.1：修改 export_service.py
新增函数: 从 Blocks 渲染 Word
def blocks_to_docx_bytes(blocks: list[dict], style_map: dict, style_data: dict) -> bytes:
"""将 Blocks 列表转换为 Word 文档字节流"""
doc = Document()
inject_styles_from_json(doc, style_data)
for block in blocks:
block_type = block['type']
content = block['content']
if block_type == 'heading':
_render_heading_block(doc, block, style_map)
elif block_type == 'paragraph':
_render_paragraph_block(doc, block, style_map)
elif block_type == 'table':
_render_table_block(doc, block, style_map)
elif block_type == 'image':
_render_image_block(doc, block)
buf = io.BytesIO()
doc.save(buf)
return buf.getvalue()
def _render_heading_block(doc, block, style_map):
"""渲染标题块"""
level = block['level']
content = block['content']
# content 可能是字符串或富文本数组
if isinstance(content, list):
para = doc.add_paragraph()
para.style = f"Heading {level}"
_render_rich_text(para, content)
else:
doc.add_heading(content, level=level)
def _render_paragraph_block(doc, block, style_map):
"""渲染段落块（支持富文本）"""
content = block['content']
style_name = block.get('word_style', 'Normal')
para = doc.add_paragraph()
# 应用样式
style_id = _resolve_style_id(style_map, style_name)
if style_id:
try:
para.style = doc.styles[style_id]
except KeyError:
pass
# 渲染内容
if isinstance(content, list):
_render_rich_text(para, content)
else:
para.add_run(content)
def _render_rich_text(para, segments: list):
"""渲染富文本格式"""
for seg in segments:
text = seg.get('text', '')
style = seg.get('style', {})
run = para.add_run(text)
if style.get('bold'):
run.bold = True
if style.get('italic'):
run.italic = True
if style.get('underline'):
run.underline = True
if style.get('color'):
run.font.color.rgb = RGBColor.from_string(style['color'])
if style.get('font_name'):
run.font.name = style['font_name']
if style.get('font_size'):
run.font.size = Pt(style['font_size'])
def _render_table_block(doc, block, style_map):
"""渲染表格块"""
table_data = block['content']
rows = table_data['rows']
if not rows:
return
# 创建表格
table = doc.add_table(rows=len(rows), cols=len(rows[0]['cells']))
table.style = 'Table Grid'
# 填充内容
for r_idx, row_data in enumerate(rows):
for c_idx, cell_data in enumerate(row_data['cells']):
cell = table.rows[r_idx].cells[c_idx]
cell_text = cell_data['text']
# 清空默认段落
cell.text = ''
para = cell.paragraphs[0]
# 渲染单元格内容（支持富文本）
if isinstance(cell_text, list):
_render_rich_text(para, cell_text)
else:
para.add_run(cell_text)
# 应用合并
if cell_data.get('rowspan', 1) > 1:
cell.merge(table.rows[r_idx + cell_data['rowspan'] - 1].cells[c_idx])
if cell_data.get('colspan', 1) > 1:
cell.merge(table.rows[r_idx].cells[c_idx + cell_data['colspan'] - 1])
步骤 5.2：修改 export.py
@router.post("/export/doc", summary="导出 .doc 文件")
async def export_document(
body: ExportDocRequest,
db: AsyncSession = Depends(get_db),
) -> dict:
doc_svc = DocumentService(db)
rec_svc = ExportRecordService(db)
# 1. 读取文档
doc = await doc_svc.get_document(body.document_id)
user_id = doc.created_by or "default-user"
# 2. 检查是否有最新记录可复用
latest = await rec_svc.get_latest_record(body.document_id)
if latest and doc.updated_at <= latest.created_at:
warning = check_quota(user_id)
return ok(ExportDocResponse(...).model_dump(by_alias=True))
# 3. 加载样式
style_data = load_style_file(body.style_id)
style_map = build_style_map(style_data)
actual_style_id = body.style_id or "default"
# 4. 从 SQLite 读取 blocks 并生成 .doc 字节流
try:
# 从 SQLite 读取 blocks
content_db = ContentDB(doc.content_db_path)
content_db.connect()
blocks = content_db.get_blocks()
content_db.close()
# Blocks → Word
doc_bytes = blocks_to_docx_bytes(blocks, style_map, style_data)
except Exception as exc:
raise ExportError(f"文档转换失败: {exc}") from exc
# 5-6. 写入文件、创建记录（与原逻辑相同）
# ...
四、测试要点
4.1 单元测试
ContentDB 测试 (
tests/test_content_db.py)- 创建数据库
- 插入、查询、更新、删除 blocks
- 搜索功能
- 目录树构建
Word Parser 测试 (
tests/test_word_parser.py)- 解析标题块
- 解析段落块（简单文本和富文本）
- 解析表格块
- 解析图片块
- 父子关系构建
Blocks 导出测试 (
tests/test_blocks_export.py)- Blocks → Word 转换
- 富文本格式保留
- 表格合并单元格
- 样式应用
4.2 集成测试
完整流程测试
async def test_full_workflow(): # 1. 上传 Word → 创建文档（SQLite） response = await client.post("/api/v1/documents", json={ "userId": "test-user", "fileUrl": "https://example.com/test.docx", "sessionId": "session-123" }) doc_id = response.json()["data"]["documentId"] # 2. 获取文档详情（包含 blocks） response = await client.get(f"/api/v1/documents/{doc_id}") assert "blocks" in response.json()["data"] assert "contentDbPath" in response.json()["data"] # 3. 获取目录树 response = await client.get(f"/api/v1/documents/{doc_id}/blocks/toc") assert "toc" in response.json()["data"] # 4. 导出 .doc response = await client.post("/api/v1/export/doc", json={ "documentId": doc_id }) assert response.json()["data"]["downloadUrl"]富文本测试
async def test_rich_text(): # 测试富文本块的创建和导出 # 验证粗体、斜体、颜色等格式正确保留表格测试
async def test_table_blocks(): # 测试表格块的解析和导出 # 验证合并单元格、表格样式正确保留
五、上线计划
5.1 发布策略
阶段 1：开发与测试（2-3周）
- 完成所有代码开发
- 单元测试和集成测试
- 内部测试环境验证
阶段 2：预发布环境验证（1周）
- 部署到预发布环境
- 团队全面测试
- 性能压测
阶段 3：生产环境发布（1天）
- 选择低峰时段发布
- 实时监控日志和性能
- 准备快速回滚方案
5.2 数据迁移方案
由于完全移除了 Markdown 存储，现有数据需要迁移：
选项 A：一次性迁移（推荐）
- 发布前：停服务，运行迁移脚本，将所有旧文档转为 SQLite 格式
- 优点：干净利落，无遗留数据
- 缺点：需要短暂停服
选项 B：废弃旧数据
- 发布后：旧文档标记为只读，不再支持编辑和导出
- 优点：无需迁移，快速上线
- 缺点：用户体验受影响
推荐使用选项 A，迁移脚本示例：
async def migrate_all_documents():
"""将所有文档从 Markdown 转为 SQLite"""
docs = await db.execute(select(Document))
for doc in docs.scalars().all():
if not doc.content:
continue
# 解析 Markdown 为 Blocks（简化处理）
blocks = markdown_to_blocks(doc.content)
# 创建 SQLite
db_path = create_content_db_for_document(doc.id, blocks)
# 更新记录
doc.content_db_path = db_path
doc.content = None # 清空
doc.format = "sqlite"
await db.commit()
5.3 回滚方案
如果发现严重问题，需要快速回滚：
- 代码回滚：Git revert 到上一个版本
- 数据库回滚：运行 alembic downgrade 恢复字段
- 数据恢复：从备份恢复 content 字段数据
六、风险评估与应对
6.1 技术风险
| 风险 | 影响 | 概率 | 应对措施 |
|---|---|---|---|
| Word 解析失败率上升 | 用户无法创建文档 | 中 | 充分测试各种 Word 格式；提供详细错误日志；降级提示 |
| SQLite 文件损坏 | 文档数据丢失 | 低 | WAL 模式；定期备份；文件校验 |
| 富文本格式不兼容 | 导出样式丢失 | 中 | 完善测试用例；提供降级方案（纯文本）；样式验证 |
| 性能问题（大文档） | 查询/导出变慢 | 中 | 分页加载；异步处理；索引优化；性能监控 |
| 数据迁移失败 | 旧文档不可用 | 高 | 充分测试迁移脚本；先备份再迁移；提供手动修复工具 |
6.2 业务风险
| 风险 | 影响 | 概率 | 应对措施 |
|---|---|---|---|
| 用户反馈不佳 | 影响产品口碑 | 低 | 充分测试；快速迭代；用户调研；提供详细文档 |
| 存储成本上升 | 运营成本增加 | 中 | 监控存储用量；设置配额；定期清理；数据压缩 |
| 功能兼容性问题 | 现有功能受影响 | 中 | 全面回归测试；API 保持一致；前端协同测试 |
| 迁移时间过长 | 影响服务可用性 | 中 | 预先评估数据量；选择低峰时段；分批迁移 |
七、时间估算
| 阶段 | 任务 | 预估时间 | 负责人 |
|---|---|---|---|
| 1 | 数据库迁移 + ContentDB | 1-2天 | 后端开发 |
| 2 | Word Parser 开发 | 2-3天 | 后端开发 |
| 3 | DocumentService 改造（删除 Markdown 代码） | 2-3天 | 后端开发 |
| 4 | Blocks API 开发 | 1-2天 | 后端开发 |
| 5 | 导出服务重构（删除 Markdown 处理） | 3-4天 | 后端开发 |
| 6 | 单元测试 | 2天 | 后端开发 |
| 7 | 集成测试 | 2天 | 测试工程师 |
| 8 | 数据迁移脚本开发与测试 | 2天 | 后端开发 |
| 9 | 文档编写 | 1天 | 后端开发 |
| 总计 | 16-21天 |
八、依赖项检查
8.1 Python 依赖
现有依赖已满足需求：
- ✅
python-docx- Word 文档解析 - ✅
sqlalchemy- ORM - ✅
sqlite3- SQLite 操作（Python 内置）
可删除的依赖：
- ❌
mistune- Markdown 解析（不再需要，可以删除）
8.2 系统依赖
- ✅ SQLite 3.x（系统自带）
- ✅ 磁盘空间（监控 tmp/ 目录）
九、需要删除的代码清单
为确保代码库清洁，以下是需要彻底删除的 Markdown 相关代码：
9.1 app/services/document_service.py
删除的函数（约 500+ 行）：
# Markdown 解析相关
def _docx_to_markdown(path: Path) -> str
def _render_run_with_format(run) -> str
def _render_para_with_inline_format(para) -> str
def _is_code_block(para) -> bool
def _convert_table_to_markdown(table) -> str
def _get_list_info(para) -> tuple[str, int]
def _has_bottom_border(para) -> bool
# Block 更新相关（旧逻辑）
def _apply_block_updates(content: str, blocks: list) -> str
# 下载解析函数（需要重构，不是删除）
async def _download_and_parse(file_url: str) -> str # 改为返回 blocks
删除的常量：
STANDARD_STYLES # 不再需要判断标准样式
9.2 app/services/export_service.py
删除的函数和类（约 300+ 行）：
# 旧的 Markdown 渲染器
class DocxRenderer(mistune.BaseRenderer) # 整个类删除
def markdown_to_docx_bytes(content: str, ...) -> bytes
# Markdown 相关的渲染方法
def heading(self, token: dict, state) -> str
def paragraph(self, token: dict, state) -> str
def block_quote(self, token: dict, state) -> str
def block_code(self, token: dict, state) -> str
def list(self, token: dict, state) -> str
def table(self, token: dict, state) -> str
# ... 所有 mistune 相关的 token 处理方法
保留但需重构：
# 这些函数保留，但改为从 blocks 生成
def blocks_to_docx_bytes(blocks: list[dict], ...) -> bytes # 新函数
def _render_heading_block(doc, block, style_map) # 新函数
def _render_paragraph_block(doc, block, style_map) # 新函数
def _render_table_block(doc, block, style_map) # 新函数
def _render_rich_text(para, segments: list) # 新函数
9.3 app/models/document.py
删除的字段：
content: Mapped[str] = mapped_column(Text, nullable=False, default="")
format: Mapped[str] = mapped_column(String(32), nullable=False, default="markdown")
9.4 app/schemas/document.py
删除的字段：
class UpdateDocumentRequest(BaseModel):
content: Optional[str] = None # 删除
blocks: Optional[list[BlockUpdate]] = None # 删除（改用新的 Block API）
class DocumentResponse(BaseModel):
content: Optional[str] = None # 删除
format: str # 删除
9.5 requirements.txt
可选删除：
mistune==3.0.2 # 如果没有其他地方使用，可以删除
9.6 清理检查清单
完成代码删除后，运行以下检查：
全局搜索残留引用：
# 搜索 markdown 关键字 grep -r "markdown" --include="*.py" app/ grep -r "mistune" --include="*.py" app/ grep -r "_docx_to_markdown" --include="*.py" app/ grep -r "DocxRenderer" --include="*.py" app/检查导入语句：
grep -r "import mistune" --include="*.py" app/ grep -r "from mistune" --include="*.py" app/运行测试：
pytest tests/ -v检查 Alembic 迁移：
alembic history alembic upgrade head
十、关键代码示例
9.1 稀疏排序初始化
def initialize_blocks_with_sparse_order(blocks: list[dict]) -> list[dict]:
"""为 blocks 初始化稀疏排序的 block_order"""
gap = 100 # 间隔
for i, block in enumerate(blocks):
block['block_order'] = i * gap
return blocks
9.2 插入 Block（保持稀疏排序）
def insert_block_between(content_db: ContentDB, after_id: str, new_block: dict):
"""在指定 block 后插入新 block"""
# 查询前后 block 的 block_order
cursor = content_db.conn.execute(
"SELECT block_order FROM document_blocks WHERE id = ?",
(after_id,)
)
after_order = cursor.fetchone()[0]
cursor = content_db.conn.execute(
"SELECT block_order FROM document_blocks WHERE block_order > ? ORDER BY block_order LIMIT 1",
(after_order,)
)
next_row = cursor.fetchone()
next_order = next_row[0] if next_row else after_order + 100
# 计算新 block_order
gap = next_order - after_order
if gap > 1:
# 有空间，插入中间
new_block['block_order'] = (after_order + next_order) // 2
else:
# 空间不足，触发重排（TODO）
raise Exception("需要重排序")
# 插入
content_db.insert_blocks([new_block])
十、总结
本方案提供了从当前 Markdown 存储向纯 SQLite + Block 存储的完整迁移路径。关键要点：
- 彻底重构 - 完全移除 Markdown 存储，采用纯 SQLite 方案
- 结构化存储 - 所有内容按 Block 类型分块存储，支持精确操作
- 富文本支持 - 内容支持 JSON 格式，保留完整样式信息
- 性能优化 - 稀疏排序、索引优化、批量操作
- 数据迁移 - 提供完整的迁移方案，确保旧数据平滑过渡
核心优势
相比原有的 Markdown 存储：
| 优势 | 说明 |
|---|---|
| 查询性能 | 索引查询代替全文扫描，速度提升 10-100 倍 |
| 精确更新 | 直接更新单个 Block，无需解析整个文档 |
| 结构化查询 | 原生支持"获取所有表格"、"按标题筛选"等查询 |
| 目录构建 | 通过 parent_id 关系，毫秒级构建树形目录 |
| 富文本支持 | 完整保留格式信息，支持粗体、斜体、颜色等 |
| 灵活扩展 | 易于添加新的 Block 类型和属性 |
注意事项
- 数据迁移风险 - 必须在发布前完成旧数据迁移，做好备份
- 代码清理 - 删除所有 Markdown 相关代码，避免混淆
- 充分测试 - Word 解析、富文本、表格等功能需要全面测试
- 性能监控 - 上线后密切监控查询性能和存储空间
- 文档更新 - 更新 API 文档和用户文档
总工期: 16-21 个工作日（含数据迁移）
建议优先级: 高（核心架构升级）
风险等级: 中高（涉及数据迁移和大量代码重构）
文档版本: v2.0
最后更新: 2026-07-02
作者: 项目团队
变更说明: 移除所有 Markdown 兼容性内容，采用纯 SQLite 方案