feat: add comprehensive error handling to LoraInfo_UTK node - 增加网络连接失败、文件读取错误等异常处理,确保程序稳定运行

This commit is contained in:
Cyber Dick Lang
2025-07-15 18:49:43 +08:00
parent c701d628cf
commit 5a8ab327fd
2 changed files with 200 additions and 108 deletions
+165 -95
View File
@@ -41,11 +41,25 @@ def save_dict_to_json(data_dict, file_path):
def get_model_version_info(hash_value):
api_url = f"https://civitai.com/api/v1/model-versions/by-hash/{hash_value}"
response = requests.get(api_url)
try:
response = requests.get(api_url, timeout=10) # 设置10秒超时
if response.status_code == 200:
return response.json()
else:
if response.status_code == 200:
return response.json()
else:
print(f"[LoraInfo_UTK] CivitAI API返回错误状态码: {response.status_code}")
return {}
except requests.exceptions.ConnectionError:
print("[LoraInfo_UTK] 无法连接到CivitAI服务器,请检查网络连接")
return {}
except requests.exceptions.Timeout:
print("[LoraInfo_UTK] 连接CivitAI服务器超时,请稍后重试")
return {}
except requests.exceptions.RequestException as e:
print(f"[LoraInfo_UTK] 请求CivitAI API时发生错误: {e}")
return {}
except Exception as e:
print(f"[LoraInfo_UTK] 获取模型信息时发生未知错误: {e}")
return {}
def calculate_sha256(file_path):
@@ -98,90 +112,135 @@ def sort_tags_by_frequency(meta_tags):
return []
def get_lora_info(lora_name):
db = load_json_from_file(db_path)
output = None
examplePrompt = None
trainedWords = None
baseModel = None
metaInfo = None
loraInfo = db.get(lora_name, {})
if isinstance(loraInfo, str):
loraInfo = {}
output = loraInfo.get('output', None)
examplePrompt = loraInfo.get('examplePrompt', None)
trainedWords = loraInfo.get('trainedWords', None)
baseModel = loraInfo.get('baseModel', None)
metaInfo = loraInfo.get('metaInfo', None)
if output is None or baseModel is None:
output = ""
lora_path = folder_paths.get_full_path("loras", lora_name)
LORAsha256 = calculate_sha256(lora_path)
model_info = get_model_version_info(LORAsha256)
if model_info.get("trainedWords", None) is None:
trainedWords = ""
else:
trainedWords = ",".join(model_info.get("trainedWords"))
baseModel = model_info.get("baseModel", "")
images = model_info.get('images')
try:
db = load_json_from_file(db_path)
output = None
examplePrompt = None
modelID = model_info.get("modelId")
trainedWords = None
baseModel = None
metaInfo = None
if modelID:
output += f"URL: https://civitai.com/models/{modelID}\n"
if trainedWords:
output += "Triggers: " + trainedWords
output += "\n"
loraInfo = db.get(lora_name, {})
if baseModel:
output += f"Base Model: {baseModel}\n"
if images:
output += "\nExamples:\n"
for image in images:
output += f"\nOutput: {image.get('url')}\n"
meta = image.get("meta")
if meta:
for key, value in meta.items():
if examplePrompt is None and key == "prompt":
examplePrompt = value
output += f"{key}: {value}\n"
output += '\n'
if isinstance(loraInfo, str):
loraInfo = {}
# 获取元数据信息
metadata = get_metadata(lora_name)
if metadata:
metaInfo = json.dumps(metadata, indent=2, ensure_ascii=False)
else:
metaInfo = ""
output = loraInfo.get('output', None)
examplePrompt = loraInfo.get('examplePrompt', None)
trainedWords = loraInfo.get('trainedWords', None)
baseModel = loraInfo.get('baseModel', None)
metaInfo = loraInfo.get('metaInfo', None)
db[lora_name] = {
"output": output,
"trainedWords": trainedWords,
"examplePrompt": examplePrompt,
"baseModel": baseModel,
"metaInfo": metaInfo
}
save_dict_to_json(db, db_path)
if output is None or baseModel is None:
output = ""
try:
lora_path = folder_paths.get_full_path("loras", lora_name)
if not lora_path:
print(f"[LoraInfo_UTK] 无法找到LoRA文件: {lora_name}")
return ("", "", "", "", "")
return (output, trainedWords, examplePrompt, baseModel, metaInfo)
LORAsha256 = calculate_sha256(lora_path)
model_info = get_model_version_info(LORAsha256)
if model_info.get("trainedWords", None) is None:
trainedWords = ""
else:
trainedWords = ",".join(model_info.get("trainedWords"))
baseModel = model_info.get("baseModel", "")
images = model_info.get('images')
examplePrompt = None
modelID = model_info.get("modelId")
if modelID:
output += f"URL: https://civitai.com/models/{modelID}\n"
if trainedWords:
output += "Triggers: " + trainedWords
output += "\n"
if baseModel:
output += f"Base Model: {baseModel}\n"
if images:
output += "\nExamples:\n"
for image in images:
output += f"\nOutput: {image.get('url')}\n"
meta = image.get("meta")
if meta:
for key, value in meta.items():
if examplePrompt is None and key == "prompt":
examplePrompt = value
output += f"{key}: {value}\n"
output += '\n'
# 获取元数据信息
try:
metadata = get_metadata(lora_name)
if metadata:
metaInfo = json.dumps(metadata, indent=2, ensure_ascii=False)
else:
metaInfo = ""
except Exception as e:
print(f"[LoraInfo_UTK] 读取元数据时发生错误: {e}")
metaInfo = ""
db[lora_name] = {
"output": output,
"trainedWords": trainedWords,
"examplePrompt": examplePrompt,
"baseModel": baseModel,
"metaInfo": metaInfo
}
save_dict_to_json(db, db_path)
except Exception as e:
print(f"[LoraInfo_UTK] 处理LoRA文件时发生错误: {e}")
output = f"处理LoRA文件时发生错误: {e}"
trainedWords = ""
examplePrompt = ""
baseModel = ""
metaInfo = ""
return (output, trainedWords, examplePrompt, baseModel, metaInfo)
except Exception as e:
print(f"[LoraInfo_UTK] 获取LoRA信息时发生严重错误: {e}")
return ("", "", "", "", "")
@server.PromptServer.instance.routes.post('/lora_info_utk')
async def fetch_lora_info(request):
post = await request.post()
lora_name = post.get("lora_name")
(output, triggerWords, examplePrompt, baseModel, metaInfo) = get_lora_info(lora_name)
try:
post = await request.post()
lora_name = post.get("lora_name")
return web.json_response({
"output": output,
"triggerWords": triggerWords,
"examplePrompt": examplePrompt,
"baseModel": baseModel,
"metaInfo": metaInfo
})
if not lora_name:
return web.json_response({
"error": "未提供LoRA名称",
"output": "",
"triggerWords": "",
"examplePrompt": "",
"baseModel": "",
"metaInfo": ""
})
(output, triggerWords, examplePrompt, baseModel, metaInfo) = get_lora_info(lora_name)
return web.json_response({
"output": output,
"triggerWords": triggerWords,
"examplePrompt": examplePrompt,
"baseModel": baseModel,
"metaInfo": metaInfo
})
except Exception as e:
print(f"[LoraInfo_UTK] Web API调用时发生错误: {e}")
return web.json_response({
"error": f"处理请求时发生错误: {e}",
"output": "",
"triggerWords": "",
"examplePrompt": "",
"baseModel": "",
"metaInfo": ""
})
class LoraInfo_UTK:
"""
@@ -212,26 +271,37 @@ class LoraInfo_UTK:
CATEGORY = "UniversalToolkit/Tools"
def lora_info(self, lora_name):
(output, triggerWords, examplePrompt, baseModel, metaInfo) = get_lora_info(lora_name)
try:
(output, triggerWords, examplePrompt, baseModel, metaInfo) = get_lora_info(lora_name)
# 构建信息文本
info_text = f"LoRA: {lora_name}\n"
if baseModel:
info_text += f"Base Model: {baseModel}\n"
if triggerWords:
info_text += f"Trigger Words: {triggerWords}\n"
if examplePrompt:
info_text += f"Example Prompt: {examplePrompt}\n"
if output:
info_text += f"\n详细信息:\n{output}"
# 构建信息文本
info_text = f"LoRA: {lora_name}\n"
if baseModel:
info_text += f"Base Model: {baseModel}\n"
if triggerWords:
info_text += f"Trigger Words: {triggerWords}\n"
if examplePrompt:
info_text += f"Example Prompt: {examplePrompt}\n"
if output:
info_text += f"\n详细信息:\n{output}"
return {
"ui": {
"text": (info_text,),
"model": (baseModel,)
},
"result": (lora_name, triggerWords or "", examplePrompt or "", info_text, metaInfo or "")
}
return {
"ui": {
"text": (info_text,),
"model": (baseModel,)
},
"result": (lora_name, triggerWords or "", examplePrompt or "", info_text, metaInfo or "")
}
except Exception as e:
print(f"[LoraInfo_UTK] 节点执行时发生错误: {e}")
error_text = f"LoRA: {lora_name}\n获取信息时发生错误: {e}"
return {
"ui": {
"text": (error_text,),
"model": ("",)
},
"result": (lora_name, "", "", error_text, "")
}
# Node mappings
+28 -6
View File
@@ -36,9 +36,24 @@ app.registerExtension({
.fetchApi("/lora_info_utk", { method: "POST", body })
.then((response) => response.json())
.then((resp) => {
baseModelWidget.value = resp.baseModel;
outputWidget.value = resp.output;
metaInfoWidget.value = resp.metaInfo;
if (resp.error) {
// 显示错误信息
baseModelWidget.value = "错误";
outputWidget.value = `获取信息失败: ${resp.error}`;
metaInfoWidget.value = "无法获取元数据";
} else {
// 正常显示信息
baseModelWidget.value = resp.baseModel;
outputWidget.value = resp.output;
metaInfoWidget.value = resp.metaInfo;
}
})
.catch((error) => {
// 处理网络错误
console.error("[LoraInfo_UTK] API调用失败:", error);
baseModelWidget.value = "网络错误";
outputWidget.value = "无法连接到服务器,请检查网络连接";
metaInfoWidget.value = "无法获取元数据";
});
};
}
@@ -46,9 +61,16 @@ app.registerExtension({
const onExecuted = nodeType.prototype.onExecuted;
nodeType.prototype.onExecuted = function (message) {
onExecuted?.apply(this, [message]);
this.showValueWidget.value = message.text[0];
this.baseModelWidget.value = message.model[0];
this.metaInfoWidget.value = message.metaInfo ? message.metaInfo[0] : "";
try {
this.showValueWidget.value = message.text[0];
this.baseModelWidget.value = message.model[0];
this.metaInfoWidget.value = message.metaInfo ? message.metaInfo[0] : "";
} catch (error) {
console.error("[LoraInfo_UTK] 更新界面时发生错误:", error);
this.showValueWidget.value = "界面更新失败";
this.baseModelWidget.value = "";
this.metaInfoWidget.value = "";
}
}
}
},