优化路径

Author: lzy98276

CHANGELOG/v2.2.1/CHANGELOG.md

@@ -16,13 +16,16 @@ v2.0 - Koharu（小鸟游星野） release 4
 - 优化 **结果显示**，提升**颜色/图片**性能
 - 优化 **数据导出**，统一**通用导出**逻辑
 - 优化 **抽奖动画**，过程同步显示**学生**
+- 优化 **数据导入**，增加更多导入数据文件夹
+- 优化 **路径统一**，统一更多文件的**路径处理**
 
 ## 🐛 修复问题
 
 - 修复 **抽取逻辑**，修复**类型/参数**错误
 - 修复 **学生导出**，修复**AttributeError**
 - 修复 **音频模块**，修复**初始化/异常**问题
 - 修复 **配置读取**，修复**空文件解析**错误
+- 修复 **数据导入**，避免占用**失败**
 - 修复 **抽奖显示**，修复格式映射**KeyError**
 - 修复 **通知显示**，修复**父组件/回调**异常
 - 修复 **抽奖通知**，修复小组显示并统一**分隔符**

app/Language/modules/basic_settings.py

@@ -173,6 +173,9 @@
             "import_success_content": {
                 "name": "数据已成功导入\n重启应用程序以使更改生效"
             },
+            "import_success_content_skipped": {
+                "name": "数据已成功导入（跳过 {count} 个占用文件）\n重启应用程序以使更改生效"
+            },
             "import_success_button": {"name": "我知道了"},
             "import_failure_title": {"name": "导入所有数据"},
             "import_failure_content": {"name": "导入所有数据失败:\n{error}"},
@@ -378,6 +381,9 @@
             "import_success_content": {
                 "name": "Data imported successfully into\nRestart APP to take effect"
             },
+            "import_success_content_skipped": {
+                "name": "Data imported successfully (skipped {count} locked files)\nRestart APP to take effect"
+            },
             "import_success_button": {"name": "Got it"},
             "import_failure_title": {"name": "Import all data"},
             "import_failure_content": {"name": "Failed to import all data: \n{error}"},

app/common/extraction/extract.py

@@ -1,6 +1,4 @@
 import json
-import os
-from pathlib import Path
 from typing import Dict, Tuple
 
 from PySide6.QtCore import QDateTime
@@ -233,19 +231,19 @@ def _get_cses_parser() -> CSESParser | None:
     """
     try:
         cses_dir = get_data_path("CSES")
-        if not os.path.exists(cses_dir):
+        if not cses_dir.exists():
             logger.info("CSES目录不存在")
             return None
 
-        cses_file_path = os.path.join(cses_dir, "cses_schedule.yml")
+        cses_file_path = cses_dir / "cses_schedule.yml"
 
-        if not os.path.exists(cses_file_path):
+        if not cses_file_path.exists():
             logger.info("CSES文件不存在")
             return None
 
         parser = CSESParser()
-        if not parser.load_from_file(cses_file_path):
-            logger.exception(f"加载CSES文件失败: {cses_file_path}")
+        if not parser.load_from_file(str(cses_file_path)):
+            logger.exception(f"加载CSES文件失败: {str(cses_file_path)}")
             return None
 
         return parser
@@ -449,12 +447,11 @@ def import_cses_schedule(file_path: str) -> Tuple[bool, str]:
                 "linkage_settings", "no_valid_time_periods"
             )
 
-        original_file_name = Path(file_path).name
         cses_data_path = get_data_path("CSES", "cses_schedule.yml")
         ensure_dir(get_data_path("CSES"))
         import shutil
 
-        shutil.copy2(file_path, cses_data_path)
+        shutil.copy2(get_path(file_path), cses_data_path)
         logger.info(f"已将CSES文件保存到: {cses_data_path}")
 
         summary = parser.get_summary()

app/tools/config.py

@@ -9,6 +9,7 @@
 from datetime import datetime
 import platform
 import sys
+import time
 import psutil
 
 from PySide6.QtWidgets import QWidget, QFileDialog
@@ -940,11 +941,14 @@ def _export_diagnostic_files(file_path: str, app_dir: Path) -> int:
         导出的文件数量
     """
     export_folders = [
-        Path("config"),
-        Path("app") / "data" / "list",
-        Path("app") / "data" / "Language",
-        Path("app") / "data" / "history",
-        Path("logs"),
+        get_path("config"),
+        get_data_path("list"),
+        get_data_path("Language"),
+        get_data_path("history"),
+        get_data_path("audio"),
+        get_data_path("CSES"),
+        get_data_path("images"),
+        get_path(LOG_DIR),
     ]
 
     exported_count = 0
@@ -1061,11 +1065,14 @@ def _collect_system_info(
             "export_folders": [
                 str(folder)
                 for folder in [
-                    Path("config"),
-                    Path("app") / "data" / "list",
-                    Path("app") / "data" / "Language",
-                    Path("app") / "data" / "history",
-                    Path("logs"),
+                    get_path("config"),
+                    get_data_path("list"),
+                    get_data_path("Language"),
+                    get_data_path("history"),
+                    get_data_path("audio"),
+                    get_data_path("CSES"),
+                    get_data_path("images"),
+                    get_path(LOG_DIR),
                 ]
             ],
             "export_location": str(file_path),
@@ -1272,11 +1279,14 @@ def export_all_data(parent: Optional[QWidget] = None) -> None:
             file_path += ".zip"
 
         dirs_to_backup = [
-            ("config", Path("config")),
-            ("list", Path("app") / "data" / "list"),
-            ("Language", Path("app") / "data" / "Language"),
-            ("history", Path("app") / "data" / "history"),
-            ("logs", Path("logs")),
+            ("config", get_path("config")),
+            ("list", get_data_path("list")),
+            ("Language", get_data_path("Language")),
+            ("history", get_data_path("history")),
+            ("audio", get_data_path("audio")),
+            ("CSES", get_data_path("CSES")),
+            ("images", get_data_path("images")),
+            ("logs", get_path(LOG_DIR)),
         ]
 
         with zipfile.ZipFile(file_path, "w", zipfile.ZIP_DEFLATED) as zipf:
@@ -1360,7 +1370,7 @@ def import_all_data(parent: Optional[QWidget] = None) -> None:
             if not _confirm_import(parent):
                 return
 
-            _extract_data_files(file_path)
+            skipped_files = _extract_data_files(file_path)
 
             success_dialog = MessageBox(
                 get_any_position_value_async(
@@ -1369,11 +1379,20 @@ def import_all_data(parent: Optional[QWidget] = None) -> None:
                     "import_success_title",
                     "name",
                 ),
-                get_any_position_value_async(
-                    "basic_settings",
-                    "data_import_export",
-                    "import_success_content",
-                    "name",
+                (
+                    get_any_position_value_async(
+                        "basic_settings",
+                        "data_import_export",
+                        "import_success_content_skipped",
+                        "name",
+                    ).format(count=len(skipped_files))
+                    if skipped_files
+                    else get_any_position_value_async(
+                        "basic_settings",
+                        "data_import_export",
+                        "import_success_content",
+                        "name",
+                    )
                 ),
                 parent,
             )
@@ -1480,11 +1499,14 @@ def _check_existing_files(file_path: str) -> list:
     """检查已存在的文件"""
     existing_files = []
     target_dirs = {
-        "config": Path("config"),
-        "list": Path("data/list"),
-        "Language": Path("data/Language"),
-        "history": Path("data/history"),
-        "logs": Path("logs"),
+        "config": get_path("config"),
+        "list": get_data_path("list"),
+        "Language": get_data_path("Language"),
+        "history": get_data_path("history"),
+        "audio": get_data_path("audio"),
+        "CSES": get_data_path("CSES"),
+        "images": get_data_path("images"),
+        "logs": get_path(LOG_DIR),
     }
 
     with zipfile.ZipFile(file_path, "r") as zipf:
@@ -1562,16 +1584,22 @@ def _confirm_import(parent: Optional[QWidget]) -> bool:
     return dialog.exec()
 
 
-def _extract_data_files(file_path: str) -> None:
+def _extract_data_files(file_path: str) -> list:
     """提取数据文件"""
     target_dirs = {
-        "config": Path("config"),
-        "list": Path("data/list"),
-        "Language": Path("data/Language"),
-        "history": Path("data/history"),
-        "logs": Path("logs"),
+        "config": get_path("config"),
+        "list": get_data_path("list"),
+        "Language": get_data_path("Language"),
+        "history": get_data_path("history"),
+        "CSES": get_data_path("CSES"),
+        "images": get_data_path("images"),
+        "audio": get_data_path("audio"),
+        "logs": get_path(LOG_DIR),
     }
 
+    skipped_files = []
+    logs_root = get_path(LOG_DIR).resolve()
+
     with zipfile.ZipFile(file_path, "r") as zipf:
         for member in zipf.namelist():
             if member == "version.json":
@@ -1586,9 +1614,50 @@ def _extract_data_files(file_path: str) -> None:
                     target_path = target_dirs[dir_name] / relative_path
                     target_path.parent.mkdir(parents=True, exist_ok=True)
 
-                    with zipf.open(member) as source, open(target_path, "wb") as target:
+                    tmp_path = target_path.with_name(f"{target_path.name}.import_tmp")
+                    if tmp_path.exists():
+                        try:
+                            tmp_path.unlink()
+                        except Exception:
+                            pass
+
+                    with zipf.open(member) as source, open(tmp_path, "wb") as target:
                         shutil.copyfileobj(source, target)
 
+                    for attempt in range(3):
+                        try:
+                            os.replace(tmp_path, target_path)
+                            break
+                        except PermissionError as e:
+                            if attempt < 2:
+                                time.sleep(0.2)
+                                continue
+
+                            is_logs_file = False
+                            try:
+                                is_logs_file = target_path.resolve().is_relative_to(
+                                    logs_root
+                                )
+                            except Exception:
+                                is_logs_file = (
+                                    str(target_path)
+                                    .replace("\\", "/")
+                                    .startswith(str(logs_root).replace("\\", "/") + "/")
+                                )
+
+                            if is_logs_file:
+                                skipped_files.append(str(target_path))
+                                try:
+                                    tmp_path.unlink()
+                                except Exception:
+                                    pass
+                                logger.warning(f"导入文件被占用，已跳过: {target_path}")
+                                break
+
+                            raise e
+
+    return skipped_files
+
 
 # ==================== 记录管理模块 ====================
 

app/tools/update_utils.py

@@ -62,12 +62,13 @@ def check_exe_integrity(exe_path: str) -> bool:
         logger.debug(f"检查EXE安装程序完整性: {exe_path}")
 
         # 检查文件是否存在
-        if not Path(exe_path).exists():
+        exe_file = get_path(exe_path)
+        if not exe_file.exists():
             logger.exception(f"EXE文件不存在: {exe_path}")
             return False
 
         # 检查文件大小
-        file_size = Path(exe_path).stat().st_size
+        file_size = exe_file.stat().st_size
         if file_size == 0:
             logger.exception(f"EXE文件大小为0: {exe_path}")
             return False
@@ -100,7 +101,7 @@ def check_update_file_integrity(file_path: str, file_type: str = None) -> bool:
     try:
         # 自动检测文件类型
         if file_type is None:
-            file_ext = Path(file_path).suffix.lower()
+            file_ext = get_path(file_path).suffix.lower()
             if file_ext == ".exe":
                 file_type = "exe"
             else:
@@ -134,7 +135,7 @@ async def run_installer_and_exit(exe_path: str) -> bool:
         logger.info(f"准备运行安装程序: {exe_path}")
 
         # 验证安装程序存在
-        if not Path(exe_path).exists():
+        if not get_path(exe_path).exists():
             logger.exception(f"安装程序不存在: {exe_path}")
             return False
 
@@ -911,12 +912,12 @@ async def install_update_async(file_path: str) -> bool:
         logger.debug(f"开始安装更新文件: {file_path}")
 
         # 验证更新文件存在
-        if not Path(file_path).exists():
+        if not get_path(file_path).exists():
             logger.exception(f"更新文件不存在: {file_path}")
             return False
 
         # 检查文件类型
-        file_ext = Path(file_path).suffix.lower()
+        file_ext = get_path(file_path).suffix.lower()
 
         # 只支持 exe 安装程序
         if file_ext != ".exe":
@@ -928,7 +929,7 @@ async def install_update_async(file_path: str) -> bool:
             logger.exception(f"安装程序不完整或已损坏: {file_path}")
             # 删除损坏的文件
             try:
-                Path(file_path).unlink()
+                get_path(file_path).unlink(missing_ok=True)
                 logger.info(f"已删除损坏的安装程序: {file_path}")
             except Exception as e:
                 logger.exception(f"删除损坏的安装程序失败: {e}")
@@ -1406,9 +1407,7 @@ def progress_callback(downloaded: int, total: int):
                         logger.debug(f"自动下载更新成功: {file_path}")
 
                         # 获取文件大小
-                        from pathlib import Path
-
-                        file_size = Path(file_path).stat().st_size
+                        file_size = get_path(file_path).stat().st_size
 
                         def format_size(size_bytes):
                             """格式化文件大小"""

app/view/another_window/remaining_list.py

@@ -29,7 +29,7 @@
     get_content_name_async,
 )
 from app.tools.config import read_drawn_record, read_drawn_record_simple
-from app.tools.path_utils import get_data_path
+from app.tools.path_utils import get_data_path, get_path
 from app.tools.personalised import load_custom_font
 from app.tools.variable import (
     APP_INIT_DELAY,
@@ -58,7 +58,7 @@ def __init__(
         info_template: Optional[str],
     ) -> None:
         super().__init__()
-        self._students_file = Path(students_file)
+        self._students_file = get_path(students_file)
         self._class_name = class_name
         self._group_index = int(group_index or 0)
         self._gender_index = int(gender_index or 0)