diff --git a/app/common/history/file_utils.py b/app/common/history/file_utils.py index 4c68526b..3016b63c 100644 --- a/app/common/history/file_utils.py +++ b/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), diff --git a/app/common/safety/secure_store.py b/app/common/safety/secure_store.py index 32e3ff07..c140c333 100644 --- a/app/common/safety/secure_store.py +++ b/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}") diff --git a/app/tools/config.py b/app/tools/config.py index 75687c47..79636d32 100644 --- a/app/tools/config.py +++ b/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}") diff --git a/app/tools/path_utils.py b/app/tools/path_utils.py index 645e715a..d26aad7d 100644 --- a/app/tools/path_utils.py +++ b/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 diff --git a/app/tools/platform_report.py b/app/tools/platform_report.py index 28dda783..427aee79 100644 --- a/app/tools/platform_report.py +++ b/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, diff --git a/app/tools/settings_access.py b/app/tools/settings_access.py index ce536524..59ff13bc 100644 --- a/app/tools/settings_access.py +++ b/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 ( diff --git a/app/tools/settings_default.py b/app/tools/settings_default.py index ae966459..428e7f3e 100644 --- a/app/tools/settings_default.py +++ b/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 diff --git a/app/tools/variable.py b/app/tools/variable.py index 20d3a21c..0d28c90d 100644 --- a/app/tools/variable.py +++ b/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 diff --git a/app/view/another_window/student/import_student_name.py b/app/view/another_window/student/import_student_name.py index a60addbb..15a2e4c6 100644 --- a/app/view/another_window/student/import_student_name.py +++ b/app/view/another_window/student/import_student_name.py @@ -448,9 +448,9 @@ def __load_file(self, file_path: str): raise ValueError( get_content_name_async("import_student_name", "unsupported_format") ) - - # 将列名转换为字符串，避免整数列名与UI层字符串不匹配 - data.columns = [str(col) for col in data.columns] + + # 将列名转换为字符串，避免整数列名与UI层字符串不匹配 + data.columns = [str(col) for col in data.columns] # 获取列名 columns = list(data.columns) diff --git a/data/device_uuid.json b/data/device_uuid.json index 4a2e45ab..5a7f6b5e 100644 --- a/data/device_uuid.json +++ b/data/device_uuid.json @@ -1 +1 @@ -{"device_uuid": "280f5a64-c02f-49d3-b35b-362febaaee3f"} \ No newline at end of file +{"device_uuid": "280f5a64-c02f-49d3-b35b-362febaaee3f"}