# 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 数据库层

#### 新建文件


1. **`app/services/content_db.py`** - SQLite 内容数据库操作类
   - `create_content_db()` - 创建新的内容数据库
   - `insert_blocks()` - 批量插入 blocks
   - `get_blocks()` - 查询所有 blocks
   - `get_block_by_id()` - 按 ID 查询单个 block
   - `update_block()` - 更新单个 block
   - `delete_block()` - 删除单个 block
   - `search_blocks()` - 全文搜索
   - `get_toc()` - 构建目录树

2. **`app/services/word_parser.py`** - Word 文档解析为 Blocks
   - `parse_word_to_blocks()` - 主入口函数
   - `_extract_heading()` - 提取标题块
   - `_extract_paragraph()` - 提取段落块
   - `_extract_table()` - 提取表格块
   - `_extract_image()` - 提取图片块
   - `_build_rich_text()` - 构建富文本格式
   - `_detect_style()` - 检测样式

3. **`migrations/versions/004_remove_content_field.py`** - 数据库迁移脚本
   - 删除 `content` 字段从 documents 表
   - 添加 `content_db_path` 字段（必填）
   - 删除 `format` 字段（不再需要区分格式）

#### 修改文件

4. **`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"
```

**迁移脚本内容**：
```python
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`

```python
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`

```python
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()

```python
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`

```python
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

```python
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

```python
@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 单元测试

1. **ContentDB 测试** (`tests/test_content_db.py`)
   - 创建数据库
   - 插入、查询、更新、删除 blocks
   - 搜索功能
   - 目录树构建

2. **Word Parser 测试** (`tests/test_word_parser.py`)
   - 解析标题块
   - 解析段落块（简单文本和富文本）
   - 解析表格块
   - 解析图片块
   - 父子关系构建

3. **Blocks 导出测试** (`tests/test_blocks_export.py`)
   - Blocks → Word 转换
   - 富文本格式保留
   - 表格合并单元格
   - 样式应用

---

### 4.2 集成测试

1. **完整流程测试**
   ```python
   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"]
   ```

2. **富文本测试**
   ```python
   async def test_rich_text():
       # 测试富文本块的创建和导出
       # 验证粗体、斜体、颜色等格式正确保留
   ```

3. **表格测试**
   ```python
   async def test_table_blocks():
       # 测试表格块的解析和导出
       # 验证合并单元格、表格样式正确保留
   ```

---

## 五、上线计划

### 5.1 发布策略

**阶段 1：开发与测试（2-3周）**
- 完成所有代码开发
- 单元测试和集成测试
- 内部测试环境验证

**阶段 2：预发布环境验证（1周）**
- 部署到预发布环境
- 团队全面测试
- 性能压测

**阶段 3：生产环境发布（1天）**
- 选择低峰时段发布
- 实时监控日志和性能
- 准备快速回滚方案

### 5.2 数据迁移方案

由于完全移除了 Markdown 存储，现有数据需要迁移：

**选项 A：一次性迁移（推荐）**
- 发布前：停服务，运行迁移脚本，将所有旧文档转为 SQLite 格式
- 优点：干净利落，无遗留数据
- 缺点：需要短暂停服

**选项 B：废弃旧数据**
- 发布后：旧文档标记为只读，不再支持编辑和导出
- 优点：无需迁移，快速上线
- 缺点：用户体验受影响

**推荐使用选项 A**，迁移脚本示例：
```python
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 回滚方案

如果发现严重问题，需要快速回滚：

1. **代码回滚**：Git revert 到上一个版本
2. **数据库回滚**：运行 alembic downgrade 恢复字段
3. **数据恢复**：从备份恢复 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+ 行）：
```python
# 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
```

**删除的常量**：
```python
STANDARD_STYLES  # 不再需要判断标准样式
```

### 9.2 app/services/export_service.py

**删除的函数和类**（约 300+ 行）：
```python
# 旧的 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 处理方法
```

**保留但需重构**：
```python
# 这些函数保留，但改为从 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

**删除的字段**：
```python
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

**删除的字段**：
```python
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 清理检查清单

完成代码删除后，运行以下检查：

1. **全局搜索残留引用**：
   ```bash
   # 搜索 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/
   ```

2. **检查导入语句**：
   ```bash
   grep -r "import mistune" --include="*.py" app/
   grep -r "from mistune" --include="*.py" app/
   ```

3. **运行测试**：
   ```bash
   pytest tests/ -v
   ```

4. **检查 Alembic 迁移**：
   ```bash
   alembic history
   alembic upgrade head
   ```

---

## 十、关键代码示例

### 9.1 稀疏排序初始化

```python
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（保持稀疏排序）

```python
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 存储的完整迁移路径。关键要点：

1. **彻底重构** - 完全移除 Markdown 存储，采用纯 SQLite 方案
2. **结构化存储** - 所有内容按 Block 类型分块存储，支持精确操作
3. **富文本支持** - 内容支持 JSON 格式，保留完整样式信息
4. **性能优化** - 稀疏排序、索引优化、批量操作
5. **数据迁移** - 提供完整的迁移方案，确保旧数据平滑过渡

### 核心优势

相比原有的 Markdown 存储：

| 优势 | 说明 |
|------|------|
| **查询性能** | 索引查询代替全文扫描，速度提升 10-100 倍 |
| **精确更新** | 直接更新单个 Block，无需解析整个文档 |
| **结构化查询** | 原生支持"获取所有表格"、"按标题筛选"等查询 |
| **目录构建** | 通过 parent_id 关系，毫秒级构建树形目录 |
| **富文本支持** | 完整保留格式信息，支持粗体、斜体、颜色等 |
| **灵活扩展** | 易于添加新的 Block 类型和属性 |

### 注意事项

1. **数据迁移风险** - 必须在发布前完成旧数据迁移，做好备份
2. **代码清理** - 删除所有 Markdown 相关代码，避免混淆
3. **充分测试** - Word 解析、富文本、表格等功能需要全面测试
4. **性能监控** - 上线后密切监控查询性能和存储空间
5. **文档更新** - 更新 API 文档和用户文档

**总工期**: 16-21 个工作日（含数据迁移）

**建议优先级**: 高（核心架构升级）

**风险等级**: 中高（涉及数据迁移和大量代码重构）

---

**文档版本**: v2.0  
**最后更新**: 2026-07-02  
**作者**: 项目团队  
**变更说明**: 移除所有 Markdown 兼容性内容，采用纯 SQLite 方案