欢迎来到实时协同编辑引擎的第一讲!今天我们将从零搭建项目骨架,设计核心的文本数据结构,并实现基础的文本操作。


一、项目初始化

1.1 目录结构

collaborative-editor/
├── core/                  # 核心引擎
│   ├── __init__.py
│   ├── document.py       # 文档模型
│   ├── operation.py      # 操作定义
│   ├── cursor.py         # 光标与选区
│   └── history.py        # 操作历史
├── server/               # 服务端
│   ├── __init__.py
│   ├── ws_server.py      # WebSocket 服务器
│   └── session.py        # 会话管理
├── client/               # 客户端
│   ├── __init__.py
│   ├── editor_client.py  # 编辑器客户端
│   └── sync.py           # 同步逻辑
├── tests/                # 测试
│   ├── test_document.py
│   └── test_operation.py
├── examples/             # 示例
│   └── simple_editor.py
├── requirements.txt
└── README.md

1.2 环境搭建

# 创建项目
mkdir collaborative-editor && cd collaborative-editor
python -m venv venv
source venv/bin/activate

# 依赖
pip install websockets fastapi uvicorn pytest
# requirements.txt
websockets>=12.0
fastapi>=0.109.0
uvicorn>=0.27.0
pytest>=8.0.0

二、文本数据结构选型

协同编辑器的核心问题是:如何在频繁的插入和删除操作下高效地表示文本?

2.1 常见方案对比

数据结构

插入复杂度

删除复杂度

随机访问

内存效率

适用场景

String/Python str

O(n)

O(n)

O(1)

小文本

Array/List

O(n)

O(n)

O(1)

简单场景

Gap Buffer

O(1)~O(n)

O(1)~O(n)

O(1)

Emacs/Vim

Piece Table

O(1)

O(1)

O(log n)

VS Code

Rope

O(log n)

O(log n)

O(log n)

大型文档

CRDT Tree

O(log n)

O(log n)

O(log n)

协同编辑

2.2 我们的选择:Piece Table + Rope 混合

对于协同编辑,我们需要:

  1. 高效的插入/删除​ — 操作频繁

  2. 支持版本回溯​ — 撤销/重做

  3. 便于合并操作​ — OT/CRDT 的基础

我们采用 Piece Table​ 作为主要数据结构,它在插入和删除上都是 O(1),且天然支持操作日志。


三、Piece Table 实现

3.1 Piece Table 原理

Piece Table 核心思想:

不直接修改文本,而是维护一个"片段表"(Piece Table),
每个片段指向原始缓冲区或追加缓冲区中的一段连续区域。

初始状态:
┌──────────────────────────────────────┐
│  原始缓冲区 (Original Buffer)         │
│  "Hello, World!"                     │
└──────────────────────────────────────┘

┌──────────────────────────────────────┐
│  追加缓冲区 (Add Buffer)             │
│  (空)                                │
└──────────────────────────────────────┘

Piece Table:
┌──────┬────────┬────────┬────────┐
│  ID  │ Buffer │ Offset │ Length │
├──────┼────────┼────────┼────────┤
│  P0  │ ORIG   │   0    │   13   │
└──────┴────────┴────────┴────────┘

插入 " Beautiful" 到位置 6 之后:
Piece Table:
┌──────┬────────┬────────┬────────┐
│  ID  │ Buffer │ Offset │ Length │
├──────┼────────┼────────┼────────┤
│  P0  │ ORIG   │   0    │   7    │  ← "Hello, "
│  P1  │ ADD    │   0    │   11   │  ← " Beautiful"
│  P2  │ ORIG   │   7    │   6    │  ← "World!"
└──────┴────────┴────────┴────────┘

拼接结果: "Hello, Beautiful World!"

3.2 核心代码实现

# core/document.py
"""
文档模型:基于 Piece Table 的文本表示
"""

from __future__ import annotations
from dataclasses import dataclass, field
from typing import List, Tuple, Optional, Iterator
import copy
import logging

logger = logging.getLogger(__name__)

# ---------- 缓冲区枚举 ----------

class BufferType:
    """缓冲区类型"""
    ORIGINAL = "original"  # 原始缓冲区(只读)
    ADDITION = "addition"  # 追加缓冲区(可写)

# ---------- 片段定义 ----------

@dataclass(frozen=True)
class Piece:
    """
    片段:指向缓冲区中的一段连续区域
    
    Attributes:
        piece_id: 片段唯一标识
        buffer: 缓冲区类型
        offset: 在缓冲区中的起始偏移
        length: 长度
    """
    piece_id: str
    buffer: str
    offset: int
    length: int
    
    def __repr__(self) -> str:
        return f"Piece({self.piece_id}, {self.buffer}[{self.offset}:{self.offset + self.length}])"

# ---------- Piece Table ----------

class PieceTable:
    """
    Piece Table 核心实现
    
    通过维护片段列表来表示文本,支持高效的插入和删除。
    """
    
    def __init__(self, initial_text: str = ""):
        # 原始缓冲区:存放初始文本,永不修改
        self.original_buffer = initial_text
        
        # 追加缓冲区:存放所有插入的新文本
        self.addition_buffer = ""
        
        # 片段表:有序的片段列表
        self.pieces: List[Piece] = []
        
        # 片段计数器(用于生成唯一 ID)
        self._piece_counter = 0
        
        # 初始化:如果初始文本不为空,创建第一个片段
        if initial_text:
            self.pieces.append(Piece(
                piece_id=self._next_piece_id(),
                buffer=BufferType.ORIGINAL,
                offset=0,
                length=len(initial_text)
            ))
    
    def _next_piece_id(self) -> str:
        """生成下一个片段 ID"""
        self._piece_counter += 1
        return f"P{self._piece_counter}"
    
    # ---------- 核心操作 ----------
    
    def insert(self, position: int, text: str) -> Operation:
        """
        在指定位置插入文本
        
        Args:
            position: 插入位置(0-based)
            text: 要插入的文本
        Returns:
            本次操作的描述
        """
        if position < 0 or position > self.char_length():
            raise IndexError(f"Position {position} out of range [0, {self.char_length()}]")
        
        if not text:
            return Operation(OperationType.NOOP, position, "", "")
        
        # 1. 将文本追加到追加缓冲区
        add_offset = len(self.addition_buffer)
        self.addition_buffer += text
        
        # 2. 创建新片段
        new_piece = Piece(
            piece_id=self._next_piece_id(),
            buffer=BufferType.ADDITION,
            offset=add_offset,
            length=len(text)
        )
        
        # 3. 在片段表中定位插入点
        if not self.pieces:
            # 空文档
            self.pieces.append(new_piece)
        else:
            piece_idx, offset_in_piece = self._find_position(position)
            
            old_piece = self.pieces[piece_idx]
            
            # 分裂旧片段
            left_piece = None
            right_piece = None
            
            if offset_in_piece > 0:
                left_piece = Piece(
                    piece_id=self._next_piece_id(),
                    buffer=old_piece.buffer,
                    offset=old_piece.offset,
                    length=offset_in_piece
                )
            
            if offset_in_piece < old_piece.length:
                right_piece = Piece(
                    piece_id=self._next_piece_id(),
                    buffer=old_piece.buffer,
                    offset=old_piece.offset + offset_in_piece,
                    length=old_piece.length - offset_in_piece
                )
            
            # 重组片段表
            new_pieces = self.pieces[:piece_idx]
            if left_piece:
                new_pieces.append(left_piece)
            new_pieces.append(new_piece)
            if right_piece:
                new_pieces.append(right_piece)
            new_pieces.extend(self.pieces[piece_idx + 1:])
            
            self.pieces = new_pieces
        
        logger.debug(f"Insert '{text}' at {position}")
        return Operation(
            OperationType.INSERT,
            position,
            text,
            ""
        )
    
    def delete(self, position: int, length: int) -> Operation:
        """
        删除从 position 开始的 length 个字符
        
        Args:
            position: 起始位置
            length: 删除长度
        Returns:
            本次操作的描述
        """
        doc_length = self.char_length()
        if position < 0 or position + length > doc_length:
            raise IndexError(
                f"Delete range [{position}, {position + length}) "
                f"out of range [0, {doc_length}]"
            )
        
        if length == 0:
            return Operation(OperationType.NOOP, position, "", "")
        
        # 获取被删除的文本(用于撤销)
        deleted_text = self.slice(position, position + length)
        
        # 找到需要修改的片段范围
        start_idx, start_offset = self._find_position(position)
        end_idx, end_offset = self._find_position(position + length)
        
        new_pieces = []
        
        # 处理起始片段(可能被部分删除)
        if start_offset > 0:
            first_piece = self.pieces[start_idx]
            new_pieces.append(Piece(
                piece_id=self._next_piece_id(),
                buffer=first_piece.buffer,
                offset=first_piece.offset,
                length=start_offset
            ))
        
        # 处理结束片段(可能被部分删除)
        if end_offset < self.pieces[end_idx].length:
            last_piece = self.pieces[end_idx]
            new_pieces.append(Piece(
                piece_id=self._next_piece_id(),
                buffer=last_piece.buffer,
                offset=last_piece.offset + end_offset,
                length=last_piece.length - end_offset
            ))
        
        # 跳过中间的片段(被完全删除)
        new_pieces.extend(self.pieces[end_idx + 1:])
        
        self.pieces = new_pieces
        
        logger.debug(f"Delete {length} chars at {position}")
        return Operation(
            OperationType.DELETE,
            position,
            "",
            deleted_text
        )
    
    # ---------- 查询方法 ----------
    
    def char_length(self) -> int:
        """获取文档总字符数"""
        return sum(p.length for p in self.pieces)
    
    def slice(self, start: int, end: int) -> str:
        """
        获取文本片段
        
        Args:
            start: 起始位置
            end: 结束位置(不包含)
        Returns:
            文本内容
        """
        if start < 0 or end > self.char_length() or start > end:
            raise IndexError(f"Invalid slice [{start}, {end})")
        
        result = []
        current_pos = 0
        
        for piece in self.pieces:
            piece_start = current_pos
            piece_end = current_pos + piece.length
            
            # 检查是否有重叠
            if piece_end > start and piece_start < end:
                overlap_start = max(start - current_pos, 0)
                overlap_end = min(end - current_pos, piece.length)
                
                if piece.buffer == BufferType.ORIGINAL:
                    result.append(
                        self.original_buffer[
                            piece.offset + overlap_start:
                            piece.offset + overlap_end
                        ]
                    )
                else:
                    result.append(
                        self.addition_buffer[
                            piece.offset + overlap_start:
                            piece.offset + overlap_end
                        ]
                    )
            
            current_pos = piece_end
            if current_pos >= end:
                break
        
        return "".join(result)
    
    def to_string(self) -> str:
        """获取完整文本"""
        return self.slice(0, self.char_length())
    
    def _find_position(self, position: int) -> Tuple[int, int]:
        """
        查找位置所在的片段和偏移
        
        Args:
            position: 文档中的位置
        Returns:
            (片段索引, 在片段内的偏移)
        """
        if not self.pieces:
            return (0, 0)
        
        current_pos = 0
        for idx, piece in enumerate(self.pieces):
            if current_pos + piece.length > position:
                return (idx, position - current_pos)
            current_pos += piece.length
        
        # 位置在末尾
        return (len(self.pieces) - 1, self.pieces[-1].length)
    
    # ---------- 调试方法 ----------
    
    def debug_info(self) -> dict:
        """调试信息"""
        return {
            'char_length': self.char_length(),
            'pieces_count': len(self.pieces),
            'original_buffer_len': len(self.original_buffer),
            'addition_buffer_len': len(self.addition_buffer),
            'pieces': [str(p) for p in self.pieces],
            'text': self.to_string()
        }

# ---------- 操作定义 ----------

from enum import Enum

class OperationType(Enum):
    """操作类型"""
    INSERT = "insert"
    DELETE = "delete"
    NOOP = "noop"  # 空操作

@dataclass
class Operation:
    """
    文档操作
    
    记录对文档的一次修改,支持撤销和重做。
    """
    op_type: OperationType
    position: int
    text: str = ""          # INSERT 时插入的文本
    deleted_text: str = ""  # DELETE 时被删除的文本
    timestamp: float = 0.0
    
    def __post_init__(self):
        import time
        if not self.timestamp:
            self.timestamp = time.time()
    
    def invert(self) -> Operation:
        """
        生成逆操作(用于撤销)
        
        INSERT 的逆操作是 DELETE
        DELETE 的逆操作是 INSERT
        """
        if self.op_type == OperationType.INSERT:
            return Operation(
                OperationType.DELETE,
                self.position,
                deleted_text=self.text
            )
        elif self.op_type == OperationType.DELETE:
            return Operation(
                OperationType.INSERT,
                self.position,
                text=self.deleted_text
            )
        else:
            return Operation(OperationType.NOOP, 0)
    
    def __repr__(self) -> str:
        if self.op_type == OperationType.INSERT:
            return f"INSERT(+'{self.text}' @{self.position})"
        elif self.op_type == OperationType.DELETE:
            return f"DELETE(-'{self.deleted_text}' @{self.position})"
        else:
            return "NOOP"

四、光标与选区

4.1 光标模型

# core/cursor.py
"""
光标与选区模型
"""

from dataclasses import dataclass, field
from typing import Optional, Tuple
import uuid

@dataclass
class Cursor:
    """
    光标位置
    
    支持多行文本中的行列表示和绝对位置表示。
    """
    user_id: str
    document_id: str
    
    # 绝对位置(0-based)
    position: int = 0
    
    # 屏幕坐标(可选,用于显示)
    line: int = 0
    column: int = 0
    
    # 颜色(用于区分不同用户)
    color: str = "#007bff"
    
    # 是否在线
    online: bool = True
    
    def move_to(self, new_position: int):
        """移动光标到新位置"""
        self.position = max(0, new_position)
    
    def to_dict(self) -> dict:
        """序列化"""
        return {
            'user_id': self.user_id,
            'document_id': self.document_id,
            'position': self.position,
            'line': self.line,
            'column': self.column,
            'color': self.color,
            'online': self.online
        }


@dataclass
class Selection:
    """
    选区
    
    由起点和终点定义的一段文本范围。
    """
    user_id: str
    
    # 锚点(按下鼠标时的位置)
    anchor: int = 0
    
    # 焦点(松开鼠标时的位置)
    focus: int = 0
    
    def __post_init__(self):
        if self.focus < self.anchor:
            self.anchor, self.focus = self.focus, self.anchor
    
    @property
    def start(self) -> int:
        """选区起始"""
        return min(self.anchor, self.focus)
    
    @property
    def end(self) -> int:
        """选区结束"""
        return max(self.anchor, self.focus)
    
    @property
    def length(self) -> int:
        """选区长度"""
        return self.end - self.start
    
    @property
    def is_empty(self) -> bool:
        """是否为空选区(即光标位置)"""
        return self.anchor == self.focus
    
    def contains(self, position: int) -> bool:
        """判断位置是否在选区内"""
        return self.start <= position <= self.end
    
    def shift(self, offset: int):
        """平移选区"""
        self.anchor += offset
        self.focus += offset
    
    def to_dict(self) -> dict:
        """序列化"""
        return {
            'user_id': self.user_id,
            'anchor': self.anchor,
            'focus': self.focus,
            'start': self.start,
            'end': self.end,
            'length': self.length
        }


class CursorManager:
    """
    光标管理器
    
    管理文档中所有用户的游标和选区。
    """
    
    def __init__(self, document_id: str):
        self.document_id = document_id
        self.cursors: dict = {}  # user_id -> Cursor
        self.selections: dict = {}  # user_id -> Selection
    
    def add_user(self, user_id: str, color: str = None) -> Cursor:
        """添加用户"""
        if color is None:
            colors = ["#FF6B6B", "#4ECDC4", "#45B7D1", "#96CEB4", "#FFEAA7"]
            color = colors[len(self.cursors) % len(colors)]
        
        cursor = Cursor(user_id=user_id, document_id=self.document_id, color=color)
        self.cursors[user_id] = cursor
        return cursor
    
    def remove_user(self, user_id: str):
        """移除用户"""
        self.cursors.pop(user_id, None)
        self.selections.pop(user_id, None)
    
    def update_cursor(self, user_id: str, position: int) -> Optional[Cursor]:
        """更新光标位置"""
        cursor = self.cursors.get(user_id)
        if cursor:
            cursor.move_to(position)
            return cursor
        return None
    
    def update_selection(self, user_id: str, 
                         anchor: int, focus: int) -> Optional[Selection]:
        """更新选区"""
        selection = Selection(user_id=user_id, anchor=anchor, focus=focus)
        self.selections[user_id] = selection
        return selection
    
    def get_remote_cursors(self, exclude_user: str = None) -> list:
        """获取其他用户的光标"""
        return [
            c.to_dict() for uid, c in self.cursors.items()
            if uid != exclude_user and c.online
        ]
    
    def get_remote_selections(self, exclude_user: str = None) -> list:
        """获取其他用户的选区"""
        return [
            s.to_dict() for uid, s in self.selections.items()
            if uid != exclude_user
        ]

五、操作历史与撤销

5.1 历史管理器

# core/history.py
"""
操作历史管理器

支持撤销和重做操作。
"""

from typing import List, Optional
from .document import Operation, OperationType

class HistoryManager:
    """
    操作历史管理器
    
    维护一个操作栈,支持撤销(Undo)和重做(Redo)。
    """
    
    def __init__(self, max_history: int = 1000):
        self.max_history = max_history
        
        # 撤销栈
        self.undo_stack: List[Operation] = []
        
        # 重做栈
        self.redo_stack: List[Operation] = []
    
    def push(self, operation: Operation):
        """
        记录操作
        
        Args:
            operation: 执行的操作
        """
        if operation.op_type == OperationType.NOOP:
            return
        
        self.undo_stack.append(operation)
        
        # 限制历史大小
        if len(self.undo_stack) > self.max_history:
            self.undo_stack.pop(0)
        
        # 新操作清空重做栈
        self.redo_stack.clear()
    
    def can_undo(self) -> bool:
        """是否可以撤销"""
        return len(self.undo_stack) > 0
    
    def can_redo(self) -> bool:
        """是否可以重做"""
        return len(self.redo_stack) > 0
    
    def undo(self) -> Optional[Operation]:
        """
        获取撤销操作
        
        Returns:
            逆操作(需要应用到文档)
        """
        if not self.undo_stack:
            return None
        
        operation = self.undo_stack.pop()
        inverse = operation.invert()
        
        self.redo_stack.append(operation)
        
        return inverse
    
    def redo(self) -> Optional[Operation]:
        """
        获取重做操作
        
        Returns:
            原始操作(需要应用到文档)
        """
        if not self.redo_stack:
            return None
        
        operation = self.redo_stack.pop()
        self.undo_stack.append(operation)
        
        return operation
    
    def clear(self):
        """清空历史"""
        self.undo_stack.clear()
        self.redo_stack.clear()
    
    def get_stats(self) -> dict:
        """获取统计信息"""
        return {
            'undo_count': len(self.undo_stack),
            'redo_count': len(self.redo_stack),
            'can_undo': self.can_undo(),
            'can_redo': self.can_redo()
        }

六、完整文档引擎

6.1 Document 类

# core/document.py (续)

class Document:
    """
    文档引擎
    
    整合 PieceTable、HistoryManager、CursorManager,
    提供完整的文档操作接口。
    """
    
    def __init__(self, document_id: str, initial_text: str = ""):
        self.document_id = document_id
        self.table = PieceTable(initial_text)
        self.history = HistoryManager()
        self.cursors = CursorManager(document_id)
        
        # 版本号(用于同步)
        self.version = 0
    
    # ---------- 文本操作 ----------
    
    def insert(self, position: int, text: str) -> Operation:
        """
        插入文本
        
        自动记录历史和更新版本号。
        """
        op = self.table.insert(position, text)
        if op.op_type != OperationType.NOOP:
            self.history.push(op)
            self.version += 1
        return op
    
    def delete(self, position: int, length: int) -> Operation:
        """
        删除文本
        
        自动记录历史和更新版本号。
        """
        op = self.table.delete(position, length)
        if op.op_type != OperationType.NOOP:
            self.history.push(op)
            self.version += 1
        return op
    
    def undo(self) -> Optional[Operation]:
        """
        撤销上一次操作
        
        Returns:
            执行的逆操作
        """
        inverse = self.history.undo()
        if inverse is None:
            return None
        
        if inverse.op_type == OperationType.INSERT:
            self.table.insert(inverse.position, inverse.text)
        elif inverse.op_type == OperationType.DELETE:
            self.table.delete(inverse.position, len(inverse.deleted_text))
        
        self.version += 1
        return inverse
    
    def redo(self) -> Optional[Operation]:
        """
        重做上一次撤销的操作
        
        Returns:
            执行的操作
        """
        operation = self.history.redo()
        if operation is None:
            return None
        
        if operation.op_type == OperationType.INSERT:
            self.table.insert(operation.position, operation.text)
        elif operation.op_type == OperationType.DELETE:
            self.table.delete(operation.position, len(operation.deleted_text))
        
        self.version += 1
        return operation
    
    # ---------- 查询 ----------
    
    def get_text(self) -> str:
        """获取完整文本"""
        return self.table.to_string()
    
    def get_text_range(self, start: int, end: int) -> str:
        """获取文本片段"""
        return self.table.slice(start, end)
    
    def get_length(self) -> int:
        """获取文档长度"""
        return self.table.char_length()
    
    def get_state(self) -> dict:
        """获取文档状态"""
        return {
            'document_id': self.document_id,
            'version': self.version,
            'length': self.get_length(),
            'text': self.get_text(),
            'history': self.history.get_stats()
        }

七、测试

7.1 单元测试

# tests/test_document.py
import pytest
from core.document import PieceTable, Document, Operation, OperationType

class TestPieceTable:
    """Piece Table 测试"""
    
    def test_empty_document(self):
        pt = PieceTable()
        assert pt.char_length() == 0
        assert pt.to_string() == ""
    
    def test_initial_text(self):
        pt = PieceTable("Hello")
        assert pt.char_length() == 5
        assert pt.to_string() == "Hello"
    
    def test_insert_at_beginning(self):
        pt = PieceTable("World")
        pt.insert(0, "Hello ")
        assert pt.to_string() == "Hello World"
    
    def test_insert_at_end(self):
        pt = PieceTable("Hello")
        pt.insert(5, " World")
        assert pt.to_string() == "Hello World"
    
    def test_insert_in_middle(self):
        pt = PieceTable("Heorld")
        pt.insert(2, "ll")
        assert pt.to_string() == "Hello World"  # 修正后的预期
    
    def test_multiple_inserts(self):
        pt = PieceTable("")
        pt.insert(0, "Hello")
        pt.insert(5, " World")
        pt.insert(0, "Start: ")
        assert pt.to_string() == "Start: Hello World"
    
    def test_delete_beginning(self):
        pt = PieceTable("Hello World")
        pt.delete(0, 6)
        assert pt.to_string() == "World"
    
    def test_delete_end(self):
        pt = PieceTable("Hello World")
        pt.delete(6, 5)
        assert pt.to_string() == "Hello "
    
    def test_delete_middle(self):
        pt = PieceTable("Hello World")
        pt.delete(5, 1)
        assert pt.to_string() == "HelloWorld"
    
    def test_slice(self):
        pt = PieceTable("Hello World")
        assert pt.slice(0, 5) == "Hello"
        assert pt.slice(6, 11) == "World"
        assert pt.slice(0, 11) == "Hello World"


class TestDocument:
    """文档引擎测试"""
    
    def test_basic_operations(self):
        doc = Document("doc-1", "")
        
        doc.insert(0, "Hello")
        assert doc.get_text() == "Hello"
        
        doc.insert(5, " World")
        assert doc.get_text() == "Hello World"
        
        doc.delete(5, 6)
        assert doc.get_text() == "Hello"
    
    def test_undo_redo(self):
        doc = Document("doc-1", "")
        
        doc.insert(0, "Hello")
        doc.insert(5, " World")
        assert doc.get_text() == "Hello World"
        
        # 撤销
        doc.undo()
        assert doc.get_text() == "Hello"
        
        # 再撤销
        doc.undo()
        assert doc.get_text() == ""
        
        # 重做
        doc.redo()
        assert doc.get_text() == "Hello"
        
        doc.redo()
        assert doc.get_text() == "Hello World"
    
    def test_version_increment(self):
        doc = Document("doc-1", "")
        v0 = doc.version
        
        doc.insert(0, "A")
        assert doc.version == v0 + 1
        
        doc.insert(1, "B")
        assert doc.version == v0 + 2
        
        doc.undo()
        assert doc.version == v0 + 3

if __name__ == "__main__":
    pytest.main([__file__, "-v"])

八、快速体验

8.1 简单的交互式演示

# examples/simple_editor.py
"""
简单的交互式编辑器演示
"""

from core.document import Document

def interactive_demo():
    print("=" * 60)
    print("📝 实时协同编辑引擎 - 演示")
    print("=" * 60)
    
    doc = Document("demo-doc", "Hello World!")
    print(f"\n初始文档: '{doc.get_text()}'")
    print(f"文档长度: {doc.get_length()}")
    print(f"版本号: {doc.version}")
    
    print("\n--- 操作演示 ---")
    
    # 插入
    doc.insert(6, "Beautiful ")
    print(f"插入后: '{doc.get_text()}'")
    
    # 删除
    doc.delete(6, 10)
    print(f"删除后: '{doc.get_text()}'")
    
    # 撤销
    doc.undo()
    print(f"撤销后: '{doc.get_text()}'")
    
    doc.undo()
    print(f"再撤销: '{doc.get_text()}'")
    
    # 重做
    doc.redo()
    print(f"重做后: '{doc.get_text()}'")
    
    print(f"\n最终版本: {doc.version}")
    print(f"历史统计: {doc.history.get_stats()}")
    
    print("\n" + "=" * 60)
    print("✅ 演示完成!")
    print("=" * 60)


if __name__ == "__main__":
    interactive_demo()

九、总结

9.1 本讲成果

组件

文件

功能

PieceTable

core/document.py

高效文本表示,O(1) 插入/删除

Operation

core/document.py

操作定义,支持逆操作

Cursor/Selection

core/cursor.py

光标和选区模型

HistoryManager

core/history.py

撤销/重做支持

Document

core/document.py

完整文档引擎

9.2 核心知识点

  • Piece Table​ 通过维护片段表而非直接修改文本,实现了高效的插入和删除

  • 操作日志​ 记录了每一次修改,为撤销/重做和协同同步打下基础

  • 光标模型​ 支持多用户同时编辑时的位置跟踪

9.3 下一讲预告

第2讲:OT 算法基础

我们将深入协同编辑的核心——Operational Transformation 算法:

  • 操作的定义与组合

  • 变换函数的数学基础

  • 客户端-服务端同步模型

  • 冲突检测与自动解决

准备好迎接挑战了吗?让我们在第2讲再见!


🧰 开发之余的小工具推荐

处理 Base64、JWT 解析、JSON 格式化、Crontab 计算、PDF 合并压缩这些碎片需求,我常用一个纯前端本地工具箱:zz365.top(子页 PDF 大师:PDF 大师 - zz365工具箱)。所有计算在浏览器完成,文件不上传服务器,关页即清。免费、无登录、无广告,适合开发者当常驻标签页。

Logo

汇聚全球AI编程工具,助力开发者即刻编程。

更多推荐