Files
FRAISEMOE-Addons-Installer-…/source/core/offline_mode_manager.py

702 lines
29 KiB
Python
Raw Normal View History

import os
import hashlib
import shutil
import tempfile
import py7zr
import traceback
from PySide6.QtWidgets import QMessageBox
from data.config import PLUGIN, PLUGIN_HASH, GAME_INFO
from utils import msgbox_frame
from utils.logger import setup_logger
# 初始化logger
logger = setup_logger("offline_mode_manager")
class OfflineModeManager:
"""离线模式管理器,用于管理离线模式下的补丁安装和检测"""
def __init__(self, main_window):
"""初始化离线模式管理器
Args:
main_window: 主窗口实例用于访问UI和状态
"""
self.main_window = main_window
self.app_name = main_window.APP_NAME if hasattr(main_window, 'APP_NAME') else ""
self.offline_patches = {} # 存储离线补丁信息 {补丁名称: 文件路径}
self.is_offline_mode = False
def _is_debug_mode(self):
"""检查是否处于调试模式
Returns:
bool: 是否处于调试模式
"""
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 scan_for_offline_patches(self, directory=None):
"""扫描指定目录(默认为软件所在目录)查找离线补丁文件
Args:
directory: 要扫描的目录如果为None则使用软件所在目录
Returns:
dict: 找到的补丁文件 {补丁名称: 文件路径}
"""
if directory is None:
# 获取软件所在目录
directory = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
debug_mode = self._is_debug_mode()
if debug_mode:
logger.debug(f"DEBUG: 扫描离线补丁文件,目录: {directory}")
# 要查找的补丁文件名
patch_files = ["vol.1.7z", "vol.2.7z", "vol.3.7z", "vol.4.7z", "after.7z"]
found_patches = {}
# 扫描目录中的文件
for file in os.listdir(directory):
if file.lower() in patch_files:
file_path = os.path.join(directory, file)
if os.path.isfile(file_path):
patch_name = file.lower()
found_patches[patch_name] = file_path
if debug_mode:
logger.debug(f"DEBUG: 找到离线补丁文件: {patch_name} 路径: {file_path}")
self.offline_patches = found_patches
return found_patches
def has_offline_patches(self):
"""检查是否有可用的离线补丁文件
Returns:
bool: 是否有可用的离线补丁
"""
if not self.offline_patches:
self.scan_for_offline_patches()
return len(self.offline_patches) > 0
def set_offline_mode(self, enabled):
"""设置离线模式状态
Args:
enabled: 是否启用离线模式
Returns:
bool: 是否成功设置离线模式
"""
debug_mode = self._is_debug_mode()
if enabled:
# 检查是否有离线补丁文件
if not self.has_offline_patches() and not debug_mode:
msgbox_frame(
f"离线模式错误 - {self.app_name}",
"\n未找到任何离线补丁文件,无法启用离线模式。\n\n请将补丁文件放置在软件所在目录后再尝试。\n",
QMessageBox.StandardButton.Ok
).exec()
return False
if debug_mode:
logger.debug("DEBUG: 已启用离线模式(调试模式下允许强制启用)")
self.is_offline_mode = enabled
# 更新窗口标题
if hasattr(self.main_window, 'setWindowTitle'):
from data.config import APP_NAME, APP_VERSION
mode_indicator = "[离线模式]" if enabled else "[在线模式]"
self.main_window.setWindowTitle(f"{APP_NAME} v{APP_VERSION} {mode_indicator}")
# 同时更新UI中的标题标签
if hasattr(self.main_window, 'ui') and hasattr(self.main_window.ui, 'title_label'):
self.main_window.ui.title_label.setText(f"{APP_NAME} v{APP_VERSION} {mode_indicator}")
# 同步更新UI菜单中的模式选择状态
if hasattr(self.main_window, 'ui_manager'):
ui_manager = self.main_window.ui_manager
if hasattr(ui_manager, 'online_mode_action') and hasattr(ui_manager, 'offline_mode_action'):
ui_manager.online_mode_action.setChecked(not enabled)
ui_manager.offline_mode_action.setChecked(enabled)
if debug_mode:
logger.debug(f"DEBUG: 离线模式已{'启用' if enabled else '禁用'}")
return True
def get_offline_patch_path(self, game_version):
"""根据游戏版本获取对应的离线补丁文件路径
Args:
game_version: 游戏版本名称"NEKOPARA Vol.1"
Returns:
str: 离线补丁文件路径如果没有找到则返回None
"""
# 确保已扫描过补丁文件
if not self.offline_patches:
self.scan_for_offline_patches()
# 根据游戏版本获取对应的补丁文件名
patch_file = None
if "Vol.1" in game_version:
patch_file = "vol.1.7z"
elif "Vol.2" in game_version:
patch_file = "vol.2.7z"
elif "Vol.3" in game_version:
patch_file = "vol.3.7z"
elif "Vol.4" in game_version:
patch_file = "vol.4.7z"
elif "After" in game_version:
patch_file = "after.7z"
# 检查是否有对应的补丁文件
if patch_file and patch_file in self.offline_patches:
return self.offline_patches[patch_file]
return None
def prepare_offline_patch(self, game_version, target_path):
"""准备离线补丁文件,复制到缓存目录
Args:
game_version: 游戏版本名称
target_path: 目标路径通常是缓存目录中的路径
Returns:
bool: 是否成功准备补丁文件
"""
source_path = self.get_offline_patch_path(game_version)
if not source_path:
return False
debug_mode = self._is_debug_mode()
try:
# 确保目标目录存在
os.makedirs(os.path.dirname(target_path), exist_ok=True)
# 复制文件
shutil.copy2(source_path, target_path)
if debug_mode:
logger.debug(f"DEBUG: 已复制离线补丁文件 {source_path}{target_path}")
return True
except Exception as e:
if debug_mode:
logger.error(f"DEBUG: 复制离线补丁文件失败: {e}")
return False
def verify_patch_hash(self, game_version, file_path):
"""验证补丁文件的哈希值
Args:
game_version: 游戏版本名称
file_path: 补丁压缩包文件路径
Returns:
bool: 哈希值是否匹配
"""
# 获取预期的哈希值
expected_hash = None
if "Vol.1" in game_version:
expected_hash = PLUGIN_HASH.get("vol1", "")
elif "Vol.2" in game_version:
expected_hash = PLUGIN_HASH.get("vol2", "")
elif "Vol.3" in game_version:
expected_hash = PLUGIN_HASH.get("vol3", "")
elif "Vol.4" in game_version:
expected_hash = PLUGIN_HASH.get("vol4", "")
elif "After" in game_version:
expected_hash = PLUGIN_HASH.get("after", "")
if not expected_hash:
logger.warning(f"DEBUG: 未找到 {game_version} 的预期哈希值")
return False
debug_mode = self._is_debug_mode()
if debug_mode:
logger.debug(f"DEBUG: 开始验证离线补丁文件: {file_path}")
logger.debug(f"DEBUG: 游戏版本: {game_version}")
logger.debug(f"DEBUG: 预期哈希值: {expected_hash}")
try:
# 检查文件是否存在
if not os.path.exists(file_path):
if debug_mode:
logger.warning(f"DEBUG: 补丁文件不存在: {file_path}")
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")
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}")
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
def is_offline_mode_available(self):
"""检查是否可以使用离线模式
Returns:
bool: 是否可以使用离线模式
"""
# 在调试模式下始终允许离线模式
if self._is_debug_mode():
return True
# 检查是否有离线补丁文件
return self.has_offline_patches()
def is_in_offline_mode(self):
"""检查当前是否处于离线模式
Returns:
bool: 是否处于离线模式
"""
return self.is_offline_mode
def install_offline_patches(self, selected_games):
"""直接安装离线补丁,完全绕过下载模块
Args:
selected_games: 用户选择安装的游戏列表
Returns:
bool: 是否成功启动安装流程
"""
debug_mode = self._is_debug_mode()
if debug_mode:
logger.debug(f"DEBUG: 开始离线安装流程,选择的游戏: {selected_games}")
if not self.is_in_offline_mode():
if debug_mode:
logger.warning("DEBUG: 当前不是离线模式,无法使用离线安装")
return False
# 确保已扫描过补丁文件
if not self.offline_patches:
self.scan_for_offline_patches()
if not self.offline_patches:
if debug_mode:
logger.warning("DEBUG: 未找到任何离线补丁文件")
msgbox_frame(
f"离线安装错误 - {self.app_name}",
"\n未找到任何离线补丁文件,无法进行离线安装。\n\n请将补丁文件放置在软件所在目录后再尝试。\n",
QMessageBox.StandardButton.Ok
).exec()
return False
# 获取游戏目录
game_dirs = self.main_window.game_detector.identify_game_directories_improved(
self.main_window.download_manager.selected_folder
)
if not game_dirs:
if debug_mode:
logger.warning("DEBUG: 未识别到任何游戏目录")
return False
# 显示文件检验窗口
self.main_window.hash_msg_box = self.main_window.hash_manager.hash_pop_window(check_type="pre", is_offline=True)
# 获取安装路径
install_paths = self.main_window.download_manager.get_install_paths()
# 创建并启动哈希线程进行预检查
self.main_window.hash_thread = self.main_window.create_hash_thread("pre", install_paths)
self.main_window.hash_thread.pre_finished.connect(
lambda updated_status: self.on_offline_pre_hash_finished(updated_status, game_dirs, selected_games)
)
self.main_window.hash_thread.start()
return True
def on_offline_pre_hash_finished(self, updated_status, game_dirs, selected_games):
"""离线模式下的哈希预检查完成处理
Args:
updated_status: 更新后的安装状态
game_dirs: 识别到的游戏目录
selected_games: 用户选择安装的游戏列表
"""
debug_mode = self._is_debug_mode()
# 更新安装状态
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)
# 过滤出需要安装的游戏
installable_games = []
for game_version in selected_games:
if game_version in game_dirs and not self.main_window.installed_status.get(game_version, False):
# 检查是否有对应的离线补丁
if self.get_offline_patch_path(game_version):
installable_games.append(game_version)
elif debug_mode:
logger.warning(f"DEBUG: 未找到 {game_version} 的离线补丁文件,跳过")
if not installable_games:
if debug_mode:
logger.info("DEBUG: 没有需要安装的游戏或未找到对应的离线补丁")
msgbox_frame(
f"离线安装信息 - {self.app_name}",
"\n没有需要安装的游戏或未找到对应的离线补丁文件。\n",
QMessageBox.StandardButton.Ok
).exec()
self.main_window.ui.start_install_text.setText("开始安装")
return
# 开始安装流程
if debug_mode:
logger.info(f"DEBUG: 开始离线安装流程,安装游戏: {installable_games}")
# 创建安装任务列表
install_tasks = []
for game_version in installable_games:
# 获取离线补丁文件路径
patch_file = self.get_offline_patch_path(game_version)
if not patch_file:
continue
# 获取游戏目录
game_folder = game_dirs.get(game_version)
if not game_folder:
continue
# 获取目标路径
if "Vol.1" in game_version:
_7z_path = os.path.join(PLUGIN, "vol.1.7z")
plugin_path = os.path.join(PLUGIN, GAME_INFO[game_version]["plugin_path"])
elif "Vol.2" in game_version:
_7z_path = os.path.join(PLUGIN, "vol.2.7z")
plugin_path = os.path.join(PLUGIN, GAME_INFO[game_version]["plugin_path"])
elif "Vol.3" in game_version:
_7z_path = os.path.join(PLUGIN, "vol.3.7z")
plugin_path = os.path.join(PLUGIN, GAME_INFO[game_version]["plugin_path"])
elif "Vol.4" in game_version:
_7z_path = os.path.join(PLUGIN, "vol.4.7z")
plugin_path = os.path.join(PLUGIN, GAME_INFO[game_version]["plugin_path"])
elif "After" in game_version:
_7z_path = os.path.join(PLUGIN, "after.7z")
plugin_path = os.path.join(PLUGIN, GAME_INFO[game_version]["plugin_path"])
else:
continue
# 添加到安装任务列表
install_tasks.append((patch_file, game_folder, game_version, _7z_path, plugin_path))
# 开始执行第一个安装任务
if install_tasks:
self.process_next_offline_install_task(install_tasks)
else:
self.main_window.ui.start_install_text.setText("开始安装")
def process_next_offline_install_task(self, install_tasks):
"""处理下一个离线安装任务
Args:
install_tasks: 安装任务列表每个任务是一个元组 (patch_file, game_folder, game_version, _7z_path, plugin_path)
"""
debug_mode = self._is_debug_mode()
if not install_tasks:
# 所有任务完成,进行后检查
if debug_mode:
logger.info("DEBUG: 所有离线安装任务完成,进行后检查")
self.main_window.after_hash_compare()
return
# 获取下一个任务
patch_file, game_folder, game_version, _7z_path, plugin_path = install_tasks.pop(0)
if debug_mode:
logger.debug(f"DEBUG: 处理离线安装任务: {game_version}")
logger.debug(f"DEBUG: 补丁文件: {patch_file}")
logger.debug(f"DEBUG: 游戏目录: {game_folder}")
# 确保目标目录存在
os.makedirs(os.path.dirname(_7z_path), exist_ok=True)
try:
# 复制补丁文件到缓存目录
shutil.copy2(patch_file, _7z_path)
if debug_mode:
logger.debug(f"DEBUG: 已复制补丁文件到缓存目录: {_7z_path}")
logger.debug(f"DEBUG: 开始验证补丁文件哈希值")
# 获取预期的哈希值
expected_hash = None
if "Vol.1" in game_version:
expected_hash = PLUGIN_HASH.get("vol1", "")
elif "Vol.2" in game_version:
expected_hash = PLUGIN_HASH.get("vol2", "")
elif "Vol.3" in game_version:
expected_hash = PLUGIN_HASH.get("vol3", "")
elif "Vol.4" in game_version:
expected_hash = PLUGIN_HASH.get("vol4", "")
elif "After" in game_version:
expected_hash = PLUGIN_HASH.get("after", "")
if debug_mode and expected_hash:
logger.debug(f"DEBUG: 预期哈希值: {expected_hash}")
# 显示哈希验证窗口 - 使用离线特定消息
self.main_window.hash_msg_box = self.main_window.hash_manager.hash_pop_window(check_type="offline_verify", is_offline=True)
# 验证补丁文件哈希
hash_valid = self.verify_patch_hash(game_version, _7z_path)
# 关闭哈希验证窗口
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 hash_valid:
if debug_mode:
logger.info(f"DEBUG: 补丁文件哈希验证成功,开始解压")
# 显示解压窗口 - 使用离线特定消息
self.main_window.hash_msg_box = self.main_window.hash_manager.hash_pop_window(check_type="offline_extraction", is_offline=True)
try:
# 创建解压线程
extraction_thread = self.main_window.create_extraction_thread(
_7z_path, game_folder, plugin_path, game_version
)
# 正确连接信号
extraction_thread.finished.connect(
lambda success, error, game_ver: self.on_extraction_thread_finished(
success, error, game_ver, install_tasks
)
)
# 启动解压线程
extraction_thread.start()
except Exception as e:
if debug_mode:
logger.error(f"DEBUG: 创建或启动解压线程失败: {e}")
# 关闭解压窗口
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
# 显示错误消息
msgbox_frame(
f"解压错误 - {self.app_name}",
f"\n{game_version} 的解压过程中发生错误: {str(e)}\n\n跳过此游戏的安装。\n",
QMessageBox.StandardButton.Ok
).exec()
# 继续下一个任务
self.process_next_offline_install_task(install_tasks)
else:
if debug_mode:
logger.warning(f"DEBUG: 补丁文件哈希验证失败")
# 显示错误消息
msgbox_frame(
f"哈希验证失败 - {self.app_name}",
f"\n{game_version} 的补丁文件哈希验证失败,可能已损坏或被篡改。\n\n跳过此游戏的安装。\n",
QMessageBox.StandardButton.Ok
).exec()
# 继续下一个任务
self.process_next_offline_install_task(install_tasks)
except Exception as e:
if debug_mode:
logger.error(f"DEBUG: 离线安装任务处理失败: {e}")
# 显示错误消息
msgbox_frame(
f"安装错误 - {self.app_name}",
f"\n{game_version} 的安装过程中发生错误: {str(e)}\n\n跳过此游戏的安装。\n",
QMessageBox.StandardButton.Ok
).exec()
# 继续下一个任务
self.process_next_offline_install_task(install_tasks)
def on_extraction_thread_finished(self, success, error_message, game_version, remaining_tasks):
"""解压线程完成后的处理
Args:
success: 是否解压成功
error_message: 错误信息
game_version: 游戏版本
remaining_tasks: 剩余的安装任务列表
"""
debug_mode = self._is_debug_mode()
# 关闭解压窗口
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 debug_mode:
logger.debug(f"DEBUG: 离线解压完成,状态: {'成功' if success else '失败'}")
if not success:
logger.error(f"DEBUG: 错误信息: {error_message}")
if not success:
# 显示错误消息
msgbox_frame(
f"解压失败 - {self.app_name}",
f"\n{game_version} 的补丁解压失败。\n\n错误信息: {error_message}\n\n跳过此游戏的安装。\n",
QMessageBox.StandardButton.Ok
).exec()
# 更新安装状态
self.main_window.installed_status[game_version] = False
else:
# 更新安装状态
self.main_window.installed_status[game_version] = True
# 处理下一个任务
self.process_next_offline_install_task(remaining_tasks)
def on_offline_extraction_finished(self, remaining_tasks):
"""离线模式下的解压完成处理(旧方法,保留兼容性)
Args:
remaining_tasks: 剩余的安装任务列表
"""
debug_mode = self._is_debug_mode()
if debug_mode:
logger.debug("DEBUG: 离线解压完成,继续处理下一个任务")
# 处理下一个任务
self.process_next_offline_install_task(remaining_tasks)