diff --git a/py/recipes/parsers/recipe_format.py b/py/recipes/parsers/recipe_format.py index 4b378c1c..5539701c 100644 --- a/py/recipes/parsers/recipe_format.py +++ b/py/recipes/parsers/recipe_format.py @@ -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() diff --git a/py/routes/handlers/recipe_handlers.py b/py/routes/handlers/recipe_handlers.py index 8c035a1e..2827123b 100644 --- a/py/routes/handlers/recipe_handlers.py +++ b/py/routes/handlers/recipe_handlers.py @@ -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( diff --git a/py/services/recipes/analysis_service.py b/py/services/recipes/analysis_service.py index 815d8a60..e4b9f60b 100644 --- a/py/services/recipes/analysis_service.py +++ b/py/services/recipes/analysis_service.py @@ -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, diff --git a/py/services/recipes/persistence_service.py b/py/services/recipes/persistence_service.py index 60e28455..0892f7c0 100644 --- a/py/services/recipes/persistence_service.py +++ b/py/services/recipes/persistence_service.py @@ -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) diff --git a/py/utils/exif_utils.py b/py/utils/exif_utils.py index 169fbb06..7ed97cc5 100644 --- a/py/utils/exif_utils.py +++ b/py/utils/exif_utils.py @@ -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(" 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 == [] diff --git a/tests/services/test_recipe_format_parser.py b/tests/services/test_recipe_format_parser.py index 4fe11f72..558636dd 100644 --- a/tests/services/test_recipe_format_parser.py +++ b/tests/services/test_recipe_format_parser.py @@ -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" diff --git a/tests/services/test_recipe_services.py b/tests/services/test_recipe_services.py index 9e899a0b..987b4881 100644 --- a/tests/services/test_recipe_services.py +++ b/tests/services/test_recipe_services.py @@ -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 diff --git a/tests/utils/test_exif_utils.py b/tests/utils/test_exif_utils.py index b1d7797d..09258b1a 100644 --- a/tests/utils/test_exif_utils.py +++ b/tests/utils/test_exif_utils.py @@ -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()