tungstenowl/FRAISEMOE-Addons-Installer-NEXT
1
0
Fork 0
mirror of https://github.com/hyb-oyqq/FRAISEMOE-Addons-Installer-NEXT.git synced 2025-12-18 04:50:28 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(core): 增强补丁管理和进度反馈功能

Browse Source 
- 在主窗口中添加解压进度窗口，提供用户友好的解压反馈。
- 更新补丁检测器和下载管理器，支持异步游戏目录识别和补丁状态检查，提升用户体验。
- 优化哈希验证和解压流程，确保在关键操作中提供详细的进度信息和错误处理。
- 增强日志记录，确保在补丁管理过程中记录详细的调试信息，便于后续排查和用户反馈。
This commit is contained in:
hyb-oyqq
2025-08-11 11:14:38 +08:00
parent 09d6883432
commit f0031ed17c
12 changed files with 1198 additions and 726 deletions
Download Patch File Download Diff File Expand all files Collapse all files

274
source/core/download_manager.py
View File

@@ -109,13 +109,40 @@ class DownloadManager:
logger.debug(f"DEBUG: Parsed JSON data: {json.dumps(safe_config, indent=2)}")
urls = {}
missing_urls = []
# 检查每个游戏版本的URL
for i in range(4):
key = f"vol.{i+1}.data"
if key in config_data and "url" in config_data[key]:
urls[f"vol{i+1}"] = config_data[key]["url"]
else:
missing_urls.append(f"NEKOPARA Vol.{i+1}")
if self.is_debug_mode():
logger.warning(f"DEBUG: 未找到 NEKOPARA Vol.{i+1} 的下载URL")
# 检查After的URL
if "after.data" in config_data and "url" in config_data["after.data"]:
urls["after"] = config_data["after.data"]["url"]
else:
missing_urls.append("NEKOPARA After")
if self.is_debug_mode():
logger.warning(f"DEBUG: 未找到 NEKOPARA After 的下载URL")
# 如果有缺失的URL，记录详细信息
if missing_urls:
if self.is_debug_mode():
logger.warning(f"DEBUG: 以下游戏版本缺少下载URL: {', '.join(missing_urls)}")
logger.warning(f"DEBUG: 当前云端配置中的键: {list(config_data.keys())}")
# 检查每个游戏数据是否包含url键
for i in range(4):
key = f"vol.{i+1}.data"
if key in config_data:
logger.warning(f"DEBUG: {key} 内容: {list(config_data[key].keys())}")
if "after.data" in config_data:
logger.warning(f"DEBUG: after.data 内容: {list(config_data['after.data'].keys())}")
if len(urls) != 5:
missing_keys_map = {
@@ -128,7 +155,17 @@ class DownloadManager:
missing_simple_keys = all_keys - extracted_keys
missing_original_keys = [missing_keys_map[k] for k in missing_simple_keys]
raise ValueError(f"配置文件缺少必要的键: {', '.join(missing_original_keys)}")
# 记录详细的缺失信息
if self.is_debug_mode():
logger.warning(f"DEBUG: 缺失的URL键: {missing_original_keys}")
# 如果所有URL都缺失，可能是云端配置问题
if len(urls) == 0:
raise ValueError(f"配置文件缺少所有下载URL键: {', '.join(missing_original_keys)}")
# 否则只是部分缺失，可以继续使用已有的URL
logger.warning(f"配置文件缺少部分键: {', '.join(missing_original_keys)}")
if self.is_debug_mode():
# 创建安全版本的URL字典用于调试输出
@@ -218,9 +255,16 @@ class DownloadManager:
self.main_window.setEnabled(True)
self.main_window.ui.start_install_text.setText("开始安装")
return
# 关闭可能存在的哈希校验窗口
self.main_window.close_hash_msg_box()
# 显示文件检验窗口
self.main_window.hash_msg_box = self.main_window.hash_manager.hash_pop_window(check_type="pre")
self.main_window.hash_msg_box = self.main_window.hash_manager.hash_pop_window(
check_type="pre",
auto_close=True, # 添加自动关闭参数
close_delay=1000 # 1秒后自动关闭
)
# 获取安装路径
install_paths = self.get_install_paths()
@@ -240,9 +284,9 @@ class DownloadManager:
game_dirs: 识别到的游戏目录
"""
self.main_window.installed_status = updated_status
if self.main_window.hash_msg_box and self.main_window.hash_msg_box.isVisible():
self.main_window.hash_msg_box.accept()
self.main_window.hash_msg_box = None
# 关闭哈希校验窗口
self.main_window.close_hash_msg_box()
debug_mode = self.is_debug_mode()
@@ -296,96 +340,145 @@ class DownloadManager:
logger.info(f"DEBUG: 用户选择不启用被禁用的补丁，这些游戏将被添加到可安装列表")
# 用户选择不启用，将这些游戏视为可以安装补丁
installable_games.extend(disabled_patch_games)
# 更新status_message
if already_installed_games:
status_message = f"已安装补丁的游戏：\n{chr(10).join(already_installed_games)}\n\n"
# 如果有可安装的游戏，显示选择对话框
if installable_games:
# 创建游戏选择对话框
dialog = QtWidgets.QDialog(self.main_window)
dialog.setWindowTitle(f"选择要安装的游戏 - {APP_NAME}")
dialog.setMinimumWidth(400)
dialog.setMinimumHeight(300)
if not installable_games:
layout = QtWidgets.QVBoxLayout()
# 添加说明标签
label = QtWidgets.QLabel("请选择要安装的游戏：")
layout.addWidget(label)
# 添加已安装游戏的状态提示
if already_installed_games:
installed_label = QtWidgets.QLabel(status_message)
installed_label.setStyleSheet("color: green;")
layout.addWidget(installed_label)
# 创建列表控件
list_widget = QtWidgets.QListWidget()
list_widget.setSelectionMode(QtWidgets.QAbstractItemView.SelectionMode.MultiSelection)
# 添加可安装的游戏
for game in installable_games:
item = QtWidgets.QListWidgetItem(game)
item.setSelected(True) # 默认全选
list_widget.addItem(item)
layout.addWidget(list_widget)
# 添加按钮
button_layout = QtWidgets.QHBoxLayout()
ok_button = QtWidgets.QPushButton("确定")
cancel_button = QtWidgets.QPushButton("取消")
button_layout.addWidget(ok_button)
button_layout.addWidget(cancel_button)
layout.addLayout(button_layout)
dialog.setLayout(layout)
# 连接按钮信号
ok_button.clicked.connect(dialog.accept)
cancel_button.clicked.connect(dialog.reject)
# 显示对话框
if dialog.exec() == QtWidgets.QDialog.DialogCode.Accepted:
selected_games = [item.text() for item in list_widget.selectedItems()]
if debug_mode:
logger.debug(f"DEBUG: 用户选择了以下游戏进行安装: {selected_games}")
selected_game_dirs = {game: game_dirs[game] for game in selected_games if game in game_dirs}
self.main_window.setEnabled(False)
# 检查是否处于离线模式
is_offline_mode = False
if hasattr(self.main_window, 'offline_mode_manager'):
is_offline_mode = self.main_window.offline_mode_manager.is_in_offline_mode()
if is_offline_mode:
if debug_mode:
logger.info("DEBUG: 使用离线模式，跳过网络配置获取")
self._fill_offline_download_queue(selected_game_dirs)
else:
# 在线模式下，重新获取云端配置
if hasattr(self.main_window, 'fetch_cloud_config'):
if debug_mode:
logger.info("DEBUG: 重新获取云端配置以确保URL最新")
# 重新获取云端配置并继续下载流程
from workers.config_fetch_thread import ConfigFetchThread
self.main_window.config_manager.fetch_cloud_config(
ConfigFetchThread,
lambda data, error: self._continue_download_after_config_fetch(data, error, selected_game_dirs)
)
else:
# 如果无法重新获取配置，使用当前配置
config = self.get_download_url()
self._continue_download_with_config(config, selected_game_dirs)
else:
if debug_mode:
logger.debug("DEBUG: 用户取消了游戏选择")
self.main_window.ui.start_install_text.setText("开始安装")
else:
# 如果没有可安装的游戏，显示提示
if already_installed_games:
msg = f"所有游戏已安装补丁，无需重复安装。\n\n已安装的游戏：\n{chr(10).join(already_installed_games)}"
else:
msg = "未检测到可安装的游戏。"
QtWidgets.QMessageBox.information(
self.main_window,
f"信息 - {APP_NAME}",
f"\n所有检测到的游戏都已安装补丁。\n\n{status_message}"
self.main_window,
f"通知 - {APP_NAME}",
msg
)
self.main_window.ui.start_install_text.setText("开始安装")
def _continue_download_after_config_fetch(self, data, error, selected_game_dirs):
"""云端配置获取完成后继续下载流程
Args:
data: 获取到的配置数据
error: 错误信息
selected_game_dirs: 选择的游戏目录
"""
debug_mode = self.is_debug_mode()
if error:
if debug_mode:
logger.error(f"DEBUG: 重新获取云端配置失败: {error}")
# 使用当前配置
config = self.get_download_url()
else:
# 使用新获取的配置
self.main_window.cloud_config = data
config = self.get_download_url()
self._continue_download_with_config(config, selected_game_dirs)
def _continue_download_with_config(self, config, selected_game_dirs):
"""使用配置继续下载流程
Args:
config: 下载配置
selected_game_dirs: 选择的游戏目录
"""
debug_mode = self.is_debug_mode()
if not config:
QtWidgets.QMessageBox.critical(
self.main_window, f"错误 - {APP_NAME}", "\n网络状态异常或服务器故障，请重试\n"
)
self.main_window.setEnabled(True)
self.main_window.ui.start_install_text.setText("开始安装")
return
dialog = QtWidgets.QDialog(self.main_window)
dialog.setWindowTitle("选择要安装的游戏")
dialog.resize(400, 300)
layout = QtWidgets.QVBoxLayout(dialog)
if already_installed_games:
already_installed_label = QtWidgets.QLabel("已安装补丁的游戏:", dialog)
already_installed_label.setFont(QFont(already_installed_label.font().family(), already_installed_label.font().pointSize(), QFont.Weight.Bold))
layout.addWidget(already_installed_label)
already_installed_list = QtWidgets.QLabel(chr(10).join(already_installed_games), dialog)
layout.addWidget(already_installed_list)
layout.addSpacing(10)
info_label = QtWidgets.QLabel("请选择你需要安装补丁的游戏:", dialog)
info_label.setFont(QFont(info_label.font().family(), info_label.font().pointSize(), QFont.Weight.Bold))
layout.addWidget(info_label)
list_widget = QtWidgets.QListWidget(dialog)
list_widget.setSelectionMode(QtWidgets.QAbstractItemView.SelectionMode.ExtendedSelection)
for game in installable_games:
list_widget.addItem(game)
layout.addWidget(list_widget)
select_all_btn = QPushButton("全选", dialog)
select_all_btn.clicked.connect(lambda: list_widget.selectAll())
layout.addWidget(select_all_btn)
buttons_layout = QHBoxLayout()
ok_button = QPushButton("确定", dialog)
cancel_button = QPushButton("取消", dialog)
buttons_layout.addWidget(ok_button)
buttons_layout.addWidget(cancel_button)
layout.addLayout(buttons_layout)
ok_button.clicked.connect(dialog.accept)
cancel_button.clicked.connect(dialog.reject)
result = dialog.exec()
if result != QDialog.DialogCode.Accepted or list_widget.selectedItems() == []:
self.main_window.setEnabled(True)
self.main_window.ui.start_install_text.setText("开始安装")
return
selected_games = [item.text() for item in list_widget.selectedItems()]
if debug_mode:
logger.debug(f"DEBUG: 用户选择了以下游戏进行安装: {selected_games}")
selected_game_dirs = {game: game_dirs[game] for game in selected_games if game in game_dirs}
self.main_window.setEnabled(False)
# 检查是否处于离线模式
is_offline_mode = False
if hasattr(self.main_window, 'offline_mode_manager'):
is_offline_mode = self.main_window.offline_mode_manager.is_in_offline_mode()
if is_offline_mode:
if debug_mode:
logger.info("DEBUG: 使用离线模式，跳过网络配置获取")
self._fill_offline_download_queue(selected_game_dirs)
else:
config = self.get_download_url()
if not config:
QtWidgets.QMessageBox.critical(
self.main_window, f"错误 - {APP_NAME}", "\n网络状态异常或服务器故障，请重试\n"
)
self.main_window.setEnabled(True)
self.main_window.ui.start_install_text.setText("开始安装")
return
self._fill_download_queue(config, selected_game_dirs)
self._fill_download_queue(config, selected_game_dirs)
if not self.download_queue:
# 所有下载任务都已完成，进行后检查
@@ -395,6 +488,11 @@ class DownloadManager:
self.main_window.patch_detector.after_hash_compare()
return
# 检查是否处于离线模式
is_offline_mode = False
if hasattr(self.main_window, 'offline_mode_manager'):
is_offline_mode = self.main_window.offline_mode_manager.is_in_offline_mode()
# 如果是离线模式，直接开始下一个下载任务
if is_offline_mode:
if debug_mode:

86
source/core/extraction_handler.py
View File

@@ -2,7 +2,7 @@ import os
import shutil
from PySide6 import QtWidgets
from PySide6.QtWidgets import QMessageBox
from PySide6.QtCore import QTimer
from PySide6.QtCore import QTimer, QCoreApplication
from utils.logger import setup_logger
@@ -20,6 +20,7 @@ class ExtractionHandler:
"""
self.main_window = main_window
self.APP_NAME = main_window.APP_NAME if hasattr(main_window, 'APP_NAME') else ""
self.extraction_progress_window = None
def start_extraction(self, _7z_path, game_folder, plugin_path, game_version, extracted_path=None):
"""开始解压任务
@@ -36,19 +37,41 @@ class ExtractionHandler:
if hasattr(self.main_window, 'offline_mode_manager'):
is_offline = self.main_window.offline_mode_manager.is_in_offline_mode()
# 显示解压中的消息窗口
self.main_window.hash_msg_box = self.main_window.hash_manager.hash_pop_window(
check_type="offline_extraction" if is_offline else "extraction",
is_offline=is_offline
)
# 创建并显示解压进度窗口，替代原来的消息框
self.extraction_progress_window = self.main_window.create_extraction_progress_window()
self.extraction_progress_window.show()
# 确保UI更新
QCoreApplication.processEvents()
# 创建并启动解压线程
self.main_window.extraction_thread = self.main_window.create_extraction_thread(
_7z_path, game_folder, plugin_path, game_version, extracted_path
)
# 连接进度信号
self.main_window.extraction_thread.progress.connect(self.update_extraction_progress)
# 连接完成信号
self.main_window.extraction_thread.finished.connect(self.on_extraction_finished_with_hash_check)
# 启动线程
self.main_window.extraction_thread.start()
def update_extraction_progress(self, progress, status_text):
"""更新解压进度
Args:
progress: 进度百分比
status_text: 状态文本
"""
if self.extraction_progress_window and hasattr(self.extraction_progress_window, 'progress_bar'):
self.extraction_progress_window.progress_bar.setValue(progress)
self.extraction_progress_window.status_label.setText(status_text)
# 确保UI更新
QCoreApplication.processEvents()
def on_extraction_finished_with_hash_check(self, success, error_message, game_version):
"""解压完成后进行哈希校验
@@ -57,10 +80,10 @@ class ExtractionHandler:
error_message: 错误信息
game_version: 游戏版本
"""
# 关闭哈希检查窗口
if self.main_window.hash_msg_box and self.main_window.hash_msg_box.isVisible():
self.main_window.hash_msg_box.close()
self.main_window.hash_msg_box = None
# 关闭解压进度窗口
if self.extraction_progress_window:
self.extraction_progress_window.close()
self.extraction_progress_window = None
# 如果解压失败，显示错误并询问是否继续
if not success:
@@ -100,6 +123,10 @@ class ExtractionHandler:
Args:
game_version: 游戏版本
"""
# 导入所需模块
from data.config import GAME_INFO, PLUGIN_HASH
from workers.hash_thread import HashThread
# 获取安装路径
install_paths = {}
if hasattr(self.main_window, 'game_detector') and hasattr(self.main_window, 'download_manager'):
@@ -107,7 +134,7 @@ class ExtractionHandler:
self.main_window.download_manager.selected_folder
)
for game, info in self.main_window.GAME_INFO.items():
for game, info in GAME_INFO.items():
if game in game_dirs and game == game_version:
game_dir = game_dirs[game]
install_path = os.path.join(game_dir, os.path.basename(info["install_path"]))
@@ -120,19 +147,30 @@ class ExtractionHandler:
self.main_window.installed_status[game_version] = True
self.main_window.download_manager.on_extraction_finished(True)
return
# 关闭可能存在的哈希校验窗口
self.main_window.close_hash_msg_box()
# 显示哈希校验窗口
self.main_window.hash_msg_box = self.main_window.hash_manager.hash_pop_window(check_type="post")
self.main_window.hash_msg_box = self.main_window.hash_manager.hash_pop_window(
check_type="post",
auto_close=True, # 添加自动关闭参数
close_delay=1000 # 1秒后自动关闭
)
# 创建并启动哈希线程进行校验
self.main_window.hash_thread = self.main_window.create_hash_thread(
# 直接创建并启动哈希线程进行校验
hash_thread = HashThread(
"after",
install_paths,
self.main_window.plugin_hash,
self.main_window.installed_status
PLUGIN_HASH,
self.main_window.installed_status,
self.main_window
)
self.main_window.hash_thread.after_finished.connect(self.on_hash_check_finished)
self.main_window.hash_thread.start()
hash_thread.after_finished.connect(self.on_hash_check_finished)
# 保存引用以便后续使用
self.hash_thread = hash_thread
hash_thread.start()
def on_hash_check_finished(self, result):
"""哈希校验完成后的处理
@@ -140,10 +178,11 @@ class ExtractionHandler:
Args:
result: 校验结果，包含通过状态、游戏版本和消息
"""
# 导入所需模块
from data.config import GAME_INFO
# 关闭哈希检查窗口
if self.main_window.hash_msg_box and self.main_window.hash_msg_box.isVisible():
self.main_window.hash_msg_box.close()
self.main_window.hash_msg_box = None
self.main_window.close_hash_msg_box()
if not result["passed"]:
# 校验失败，删除已解压的文件并提示重新下载
@@ -160,9 +199,9 @@ class ExtractionHandler:
self.main_window.download_manager.selected_folder
)
if game_version in game_dirs and game_version in self.main_window.GAME_INFO:
if game_version in game_dirs and game_version in GAME_INFO:
game_dir = game_dirs[game_version]
install_path = os.path.join(game_dir, os.path.basename(self.main_window.GAME_INFO[game_version]["install_path"]))
install_path = os.path.join(game_dir, os.path.basename(GAME_INFO[game_version]["install_path"]))
# 如果找到安装路径，尝试删除已解压的文件
if install_path and os.path.exists(install_path):
@@ -209,6 +248,7 @@ class ExtractionHandler:
self.main_window.download_manager.on_extraction_finished(True)
else:
# 校验通过，更新安装状态
game_version = result["game"]
self.main_window.installed_status[game_version] = True
# 通知DownloadManager继续下一个下载任务
self.main_window.download_manager.on_extraction_finished(True)

25
source/core/game_detector.py
View File

@@ -1,7 +1,21 @@
from PySide6.QtCore import QThread, Signal
import os
import re
from utils.logger import setup_logger
class GameDetectionThread(QThread):
"""用于在后台线程中执行游戏目录识别的线程"""
finished = Signal(dict)
def __init__(self, detector_func, selected_folder):
super().__init__()
self.detector_func = detector_func
self.selected_folder = selected_folder
def run(self):
result = self.detector_func(self.selected_folder)
self.finished.emit(result)
class GameDetector:
"""游戏检测器，用于识别游戏目录和版本"""
@@ -16,6 +30,17 @@ class GameDetector:
self.debug_manager = debug_manager
self.directory_cache = {} # 添加目录缓存
self.logger = setup_logger("game_detector")
self.detection_thread = None
def identify_game_directories_async(self, selected_folder, callback):
"""异步识别游戏目录"""
def on_finished(game_dirs):
callback(game_dirs)
self.detection_thread = None
self.detection_thread = GameDetectionThread(self.identify_game_directories_improved, selected_folder)
self.detection_thread.finished.connect(on_finished)
self.detection_thread.start()
def _is_debug_mode(self):
"""检查是否处于调试模式

391
source/core/offline_mode_manager.py
View File

@@ -322,15 +322,13 @@ class OfflineModeManager:
debug_mode = self._is_debug_mode()
# 导入所需模块
from data.config import GAME_INFO
from data.config import GAME_INFO, PLUGIN
# 存储结果到对话框，以便在exec()返回后获取
dialog.hash_result = result
# 关闭哈希验证窗口
if self.main_window.hash_msg_box and self.main_window.hash_msg_box.isVisible():
self.main_window.hash_msg_box.close()
self.main_window.hash_msg_box = None
self.main_window.close_hash_msg_box()
if not result:
# 哈希验证失败
@@ -348,78 +346,110 @@ class OfflineModeManager:
self.process_next_offline_install_task(install_tasks)
return
# 哈希验证成功，直接进行安装（复制文件）
# 哈希验证成功，直接进行安装
if debug_mode:
logger.debug(f"DEBUG: 哈希验证成功，直接进行安装")
if extracted_path:
logger.debug(f"DEBUG: 使用已解压的补丁文件: {extracted_path}")
logger.debug(f"DEBUG: 哈希验证成功，开始安装")
# 显示安装进度窗口
self.main_window.hash_msg_box = self.main_window.hash_manager.hash_pop_window(check_type="offline_installation", is_offline=True)
try:
# 直接复制已解压的文件到游戏目录
# 确保游戏目录存在
os.makedirs(game_folder, exist_ok=True)
# 获取目标文件路径
target_file = None
# 根据游戏版本确定目标文件名
target_filename = None
if "Vol.1" in game_version:
target_file = os.path.join(game_folder, "adultsonly.xp3")
target_filename = "adultsonly.xp3"
elif "Vol.2" in game_version:
target_file = os.path.join(game_folder, "adultsonly.xp3")
target_filename = "adultsonly.xp3"
elif "Vol.3" in game_version:
target_file = os.path.join(game_folder, "update00.int")
target_filename = "update00.int"
elif "Vol.4" in game_version:
target_file = os.path.join(game_folder, "vol4adult.xp3")
target_filename = "vol4adult.xp3"
elif "After" in game_version:
target_file = os.path.join(game_folder, "afteradult.xp3")
target_filename = "afteradult.xp3"
if not target_file:
if not target_filename:
raise ValueError(f"未知的游戏版本: {game_version}")
# 复制文件
shutil.copy2(extracted_path, target_file)
# 对于NEKOPARA After，还需要复制签名文件
if game_version == "NEKOPARA After":
# 从已解压文件的目录中获取签名文件
extracted_dir = os.path.dirname(extracted_path)
sig_filename = os.path.basename(GAME_INFO[game_version]["sig_path"])
sig_path = os.path.join(extracted_dir, sig_filename)
# 如果签名文件存在，则复制它
if os.path.exists(sig_path):
shutil.copy(sig_path, game_folder)
else:
# 如果签名文件不存在，则使用原始路径
sig_path = os.path.join(PLUGIN, GAME_INFO[game_version]["sig_path"])
shutil.copy(sig_path, game_folder)
# 更新安装状态
self.main_window.installed_status[game_version] = True
# 直接解压文件到游戏目录
import py7zr
# 添加到已安装游戏列表
if game_version not in self.installed_games:
self.installed_games.append(game_version)
if debug_mode:
logger.debug(f"DEBUG: 成功安装 {game_version} 补丁文件")
logger.debug(f"DEBUG: 直接解压文件 {_7z_path} 到游戏目录 {game_folder}")
# 解压文件
with py7zr.SevenZipFile(_7z_path, mode="r") as archive:
# 获取压缩包内的文件列表
file_list = archive.getnames()
if debug_mode:
logger.debug(f"DEBUG: 压缩包内文件列表: {file_list}")
# 关闭安装进度窗口
if self.main_window.hash_msg_box and self.main_window.hash_msg_box.isVisible():
self.main_window.hash_msg_box.close()
self.main_window.hash_msg_box = None
# 解析压缩包内的文件结构
target_file_in_archive = None
for file_path in file_list:
if target_filename in file_path:
target_file_in_archive = file_path
break
# 继续下一个任务
self.process_next_offline_install_task(install_tasks)
if not target_file_in_archive:
if debug_mode:
logger.warning(f"DEBUG: 在压缩包中未找到目标文件 {target_filename}")
raise FileNotFoundError(f"在压缩包中未找到目标文件 {target_filename}")
# 准备解压特定文件到游戏目录
target_path = os.path.join(game_folder, target_filename)
# 创建一个临时目录用于解压单个文件
with tempfile.TemporaryDirectory() as temp_dir:
# 解压特定文件到临时目录
archive.extract(path=temp_dir, targets=[target_file_in_archive])
# 找到解压后的文件
extracted_file_path = os.path.join(temp_dir, target_file_in_archive)
# 复制到目标位置
shutil.copy2(extracted_file_path, target_path)
if debug_mode:
logger.debug(f"DEBUG: 已解压并复制文件到 {target_path}")
# 对于NEKOPARA After，还需要复制签名文件
if game_version == "NEKOPARA After":
sig_filename = f"{target_filename}.sig"
sig_file_in_archive = None
# 查找签名文件
for file_path in file_list:
if sig_filename in file_path:
sig_file_in_archive = file_path
break
if sig_file_in_archive:
# 解压签名文件
archive.extract(path=temp_dir, targets=[sig_file_in_archive])
extracted_sig_path = os.path.join(temp_dir, sig_file_in_archive)
sig_target = os.path.join(game_folder, sig_filename)
shutil.copy2(extracted_sig_path, sig_target)
if debug_mode:
logger.debug(f"DEBUG: 已解压并复制签名文件到 {sig_target}")
else:
if debug_mode:
logger.warning(f"DEBUG: 未找到签名文件 {sig_filename}")
# 进行安装后的哈希校验
self._perform_hash_check(game_version, install_tasks)
except Exception as e:
if debug_mode:
logger.error(f"DEBUG: 安装补丁文件失败: {e}")
import traceback
logger.error(f"DEBUG: 错误堆栈: {traceback.format_exc()}")
# 关闭安装进度窗口
if self.main_window.hash_msg_box and self.main_window.hash_msg_box.isVisible():
self.main_window.hash_msg_box.close()
self.main_window.hash_msg_box = None
self.main_window.close_hash_msg_box()
# 显示错误消息
msgbox_frame(
@@ -431,6 +461,130 @@ class OfflineModeManager:
# 继续下一个任务
self.process_next_offline_install_task(install_tasks)
def _perform_hash_check(self, game_version, install_tasks):
"""安装完成后进行哈希校验
Args:
game_version: 游戏版本
install_tasks: 剩余的安装任务列表
"""
debug_mode = self._is_debug_mode()
# 导入所需模块
from data.config import GAME_INFO, PLUGIN_HASH
from workers.hash_thread import HashThread
# 获取安装路径
install_paths = {}
game_dirs = self.main_window.game_detector.identify_game_directories_improved(
self.main_window.download_manager.selected_folder
)
for game, info in GAME_INFO.items():
if game in game_dirs and game == game_version:
game_dir = game_dirs[game]
install_path = os.path.join(game_dir, os.path.basename(info["install_path"]))
install_paths[game] = install_path
break
if not install_paths:
# 如果找不到安装路径，直接认为安装成功
logger.warning(f"未找到 {game_version} 的安装路径，跳过哈希校验")
self.main_window.installed_status[game_version] = True
# 添加到已安装游戏列表
if game_version not in self.installed_games:
self.installed_games.append(game_version)
# 关闭安装进度窗口
self.main_window.close_hash_msg_box()
# 继续下一个任务
self.process_next_offline_install_task(install_tasks)
return
# 关闭可能存在的哈希校验窗口，然后创建新窗口
self.main_window.close_hash_msg_box()
# 显示哈希校验窗口
self.main_window.hash_msg_box = self.main_window.hash_manager.hash_pop_window(check_type="post", is_offline=True)
# 直接创建并启动哈希线程进行校验，而不是通过主窗口
hash_thread = HashThread(
"after",
install_paths,
PLUGIN_HASH,
self.main_window.installed_status,
self.main_window
)
hash_thread.after_finished.connect(
lambda result: self._on_hash_check_finished(result, game_version, install_tasks)
)
# 保存引用以便后续使用
self.hash_thread = hash_thread
hash_thread.start()
def _on_hash_check_finished(self, result, game_version, install_tasks):
"""哈希校验完成后的处理
Args:
result: 校验结果，包含通过状态、游戏版本和消息
game_version: 游戏版本
install_tasks: 剩余的安装任务列表
"""
debug_mode = self._is_debug_mode()
# 关闭哈希检查窗口
self.main_window.close_hash_msg_box()
if not result["passed"]:
# 校验失败，删除已解压的文件并提示重新安装
error_message = result["message"]
# 获取安装路径
install_path = None
game_dirs = self.main_window.game_detector.identify_game_directories_improved(
self.main_window.download_manager.selected_folder
)
from data.config import GAME_INFO
if game_version in game_dirs and game_version in GAME_INFO:
game_dir = game_dirs[game_version]
install_path = os.path.join(game_dir, os.path.basename(GAME_INFO[game_version]["install_path"]))
# 如果找到安装路径，尝试删除已解压的文件
if install_path and os.path.exists(install_path):
try:
os.remove(install_path)
logger.info(f"已删除校验失败的文件: {install_path}")
except Exception as e:
logger.error(f"删除文件失败: {e}")
# 显示错误消息
msgbox_frame(
f"校验失败 - {self.app_name}",
f"{error_message}\n\n跳过此游戏的安装。",
QMessageBox.StandardButton.Ok
).exec()
# 更新安装状态
self.main_window.installed_status[game_version] = False
else:
# 校验通过，更新安装状态
self.main_window.installed_status[game_version] = True
# 添加到已安装游戏列表
if game_version not in self.installed_games:
self.installed_games.append(game_version)
# 显示安装成功消息
if debug_mode:
logger.debug(f"DEBUG: {game_version} 安装成功并通过哈希校验")
# 继续处理下一个任务
self.process_next_offline_install_task(install_tasks)
def _on_extraction_finished_with_hash_check(self, success, error_message, game_version, install_tasks):
"""解压完成后进行哈希校验
@@ -692,67 +846,96 @@ class OfflineModeManager:
logger.debug(f"DEBUG: 补丁文件: {patch_file}")
logger.debug(f"DEBUG: 游戏目录: {game_folder}")
# 确保目标目录存在
os.makedirs(os.path.dirname(_7z_path), exist_ok=True)
# 显示安装进度窗口
self.main_window.hash_msg_box = self.main_window.hash_manager.hash_pop_window(check_type="offline_installation", is_offline=True)
try:
# 复制补丁文件到缓存目录
shutil.copy2(patch_file, _7z_path)
# 确保游戏目录存在
os.makedirs(game_folder, exist_ok=True)
# 从GAME_INFO获取目标文件名
target_filename = os.path.basename(GAME_INFO[game_version]["install_path"])
if not target_filename:
raise ValueError(f"未知的游戏版本或配置错误: {game_version}")
# 直接从源7z文件解压
with py7zr.SevenZipFile(patch_file, mode="r") as archive:
file_list = archive.getnames()
target_file_in_archive = None
# 查找压缩包中的目标文件
for f_path in file_list:
if target_filename in f_path:
target_file_in_archive = f_path
break
if not target_file_in_archive:
raise FileNotFoundError(f"在压缩包 {os.path.basename(patch_file)} 中未找到目标文件 {target_filename}")
# 使用临时目录来解压单个文件
with tempfile.TemporaryDirectory() as temp_dir:
archive.extract(path=temp_dir, targets=[target_file_in_archive])
extracted_file_path = os.path.join(temp_dir, target_file_in_archive)
# 最终目标路径
target_path = os.path.join(game_folder, target_filename)
# 复制到游戏目录
shutil.copy2(extracted_file_path, target_path)
if debug_mode:
logger.debug(f"DEBUG: 已解压并复制文件到 {target_path}")
# 对于NEKOPARA After，还需要处理签名文件
if game_version == "NEKOPARA After":
sig_filename = f"{target_filename}.sig"
sig_file_in_archive = None
for f_path in file_list:
if sig_filename in f_path:
sig_file_in_archive = f_path
break
if sig_file_in_archive:
try:
archive.extract(path=temp_dir, targets=[sig_file_in_archive])
extracted_sig_path = os.path.join(temp_dir, sig_file_in_archive)
sig_target = os.path.join(game_folder, sig_filename)
shutil.copy2(extracted_sig_path, sig_target)
if debug_mode:
logger.debug(f"DEBUG: 已解压并复制签名文件到 {sig_target}")
except py7zr.exceptions.CrcError as sig_e:
if debug_mode:
logger.warning(f"DEBUG: 签名文件 '{sig_e.filename}' CRC校验失败，已忽略此文件。")
# 进行安装后的哈希校验
self._perform_hash_check(game_version, install_tasks)
except py7zr.exceptions.CrcError as e:
if debug_mode:
logger.debug(f"DEBUG: 已复制补丁文件到缓存目录: {_7z_path}")
logger.debug(f"DEBUG: 开始验证补丁文件哈希值")
# 验证补丁文件哈希
hash_valid = False
extracted_path = None
logger.error(f"DEBUG: CRC校验失败，文件可能已损坏: {e}")
logger.error(f"DEBUG: 错误堆栈: {traceback.format_exc()}")
self.main_window.close_hash_msg_box()
msgbox_frame(
f"安装错误 - {self.app_name}",
f"\n补丁文件 {os.path.basename(patch_file)} 在解压时CRC校验失败。\n"
f"这通常意味着文件已损坏，请尝试重新下载该文件。\n\n"
f"游戏: {game_version}\n"
f"错误文件: {e.filename}\n\n"
"跳过此游戏的安装。",
QMessageBox.StandardButton.Ok
).exec()
# 显示哈希验证窗口 - 使用离线特定消息
self.main_window.hash_msg_box = self.main_window.hash_manager.hash_pop_window(check_type="offline_verify", is_offline=True)
# 验证补丁文件哈希
# 使用特殊版本的verify_patch_hash方法，它会返回哈希验证结果和解压后的文件路径
from utils.helpers import ProgressHashVerifyDialog
from data.config import PLUGIN_HASH
from workers.hash_thread import OfflineHashVerifyThread
# 创建并显示进度对话框
progress_dialog = ProgressHashVerifyDialog(
f"验证补丁文件 - {self.app_name}",
f"正在验证 {game_version} 的补丁文件完整性...",
self.main_window
)
# 创建哈希验证线程
hash_thread = OfflineHashVerifyThread(game_version, _7z_path, PLUGIN_HASH, self.main_window)
# 存储解压后的文件路径
extracted_file_path = ""
# 连接信号
hash_thread.progress.connect(progress_dialog.update_progress)
hash_thread.finished.connect(
lambda result, error, path: self._on_offline_install_hash_finished(
result, error, path, progress_dialog, game_version, _7z_path, game_folder, plugin_path, install_tasks
)
)
# 启动线程
hash_thread.start()
# 显示对话框，阻塞直到对话框关闭
progress_dialog.exec()
# 如果用户取消了验证，停止线程并继续下一个任务
if hash_thread.isRunning():
hash_thread.terminate()
hash_thread.wait()
self.process_next_offline_install_task(install_tasks)
return
self.process_next_offline_install_task(install_tasks)
except Exception as e:
if debug_mode:
logger.error(f"DEBUG: 离线安装任务处理失败: {e}")
logger.error(f"DEBUG: 错误堆栈: {traceback.format_exc()}")
# 关闭安装进度窗口
self.main_window.close_hash_msg_box()
# 显示错误消息
msgbox_frame(

419
source/core/patch_detector.py
View File

@@ -5,13 +5,25 @@ import py7zr
import traceback
from utils.logger import setup_logger
from PySide6.QtWidgets import QMessageBox
from PySide6.QtCore import QTimer
from PySide6.QtCore import QTimer, QThread, Signal
from data.config import PLUGIN_HASH, APP_NAME
from workers.hash_thread import HashThread
# 初始化logger
logger = setup_logger("patch_detector")
class PatchCheckThread(QThread):
"""用于在后台线程中执行补丁检查的线程"""
finished = Signal(bool) # (is_installed)
def __init__(self, checker_func, *args):
super().__init__()
self.checker_func = checker_func
self.args = args
def run(self):
result = self.checker_func(*self.args)
self.finished.emit(result)
class PatchDetector:
"""补丁检测与校验模块，用于统一处理在线和离线模式下的补丁检测和校验"""
@@ -25,10 +37,9 @@ class PatchDetector:
self.app_name = main_window.APP_NAME if hasattr(main_window, 'APP_NAME') else ""
self.game_info = {}
self.plugin_hash = {}
# 从配置中加载游戏信息和补丁哈希值
self._load_game_info()
self.patch_check_thread = None
def _load_game_info(self):
"""从配置中加载游戏信息和补丁哈希值"""
try:
@@ -37,7 +48,7 @@ class PatchDetector:
self.plugin_hash = PLUGIN_HASH
except ImportError:
logger.error("无法加载游戏信息或补丁哈希值配置")
def _is_debug_mode(self):
"""检查是否处于调试模式
@@ -47,27 +58,25 @@ class PatchDetector:
try:
if hasattr(self.main_window, 'debug_manager') and self.main_window.debug_manager:
if hasattr(self.main_window.debug_manager, '_is_debug_mode'):
# 尝试直接从debug_manager获取状态
return self.main_window.debug_manager._is_debug_mode()
elif hasattr(self.main_window, 'config'):
# 如果debug_manager还没准备好，尝试从配置中获取
return self.main_window.config.get('debug_mode', False)
# 如果以上都不可行，返回False
return False
except Exception:
# 捕获任何异常，默认返回False
return False
def check_patch_installed(self, game_dir, game_version):
"""检查游戏是否已安装补丁
Args:
game_dir: 游戏目录路径
game_version: 游戏版本
Returns:
bool: 如果已安装补丁或有被禁用的补丁文件返回True，否则返回False
"""
def check_patch_installed_async(self, game_dir, game_version, callback):
"""异步检查游戏是否已安装补丁"""
def on_finished(is_installed):
callback(is_installed)
self.patch_check_thread = None
self.patch_check_thread = PatchCheckThread(self._check_patch_installed_sync, game_dir, game_version)
self.patch_check_thread.finished.connect(on_finished)
self.patch_check_thread.start()
def _check_patch_installed_sync(self, game_dir, game_version):
"""同步检查游戏是否已安装补丁（在工作线程中运行）"""
debug_mode = self._is_debug_mode()
if debug_mode:
@@ -78,152 +87,34 @@ class PatchDetector:
logger.debug(f"DEBUG: {game_version} 不在支持的游戏列表中，跳过检查")
return False
# 获取可能的补丁文件路径
install_path_base = os.path.basename(self.game_info[game_version]["install_path"])
patch_file_path = os.path.join(game_dir, install_path_base)
if debug_mode:
logger.debug(f"DEBUG: 基础补丁文件路径: {patch_file_path}")
# 尝试查找补丁文件，支持不同大小写
patch_files_to_check = [
patch_file_path,
patch_file_path.lower(),
patch_file_path.upper(),
patch_file_path.replace("_", ""),
patch_file_path.replace("_", "-"),
]
if debug_mode:
logger.debug(f"DEBUG: 将检查以下补丁文件路径: {patch_files_to_check}")
# 查找补丁文件
for patch_path in patch_files_to_check:
if os.path.exists(patch_path):
if debug_mode:
logger.debug(f"DEBUG: 找到补丁文件: {patch_path}")
logger.debug(f"DEBUG: {game_version} 已安装补丁")
return True
# 检查是否存在被禁用的补丁文件（带.fain后缀）
disabled_path = f"{patch_path}.fain"
if os.path.exists(disabled_path):
if debug_mode:
logger.debug(f"DEBUG: 找到被禁用的补丁文件: {disabled_path}")
logger.debug(f"DEBUG: {game_version} 已安装补丁（但被禁用）")
return True
if debug_mode:
logger.debug(f"DEBUG: 未找到补丁文件，继续检查补丁文件夹")
# 检查是否有补丁文件夹
patch_folders_to_check = [
os.path.join(game_dir, "patch"),
os.path.join(game_dir, "Patch"),
os.path.join(game_dir, "PATCH"),
]
if debug_mode:
logger.debug(f"DEBUG: 将检查以下补丁文件夹: {patch_folders_to_check}")
for patch_folder in patch_folders_to_check:
if os.path.exists(patch_folder):
if debug_mode:
logger.debug(f"DEBUG: 找到补丁文件夹: {patch_folder}")
logger.debug(f"DEBUG: {game_version} 已安装补丁")
return True
if debug_mode:
logger.debug(f"DEBUG: 未找到补丁文件夹，继续检查game/patch文件夹")
# 检查game/patch文件夹
game_folders = ["game", "Game", "GAME"]
patch_folders = ["patch", "Patch", "PATCH"]
if debug_mode:
logger.debug(f"DEBUG: 将检查以下game/patch组合: {[(g, p) for g in game_folders for p in patch_folders]}")
for game_folder in game_folders:
for patch_folder in patch_folders:
game_patch_folder = os.path.join(game_dir, game_folder, patch_folder)
if os.path.exists(game_patch_folder):
if debug_mode:
logger.debug(f"DEBUG: 找到game/patch文件夹: {game_patch_folder}")
logger.debug(f"DEBUG: {game_version} 已安装补丁")
return True
if debug_mode:
logger.debug(f"DEBUG: 未找到game/patch文件夹，继续检查配置文件和脚本文件")
# 检查配置文件
config_files = ["config.json", "Config.json", "CONFIG.JSON"]
script_files = ["scripts.json", "Scripts.json", "SCRIPTS.JSON"]
if debug_mode:
logger.debug(f"DEBUG: 将在game文件夹中检查以下配置文件: {config_files}")
logger.debug(f"DEBUG: 将在game文件夹中检查以下脚本文件: {script_files}")
for game_folder in game_folders:
game_path = os.path.join(game_dir, game_folder)
if os.path.exists(game_path):
# 检查配置文件
for config_file in config_files:
config_path = os.path.join(game_path, config_file)
if os.path.exists(config_path):
if debug_mode:
logger.debug(f"DEBUG: 找到配置文件: {config_path}")
logger.debug(f"DEBUG: {game_version} 已安装补丁")
return True
# 检查脚本文件
for script_file in script_files:
script_path = os.path.join(game_path, script_file)
if os.path.exists(script_path):
if debug_mode:
logger.debug(f"DEBUG: 找到脚本文件: {script_path}")
logger.debug(f"DEBUG: {game_version} 已安装补丁")
return True
# 没有找到补丁文件或文件夹
if debug_mode:
logger.debug(f"DEBUG: {game_version} 在 {game_dir} 中没有安装补丁")
# 检查补丁文件和禁用的补丁文件
if os.path.exists(patch_file_path) or os.path.exists(f"{patch_file_path}.fain"):
return True
return False
def check_patch_installed(self, game_dir, game_version):
"""检查游戏是否已安装补丁（此方法可能导致阻塞，推荐使用异步版本）"""
return self._check_patch_installed_sync(game_dir, game_version)
def check_patch_disabled(self, game_dir, game_version):
"""检查游戏的补丁是否已被禁用
Args:
game_dir: 游戏目录路径
game_version: 游戏版本
Returns:
bool: 如果补丁被禁用返回True，否则返回False
str: 禁用的补丁文件路径，如果没有禁用返回None
"""
"""检查游戏的补丁是否已被禁用"""
debug_mode = self._is_debug_mode()
if game_version not in self.game_info:
return False, None
# 获取可能的补丁文件路径
install_path_base = os.path.basename(self.game_info[game_version]["install_path"])
patch_file_path = os.path.join(game_dir, install_path_base)
disabled_path = f"{patch_file_path}.fain"
# 检查是否存在禁用的补丁文件（.fain后缀）
disabled_patch_files = [
f"{patch_file_path}.fain",
f"{patch_file_path.lower()}.fain",
f"{patch_file_path.upper()}.fain",
f"{patch_file_path.replace('_', '')}.fain",
f"{patch_file_path.replace('_', '-')}.fain",
]
# 检查是否有禁用的补丁文件
for disabled_path in disabled_patch_files:
if os.path.exists(disabled_path):
if debug_mode:
logger.debug(f"找到禁用的补丁文件: {disabled_path}")
return True, disabled_path
if os.path.exists(disabled_path):
if debug_mode:
logger.debug(f"找到禁用的补丁文件: {disabled_path}")
return True, disabled_path
if debug_mode:
logger.debug(f"{game_version} 在 {game_dir} 的补丁未被禁用")
@@ -231,14 +122,7 @@ class PatchDetector:
return False, None
def detect_installable_games(self, game_dirs):
"""检测可安装补丁的游戏
Args:
game_dirs: 游戏版本到游戏目录的映射字典
Returns:
tuple: (已安装补丁的游戏列表, 可安装补丁的游戏列表, 禁用补丁的游戏列表)
"""
"""检测可安装补丁的游戏"""
debug_mode = self._is_debug_mode()
if debug_mode:
@@ -249,21 +133,16 @@ class PatchDetector:
disabled_patch_games = []
for game_version, game_dir in game_dirs.items():
# 首先通过文件检查确认补丁是否已安装
is_patch_installed = self.check_patch_installed(game_dir, game_version)
# 同时考虑哈希检查结果
hash_check_passed = self.main_window.installed_status.get(game_version, False)
# 如果补丁文件存在或哈希检查通过，认为已安装
if is_patch_installed or hash_check_passed:
if debug_mode:
logger.info(f"DEBUG: {game_version} 已安装补丁，不需要再次安装")
logger.info(f"DEBUG: 文件检查结果: {is_patch_installed}, 哈希检查结果: {hash_check_passed}")
already_installed_games.append(game_version)
# 更新安装状态
self.main_window.installed_status[game_version] = True
else:
# 检查是否存在被禁用的补丁
is_disabled, disabled_path = self.check_patch_disabled(game_dir, game_version)
if is_disabled:
if debug_mode:
@@ -283,19 +162,7 @@ class PatchDetector:
return already_installed_games, installable_games, disabled_patch_games
def verify_patch_hash(self, game_version, file_path):
"""验证补丁文件的哈希值
Args:
game_version: 游戏版本名称
file_path: 补丁压缩包文件路径
Returns:
bool: 哈希值是否匹配
"""
# 获取预期的哈希值
expected_hash = None
# 直接使用完整游戏名称作为键
"""验证补丁文件的哈希值"""
expected_hash = self.plugin_hash.get(game_version, "")
if not expected_hash:
@@ -310,139 +177,79 @@ class PatchDetector:
logger.debug(f"DEBUG: 预期哈希值: {expected_hash}")
try:
# 检查文件是否存在
if not os.path.exists(file_path):
if debug_mode:
logger.warning(f"DEBUG: 补丁文件不存在: {file_path}")
if not os.path.exists(file_path) or os.path.getsize(file_path) == 0:
return False
# 检查文件大小
file_size = os.path.getsize(file_path)
if debug_mode:
logger.debug(f"DEBUG: 补丁文件大小: {file_size} 字节")
if file_size == 0:
if debug_mode:
logger.warning(f"DEBUG: 补丁文件大小为0，无效文件")
return False
# 创建临时目录用于解压文件
with tempfile.TemporaryDirectory() as temp_dir:
if debug_mode:
logger.debug(f"DEBUG: 创建临时目录: {temp_dir}")
# 解压补丁文件
try:
if debug_mode:
logger.debug(f"DEBUG: 开始解压文件: {file_path}")
with py7zr.SevenZipFile(file_path, mode="r") as archive:
# 获取压缩包内文件列表
file_list = archive.getnames()
if debug_mode:
logger.debug(f"DEBUG: 压缩包内文件列表: {file_list}")
# 解压所有文件
archive.extractall(path=temp_dir)
if debug_mode:
logger.debug(f"DEBUG: 解压完成")
# 列出解压后的文件
extracted_files = []
for root, dirs, files in os.walk(temp_dir):
for file in files:
extracted_files.append(os.path.join(root, file))
logger.debug(f"DEBUG: 解压后的文件列表: {extracted_files}")
except Exception as e:
if debug_mode:
logger.error(f"DEBUG: 解压补丁文件失败: {e}")
logger.error(f"DEBUG: 错误类型: {type(e).__name__}")
logger.error(f"DEBUG: 错误堆栈: {traceback.format_exc()}")
return False
# 获取补丁文件路径
patch_file = None
if "Vol.1" in game_version:
patch_file = os.path.join(temp_dir, "vol.1", "adultsonly.xp3")
elif "Vol.2" in game_version:
patch_file = os.path.join(temp_dir, "vol.2", "adultsonly.xp3")
elif "Vol.3" in game_version:
patch_file = os.path.join(temp_dir, "vol.3", "update00.int")
elif "Vol.4" in game_version:
patch_file = os.path.join(temp_dir, "vol.4", "vol4adult.xp3")
elif "After" in game_version:
patch_file = os.path.join(temp_dir, "after", "afteradult.xp3")
patch_file = self._find_patch_file_in_temp_dir(temp_dir, game_version)
if not patch_file or not os.path.exists(patch_file):
if debug_mode:
logger.warning(f"DEBUG: 未找到解压后的补丁文件: {patch_file}")
# 尝试查找可能的替代文件
alternative_files = []
for root, dirs, files in os.walk(temp_dir):
for file in files:
if file.endswith('.xp3') or file.endswith('.int'):
alternative_files.append(os.path.join(root, file))
if alternative_files:
logger.debug(f"DEBUG: 找到可能的替代文件: {alternative_files}")
# 检查解压目录结构
logger.debug(f"DEBUG: 检查解压目录结构:")
for root, dirs, files in os.walk(temp_dir):
logger.debug(f"DEBUG: 目录: {root}")
logger.debug(f"DEBUG: 子目录: {dirs}")
logger.debug(f"DEBUG: 文件: {files}")
logger.warning(f"DEBUG: 未找到解压后的补丁文件")
return False
if debug_mode:
logger.debug(f"DEBUG: 找到解压后的补丁文件: {patch_file}")
# 计算补丁文件哈希值
try:
with open(patch_file, "rb") as f:
file_hash = hashlib.sha256(f.read()).hexdigest()
# 比较哈希值
result = file_hash.lower() == expected_hash.lower()
if debug_mode:
logger.debug(f"DEBUG: 补丁文件 {patch_file} 哈希值验证: {'成功' if result else '失败'}")
logger.debug(f"DEBUG: 预期哈希值: {expected_hash}")
logger.debug(f"DEBUG: 实际哈希值: {file_hash}")
return result
except Exception as e:
if debug_mode:
logger.error(f"DEBUG: 计算补丁文件哈希值失败: {e}")
logger.error(f"DEBUG: 错误类型: {type(e).__name__}")
return False
except Exception as e:
if debug_mode:
logger.error(f"DEBUG: 验证补丁哈希值失败: {e}")
logger.error(f"DEBUG: 错误类型: {type(e).__name__}")
logger.error(f"DEBUG: 错误堆栈: {traceback.format_exc()}")
return False
return False
def _find_patch_file_in_temp_dir(self, temp_dir, game_version):
"""在临时目录中查找解压后的补丁文件"""
game_patch_map = {
"Vol.1": os.path.join("vol.1", "adultsonly.xp3"),
"Vol.2": os.path.join("vol.2", "adultsonly.xp3"),
"Vol.3": os.path.join("vol.3", "update00.int"),
"Vol.4": os.path.join("vol.4", "vol4adult.xp3"),
"After": os.path.join("after", "afteradult.xp3"),
}
for version_keyword, relative_path in game_patch_map.items():
if version_keyword in game_version:
return os.path.join(temp_dir, relative_path)
# 如果没有找到，则进行通用搜索
for root, dirs, files in os.walk(temp_dir):
for file in files:
if file.endswith('.xp3') or file.endswith('.int'):
return os.path.join(root, file)
return None
def create_hash_thread(self, mode, install_paths):
"""创建哈希检查线程
Args:
mode: 检查模式，"pre"或"after"
install_paths: 安装路径字典
Returns:
HashThread: 哈希检查线程实例
"""
from workers.hash_thread import HashThread
return HashThread(mode, install_paths, PLUGIN_HASH, self.main_window.installed_status, self.main_window)
def after_hash_compare(self):
"""进行安装后哈希比较"""
# 禁用窗口已在安装流程开始时完成
# 检查是否处于离线模式
is_offline = False
if hasattr(self.main_window, 'offline_mode_manager'):
is_offline = self.main_window.offline_mode_manager.is_in_offline_mode()
is_offline = self.main_window.offline_mode_manager.is_in_offline_mode()
self.main_window.close_hash_msg_box()
self.main_window.hash_msg_box = self.main_window.hash_manager.hash_pop_window(check_type="after", is_offline=is_offline)
install_paths = self.main_window.download_manager.get_install_paths()
@@ -452,73 +259,34 @@ class PatchDetector:
self.main_window.hash_thread.start()
def on_after_hash_finished(self, result):
"""哈希比较完成后的处理
Args:
result: 哈希比较结果
"""
# 确保哈希检查窗口关闭，无论是否还在显示
if self.main_window.hash_msg_box:
try:
if self.main_window.hash_msg_box.isVisible():
self.main_window.hash_msg_box.close()
else:
# 如果窗口已经不可见但没有关闭，也要尝试关闭
self.main_window.hash_msg_box.close()
except:
pass # 忽略任何关闭窗口时的错误
self.main_window.hash_msg_box = None
self.main_window.close_hash_msg_box()
if not result["passed"]:
# 启用窗口以显示错误消息
self.main_window.setEnabled(True)
game = result.get("game", "未知游戏")
message = result.get("message", "发生未知错误。")
msg_box = QMessageBox.critical(
self.main_window,
f"文件校验失败 - {APP_NAME}",
message,
QMessageBox.StandardButton.Ok,
)
QMessageBox.critical(self.main_window, f"文件校验失败 - {APP_NAME}", message)
# 恢复窗口状态
self.main_window.setEnabled(True)
self.main_window.ui.start_install_text.setText("开始安装")
# 添加短暂延迟确保UI更新
QTimer.singleShot(100, self.main_window.show_result)
def on_offline_pre_hash_finished(self, updated_status, game_dirs):
"""离线模式下的哈希预检查完成处理
Args:
updated_status: 更新后的安装状态
game_dirs: 识别到的游戏目录
"""
# 更新安装状态
self.main_window.installed_status = updated_status
# 关闭哈希检查窗口
if self.main_window.hash_msg_box and self.main_window.hash_msg_box.isVisible():
self.main_window.hash_msg_box.accept()
self.main_window.hash_msg_box = None
# 重新启用主窗口
self.main_window.setEnabled(True)
# 使用patch_detector检测可安装的游戏
already_installed_games, installable_games, disabled_patch_games = self.detect_installable_games(game_dirs)
debug_mode = self._is_debug_mode()
status_message = ""
if already_installed_games:
status_message += f"已安装补丁的游戏：\n{chr(10).join(already_installed_games)}\n\n"
# 处理禁用补丁的情况
if disabled_patch_games:
# 构建提示消息
disabled_msg = f"检测到以下游戏的补丁已被禁用：\n{chr(10).join(disabled_patch_games)}\n\n是否要启用这些补丁？"
from PySide6 import QtWidgets
@@ -530,23 +298,15 @@ class PatchDetector:
)
if reply == QtWidgets.QMessageBox.StandardButton.Yes:
# 用户选择启用补丁
if debug_mode:
logger.debug(f"DEBUG: 用户选择启用被禁用的补丁")
# 为每个禁用的游戏创建目录映射
disabled_game_dirs = {game: game_dirs[game] for game in disabled_patch_games}
# 批量启用补丁
success_count, fail_count, results = self.main_window.patch_manager.batch_toggle_patches(
disabled_game_dirs,
operation="enable"
)
# 显示启用结果
self.main_window.patch_manager.show_toggle_result(success_count, fail_count, results)
# 更新安装状态
for game_version in disabled_patch_games:
self.main_window.installed_status[game_version] = True
if game_version in installable_games:
@@ -554,61 +314,47 @@ class PatchDetector:
if game_version not in already_installed_games:
already_installed_games.append(game_version)
else:
if debug_mode:
logger.info(f"DEBUG: 用户选择不启用被禁用的补丁，这些游戏将被添加到可安装列表")
# 用户选择不启用，将这些游戏视为可以安装补丁
installable_games.extend(disabled_patch_games)
# 更新status_message
if disabled_patch_games:
status_message += f"禁用补丁的游戏：\n{chr(10).join(disabled_patch_games)}\n\n"
if not installable_games:
# 没有可安装的游戏，显示信息并重置UI
if already_installed_games:
# 有已安装的游戏，显示已安装信息
QMessageBox.information(
self.main_window,
f"信息 - {APP_NAME}",
f"\n所有游戏已安装补丁，无需重复安装。\n\n{status_message}",
QMessageBox.StandardButton.Ok,
)
else:
# 没有已安装的游戏，可能是未检测到游戏
QMessageBox.warning(
self.main_window,
f"警告 - {APP_NAME}",
"\n未检测到任何需要安装补丁的游戏。\n\n请确保游戏文件夹位于选择的目录中。\n",
QMessageBox.StandardButton.Ok,
)
self.main_window.ui.start_install_text.setText("开始安装")
return
# 显示游戏选择对话框
from PySide6 import QtWidgets
dialog = QtWidgets.QDialog(self.main_window)
dialog.setWindowTitle(f"选择要安装的游戏 - {APP_NAME}")
dialog.setMinimumWidth(300)
layout = QtWidgets.QVBoxLayout()
# 添加说明标签
label = QtWidgets.QLabel("请选择要安装补丁的游戏：")
layout.addWidget(label)
# 添加游戏列表
list_widget = QtWidgets.QListWidget()
list_widget.setSelectionMode(QtWidgets.QAbstractItemView.SelectionMode.MultiSelection)
for game in installable_games:
item = QtWidgets.QListWidgetItem(game)
list_widget.addItem(item)
item.setSelected(True) # 默认全选
item.setSelected(True)
layout.addWidget(list_widget)
# 添加按钮
button_box = QtWidgets.QDialogButtonBox(
QtWidgets.QDialogButtonBox.StandardButton.Ok |
QtWidgets.QDialogButtonBox.StandardButton.Cancel
@@ -619,18 +365,11 @@ class PatchDetector:
dialog.setLayout(layout)
# 显示对话框
result = dialog.exec()
if result != QtWidgets.QDialog.DialogCode.Accepted or list_widget.selectedItems() == []:
if result != QtWidgets.QDialog.DialogCode.Accepted or not list_widget.selectedItems():
self.main_window.ui.start_install_text.setText("开始安装")
return
# 获取用户选择的游戏
selected_games = [item.text() for item in list_widget.selectedItems()]
# 开始安装
if debug_mode:
logger.debug(f"DEBUG: 用户选择了以下游戏进行安装: {selected_games}")
# 调用离线模式管理器安装补丁
self.main_window.offline_mode_manager.install_offline_patches(selected_games)

4
source/core/ui_manager.py
View File

@@ -953,6 +953,10 @@ class UIManager:
self.main_window.config["offline_mode"] = False
self.main_window.save_config(self.main_window.config)
# 重新获取云端配置
if hasattr(self.main_window, 'fetch_cloud_config'):
self.main_window.fetch_cloud_config()
# 如果当前版本过低，设置版本警告标志
if hasattr(self.main_window, 'last_error_message') and self.main_window.last_error_message == "update_required":
# 设置版本警告标志

73
source/handlers/patch_toggle_handler.py
View File

@@ -3,7 +3,7 @@ from PySide6.QtWidgets import (
QDialog, QVBoxLayout, QLabel, QListWidget, QPushButton, QHBoxLayout,
QAbstractItemView, QRadioButton, QButtonGroup, QFileDialog, QMessageBox
)
from PySide6.QtCore import QObject
from PySide6.QtCore import QObject, Signal, QThread
from PySide6.QtGui import QFont
from utils import msgbox_frame
from utils.logger import setup_logger
@@ -11,6 +11,20 @@ from utils.logger import setup_logger
# 初始化logger
logger = setup_logger("patch_toggle_handler")
class PatchToggleThread(QThread):
"""在后台线程中处理补丁切换逻辑"""
finished = Signal(object)
def __init__(self, handler, selected_folder):
super().__init__()
self.handler = handler
self.selected_folder = selected_folder
def run(self):
# 在后台线程中执行耗时操作
game_dirs = self.handler.game_detector.identify_game_directories_improved(self.selected_folder)
self.finished.emit(game_dirs)
class PatchToggleHandler(QObject):
"""
处理补丁启用/禁用功能的类
@@ -28,32 +42,59 @@ class PatchToggleHandler(QObject):
self.game_detector = main_window.game_detector
self.patch_manager = main_window.patch_manager
self.app_name = main_window.patch_manager.app_name
self.toggle_thread = None
def handle_toggle_patch_button_click(self):
"""
处理禁/启用补丁按钮点击事件
打开文件选择对话框选择游戏目录，然后禁用或启用对应游戏的补丁
"""
# 获取游戏目录
debug_mode = self.debug_manager._is_debug_mode()
selected_folder = QFileDialog.getExistingDirectory(self.main_window, "选择游戏上级目录", "")
# 提示用户选择目录
file_dialog_info = "选择游戏上级目录" if debug_mode else "选择游戏目录"
selected_folder = QFileDialog.getExistingDirectory(self.main_window, file_dialog_info, "")
if not selected_folder:
return
self.main_window.show_loading_dialog("正在识别游戏目录并检查补丁状态...")
if not selected_folder or selected_folder == "":
return # 用户取消了选择
self.toggle_thread = PatchToggleThread(self, selected_folder)
self.toggle_thread.finished.connect(self.on_game_detection_finished)
self.toggle_thread.start()
def on_game_detection_finished(self, game_dirs):
"""游戏识别完成后的回调"""
self.main_window.hide_loading_dialog()
if not game_dirs:
QMessageBox.information(
self.main_window,
f"提示 - {self.app_name}",
"\n未在选择的目录中找到任何支持的游戏。\n",
)
return
games_with_patch = {}
for game_version, game_dir in game_dirs.items():
if self.patch_manager.check_patch_installed(game_dir, game_version):
is_disabled, _ = self.patch_manager.check_patch_disabled(game_dir, game_version)
status = "已禁用" if is_disabled else "已启用"
games_with_patch[game_version] = {"dir": game_dir, "status": status}
if debug_mode:
logger.debug(f"DEBUG: 禁/启用功能 - 用户选择了目录: {selected_folder}")
if not games_with_patch:
QMessageBox.information(
self.main_window,
f"提示 - {self.app_name}",
"\n目录中未找到已安装补丁的游戏。\n",
)
return
selected_games, operation = self._show_multi_game_dialog(games_with_patch)
# 首先尝试将选择的目录视为上级目录，使用增强的目录识别功能
game_dirs = self.game_detector.identify_game_directories_improved(selected_folder)
if not selected_games:
return
if game_dirs and len(game_dirs) > 0:
self._handle_multiple_games(game_dirs, debug_mode)
else:
self._handle_single_game(selected_folder, debug_mode)
selected_game_dirs = {game: games_with_patch[game]["dir"] for game in selected_games if game in games_with_patch}
self._execute_batch_toggle(selected_game_dirs, operation)
def _handle_multiple_games(self, game_dirs, debug_mode):
"""

77
source/handlers/uninstall_handler.py
View File

@@ -3,7 +3,7 @@ from PySide6.QtWidgets import (
QDialog, QVBoxLayout, QLabel, QListWidget, QPushButton, QHBoxLayout,
QAbstractItemView, QFileDialog, QMessageBox
)
from PySide6.QtCore import QObject
from PySide6.QtCore import QObject, Signal, QThread
from PySide6.QtGui import QFont
from utils import msgbox_frame
from utils.logger import setup_logger
@@ -11,6 +11,20 @@ from utils.logger import setup_logger
# 初始化logger
logger = setup_logger("uninstall_handler")
class UninstallThread(QThread):
"""在后台线程中处理卸载逻辑"""
finished = Signal(object)
def __init__(self, handler, selected_folder):
super().__init__()
self.handler = handler
self.selected_folder = selected_folder
def run(self):
# 在后台线程中执行耗时操作
game_dirs = self.handler.game_detector.identify_game_directories_improved(self.selected_folder)
self.finished.emit(game_dirs)
class UninstallHandler(QObject):
"""
处理补丁卸载功能的类
@@ -28,6 +42,7 @@ class UninstallHandler(QObject):
self.game_detector = main_window.game_detector
self.patch_manager = main_window.patch_manager
self.app_name = main_window.patch_manager.app_name
self.uninstall_thread = None
# 记录初始化日志
debug_mode = self.debug_manager._is_debug_mode() if hasattr(self.debug_manager, '_is_debug_mode') else False
@@ -60,16 +75,58 @@ class UninstallHandler(QObject):
if debug_mode:
logger.debug(f"DEBUG: 卸载功能 - 用户选择了目录: {selected_folder}")
# 首先尝试将选择的目录视为上级目录，使用增强的目录识别功能
logger.info("尝试识别游戏目录")
game_dirs = self.game_detector.identify_game_directories_improved(selected_folder)
self.main_window.show_loading_dialog("正在识别游戏目录...")
if game_dirs and len(game_dirs) > 0:
logger.info(f"在上级目录中找到游戏: {list(game_dirs.keys())}")
self._handle_multiple_games(game_dirs, debug_mode)
else:
logger.info("未在上级目录找到游戏，尝试将选择的目录作为单个游戏目录处理")
self._handle_single_game(selected_folder, debug_mode)
self.uninstall_thread = UninstallThread(self, selected_folder)
self.uninstall_thread.finished.connect(self.on_game_detection_finished)
self.uninstall_thread.start()
def on_game_detection_finished(self, game_dirs):
"""游戏识别完成后的回调"""
self.main_window.hide_loading_dialog()
if not game_dirs:
QMessageBox.information(
self.main_window,
f"提示 - {self.app_name}",
"\n未在选择的目录中找到任何支持的游戏。\n",
)
return
games_with_patch = {}
for game_version, game_dir in game_dirs.items():
if self.patch_manager.check_patch_installed(game_dir, game_version):
games_with_patch[game_version] = game_dir
if not games_with_patch:
QMessageBox.information(
self.main_window,
f"提示 - {self.app_name}",
"\n目录中未找到已安装补丁的游戏。\n",
)
return
selected_games = self._show_game_selection_dialog(games_with_patch)
if not selected_games:
return
selected_game_dirs = {game: games_with_patch[game] for game in selected_games if game in games_with_patch}
game_list = '\n'.join(selected_games)
reply = QMessageBox.question(
self.main_window,
f"确认卸载 - {self.app_name}",
f"\n确定要卸载以下游戏的补丁吗？\n\n{game_list}\n",
QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No,
QMessageBox.StandardButton.No
)
if reply == QMessageBox.StandardButton.No:
return
success_count, fail_count, results = self.patch_manager.batch_uninstall_patches(selected_game_dirs)
self.patch_manager.show_uninstall_result(success_count, fail_count, results)
def _handle_multiple_games(self, game_dirs, debug_mode):
"""

172
source/main_window.py
View File

@@ -7,7 +7,7 @@ import webbrowser
from PySide6 import QtWidgets
from PySide6.QtCore import QTimer, Qt, QPoint, QRect, QSize
from PySide6.QtWidgets import QMainWindow, QMessageBox, QGraphicsOpacityEffect, QGraphicsColorizeEffect
from PySide6.QtWidgets import QMainWindow, QMessageBox, QGraphicsOpacityEffect, QGraphicsColorizeEffect, QDialog, QVBoxLayout, QProgressBar, QLabel
from PySide6.QtGui import QPalette, QColor, QPainterPath, QRegion, QFont
from PySide6.QtGui import QAction # Added for menu actions
from PySide6.QtWidgets import QDialog, QVBoxLayout, QHBoxLayout, QProgressBar, QLabel # Added for progress window
@@ -32,6 +32,7 @@ from core import (
from core.ipv6_manager import IPv6Manager
from handlers import PatchToggleHandler, UninstallHandler
from utils.logger import setup_logger
from core.patch_detector import PatchDetector
# 初始化logger
logger = setup_logger("main_window")
@@ -62,6 +63,11 @@ class MainWindow(QMainWindow):
self.hash_manager = HashManager(BLOCK_SIZE)
self.admin_privileges = AdminPrivileges()
# 初始化哈希校验窗口引用
self.hash_msg_box = None
self.loading_dialog = None
self.patch_detector = PatchDetector(self)
# 初始化各种管理器 - 调整初始化顺序，避免循环依赖
# 1. 首先创建必要的基础管理器
self.animator = MultiStageAnimations(self.ui, self)
@@ -108,7 +114,6 @@ class MainWindow(QMainWindow):
self.config_valid = False # 添加配置有效标志
self.patch_manager.initialize_status()
self.installed_status = self.patch_manager.get_status() # 获取初始化后的状态
self.hash_msg_box = None
self.last_error_message = "" # 添加错误信息记录
self.version_warning = False # 添加版本警告标志
self.install_button_enabled = True # 默认启用安装按钮
@@ -367,6 +372,34 @@ class MainWindow(QMainWindow):
return progress_window
def create_extraction_progress_window(self):
"""创建解压进度窗口
Returns:
QDialog: 解压进度窗口实例
"""
progress_window = QDialog(self)
progress_window.setWindowTitle(f"解压进度 - {APP_NAME}")
progress_window.setFixedSize(400, 150)
layout = QVBoxLayout()
# 添加进度条
progress_bar = QProgressBar()
progress_bar.setRange(0, 100)
progress_bar.setValue(0)
layout.addWidget(progress_bar)
# 添加标签
status_label = QLabel("准备解压...")
layout.addWidget(status_label)
progress_window.setLayout(layout)
progress_window.progress_bar = progress_bar
progress_window.status_label = status_label
return progress_window
def create_extraction_thread(self, _7z_path, game_folder, plugin_path, game_version, extracted_path=None):
"""创建解压线程
@@ -382,6 +415,26 @@ class MainWindow(QMainWindow):
"""
return ExtractionThread(_7z_path, game_folder, plugin_path, game_version, self, extracted_path)
def create_hash_thread(self, mode, install_paths, plugin_hash=None, installed_status=None):
"""创建哈希校验线程
Args:
mode: 校验模式，"pre"或"after"
install_paths: 安装路径字典
plugin_hash: 插件哈希值字典，如果为None则使用self.plugin_hash
installed_status: 安装状态字典，如果为None则使用self.installed_status
Returns:
HashThread: 哈希校验线程实例
"""
if plugin_hash is None:
plugin_hash = PLUGIN_HASH
if installed_status is None:
installed_status = self.installed_status
return HashThread(mode, install_paths, plugin_hash, installed_status, self)
def show_result(self):
"""显示安装结果，调用patch_manager的show_result方法"""
self.patch_manager.show_result()
@@ -516,57 +569,20 @@ class MainWindow(QMainWindow):
# 重试获取配置
self.fetch_cloud_config()
else:
# 按钮处于"开始安装"状态，正常执行安装流程
# 检查是否处于离线模式
if is_offline_mode:
# 如果是离线模式，使用离线安装流程
# 先选择游戏目录
if self.offline_mode_manager.is_in_offline_mode():
self.selected_folder = QtWidgets.QFileDialog.getExistingDirectory(
self, f"选择游戏所在【上级目录】 {APP_NAME}"
)
if not self.selected_folder:
QtWidgets.QMessageBox.warning(
self, f"通知 - {APP_NAME}", "\n未选择任何目录,请重新选择\n"
)
QtWidgets.QMessageBox.warning(self, f"通知 - {APP_NAME}", "\n未选择任何目录,请重新选择\n")
return
# 保存选择的目录到下载管理器
self.download_manager.selected_folder = self.selected_folder
# 设置按钮状态
self.ui.start_install_text.setText("正在安装")
self.show_loading_dialog("正在识别游戏目录...")
self.setEnabled(False)
# 清除游戏检测器的目录缓存
if hasattr(self, 'game_detector') and hasattr(self.game_detector, 'clear_directory_cache'):
self.game_detector.clear_directory_cache()
# 识别游戏目录
game_dirs = self.game_detector.identify_game_directories_improved(self.selected_folder)
if not game_dirs:
self.last_error_message = "directory_not_found"
QtWidgets.QMessageBox.warning(
self,
f"目录错误 - {APP_NAME}",
"\n未能识别到任何游戏目录。\n\n请确认您选择的是游戏的上级目录，并且该目录中包含NEKOPARA系列游戏文件夹。\n"
)
self.setEnabled(True)
self.ui.start_install_text.setText("开始安装")
return
# 显示文件检验窗口
self.hash_msg_box = self.hash_manager.hash_pop_window(check_type="pre", is_offline=True)
# 获取安装路径
install_paths = self.download_manager.get_install_paths()
# 创建并启动哈希线程进行预检查
self.hash_thread = self.patch_detector.create_hash_thread("pre", install_paths)
self.hash_thread.pre_finished.connect(
lambda updated_status: self.patch_detector.on_offline_pre_hash_finished(updated_status, game_dirs)
)
self.hash_thread.start()
# 异步识别游戏目录
self.game_detector.identify_game_directories_async(self.selected_folder, self.on_game_directories_identified)
else:
# 在线模式下，检查版本是否过低
if hasattr(self, 'version_warning') and self.version_warning:
@@ -581,6 +597,61 @@ class MainWindow(QMainWindow):
# 版本正常，使用原有的下载流程
self.download_manager.file_dialog()
def show_loading_dialog(self, message):
"""显示加载对话框"""
if not self.loading_dialog:
self.loading_dialog = QDialog(self)
self.loading_dialog.setWindowTitle(f"请稍候 - {APP_NAME}")
self.loading_dialog.setFixedSize(300, 100)
self.loading_dialog.setModal(True)
layout = QVBoxLayout()
self.loading_label = QLabel(message)
self.loading_label.setAlignment(Qt.AlignCenter)
layout.addWidget(self.loading_label)
self.loading_dialog.setLayout(layout)
else:
self.loading_label.setText(message)
self.loading_dialog.show()
QtWidgets.QApplication.processEvents()
def hide_loading_dialog(self):
"""隐藏加载对话框"""
if self.loading_dialog:
self.loading_dialog.hide()
self.loading_dialog = None
def on_game_directories_identified(self, game_dirs):
"""游戏目录识别完成后的回调"""
self.hide_loading_dialog()
if not game_dirs:
self.setEnabled(True)
self.ui.start_install_text.setText("开始安装")
QtWidgets.QMessageBox.warning(
self,
f"目录错误 - {APP_NAME}",
"\n未能识别到任何游戏目录。\n\n请确认您选择的是游戏的上级目录，并且该目录中包含NEKOPARA系列游戏文件夹。\n"
)
return
self.show_loading_dialog("正在检查补丁状态...")
install_paths = self.download_manager.get_install_paths()
# 使用异步方式进行哈希预检查
hash_thread = self.patch_detector.create_hash_thread("pre", install_paths)
hash_thread.pre_finished.connect(
lambda updated_status: self.on_pre_hash_finished(updated_status, game_dirs)
)
hash_thread.start()
def on_pre_hash_finished(self, updated_status, game_dirs):
"""哈希预检查完成后的回调"""
self.hide_loading_dialog()
self.setEnabled(True)
self.patch_detector.on_offline_pre_hash_finished(updated_status, game_dirs)
# 移除on_offline_pre_hash_finished方法
def check_and_set_offline_mode(self):
@@ -643,6 +714,17 @@ class MainWindow(QMainWindow):
logger.error(f"错误: 检查离线模式时发生异常: {e}")
return False
def close_hash_msg_box(self):
"""关闭哈希校验窗口，确保在创建新窗口前关闭旧窗口"""
if hasattr(self, 'hash_msg_box') and self.hash_msg_box:
try:
if self.hash_msg_box.isVisible():
self.hash_msg_box.close()
QtWidgets.QApplication.processEvents() # 确保UI更新，窗口真正关闭
except Exception as e:
logger.error(f"关闭哈希校验窗口时发生错误: {e}")
self.hash_msg_box = None

23
source/utils/helpers.py
View File

@@ -201,12 +201,14 @@ class HashManager:
logger.error(f"Error calculating hash for {file_path}: {e}")
return results
def hash_pop_window(self, check_type="default", is_offline=False):
def hash_pop_window(self, check_type="default", is_offline=False, auto_close=False, close_delay=500):
"""显示文件检验窗口
Args:
check_type: 检查类型，可以是 'pre'(预检查), 'after'(后检查), 'extraction'(解压后检查), 'offline_extraction'(离线解压), 'offline_verify'(离线验证)
is_offline: 是否处于离线模式
auto_close: 是否自动关闭窗口
close_delay: 自动关闭延迟（毫秒）
Returns:
QMessageBox: 消息框实例
@@ -223,6 +225,8 @@ class HashManager:
message = "\n正在验证本地补丁压缩文件完整性...\n"
elif check_type == "offline_extraction":
message = "\n正在解压安装补丁文件...\n"
elif check_type == "offline_installation":
message = "\n正在安装补丁文件...\n"
else:
message = "\n正在处理离线补丁文件...\n"
else:
@@ -233,10 +237,27 @@ class HashManager:
message = "\n正在检验本地文件完整性...\n"
elif check_type == "extraction":
message = "\n正在验证下载的解压文件完整性...\n"
elif check_type == "post":
message = "\n正在检验补丁文件完整性...\n"
# 创建新的消息框
msg_box = msgbox_frame(f"通知 - {APP_NAME}", message)
# 使用open()而不是exec()，避免阻塞UI线程
msg_box.open()
# 处理事件循环，确保窗口显示
QtWidgets.QApplication.processEvents()
# 如果设置了自动关闭，添加定时器
if auto_close:
timer = QtCore.QTimer()
timer.setSingleShot(True)
timer.timeout.connect(msg_box.close)
timer.start(close_delay)
# 保存定时器引用，防止被垃圾回收
msg_box.close_timer = timer
return msg_box
def cfg_pre_hash_compare(self, install_paths, plugin_hash, installed_status):

114
source/workers/extraction_thread.py
View File

@@ -1,11 +1,12 @@
import os
import shutil
import py7zr
from PySide6.QtCore import QThread, Signal
from PySide6.QtCore import QThread, Signal, QCoreApplication
from data.config import PLUGIN, GAME_INFO
class ExtractionThread(QThread):
finished = Signal(bool, str, str) # success, error_message, game_version
progress = Signal(int, str) # 添加进度信号，传递进度百分比和状态信息
def __init__(self, _7z_path, game_folder, plugin_path, game_version, parent=None, extracted_path=None):
super().__init__(parent)
@@ -17,11 +18,27 @@ class ExtractionThread(QThread):
def run(self):
try:
# 确保游戏目录存在
os.makedirs(self.game_folder, exist_ok=True)
# 发送初始进度信号
self.progress.emit(0, f"开始处理 {self.game_version} 的补丁文件...")
# 确保UI更新
QCoreApplication.processEvents()
# 如果提供了已解压文件路径，直接使用它
if self.extracted_path and os.path.exists(self.extracted_path):
# 发送进度信号
self.progress.emit(20, f"正在复制 {self.game_version} 的补丁文件...")
QCoreApplication.processEvents()
# 直接复制已解压的文件到游戏目录
os.makedirs(self.game_folder, exist_ok=True)
shutil.copy(self.extracted_path, self.game_folder)
target_file = os.path.join(self.game_folder, os.path.basename(self.plugin_path))
shutil.copy(self.extracted_path, target_file)
# 发送进度信号
self.progress.emit(60, f"正在完成 {self.game_version} 的补丁安装...")
QCoreApplication.processEvents()
# 对于NEKOPARA After，还需要复制签名文件
if self.game_version == "NEKOPARA After":
@@ -37,18 +54,93 @@ class ExtractionThread(QThread):
# 如果签名文件不存在，则使用原始路径
sig_path = os.path.join(PLUGIN, GAME_INFO[self.game_version]["sig_path"])
shutil.copy(sig_path, self.game_folder)
# 发送完成进度信号
self.progress.emit(100, f"{self.game_version} 补丁文件处理完成")
QCoreApplication.processEvents()
else:
# 如果没有提供已解压文件路径，执行正常的解压流程
# 如果没有提供已解压文件路径，直接解压到游戏目录
# 获取目标文件名
target_filename = os.path.basename(self.plugin_path)
target_path = os.path.join(self.game_folder, target_filename)
# 发送进度信号
self.progress.emit(10, f"正在打开 {self.game_version} 的补丁压缩包...")
QCoreApplication.processEvents()
# 使用7z解压
with py7zr.SevenZipFile(self._7z_path, mode="r") as archive:
archive.extractall(path=PLUGIN)
# 获取压缩包内的文件列表
file_list = archive.getnames()
# 发送进度信号
self.progress.emit(20, f"正在分析 {self.game_version} 的补丁文件...")
QCoreApplication.processEvents()
# 解析压缩包内的文件结构
target_file_in_archive = None
for file_path in file_list:
if target_filename in file_path:
target_file_in_archive = file_path
break
if not target_file_in_archive:
raise FileNotFoundError(f"在压缩包中未找到目标文件 {target_filename}")
# 发送进度信号
self.progress.emit(30, f"正在解压 {self.game_version} 的补丁文件...")
QCoreApplication.processEvents()
# 创建一个临时目录用于解压单个文件
import tempfile
with tempfile.TemporaryDirectory() as temp_dir:
# 解压特定文件到临时目录
archive.extract(path=temp_dir, targets=[target_file_in_archive])
# 发送进度信号
self.progress.emit(60, f"正在复制 {self.game_version} 的补丁文件...")
QCoreApplication.processEvents()
# 找到解压后的文件
extracted_file_path = os.path.join(temp_dir, target_file_in_archive)
# 复制到目标位置
shutil.copy2(extracted_file_path, target_path)
# 发送进度信号
self.progress.emit(80, f"正在完成 {self.game_version} 的补丁安装...")
QCoreApplication.processEvents()
# 对于NEKOPARA After，还需要复制签名文件
if self.game_version == "NEKOPARA After":
sig_filename = f"{target_filename}.sig"
sig_file_in_archive = None
# 查找签名文件
for file_path in file_list:
if sig_filename in file_path:
sig_file_in_archive = file_path
break
if sig_file_in_archive:
# 解压签名文件
archive.extract(path=temp_dir, targets=[sig_file_in_archive])
extracted_sig_path = os.path.join(temp_dir, sig_file_in_archive)
sig_target = os.path.join(self.game_folder, sig_filename)
shutil.copy2(extracted_sig_path, sig_target)
else:
# 如果签名文件不存在，则使用原始路径
sig_path = os.path.join(PLUGIN, GAME_INFO[self.game_version]["sig_path"])
if os.path.exists(sig_path):
sig_target = os.path.join(self.game_folder, sig_filename)
shutil.copy2(sig_path, sig_target)
os.makedirs(self.game_folder, exist_ok=True)
shutil.copy(self.plugin_path, self.game_folder)
if self.game_version == "NEKOPARA After":
sig_path = os.path.join(PLUGIN, GAME_INFO[self.game_version]["sig_path"])
shutil.copy(sig_path, self.game_folder)
# 发送完成进度信号
self.progress.emit(100, f"{self.game_version} 补丁文件解压完成")
QCoreApplication.processEvents()
self.finished.emit(True, "", self.game_version)
except (py7zr.Bad7zFile, FileNotFoundError, Exception) as e:
self.progress.emit(100, f"处理 {self.game_version} 的补丁文件失败")
QCoreApplication.processEvents()
self.finished.emit(False, f"\n文件操作失败，请重试\n\n【错误信息】：{e}\n", self.game_version)

266
source/workers/hash_thread.py
View File

@@ -4,6 +4,7 @@ import py7zr
import tempfile
import traceback
from PySide6.QtCore import QThread, Signal
from PySide6.QtWidgets import QApplication
from utils.logger import setup_logger
# 初始化logger
@@ -166,6 +167,9 @@ class OfflineHashVerifyThread(QThread):
if not expected_hash:
logger.warning(f"DEBUG: 未找到 {self.game_version} 的预期哈希值")
# 确保发送100%进度信号，以便UI更新
self.progress.emit(100)
QApplication.processEvents()
self.finished.emit(False, f"未找到 {self.game_version} 的预期哈希值", "")
return
@@ -179,6 +183,9 @@ class OfflineHashVerifyThread(QThread):
if not os.path.exists(self.file_path):
if debug_mode:
logger.warning(f"DEBUG: 补丁文件不存在: {self.file_path}")
# 确保发送100%进度信号，以便UI更新
self.progress.emit(100)
QApplication.processEvents()
self.finished.emit(False, f"补丁文件不存在: {self.file_path}", "")
return
@@ -190,6 +197,9 @@ class OfflineHashVerifyThread(QThread):
if file_size == 0:
if debug_mode:
logger.warning(f"DEBUG: 补丁文件大小为0，无效文件")
# 确保发送100%进度信号，以便UI更新
self.progress.emit(100)
QApplication.processEvents()
self.finished.emit(False, "补丁文件大小为0，无效文件", "")
return
@@ -206,112 +216,192 @@ class OfflineHashVerifyThread(QThread):
if debug_mode:
logger.debug(f"DEBUG: 开始解压文件: {self.file_path}")
# 确定目标文件名
target_filename = None
if "Vol.1" in self.game_version:
target_filename = "adultsonly.xp3"
elif "Vol.2" in self.game_version:
target_filename = "adultsonly.xp3"
elif "Vol.3" in self.game_version:
target_filename = "update00.int"
elif "Vol.4" in self.game_version:
target_filename = "vol4adult.xp3"
elif "After" in self.game_version:
target_filename = "afteradult.xp3"
if not target_filename:
if debug_mode:
logger.warning(f"DEBUG: 未知的游戏版本: {self.game_version}")
self.progress.emit(100)
QApplication.processEvents()
self.finished.emit(False, f"未知的游戏版本: {self.game_version}", "")
return
with py7zr.SevenZipFile(self.file_path, mode="r") as archive:
# 获取压缩包内文件列表
file_list = archive.getnames()
if debug_mode:
logger.debug(f"DEBUG: 压缩包内文件列表: {file_list}")
# 解压所有文件
archive.extractall(path=temp_dir)
# 查找目标文件
target_file_in_archive = None
for file_path in file_list:
if target_filename in file_path:
target_file_in_archive = file_path
break
if not target_file_in_archive:
if debug_mode:
logger.warning(f"DEBUG: 在压缩包中未找到目标文件: {target_filename}")
# 尝试查找可能的替代文件
alternative_files = []
for file_path in file_list:
if file_path.endswith('.xp3') or file_path.endswith('.int'):
alternative_files.append(file_path)
if alternative_files:
if debug_mode:
logger.debug(f"DEBUG: 找到可能的替代文件: {alternative_files}")
target_file_in_archive = alternative_files[0]
else:
# 如果找不到任何替代文件，解压全部文件
if debug_mode:
logger.debug(f"DEBUG: 未找到任何替代文件，解压全部文件")
archive.extractall(path=temp_dir)
# 尝试在解压后的目录中查找目标文件
for root, dirs, files in os.walk(temp_dir):
for file in files:
if file.endswith('.xp3') or file.endswith('.int'):
patch_file = os.path.join(root, file)
if debug_mode:
logger.debug(f"DEBUG: 找到可能的补丁文件: {patch_file}")
break
if patch_file:
break
if not patch_file:
if debug_mode:
logger.warning(f"DEBUG: 未找到解压后的补丁文件")
self.progress.emit(100)
QApplication.processEvents()
self.finished.emit(False, "未找到解压后的补丁文件", "")
return
else:
# 只解压目标文件
if debug_mode:
logger.debug(f"DEBUG: 解压目标文件: {target_file_in_archive}")
archive.extract(path=temp_dir, targets=[target_file_in_archive])
patch_file = os.path.join(temp_dir, target_file_in_archive)
# 发送进度信号 - 50%
self.progress.emit(50)
# 如果还没有设置patch_file，尝试查找
if not 'patch_file' in locals():
if "Vol.1" in self.game_version:
patch_file = os.path.join(temp_dir, "vol.1", "adultsonly.xp3")
elif "Vol.2" in self.game_version:
patch_file = os.path.join(temp_dir, "vol.2", "adultsonly.xp3")
elif "Vol.3" in self.game_version:
patch_file = os.path.join(temp_dir, "vol.3", "update00.int")
elif "Vol.4" in self.game_version:
patch_file = os.path.join(temp_dir, "vol.4", "vol4adult.xp3")
elif "After" in self.game_version:
patch_file = os.path.join(temp_dir, "after", "afteradult.xp3")
if not os.path.exists(patch_file):
if debug_mode:
logger.warning(f"DEBUG: 未找到解压后的补丁文件: {patch_file}")
# 尝试查找可能的替代文件
alternative_files = []
for root, dirs, files in os.walk(temp_dir):
for file in files:
if file.endswith('.xp3') or file.endswith('.int'):
alternative_files.append(os.path.join(root, file))
if alternative_files:
logger.debug(f"DEBUG: 找到可能的替代文件: {alternative_files}")
patch_file = alternative_files[0]
else:
# 检查解压目录结构
logger.debug(f"DEBUG: 检查解压目录结构:")
for root, dirs, files in os.walk(temp_dir):
logger.debug(f"DEBUG: 目录: {root}")
logger.debug(f"DEBUG: 子目录: {dirs}")
logger.debug(f"DEBUG: 文件: {files}")
if not os.path.exists(patch_file):
# 确保发送100%进度信号，以便UI更新
self.progress.emit(100)
QApplication.processEvents()
self.finished.emit(False, f"未找到解压后的补丁文件", "")
return
# 发送进度信号 - 70%
self.progress.emit(70)
if debug_mode:
logger.debug(f"DEBUG: 解压完成")
# 列出解压后的文件
extracted_files = []
for root, dirs, files in os.walk(temp_dir):
for file in files:
extracted_files.append(os.path.join(root, file))
logger.debug(f"DEBUG: 解压后的文件列表: {extracted_files}")
logger.debug(f"DEBUG: 找到解压后的补丁文件: {patch_file}")
# 计算补丁文件哈希值
try:
# 读取文件内容并计算哈希值，同时更新进度
file_size = os.path.getsize(patch_file)
chunk_size = 1024 * 1024 # 1MB
hash_obj = hashlib.sha256()
with open(patch_file, "rb") as f:
bytes_read = 0
while chunk := f.read(chunk_size):
hash_obj.update(chunk)
bytes_read += len(chunk)
# 计算进度 (70-95%)
progress = 70 + int(25 * bytes_read / file_size)
self.progress.emit(min(95, progress))
file_hash = hash_obj.hexdigest()
# 比较哈希值
result = file_hash.lower() == expected_hash.lower()
# 发送进度信号 - 100%
self.progress.emit(100)
# 确保UI更新
QApplication.processEvents()
if debug_mode:
logger.debug(f"DEBUG: 补丁文件 {patch_file} 哈希值验证: {'成功' if result else '失败'}")
logger.debug(f"DEBUG: 预期哈希值: {expected_hash}")
logger.debug(f"DEBUG: 实际哈希值: {file_hash}")
# 将验证结果和解压后的文件路径传递回去
# 注意：由于使用了临时目录，此路径在函数返回后将不再有效
# 但这里返回的路径只是用于标识验证成功，实际安装时会重新解压
self.finished.emit(result, "" if result else "补丁文件哈希验证失败，文件可能已损坏或被篡改", patch_file if result else "")
except Exception as e:
if debug_mode:
logger.error(f"DEBUG: 计算补丁文件哈希值失败: {e}")
logger.error(f"DEBUG: 错误类型: {type(e).__name__}")
# 确保发送100%进度信号，以便UI更新
self.progress.emit(100)
QApplication.processEvents()
self.finished.emit(False, f"计算补丁文件哈希值失败: {str(e)}", "")
except Exception as e:
if debug_mode:
logger.error(f"DEBUG: 解压补丁文件失败: {e}")
logger.error(f"DEBUG: 错误类型: {type(e).__name__}")
logger.error(f"DEBUG: 错误堆栈: {traceback.format_exc()}")
# 确保发送100%进度信号，以便UI更新
self.progress.emit(100)
QApplication.processEvents()
self.finished.emit(False, f"解压补丁文件失败: {str(e)}", "")
return
# 发送进度信号 - 50%
self.progress.emit(50)
# 获取补丁文件路径
patch_file = None
if "Vol.1" in self.game_version:
patch_file = os.path.join(temp_dir, "vol.1", "adultsonly.xp3")
elif "Vol.2" in self.game_version:
patch_file = os.path.join(temp_dir, "vol.2", "adultsonly.xp3")
elif "Vol.3" in self.game_version:
patch_file = os.path.join(temp_dir, "vol.3", "update00.int")
elif "Vol.4" in self.game_version:
patch_file = os.path.join(temp_dir, "vol.4", "vol4adult.xp3")
elif "After" in self.game_version:
patch_file = os.path.join(temp_dir, "after", "afteradult.xp3")
if not patch_file or not os.path.exists(patch_file):
if debug_mode:
logger.warning(f"DEBUG: 未找到解压后的补丁文件: {patch_file}")
# 尝试查找可能的替代文件
alternative_files = []
for root, dirs, files in os.walk(temp_dir):
for file in files:
if file.endswith('.xp3') or file.endswith('.int'):
alternative_files.append(os.path.join(root, file))
if alternative_files:
logger.debug(f"DEBUG: 找到可能的替代文件: {alternative_files}")
# 检查解压目录结构
logger.debug(f"DEBUG: 检查解压目录结构:")
for root, dirs, files in os.walk(temp_dir):
logger.debug(f"DEBUG: 目录: {root}")
logger.debug(f"DEBUG: 子目录: {dirs}")
logger.debug(f"DEBUG: 文件: {files}")
self.finished.emit(False, f"未找到解压后的补丁文件", "")
return
# 发送进度信号 - 70%
self.progress.emit(70)
if debug_mode:
logger.debug(f"DEBUG: 找到解压后的补丁文件: {patch_file}")
# 计算补丁文件哈希值
try:
# 读取文件内容并计算哈希值，同时更新进度
file_size = os.path.getsize(patch_file)
chunk_size = 1024 * 1024 # 1MB
hash_obj = hashlib.sha256()
with open(patch_file, "rb") as f:
bytes_read = 0
while chunk := f.read(chunk_size):
hash_obj.update(chunk)
bytes_read += len(chunk)
# 计算进度 (70-95%)
progress = 70 + int(25 * bytes_read / file_size)
self.progress.emit(min(95, progress))
file_hash = hash_obj.hexdigest()
# 比较哈希值
result = file_hash.lower() == expected_hash.lower()
# 发送进度信号 - 100%
self.progress.emit(100)
if debug_mode:
logger.debug(f"DEBUG: 补丁文件 {patch_file} 哈希值验证: {'成功' if result else '失败'}")
logger.debug(f"DEBUG: 预期哈希值: {expected_hash}")
logger.debug(f"DEBUG: 实际哈希值: {file_hash}")
self.finished.emit(result, "" if result else "补丁文件哈希验证失败，文件可能已损坏或被篡改", patch_file)
except Exception as e:
if debug_mode:
logger.error(f"DEBUG: 计算补丁文件哈希值失败: {e}")
logger.error(f"DEBUG: 错误类型: {type(e).__name__}")
self.finished.emit(False, f"计算补丁文件哈希值失败: {str(e)}", "")
except Exception as e:
if debug_mode:
logger.error(f"DEBUG: 验证补丁哈希值失败: {e}")
logger.error(f"DEBUG: 错误类型: {type(e).__name__}")
logger.error(f"DEBUG: 错误堆栈: {traceback.format_exc()}")
# 确保发送100%进度信号，以便UI更新
self.progress.emit(100)
QApplication.processEvents()
self.finished.emit(False, f"验证补丁哈希值失败: {str(e)}", "")