Coverage for merco/sandbox/snapshot.py: 29%

1"""快照系统 — 文件修改历史跟踪 

2 

3每次编辑前保存文件副本，支持： 

4- track(path, content) — 记录修改前内容 

5- history(session_id) — 查询某轮会话的所有改动 

6- revert(session_id) — 撤销整轮修改 

7""" 

8 

9import json 

10import logging 

11import shutil 

12from datetime import datetime, timezone 

13from pathlib import Path 

14 

15logger = logging.getLogger("merco.sandbox.snapshot") 

16 

17# 快照存储根目录 

18SNAPSHOT_DIR = Path.home() / ".merco" / "snapshots" 

19 

20 

21def _ensure_dir() -> Path: 

22 SNAPSHOT_DIR.mkdir(parents=True, exist_ok=True) 

23 return SNAPSHOT_DIR 

24 

25 

26def _session_path(session_id: str) -> Path: 

27 return _ensure_dir() / f"{session_id}.json" 

28 

29 

30# ── 当前会话 ID（由 Agent 在启动时设置）── 

31_current_session_id: str | None = None 

32 

33 

34def set_current_session(session_id: str) -> None: 

35 """设置当前会话 ID（由 Agent 在初始化时调用）""" 

36 global _current_session_id 

37 _current_session_id = session_id 

38 

39 

40def get_current_session() -> str | None: 

41 return _current_session_id 

42 

43 

44# ── 公开 API ──────────────────────────────────────────────────────── 

45 

46 

47def track(path: str, content: str, session_id: str | None = None) -> dict: 

48 """记录一次编辑前的快照 

49 

50 Args: 

51 path: 被修改的文件路径 

52 content: 修改前的文件内容 

53 session_id: 会话标识，默认用当前时间戳 

54 

55 Returns: 

56 {session_id, snapshot_id, path, timestamp} 

57 """ 

58 if session_id is None: 

59 session_id = datetime.now(tz=timezone.utc).strftime("%Y%m%d_%H%M%S_%f") 

60 

61 snapshots = _load(session_id) 

62 entry = { 

63 "path": str(Path(path).resolve()), 

64 "content": content, 

65 "timestamp": datetime.now(tz=timezone.utc).isoformat(), 

66 } 

67 snapshots.append(entry) 

68 _save(session_id, snapshots) 

69 

70 return { 

71 "session_id": session_id, 

72 "snapshot_id": len(snapshots) - 1, 

73 "path": entry["path"], 

74 "timestamp": entry["timestamp"], 

75 } 

76 

77 

78def history(session_id: str) -> list[dict]: 

79 """查询某次会话的所有快照记录 

80 

81 Returns: 

82 [{path, timestamp, content}] 

83 """ 

84 return _load(session_id) 

85 

86 

87def list_sessions() -> list[dict]: 

88 """列出所有有快照的会话 

89 

90 Returns: 

91 [{session_id, file_count, timestamp}] 

92 """ 

93 sessions = [] 

94 for f in sorted(_ensure_dir().glob("*.json"), reverse=True): 

95 try: 

96 data = json.loads(f.read_text()) 

97 sessions.append({ 

98 "session_id": f.stem, 

99 "file_count": len(data), 

100 "timestamp": data[0]["timestamp"] if data else "", 

101 }) 

102 except Exception: 

103 continue 

104 return sessions 

105 

106 

107def revert(session_id: str, snapshot_index: int | None = None) -> list[dict]: 

108 """撤销快照中的修改 

109 

110 Args: 

111 session_id: 会话标识 

112 snapshot_index: 要撤销的快照索引（None=撤销全部） 

113 

114 Returns: 

115 [{"path": str, "reverted": bool, "error": str|None}] 

116 """ 

117 snapshots = _load(session_id) 

118 if not snapshots: 

119 return [] 

120 

121 if snapshot_index is not None: 

122 snapshots = [snapshots[snapshot_index]] 

123 

124 results = [] 

125 for entry in snapshots: 

126 file_path = Path(entry["path"]) 

127 try: 

128 file_path.parent.mkdir(parents=True, exist_ok=True) 

129 file_path.write_text(entry["content"], encoding="utf-8") 

130 results.append({ 

131 "path": entry["path"], 

132 "reverted": True, 

133 "error": None, 

134 }) 

135 logger.info("已恢复 %s", entry["path"]) 

136 except Exception as e: 

137 results.append({ 

138 "path": entry["path"], 

139 "reverted": False, 

140 "error": str(e), 

141 }) 

142 

143 if snapshot_index is None: 

144 # 全部撤销后删除会话文件 

145 _session_path(session_id).unlink(missing_ok=True) 

146 

147 return results 

148 

149 

150# ── 内部 ──────────────────────────────────────────────────────────── 

151 

152 

153def _load(session_id: str) -> list[dict]: 

154 sp = _session_path(session_id) 

155 if sp.exists(): 

156 try: 

157 return json.loads(sp.read_text()) 

158 except (json.JSONDecodeError, OSError): 

159 return [] 

160 return [] 

161 

162 

163def _save(session_id: str, data: list[dict]) -> None: 

164 sp = _session_path(session_id) 

165 sp.write_text(json.dumps(data, ensure_ascii=False, indent=2))