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.
This commit is contained in:
Will Miao
2026-08-31 10:01:18 +08:00
parent 2a3c632dc5
commit 39e7c1376c
9 changed files with 537 additions and 41 deletions
+21
View File
@@ -208,3 +208,24 @@ class RecipeFormatParser(RecipeMetadataParser):
except Exception as e: except Exception as e:
logger.error(f"Error parsing recipe format metadata: {e}", exc_info=True) logger.error(f"Error parsing recipe format metadata: {e}", exc_info=True)
return {"error": str(e), "loras": []} 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) return web.json_response({"success": False, "error": str(exc)}, status=500)
async def reimport_recipe(self, request: web.Request) -> web.Response: 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 Gives the recipe a fresh start: URL-sourced recipes re-download the
CivitAI, re-parses EXIF metadata with the current parser, and image from CivitAI; local ones re-parse the saved recipe image. Both
re-resolves LoRAs / checkpoint. User edits (title, tags, favorite) use the original embedded generation metadata (the appended recipe
are carried over from the old 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: try:
await self._ensure_dependencies_ready() await self._ensure_dependencies_ready()
@@ -1108,13 +1110,34 @@ class RecipeManagementHandler:
if not old_recipe: if not old_recipe:
raise RecipeNotFoundError(f"Recipe {recipe_id} not found") raise RecipeNotFoundError(f"Recipe {recipe_id} not found")
source_path = old_recipe.get("source_path") old_file_path = old_recipe.get("file_path", "")
if not source_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( return web.json_response(
{ {
"success": False, "success": False,
"error": ( "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." "Use repair or manual import instead."
), ),
}, },
@@ -1128,28 +1151,9 @@ class RecipeManagementHandler:
if "tags" in user_edits and not isinstance(user_edits["tags"], list): if "tags" in user_edits and not isinstance(user_edits["tags"], list):
del user_edits["tags"] del user_edits["tags"]
old_file_path = old_recipe.get("file_path", "") if local_source:
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:
return await self._do_reimport_from_local( return await self._do_reimport_from_local(
source_path, local_source,
recipe_scanner, recipe_scanner,
recipe_id=recipe_id, recipe_id=recipe_id,
target_dir=old_folder, target_dir=old_folder,
@@ -2500,8 +2504,10 @@ class RecipeManagementHandler:
) -> web.Response: ) -> web.Response:
"""Re-import a recipe from a local image file. """Re-import a recipe from a local image file.
Reads the original source file, re-parses its EXIF metadata, saves a Reads the original source file, re-parses its original embedded
fresh recipe, then deletes the old one. 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) normalized = os.path.normpath(file_path)
if not os.path.isfile(normalized): if not os.path.isfile(normalized):
@@ -2517,6 +2523,7 @@ class RecipeManagementHandler:
analysis_result = await self._analysis_service.analyze_local_image( analysis_result = await self._analysis_service.analyze_local_image(
file_path=normalized, file_path=normalized,
recipe_scanner=recipe_scanner, recipe_scanner=recipe_scanner,
ignore_recipe_metadata=True,
) )
analysis_payload: dict[str, Any] = analysis_result.payload analysis_payload: dict[str, Any] = analysis_result.payload
@@ -2561,6 +2568,10 @@ class RecipeManagementHandler:
metadata=metadata, metadata=metadata,
extension=extension, extension=extension,
target_dir=target_dir, 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( await self._persistence_service.delete_recipe(
+17
View File
@@ -368,6 +368,7 @@ class RecipeAnalysisService:
*, *,
file_path: str | None, file_path: str | None,
recipe_scanner, recipe_scanner,
ignore_recipe_metadata: bool = False,
) -> AnalysisResult: ) -> AnalysisResult:
"""Analyze a file already present on disk.""" """Analyze a file already present on disk."""
@@ -389,6 +390,22 @@ class RecipeAnalysisService:
} }
return result 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( result = await self._parse_metadata(
metadata, metadata,
recipe_scanner=recipe_scanner, recipe_scanner=recipe_scanner,
+16 -3
View File
@@ -58,6 +58,7 @@ class RecipePersistenceService:
extension: str | None = None, extension: str | None = None,
recipe_id: str | None = None, recipe_id: str | None = None,
target_dir: str | None = None, target_dir: str | None = None,
skip_optimize: bool = False,
) -> PersistenceResult: ) -> PersistenceResult:
"""Persist a user uploaded recipe. """Persist a user uploaded recipe.
@@ -67,6 +68,11 @@ class RecipePersistenceService:
target_dir: If provided, save recipe files to this directory instead target_dir: If provided, save recipe files to this directory instead
of the default recipes_dir. Used by re-import to preserve the of the default recipes_dir. Used by re-import to preserve the
original folder location. 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 = [] missing_fields = []
@@ -87,9 +93,12 @@ class RecipePersistenceService:
recipe_id = recipe_id or str(uuid.uuid4()) 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"] is_video = extension in [".mp4", ".webm"]
if is_video: if is_video or skip_optimize:
optimized_image = resolved_image_bytes optimized_image = resolved_image_bytes
# extension is already set # extension is already set
else: else:
@@ -175,7 +184,11 @@ class RecipePersistenceService:
json.dump(recipe_data, file_obj, indent=4, ensure_ascii=False) json.dump(recipe_data, file_obj, indent=4, ensure_ascii=False)
if not is_video: 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) matching_recipes = await self._find_matching_recipes(recipe_scanner, fingerprint, exclude_id=recipe_id)
await recipe_scanner.add_recipe(recipe_data) await recipe_scanner.add_recipe(recipe_data)
+67 -3
View File
@@ -348,8 +348,14 @@ class ExifUtils:
return image_path return image_path
@staticmethod @staticmethod
def append_recipe_metadata(image_path, recipe_data) -> str: def append_recipe_metadata(image_path, recipe_data, pixel_preserving=False) -> str:
"""Append recipe metadata to an image's EXIF data""" """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: try:
if image_path: if image_path:
ext = os.path.splitext(image_path)[1].lower() ext = os.path.splitext(image_path)[1].lower()
@@ -417,13 +423,71 @@ class ExifUtils:
# Append to existing metadata or create new one # Append to existing metadata or create new one
new_metadata = f"{metadata} \n {recipe_metadata_marker}" if metadata else recipe_metadata_marker 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 # Write back to the image
return ExifUtils.update_image_metadata(image_path, new_metadata) return ExifUtils.update_image_metadata(image_path, new_metadata)
except Exception as e: except Exception as e:
logger.error(f"Error appending recipe metadata: {e}", exc_info=True) logger.error(f"Error appending recipe metadata: {e}", exc_info=True)
return image_path 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 @staticmethod
def remove_recipe_metadata(user_comment): def remove_recipe_metadata(user_comment):
"""Remove recipe metadata from user comment""" """Remove recipe metadata from user comment"""
+79 -1
View File
@@ -197,6 +197,7 @@ class StubAnalysisService:
self.upload_calls: List[bytes] = [] self.upload_calls: List[bytes] = []
self.remote_calls: List[Optional[str]] = [] self.remote_calls: List[Optional[str]] = []
self.local_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.result = SimpleNamespace(payload={"loras": []}, status=200)
self._recipe_parser_factory: Any = None self._recipe_parser_factory: Any = None
StubAnalysisService.instances.append(self) StubAnalysisService.instances.append(self)
@@ -218,11 +219,16 @@ class StubAnalysisService:
return self.result return self.result
async def analyze_local_image( 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 ) -> SimpleNamespace: # noqa: D401
if self.raise_for_local: if self.raise_for_local:
raise self.raise_for_local raise self.raise_for_local
self.local_calls.append(file_path) self.local_calls.append(file_path)
self.local_ignore_recipe_metadata_calls.append(ignore_recipe_metadata)
return self.result return self.result
async def analyze_widget_metadata(self, *, recipe_scanner) -> SimpleNamespace: async def analyze_widget_metadata(self, *, recipe_scanner) -> SimpleNamespace:
@@ -257,6 +263,7 @@ class StubPersistenceService:
extension=None, extension=None,
recipe_id=None, recipe_id=None,
target_dir=None, target_dir=None,
skip_optimize=False,
) -> SimpleNamespace: # noqa: D401 ) -> SimpleNamespace: # noqa: D401
self.save_calls.append( self.save_calls.append(
{ {
@@ -269,6 +276,7 @@ class StubPersistenceService:
"extension": extension, "extension": extension,
"recipe_id": recipe_id, "recipe_id": recipe_id,
"target_dir": target_dir, "target_dir": target_dir,
"skip_optimize": skip_optimize,
} }
) )
return self.save_result return self.save_result
@@ -2168,3 +2176,73 @@ async def test_checkpoint_mark_hash_invalid_route_requires_recipe_id(
json={}, json={},
) )
assert response.status == 400 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 == []
+36 -1
View File
@@ -3,7 +3,7 @@ from typing import Any, Dict
import pytest 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 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] lora_entry = result["loras"][0]
assert lora_entry["existsLocally"] is False assert lora_entry["existsLocally"] is False
assert lora_entry["localPath"] is None 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 self.optimized_calls += 1
return image_data, ".webp" return image_data, ".webp"
def append_recipe_metadata(self, image_path, recipe_data): def append_recipe_metadata(self, image_path, recipe_data, pixel_preserving=False):
self.appended = (image_path, recipe_data) self.appended = (image_path, recipe_data, pixel_preserving)
def extract_image_metadata(self, path): def extract_image_metadata(self, path):
return {} 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" 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 @pytest.mark.asyncio
async def test_analyze_remote_image_download_failure_cleans_temp(tmp_path, monkeypatch): async def test_analyze_remote_image_download_failure_cleans_temp(tmp_path, monkeypatch):
exif_utils = DummyExifUtils() 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["hash_invalid"] is False
assert result.payload["updated_checkpoint"]["hashInvalid"] 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
+63
View File
@@ -65,6 +65,69 @@ def test_append_recipe_metadata_includes_checkpoint(monkeypatch, tmp_path):
assert payload["base_model"] == "Illustrious" 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): def test_optimize_image_preserves_workflow_when_converting_png_to_webp(tmp_path):
image_path = tmp_path / "source.png" image_path = tmp_path / "source.png"
png_info = PngImagePlugin.PngInfo() png_info = PngImagePlugin.PngInfo()