mirror of
https://github.com/willmiao/ComfyUI-Lora-Manager.git
synced 2026-09-21 03:01:27 -03:00
Compare commits
2 Commits
2a3c632dc5
...
1fd7cc0123
| Author | SHA1 | Date | |
|---|---|---|---|
| 1fd7cc0123 | |||
| 39e7c1376c |
@@ -8,6 +8,7 @@ from typing import Dict, Any
|
||||
from ..base import RecipeMetadataParser
|
||||
from ..constants import GEN_PARAM_KEYS
|
||||
from ...services.metadata_service import get_default_metadata_provider
|
||||
from ...utils.constants import is_empty_placeholder_hash
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -524,6 +525,26 @@ class AutomaticMetadataParser(RecipeMetadataParser):
|
||||
weight = prompt_entries[0][1] if len(prompt_entries) == 1 else 1.0
|
||||
lora_entry = make_lora_entry(lora_type, lora_name, weight, lora_hash)
|
||||
|
||||
if is_empty_placeholder_hash(lora_hash):
|
||||
# The empty-hash placeholder (SHA256 of an empty byte
|
||||
# string) is not a real hash: never look it up in the
|
||||
# local hash index or on CivitAI. Match by filename;
|
||||
# otherwise keep the item as unresolved (no hash, flagged
|
||||
# hashInvalid so the UI shows the unresolvable-hash state
|
||||
# and offers reconnect instead of download) rather than
|
||||
# dropping it.
|
||||
if recipe_scanner and lora_type == 'lora' and basename_key not in queried_local_basenames:
|
||||
local_lora = await recipe_scanner.get_local_lora(lora_name, recipe_base_model)
|
||||
if local_lora:
|
||||
local_entry = self.populate_lora_from_local(lora_entry, local_lora)
|
||||
merge_or_append_local(local_entry)
|
||||
continue
|
||||
lora_entry['hash'] = ''
|
||||
lora_entry['hashInvalid'] = True
|
||||
if not resource_lora_count:
|
||||
loras.append(lora_entry)
|
||||
continue
|
||||
|
||||
if lora_hash and recipe_scanner and lora_type == 'lora':
|
||||
local_lora = await recipe_scanner.get_local_lora_by_hash(lora_hash)
|
||||
if local_lora:
|
||||
|
||||
@@ -208,3 +208,24 @@ class RecipeFormatParser(RecipeMetadataParser):
|
||||
except Exception as e:
|
||||
logger.error(f"Error parsing recipe format metadata: {e}", exc_info=True)
|
||||
return {"error": str(e), "loras": []}
|
||||
|
||||
|
||||
def strip_recipe_metadata(metadata_text: str) -> str:
|
||||
"""Strip the ``Recipe metadata: {...}`` block appended by LoRA Manager.
|
||||
|
||||
The saved recipe image carries the original generation metadata followed
|
||||
by an appended recipe JSON block (see ``ExifUtils.append_recipe_metadata``).
|
||||
Re-import wants to re-parse the original embedded metadata, so this returns
|
||||
only the text before the appended marker. The input is returned unchanged
|
||||
when no marker is present.
|
||||
"""
|
||||
if not metadata_text:
|
||||
return metadata_text
|
||||
match = re.search(
|
||||
RecipeFormatParser.METADATA_MARKER,
|
||||
metadata_text,
|
||||
re.IGNORECASE | re.DOTALL,
|
||||
)
|
||||
if not match:
|
||||
return metadata_text
|
||||
return metadata_text[: match.start()].strip()
|
||||
|
||||
@@ -1090,12 +1090,14 @@ class RecipeManagementHandler:
|
||||
return web.json_response({"success": False, "error": str(exc)}, status=500)
|
||||
|
||||
async def reimport_recipe(self, request: web.Request) -> web.Response:
|
||||
"""Delete a recipe and re-import it from its source URL.
|
||||
"""Delete a recipe and re-import it from its source.
|
||||
|
||||
This gives the recipe a fresh start — re-downloads the image from
|
||||
CivitAI, re-parses EXIF metadata with the current parser, and
|
||||
re-resolves LoRAs / checkpoint. User edits (title, tags, favorite)
|
||||
are carried over from the old recipe.
|
||||
Gives the recipe a fresh start: URL-sourced recipes re-download the
|
||||
image from CivitAI; local ones re-parse the saved recipe image. Both
|
||||
use the original embedded generation metadata (the appended recipe
|
||||
metadata block is ignored) with the current parser, and re-resolve
|
||||
LoRAs / checkpoint. User edits (title, tags, favorite) are carried
|
||||
over from the old recipe.
|
||||
"""
|
||||
try:
|
||||
await self._ensure_dependencies_ready()
|
||||
@@ -1108,13 +1110,34 @@ class RecipeManagementHandler:
|
||||
if not old_recipe:
|
||||
raise RecipeNotFoundError(f"Recipe {recipe_id} not found")
|
||||
|
||||
source_path = old_recipe.get("source_path")
|
||||
if not source_path:
|
||||
old_file_path = old_recipe.get("file_path", "")
|
||||
old_folder = os.path.dirname(old_file_path) if old_file_path else None
|
||||
|
||||
source_path = old_recipe.get("source_path") or ""
|
||||
image_id = extract_civitai_image_id(source_path) if source_path else None
|
||||
|
||||
# Local re-import sources: an explicit local source_path, or — when
|
||||
# no source_path was recorded (drag & drop / file-picker imports) —
|
||||
# the recipe's own saved image, which still carries the original
|
||||
# embedded generation metadata next to the recipe metadata block.
|
||||
local_source = None
|
||||
if not image_id and source_path and os.path.isfile(source_path):
|
||||
local_source = source_path
|
||||
elif (
|
||||
not image_id
|
||||
and not source_path
|
||||
and old_file_path
|
||||
and os.path.isfile(old_file_path)
|
||||
):
|
||||
local_source = old_file_path
|
||||
|
||||
if not image_id and not local_source:
|
||||
return web.json_response(
|
||||
{
|
||||
"success": False,
|
||||
"error": (
|
||||
"Recipe has no source URL — cannot re-import. "
|
||||
"Recipe has no re-importable source (no source URL "
|
||||
"and no accessible local image). "
|
||||
"Use repair or manual import instead."
|
||||
),
|
||||
},
|
||||
@@ -1128,28 +1151,9 @@ class RecipeManagementHandler:
|
||||
if "tags" in user_edits and not isinstance(user_edits["tags"], list):
|
||||
del user_edits["tags"]
|
||||
|
||||
old_file_path = old_recipe.get("file_path", "")
|
||||
old_folder = os.path.dirname(old_file_path) if old_file_path else None
|
||||
|
||||
image_id = extract_civitai_image_id(source_path)
|
||||
is_local_file = not image_id and os.path.isfile(source_path)
|
||||
|
||||
if not image_id and not is_local_file:
|
||||
return web.json_response(
|
||||
{
|
||||
"success": False,
|
||||
"error": (
|
||||
"Recipe source is neither a valid CivitAI image URL "
|
||||
"nor an accessible local file. "
|
||||
"Use repair or manual import instead."
|
||||
),
|
||||
},
|
||||
status=400,
|
||||
)
|
||||
|
||||
if is_local_file:
|
||||
if local_source:
|
||||
return await self._do_reimport_from_local(
|
||||
source_path,
|
||||
local_source,
|
||||
recipe_scanner,
|
||||
recipe_id=recipe_id,
|
||||
target_dir=old_folder,
|
||||
@@ -2500,8 +2504,10 @@ class RecipeManagementHandler:
|
||||
) -> web.Response:
|
||||
"""Re-import a recipe from a local image file.
|
||||
|
||||
Reads the original source file, re-parses its EXIF metadata, saves a
|
||||
fresh recipe, then deletes the old one.
|
||||
Reads the original source file, re-parses its original embedded
|
||||
generation metadata (the appended recipe metadata block is ignored so
|
||||
the current parser gets a fresh pass), saves a new recipe, then deletes
|
||||
the old one.
|
||||
"""
|
||||
normalized = os.path.normpath(file_path)
|
||||
if not os.path.isfile(normalized):
|
||||
@@ -2517,6 +2523,7 @@ class RecipeManagementHandler:
|
||||
analysis_result = await self._analysis_service.analyze_local_image(
|
||||
file_path=normalized,
|
||||
recipe_scanner=recipe_scanner,
|
||||
ignore_recipe_metadata=True,
|
||||
)
|
||||
analysis_payload: dict[str, Any] = analysis_result.payload
|
||||
|
||||
@@ -2561,6 +2568,10 @@ class RecipeManagementHandler:
|
||||
metadata=metadata,
|
||||
extension=extension,
|
||||
target_dir=target_dir,
|
||||
# The source is the recipe's own already-optimized preview image;
|
||||
# store its bytes verbatim instead of re-compressing (which would
|
||||
# only degrade quality) and skip the metadata re-append.
|
||||
skip_optimize=True,
|
||||
)
|
||||
|
||||
await self._persistence_service.delete_recipe(
|
||||
|
||||
@@ -21,7 +21,7 @@ from .model_metadata_provider import (
|
||||
from .downloader import get_downloader
|
||||
from .errors import RateLimitError, ResourceNotFoundError
|
||||
from ..utils.civitai_utils import resolve_license_payload
|
||||
from ..utils.constants import MODEL_WEIGHT_FILE_TYPES
|
||||
from ..utils.constants import MODEL_WEIGHT_FILE_TYPES, is_empty_placeholder_hash
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -180,6 +180,11 @@ class CivitaiClient:
|
||||
async def get_model_by_hash(
|
||||
self, model_hash: str
|
||||
) -> Tuple[Optional[Dict[str, Any]], Optional[str]]:
|
||||
if is_empty_placeholder_hash(model_hash):
|
||||
# The empty-hash placeholder (SHA256 of an empty byte string)
|
||||
# matches no real file; CivitAI's by-hash index can contain
|
||||
# polluted entries for it, so never resolve it.
|
||||
return None, "Model not found"
|
||||
try:
|
||||
success, version = await self._make_request(
|
||||
"GET",
|
||||
@@ -503,6 +508,8 @@ class CivitaiClient:
|
||||
async def _fetch_version_by_hash(self, model_hash: Optional[str]) -> Optional[Dict[str, Any]]:
|
||||
if not model_hash:
|
||||
return None
|
||||
if is_empty_placeholder_hash(model_hash):
|
||||
return None
|
||||
|
||||
success, version = await self._make_request(
|
||||
"GET",
|
||||
|
||||
@@ -1,6 +1,8 @@
|
||||
from typing import Dict, Optional, Set, List
|
||||
import os
|
||||
|
||||
from ..utils.constants import is_empty_placeholder_hash
|
||||
|
||||
class ModelHashIndex:
|
||||
"""Index for looking up models by hash or filename"""
|
||||
|
||||
@@ -81,6 +83,8 @@ class ModelHashIndex:
|
||||
# mapping. First-time registrations stay O(1).
|
||||
if autov3:
|
||||
autov3 = autov3.lower()
|
||||
if is_empty_placeholder_hash(autov3):
|
||||
autov3 = None
|
||||
if is_re_registration and (existing_hash != sha256 or autov3):
|
||||
stale_autov3_keys = [
|
||||
key for key, mapped_path in self._autov3_to_path.items()
|
||||
@@ -93,7 +97,7 @@ class ModelHashIndex:
|
||||
|
||||
def add_autov3(self, autov3: str, file_path: str) -> None:
|
||||
"""Add or update an AutoV3-only index entry (used when only AutoV3 is known)"""
|
||||
if not autov3:
|
||||
if not autov3 or is_empty_placeholder_hash(autov3):
|
||||
return
|
||||
autov3 = autov3.lower()
|
||||
self._autov3_to_path[autov3] = file_path
|
||||
@@ -250,6 +254,8 @@ class ModelHashIndex:
|
||||
|
||||
def has_hash(self, hash_value: str) -> bool:
|
||||
"""Check if hash exists in index (SHA256, AutoV2, or AutoV3)"""
|
||||
if is_empty_placeholder_hash(hash_value):
|
||||
return False
|
||||
normalized = hash_value.lower()
|
||||
if normalized in self._hash_to_path:
|
||||
return True
|
||||
@@ -261,6 +267,8 @@ class ModelHashIndex:
|
||||
|
||||
def get_path(self, hash_value: str) -> Optional[str]:
|
||||
"""Get file path for a hash (SHA256, AutoV2, or AutoV3)"""
|
||||
if is_empty_placeholder_hash(hash_value):
|
||||
return None
|
||||
normalized = hash_value.lower()
|
||||
path = self._hash_to_path.get(normalized)
|
||||
if path is not None:
|
||||
|
||||
@@ -368,6 +368,7 @@ class RecipeAnalysisService:
|
||||
*,
|
||||
file_path: str | None,
|
||||
recipe_scanner,
|
||||
ignore_recipe_metadata: bool = False,
|
||||
) -> AnalysisResult:
|
||||
"""Analyze a file already present on disk."""
|
||||
|
||||
@@ -389,6 +390,22 @@ class RecipeAnalysisService:
|
||||
}
|
||||
return result
|
||||
|
||||
if ignore_recipe_metadata:
|
||||
# Re-import: re-parse the original embedded generation metadata
|
||||
# instead of the recipe JSON block LoRA Manager appended on save.
|
||||
from ...recipes.parsers.recipe_format import strip_recipe_metadata
|
||||
|
||||
metadata = strip_recipe_metadata(metadata)
|
||||
if not metadata:
|
||||
result = self._metadata_not_found_response(normalized_path)
|
||||
result.payload["diagnostics"] = {
|
||||
"channel": "local",
|
||||
"exif_present": True,
|
||||
"ignore_recipe_metadata": True,
|
||||
"reason": "only_recipe_metadata",
|
||||
}
|
||||
return result
|
||||
|
||||
result = await self._parse_metadata(
|
||||
metadata,
|
||||
recipe_scanner=recipe_scanner,
|
||||
|
||||
@@ -58,6 +58,7 @@ class RecipePersistenceService:
|
||||
extension: str | None = None,
|
||||
recipe_id: str | None = None,
|
||||
target_dir: str | None = None,
|
||||
skip_optimize: bool = False,
|
||||
) -> PersistenceResult:
|
||||
"""Persist a user uploaded recipe.
|
||||
|
||||
@@ -67,6 +68,11 @@ class RecipePersistenceService:
|
||||
target_dir: If provided, save recipe files to this directory instead
|
||||
of the default recipes_dir. Used by re-import to preserve the
|
||||
original folder location.
|
||||
skip_optimize: If True, store the image bytes verbatim without
|
||||
resizing/re-encoding (recipe metadata is still embedded via a
|
||||
byte-level EXIF update that leaves the pixels untouched). Used
|
||||
by local re-import, where the source is the recipe's own
|
||||
already-optimized preview image.
|
||||
"""
|
||||
|
||||
missing_fields = []
|
||||
@@ -87,9 +93,12 @@ class RecipePersistenceService:
|
||||
|
||||
recipe_id = recipe_id or str(uuid.uuid4())
|
||||
|
||||
# Handle video formats by bypassing optimization and metadata embedding
|
||||
# Handle video formats by bypassing optimization and metadata embedding.
|
||||
# Local re-import also bypasses optimization: the source is the
|
||||
# recipe's own already-optimized preview image, so re-compressing it
|
||||
# would only degrade quality.
|
||||
is_video = extension in [".mp4", ".webm"]
|
||||
if is_video:
|
||||
if is_video or skip_optimize:
|
||||
optimized_image = resolved_image_bytes
|
||||
# extension is already set
|
||||
else:
|
||||
@@ -175,7 +184,11 @@ class RecipePersistenceService:
|
||||
json.dump(recipe_data, file_obj, indent=4, ensure_ascii=False)
|
||||
|
||||
if not is_video:
|
||||
self._exif_utils.append_recipe_metadata(normalized_image_path, recipe_data)
|
||||
self._exif_utils.append_recipe_metadata(
|
||||
normalized_image_path,
|
||||
recipe_data,
|
||||
pixel_preserving=skip_optimize,
|
||||
)
|
||||
|
||||
matching_recipes = await self._find_matching_recipes(recipe_scanner, fingerprint, exclude_id=recipe_id)
|
||||
await recipe_scanner.add_recipe(recipe_data)
|
||||
|
||||
+26
-5
@@ -1,3 +1,5 @@
|
||||
from typing import Any
|
||||
|
||||
NSFW_LEVELS = {
|
||||
"PG": 1,
|
||||
"PG13": 2,
|
||||
@@ -99,11 +101,30 @@ DEFAULT_HASH_CHUNK_SIZE_MB = 4
|
||||
# absurd 64-bit header length from forcing a multi-GB allocation during scan.
|
||||
MAX_SAFETENSORS_HEADER_BYTES = 64 * 1024 * 1024
|
||||
|
||||
# First 12 chars of the SHA256 of an empty byte string. Some (re-packaging)
|
||||
# training tools write this placeholder into safetensors metadata instead of a
|
||||
# real hash; it must never be treated as a valid AutoV3 — several broken
|
||||
# models sharing it would collide in the hash index and falsely match recipes.
|
||||
INVALID_AUTOV3_EMPTY_HASH = "e3b0c44298fc"
|
||||
# SHA256 of an empty byte string. Some (re-packaging) training tools write a
|
||||
# truncated form of this placeholder into safetensors metadata (as
|
||||
# ``modelspec.hash_sha256`` / ``sshs_model_hash``), and hashing an empty or
|
||||
# unreadable file produces it directly. It must never be treated as a valid
|
||||
# hash: several broken models share it, CivitAI's by-hash index can contain
|
||||
# such polluted entries, and matching it falsely attributes recipes.
|
||||
EMPTY_HASH_SHA256 = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
|
||||
INVALID_AUTOV3_EMPTY_HASH = EMPTY_HASH_SHA256[:12]
|
||||
INVALID_AUTOV2_EMPTY_HASH = EMPTY_HASH_SHA256[:10]
|
||||
|
||||
|
||||
def is_empty_placeholder_hash(value: Any) -> bool:
|
||||
"""True for a 10/12/64-hex-char spelling of the empty-hash placeholder.
|
||||
|
||||
These are the AutoV2, AutoV3 and full-SHA256 forms of the placeholder;
|
||||
such values identify no real model and must never be resolved against
|
||||
local files or CivitAI.
|
||||
"""
|
||||
if not isinstance(value, str):
|
||||
return False
|
||||
v = value.strip().lower()
|
||||
if len(v) not in (10, 12, 64):
|
||||
return False
|
||||
return v == EMPTY_HASH_SHA256[: len(v)]
|
||||
|
||||
# Auto-organize settings
|
||||
AUTO_ORGANIZE_BATCH_SIZE = (
|
||||
|
||||
+67
-3
@@ -348,8 +348,14 @@ class ExifUtils:
|
||||
return image_path
|
||||
|
||||
@staticmethod
|
||||
def append_recipe_metadata(image_path, recipe_data) -> str:
|
||||
"""Append recipe metadata to an image's EXIF data"""
|
||||
def append_recipe_metadata(image_path, recipe_data, pixel_preserving=False) -> str:
|
||||
"""Append recipe metadata to an image's EXIF data
|
||||
|
||||
When ``pixel_preserving`` is True (and the image is a WebP) only the
|
||||
EXIF container is rewritten at the byte level, so the preview pixels
|
||||
are never re-encoded. Local re-import uses this because its source is
|
||||
the recipe's own already-optimized preview image.
|
||||
"""
|
||||
try:
|
||||
if image_path:
|
||||
ext = os.path.splitext(image_path)[1].lower()
|
||||
@@ -417,13 +423,71 @@ class ExifUtils:
|
||||
|
||||
# Append to existing metadata or create new one
|
||||
new_metadata = f"{metadata} \n {recipe_metadata_marker}" if metadata else recipe_metadata_marker
|
||||
|
||||
|
||||
# Write back to the image. Re-import keeps the already-optimized
|
||||
# preview pixels untouched and updates only the WebP EXIF chunk
|
||||
# instead of re-encoding the whole image.
|
||||
if pixel_preserving and image_path.lower().endswith(".webp"):
|
||||
metadata_fields = ExifUtils._load_structured_metadata(image_path)
|
||||
metadata_fields["parameters"] = new_metadata
|
||||
exif_bytes = ExifUtils._build_exif_bytes(metadata_fields)
|
||||
with open(image_path, "rb") as file_obj:
|
||||
image_bytes = file_obj.read()
|
||||
try:
|
||||
updated = ExifUtils._replace_webp_exif(image_bytes, exif_bytes)
|
||||
except ValueError:
|
||||
# Container without an EXIF chunk; fall back to re-encoding.
|
||||
return ExifUtils.update_image_metadata(image_path, new_metadata)
|
||||
with open(image_path, "wb") as file_obj:
|
||||
file_obj.write(updated)
|
||||
return image_path
|
||||
|
||||
# Write back to the image
|
||||
return ExifUtils.update_image_metadata(image_path, new_metadata)
|
||||
except Exception as e:
|
||||
logger.error(f"Error appending recipe metadata: {e}", exc_info=True)
|
||||
return image_path
|
||||
|
||||
@staticmethod
|
||||
def _replace_webp_exif(image_bytes: bytes, exif_bytes: bytes) -> bytes:
|
||||
"""Replace the EXIF chunk of a WebP file without re-encoding pixels."""
|
||||
if image_bytes[:4] != b"RIFF" or image_bytes[8:12] != b"WEBP":
|
||||
raise ValueError("Not a WebP file")
|
||||
# The WebP EXIF chunk stores raw TIFF data; strip the JPEG-style
|
||||
# "Exif\\0\\0" prefix that piexif.dump may prepend.
|
||||
tiff = exif_bytes[6:] if exif_bytes[:6] == b"Exif\x00\x00" else exif_bytes
|
||||
|
||||
out = bytearray(image_bytes[:12])
|
||||
pos = 12
|
||||
exif_payload = None
|
||||
while pos + 8 <= len(image_bytes):
|
||||
fourcc = image_bytes[pos : pos + 4]
|
||||
size = struct.unpack("<I", image_bytes[pos + 4 : pos + 8])[0]
|
||||
chunk_data = image_bytes[pos + 8 : pos + 8 + size]
|
||||
pad = size % 2
|
||||
if fourcc == b"EXIF":
|
||||
exif_payload = tiff
|
||||
else:
|
||||
out += (
|
||||
fourcc
|
||||
+ struct.pack("<I", size)
|
||||
+ chunk_data
|
||||
+ (b"\x00" * pad)
|
||||
)
|
||||
pos += 8 + size + pad
|
||||
|
||||
if exif_payload is None:
|
||||
raise ValueError("WebP has no EXIF chunk")
|
||||
|
||||
out += (
|
||||
b"EXIF"
|
||||
+ struct.pack("<I", len(exif_payload))
|
||||
+ exif_payload
|
||||
+ (b"\x00" * (len(exif_payload) % 2))
|
||||
)
|
||||
out[4:8] = struct.pack("<I", len(out) - 8)
|
||||
return bytes(out)
|
||||
|
||||
@staticmethod
|
||||
def remove_recipe_metadata(user_comment):
|
||||
"""Remove recipe metadata from user comment"""
|
||||
|
||||
@@ -197,6 +197,7 @@ class StubAnalysisService:
|
||||
self.upload_calls: List[bytes] = []
|
||||
self.remote_calls: List[Optional[str]] = []
|
||||
self.local_calls: List[Optional[str]] = []
|
||||
self.local_ignore_recipe_metadata_calls: List[bool] = []
|
||||
self.result = SimpleNamespace(payload={"loras": []}, status=200)
|
||||
self._recipe_parser_factory: Any = None
|
||||
StubAnalysisService.instances.append(self)
|
||||
@@ -218,11 +219,16 @@ class StubAnalysisService:
|
||||
return self.result
|
||||
|
||||
async def analyze_local_image(
|
||||
self, *, file_path: Optional[str], recipe_scanner
|
||||
self,
|
||||
*,
|
||||
file_path: Optional[str],
|
||||
recipe_scanner,
|
||||
ignore_recipe_metadata: bool = False,
|
||||
) -> SimpleNamespace: # noqa: D401
|
||||
if self.raise_for_local:
|
||||
raise self.raise_for_local
|
||||
self.local_calls.append(file_path)
|
||||
self.local_ignore_recipe_metadata_calls.append(ignore_recipe_metadata)
|
||||
return self.result
|
||||
|
||||
async def analyze_widget_metadata(self, *, recipe_scanner) -> SimpleNamespace:
|
||||
@@ -257,6 +263,7 @@ class StubPersistenceService:
|
||||
extension=None,
|
||||
recipe_id=None,
|
||||
target_dir=None,
|
||||
skip_optimize=False,
|
||||
) -> SimpleNamespace: # noqa: D401
|
||||
self.save_calls.append(
|
||||
{
|
||||
@@ -269,6 +276,7 @@ class StubPersistenceService:
|
||||
"extension": extension,
|
||||
"recipe_id": recipe_id,
|
||||
"target_dir": target_dir,
|
||||
"skip_optimize": skip_optimize,
|
||||
}
|
||||
)
|
||||
return self.save_result
|
||||
@@ -2168,3 +2176,73 @@ async def test_checkpoint_mark_hash_invalid_route_requires_recipe_id(
|
||||
json={},
|
||||
)
|
||||
assert response.status == 400
|
||||
|
||||
|
||||
async def test_reimport_without_source_path_falls_back_to_recipe_file(
|
||||
monkeypatch, tmp_path: Path
|
||||
) -> None:
|
||||
"""Drag & drop imports record no source_path; re-import must fall back to
|
||||
the recipe's own saved image and re-parse ignoring the recipe metadata."""
|
||||
async with recipe_harness(monkeypatch, tmp_path) as harness:
|
||||
recipe_file = harness.tmp_dir / "recipes" / "rec1.webp"
|
||||
recipe_file.parent.mkdir(parents=True, exist_ok=True)
|
||||
recipe_file.write_bytes(b"fake-image")
|
||||
|
||||
harness.scanner.recipes["rec1"] = {
|
||||
"id": "rec1",
|
||||
"title": "Old title",
|
||||
"file_path": str(recipe_file),
|
||||
"tags": ["tag1"],
|
||||
# no source_path on purpose
|
||||
}
|
||||
harness.analysis.result = SimpleNamespace(
|
||||
payload={
|
||||
"success": True,
|
||||
"recipe_id": "new-rec",
|
||||
"loras": [],
|
||||
},
|
||||
status=200,
|
||||
)
|
||||
harness.persistence.save_result = SimpleNamespace(
|
||||
payload={"success": True, "recipe_id": "new-rec"}, status=200
|
||||
)
|
||||
|
||||
response = await harness.client.post("/api/lm/recipe/rec1/reimport")
|
||||
payload = await response.json()
|
||||
|
||||
assert response.status == 200
|
||||
assert payload["success"] is True
|
||||
assert payload["old_recipe_id"] == "rec1"
|
||||
assert payload["recipe_id"] == "new-rec"
|
||||
# Local analysis is used on the saved image, ignoring recipe metadata.
|
||||
assert harness.analysis.local_calls == [str(recipe_file)]
|
||||
assert harness.analysis.local_ignore_recipe_metadata_calls == [True]
|
||||
# The old recipe is deleted after the fresh save.
|
||||
assert harness.persistence.delete_calls == ["rec1"]
|
||||
# The already-optimized preview image must be stored verbatim.
|
||||
assert harness.persistence.save_calls[-1]["skip_optimize"] is True
|
||||
assert harness.persistence.save_calls[-1]["image_bytes"] == b"fake-image"
|
||||
# User edits (title, tags) are carried over to the new recipe.
|
||||
assert harness.persistence.update_calls[-1]["recipe_id"] == "new-rec"
|
||||
assert harness.persistence.update_calls[-1]["updates"]["title"] == "Old title"
|
||||
assert harness.persistence.update_calls[-1]["updates"]["tags"] == ["tag1"]
|
||||
|
||||
|
||||
async def test_reimport_without_any_source_returns_400(
|
||||
monkeypatch, tmp_path: Path
|
||||
) -> None:
|
||||
"""Recipes with neither source_path nor an accessible image cannot re-import."""
|
||||
async with recipe_harness(monkeypatch, tmp_path) as harness:
|
||||
harness.scanner.recipes["rec2"] = {
|
||||
"id": "rec2",
|
||||
"title": "No source",
|
||||
"file_path": str(harness.tmp_dir / "recipes" / "missing.webp"),
|
||||
}
|
||||
|
||||
response = await harness.client.post("/api/lm/recipe/rec2/reimport")
|
||||
payload = await response.json()
|
||||
|
||||
assert response.status == 400
|
||||
assert payload["success"] is False
|
||||
assert harness.analysis.local_calls == []
|
||||
assert harness.persistence.delete_calls == []
|
||||
|
||||
@@ -503,3 +503,64 @@ async def test_parse_metadata_extracts_checkpoint_from_model_hash(monkeypatch):
|
||||
assert result["model"] == checkpoint
|
||||
assert result["base_model"] == "flux"
|
||||
assert result["loras"] == []
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_parse_metadata_keeps_empty_placeholder_hash_lora_unresolved(monkeypatch):
|
||||
"""A LoRA hash equal to the SHA256("") placeholder must never be resolved
|
||||
against CivitAI or the local hash index, but the LoRA item itself must be
|
||||
kept: matched by filename locally when present, otherwise kept as an
|
||||
unresolved entry (no hash) instead of being dropped."""
|
||||
queried_hashes = []
|
||||
|
||||
async def fake_metadata_provider():
|
||||
class Provider:
|
||||
async def get_model_by_hash(self, model_hash):
|
||||
queried_hashes.append(model_hash)
|
||||
return None, "Model not found"
|
||||
|
||||
async def get_model_version_info(self, version_id):
|
||||
raise AssertionError("get_model_version_info should not be called")
|
||||
|
||||
return Provider()
|
||||
|
||||
monkeypatch.setattr(
|
||||
"py.recipes.parsers.automatic.get_default_metadata_provider",
|
||||
fake_metadata_provider,
|
||||
)
|
||||
|
||||
parser = AutomaticMetadataParser()
|
||||
metadata_text = (
|
||||
"photo of a DeLorean DMC12, <lora:dmc12bttf:1.2>, at night\n"
|
||||
"Steps: 20, Sampler: Euler, CFG scale: 1, Seed: 2242760352, Size: 1280x720, "
|
||||
"Model: flux1-dev, Model hash: 3f97fdc57a, "
|
||||
'Lora hashes: "dmc12bttf: e3b0c44298fc"'
|
||||
)
|
||||
|
||||
# Local file with the same name: the item is matched by filename.
|
||||
scanner_with_local = LocalRecipeScanner({"dmc12bttf": local_lora("dmc12bttf")})
|
||||
result = await parser.parse_metadata(metadata_text, recipe_scanner=scanner_with_local)
|
||||
|
||||
assert "e3b0c44298fc" not in queried_hashes
|
||||
assert "e3b0c44298" not in queried_hashes
|
||||
assert scanner_with_local.hash_queries == []
|
||||
assert scanner_with_local.queries == ["dmc12bttf"]
|
||||
assert len(result["loras"]) == 1
|
||||
assert result["loras"][0]["file_name"] == "dmc12bttf"
|
||||
assert result["loras"][0]["weight"] == 1.2
|
||||
assert result["loras"][0]["existsLocally"] is True
|
||||
assert result["loras"][0]["isDeleted"] is False
|
||||
|
||||
# No local file: the item is kept as unresolved (empty hash, flagged
|
||||
# hashInvalid so the UI renders the unresolvable-hash badge).
|
||||
scanner_without_local = LocalRecipeScanner({})
|
||||
result = await parser.parse_metadata(metadata_text, recipe_scanner=scanner_without_local)
|
||||
|
||||
assert len(result["loras"]) == 1
|
||||
lora = result["loras"][0]
|
||||
assert lora["file_name"] == "dmc12bttf"
|
||||
assert lora["weight"] == 1.2
|
||||
assert lora["hash"] == ""
|
||||
assert lora["hashInvalid"] is True
|
||||
assert lora["existsLocally"] is False
|
||||
assert lora["isDeleted"] is False
|
||||
|
||||
@@ -789,3 +789,32 @@ async def test_get_creator_model_count_never_raises(downloader):
|
||||
|
||||
client = await CivitaiClient.get_instance()
|
||||
assert await client.get_creator_model_count("pixel") is None
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"placeholder_hash",
|
||||
[
|
||||
"e3b0c44298", # AutoV2 (10 chars)
|
||||
"e3b0c44298fc", # AutoV3 (12 chars)
|
||||
"e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855", # full SHA256
|
||||
],
|
||||
)
|
||||
async def test_get_model_by_hash_rejects_empty_placeholder_without_request(downloader, placeholder_hash):
|
||||
"""The empty-hash placeholder must never be resolved via the by-hash API:
|
||||
CivitAI's index can contain polluted entries for it (e.g. a broken SD 1.5
|
||||
LoRA whose AutoV3 equals the placeholder)."""
|
||||
requested = []
|
||||
|
||||
async def fake_make_request(method, url, use_auth=True, **kwargs):
|
||||
requested.append(url)
|
||||
return True, {}
|
||||
|
||||
downloader.make_request = fake_make_request
|
||||
|
||||
client = await CivitaiClient.get_instance()
|
||||
|
||||
result, error = await client.get_model_by_hash(placeholder_hash)
|
||||
|
||||
assert result is None
|
||||
assert error == "Model not found"
|
||||
assert requested == []
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
import pytest
|
||||
from py.services.model_hash_index import ModelHashIndex
|
||||
from py.utils.constants import EMPTY_HASH_SHA256
|
||||
|
||||
|
||||
class TestModelHashIndexRemoveByPath:
|
||||
@@ -253,3 +254,38 @@ class TestModelHashIndexAutov3:
|
||||
assert index.has_hash("abcdef123456") is False
|
||||
assert index.get_path("fedcba654321") == "/models/ckpt.safetensors"
|
||||
assert index.get_all_autov3() == {"fedcba654321": "/models/ckpt.safetensors"}
|
||||
|
||||
|
||||
class TestModelHashIndexEmptyPlaceholder:
|
||||
def test_add_autov3_rejects_empty_placeholder(self):
|
||||
index = ModelHashIndex()
|
||||
index.add_autov3("e3b0c44298fc", "/models/lora.safetensors")
|
||||
assert "e3b0c44298fc" not in index.get_all_autov3()
|
||||
|
||||
def test_add_entry_rejects_empty_placeholder_autov3(self):
|
||||
index = ModelHashIndex()
|
||||
index.add_entry("abc123", "/models/lora.safetensors", autov3="e3b0c44298fc")
|
||||
assert "e3b0c44298fc" not in index.get_all_autov3()
|
||||
|
||||
def test_has_hash_false_for_placeholder(self):
|
||||
index = ModelHashIndex()
|
||||
index.add_entry("abc123", "/models/lora.safetensors")
|
||||
assert not index.has_hash("e3b0c44298")
|
||||
assert not index.has_hash("e3b0c44298fc")
|
||||
assert not index.has_hash(EMPTY_HASH_SHA256)
|
||||
|
||||
def test_get_path_none_for_placeholder(self):
|
||||
index = ModelHashIndex()
|
||||
index.add_entry("abc123", "/models/lora.safetensors")
|
||||
assert index.get_path("e3b0c44298") is None
|
||||
assert index.get_path("e3b0c44298fc") is None
|
||||
assert index.get_path(EMPTY_HASH_SHA256) is None
|
||||
|
||||
def test_placeholder_autov3_does_not_clobber_existing_mapping(self):
|
||||
# Registering a path with a placeholder autov3 must not replace or
|
||||
# clear autov3 mappings already registered for other paths.
|
||||
index = ModelHashIndex()
|
||||
index.add_entry("a" * 64, "/models/real.safetensors", autov3="abcdef123456")
|
||||
index.add_entry("b" * 64, "/models/other.safetensors", autov3="e3b0c44298fc")
|
||||
assert index.get_path("abcdef123456") == "/models/real.safetensors"
|
||||
assert index.get_all_autov3() == {"abcdef123456": "/models/real.safetensors"}
|
||||
|
||||
@@ -3,7 +3,7 @@ from typing import Any, Dict
|
||||
|
||||
import pytest
|
||||
|
||||
from py.recipes.parsers.recipe_format import RecipeFormatParser
|
||||
from py.recipes.parsers.recipe_format import RecipeFormatParser, strip_recipe_metadata
|
||||
from py.config import config
|
||||
|
||||
|
||||
@@ -425,3 +425,38 @@ async def test_recipe_format_parser_sha256_less_cache_item_no_keyerror(monkeypat
|
||||
lora_entry = result["loras"][0]
|
||||
assert lora_entry["existsLocally"] is False
|
||||
assert lora_entry["localPath"] is None
|
||||
|
||||
|
||||
def test_strip_recipe_metadata_removes_appended_marker():
|
||||
original = (
|
||||
"masterpiece, best quality\n"
|
||||
"Negative prompt: lowres\n"
|
||||
"Steps: 20, Sampler: DPM++ 2M Karras, CFG scale: 7, Seed: 123, "
|
||||
"Size: 512x768, Model hash: abc123, Model: foo_v1, Clip skip: 2\n"
|
||||
' Recipe metadata: {"title": "Saved", "loras": []}'
|
||||
)
|
||||
stripped = strip_recipe_metadata(original)
|
||||
|
||||
assert "Recipe metadata:" not in stripped
|
||||
assert stripped.startswith("masterpiece, best quality")
|
||||
assert "Steps: 20" in stripped
|
||||
assert '{"title": "Saved"}' not in stripped
|
||||
|
||||
|
||||
def test_strip_recipe_metadata_returns_input_without_marker():
|
||||
text = "Steps: 20, Sampler: DPM++ 2M Karras, Seed: 1"
|
||||
assert strip_recipe_metadata(text) == text
|
||||
|
||||
|
||||
def test_strip_recipe_metadata_empty_when_only_marker():
|
||||
text = ' Recipe metadata: {"title": "Saved"}'
|
||||
assert strip_recipe_metadata(text) == ""
|
||||
|
||||
|
||||
def test_strip_recipe_metadata_handles_multiline_json_marker():
|
||||
original = (
|
||||
"Steps: 20, Sampler: DPM++ 2M Karras, Seed: 1\n"
|
||||
' Recipe metadata: {"title": "Saved", "loras": [{"name": "a", "hash": "h"}]}'
|
||||
)
|
||||
stripped = strip_recipe_metadata(original)
|
||||
assert stripped == "Steps: 20, Sampler: DPM++ 2M Karras, Seed: 1"
|
||||
|
||||
@@ -32,8 +32,8 @@ class DummyExifUtils:
|
||||
self.optimized_calls += 1
|
||||
return image_data, ".webp"
|
||||
|
||||
def append_recipe_metadata(self, image_path, recipe_data):
|
||||
self.appended = (image_path, recipe_data)
|
||||
def append_recipe_metadata(self, image_path, recipe_data, pixel_preserving=False):
|
||||
self.appended = (image_path, recipe_data, pixel_preserving)
|
||||
|
||||
def extract_image_metadata(self, path):
|
||||
return {}
|
||||
@@ -87,6 +87,87 @@ async def test_save_recipe_video_bypasses_optimization(tmp_path):
|
||||
assert exif_utils.appended is None, "Metadata embedding should be bypassed for video"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_save_recipe_skip_optimize_preserves_image_bytes(tmp_path):
|
||||
"""Local re-import sources are already-optimized recipe images; saving them
|
||||
must keep the bytes verbatim instead of re-compressing, while the recipe
|
||||
metadata block is still embedded via a pixel-preserving EXIF update."""
|
||||
exif_utils = DummyExifUtils()
|
||||
|
||||
class DummyScanner:
|
||||
def __init__(self, root):
|
||||
self.recipes_dir = str(root / "recipes")
|
||||
|
||||
async def add_recipe(self, recipe_data):
|
||||
return None
|
||||
|
||||
async def find_recipes_by_fingerprint(self, fingerprint):
|
||||
return []
|
||||
|
||||
scanner = DummyScanner(tmp_path)
|
||||
service = RecipePersistenceService(
|
||||
exif_utils=exif_utils,
|
||||
card_preview_width=512,
|
||||
logger=logging.getLogger("test"),
|
||||
)
|
||||
|
||||
image_bytes = b"\x89PNG-not-optimized-again"
|
||||
result = await service.save_recipe(
|
||||
recipe_scanner=scanner,
|
||||
image_bytes=image_bytes,
|
||||
image_base64=None,
|
||||
name="Re-imported",
|
||||
tags=[],
|
||||
metadata={"gen_params": {"steps": 20}, "base_model": "SDXL", "loras": []},
|
||||
extension=".webp",
|
||||
skip_optimize=True,
|
||||
)
|
||||
|
||||
assert result.payload["image_path"].endswith(".webp")
|
||||
assert Path(result.payload["image_path"]).read_bytes() == image_bytes
|
||||
assert exif_utils.optimized_calls == 0, "Optimization should be bypassed"
|
||||
# Metadata is still embedded, but through the pixel-preserving path.
|
||||
assert exif_utils.appended is not None
|
||||
assert exif_utils.appended[2] is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_save_recipe_skip_optimize_default_optimizes(tmp_path):
|
||||
"""Normal saves must keep optimizing; only re-import opts out."""
|
||||
exif_utils = DummyExifUtils()
|
||||
|
||||
class DummyScanner:
|
||||
def __init__(self, root):
|
||||
self.recipes_dir = str(root / "recipes")
|
||||
|
||||
async def add_recipe(self, recipe_data):
|
||||
return None
|
||||
|
||||
async def find_recipes_by_fingerprint(self, fingerprint):
|
||||
return []
|
||||
|
||||
scanner = DummyScanner(tmp_path)
|
||||
service = RecipePersistenceService(
|
||||
exif_utils=exif_utils,
|
||||
card_preview_width=512,
|
||||
logger=logging.getLogger("test"),
|
||||
)
|
||||
|
||||
await service.save_recipe(
|
||||
recipe_scanner=scanner,
|
||||
image_bytes=b"raw-image",
|
||||
image_base64=None,
|
||||
name="Normal",
|
||||
tags=[],
|
||||
metadata={"gen_params": {"steps": 20}, "base_model": "SDXL", "loras": []},
|
||||
extension=".webp",
|
||||
)
|
||||
|
||||
assert exif_utils.optimized_calls == 1
|
||||
assert exif_utils.appended is not None
|
||||
assert exif_utils.appended[2] is False
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_analyze_remote_image_download_failure_cleans_temp(tmp_path, monkeypatch):
|
||||
exif_utils = DummyExifUtils()
|
||||
@@ -1979,3 +2060,116 @@ async def test_mark_checkpoint_hash_invalid_can_clear_flag(tmp_path):
|
||||
|
||||
assert result.payload["hash_invalid"] is False
|
||||
assert result.payload["updated_checkpoint"]["hashInvalid"] is False
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_analyze_local_image_ignore_recipe_metadata_strips_marker(tmp_path):
|
||||
"""Re-import must re-parse the original embedded metadata, not the
|
||||
recipe JSON block appended on save."""
|
||||
original = (
|
||||
"masterpiece, best quality\n"
|
||||
"Negative prompt: lowres\n"
|
||||
"Steps: 20, Sampler: DPM++ 2M Karras, CFG scale: 7, Seed: 1, "
|
||||
"Size: 512x768, Model hash: abc123, Model: foo_v1, Clip skip: 2\n"
|
||||
' Recipe metadata: {"title": "Saved", "loras": [], "gen_params": {}}'
|
||||
)
|
||||
|
||||
class SpyFactory:
|
||||
def __init__(self):
|
||||
self.received = None
|
||||
|
||||
def create_parser(self, metadata):
|
||||
self.received = metadata
|
||||
return _AutomaticMetadataSpyParser()
|
||||
|
||||
class _AutomaticMetadataSpyParser:
|
||||
async def parse_metadata(self, user_comment, recipe_scanner=None, civitai_client=None):
|
||||
return {"loras": [], "base_model": "Illustrious", "gen_params": {"seed": 1}}
|
||||
|
||||
class DummyScanner:
|
||||
async def find_recipes_by_fingerprint(self, fingerprint):
|
||||
return []
|
||||
|
||||
image_path = tmp_path / "rec.webp"
|
||||
image_path.write_bytes(b"fake-image")
|
||||
|
||||
factory = SpyFactory()
|
||||
service = _make_analysis_service(factory, _exif_utils_returning(original))
|
||||
|
||||
result = await service.analyze_local_image(
|
||||
file_path=str(image_path),
|
||||
recipe_scanner=DummyScanner(),
|
||||
ignore_recipe_metadata=True,
|
||||
)
|
||||
|
||||
# The parser must receive the original A1111 text without the appended
|
||||
# recipe metadata block, so it re-parses rather than reusing the snapshot.
|
||||
assert factory.received is not None
|
||||
assert "Recipe metadata:" not in factory.received
|
||||
assert factory.received.startswith("masterpiece, best quality")
|
||||
assert '{"title": "Saved"}' not in factory.received
|
||||
assert result.payload["parser"] == "_AutomaticMetadataSpyParser"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_analyze_local_image_ignore_recipe_metadata_only_marker(tmp_path):
|
||||
"""An image carrying only the recipe metadata block (no original embedded
|
||||
metadata) cannot be re-imported; report it instead of reusing the block."""
|
||||
original = 'Recipe metadata: {"title": "Saved", "loras": []}'
|
||||
|
||||
class NeverFactory:
|
||||
def create_parser(self, metadata):
|
||||
raise AssertionError("Parser must not run on stripped metadata")
|
||||
|
||||
image_path = tmp_path / "rec.webp"
|
||||
image_path.write_bytes(b"fake-image")
|
||||
|
||||
service = _make_analysis_service(NeverFactory(), _exif_utils_returning(original))
|
||||
|
||||
result = await service.analyze_local_image(
|
||||
file_path=str(image_path),
|
||||
recipe_scanner=SimpleNamespace(),
|
||||
ignore_recipe_metadata=True,
|
||||
)
|
||||
|
||||
assert "error" in result.payload
|
||||
assert result.payload["diagnostics"]["reason"] == "only_recipe_metadata"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_analyze_local_image_default_keeps_recipe_metadata_behavior(tmp_path):
|
||||
"""Normal import path keeps preferring the recipe metadata block."""
|
||||
original = (
|
||||
"Steps: 20, Sampler: DPM++ 2M Karras, Seed: 1\n"
|
||||
' Recipe metadata: {"title": "Saved", "loras": [], "gen_params": {}}'
|
||||
)
|
||||
|
||||
class SpyFactory:
|
||||
def __init__(self):
|
||||
self.received = None
|
||||
|
||||
def create_parser(self, metadata):
|
||||
self.received = metadata
|
||||
return _AutomaticMetadataSpyParser()
|
||||
|
||||
class _AutomaticMetadataSpyParser:
|
||||
async def parse_metadata(self, user_comment, recipe_scanner=None, civitai_client=None):
|
||||
return {"loras": [], "base_model": "Illustrious", "gen_params": {"seed": 1}}
|
||||
|
||||
class DummyScanner:
|
||||
async def find_recipes_by_fingerprint(self, fingerprint):
|
||||
return []
|
||||
|
||||
image_path = tmp_path / "rec.webp"
|
||||
image_path.write_bytes(b"fake-image")
|
||||
|
||||
factory = SpyFactory()
|
||||
service = _make_analysis_service(factory, _exif_utils_returning(original))
|
||||
|
||||
result = await service.analyze_local_image(
|
||||
file_path=str(image_path),
|
||||
recipe_scanner=DummyScanner(),
|
||||
)
|
||||
|
||||
assert factory.received == original
|
||||
assert "Recipe metadata:" in factory.received
|
||||
|
||||
@@ -0,0 +1,45 @@
|
||||
"""Tests for the empty-hash placeholder predicate in constants."""
|
||||
|
||||
from py.utils.constants import (
|
||||
EMPTY_HASH_SHA256,
|
||||
INVALID_AUTOV2_EMPTY_HASH,
|
||||
INVALID_AUTOV3_EMPTY_HASH,
|
||||
is_empty_placeholder_hash,
|
||||
)
|
||||
|
||||
|
||||
class TestIsEmptyPlaceholderHash:
|
||||
def test_full_length_sha256(self):
|
||||
assert is_empty_placeholder_hash(EMPTY_HASH_SHA256)
|
||||
|
||||
def test_autov3_length(self):
|
||||
assert is_empty_placeholder_hash("e3b0c44298fc")
|
||||
|
||||
def test_autov2_length(self):
|
||||
assert is_empty_placeholder_hash("e3b0c44298")
|
||||
|
||||
def test_case_insensitive(self):
|
||||
assert is_empty_placeholder_hash("E3B0C44298FC")
|
||||
assert is_empty_placeholder_hash(EMPTY_HASH_SHA256.upper())
|
||||
|
||||
def test_derived_constants_are_prefixes(self):
|
||||
assert INVALID_AUTOV2_EMPTY_HASH == EMPTY_HASH_SHA256[:10]
|
||||
assert INVALID_AUTOV3_EMPTY_HASH == EMPTY_HASH_SHA256[:12]
|
||||
|
||||
def test_rejects_other_lengths(self):
|
||||
# 8-char AutoV1-style prefix and non-placeholder lengths are not it
|
||||
assert not is_empty_placeholder_hash("e3b0c442")
|
||||
assert not is_empty_placeholder_hash("e3b0c44298fc1c")
|
||||
assert not is_empty_placeholder_hash("")
|
||||
|
||||
def test_rejects_real_hashes_that_share_the_prefix(self):
|
||||
# A real hash whose first characters coincide must not be rejected
|
||||
assert not is_empty_placeholder_hash("e3b0c44298aa")
|
||||
assert not is_empty_placeholder_hash("e3b0c44298fc" + "a" * 52)
|
||||
assert not is_empty_placeholder_hash("915a9a1f5f")
|
||||
assert not is_empty_placeholder_hash("915a9a1f5f58")
|
||||
assert not is_empty_placeholder_hash("a" * 64)
|
||||
|
||||
def test_rejects_non_strings(self):
|
||||
assert not is_empty_placeholder_hash(None)
|
||||
assert not is_empty_placeholder_hash(123)
|
||||
@@ -65,6 +65,69 @@ def test_append_recipe_metadata_includes_checkpoint(monkeypatch, tmp_path):
|
||||
assert payload["base_model"] == "Illustrious"
|
||||
|
||||
|
||||
def test_append_recipe_metadata_pixel_preserving_webp(tmp_path):
|
||||
"""pixel_preserving=True must rewrite only the EXIF chunk of a WebP,
|
||||
leaving the pixel chunks byte-identical and the old block replaced."""
|
||||
img = Image.new("RGB", (64, 48), (120, 30, 200))
|
||||
original_params = (
|
||||
"masterpiece, best quality\n"
|
||||
"Negative prompt: lowres\n"
|
||||
"Steps: 20, Sampler: DPM++ 2M Karras, CFG scale: 7, Seed: 1, "
|
||||
"Size: 512x768, Model hash: abc123, Model: foo_v1, Clip skip: 2\n"
|
||||
' Recipe metadata: {"title": "Old", "loras": []}'
|
||||
)
|
||||
exif = piexif.dump(
|
||||
{
|
||||
"0th": {},
|
||||
"Exif": {
|
||||
piexif.ExifIFD.UserComment: b"UNICODE\x00"
|
||||
+ original_params.encode("utf-16be")
|
||||
},
|
||||
}
|
||||
)
|
||||
image_path = tmp_path / "recipe.webp"
|
||||
img.save(str(image_path), format="WEBP", exif=exif, quality=85)
|
||||
|
||||
with open(image_path, "rb") as fh:
|
||||
before = fh.read()
|
||||
|
||||
new_recipe = {
|
||||
"title": "New",
|
||||
"base_model": "SDXL",
|
||||
"loras": [],
|
||||
"gen_params": {"steps": 25},
|
||||
}
|
||||
ExifUtils.append_recipe_metadata(
|
||||
str(image_path), new_recipe, pixel_preserving=True
|
||||
)
|
||||
|
||||
with open(image_path, "rb") as fh:
|
||||
after = fh.read()
|
||||
|
||||
def chunks(data: bytes) -> Dict[bytes, bytes]:
|
||||
pos, result = 12, {}
|
||||
while pos + 8 <= len(data):
|
||||
fourcc = data[pos : pos + 4]
|
||||
size = int.from_bytes(data[pos + 4 : pos + 8], "little")
|
||||
result[fourcc] = data[pos + 8 : pos + 8 + size]
|
||||
pos += 8 + size + (size % 2)
|
||||
return result
|
||||
|
||||
before_chunks = chunks(before)
|
||||
after_chunks = chunks(after)
|
||||
for fourcc, payload in before_chunks.items():
|
||||
if fourcc == b"EXIF":
|
||||
assert after_chunks[fourcc] != payload, "EXIF must be replaced"
|
||||
else:
|
||||
assert after_chunks[fourcc] == payload, f"{fourcc} was re-encoded"
|
||||
|
||||
# The appended block is updated; the original parameters stay in front.
|
||||
metadata = ExifUtils.extract_image_metadata(str(image_path))
|
||||
assert "Steps: 20" in metadata
|
||||
assert 'Recipe metadata: {"title": "New"' in metadata
|
||||
assert '"title": "Old"' not in metadata
|
||||
|
||||
|
||||
def test_optimize_image_preserves_workflow_when_converting_png_to_webp(tmp_path):
|
||||
image_path = tmp_path / "source.png"
|
||||
png_info = PngImagePlugin.PngInfo()
|
||||
|
||||
Reference in New Issue
Block a user