mirror of
https://github.com/willmiao/ComfyUI-Lora-Manager.git
synced 2026-06-22 11:21:15 -03:00
Compare commits
4 Commits
716f18e0ed
...
c8beaa64e1
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
c8beaa64e1 | ||
|
|
fb443ed6ae | ||
|
|
151a467598 | ||
|
|
98e1d168b0 |
@@ -1668,6 +1668,10 @@
|
||||
"noRecipeId": "Keine Rezept-ID verfügbar",
|
||||
"sendToWorkflowFailed": "Fehler beim Senden des Rezepts an den Workflow: {message}",
|
||||
"copyFailed": "Fehler beim Kopieren der Rezept-Syntax: {message}",
|
||||
"createError": "Fehler beim Erstellen des Rezepts:{message}",
|
||||
"createFailed": "Fehler beim Erstellen des Rezepts:{error}",
|
||||
"createMissingData": "Erforderliche Daten zum Erstellen des Rezepts fehlen",
|
||||
"created": "Rezept erfolgreich erstellt",
|
||||
"noMissingLoras": "Keine fehlenden LoRAs zum Herunterladen",
|
||||
"missingLorasInfoFailed": "Fehler beim Abrufen der Informationen für fehlende LoRAs",
|
||||
"preparingForDownloadFailed": "Fehler beim Vorbereiten der LoRAs für den Download",
|
||||
|
||||
@@ -1668,6 +1668,10 @@
|
||||
"noRecipeId": "No recipe ID available",
|
||||
"sendToWorkflowFailed": "Failed to send recipe to workflow: {message}",
|
||||
"copyFailed": "Error copying recipe syntax: {message}",
|
||||
"createError": "Error creating recipe: {message}",
|
||||
"createFailed": "Failed to create recipe: {error}",
|
||||
"createMissingData": "Missing required data to create recipe",
|
||||
"created": "Recipe created successfully",
|
||||
"noMissingLoras": "No missing LoRAs to download",
|
||||
"missingLorasInfoFailed": "Failed to get information for missing LoRAs",
|
||||
"preparingForDownloadFailed": "Error preparing LoRAs for download",
|
||||
|
||||
@@ -1668,6 +1668,10 @@
|
||||
"noRecipeId": "No hay ID de receta disponible",
|
||||
"sendToWorkflowFailed": "Error al enviar la receta al flujo de trabajo: {message}",
|
||||
"copyFailed": "Error copiando sintaxis de receta: {message}",
|
||||
"createError": "Error al crear la receta:{message}",
|
||||
"createFailed": "Error al crear la receta:{error}",
|
||||
"createMissingData": "Faltan datos necesarios para crear la receta",
|
||||
"created": "Receta creada exitosamente",
|
||||
"noMissingLoras": "No hay LoRAs faltantes para descargar",
|
||||
"missingLorasInfoFailed": "Error al obtener información de LoRAs faltantes",
|
||||
"preparingForDownloadFailed": "Error preparando LoRAs para descarga",
|
||||
|
||||
@@ -1668,6 +1668,10 @@
|
||||
"noRecipeId": "Aucun ID de recipe disponible",
|
||||
"sendToWorkflowFailed": "Échec de l'envoi de la recette vers le workflow : {message}",
|
||||
"copyFailed": "Erreur lors de la copie de la syntaxe de la recipe : {message}",
|
||||
"createError": "Erreur lors de la création du Recipe :{message}",
|
||||
"createFailed": "Échec de la création du Recipe :{error}",
|
||||
"createMissingData": "Données requises manquantes pour créer le Recipe",
|
||||
"created": "Recipe créé avec succès",
|
||||
"noMissingLoras": "Aucun LoRA manquant à télécharger",
|
||||
"missingLorasInfoFailed": "Échec de l'obtention des informations pour les LoRAs manquants",
|
||||
"preparingForDownloadFailed": "Erreur lors de la préparation des LoRAs pour le téléchargement",
|
||||
|
||||
@@ -1668,6 +1668,10 @@
|
||||
"noRecipeId": "אין מזהה מתכון זמין",
|
||||
"sendToWorkflowFailed": "נכשל שליחת המתכון ל-workflow: {message}",
|
||||
"copyFailed": "שגיאה בהעתקת תחביר המתכון: {message}",
|
||||
"createError": "שגיאה ביצירת המתכון:{message}",
|
||||
"createFailed": "יצירת המתכון נכשלה:{error}",
|
||||
"createMissingData": "חסרים נתונים נדרשים ליצירת המתכון",
|
||||
"created": "המתכון נוצר בהצלחה",
|
||||
"noMissingLoras": "אין LoRAs חסרים להורדה",
|
||||
"missingLorasInfoFailed": "קבלת מידע עבור LoRAs חסרים נכשלה",
|
||||
"preparingForDownloadFailed": "שגיאה בהכנת LoRAs להורדה",
|
||||
|
||||
@@ -1668,6 +1668,10 @@
|
||||
"noRecipeId": "レシピIDが利用できません",
|
||||
"sendToWorkflowFailed": "ワークフローへのレシピ送信に失敗しました:{message}",
|
||||
"copyFailed": "レシピ構文のコピーエラー:{message}",
|
||||
"createError": "レシピ作成中にエラーが発生しました:{message}",
|
||||
"createFailed": "レシピの作成に失敗しました:{error}",
|
||||
"createMissingData": "レシピ作成に必要なデータが不足しています",
|
||||
"created": "レシピを作成しました",
|
||||
"noMissingLoras": "ダウンロードする不足LoRAがありません",
|
||||
"missingLorasInfoFailed": "不足LoRAの情報取得に失敗しました",
|
||||
"preparingForDownloadFailed": "ダウンロード用LoRAの準備中にエラーが発生しました",
|
||||
|
||||
@@ -1668,6 +1668,10 @@
|
||||
"noRecipeId": "사용 가능한 레시피 ID가 없습니다",
|
||||
"sendToWorkflowFailed": "워크플로우에 레시피 보내기 실패: {message}",
|
||||
"copyFailed": "레시피 문법 복사 오류: {message}",
|
||||
"createError": "레시피 생성 중 오류 발생:{message}",
|
||||
"createFailed": "레시피 생성 실패:{error}",
|
||||
"createMissingData": "레시피 생성에 필요한 데이터가 없습니다",
|
||||
"created": "레시피가 생성되었습니다",
|
||||
"noMissingLoras": "다운로드할 누락된 LoRA가 없습니다",
|
||||
"missingLorasInfoFailed": "누락된 LoRA 정보를 가져오는데 실패했습니다",
|
||||
"preparingForDownloadFailed": "LoRA 다운로드 준비 오류",
|
||||
|
||||
@@ -1668,6 +1668,10 @@
|
||||
"noRecipeId": "ID рецепта недоступен",
|
||||
"sendToWorkflowFailed": "Не удалось отправить рецепт в рабочий процесс: {message}",
|
||||
"copyFailed": "Ошибка копирования синтаксиса рецепта: {message}",
|
||||
"createError": "Ошибка при создании рецепта:{message}",
|
||||
"createFailed": "Не удалось создать рецепт:{error}",
|
||||
"createMissingData": "Отсутствуют необходимые данные для создания рецепта",
|
||||
"created": "Рецепт успешно создан",
|
||||
"noMissingLoras": "Нет отсутствующих LoRAs для загрузки",
|
||||
"missingLorasInfoFailed": "Не удалось получить информацию для отсутствующих LoRAs",
|
||||
"preparingForDownloadFailed": "Ошибка подготовки LoRAs для загрузки",
|
||||
|
||||
@@ -1668,6 +1668,10 @@
|
||||
"noRecipeId": "无配方 ID",
|
||||
"sendToWorkflowFailed": "发送配方到工作流失败:{message}",
|
||||
"copyFailed": "复制配方语法出错:{message}",
|
||||
"createError": "创建配方时出错:{message}",
|
||||
"createFailed": "创建配方失败:{error}",
|
||||
"createMissingData": "缺少创建配方所需的数据",
|
||||
"created": "配方创建成功",
|
||||
"noMissingLoras": "没有缺失的 LoRA 可下载",
|
||||
"missingLorasInfoFailed": "获取缺失 LoRA 信息失败",
|
||||
"preparingForDownloadFailed": "准备下载 LoRA 时出错",
|
||||
|
||||
@@ -1668,6 +1668,10 @@
|
||||
"noRecipeId": "無配方 ID",
|
||||
"sendToWorkflowFailed": "傳送配方到工作流失敗:{message}",
|
||||
"copyFailed": "複製配方語法錯誤:{message}",
|
||||
"createError": "建立配方時發生錯誤:{message}",
|
||||
"createFailed": "建立配方失敗:{error}",
|
||||
"createMissingData": "缺少建立配方所需的資料",
|
||||
"created": "配方建立成功",
|
||||
"noMissingLoras": "無缺少的 LoRA 可下載",
|
||||
"missingLorasInfoFailed": "取得缺少 LoRA 資訊失敗",
|
||||
"preparingForDownloadFailed": "準備下載 LoRA 時發生錯誤",
|
||||
|
||||
@@ -58,9 +58,52 @@ class RecipeMetadataParser(ABC):
|
||||
civitai_info, error_msg = civitai_info_tuple if isinstance(civitai_info_tuple, tuple) else (civitai_info_tuple, None)
|
||||
|
||||
if not civitai_info or error_msg == "Model not found":
|
||||
# Model not found or deleted
|
||||
lora_entry['isDeleted'] = True
|
||||
lora_entry['thumbnailUrl'] = '/loras_static/images/no-preview.png'
|
||||
# CivitAI may fail to resolve a hash that is still being
|
||||
# computed (known CivitAI issue). Before marking as deleted,
|
||||
# try to reconcile with a local model that has the same
|
||||
# filename and matching AutoV3 hash.
|
||||
reconciled = False
|
||||
file_name = lora_entry.get("file_name")
|
||||
if file_name and recipe_scanner and hash_value:
|
||||
lora_scanner = getattr(recipe_scanner, "_lora_scanner", None)
|
||||
if lora_scanner:
|
||||
try:
|
||||
# Local import to avoid circular dependency:
|
||||
# base.py → file_utils → settings_manager → ...
|
||||
# → recipe_scanner → enrichment → base.py
|
||||
from ..utils.file_utils import calculate_autov3 # fmt: skip
|
||||
cache = await lora_scanner.get_cached_data()
|
||||
for item in getattr(cache, "raw_data", []):
|
||||
if item.get("file_name") == file_name:
|
||||
local_path = item.get("file_path")
|
||||
if local_path and os.path.exists(local_path):
|
||||
local_autov3 = calculate_autov3(local_path)
|
||||
if local_autov3 and local_autov3 == hash_value:
|
||||
lora_entry["existsLocally"] = True
|
||||
lora_entry["localPath"] = local_path
|
||||
lora_entry["hash"] = item.get("sha256", hash_value)
|
||||
if "preview_url" in item:
|
||||
lora_entry["thumbnailUrl"] = config.get_preview_static_url(item["preview_url"])
|
||||
civ = item.get("civitai") or {}
|
||||
if isinstance(civ, dict):
|
||||
if civ.get("id") is not None:
|
||||
lora_entry["id"] = civ["id"]
|
||||
if civ.get("modelId") is not None:
|
||||
lora_entry["modelId"] = civ["modelId"]
|
||||
if civ.get("name"):
|
||||
lora_entry["version"] = civ["name"]
|
||||
# model_name is the CivitAI model display
|
||||
# name stored directly in the cache column.
|
||||
cached_model_name = item.get("model_name")
|
||||
if cached_model_name:
|
||||
lora_entry["name"] = cached_model_name
|
||||
reconciled = True
|
||||
break
|
||||
except Exception:
|
||||
pass
|
||||
if not reconciled:
|
||||
lora_entry['isDeleted'] = True
|
||||
lora_entry['thumbnailUrl'] = '/loras_static/images/no-preview.png'
|
||||
return lora_entry
|
||||
|
||||
# Get model type and validate
|
||||
|
||||
@@ -6,6 +6,7 @@ from typing import Dict, Any, Union
|
||||
from ..base import RecipeMetadataParser
|
||||
from ..constants import GEN_PARAM_KEYS
|
||||
from ...services.metadata_service import get_default_metadata_provider
|
||||
from ...config import config
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -73,7 +74,8 @@ class CivitaiApiMetadataParser(RecipeMetadataParser):
|
||||
return False
|
||||
|
||||
async def parse_metadata( # type: ignore[override]
|
||||
self, user_comment, recipe_scanner=None, civitai_client=None
|
||||
self, user_comment, recipe_scanner=None, civitai_client=None,
|
||||
local_cache: dict[str, Any] | None = None,
|
||||
) -> Dict[str, Any]:
|
||||
"""Parse metadata from Civitai image format
|
||||
|
||||
@@ -81,6 +83,8 @@ class CivitaiApiMetadataParser(RecipeMetadataParser):
|
||||
user_comment: The metadata from the image (dict)
|
||||
recipe_scanner: Optional recipe scanner service
|
||||
civitai_client: Optional Civitai API client (deprecated, use metadata_provider instead)
|
||||
local_cache: Optional dict mapping sha256/autov3 hash → scanner cache item.
|
||||
When provided, matching models skip CivitAI API calls.
|
||||
|
||||
Returns:
|
||||
Dict containing parsed recipe data
|
||||
@@ -210,35 +214,45 @@ class CivitaiApiMetadataParser(RecipeMetadataParser):
|
||||
}
|
||||
|
||||
# Try to look up base model from the checkpoint hash
|
||||
if checkpoint_entry["hash"] and metadata_provider:
|
||||
try:
|
||||
civitai_info = (
|
||||
await metadata_provider.get_model_by_hash(
|
||||
checkpoint_entry["hash"]
|
||||
cp_hash = checkpoint_entry.get("hash")
|
||||
if cp_hash and metadata_provider:
|
||||
local_cached = local_cache.get(cp_hash) if local_cache else None
|
||||
if local_cached:
|
||||
self._populate_entry_from_cache(
|
||||
checkpoint_entry, local_cached
|
||||
)
|
||||
bm = checkpoint_entry.get("baseModel", "")
|
||||
if bm and not result["base_model"]:
|
||||
result["base_model"] = bm
|
||||
else:
|
||||
try:
|
||||
civitai_info = (
|
||||
await metadata_provider.get_model_by_hash(
|
||||
cp_hash
|
||||
)
|
||||
)
|
||||
civitai_data, error_msg = (
|
||||
(civitai_info, None)
|
||||
if not isinstance(civitai_info, tuple)
|
||||
else civitai_info
|
||||
)
|
||||
if civitai_data and error_msg != "Model not found":
|
||||
if 'model' in civitai_data and 'name' in civitai_data['model']:
|
||||
checkpoint_entry['name'] = civitai_data['model']['name']
|
||||
checkpoint_entry['id'] = civitai_data.get('id', 0)
|
||||
checkpoint_entry['modelId'] = civitai_data.get('modelId', 0)
|
||||
if 'name' in civitai_data:
|
||||
checkpoint_entry['version'] = civitai_data['name']
|
||||
base_model = civitai_data.get('baseModel', '')
|
||||
if base_model:
|
||||
checkpoint_entry['baseModel'] = base_model
|
||||
if not result['base_model']:
|
||||
result['base_model'] = base_model
|
||||
except Exception as e:
|
||||
logger.error(
|
||||
f"Error fetching checkpoint info for hash "
|
||||
f"{cp_hash}: {e}"
|
||||
)
|
||||
)
|
||||
civitai_data, error_msg = (
|
||||
(civitai_info, None)
|
||||
if not isinstance(civitai_info, tuple)
|
||||
else civitai_info
|
||||
)
|
||||
if civitai_data and error_msg != "Model not found":
|
||||
if 'model' in civitai_data and 'name' in civitai_data['model']:
|
||||
checkpoint_entry['name'] = civitai_data['model']['name']
|
||||
checkpoint_entry['id'] = civitai_data.get('id', 0)
|
||||
checkpoint_entry['modelId'] = civitai_data.get('modelId', 0)
|
||||
if 'name' in civitai_data:
|
||||
checkpoint_entry['version'] = civitai_data['name']
|
||||
base_model = civitai_data.get('baseModel', '')
|
||||
if base_model:
|
||||
checkpoint_entry['baseModel'] = base_model
|
||||
if not result['base_model']:
|
||||
result['base_model'] = base_model
|
||||
except Exception as e:
|
||||
logger.error(
|
||||
f"Error fetching checkpoint info for hash "
|
||||
f"{checkpoint_entry['hash']}: {e}"
|
||||
)
|
||||
|
||||
if result["model"] is None:
|
||||
result["model"] = checkpoint_entry
|
||||
@@ -279,34 +293,45 @@ class CivitaiApiMetadataParser(RecipeMetadataParser):
|
||||
}
|
||||
|
||||
# Try to get info from Civitai if hash is available
|
||||
if lora_entry["hash"] and metadata_provider:
|
||||
try:
|
||||
civitai_info = (
|
||||
await metadata_provider.get_model_by_hash(lora_hash)
|
||||
if lora_hash and metadata_provider:
|
||||
local_cached = local_cache.get(lora_hash) if local_cache else None
|
||||
if local_cached:
|
||||
self._populate_entry_from_cache(
|
||||
lora_entry, local_cached
|
||||
)
|
||||
|
||||
populated_entry = await self.populate_lora_from_civitai(
|
||||
lora_entry,
|
||||
civitai_info,
|
||||
recipe_scanner,
|
||||
base_model_counts,
|
||||
lora_hash,
|
||||
)
|
||||
|
||||
if populated_entry is None:
|
||||
continue # Skip invalid LoRA types
|
||||
|
||||
lora_entry = populated_entry
|
||||
|
||||
# If we have a version ID from Civitai, track it for deduplication
|
||||
if "id" in lora_entry and lora_entry["id"]:
|
||||
# Track by version ID for deduplication
|
||||
if lora_entry.get("id"):
|
||||
added_loras[str(lora_entry["id"])] = len(
|
||||
result["loras"]
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(
|
||||
f"Error fetching Civitai info for LoRA hash {lora_entry['hash']}: {e}"
|
||||
)
|
||||
else:
|
||||
try:
|
||||
civitai_info = (
|
||||
await metadata_provider.get_model_by_hash(lora_hash)
|
||||
)
|
||||
|
||||
populated_entry = await self.populate_lora_from_civitai(
|
||||
lora_entry,
|
||||
civitai_info,
|
||||
recipe_scanner,
|
||||
base_model_counts,
|
||||
lora_hash,
|
||||
)
|
||||
|
||||
if populated_entry is None:
|
||||
continue # Skip invalid LoRA types
|
||||
|
||||
lora_entry = populated_entry
|
||||
|
||||
# If we have a version ID from Civitai, track it for deduplication
|
||||
if "id" in lora_entry and lora_entry["id"]:
|
||||
added_loras[str(lora_entry["id"])] = len(
|
||||
result["loras"]
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(
|
||||
f"Error fetching Civitai info for LoRA hash {lora_entry['hash']}: {e}"
|
||||
)
|
||||
|
||||
# Track by hash if we have it
|
||||
if lora_hash:
|
||||
@@ -684,3 +709,41 @@ class CivitaiApiMetadataParser(RecipeMetadataParser):
|
||||
except Exception as e:
|
||||
logger.error(f"Error parsing Civitai image metadata: {e}", exc_info=True)
|
||||
return {"error": str(e), "loras": []}
|
||||
|
||||
@staticmethod
|
||||
def _populate_entry_from_cache(
|
||||
entry: dict[str, Any],
|
||||
cache_item: dict[str, Any],
|
||||
) -> None:
|
||||
"""Fill a lora/checkpoint entry from a scanner cache item.
|
||||
|
||||
Avoids CivitAI API calls for models that exist locally.
|
||||
Mirrors the population logic in
|
||||
``RecipeMetadataParser.populate_lora_from_civitai()`` but operates
|
||||
entirely on cached data.
|
||||
"""
|
||||
civ = cache_item.get("civitai") or {}
|
||||
if isinstance(civ, dict):
|
||||
if civ.get("id") is not None:
|
||||
entry["id"] = civ["id"]
|
||||
if civ.get("modelId") is not None:
|
||||
entry["modelId"] = civ["modelId"]
|
||||
if civ.get("name"):
|
||||
entry["version"] = civ["name"]
|
||||
cached_name = cache_item.get("model_name")
|
||||
if cached_name:
|
||||
entry["name"] = cached_name
|
||||
entry["existsLocally"] = True
|
||||
local_path = cache_item.get("file_path")
|
||||
if local_path:
|
||||
entry["localPath"] = local_path
|
||||
sha256 = cache_item.get("sha256")
|
||||
if sha256:
|
||||
entry["hash"] = sha256
|
||||
if "preview_url" in cache_item:
|
||||
entry["thumbnailUrl"] = config.get_preview_static_url(
|
||||
cache_item["preview_url"]
|
||||
)
|
||||
base_model = cache_item.get("base_model", "")
|
||||
if base_model:
|
||||
entry["baseModel"] = base_model
|
||||
|
||||
@@ -16,7 +16,7 @@ from aiohttp import web
|
||||
|
||||
from ...config import config
|
||||
from ...services.server_i18n import server_i18n as default_server_i18n
|
||||
from ...services.settings_manager import SettingsManager
|
||||
from ...services.settings_manager import SettingsManager, get_settings_manager
|
||||
from ...services.recipes import (
|
||||
RecipeAnalysisService,
|
||||
RecipeDownloadError,
|
||||
@@ -26,7 +26,12 @@ from ...services.recipes import (
|
||||
RecipeValidationError,
|
||||
)
|
||||
from ...services.metadata_service import get_default_metadata_provider
|
||||
from ...utils.civitai_utils import extract_civitai_image_id, rewrite_preview_url
|
||||
from ...utils.civitai_utils import (
|
||||
build_civitai_image_page_url,
|
||||
extract_civitai_image_id,
|
||||
extract_civitai_image_id_from_cdn_url,
|
||||
rewrite_preview_url,
|
||||
)
|
||||
from ...utils.exif_utils import ExifUtils
|
||||
from ...recipes.merger import GenParamsMerger
|
||||
from ...recipes.enrichment import RecipeEnricher
|
||||
@@ -96,6 +101,7 @@ class RecipeHandlerSet:
|
||||
"browse_directory": self.batch_import.browse_directory,
|
||||
"check_image_exists": self.management.check_image_exists,
|
||||
"import_from_url": self.management.import_from_url,
|
||||
"create_from_example": self.management.create_from_example,
|
||||
}
|
||||
|
||||
|
||||
@@ -1668,6 +1674,272 @@ class RecipeManagementHandler:
|
||||
)
|
||||
return web.json_response(result.payload, status=result.status)
|
||||
|
||||
async def create_from_example(self, request: web.Request) -> web.Response:
|
||||
"""Create a recipe from a model's example image using cached metadata.
|
||||
|
||||
Uses the image's meta data (already cached in .metadata.json from the
|
||||
CivitAI model-versions API) to create a recipe without additional
|
||||
CivitAI API calls.
|
||||
|
||||
If the image metadata doesn't contain any resources of the parent
|
||||
model's type (LoRA-type or Checkpoint), the parent model is
|
||||
auto-populated as a fallback.
|
||||
|
||||
Request body:
|
||||
image_data (dict): The full image object from model-versions API
|
||||
(includes meta, additionalResources, url, etc.)
|
||||
model_hash (str): SHA256 hash of the parent model
|
||||
model_name (str): Filename of the parent model
|
||||
model_type (str): Page type (``"loras"``, ``"checkpoints"``, etc.)
|
||||
local_image_path (str, optional): Local filesystem path to read
|
||||
the image bytes for the recipe preview
|
||||
"""
|
||||
try:
|
||||
await self._ensure_dependencies_ready()
|
||||
recipe_scanner = self._recipe_scanner_getter()
|
||||
if recipe_scanner is None:
|
||||
raise RuntimeError("Recipe scanner unavailable")
|
||||
|
||||
data = await request.json()
|
||||
image_data = data.get("image_data")
|
||||
model_hash = data.get("model_hash")
|
||||
model_name = data.get("model_name")
|
||||
model_type = data.get("model_type", "")
|
||||
|
||||
if not image_data or not model_hash or not model_name:
|
||||
raise RecipeValidationError(
|
||||
"Missing required fields: image_data, model_hash, model_name"
|
||||
)
|
||||
|
||||
# Merge nested meta into top level so the parser finds everything.
|
||||
# CivitaiApiMetadataParser expects prompt, seed, resources, etc.
|
||||
# at the top level or wrapped under a "meta" key.
|
||||
inner_meta = image_data.get("meta") or {}
|
||||
parsed_input = {**image_data, **inner_meta}
|
||||
parsed_input.pop("meta", None)
|
||||
|
||||
# Build a local cache of {hash → cache_item} so the parser can
|
||||
# skip CivitAI API calls for models that exist on disk.
|
||||
local_cache: Dict[str, Dict[str, Any]] = {}
|
||||
lora_scanner = getattr(recipe_scanner, "_lora_scanner", None)
|
||||
if lora_scanner and model_hash:
|
||||
try:
|
||||
parent_cache_data = await lora_scanner.get_cached_data()
|
||||
for item in getattr(parent_cache_data, "raw_data", []):
|
||||
if item.get("sha256", "").lower() == model_hash.lower():
|
||||
local_cache[model_hash.lower()] = item
|
||||
# Compute AutoV3 so the parser can also match on
|
||||
# that hash type (CivitAI metadata resources use
|
||||
# AutoV3).
|
||||
file_path = item.get("file_path")
|
||||
if file_path and os.path.exists(file_path):
|
||||
try:
|
||||
from ...utils.file_utils import (
|
||||
calculate_autov3,
|
||||
)
|
||||
autov3 = calculate_autov3(file_path)
|
||||
if autov3:
|
||||
local_cache[autov3.lower()] = item
|
||||
except Exception:
|
||||
pass
|
||||
break
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
parser = self._analysis_service._recipe_parser_factory.create_parser(
|
||||
parsed_input
|
||||
)
|
||||
if not parser:
|
||||
raise RecipeValidationError("Unable to parse image metadata")
|
||||
|
||||
from ...recipes.parsers.civitai_image import CivitaiApiMetadataParser
|
||||
|
||||
if isinstance(parser, CivitaiApiMetadataParser):
|
||||
parsed = await parser.parse_metadata(
|
||||
parsed_input,
|
||||
recipe_scanner=recipe_scanner,
|
||||
local_cache=local_cache,
|
||||
)
|
||||
else:
|
||||
parsed = await parser.parse_metadata(
|
||||
parsed_input, recipe_scanner=recipe_scanner
|
||||
)
|
||||
|
||||
loras = list(parsed.get("loras") or [])
|
||||
checkpoint = parsed.get("model")
|
||||
is_lora_type = model_type.startswith("lora")
|
||||
is_ckpt_type = model_type.startswith("checkpoint")
|
||||
|
||||
# Extract parent model metadata from local_cache (used below to
|
||||
# reconcile isDeleted entries and enrich auto-populated ones).
|
||||
parent_civitai_id: int | None = None
|
||||
parent_model_id: int | None = None
|
||||
parent_version_name: str | None = None
|
||||
parent_model_name: str | None = None
|
||||
# Prefer sha256 key; fall back to any cached entry.
|
||||
parent_item = local_cache.get(model_hash.lower()) if model_hash else None
|
||||
if parent_item is None and local_cache:
|
||||
parent_item = next(iter(local_cache.values()))
|
||||
if parent_item:
|
||||
civ = parent_item.get("civitai") or {}
|
||||
if isinstance(civ, dict):
|
||||
parent_civitai_id = civ.get("id")
|
||||
parent_model_id = civ.get("modelId")
|
||||
parent_version_name = civ.get("name")
|
||||
parent_model_name = parent_item.get("model_name")
|
||||
|
||||
# Reconcile isDeleted entries against the parent model.
|
||||
# When the CivitAI hash lookup fails (known issue — hashes not
|
||||
# yet computed), the parser marks the entry isDeleted even though
|
||||
# the model exists locally.
|
||||
if is_lora_type:
|
||||
for lora in loras:
|
||||
if lora.get("isDeleted") and lora.get("file_name") == model_name:
|
||||
lora["isDeleted"] = False
|
||||
lora["existsLocally"] = True
|
||||
lora["hash"] = model_hash
|
||||
if parent_civitai_id is not None:
|
||||
lora["id"] = parent_civitai_id
|
||||
if parent_model_id is not None:
|
||||
lora["modelId"] = parent_model_id
|
||||
if parent_version_name is not None:
|
||||
lora["version"] = parent_version_name
|
||||
if parent_model_name is not None:
|
||||
lora["name"] = parent_model_name
|
||||
elif is_ckpt_type and checkpoint and checkpoint.get("isDeleted"):
|
||||
if checkpoint.get("file_name") == model_name:
|
||||
checkpoint["isDeleted"] = False
|
||||
checkpoint["existsLocally"] = True
|
||||
checkpoint["hash"] = model_hash
|
||||
if parent_civitai_id is not None:
|
||||
checkpoint["id"] = parent_civitai_id
|
||||
if parent_model_id is not None:
|
||||
checkpoint["modelId"] = parent_model_id
|
||||
if parent_version_name is not None:
|
||||
checkpoint["version"] = parent_version_name
|
||||
|
||||
# Auto-populate parent model only when the image metadata didn't
|
||||
# contain any resources of that type.
|
||||
if is_lora_type and not loras:
|
||||
lora_entry = {
|
||||
"name": model_name,
|
||||
"type": "lora",
|
||||
"weight": 1.0,
|
||||
"hash": model_hash,
|
||||
"existsLocally": True,
|
||||
"localPath": None,
|
||||
"file_name": model_name,
|
||||
"thumbnailUrl": "/loras_static/images/no-preview.png",
|
||||
"baseModel": parsed.get("base_model", ""),
|
||||
"size": 0,
|
||||
"downloadUrl": "",
|
||||
"isDeleted": False,
|
||||
}
|
||||
if parent_civitai_id is not None:
|
||||
lora_entry["id"] = parent_civitai_id
|
||||
if parent_model_id is not None:
|
||||
lora_entry["modelId"] = parent_model_id
|
||||
if parent_version_name is not None:
|
||||
lora_entry["version"] = parent_version_name
|
||||
if parent_model_name is not None:
|
||||
lora_entry["name"] = parent_model_name
|
||||
loras.insert(0, lora_entry)
|
||||
elif is_ckpt_type and not checkpoint:
|
||||
checkpoint = {
|
||||
"name": model_name,
|
||||
"type": "checkpoint",
|
||||
"hash": model_hash,
|
||||
"file_name": model_name,
|
||||
"existsLocally": True,
|
||||
"baseModel": parsed.get("base_model", ""),
|
||||
"isDeleted": False,
|
||||
}
|
||||
if parent_civitai_id is not None:
|
||||
checkpoint["id"] = parent_civitai_id
|
||||
if parent_model_id is not None:
|
||||
checkpoint["modelId"] = parent_model_id
|
||||
if parent_version_name is not None:
|
||||
checkpoint["version"] = parent_version_name
|
||||
if parent_model_name is not None:
|
||||
checkpoint["name"] = parent_model_name
|
||||
|
||||
image_url = image_data.get("url") or ""
|
||||
image_id = extract_civitai_image_id_from_cdn_url(image_url)
|
||||
settings_mgr = get_settings_manager()
|
||||
civitai_host = settings_mgr.get("civitai_host") if settings_mgr else None
|
||||
page_url = build_civitai_image_page_url(image_id, host=civitai_host) or image_url
|
||||
|
||||
recipe_metadata: dict[str, Any] = {
|
||||
"base_model": parsed.get("base_model") or "",
|
||||
"loras": loras,
|
||||
"gen_params": parsed.get("gen_params") or {},
|
||||
"source_path": page_url,
|
||||
}
|
||||
nsfw_level = image_data.get("nsfwLevel")
|
||||
if isinstance(nsfw_level, int):
|
||||
recipe_metadata["preview_nsfw_level"] = nsfw_level
|
||||
if checkpoint:
|
||||
recipe_metadata["checkpoint"] = checkpoint
|
||||
|
||||
image_bytes: bytes | None = None
|
||||
extension: str | None = None
|
||||
local_image_path = data.get("local_image_path")
|
||||
if local_image_path and os.path.exists(local_image_path):
|
||||
with open(local_image_path, "rb") as f:
|
||||
image_bytes = f.read()
|
||||
ext = os.path.splitext(local_image_path)[1].lower()
|
||||
if ext in (".jpg", ".jpeg", ".png", ".webp", ".gif"):
|
||||
extension = ext
|
||||
elif image_data.get("url"):
|
||||
try:
|
||||
downloader = await self._downloader_factory()
|
||||
url = image_data["url"]
|
||||
tmp = tempfile.NamedTemporaryFile(delete=False)
|
||||
tmp.close()
|
||||
success, result = await downloader.download_file(
|
||||
url, tmp.name, use_auth=False
|
||||
)
|
||||
if success:
|
||||
with open(tmp.name, "rb") as f:
|
||||
image_bytes = f.read()
|
||||
url_path = url.split("?")[0].split("#")[0]
|
||||
ext = os.path.splitext(url_path)[1].lower()
|
||||
if ext:
|
||||
extension = ext
|
||||
if os.path.exists(tmp.name):
|
||||
os.unlink(tmp.name)
|
||||
except Exception as exc:
|
||||
self._logger.warning(
|
||||
"Failed to download image for recipe: %s", exc
|
||||
)
|
||||
|
||||
prompt = (
|
||||
(parsed.get("gen_params") or {}).get("prompt") or ""
|
||||
)
|
||||
if prompt:
|
||||
name = " ".join(str(prompt).split()[:10])
|
||||
else:
|
||||
name = f"Recipe from {model_name}"
|
||||
|
||||
save_result = await self._persistence_service.save_recipe(
|
||||
recipe_scanner=recipe_scanner,
|
||||
image_bytes=image_bytes,
|
||||
image_base64=None,
|
||||
name=name,
|
||||
tags=[],
|
||||
metadata=recipe_metadata,
|
||||
extension=extension,
|
||||
)
|
||||
return web.json_response(save_result.payload, status=save_result.status)
|
||||
|
||||
except RecipeValidationError as exc:
|
||||
return web.json_response({"error": str(exc)}, status=400)
|
||||
except Exception as exc:
|
||||
self._logger.error(
|
||||
"Error creating recipe from example: %s", exc, exc_info=True
|
||||
)
|
||||
return web.json_response({"error": str(exc)}, status=500)
|
||||
|
||||
|
||||
class RecipeAnalysisHandler:
|
||||
"""Analyze images to extract recipe metadata."""
|
||||
|
||||
@@ -75,6 +75,9 @@ ROUTE_DEFINITIONS: tuple[RouteDefinition, ...] = (
|
||||
"GET", "/api/lm/recipes/check-image-exists", "check_image_exists"
|
||||
),
|
||||
RouteDefinition("GET", "/api/lm/recipes/import-from-url", "import_from_url"),
|
||||
RouteDefinition(
|
||||
"POST", "/api/lm/recipes/create-from-example", "create_from_example"
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
|
||||
@@ -115,6 +115,10 @@ class RecipePersistenceService:
|
||||
if metadata.get("source_path"):
|
||||
recipe_data["source_path"] = metadata.get("source_path")
|
||||
|
||||
nsfw_level = metadata.get("preview_nsfw_level")
|
||||
if nsfw_level is not None and isinstance(nsfw_level, int):
|
||||
recipe_data["preview_nsfw_level"] = nsfw_level
|
||||
|
||||
json_filename = f"{recipe_id}.recipe.json"
|
||||
json_path = os.path.join(recipes_dir, json_filename)
|
||||
json_path = os.path.normpath(json_path)
|
||||
|
||||
@@ -66,6 +66,46 @@ def build_civitai_model_page_url(
|
||||
return None
|
||||
|
||||
|
||||
_RE_CDN_IMAGE_ID = re.compile(r"/(\d+)\.(?:jpeg|jpg|png|webp|gif)(?:\?|#|$)")
|
||||
|
||||
|
||||
def extract_civitai_image_id_from_cdn_url(url: str | None) -> str | None:
|
||||
"""Extract the numeric image ID from a Cloudflare CDN image URL.
|
||||
|
||||
CivitAI image CDN URLs follow the pattern::
|
||||
|
||||
https://image.civitai.com/{cf_uuid}/{params}/{image_id}.{ext}
|
||||
|
||||
The image database ID is always the last path segment (minus extension)
|
||||
because ``getEdgeUrl(…, name=id.toString())`` embeds it explicitly
|
||||
in the model-versions REST API response.
|
||||
"""
|
||||
if not url:
|
||||
return None
|
||||
match = _RE_CDN_IMAGE_ID.search(url)
|
||||
return match.group(1) if match else None
|
||||
|
||||
|
||||
def build_civitai_image_page_url(
|
||||
image_id: str | int | None,
|
||||
*,
|
||||
host: str | None = None,
|
||||
) -> str | None:
|
||||
"""Build a Civitai image page URL.
|
||||
|
||||
Returns something like ``https://civitai.com/images/12345``.
|
||||
The host is resolved through :func:`normalize_civitai_page_host` and
|
||||
therefore respects the user's ``civitai_host`` setting.
|
||||
"""
|
||||
if not image_id:
|
||||
return None
|
||||
normalized_host = normalize_civitai_page_host(host)
|
||||
normalized_id = str(image_id).strip()
|
||||
if not normalized_id:
|
||||
return None
|
||||
return urlunparse(("https", normalized_host, f"/images/{normalized_id}", "", "", ""))
|
||||
|
||||
|
||||
def _parse_supported_civitai_page_url(url: str | None):
|
||||
if not url:
|
||||
return None
|
||||
@@ -328,8 +368,10 @@ def rewrite_preview_url(
|
||||
|
||||
|
||||
__all__ = [
|
||||
"build_civitai_image_page_url",
|
||||
"build_license_flags",
|
||||
"extract_civitai_image_id",
|
||||
"extract_civitai_image_id_from_cdn_url",
|
||||
"extract_civitai_page_host",
|
||||
"extract_civitai_model_url_parts",
|
||||
"is_supported_civitai_page_host",
|
||||
|
||||
@@ -1,7 +1,10 @@
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import struct
|
||||
from typing import Any
|
||||
|
||||
from .constants import (
|
||||
CARD_PREVIEW_WIDTH,
|
||||
@@ -31,7 +34,7 @@ def _get_hash_chunk_size_bytes() -> int:
|
||||
|
||||
|
||||
async def calculate_sha256(file_path: str) -> str:
|
||||
"""Calculate SHA256 hash of a file"""
|
||||
"""Calculate SHA256 hash of a file (full file content)."""
|
||||
sha256_hash = hashlib.sha256()
|
||||
chunk_size = _get_hash_chunk_size_bytes()
|
||||
with open(file_path, "rb") as f:
|
||||
@@ -39,6 +42,79 @@ async def calculate_sha256(file_path: str) -> str:
|
||||
sha256_hash.update(byte_block)
|
||||
return sha256_hash.hexdigest()
|
||||
|
||||
|
||||
def calculate_autov2(file_path: str) -> str:
|
||||
"""Calculate CivitAI AutoV2 hash.
|
||||
|
||||
AutoV2 is the first 10 characters of the full file SHA256.
|
||||
Used by CivitAI as a shortened file identifier.
|
||||
|
||||
Reference: https://developer.civitai.com/site/reference/model-versions
|
||||
"""
|
||||
full_hash = hashlib.sha256()
|
||||
chunk_size = _get_hash_chunk_size_bytes()
|
||||
with open(file_path, "rb") as f:
|
||||
for byte_block in iter(lambda: f.read(chunk_size), b""):
|
||||
full_hash.update(byte_block)
|
||||
return full_hash.hexdigest()[:10]
|
||||
|
||||
|
||||
def read_safetensors_metadata(file_path: str) -> dict[str, Any]:
|
||||
"""Read the ``__metadata__`` dict from a safetensors file header.
|
||||
|
||||
Safetensors file format:
|
||||
- 8 bytes: header length (little-endian 64-bit)
|
||||
- N bytes: UTF-8 JSON header
|
||||
- The header JSON contains a ``__metadata__`` key holding arbitrary metadata.
|
||||
|
||||
Returns an empty dict if the file is not a valid safetensors file or has no
|
||||
metadata.
|
||||
"""
|
||||
try:
|
||||
with open(file_path, "rb") as f:
|
||||
header_len_bytes = f.read(8)
|
||||
if len(header_len_bytes) < 8:
|
||||
return {}
|
||||
header_len = struct.unpack("<Q", header_len_bytes)[0]
|
||||
header_bytes = f.read(header_len)
|
||||
if len(header_bytes) < header_len:
|
||||
return {}
|
||||
header = json.loads(header_bytes.decode("utf-8"))
|
||||
return header.get("__metadata__", {})
|
||||
except (OSError, json.JSONDecodeError, UnicodeDecodeError, struct.error):
|
||||
return {}
|
||||
|
||||
|
||||
def calculate_autov3(file_path: str) -> str | None:
|
||||
"""Calculate CivitAI AutoV3 hash from a safetensors file.
|
||||
|
||||
AutoV3 is extracted from the safetensors file's embedded metadata, not
|
||||
computed from the file bytes directly. The orchestrator reads the
|
||||
``sshs_model_hash`` (kohya-ss format) or ``modelspec.hash_sha256`` field
|
||||
from the safetensors header and stores the first 12 characters.
|
||||
|
||||
The embedded hash itself is the SHA256 of the file after skipping the
|
||||
8-byte header length + JSON header (a.k.a. the addnet hash / tensor-only
|
||||
hash).
|
||||
|
||||
Reference:
|
||||
- CivitAI DB trigger: ``SUBSTRING(NEW.hash FROM 1 FOR 12)``
|
||||
- https://developer.civitai.com/site/reference/model-versions
|
||||
|
||||
Returns ``None`` when no AutoV3 hash can be determined (e.g. the file is
|
||||
not safetensors, or the metadata doesn't contain a recognised hash field).
|
||||
"""
|
||||
metadata = read_safetensors_metadata(file_path)
|
||||
if not metadata:
|
||||
return None
|
||||
|
||||
embedded_hash = metadata.get("sshs_model_hash") or metadata.get("modelspec.hash_sha256")
|
||||
if embedded_hash and isinstance(embedded_hash, str) and len(embedded_hash) >= 12:
|
||||
return embedded_hash[:12]
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def find_preview_file(base_name: str, dir_path: str) -> str:
|
||||
"""Find preview file for given base name in directory.
|
||||
|
||||
|
||||
404
scripts/restore_suffixed_filenames.py
Normal file
404
scripts/restore_suffixed_filenames.py
Normal file
@@ -0,0 +1,404 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Restore original filenames by removing leftover 4-char hash suffixes.
|
||||
|
||||
When LoRA Manager's old duplicate filename resolver ran, it appended
|
||||
``-{first4ofSHA256}`` to duplicate filenames, e.g.::
|
||||
|
||||
my_lora.safetensors → my_lora-a3f7.safetensors
|
||||
|
||||
With full-path LoRA syntax now available (``<lora:subfolder/name:1.0>``),
|
||||
these suffixes are unnecessary. This script detects such files and, with
|
||||
your confirmation, restores their original names.
|
||||
|
||||
The same suffix pattern is also used by the download conflict handler
|
||||
(``{name}-{hash}.{ext}``). To avoid false positives, this script skips
|
||||
any file whose original name already exists in the same directory — those
|
||||
were likely added by a download conflict, not the old resolver.
|
||||
|
||||
Usage::
|
||||
|
||||
# Detect only (dry-run, default)
|
||||
python scripts/restore_suffixed_filenames.py
|
||||
|
||||
# Detect + restore (with confirmation prompt)
|
||||
python scripts/restore_suffixed_filenames.py --apply
|
||||
|
||||
After restoring filenames, run **Rebuild Cache** in the LoRA Manager
|
||||
Doctor panel to refresh the model cache.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
logging.basicConfig(
|
||||
level=logging.INFO,
|
||||
format="%(message)s",
|
||||
)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
APP_NAME = "ComfyUI-LoRA-Manager"
|
||||
MODEL_EXTENSIONS = {".safetensors", ".ckpt", ".pt", ".pth", ".bin"}
|
||||
PREVIEW_EXTENSIONS = {
|
||||
".png", ".jpg", ".jpeg", ".webp", ".gif", ".bmp",
|
||||
".mp4", ".webm", ".mov",
|
||||
}
|
||||
|
||||
# Matches filenames like "my_lora-a3f7.safetensors"
|
||||
# Groups: (base_name, 4-char-hex, extension)
|
||||
_SUFFIX_RE = re.compile(r"^(.+)-([0-9a-f]{4})(\.[^.]+)$")
|
||||
|
||||
|
||||
# ── helpers (copied from migrate_legacy_metadata.py for consistency) ──────────
|
||||
|
||||
|
||||
def resolve_settings_path() -> Path:
|
||||
repo_root = Path(__file__).parent.parent.resolve()
|
||||
portable = repo_root / "settings.json"
|
||||
if portable.exists():
|
||||
payload = _load_json(portable)
|
||||
if isinstance(payload, dict) and payload.get("use_portable_settings") is True:
|
||||
return portable
|
||||
|
||||
config_home = os.environ.get("XDG_CONFIG_HOME")
|
||||
if config_home:
|
||||
return Path(config_home).expanduser() / APP_NAME / "settings.json"
|
||||
return Path.home() / ".config" / APP_NAME / "settings.json"
|
||||
|
||||
|
||||
def _load_json(path: Path) -> dict[str, Any]:
|
||||
try:
|
||||
with path.open("r", encoding="utf-8") as f:
|
||||
return json.load(f)
|
||||
except (FileNotFoundError, json.JSONDecodeError, OSError):
|
||||
return {}
|
||||
|
||||
|
||||
def _expand_path(value: str) -> str:
|
||||
return str(Path(value).expanduser().resolve(strict=False))
|
||||
|
||||
|
||||
def _normalize_path_list(value: Any) -> list[str]:
|
||||
if isinstance(value, str):
|
||||
return [_expand_path(value)] if value else []
|
||||
if isinstance(value, list):
|
||||
return [_expand_path(item) for item in value if isinstance(item, str) and item]
|
||||
return []
|
||||
|
||||
|
||||
def _dedupe(values: list[str]) -> list[str]:
|
||||
seen: set[str] = set()
|
||||
result: list[str] = []
|
||||
for value in values:
|
||||
if value not in seen:
|
||||
result.append(value)
|
||||
seen.add(value)
|
||||
return result
|
||||
|
||||
|
||||
def get_model_roots(settings: dict[str, Any]) -> dict[str, list[str]]:
|
||||
"""Extract model folder roots from LoRA Manager settings.
|
||||
|
||||
Returns ``{model_type: [path, ...]}`` where *model_type* is one of
|
||||
``loras``, ``checkpoints``, ``embeddings``, ``unet``, etc.
|
||||
|
||||
Both primary (``folder_paths``) and extra (``extra_folder_paths``)
|
||||
paths are included. Extra paths can be configured via the UI at
|
||||
Settings → Model Libraries → Extra Folder Paths.
|
||||
"""
|
||||
roots: dict[str, list[str]] = {}
|
||||
active_library = settings.get("active_library") or "default"
|
||||
sources = [settings]
|
||||
library = settings.get("libraries", {}).get(active_library)
|
||||
if isinstance(library, dict):
|
||||
sources.insert(0, library)
|
||||
for source in sources:
|
||||
# Primary folder paths.
|
||||
folder_paths = source.get("folder_paths")
|
||||
if isinstance(folder_paths, dict):
|
||||
for key, value in folder_paths.items():
|
||||
roots.setdefault(key, []).extend(_normalize_path_list(value))
|
||||
# Extra folder paths (Settings → Model Libraries → Extra Folder Paths).
|
||||
extra_folder_paths = source.get("extra_folder_paths")
|
||||
if isinstance(extra_folder_paths, dict):
|
||||
for key, value in extra_folder_paths.items():
|
||||
roots.setdefault(key, []).extend(_normalize_path_list(value))
|
||||
for default_key, folder_key in (
|
||||
("default_lora_root", "loras"),
|
||||
("default_checkpoint_root", "checkpoints"),
|
||||
("default_unet_root", "unet"),
|
||||
("default_embedding_root", "embeddings"),
|
||||
):
|
||||
value = settings.get(default_key)
|
||||
if isinstance(value, str) and value:
|
||||
roots.setdefault(folder_key, []).append(_expand_path(value))
|
||||
return {key: _dedupe(values) for key, values in roots.items()}
|
||||
|
||||
|
||||
def find_model_files(directory: Path) -> list[Path]:
|
||||
"""Recursively find all model files in *directory*."""
|
||||
files: list[Path] = []
|
||||
for ext in MODEL_EXTENSIONS:
|
||||
files.extend(directory.rglob(f"*{ext}"))
|
||||
return files
|
||||
|
||||
|
||||
# ── core detection logic ──────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def check_file(path: Path) -> tuple[str, str, str] | None:
|
||||
"""If *path* matches the suffix pattern, return ``(base_name, hex, ext)``.
|
||||
|
||||
Returns ``None`` when:
|
||||
* The filename does not match the pattern, or
|
||||
* The original name (without the suffix) already exists in the same
|
||||
directory (likely a download-conflict rename, not a doctor rename).
|
||||
"""
|
||||
match = _SUFFIX_RE.match(path.name)
|
||||
if not match:
|
||||
return None
|
||||
|
||||
base_name = match.group(1)
|
||||
hex_part = match.group(2)
|
||||
extension = match.group(3)
|
||||
orig_name = base_name + extension
|
||||
orig_path = path.with_name(orig_name)
|
||||
|
||||
# Safety: skip if the original name already exists.
|
||||
if orig_path.exists():
|
||||
return None
|
||||
|
||||
return base_name, hex_part, extension
|
||||
|
||||
|
||||
def scan_roots(
|
||||
roots: dict[str, list[str]],
|
||||
) -> dict[str, list[tuple[Path, str, str, str]]]:
|
||||
"""Scan all model roots and return detected files grouped by model type.
|
||||
|
||||
Returns ``{model_type: [(full_path, base_name, hex, ext), ...]}``.
|
||||
"""
|
||||
results: dict[str, list[tuple[Path, str, str, str]]] = {}
|
||||
|
||||
for model_type, root_list in roots.items():
|
||||
type_results: list[tuple[Path, str, str, str]] = []
|
||||
for root in root_list:
|
||||
root_path = Path(root)
|
||||
if not root_path.is_dir():
|
||||
continue
|
||||
for model_file in find_model_files(root_path):
|
||||
match = check_file(model_file)
|
||||
if match:
|
||||
type_results.append((model_file, *match))
|
||||
if type_results:
|
||||
results[model_type] = type_results
|
||||
|
||||
return results
|
||||
|
||||
|
||||
def rename_file(
|
||||
path: Path, base_name: str, extension: str, dry_run: bool
|
||||
) -> bool:
|
||||
"""Rename *path* to ``{base_name}{extension}``.
|
||||
|
||||
Also renames sidecar files (``.metadata.json``, ``.civitai.info``) and
|
||||
preview images. Returns ``True`` on success.
|
||||
"""
|
||||
new_path = path.with_name(base_name + extension)
|
||||
old_stem = path.with_suffix("") # /dir/base_name-hex (no ext)
|
||||
new_stem = new_path.with_suffix("") # /dir/base_name (no ext)
|
||||
|
||||
if dry_run:
|
||||
logger.info(" would rename: %s", path.name)
|
||||
logger.info(" -> %s", new_path.name)
|
||||
return True
|
||||
|
||||
try:
|
||||
os.rename(path, new_path)
|
||||
except OSError as exc:
|
||||
logger.error(" FAILED to rename %s: %s", path.name, exc)
|
||||
return False
|
||||
|
||||
# Rename sidecar metadata files.
|
||||
for suffix in (".metadata.json", ".civitai.info"):
|
||||
old_sidecar = old_stem.with_name(old_stem.name + suffix)
|
||||
new_sidecar = new_stem.with_name(new_stem.name + suffix)
|
||||
if old_sidecar.exists():
|
||||
try:
|
||||
os.rename(old_sidecar, new_sidecar)
|
||||
except OSError as exc:
|
||||
logger.warning(" could not rename sidecar %s: %s", old_sidecar.name, exc)
|
||||
|
||||
# Rename preview images.
|
||||
for preview_ext in PREVIEW_EXTENSIONS:
|
||||
old_preview = old_stem.with_name(old_stem.name + preview_ext)
|
||||
new_preview = new_stem.with_name(new_stem.name + preview_ext)
|
||||
if old_preview.exists():
|
||||
try:
|
||||
os.rename(old_preview, new_preview)
|
||||
except OSError as exc:
|
||||
logger.warning(" could not rename preview %s: %s", old_preview.name, exc)
|
||||
|
||||
logger.info(" renamed: %s -> %s", path.name, new_path.name)
|
||||
return True
|
||||
|
||||
|
||||
# ── report helpers ────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def print_report(results: dict[str, list[tuple[Path, str, str, str]]]) -> int:
|
||||
"""Print a human-readable report of detected files. Returns total count."""
|
||||
if not results:
|
||||
logger.info("No leftover suffixed filenames detected.")
|
||||
return 0
|
||||
|
||||
total = 0
|
||||
for model_type in sorted(results):
|
||||
entries = results[model_type]
|
||||
total += len(entries)
|
||||
label = model_type.capitalize()
|
||||
logger.info("")
|
||||
logger.info("─" * 50)
|
||||
logger.info(" %s (%d file(s))", label, len(entries))
|
||||
logger.info("─" * 50)
|
||||
for path, base_name, hex_part, ext in sorted(entries):
|
||||
logger.info(" %s → %s%s", path.name, base_name, ext)
|
||||
|
||||
logger.info("")
|
||||
logger.info("=" * 50)
|
||||
logger.info(" Total: %d file(s) with leftover suffixes.", total)
|
||||
logger.info("=" * 50)
|
||||
return total
|
||||
|
||||
|
||||
def prompt_user(count: int) -> bool:
|
||||
"""Ask the user whether to proceed with the rename."""
|
||||
try:
|
||||
answer = input(
|
||||
f"\nRestore {count} file(s) to their original names? [y/N] "
|
||||
).strip().lower()
|
||||
except (EOFError, KeyboardInterrupt):
|
||||
print()
|
||||
return False
|
||||
return answer in ("y", "yes")
|
||||
|
||||
|
||||
# ── main ──────────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(
|
||||
description=(
|
||||
"Detect and restore model filenames that have leftover "
|
||||
"4-character hash suffixes from the old conflict resolver."
|
||||
),
|
||||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||||
epilog=(
|
||||
"Examples:\n"
|
||||
" python scripts/restore_suffixed_filenames.py\n"
|
||||
" python scripts/restore_suffixed_filenames.py --apply\n"
|
||||
" python scripts/restore_suffixed_filenames.py --apply --yes\n"
|
||||
),
|
||||
)
|
||||
parser.add_argument(
|
||||
"--apply",
|
||||
action="store_true",
|
||||
help="Actually rename files (with confirmation prompt unless --yes is given)",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--yes", "-y",
|
||||
action="store_true",
|
||||
help="Skip confirmation prompt (implies --apply)",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--dry-run",
|
||||
action="store_true",
|
||||
help="Detect only — show what would be renamed without making changes",
|
||||
)
|
||||
parser.add_argument(
|
||||
"-v", "--verbose",
|
||||
action="store_true",
|
||||
help="Enable debug-level logging",
|
||||
)
|
||||
args = parser.parse_args()
|
||||
|
||||
if args.verbose:
|
||||
logging.getLogger().setLevel(logging.DEBUG)
|
||||
|
||||
# Resolve settings.
|
||||
settings_path = resolve_settings_path()
|
||||
logger.info("Settings: %s", settings_path)
|
||||
settings = _load_json(settings_path)
|
||||
if not settings:
|
||||
logger.error("Could not load settings.json. Is LoRA Manager configured?")
|
||||
return 1
|
||||
|
||||
roots = get_model_roots(settings)
|
||||
if not roots:
|
||||
logger.error("No model folders found in settings.")
|
||||
return 1
|
||||
|
||||
# Log which roots are being scanned.
|
||||
for model_type, root_list in roots.items():
|
||||
for root in root_list:
|
||||
logger.info("Scanning %s: %s", model_type, root)
|
||||
|
||||
# Detect.
|
||||
results = scan_roots(roots)
|
||||
total = print_report(results)
|
||||
|
||||
if total == 0:
|
||||
return 0
|
||||
|
||||
# Determine mode.
|
||||
dry_run = not args.apply and not args.yes
|
||||
|
||||
if dry_run:
|
||||
logger.info("\n[Dry-run mode — no files modified]")
|
||||
logger.info("Run with --apply to restore filenames.")
|
||||
return 0
|
||||
|
||||
# Confirm unless --yes.
|
||||
if not args.yes:
|
||||
if not prompt_user(total):
|
||||
logger.info("Aborted.")
|
||||
return 0
|
||||
|
||||
# Rename.
|
||||
logger.info("")
|
||||
success = 0
|
||||
fail = 0
|
||||
for model_type in sorted(results):
|
||||
entries = results[model_type]
|
||||
logger.info("")
|
||||
logger.info("─" * 50)
|
||||
logger.info(" Restoring %s (%d file(s))", model_type, len(entries))
|
||||
logger.info("─" * 50)
|
||||
for path, base_name, hex_part, ext in sorted(entries):
|
||||
ok = rename_file(path, base_name, ext, dry_run=False)
|
||||
if ok:
|
||||
success += 1
|
||||
else:
|
||||
fail += 1
|
||||
|
||||
logger.info("")
|
||||
logger.info("=" * 50)
|
||||
logger.info(" Done: %d restored, %d failed.", success, fail)
|
||||
logger.info("=" * 50)
|
||||
logger.info("")
|
||||
logger.info(" ⚠ Please run Rebuild Cache in the LoRA Manager")
|
||||
logger.info(" Doctor panel to refresh the model cache.")
|
||||
|
||||
return 0 if fail == 0 else 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -522,7 +522,7 @@ export async function showModelModal(model, modelType) {
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<div class="showcase-section" data-model-hash="${modelWithFullData.sha256 || ''}" data-filepath="${escapedFilePathAttr}">
|
||||
<div class="showcase-section" data-model-hash="${modelWithFullData.sha256 || ''}" data-model-name="${escapeAttribute(modelWithFullData.file_name || modelWithFullData.model_name || '')}" data-model-type="${modelType}" data-filepath="${escapedFilePathAttr}">
|
||||
<div class="showcase-tabs">
|
||||
${tabsContent}
|
||||
</div>
|
||||
|
||||
@@ -135,6 +135,39 @@ export function initLazyLoading(container) {
|
||||
lazyElements.forEach(element => observer.observe(element));
|
||||
}
|
||||
|
||||
/**
|
||||
* Check which Create As Recipe buttons correspond to already-imported
|
||||
* images and disable them.
|
||||
*/
|
||||
async function checkImportedRecipes(container) {
|
||||
const recipeButtons = container.querySelectorAll('.create-recipe-btn');
|
||||
if (!recipeButtons.length) return;
|
||||
|
||||
const imageIds = [];
|
||||
recipeButtons.forEach(btn => {
|
||||
const id = btn.dataset.imageId;
|
||||
if (id) imageIds.push(id);
|
||||
});
|
||||
if (!imageIds.length) return;
|
||||
|
||||
try {
|
||||
const response = await fetch(`/api/lm/recipes/check-image-exists?image_ids=${imageIds.join(',')}`);
|
||||
const data = await response.json();
|
||||
if (!data.success || !data.results) return;
|
||||
recipeButtons.forEach(btn => {
|
||||
const id = btn.dataset.imageId;
|
||||
if (id && data.results[id]?.in_library) {
|
||||
btn.disabled = true;
|
||||
btn.title = 'Already imported as recipe';
|
||||
btn.classList.add('disabled');
|
||||
}
|
||||
});
|
||||
} catch (err) {
|
||||
console.error('Failed to check imported recipes:', err);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Get the actual rendered rectangle of a media element with object-fit: contain
|
||||
* @param {HTMLElement} mediaElement - The img or video element
|
||||
@@ -471,6 +504,70 @@ export function initMediaControlHandlers(container) {
|
||||
});
|
||||
});
|
||||
|
||||
// Create As Recipe buttons
|
||||
const recipeButtons = container.querySelectorAll('.create-recipe-btn');
|
||||
recipeButtons.forEach(btn => {
|
||||
btn.addEventListener('click', async function(e) {
|
||||
e.stopPropagation();
|
||||
|
||||
const imageMetaRaw = this.dataset.imageMeta;
|
||||
const imageUrl = this.dataset.imageUrl;
|
||||
const imageNsfw = this.dataset.imageNsfw;
|
||||
const localPath = this.dataset.localPath || '';
|
||||
const showcaseSection = this.closest('.showcase-section');
|
||||
const modelHash = showcaseSection ? showcaseSection.dataset.modelHash : '';
|
||||
const modelName = showcaseSection ? showcaseSection.dataset.modelName : '';
|
||||
const modelType = showcaseSection ? showcaseSection.dataset.modelType : '';
|
||||
|
||||
if (!imageMetaRaw || !modelHash) {
|
||||
showToast('toast.recipes.createMissingData', {}, 'error');
|
||||
return;
|
||||
}
|
||||
|
||||
// Show loading state
|
||||
const originalHtml = this.innerHTML;
|
||||
this.innerHTML = '<i class="fas fa-spinner fa-spin"></i>';
|
||||
this.disabled = true;
|
||||
|
||||
try {
|
||||
const imageMeta = JSON.parse(decodeURIComponent(imageMetaRaw));
|
||||
|
||||
const response = await fetch('/api/lm/recipes/create-from-example', {
|
||||
method: 'POST',
|
||||
headers: { 'Content-Type': 'application/json' },
|
||||
body: JSON.stringify({
|
||||
image_data: {
|
||||
meta: imageMeta,
|
||||
url: imageUrl,
|
||||
nsfwLevel: imageNsfw ? parseInt(imageNsfw, 10) : undefined,
|
||||
},
|
||||
model_hash: modelHash,
|
||||
model_name: modelName || modelHash,
|
||||
model_type: modelType,
|
||||
local_image_path: localPath,
|
||||
}),
|
||||
});
|
||||
|
||||
const result = await response.json();
|
||||
|
||||
if (result.success && result.recipe_id) {
|
||||
showToast('toast.recipes.created', { recipeId: result.recipe_id }, 'success');
|
||||
} else {
|
||||
showToast('toast.recipes.createFailed', { error: result.error || 'Unknown error' }, 'error');
|
||||
}
|
||||
} catch (error) {
|
||||
console.error('Failed to create recipe:', error);
|
||||
showToast('toast.recipes.createError', { message: error.message }, 'error');
|
||||
} finally {
|
||||
this.innerHTML = originalHtml;
|
||||
this.disabled = false;
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
// Check which images are already imported as recipes → disable button
|
||||
checkImportedRecipes(container);
|
||||
|
||||
// Initialize set preview buttons
|
||||
initSetPreviewHandlers(container);
|
||||
|
||||
|
||||
@@ -183,6 +183,9 @@ function renderMediaItem(img, index, exampleFiles) {
|
||||
Math.min(maxHeightPercent, aspectRatio)
|
||||
);
|
||||
|
||||
// Extract CivitAI image ID from CDN URL for import status check
|
||||
const cdnImageId = (img.url || '').match(/\/(\d+)\.(?:jpeg|jpg|png|webp|gif)(?:\?|#|$)/)?.[1] || '';
|
||||
|
||||
// Check if media should be blurred
|
||||
const nsfwLevel = img.nsfwLevel !== undefined ? img.nsfwLevel : 0;
|
||||
const matureBlurThreshold = getMatureBlurThreshold(state.settings);
|
||||
@@ -224,12 +227,25 @@ function renderMediaItem(img, index, exampleFiles) {
|
||||
// Determine if this is a custom image (has id property)
|
||||
const isCustomImage = Boolean(typeof img.id === 'string' && img.id);
|
||||
|
||||
const hasGenMeta = img.hasMeta || (img.meta && (img.meta.prompt || img.meta.seed || img.meta.resources));
|
||||
|
||||
// Create the media control buttons HTML
|
||||
const mediaControlsHtml = `
|
||||
<div class="media-controls">
|
||||
<button class="media-control-btn set-preview-btn" title="Set as preview">
|
||||
<i class="fas fa-image"></i>
|
||||
</button>
|
||||
${hasGenMeta ? `
|
||||
<button class="media-control-btn create-recipe-btn"
|
||||
title="Create As Recipe"
|
||||
data-image-meta="${encodeURIComponent(JSON.stringify(img.meta || {}))}"
|
||||
data-image-url="${img.url || ''}"
|
||||
data-image-nsfw="${img.nsfwLevel ?? ''}"
|
||||
data-image-id="${cdnImageId}"
|
||||
data-local-path="${localFile ? localFile.path : ''}">
|
||||
<i class="fas fa-book-open"></i>
|
||||
</button>
|
||||
` : ''}
|
||||
<button class="media-control-btn set-nsfw-btn"
|
||||
title="Set content rating"
|
||||
data-media-index="${index}"
|
||||
|
||||
Reference in New Issue
Block a user