fix: 使用原子写入防止重启时配置丢失

app/common/history/file_utils.py

@@ -9,7 +9,7 @@
 
 from loguru import logger
 
-from app.tools.path_utils import get_path
+from app.tools.path_utils import get_path, atomic_write_json
 
 
 _history_cache_lock = threading.RLock()
@@ -149,8 +149,7 @@ def save_history_data(history_type: str, file_name: str, data: Dict[str, Any]) -
     """
     try:
         file_path = get_history_file_path(history_type, file_name, strict=True)
-        with open(file_path, "w", encoding="utf-8") as f:
-            json.dump(data, f, ensure_ascii=False, indent=4)
+        atomic_write_json(file_path, data)
         with _history_cache_lock:
             _history_data_cache[file_path] = (
                 _get_file_signature(file_path),

app/common/safety/secure_store.py

@@ -6,7 +6,7 @@
 import ctypes
 import uuid
 from loguru import logger
-from app.tools.path_utils import get_settings_path, ensure_dir
+from app.tools.path_utils import get_settings_path, ensure_dir, atomic_write_bytes
 
 try:
     from Cryptodome.Cipher import AES
@@ -143,37 +143,15 @@ def write_secrets(d: dict) -> None:
         comp = zlib.compress(raw, level=6)
         key = _platform_key()
         payload = _encrypt_payload(comp, key)
-        with open(p, "wb") as f:
-            f.write(b"SRV1" + payload)
+        atomic_write_bytes(p, b"SRV1" + payload)
         _set_hidden(str(p))
         logger.debug(f"写入安全配置成功：{p}")
     except PermissionError as e:
-        logger.warning(
-            f"写入安全配置失败：权限被拒绝，文件可能被占用或无写权限：{p}, 错误：{e}"
+        logger.error(
+            f"写入安全配置失败：文件被占用或无写权限，请关闭占用程序或检查权限：{p}, 错误：{e}"
         )
-        # 尝试使用临时文件写入然后替换
-        try:
-            import tempfile
-
-            with tempfile.NamedTemporaryFile(
-                mode="wb", delete=False, dir=os.path.dirname(p)
-            ) as tmp_file:
-                tmp_file.write(b"SRV1" + payload)
-                tmp_path = tmp_file.name
-
-            # 替换原文件
-            os.replace(tmp_path, p)
-            _set_hidden(str(p))
-            logger.debug(f"使用临时文件写入安全配置成功：{p}")
-        except Exception as temp_e:
-            logger.warning(f"使用临时文件写入安全配置也失败：{temp_e}")
-            # 降级到明文JSON写入
-            try:
-                with open(p, "w", encoding="utf-8") as f:
-                    json.dump(d, f, ensure_ascii=False, indent=4)
-                logger.warning(f"写入安全配置降级为明文JSON：{p}")
-            except Exception as e2:
-                logger.warning(f"降级写入明文JSON也失败：{e2}")
+    except Exception as e:
+        logger.error(f"写入安全配置失败：{p}, 错误：{e}")
 
 
 def read_behind_scenes_settings() -> dict:
@@ -230,39 +208,12 @@ def write_behind_scenes_settings(d: dict) -> None:
         comp = zlib.compress(raw, level=6)
         key = _platform_key()
         payload = _encrypt_payload(comp, key)
-        with open(p, "wb") as f:
-            f.write(b"SRV1" + payload)
+        atomic_write_bytes(p, b"SRV1" + payload)
         _set_hidden(str(p))
         logger.debug(f"写入内幕设置成功：{p}")
     except PermissionError as e:
         logger.error(
-            f"写入内幕设置失败：权限被拒绝，文件可能被占用或无写权限：{p}, 错误：{e}"
+            f"写入内幕设置失败：文件被占用或无写权限，请关闭占用程序或检查权限：{p}, 错误：{e}"
         )
-        try:
-            import tempfile
-
-            with tempfile.NamedTemporaryFile(
-                mode="wb", delete=False, dir=os.path.dirname(p)
-            ) as tmp_file:
-                tmp_file.write(b"SRV1" + payload)
-                tmp_path = tmp_file.name
-
-            os.replace(tmp_path, p)
-            _set_hidden(str(p))
-            logger.debug(f"使用临时文件写入内幕设置成功：{p}")
-        except Exception as temp_e:
-            logger.error(f"使用临时文件写入内幕设置也失败：{temp_e}")
-            try:
-                with open(p, "w", encoding="utf-8") as f:
-                    json.dump(d, f, ensure_ascii=False, indent=4)
-                logger.warning(f"写入内幕设置降级为明文JSON：{p}")
-            except Exception as e2:
-                logger.error(f"降级写入明文JSON也失败：{e2}")
     except Exception as e:
         logger.error(f"写入内幕设置失败：{p}, 错误：{e}")
-        try:
-            with open(p, "w", encoding="utf-8") as f:
-                json.dump(d, f, ensure_ascii=False, indent=4)
-            logger.warning(f"写入内幕设置降级为明文JSON：{p}")
-        except Exception as e2:
-            logger.exception(f"降级写入明文JSON也失败：{e2}")

app/tools/config.py

@@ -29,6 +29,7 @@
     get_data_path,
     get_settings_path,
     get_path,
+    atomic_write_json,
 )
 from app.tools.personalised import get_theme_icon
 from app.tools.settings_access import readme_settings_async
@@ -956,8 +957,7 @@ def import_settings(parent: Optional[QWidget] = None) -> None:
 
             if dialog.exec():
                 settings_path = get_settings_path()
-                with open(settings_path, "w", encoding="utf-8") as f:
-                    json.dump(imported_settings, f, ensure_ascii=False, indent=4)
+                atomic_write_json(settings_path, imported_settings)
 
                 success_dialog = MessageBox(
                     get_any_position_value_async(
@@ -2093,8 +2093,7 @@ def _save_drawn_records(file_path: str, drawn_records: dict) -> None:
         drawn_records: 已抽取的学生记录字典
     """
     try:
-        with open(file_path, "w", encoding="utf-8") as file:
-            json.dump(drawn_records, file, ensure_ascii=False, indent=2)
+        atomic_write_json(file_path, drawn_records, indent=2)
     except IOError as e:
         logger.exception(f"保存已抽取记录失败: {e}")
 

app/tools/path_utils.py

@@ -20,8 +20,10 @@
 # 导入模块
 # ==================================================
 import os
+import json
 import shutil
 import sys
+import tempfile
 from pathlib import Path
 from typing import Union
 from loguru import logger
@@ -534,3 +536,69 @@ def get_font_path(filename: str = DEFAULT_FONT_FILENAME_PRIMARY) -> Path:
         Path: 字体文件的绝对路径
     """
     return path_getter.get_font_path(filename)
+
+
+def atomic_write_json(
+    target_path: Union[str, Path],
+    data: dict,
+    indent: int = 4,
+    ensure_ascii: bool = False,
+) -> None:
+    """原子写入 JSON 文件，防止写入过程中崩溃导致数据丢失。
+
+    先写入临时文件，再通过 os.replace() 原子替换目标文件。
+    在 POSIX 系统上 os.replace() 是原子的；在 Windows 上也是
+    近似原子的（NTFS 上 REPLACEFILE 操作为原子）。
+
+    Args:
+        target_path: 目标文件路径（相对或绝对）
+        data: 要写入的字典数据
+        indent: JSON 缩进层级
+        ensure_ascii: 是否转义非 ASCII 字符
+    """
+    absolute_path = path_manager.get_absolute_path(target_path)
+    ensure_dir(absolute_path.parent)
+    dir_path = str(absolute_path.parent)
+
+    tmp_fd, tmp_path = tempfile.mkstemp(dir=dir_path, suffix=".tmp")
+    try:
+        with os.fdopen(tmp_fd, "w", encoding="utf-8") as f:
+            json.dump(data, f, indent=indent, ensure_ascii=ensure_ascii)
+            f.flush()
+            os.fsync(f.fileno())
+        os.replace(tmp_path, str(absolute_path))
+    except BaseException:
+        try:
+            os.unlink(tmp_path)
+        except OSError:
+            pass
+        raise
+
+
+def atomic_write_bytes(
+    target_path: Union[str, Path],
+    data: bytes,
+) -> None:
+    """原子写入二进制文件，防止写入过程中崩溃导致数据丢失。
+
+    Args:
+        target_path: 目标文件路径（相对或绝对）
+        data: 要写入的二进制数据
+    """
+    absolute_path = path_manager.get_absolute_path(target_path)
+    ensure_dir(absolute_path.parent)
+    dir_path = str(absolute_path.parent)
+
+    tmp_fd, tmp_path = tempfile.mkstemp(dir=dir_path, suffix=".tmp")
+    try:
+        with os.fdopen(tmp_fd, "wb") as f:
+            f.write(data)
+            f.flush()
+            os.fsync(f.fileno())
+        os.replace(tmp_path, str(absolute_path))
+    except BaseException:
+        try:
+            os.unlink(tmp_path)
+        except OSError:
+            pass
+        raise

app/tools/platform_report.py

@@ -9,7 +9,11 @@
 import requests
 from loguru import logger
 
-from app.tools.settings_access import get_int_setting, readme_settings_async, update_settings
+from app.tools.settings_access import (
+    get_int_setting,
+    readme_settings_async,
+    update_settings,
+)
 from app.tools.variable import (
     SECTL_API_BASE_URL,
     SECTL_ONLINE_REPORT_TIMEOUT_SECONDS,

app/tools/settings_access.py

@@ -18,6 +18,7 @@
 from app.tools.variable import *
 from app.tools.path_utils import *
 from app.tools.settings_default import *
+from app.tools.path_utils import atomic_write_json
 
 
 _UNSET = object()
@@ -374,9 +375,7 @@ def update_settings(first_level_key: str, second_level_key: str, value: Any):
         # 直接保存值，不保存嵌套结构
         settings_data[first_level_key][second_level_key] = value
 
-        # 写入设置文件
-        with open_file(settings_path, "w", encoding="utf-8") as f:
-            json.dump(settings_data, f, ensure_ascii=False, indent=4)
+        atomic_write_json(settings_path, settings_data)
         _replace_settings_cache(settings_data)
 
         if not (

app/tools/settings_default.py

@@ -10,11 +10,13 @@
 import platform
 import ctypes
 import uuid
+import zipfile
 from loguru import logger
 
 from app.tools.variable import *
 from app.tools.path_utils import *
 from app.tools.settings_default_storage import *
+from app.tools.path_utils import atomic_write_json
 
 Language = DEFAULT_LANGUAGE
 
@@ -53,6 +55,47 @@ def get_default_setting(first_level_key: str, second_level_key: str):
 _DEVICE_UUID_FILE = "device_uuid.json"
 
 
+def _try_recover_settings_from_backup() -> dict | None:
+    """尝试从最近的备份中恢复 settings.json 内容。
+
+    遍历备份目录中的 zip 文件（按修改时间倒序），查找包含
+    config/settings.json 的备份，返回其中可解析的设置字典。
+    如果所有备份都无法恢复则返回 None。
+    """
+    try:
+        backup_dir = get_data_path("backup")
+        if not backup_dir.exists():
+            return None
+
+        zip_files = sorted(
+            backup_dir.glob("*.zip"),
+            key=lambda p: p.stat().st_mtime,
+            reverse=True,
+        )
+
+        for zip_path in zip_files:
+            try:
+                with zipfile.ZipFile(str(zip_path), "r") as zf:
+                    for candidate in (
+                        "config/settings.json",
+                        "settings.json",
+                    ):
+                        if candidate in zf.namelist():
+                            with zf.open(candidate) as member:
+                                content = member.read().decode("utf-8")
+                            if content and content.strip():
+                                recovered = json.loads(content)
+                                if isinstance(recovered, dict):
+                                    logger.info(f"从备份 {zip_path.name} 恢复设置成功")
+                                    return recovered
+            except Exception:
+                continue
+    except Exception as e:
+        logger.warning(f"扫描备份恢复设置失败: {e}")
+
+    return None
+
+
 def ensure_device_uuid():
     """确保 offline_user_id 为标准的 36 位 UUID 格式
 
@@ -89,8 +132,7 @@ def ensure_device_uuid():
         new_uuid = str(uuid.uuid4()).lower()
         settings.setdefault("basic_settings", {})["offline_user_id"] = new_uuid
         try:
-            with open_file(settings_file, "w", encoding="utf-8") as f:
-                json.dump(settings, f, indent=4, ensure_ascii=False)
+            atomic_write_json(settings_file, settings)
             logger.info(f"已生成新的 offline_user_id: {new_uuid}")
         except Exception as e:
             logger.error(f"写入 offline_user_id 失败: {e}")
@@ -134,28 +176,31 @@ def manage_settings_file():
                             second_level_value["default_value"]
                         )
 
-            with open_file(settings_file, "w", encoding="utf-8") as f:
-                json.dump(flat_settings, f, indent=4, ensure_ascii=False)
+            atomic_write_json(settings_file, flat_settings)
             return
 
         try:
             with open_file(settings_file, "r", encoding="utf-8") as f:
                 current_settings = json.load(f)
         except Exception as e:
-            logger.warning(f"读取设置文件失败: {e}，将重新创建默认设置文件")
-            flat_settings = {}
-            for first_level_key, first_level_value in default_settings.items():
-                flat_settings[first_level_key] = {}
-                for second_level_key, second_level_value in first_level_value.items():
-                    # 如果默认值为 None，则不写入设置文件
-                    if second_level_value["default_value"] is not None:
-                        flat_settings[first_level_key][second_level_key] = (
-                            second_level_value["default_value"]
-                        )
-
-            with open_file(settings_file, "w", encoding="utf-8") as f:
-                json.dump(flat_settings, f, indent=4, ensure_ascii=False)
-            return
+            logger.warning(f"读取设置文件失败: {e}，尝试从备份恢复")
+            recovered = _try_recover_settings_from_backup()
+            if recovered is not None:
+                logger.info("从备份恢复设置成功，将使用恢复的设置继续合并")
+                current_settings = recovered
+            else:
+                logger.warning("无可用备份，将创建默认设置文件")
+                flat_settings = {}
+                for first_level_key, first_level_value in default_settings.items():
+                    flat_settings[first_level_key] = {}
+                    for second_level_key, second_level_value in first_level_value.items():
+                        if second_level_value["default_value"] is not None:
+                            flat_settings[first_level_key][second_level_key] = (
+                                second_level_value["default_value"]
+                            )
+
+                atomic_write_json(settings_file, flat_settings)
+                return
 
         # 检查并更新设置文件
         settings_updated = False
@@ -232,9 +277,7 @@ def manage_settings_file():
                         del updated_settings[first_level_key][second_level_key]
 
         if settings_updated:
-            # logger.debug("设置文件已更新")
-            with open_file(settings_file, "w", encoding="utf-8") as f:
-                json.dump(updated_settings, f, indent=4, ensure_ascii=False)
+            atomic_write_json(settings_file, updated_settings)
         else:
             # logger.debug("设置文件已是最新，无需更新")
             pass

app/tools/variable.py

@@ -345,7 +345,7 @@ def _normalize_arch(machine: str) -> str:
 PROCESS_EXIT_WAIT_SECONDS = 1  # 进程退出等待时间（秒）
 
 # -------------------- SECTL 在线状态上报配置 --------------------
-SECTL_API_BASE_URL = "https://appwrite.sectl.cn"
+SECTL_API_BASE_URL = "https://appwrite.sectl.cn"
 SECTL_PLATFORM_ID = "69c8cd6a0012dd3ea10a"
 SECTL_ONLINE_REPORT_INTERVAL_MS = 120000
 SECTL_ONLINE_REPORT_TIMEOUT_SECONDS = 10