225 lines
9.3 KiB
Python
225 lines
9.3 KiB
Python
import os
|
|
import shutil
|
|
import json
|
|
import zipfile
|
|
from datetime import datetime
|
|
|
|
class CCXManagerNode:
|
|
def __init__(self, config_filename="config.json"):
|
|
self.config_path = os.path.join(os.path.dirname(__file__), config_filename)
|
|
self.config = self.load_config()
|
|
self.status = "未运行"
|
|
|
|
def load_config(self):
|
|
"""加载配置文件，如果不存在则创建默认配置"""
|
|
default_config = {
|
|
"source_path": "",
|
|
"target_path": "",
|
|
"ccx_filename": "",
|
|
"auto_run_on_restart": False,
|
|
"last_run_time": ""
|
|
}
|
|
try:
|
|
if os.path.exists(self.config_path):
|
|
with open(self.config_path, 'r', encoding='utf-8') as f:
|
|
return json.load(f)
|
|
else:
|
|
with open(self.config_path, 'w', encoding='utf-8') as f:
|
|
json.dump(default_config, f, indent=4, ensure_ascii=False)
|
|
return default_config
|
|
except Exception as e:
|
|
self.status = f"加载配置失败: {str(e)}"
|
|
print(f"[CCXManager] 加载配置失败: {str(e)}")
|
|
return default_config
|
|
|
|
def save_config(self):
|
|
"""保存当前配置到文件"""
|
|
try:
|
|
self.config["last_run_time"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
|
with open(self.config_path, 'w', encoding='utf-8') as f:
|
|
json.dump(self.config, f, indent=4, ensure_ascii=False)
|
|
self.status = f"配置已保存: {self.config['last_run_time']}"
|
|
print(f"[CCXManager] 配置已保存: {self.config['last_run_time']}")
|
|
except Exception as e:
|
|
self.status = f"保存配置失败: {str(e)}"
|
|
print(f"[CCXManager] 保存配置失败: {str(e)}")
|
|
|
|
def clear_directory(self, dir_path):
|
|
"""清空目标目录"""
|
|
if not os.path.exists(dir_path):
|
|
os.makedirs(dir_path)
|
|
print(f"[CCXManager] 创建目标目录: {dir_path}")
|
|
return True
|
|
try:
|
|
for item in os.listdir(dir_path):
|
|
item_path = os.path.join(dir_path, item)
|
|
if os.path.isfile(item_path):
|
|
os.remove(item_path)
|
|
elif os.path.isdir(item_path):
|
|
shutil.rmtree(item_path)
|
|
print(f"[CCXManager] 已清空目录: {dir_path}")
|
|
return True
|
|
except Exception as e:
|
|
self.status = f"清空目录失败: {str(e)}"
|
|
print(f"[CCXManager] 清空目录失败: {str(e)}")
|
|
return False
|
|
|
|
def extract_ccx(self, ccx_path, target_dir):
|
|
"""解压CCX文件到目标目录"""
|
|
try:
|
|
with zipfile.ZipFile(ccx_path, 'r') as zip_ref:
|
|
zip_ref.extractall(target_dir)
|
|
print(f"[CCXManager] CCX文件解压成功: {ccx_path}")
|
|
return True
|
|
except Exception as e:
|
|
self.status = f"解压失败: {str(e)}"
|
|
print(f"[CCXManager] 解压失败: {str(e)}")
|
|
return False
|
|
|
|
def run(self, source_path=None, target_path=None, ccx_filename=None, auto_run_on_restart=None):
|
|
"""执行CCX文件复制和解压流程"""
|
|
if source_path is None: source_path = self.config.get("source_path")
|
|
if target_path is None: target_path = self.config.get("target_path")
|
|
if ccx_filename is None: ccx_filename = self.config.get("ccx_filename")
|
|
if auto_run_on_restart is not None:
|
|
self.config["auto_run_on_restart"] = auto_run_on_restart
|
|
|
|
# 将路径转换为绝对路径
|
|
if source_path: source_path = os.path.abspath(source_path)
|
|
if target_path: target_path = os.path.abspath(target_path)
|
|
|
|
# 更新配置中的绝对路径
|
|
self.config["source_path"] = source_path
|
|
self.config["target_path"] = target_path
|
|
self.config["ccx_filename"] = ccx_filename
|
|
|
|
if not all([source_path, target_path, ccx_filename]):
|
|
self.status = "错误: 路径或文件名不能为空"
|
|
print(f"[CCXManager] {self.status}")
|
|
return self.status
|
|
|
|
ccx_source_path = os.path.join(source_path, ccx_filename)
|
|
if not os.path.exists(ccx_source_path):
|
|
self.status = f"错误: CCX文件不存在 - {ccx_source_path}"
|
|
print(f"[CCXManager] {self.status}")
|
|
return self.status
|
|
|
|
if not self.clear_directory(target_path):
|
|
return self.status
|
|
|
|
try:
|
|
ccx_target_path = os.path.join(target_path, ccx_filename)
|
|
shutil.copy2(ccx_source_path, ccx_target_path)
|
|
self.status = f"CCX文件已复制: {ccx_target_path}"
|
|
print(f"[CCXManager] {self.status}")
|
|
except Exception as e:
|
|
self.status = f"复制文件失败: {str(e)}"
|
|
print(f"[CCXManager] {self.status}")
|
|
return self.status
|
|
|
|
if self.extract_ccx(ccx_target_path, target_path):
|
|
self.status = f"成功: CCX文件已解压到 {target_path}"
|
|
print(f"[CCXManager] CCX文件已经全部更新")
|
|
self.save_config()
|
|
else:
|
|
self.status = f"失败: {self.status}"
|
|
|
|
return self.status
|
|
|
|
def auto_run(self):
|
|
"""重启后自动运行"""
|
|
if self.config.get("auto_run_on_restart", False):
|
|
print(f"[CCXManager] 检测到自动运行配置，正在执行CCX更新...")
|
|
if all([self.config.get("source_path"), self.config.get("target_path"), self.config.get("ccx_filename")]):
|
|
self.run()
|
|
else:
|
|
print(f"[CCXManager] 自动运行失败: 配置不完整")
|
|
else:
|
|
print(f"[CCXManager] 自动运行已禁用")
|
|
|
|
class CCXManager:
|
|
@classmethod
|
|
def INPUT_TYPES(s):
|
|
# 加载配置文件获取默认值
|
|
config_path = os.path.join(os.path.dirname(__file__), "config.json")
|
|
config = {}
|
|
if os.path.exists(config_path):
|
|
with open(config_path, 'r', encoding='utf-8') as f:
|
|
config = json.load(f)
|
|
|
|
return {
|
|
"required": {
|
|
"source_path": ("STRING", {"default": config.get("source_path", ""), "placeholder": "CCX文件所在路径"}),
|
|
"target_path": ("STRING", {"default": config.get("target_path", ""), "placeholder": "目标解压路径"}),
|
|
"ccx_filename": ("STRING", {"default": config.get("ccx_filename", "sd-ppp_PS.ccx"), "placeholder": "CCX文件名(含后缀)"}),
|
|
"auto_run_on_restart": (["enable", "disable"], {"default": "enable" if config.get("auto_run_on_restart", False) else "disable"}),
|
|
},
|
|
}
|
|
|
|
RETURN_TYPES = ("STRING",)
|
|
RETURN_NAMES = ("运行状态",)
|
|
FUNCTION = "process_ccx"
|
|
CATEGORY = "SD-PPP插件更新"
|
|
TITLE = "SDPPP插件自动更新助手1.0"
|
|
|
|
def __init__(self):
|
|
self.manager = CCXManagerNode()
|
|
|
|
def process_ccx(self, source_path, target_path, ccx_filename, auto_run_on_restart):
|
|
auto_run = auto_run_on_restart == "enable"
|
|
status = self.manager.run(source_path, target_path, ccx_filename, auto_run)
|
|
return (status,)
|
|
|
|
class CCXManagerCopy:
|
|
@classmethod
|
|
def INPUT_TYPES(s):
|
|
# 加载配置文件获取默认值
|
|
config_path = os.path.join(os.path.dirname(__file__), "config_copy.json")
|
|
config = {}
|
|
if os.path.exists(config_path):
|
|
with open(config_path, 'r', encoding='utf-8') as f:
|
|
config = json.load(f)
|
|
|
|
return {
|
|
"required": {
|
|
"source_path": ("STRING", {"default": config.get("source_path", ""), "placeholder": "CCX文件所在路径"}),
|
|
"target_path": ("STRING", {"default": config.get("target_path", ""), "placeholder": "目标解压路径"}),
|
|
"ccx_filename": ("STRING", {"default": config.get("ccx_filename", "sd-ppp2_PS.ccx"), "placeholder": "CCX文件名(含后缀)"}),
|
|
"auto_run_on_restart": (["enable", "disable"], {"default": "enable" if config.get("auto_run_on_restart", False) else "disable"}),
|
|
},
|
|
}
|
|
|
|
RETURN_TYPES = ("STRING",)
|
|
RETURN_NAMES = ("运行状态",)
|
|
FUNCTION = "process_ccx"
|
|
CATEGORY = "SD-PPP插件更新"
|
|
TITLE = "SDPPP插件自动更新助手2.0"
|
|
|
|
def __init__(self):
|
|
self.manager = CCXManagerNode(config_filename="config_copy.json")
|
|
|
|
def process_ccx(self, source_path, target_path, ccx_filename, auto_run_on_restart):
|
|
auto_run = auto_run_on_restart == "enable"
|
|
status = self.manager.run(source_path, target_path, ccx_filename, auto_run)
|
|
return (status,)
|
|
|
|
NODE_CLASS_MAPPINGS = {
|
|
"CCXManager": CCXManager,
|
|
"CCXManagerCopy": CCXManagerCopy
|
|
}
|
|
|
|
NODE_DISPLAY_NAME_MAPPINGS = {
|
|
"CCXManager": "SDPPP插件自动更新助手1.0",
|
|
"CCXManagerCopy": "SDPPP插件自动更新助手2.0"
|
|
}
|
|
|
|
# 程序启动时自动检查并运行所有节点
|
|
print(f"[CCXManager] SDPPP插件自动更新助手加载")
|
|
|
|
# 主节点自动运行
|
|
main_manager = CCXManagerNode()
|
|
main_manager.auto_run()
|
|
|
|
# 分身节点自动运行
|
|
clone_manager = CCXManagerNode(config_filename="config_copy.json")
|
|
clone_manager.auto_run() |