Compare commits

...

2 Commits

Author SHA1 Message Date
Will Miao 1fd7cc0123 fix(recipes): reject the empty-hash placeholder when resolving LoRA hashes
The SHA256 of an empty byte string (written by repackaging tools into
safetensors metadata, or produced by hashing an empty/unreadable file)
was previously resolved against CivitAI's by-hash API, which can contain
polluted entries for it (e.g. a broken SD 1.5 LoRA whose AutoV3 equals
the placeholder) and falsely attributed the wrong model to a recipe.

Guard all lookup paths for the 10/12/64-char AutoV2/AutoV3/full-SHA256
spellings: CivitaiClient.get_model_by_hash/_fetch_version_by_hash return
not-found without a request, and ModelHashIndex ignores the placeholder
in has_hash/get_path/add_autov3.

The Automatic1111 metadata parser keeps the LoRA item itself when its
hash is the placeholder: it matches by filename locally, or retains the
entry with an empty hash flagged hashInvalid (unresolvable-hash state in
the UI, with reconnect as the remedy) instead of dropping it or resolving
it to a polluted CivitAI entry.
2026-09-01 21:14:30 +08:00
Will Miao 39e7c1376c Support re-import for recipes without a source URL
Recipes imported by drag & drop / file-picker record no source_path and
were rejected by re-import. Fall back to the recipe's own saved image,
which still carries the original embedded generation metadata.

Re-import now re-parses that original metadata instead of the appended
recipe JSON block, so parser upgrades produce fresh results. The
already-optimized preview image is kept verbatim: only its WebP EXIF
chunk is rewritten in place to replace the recipe metadata block, and
the recipe JSON is rewritten with the new analysis plus carried-over
user edits.
2026-08-31 10:01:18 +08:00
17 changed files with 772 additions and 48 deletions
+21
View File
@@ -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:
+21
View File
@@ -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()
+42 -31
View File
@@ -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(
+8 -1
View File
@@ -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",
+9 -1
View File
@@ -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:
+17
View File
@@ -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,
+16 -3
View File
@@ -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
View File
@@ -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
View File
@@ -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"""
+79 -1
View File
@@ -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
+29
View File
@@ -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 == []
+36
View File
@@ -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"}
+36 -1
View File
@@ -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"
+196 -2
View File
@@ -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
+45
View File
@@ -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)
+63
View File
@@ -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()