Преглед изворни кода

fix: archive writes atomic, restore is delete-last

lingfengQAQ пре 1 недеља
родитељ
комит
01f9780b6f

+ 3 - 3
docs/superpowers/plans/2026-06-10-audit-fix-plan.md

@@ -105,9 +105,9 @@ def rollback(self, chapter: int) -> bool:
 - Modify: `webnovel-writer/scripts/archive_manager.py:125-128, 494-508`
 - Test: `webnovel-writer/scripts/data_modules/tests/test_archive_manager.py`
 
-- [ ] **Step 1: `save_archive` 改用 `atomic_write_json`**(归档是数据被移出 state 后的唯一副本)。
-- [ ] **Step 2: `restore_character` 顺序反转**:先恢复 SQLite,确认成功后才从归档 JSON 删除该角色;SQLite 失败时归档保持原样并返回错误。写测试:monkeypatch SQLite 恢复抛异常,断言归档文件未被修改。
-- [ ] **Step 3: 提交** `fix: archive writes atomic, restore is delete-last`。
+- [x] **Step 1: `save_archive` 改用 `atomic_write_json`**(归档是数据被移出 state 后的唯一副本)。
+- [x] **Step 2: `restore_character` 顺序反转**:先恢复 SQLite,确认成功后才从归档 JSON 删除该角色;SQLite 失败时归档保持原样并返回错误。写测试:monkeypatch SQLite 恢复抛异常,断言归档文件未被修改。
+- [x] **Step 3: 提交** `fix: archive writes atomic, restore is delete-last`。
 
 ---
 

+ 18 - 13
webnovel-writer/scripts/archive_manager.py

@@ -123,9 +123,8 @@ class ArchiveManager:
             return json.load(f)
 
     def save_archive(self, archive_file, data):
-        """保存归档文件"""
-        with open(archive_file, 'w', encoding='utf-8') as f:
-            json.dump(data, f, ensure_ascii=False, indent=2)
+        """保存归档文件(原子化写入)"""
+        atomic_write_json(archive_file, data, use_lock=True, backup=True)
 
     def check_trigger_conditions(self, state):
         """检查是否需要触发归档"""
@@ -489,23 +488,29 @@ class ArchiveManager:
 
         if not char_to_restore:
             print(f"❌ 归档中未找到角色: {name}")
-            return
-
-        # 移除 archived_at 字段
-        char_to_restore.pop("archived_at", None)
+            return False
 
-        # 原子性修复:先从归档中移除
-        archived = [char for char in archived if char["name"] != name]
-        self.save_archive(self.characters_archive, archived)
+        restored_character = dict(char_to_restore)
+        restored_character.pop("archived_at", None)
 
         # v5.1 引入: 恢复到 SQLite (通过 IndexManager)
-        char_id = char_to_restore.get("id", char_to_restore.get("name", "unknown"))
+        char_id = restored_character.get("id", restored_character.get("name", "unknown"))
         try:
             # 更新实体状态为 active
             self._index_manager.update_entity_field(char_id, "status", "active")
-            print(f"✅ 角色已恢复: {name}")
         except Exception as e:
-            print(f"⚠️ 实体状态恢复失败: {e}")
+            print(f"❌ 实体状态恢复失败,归档已保留: {e}")
+            return False
+
+        archived = [char for char in archived if char.get("name") != name]
+        try:
+            self.save_archive(self.characters_archive, archived)
+        except Exception as e:
+            print(f"❌ 归档更新失败,角色已恢复但归档未删除: {e}")
+            return False
+
+        print(f"✅ 角色已恢复: {name}")
+        return True
 
     def show_stats(self):
         """显示归档统计"""

+ 64 - 0
webnovel-writer/scripts/data_modules/tests/test_archive_manager.py

@@ -1,6 +1,7 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
 
+import json
 from pathlib import Path
 
 import pytest
@@ -72,3 +73,66 @@ def test_archive_identify_old_reviews_handles_mixed_formats(archive_env):
     assert len(results) == 3
     assert all(row["chapters_since_review"] >= 5 for row in results)
 
+
+def test_save_archive_uses_atomic_write_json(archive_env, monkeypatch):
+    module = _load_archive_module()
+    manager = module.ArchiveManager(project_root=archive_env)
+    calls = []
+
+    def fake_atomic_write_json(path, data, *, use_lock=True, backup=True, indent=2):
+        calls.append((path, data, use_lock, backup, indent))
+        Path(path).write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8")
+
+    monkeypatch.setattr(module, "atomic_write_json", fake_atomic_write_json)
+
+    manager.save_archive(manager.characters_archive, [{"name": "李雪"}])
+
+    assert calls == [(manager.characters_archive, [{"name": "李雪"}], True, True, 2)]
+
+
+def test_restore_character_keeps_archive_when_sqlite_restore_fails(archive_env, monkeypatch):
+    module = _load_archive_module()
+    manager = module.ArchiveManager(project_root=archive_env)
+    archived = [
+        {
+            "id": "li_xue",
+            "name": "李雪",
+            "tier": "支线",
+            "archived_at": "2026-06-10T00:00:00",
+        }
+    ]
+    manager.characters_archive.write_text(json.dumps(archived, ensure_ascii=False), encoding="utf-8")
+    before = manager.characters_archive.read_text(encoding="utf-8")
+
+    def fail_restore(*args, **kwargs):
+        raise RuntimeError("sqlite down")
+
+    monkeypatch.setattr(manager._index_manager, "update_entity_field", fail_restore)
+
+    assert manager.restore_character("李雪") is False
+    assert manager.characters_archive.read_text(encoding="utf-8") == before
+
+
+def test_restore_character_deletes_archive_after_sqlite_restore_succeeds(archive_env, monkeypatch):
+    module = _load_archive_module()
+    manager = module.ArchiveManager(project_root=archive_env)
+    archived = [
+        {
+            "id": "li_xue",
+            "name": "李雪",
+            "tier": "支线",
+            "archived_at": "2026-06-10T00:00:00",
+        }
+    ]
+    manager.characters_archive.write_text(json.dumps(archived, ensure_ascii=False), encoding="utf-8")
+    calls = []
+
+    def restore_status(entity_id, field, value):
+        calls.append((entity_id, field, value))
+
+    monkeypatch.setattr(manager._index_manager, "update_entity_field", restore_status)
+
+    assert manager.restore_character("李雪") is True
+    assert calls == [("li_xue", "status", "active")]
+    assert json.loads(manager.characters_archive.read_text(encoding="utf-8")) == []
+